:py:mod:`bw_timex.edge_extractor` ================================= .. py:module:: bw_timex.edge_extractor Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: bw_timex.edge_extractor.Edge bw_timex.edge_extractor.EdgeExtractor bw_timex.edge_extractor.EdgeExtractorBFS Functions ~~~~~~~~~ .. autoapisummary:: bw_timex.edge_extractor.extract_temporal_evolution bw_timex.edge_extractor._join_datetime_and_timedelta_distributions Attributes ~~~~~~~~~~ .. autoapisummary:: bw_timex.edge_extractor.datetime_type bw_timex.edge_extractor.timedelta_type .. py:data:: datetime_type .. py:data:: timedelta_type .. py:class:: Edge Class for storing a temporal edge with source and target. Leaf edges link to a source process which is a leaf in our graph traversal (either through cutoff or a filter function). .. py:attribute:: edge_type :type: str .. py:attribute:: distribution :type: bw_temporalis.TemporalDistribution .. py:attribute:: leaf :type: bool .. py:attribute:: consumer :type: int .. py:attribute:: producer :type: int .. py:attribute:: td_producer :type: bw_temporalis.TemporalDistribution .. py:attribute:: td_consumer :type: bw_temporalis.TemporalDistribution .. py:attribute:: abs_td_producer :type: bw_temporalis.TemporalDistribution .. py:attribute:: abs_td_consumer :type: bw_temporalis.TemporalDistribution .. py:attribute:: temporal_evolution :type: dict .. py:function:: extract_temporal_evolution(exc_data: dict) -> dict | None Read ``temporal_evolution`` data from an exchange's data dict. Returns a ``{datetime: factor}`` dict, or ``None`` if the exchange carries no temporal evolution. ``temporal_evolution_amounts`` are normalized to factors using the exchange's base ``amount``. ``temporal_evolution_factors`` and ``temporal_evolution_amounts`` are mutually exclusive. .. py:class:: EdgeExtractor(*args, edge_filter_function: Callable = None, **kwargs) Bases: :py:obj:`bw_temporalis.TemporalisLCA` .. autoapi-inheritance-diagram:: bw_timex.edge_extractor.EdgeExtractor :parts: 1 :private-bases: Child class of TemporalisLCA that traverses the supply chain just as the parent class but can create a timeline of edges, in addition timeline of flows or nodes. The edge timeline is then used to match the timestamp of edges to that of background databases and to replace these edges with edges from these background databases using Brightway Datapackages. Initialize the EdgeExtractor class and traverses the supply chain using functions of the parent class TemporalisLCA. :param \*args: :type \*args: Variable length argument list :param edge_filter_function: A callable that filters edges. If not provided, a function that always returns False is used. :type edge_filter_function: Callable, optional :param \*\*kwargs: :type \*\*kwargs: Arbitrary keyword arguments :returns: stores the output of the TemporalisLCA graph traversal (incl. relation of edges (edge_mapping) and nodes (node_mapping) in the instance of the class. :rtype: None .. py:method:: build_edge_timeline() -> list Creates a timeline of the edges from the output of the graph traversal. Starting from the edges of the functional unit node, it goes through each node using a heap, selecting the node with the highest impact first. It, then, propagates the TemporalDistributions of the edges from node to node through time using convolution-operators. It stops in case the current edge is known to have no temporal distribution (=leaf) (e.g. part of background database). :param None: :returns: A list of Edge instances with timestamps and amounts, and ids of its producing and consuming node. :rtype: list .. py:method:: join_datetime_and_timedelta_distributions(td_producer: bw_temporalis.TemporalDistribution, td_consumer: bw_temporalis.TemporalDistribution) -> bw_temporalis.TemporalDistribution Joins a relative or absolute TemporalDistribution (td_producer) with an absolute TemporalDistribution (td_consumer) to create a new TemporalDistribution. If the producer does not have a TemporalDistribution, the consumer's TemporalDistribution is returned to continue the timeline. If both the producer and consumer have TemporalDistributions, they are joined together. :param td_producer: TemporalDistribution of the producer. Expected to be a timedelta or datetime TemporalDistribution. :type td_producer: TemporalDistribution :param td_consumer: TemporalDistribution of the consumer. Expected to be a datetime TemporalDistribution. :type td_consumer: TemporalDistribution :returns: A new TemporalDistribution that is the result of joining the producer and consumer TemporalDistributions. :rtype: TemporalDistribution :raises ValueError: If the dtype of `td_consumer.date` is not `datetime64[s]` or the dtype of `td_producer.date` is neither `datetime64[s]` nor `timedelta64[s]`. .. py:class:: EdgeExtractorBFS(lca_object, starting_datetime: datetime.datetime | str = 'now', edge_filter_function: Callable = None, cutoff: float = 1e-09, static_activity_indices: set[int] | None = None) Breadth-First-Search (BFS) graph traversal for extracting temporal edges from the supply chain. Unlike EdgeExtractor (which inherits from TemporalisLCA and uses priority-first traversal with per-subgraph LCA calculations), this class works directly with the technosphere matrix from a bw2calc LCA object and traverses using BFS. This avoids the overhead of computing individual subgraph LCAs for priority ordering. Returns the same list[Edge] format as EdgeExtractor, so all downstream code (TimelineBuilder, MatrixModifier, etc.) works unchanged. .. py:method:: _get_activity_dataset(activity_id: int) -> bw2data.backends.schema.ActivityDataset .. py:method:: _get_exchange(input_id: int, output_id: int) Look up exchange between two activities. Returns ExchangeDataset or None. .. py:method:: _get_exchange_td_and_type(input_id: int, output_id: int) Get temporal distribution, edge type and temporal evolution for an exchange. Returns (td_or_amount, edge_type, temporal_evolution) where td_or_amount is either a TemporalDistribution or a float (the signed matrix value), and temporal_evolution is a {datetime: factor} dict or None. .. py:method:: _get_production_amount(activity_id: int) -> float Get the reference product production amount (diagonal of tech matrix). .. py:method:: _get_technosphere_inputs(activity_id: int) -> list[int] Get all technosphere input activity IDs for a given activity. .. py:method:: build_edge_timeline() -> list Breadth-First-Search (BFS) traversal of the supply chain, extracting temporal edges. Returns a list of Edge instances compatible with the existing EdgeExtractor output format. .. py:function:: _join_datetime_and_timedelta_distributions(td_producer: bw_temporalis.TemporalDistribution, td_consumer: bw_temporalis.TemporalDistribution) -> bw_temporalis.TemporalDistribution Join a relative or absolute TemporalDistribution (td_producer) with an absolute TemporalDistribution (td_consumer). If the producer does not have a TemporalDistribution, the consumer's TemporalDistribution is returned. If both have TDs, they are joined via broadcasting.