bw_timex.edge_extractor

Module Contents

Classes

Edge	Class for storing a temporal edge with source and target.
VariantBackgroundMixin	Shared variant-aware (respective-variant) background-descent machinery.
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

temporal_evolution_reference: str = 'producer'

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

Shared variant-aware (respective-variant) background-descent machinery.

The base graph traversal (priority or BFS) runs on base_lca, which only contains the referenced background variant. When descent continues INTO a background process reached at a date that routes to a NON-referenced variant, the respective variant’s exchanges/amounts/TDs must be read from the bw2data activity proxy (self.bw_node_proxies) rather than from the (referenced-only) technosphere matrix or graph-traversal node objects.

Both EdgeExtractorBFS and the priority EdgeExtractor mix this in. They differ in how they reach the first background crossing (matrix BFS vs. TemporalisLCA heap), but the variant split + the proxy-only descent through the resulting variant subtree are identical, so they live here.

Mixers must provide: - self.bw_node_proxies: {activity_id: bw2data Activity proxy}. - self.database_dates_static, self.interpolation_type,

self.interdatabase_activity_mapping (set by TimelineBuilder).

• self.static_activity_indices (set; empty under traverse_background).

• self.variant_resolved_producers (set, collected during descent).

• self.cutoff and a self.edge_ff edge-filter callable.

_variant_shares_for_date(producer_date) → dict

Return {db_name: weight} interpolation shares for a cohort date.

Maps the producer’s absolute cohort date onto the available static background databases, using the same interpolation as the timeline builder so leaf and descended routing agree.

_resolve_in_variant(node_id: int, db_name: str) → int

Return the id of node_id’s sibling in database db_name.

If node_id already lives in db_name it is returned unchanged; otherwise the sibling is looked up via the interdatabase mapping.

_proxy_production_amount(activity_id: int) → float

Production amount of a (possibly non-referenced) variant node, read from its bw2data proxy.

_proxy_technosphere_inputs(activity_id: int) → list[int]

Technosphere input product ids of a variant node, read from its proxy.

static_activity_indices contains MATRIX INDICES (as used by the priority TemporalisLCA engine), not activity ids. When traverse_background=True, TimelineBuilder forces static_activity_indices = set() so this filter is always a no-op on the variant-descent path — filtering here by activity id would be incorrect whenever the set is non-empty.

_get_exchange_td_and_type_from_proxy(input_id: int, output_id: int)

Read the exchange between input_id and output_id directly from the consuming node’s bw2data proxy (self.bw_node_proxies[output_id]) rather than from the technosphere matrix, because non-referenced variant nodes are absent from base_lca’s matrix. Returns the same (td_or_amount, edge_type, temporal_evolution) tuple.

_normalized_production_edge_td_from_proxy(process_id: int)

Cohort production-edge TD of a variant node, normalized to unit weights, read from its proxy. None when there is no such TD.

_producer_process_in_variant(product_id: int, db_name: str)

Resolve the process producing product_id within variant db_name from the proxy.

product_id was read from a variant proxy, so it already lives in db_name. For chimaera nodes the product produces itself; returns None if the product has no producer (a pure leaf).

_is_static_background(node_id: int) → bool

True if node_id lives in one of the static background databases.

_emit_variant_split(*, node_id: int, producer_process: int, edge_type: str, temporal_evolution, td_producer: bw_temporalis.TemporalDistribution, distribution: bw_temporalis.TemporalDistribution, abs_td_producer: bw_temporalis.TemporalDistribution, abs_td: bw_temporalis.TemporalDistribution, td_parent, new_supply: float, total_demand: float) → list

Perform the variant-aware split at the first background crossing.

For each producer cohort date, route the edge to its temporally appropriate variant database(s), emit one scaled Edge per variant, record the variant id in self.variant_resolved_producers, and descend the resulting variant subtree via proxy reads. Returns the list of Edge instances (the split edge + every edge of every variant subtree).

Shared verbatim by both the BFS and priority engines.

A multi-date consumer (e.g. a foreground temporal_distribution feeding a non-leaf background activity, so the consuming background process is reached at several cohort dates) is handled by splitting it into one single-consumer-date routing per consumer cohort. The join that builds abs_td_producer is consumer-major, so consumer cohort i is the contiguous block i*M : (i+1)*M; each block then reduces to the single-consumer-date case, where the relative td_producer and the absolute abs_td_producer share one index axis.

_emit_variant_split_for_consumer_date(*, node_id: int, producer_process: int, edge_type: str, temporal_evolution, td_producer: bw_temporalis.TemporalDistribution, distribution: bw_temporalis.TemporalDistribution, abs_td_producer: bw_temporalis.TemporalDistribution, abs_td: bw_temporalis.TemporalDistribution, td_parent, new_supply: float, total_demand: float) → list

Variant split for a single consumer cohort date (len(abs_td) == 1).

With one consumer date the relative td_producer and the absolute abs_td_producer share one index axis, so all three arrays mask by the same kept indices and extract_edge_data can explode the consumer/producer dates and amounts consistently.

_descend_variant_subtree(*, node_id: int, td: bw_temporalis.TemporalDistribution, td_parent, abs_td: bw_temporalis.TemporalDistribution, supply: float, variant_db: str, total_demand: float) → list

Proxy-only BFS descent through a locked background variant subtree.

Once inside a variant descent the variant is LOCKED: every descendant stays in variant_db and is read from its bw2data proxy (those nodes are absent from base_lca). No re-splitting happens. Variant-resolved background producers are recorded so the timeline builder temporalizes them (rather than re-interpolating them as temporal markets).

Engine-agnostic: it does not touch the priority heap or the BFS matrix, so both extractors call it identically after their first-crossing split.

Bounded by cutoff (per-edge supply threshold) and max_calc (total descended-node budget, shared via self._calc_count); the latter caps how deep the background descent runs.

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

Bases: VariantBackgroundMixin, 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.

• traverse_background (bool, optional) – Flag indicating whether to traverse background databases. Default is False.

• **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, nodes: dict | None = None, traverse_background: bool = False, max_calc: int = 1000000)

Bases: VariantBackgroundMixin

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.

_get_production_exchange(process_id: int)

Return the single production output exchange of process_id, or None if it has none. Read from the reused node proxy (no extra node fetch) and memoized per process.

Identifying the output via the exchange type (rather than the matrix sign) matters because a negative-amount technosphere input also produces a positive matrix entry, so sign alone cannot tell them apart.

_get_technosphere_inputs(activity_id: int) → list[int]

Get the input product IDs consumed by a process.

Read straight from the node’s technosphere exchanges, so the production output is naturally excluded and avoided-burden inputs (which the matrix stores with a flipped sign) are kept.

_get_producer_process(product_id: int) → int | None

Return the process that produces product_id.

For a chimaera node this is the node itself; for an explicit product it is the separate process whose production edge feeds the product’s row. Returns None if the product has no producer (a leaf).

_get_normalized_production_edge_td(process_id: int)

Return the cohort TemporalDistribution on a process’s production output edge, normalized to unit weights, or None if there is none.

Chimaera self-production edges normally carry no TD, so this returns None and chimaera traversal is unaffected.

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.