Support explicit product-modelling and add examples

bw_timex/dynamic_biosphere_builder.py

@@ -194,8 +194,12 @@ def build_dynamic_biosphere_matrix(
                 # Get temporal evolution factor for this timestamp
                 temporal_evolution_factor = 1.0
                 if hasattr(row, "temporal_evolution") and row.temporal_evolution is not None:
+                    reference = getattr(row, "temporal_evolution_reference", "producer")
+                    reference_time = (
+                        row.date_consumer if reference == "consumer" else time_in_datetime
+                    )
                     temporal_evolution_factor = get_temporal_evolution_factor(
-                        row.temporal_evolution, time_in_datetime
+                        row.temporal_evolution, reference_time
                     )
 
                 for input_id, exc_amount, temporal_distribution in (

bw_timex/edge_extractor.py

@@ -11,6 +11,11 @@
 from bw2data.backends.schema import ExchangeDataset as ED
 from bw_temporalis import TemporalDistribution, TemporalisLCA, loader_registry
 
+from .utils import get_reference_product_production_amount
+
+if not hasattr(np, "in1d"):
+    np.in1d = np.isin
+
 datetime_type = np.dtype("datetime64[s]")
 timedelta_type = np.dtype("timedelta64[s]")
 
@@ -36,6 +41,7 @@ class Edge:
     abs_td_producer: TemporalDistribution = None
     abs_td_consumer: TemporalDistribution = None
     temporal_evolution: dict = None
+    temporal_evolution_reference: str = "producer"
 
 
 class EdgeExtractor(TemporalisLCA):
@@ -101,27 +107,71 @@ def build_edge_timeline(self) -> list:
             self.unique_id
         ]:  # starting at the edges of the functional unit
             node = self.nodes[edge.producer_unique_id]
+            td_producer = edge.amount
+            initial_distribution = self.t0 * edge.amount
+            abs_td_producer = self.t0
+            abs_td_consumer = None
+
+            row_id = self.lca_object.dicts.product.reversed[edge.product_index]
+            col_id = node.activity_datapackage_id
+            exchange = self.get_technosphere_exchange(input_id=row_id, output_id=col_id)
+
+            # In the explicit process/product paradigm, the demanded product is produced
+            # by an off-diagonal production edge. A TD on that edge distributes the
+            # process invocation itself before the process inputs are traversed.
+            if (
+                row_id != col_id
+                and hasattr(exchange, "data")
+                and exchange.data.get("type") == "production"
+            ):
+                production_amount = abs(
+                    get_reference_product_production_amount(
+                        col_id, reference_product=row_id, lca=self.lca_object
+                    )
+                )
+                td_producer = (
+                    self._exchange_value(
+                        exchange=exchange,
+                        row_id=row_id,
+                        col_id=col_id,
+                        matrix_label="technosphere_matrix",
+                    )
+                    / production_amount
+                    * edge.amount
+                )
+                if isinstance(td_producer, Number):
+                    td_producer = TemporalDistribution(
+                        date=np.array([0], dtype="timedelta64[Y]"),
+                        amount=np.array([td_producer]),
+                    )
+                initial_distribution = (self.t0 * td_producer).simplify()
+                abs_td_producer = self.join_datetime_and_timedelta_distributions(
+                    td_producer, self.t0
+                )
+                abs_td_consumer = self.t0
+
             heappush(
                 heap,
                 (
                     1 / node.cumulative_score,
-                    self.t0 * edge.amount,
-                    self.t0,
+                    initial_distribution,
                     self.t0,
+                    abs_td_producer,
                     node,
                 ),
             )
 
             timeline.append(
                 Edge(
                     edge_type="production",  # FU exchange always type production (?)
-                    distribution=self.t0 * edge.amount,
+                    distribution=initial_distribution,
                     leaf=False,
                     consumer=self.unique_id,
                     producer=node.activity_datapackage_id,
-                    td_producer=edge.amount,
+                    td_producer=td_producer,
                     td_consumer=self.t0,
-                    abs_td_producer=self.t0,
+                    abs_td_producer=abs_td_producer,
+                    abs_td_consumer=abs_td_consumer,
                 )
             )
 
@@ -131,10 +181,17 @@ def build_edge_timeline(self) -> list:
             for edge in self.edge_mapping[node.unique_id]:
                 row_id = self.nodes[edge.producer_unique_id].activity_datapackage_id
                 col_id = node.activity_datapackage_id
+                product_id = self.lca_object.dicts.product.reversed[edge.product_index]
                 exchange = self.get_technosphere_exchange(
-                    input_id=row_id,
+                    input_id=product_id,
                     output_id=col_id,
                 )
+                if not hasattr(exchange, "data"):
+                    exchange = self.get_technosphere_exchange(
+                        input_id=row_id,
+                        output_id=col_id,
+                    )
+                    product_id = row_id
 
                 edge_type = exchange.data[
                     "type"
@@ -163,15 +220,22 @@ def build_edge_timeline(self) -> list:
                         }
                 elif has_factors:
                     temporal_evolution = exc_data["temporal_evolution_factors"]
+                temporal_evolution_reference = exc_data.get(
+                    "temporal_evolution_reference", "producer"
+                )
 
                 td_producer = (  # td_producer is the TemporalDistribution of the edge
                     self._exchange_value(
                         exchange=exchange,
-                        row_id=row_id,
+                        row_id=product_id,
                         col_id=col_id,
                         matrix_label="technosphere_matrix",
                     )
-                    / abs(node.reference_product_production_amount)
+                    / abs(
+                        get_reference_product_production_amount(
+                            node.activity_datapackage_id, lca=self.lca_object
+                        )
+                    )
                 )
                 producer = self.nodes[edge.producer_unique_id]
                 leaf = self.edge_ff(row_id)
@@ -183,6 +247,38 @@ def build_edge_timeline(self) -> list:
                         amount=np.array([td_producer]),
                     )
 
+                if product_id != row_id:
+                    production_exchange = self.get_technosphere_exchange(
+                        input_id=product_id,
+                        output_id=row_id,
+                    )
+                    if (
+                        hasattr(production_exchange, "data")
+                        and production_exchange.data.get("type") == "production"
+                    ):
+                        production_amount = abs(
+                            get_reference_product_production_amount(
+                                row_id,
+                                reference_product=product_id,
+                                lca=self.lca_object,
+                            )
+                        )
+                        production_td = (
+                            self._exchange_value(
+                                exchange=production_exchange,
+                                row_id=product_id,
+                                col_id=row_id,
+                                matrix_label="technosphere_matrix",
+                            )
+                            / production_amount
+                        )
+                        if isinstance(production_td, Number):
+                            production_td = TemporalDistribution(
+                                date=np.array([0], dtype="timedelta64[Y]"),
+                                amount=np.array([production_td]),
+                            )
+                        td_producer = (td_producer * production_td).simplify()
+
                 distribution = (
                     td * td_producer
                 ).simplify()  # convolution-multiplication of TemporalDistribution of consuming node (td) and consumed edge (edge) gives TD of producing node
@@ -201,6 +297,7 @@ def build_edge_timeline(self) -> list:
                         ),
                         abs_td_consumer=abs_td,
                         temporal_evolution=temporal_evolution,
+                        temporal_evolution_reference=temporal_evolution_reference,
                     )
                 )
                 if not leaf:
@@ -354,10 +451,11 @@ def _get_exchange_td_and_type(self, input_id: int, output_id: int):
         return amount, edge_type
 
     def _get_production_amount(self, activity_id: int) -> float:
-        """Get the reference product production amount (diagonal of tech matrix)."""
-        product_idx = self.lca_object.dicts.product[activity_id]
-        col_idx = self.lca_object.dicts.activity[activity_id]
-        return self.tech_matrix_csc[product_idx, col_idx]
+        """Get the reference product production amount."""
+        activity = self.lca_object.dicts.activity.reversed[
+            self.lca_object.dicts.activity[activity_id]
+        ]
+        return get_reference_product_production_amount(activity)
 
     def _get_technosphere_inputs(self, activity_id: int) -> list[int]:
         """Get all technosphere input activity IDs for a given activity."""

bw_timex/matrix_modifier.py

@@ -7,7 +7,10 @@
 import pandas as pd
 
 from .helper_classes import InterDatabaseMapping
-from .utils import get_temporal_evolution_factor
+from .utils import (
+    get_reference_product_production_amount,
+    get_temporal_evolution_factor,
+)
 
 
 class MatrixModifier:
@@ -220,7 +223,9 @@ def add_row_to_technosphere_datapackage(
 
         if row.consumer == -1:  # functional unit
             new_producer_id = row.time_mapped_producer
-            fu_production_amount = self.nodes[row.producer].rp_exchange().amount
+            fu_production_amount = get_reference_product_production_amount(
+                self.nodes[row.producer]
+            )
             new_nodes.add((new_producer_id, fu_production_amount))
             self.temporalized_process_ids.add(
                 new_producer_id
@@ -233,15 +238,21 @@ def add_row_to_technosphere_datapackage(
         previous_producer_id = row.producer
         previous_producer_node = self.nodes[previous_producer_id]
 
-        production_exchange_amount = self.nodes[row.consumer].rp_exchange().amount
+        production_exchange_amount = get_reference_product_production_amount(
+            self.nodes[row.consumer], reference_product=row.consumer
+        )
         scaled_amount = row.amount * abs(
             production_exchange_amount
         )  # abs value used for scaling to preserve the sign of the exchange
 
         # Apply temporal evolution scaling if present
         if hasattr(row, "temporal_evolution") and row.temporal_evolution is not None:
+            reference = getattr(row, "temporal_evolution_reference", "producer")
+            reference_date = (
+                row.date_consumer if reference == "consumer" else row.date_producer
+            )
             factor = get_temporal_evolution_factor(
-                row.temporal_evolution, row.date_producer
+                row.temporal_evolution, reference_date
             )
             scaled_amount *= factor
 
@@ -316,8 +327,8 @@ def add_row_to_technosphere_datapackage(
                         "The producer activity is of type IOTableActivity, but has more than one production exchange. This is currently not supported."
                     )
             elif isinstance(previous_producer_node, bd.backends.proxies.Activity):
-                producer_production_amount = (
-                    self.nodes[row.producer].rp_exchange().amount
+                producer_production_amount = get_reference_product_production_amount(
+                    self.nodes[row.producer], reference_product=row.producer
                 )
             else:
                 raise ValueError(

bw_timex/timeline_builder.py

@@ -222,6 +222,7 @@ def build_timeline(self) -> pd.DataFrame:
                     "producer",
                     "consumer",
                     "_te_key",
+                    "temporal_evolution_reference",
                 ],
                 dropna=False,
             )
@@ -314,6 +315,7 @@ def build_timeline(self) -> pd.DataFrame:
                 "amount",
                 "temporal_market_shares",
                 "temporal_evolution",
+                "temporal_evolution_reference",
             ]
         ]
 
@@ -363,6 +365,7 @@ def extract_edge_data(self, edge: Edge) -> dict:
             "amount": edge.abs_td_producer.amount,
             "edge_type": edge.edge_type,
             "temporal_evolution": edge.temporal_evolution,
+            "temporal_evolution_reference": edge.temporal_evolution_reference,
         }
 
     def adjust_sign_of_amount_based_on_edge_type(self, edge_type):