:py:mod:`bw_timex.timex_lca` ============================ .. py:module:: bw_timex.timex_lca Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: bw_timex.timex_lca.TimexLCA Attributes ~~~~~~~~~~ .. autoapisummary:: bw_timex.timex_lca.FACTORIZE_SOLVES_THRESHOLD .. py:data:: FACTORIZE_SOLVES_THRESHOLD :value: 8 .. py:class:: TimexLCA(demand: dict, method: tuple, database_dates: dict = None, use_global_lci_cache: bool = True) Class to perform time-explicit LCA calculations. A TimexLCA contains the LCI of processes occurring at explicit points in time. It tracks the timing of processes, relinks their technosphere and biosphere exchanges to match the technology landscape at that point in time, and also keeps track of the timing of the resulting emissions. As such, it combines prospective and dynamic LCA approaches. TimexLCA first calculates a static LCA, which informs a priority-first graph traversal. From the graph traversal, temporal relationships between exchanges and processes are derived. Based on the timing of the processes, bw_timex matches the processes at the intersection between foreground and background to the best available background databases. This temporal relinking is achieved by using datapackages to add new time-specific processes. The new processes and their exchanges to other technosphere processes or biosphere flows extent the technosphere and biosphere matrices. Temporal information of both processes and biosphere flows is retained, allowing for dynamic LCIA. TimexLCA calculates: 1) a static "base" LCA score (`TimexLCA.base_score`, same as `bw2calc.lca.score`), 2) a static time-explicit LCA score (`TimexLCA.static_score`), which links LCIs to the respective background databases, but without dynamic characterization of the time-explicit inventory 3) a dynamic time-explicit LCA score (`TimexLCA.dynamic_score`), with dynamic inventory and dynamic characterization. These are provided for radiative forcing and GWP but can also be user-defined. .. rubric:: Example >>> demand = {('my_foreground_database', 'my_process'): 1} >>> method = ("some_method_family", "some_category", "some_method") >>> database_dates = { 'my_background_database_one': datetime.strptime("2020", "%Y"), 'my_background_database_two': datetime.strptime("2030", "%Y"), 'my_background_database_three': datetime.strptime("2040", "%Y"), 'my_foreground_database':'dynamic' } >>> tlca = TimexLCA(demand, method, database_dates) >>> tlca.build_timeline() # has many optional arguments >>> tlca.lci() >>> tlca.static_lcia() >>> print(tlca.static_score) >>> tlca.dynamic_lcia(metric="radiative_forcing") # also available: "GWP" >>> print(tlca.dynamic_score) Instantiating a `TimexLCA` object calculates a static LCA, initializes time mappings for activities and biosphere flows, and stores useful subsets of ids in the node_collections. :param demand: The demand for which the LCA will be calculated. The keys can be Brightway `Node` instances, `(database, code)` tuples, or integer ids. :type demand: dict[object: float] :param method: Tuple defining the LCIA method, such as `('foo', 'bar')` or default methods, such as `("EF v3.1", "climate change", "global warming potential (GWP100)")` :type method: tuple :param database_dates: Dictionary mapping database names to dates. :type database_dates: dict, optional :param use_global_lci_cache: If True (default), background unit LCI matrices are cached at module level and reused across `TimexLCA` objects within the same Python session. The cache is keyed by background process identity plus the database's `modified` token, so edits to a background database invalidate stale entries automatically. Set to False to isolate this object's caching (e.g. when mutating background databases via raw SQL that bypasses bw2data). The module-level cache can be cleared with `bw_timex.clear_background_lci_cache()`. :type use_global_lci_cache: bool, optional .. py:property:: base_score :type: float Score of the base LCA, i.e., the "normal" LCA without time-explicit information. Same as bw2calc.LCA.score .. py:property:: static_score :type: float Score resulting from the static LCIA of the time-explicit inventory. .. py:property:: dynamic_score :type: float Score resulting from the dynamic LCIA of the time-explicit inventory. .. py:method:: build_timeline(starting_datetime: datetime.datetime | str = 'now', temporal_grouping: str = 'year', interpolation_type: str = 'linear', edge_filter_function: Callable = None, cutoff: float = 1e-09, max_calc: int = 2000, graph_traversal: str = 'priority', *args, **kwargs) -> pandas.DataFrame Creates a `TimelineBuilder` instance that does the graph traversal (similar to bw_temporalis) and extracts all edges with their temporal information. Creates the `TimexLCA.timeline` of technosphere exchanges. :param starting_datetime: Point in time when the demand occurs. This is the initial starting point of the graph traversal and the timeline. Something like `"now"` or `"2023-01-01"`. Default is `"now"`. :type starting_datetime: datetime | str, optional :param temporal_grouping: Time resolution for grouping exchanges over time in the timeline. Default is 'year', other options are 'month', 'day', 'hour'. :type temporal_grouping: str, optional :param interpolation_type: Type of interpolation when sourcing the new producers in the time-mapped background databases. Default is 'linear', which means linear interpolation between the closest 2 databases, other options are 'nearest' (or 'closest'), which selects only the closest database. :type interpolation_type: str, optional :param edge_filter_function: Function to skip edges in the graph traversal. Default is to skip all edges within background databases. :type edge_filter_function: Callable, optional :param cutoff: The cutoff value for the graph traversal. Default is 1e-9. :type cutoff: float, optional :param max_calc: The maximum number of calculations to be performed by the graph traversal. Default is 2000. :type max_calc: float, optional :param graph_traversal: The graph traversal algorithm to use. Default is 'priority' (priority-first, using bw_temporalis TemporalisLCA). Alternative is 'bfs' (Breadth-First-Search, independent of TemporalisLCA, avoids per-subgraph LCA overhead). :type graph_traversal: str, optional :param \*args: Positional arguments for the graph traversal, for `bw_temporalis.TemporalisLCA` passed to the `EdgeExtractor` class, which inherits from `TemporalisLCA`. See `bw_temporalis` documentation for more information. :type \*args: iterable :param \*\*kwargs: Additional keyword arguments for the graph traversal, for `bw_temporalis.TemporalisLCA` passed to the EdgeExtractor class, which inherits from TemporalisLCA. See bw_temporalis documentation for more information. :type \*\*kwargs: dict :returns: A DataFrame containing the timeline of technosphere exchanges :rtype: pandas.DataFrame .. seealso:: :obj:`bw_timex.timeline_builder.TimelineBuilder` Class that builds the timeline. .. py:method:: lci(build_dynamic_biosphere: Optional[bool] = True, expand_technosphere: Optional[bool] = True) -> None Calculates the time-explicit LCI. There are two ways to generate time-explicit LCIs: If `expand_technosphere' is True, the biosphere and technosphere matrices are expanded by inserting time-specific processes via the `MatrixModifier` class by calling `TimexLCA.build_datapackage(). Otherwise ('expand_technosphere' is False), it generates a dynamic inventory directly from the timeline without technosphere matrix calculations. Next to the choice above concerning how to retrieve the time-explicit inventory, users can also decide if they want to retain all temporal information at the biosphere level (build_dynamic_biosphere = True). Set `build_dynamic_biosphere` to False if you only want to get a new overall score of the time-explicit inventory and don't care about the timing of the emissions. This saves time and memory. :param build_dynamic_biosphere: if True, build the dynamic biosphere matrix and calculate the dynamic LCI. Default is True. :type build_dynamic_biosphere: bool :param expand_technosphere: if True, creates an expanded time-explicit technosphere and biosphere matrix and calculates the LCI from it. if False, creates no new technosphere, but calculates the dynamic inventory directly from the timeline. Building from the timeline currently only works if `build_dynamic_biosphere` is also True. :type expand_technosphere: bool :returns: calls LCI calculations from bw2calc and calculates the dynamic inventory, if `build_dynamic_biosphere` is True. :rtype: None .. seealso:: :obj:`build_datapackage` Method to create the datapackages that contain the modifications to the technosphere and biosphere matrix using the `MatrixModifier` class. :obj:`calculate_dynamic_inventory` Method to calculate the dynamic inventory if `build_dynamic_biosphere` is True. .. py:method:: disaggregate_background_lci() -> None This method disaggregates the background LCI's of the temporal markets. The disaggregated background LCI's allow a contribution analysis on the orginal inventory level as compared to the aggregated temporal market emissions. :param None: :returns: Stores the disaggregated background inventory in the attribute `dynamic_inventory_disaggregated` as a matrix and in `dynamic_inventory_disaggregated_df` as a DataFrame. :rtype: None .. py:method:: static_lcia() -> None Calculates static LCIA using time-explicit LCIs with the standard static characterization factors of the selected LCIA method using `bw2calc.lcia()`. :param None: :returns: Stores the static score in the attribute `static_score`. :rtype: None .. py:method:: dynamic_lcia(metric: str = 'radiative_forcing', time_horizon: int = 100, fixed_time_horizon: bool = False, time_horizon_start: datetime.datetime = None, characterization_functions: dict = None, characterization_function_co2: dict = None, use_disaggregated_lci: bool = False) -> pandas.DataFrame Calculates dynamic LCIA with the `DynamicCharacterization` class using the dynamic inventory and dynamic characterization functions. Dynamic characterization is handled by the separate package `dynamic_characterization` (https://dynamic-characterization.readthedocs.io). Dynamic characterization functions in the form of a dictionary {biosphere_flow_database_id: characterization_function} can be given by the user. If none are given, a set of default dynamic characterization functions based on IPCC AR6 are provided from `dynamic_characterization` package. These are mapped to the biosphere3 flows of the chosen static climate change impact category. If there is no characterization function for a biosphere flow, it will be ignored. Two dynamic climate change metrics are supported: "GWP" and "radiative_forcing". The time horizon for the impact assessment can be set with the `time_horizon` parameter, defaulting to 100 years. The `fixed_time_horizon` parameter determines whether the emission time horizon for all emissions is calculated from a specific starting point `time_horizon_start` (`fixed_time_horizon=True`) or from the time of the emission (`fixed_time_horizon=False`). The former is the implementation of the Levasseur approach (see https://doi.org/10.1021/es9030003), while the latter is how conventional LCA is done. :param metric: the metric for which the dynamic LCIA should be calculated. Default is "radiative_forcing". Available: "GWP" and "radiative_forcing" :type metric: str, optional :param time_horizon: the time horizon for the impact assessment. Unit is years. Default is 100. :type time_horizon: int, optional :param fixed_time_horizon: Whether the emission time horizon for all emissions is calculated from the functional unit (fixed_time_horizon=True) or from the time of the emission (fixed_time_horizon=False). Default is False. :type fixed_time_horizon: bool, optional :param time_horizon_start: The starting timestamp of the time horizon for the dynamic characterization. Only needed for fixed time horizons. Default is datetime.now(). :type time_horizon_start: pd.Timestamp, optional :param characterization_functions: Dict of the form {biosphere_flow_database_id: characterization_function}. Default is None, which triggers the use of the provided dynamic characterization functions based on IPCC AR6 Chapter 7. :type characterization_functions: dict, optional :param characterization_function_co2: Characterization function for CO2 emissions. Necessary if GWP metric is chosen. Default is None, which triggers the use of the provided dynamic characterization function of CO2 based on IPCC AR6 Chapter 7. :type characterization_function_co2: Callable, optional :param use_disaggregated_lci: Whether to use the disaggregated background LCI for the dynamic LCIA. Default is False. Use True if you want to perform a contribution analysis on the disaggregated background. :type use_disaggregated_lci: bool, optional :returns: A DataFrame with the characterized inventory for the chosen metric and parameters. :rtype: pandas.DataFrame .. seealso:: :obj:`dynamic_characterization` Package handling the dynamic characterization: https://dynamic-characterization.readthedocs.io/en/latest/ .. py:method:: build_datapackage() -> list Creates the datapackages that contain the modifications to the technosphere and biosphere matrix using the `MatrixModifier` class. :param None: :returns: List of datapackages that contain the modifications to the technosphere and biosphere matrix :rtype: list .. seealso:: :obj:`bw_timex.matrix_modifier.MatrixModifier` Class that handles the technosphere and biosphere matrix modifications. .. py:method:: _solve_cache_key(expand_technosphere: bool) -> tuple Fingerprint identifying a unique fu solve on this scenario. Reuse from ``LCI_SOLVE_CACHE`` is only safe when the consuming scenario produces *identical* technosphere/biosphere matrices and demand RHS. We hash the matrix data directly so any change in the timeline (different relinking, different temporal_grouping, etc.) produces a different key — `len(activity_time_mapping)` alone collides for differently-relinked scenarios of equal size. .. py:method:: _has_matching_cached_unit_lcis(expand_technosphere: bool) -> bool Whether the cache contains entries for any of this scenario's databases. Cache keys are ``("db_code", db, code, modified)`` (structure- independent triplet form); a hit on any background database in ``self.database_dates`` with a matching ``modified`` token means the build will likely reuse cached unit LCIs. .. py:method:: calculate_dynamic_inventory(expand_technosphere=True) -> None Calculates the dynamic inventory, by first creating a dynamic biosphere matrix using the `DynamicBiosphereBuilder` class and then multiplying it with the dynamic supply array. The dynamic inventory matrix is stored in the attribute `dynamic_inventory`. It is also converted to a DataFrame and stored in the attribute `dynamic_inventory_df`. :param expand_technosphere: A boolean indicating if the dynamic biosphere matrix is built directly from the expanded matrices or from the timeline. Default is True (from expanded matrices). :type expand_technosphere: bool :returns: calculates the dynamic inventory and stores it in the attribute `dynamic_inventory` as a matrix and in `dynamic_inventory_df` as a DataFrame. Also calculates and stores the lci of the temporal markets in the attribute self.temporal_market_lcis for use in contribution analysis of the background processes. :rtype: None .. seealso:: :obj:`bw_timex.dynamic_biosphere_builder.DynamicBiosphereBuilder` Class for creating the dynamic biosphere matrix and inventory. .. py:method:: create_dynamic_inventory_dataframe(expand_technosphere=True, use_disaggregated_lci=False) -> pandas.DataFrame Brings the dynamic inventory from its matrix form in `dynamic_inventory` into the format of a pandas.DataFrame, with the right structure to later apply dynamic characterization functions. Format is: +------------+--------+------+----------+ | date | amount | flow | activity | +============+========+======+==========+ | datetime | 33 | 1 | 2 | +------------+--------+------+----------+ | datetime | 32 | 1 | 2 | +------------+--------+------+----------+ | datetime | 31 | 1 | 2 | +------------+--------+------+----------+ - date: datetime, e.g. '2024-01-01 00:00:00' - flow: flow id - activity: activity id :param expand_technosphere: A boolean indicating if the dynamic biosphere matrix is built directly from the expanded matrices or from the timeline. Default is True. :type expand_technosphere: bool :rtype: pandas.DataFrame, dynamic inventory in DataFrame format .. py:method:: prepare_base_lca_inputs(demand=None, method=None, weighting=None, normalization=None, demands=None, remapping=True, demand_database_last=True) -> tuple Prepare LCA input arguments in Brightway2.5 style. Adapted bw2data.compat.py The difference to the original method is that we load all available databases into the matrices instead of just the ones depending on the demand. We need this for the creation of the time mapping dict that creates a mapping between the producer id and the reference timing of the databases in the `database_dates`. :param demand: The demand for which the LCA will be calculated. The keys can be Brightway `Node` instances, `(database, code)` tuples, or integer ids. :type demand: dict[object: float] :param method: Tuple defining the LCIA method, such as `('foo', 'bar')`. Only needed if not passing `data_objs`. :type method: tuple :param weighting: Tuple defining the LCIA weighting, such as `('foo', 'bar')`. Only needed if not passing `data_objs`. :type weighting: tuple :param normalization: :type normalization: str :param demands: :type demands: list of dicts of demands :param remapping: If True, remap dictionaries :type remapping: bool :param demand_database_last: If True, add the demand databases last in the list `database_names`. :type demand_database_last: bool :returns: Indexed demand, data objects, and remapping dictionaries :rtype: tuple .. seealso:: :obj:`bw2data.compat.prepare_lca_inputs` Original code this function is adapted from (https://github.com/brightway-lca/brightway2-data/blob/main/bw2data/compat.py). .. py:method:: prepare_bw_timex_inputs(demand=None, method=None, weighting=None, normalization=None, demands=None, remapping=True, demand_database_last=True) -> tuple Prepare LCA input arguments in Brightway 2.5 style. ORIGINALLY FROM bw2data.compat.py Changes include: - always load all databases in demand_database_names - indexed_demand has the id of the new consumer_id of the "exploded" demand :param demand: The demand for which the LCA will be calculated. The keys can be Brightway `Node` instances, `(database, code)` tuples, or integer ids. :type demand: dict[object: float] :param method: Tuple defining the LCIA method, such as `('foo', 'bar')`. Only needed if not passing `data_objs`. :type method: tuple :param demand_timing: Dictionary mapping demand ids to their timing. :type demand_timing: dict :param weighting: Tuple defining the LCIA weighting, such as `('foo', 'bar')`. Only needed if not passing `data_objs`. :type weighting: tuple :param normalization: :type normalization: str :param demands: :type demands: list of dicts of demands :param remapping: If True, remap dictionaries :type remapping: bool :param demand_database_last: If True, add the demand databases last in the list `database_names`. :type demand_database_last: bool :returns: Indexed demand, data objects, and remapping dictionaries :rtype: tuple .. seealso:: :obj:`bw2data.compat.prepare_lca_inputs` Original code this function is adapted from (https://github.com/brightway-lca/brightway2-data/blob/main/bw2data/compat.py). .. py:method:: create_node_collections() -> None Creates a dict of collections of nodes that will be useful down the line, e.g. to determine static nodes for the graph traversal or create the dynamic biosphere matrix. Available collections are: - ``background``: set of node ids of all processes that depend on the demand processes and are in the background databases - ``foreground``: set of node ids of all processes that are not in the background databases - ``first_level_background_static``: set of node ids of all processes that are in the background databases and are directly linked to the demand processes :param None: :returns: adds the `node_collections containing` the above-mentioned collections, as well as interdatabase_activity_mapping :rtype: None .. py:method:: add_interdatabase_activity_mapping_from_timeline() -> None Fills the interdatabase_activity_mapping, which is a SetList of the matching processes across background databases in the format of {(id, database_name_1), (id, database_name_2)} with only those activities and background databases that are actually mapped in the timeline. :param None: :returns: Adds the ids of producers in other background databases (only those interpolated to in the timeline) to the `interdatabase_activity_mapping`. :rtype: None .. py:method:: collect_temporalized_processes_from_timeline() -> None Prepares the input for the LCA from the timeline. :param None: :returns: Adds "temporal_markets" and "temporalized_processes" to the node_collections based on the timeline. :rtype: None .. py:method:: add_static_activities_to_activity_time_mapping() -> None Adds all activities from the static LCA to `activity_time_mapping`, an instance of `TimeMappingDict`. This gives a unique mapping in the form of (('database', 'code'), datetime_as_integer): time_mapping_id) that is later used to uniquely identify time-resolved processes. Here, the activity_time_mapping is the pre-population with the static activities. The time-explicit activities (from other temporalized background databases) are added later on by the TimelineBuilder. Activities in the foreground database are mapped with (('database', 'code'), "dynamic"): time_mapping_id)" as their timing is not yet known. :param None: :returns: adds the static activities to the `activity_time_mapping` :rtype: None .. py:method:: create_demand_timing() -> dict Generate a dictionary that maps producer (key) to timing (value) for the demands in the product system. It searches the timeline for those rows that contain the functional units (demand-processes as producer and -1 as consumer) and returns the time of the demand as an integer. Time of demand can have flexible resolution (year=YYYY, month=YYYYMM, day=YYYYMMDD, hour=YYYYMMDDHH) defined in `temporal_grouping`. :param None: :returns: Dictionary mapping producer ids to reference timing for the specified demands. :rtype: dict .. py:method:: create_labelled_technosphere_dataframe() -> pandas.DataFrame Returns the technosphere matrix as a dataframe with comprehensible labels instead of ids. :param None: :returns: technosphere matrix as a pandas.DataFrame with comprehensible labels instead of ids. :rtype: pd.DataFrame .. py:method:: create_labelled_biosphere_dataframe() -> pandas.DataFrame Returns the biosphere matrix as a pandas.DataFrame with comprehensible labels instead of ids. :param None: :returns: biosphere matrix as a pandas.DataFrame with comprehensible labels instead of ids. :rtype: pd.DataFrame .. py:method:: create_labelled_dynamic_biosphere_dataframe() -> pandas.DataFrame Returns the dynamic biosphere matrix as a dataframe with comprehensible labels instead of ids. :param None: :returns: dynamic biosphere matrix as a pandas.DataFrame with comprehensible labels instead of ids. :rtype: pd.DataFrame .. py:method:: get_activity_name_from_time_mapped_id(time_mapped_id: int) -> str Get the activity name for a time-mapped activity ID. Uses the pre-built code-to-name cache for efficient lookups. :param time_mapped_id: The time-mapped activity ID from activity_time_mapping :type time_mapped_id: int :returns: The name of the activity :rtype: str .. py:method:: create_labelled_dynamic_inventory_dataframe() -> pandas.DataFrame Returns the dynamic_inventory_df with comprehensible labels for flows and activities instead of ids. :param None: :returns: dynamic inventory matrix as a pandas.DataFrame with comprehensible labels instead of ids. :rtype: pd.DataFrame .. py:method:: plot_dynamic_inventory(bio_flows, cumulative=False) -> None Simple plot of dynamic inventory of a biosphere flow over time, with optional cumulative plotting. :param bio_flows: database ids of the biosphere flows to plot. :type bio_flows: list of int :param cumulative: if True, plot cumulative amounts over time :type cumulative: bool :returns: shows a plot :rtype: None .. py:method:: plot_dynamic_characterized_inventory(cumsum: bool = False, sum_emissions_within_activity: bool = False, sum_activities: bool = False) -> None Plot the characterized inventory of the dynamic LCI in a very simple plot. Legend and title are selected automatically based on the chosen metric. :param cumsum: if True, plot cumulative amounts over time :type cumsum: bool :param sum_emissions_within_activity: if True, sum emissions within each activity over time :type sum_emissions_within_activity: bool :param sum_activities: if True, sum emissions over all activities over time :type sum_activities: bool :returns: shows a plot :rtype: None