LCA Processor¶

Bases: str, Enum

Supported metrics for dynamic impact characterization.

Attributes: GWP: Global Warming Potential - time-dependent radiative forcing metric CRF: Cumulative Radiative Forcing - integrated radiative forcing over time horizon

Bases: str, Enum

Supported temporal resolutions for the optimization model.

Attributes: year: Annual time steps (currently the only supported resolution)

Bases: BaseModel

Configuration for a single LCIA characterization method.

Attributes: category_name: User-defined identifier for the impact category (e.g., 'climate_change_dynamic_gwp'). brightway_method: Brightway method identifier tuple, either 2 or 3 elements (e.g., ('GWP', 'example') or ('IPCC', 'climate change', 'GWP 100a')). metric: Impact metric used for dynamic characterization. None implies static method. Supported values: 'GWP', 'CRF'.

Bases: BaseModel

Configuration related to temporal aspects of the model.

Attributes: start_date: The start date of the time horizon. temporal_resolution: Temporal resolution for the model. Options: 'year', 'month', 'day'. time_horizon: Length of the time horizon (in units of temporal_resolution). fixed_time_horizon: If True, the time horizon is calculated from the time of the functional unit (FU) instead of the time of emission database_dates: Mapping from database names to their respective reference dates.

Bases: BaseModel

Configuration for background inventory data.

Attributes: cutoff: Cutoff threshold for the number of top elementary flows to retain based on impact magnitude. calculation_method: Method for calculating the inventory tensor. Options: 'sequential', 'parallel'. path_to_save: Optional path to save the inventory tensor. path_to_load: Optional path to load the inventory tensor.

Bases: BaseModel

Configuration class for Life Cycle Assessment (LCA) data processing.

Attributes: demand: Dictionary {product_node: temporal_distribution} containing time-explicit demands for each product. Keys must be Brightway product node objects (bd.get_node(...)). temporal: Temporal configuration for model time behavior. characterization_methods: List of characterization method configurations. background_inventory: Configuration for background inventory data calculation. foreground_db_name: Name of the foreground Brightway database.

Class to perform time-explicit Life Cycle Assessment (LCA) computations and gather necessary data for building an optimization model.

This class is primarily responsible for executing the LCA-based computations required to collect all the data needed for building OptimizationModelInputs. It is reliant on Brightway2, an open-source framework for Life Cycle Assessment, to perform the calculations and retrieve LCA results.

Initialize the LCADataProcessor with the LCA configuration.

Parameters:

198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265

def __init__(
    self, config: LCAConfig, foreground_db_name: str = "foreground"
) -> None:
    """
    Initialize the LCADataProcessor with the LCA configuration.

    Parameters
    ----------
    config : LCAConfig
        The configuration object containing all settings for demand,
        temporal parameters, characterization methods, and background inventory.
    foreground_db_name : str, optional
        The name of the foreground Brightway database, by default "foreground".
    """
    self.config = config
    if foreground_db_name not in bd.databases:
        raise ValueError(
            f"Foreground database '{foreground_db_name}' is not defined."
        )
    self.foreground_db = bd.Database(foreground_db_name)
    self.background_dbs = {}
    if config.temporal.database_dates is not None:
        self.background_dbs = {
            db: date
            for db, date in config.temporal.database_dates.items()
            if db != self.foreground_db.name
        }
    else:
        for db_name in bd.databases:
            db = bd.Database(db_name)
            if (date := db.metadata.get("representative_time")) is not None:
                self.background_dbs[db.name] = datetime.fromisoformat(date)

    self.biosphere_db = bd.Database(bd.config.biosphere)

    self._demand = {}
    self._processes = {}
    self._products = {}  # Maps product codes to product names
    self._intermediate_flows = {}
    self._elementary_flows = {}

    self._reference_products = set()
    self._system_time = set()
    self._process_time = set()
    self._category = set()

    self._foreground_technosphere = {}
    self._internal_demand_technosphere = {}  # (process, product, year) -> amount
    self._foreground_biosphere = {}
    self._foreground_production = {}
    self._background_inventory = {}
    self._mapping = {}
    self._characterization = {}
    self._operation_flow = {}
    self._operation_time_limits = {}

    # Vintage-dependent parameters extracted from exchange attributes
    self._foreground_technosphere_vintages = {}
    self._foreground_biosphere_vintages = {}
    self._foreground_production_vintages = {}
    self._vintage_improvements = {}
    self._reference_vintages = set()

    self._parse_demand()
    self._construct_foreground_tensors()
    self._prepare_background_inventory()
    self._construct_characterization_tensor()
    self._construct_mapping_matrix()

778
779
780
781
782

def parallel_inventory_tensor_calculation(self, cutoff=1e4, n_jobs=None) -> dict:
    """
    Not yet implemented. Could improve performance significantly by parallelizing
    """
    raise NotImplementedError("This method is not yet functionally implemented.")