bw_timex.edge_extractor

Module Contents

Classes

Edge	Class for storing a temporal edge with source and target.
EdgeExtractor	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.
EdgeExtractorBFS	Breadth-First-Search (BFS) graph traversal for extracting temporal edges from

Functions

extract_temporal_evolution(→ dict | None)	Read temporal_evolution data from an exchange's data dict.
_join_datetime_and_timedelta_distributions(...)	Join a relative or absolute TemporalDistribution (td_producer) with an

Attributes

datetime_type
timedelta_type

bw_timex.edge_extractor.datetime_type

bw_timex.edge_extractor.timedelta_type

class bw_timex.edge_extractor.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).

edge_type: str

distribution: bw_temporalis.TemporalDistribution

leaf: bool

consumer: int

producer: int

td_producer: bw_temporalis.TemporalDistribution

td_consumer: bw_temporalis.TemporalDistribution

abs_td_producer: bw_temporalis.TemporalDistribution

abs_td_consumer: bw_temporalis.TemporalDistribution

temporal_evolution: dict

bw_timex.edge_extractor.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.

class bw_timex.edge_extractor.EdgeExtractor(*args, edge_filter_function: Callable = None, **kwargs)

Bases: bw_temporalis.TemporalisLCA

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.

Parameters:

• *args (Variable length argument list)

• edge_filter_function (Callable, optional) – A callable that filters edges. If not provided, a function that always returns False is used.

• **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.

Return type:

None

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).

Parameters:

None

Returns:

A list of Edge instances with timestamps and amounts, and ids of its producing and consuming node.

Return type:

list

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.

Parameters:

• td_producer (TemporalDistribution) – TemporalDistribution of the producer. Expected to be a timedelta or datetime TemporalDistribution.

• td_consumer (TemporalDistribution) – TemporalDistribution of the consumer. Expected to be a datetime TemporalDistribution.

Returns:

A new TemporalDistribution that is the result of joining the producer and consumer TemporalDistributions.

Return type:

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].

class bw_timex.edge_extractor.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.

_get_activity_dataset(activity_id: int) → bw2data.backends.schema.ActivityDataset

_get_exchange(input_id: int, output_id: int)

Look up exchange between two activities. Returns ExchangeDataset or None.

_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.

_get_production_amount(activity_id: int) → float

Get the reference product production amount (diagonal of tech matrix).

_get_technosphere_inputs(activity_id: int) → list[int]

Get all technosphere input activity IDs for a given activity.

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.

bw_timex.edge_extractor._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.