diff --git a/brightpath/bwconverter.py b/brightpath/bwconverter.py index 1a6d43b..90c0cbd 100644 --- a/brightpath/bwconverter.py +++ b/brightpath/bwconverter.py @@ -71,8 +71,8 @@ class BrightwayConverter: SimaPro names. :vartype simapro_technosphere: dict[tuple[str, str], str] :ivar simapro_biosphere: Mapping from biosphere exchanges to SimaPro - names. - :vartype simapro_biosphere: dict[str, str] + names, optionally regionalised by location. + :vartype simapro_biosphere: dict[str, str | dict[str | None, str]] :ivar simapro_subcompartment: Mapping of biosphere subcompartments to SimaPro names. :vartype simapro_subcompartment: dict[str, str] @@ -131,6 +131,36 @@ def __init__( # directory unless specified otherwise self.export_dir = Path(export_dir) or Path.cwd() + def _resolve_biosphere_flow_name( + self, exchange: dict, activity_location: str | None + ) -> str: + """Return the SimaPro name for a biosphere exchange, considering geography.""" + + mapping = self.simapro_biosphere.get(exchange["name"]) + if mapping is None: + return exchange["name"] + + if isinstance(mapping, str): + return mapping + + exchange_location = exchange.get("location") or activity_location + candidates = [] + if exchange_location: + candidates.append(exchange_location) + if "-" in exchange_location: + candidates.extend( + [part for part in exchange_location.split("-") if part] + ) + if activity_location and activity_location not in candidates: + candidates.append(activity_location) + candidates.extend(["GLO", "RoW", "RER", "WEU", None]) + + for candidate in candidates: + if candidate in mapping: + return mapping[candidate] + + return next(iter(mapping.values())) + def format_inventories_for_simapro(self, database: str): """Transform the Brightway inventories into the SimaPro structure. @@ -434,9 +464,13 @@ def format_inventories_for_simapro(self, database: str): else: sub_compartment = "" + biosphere_name = self._resolve_biosphere_flow_name( + exc, activity.get("location") + ) + rows.append( [ - f"{self.simapro_biosphere.get(exc['name'], exc['name'])}", + biosphere_name, sub_compartment, self.simapro_units[exc["unit"]], "{:.3E}".format(exc["amount"]), @@ -475,9 +509,13 @@ def format_inventories_for_simapro(self, database: str): else: sub_compartment = "" + biosphere_name = self._resolve_biosphere_flow_name( + exc, activity.get("location") + ) + rows.append( [ - f"{self.simapro_biosphere.get(exc['name'], exc['name'])}", + biosphere_name, sub_compartment, self.simapro_units[exc["unit"]], "{:.3E}".format(exc["amount"]), diff --git a/brightpath/utils.py b/brightpath/utils.py index b3391e2..c041fa0 100644 --- a/brightpath/utils.py +++ b/brightpath/utils.py @@ -2,8 +2,11 @@ import json import logging import re +from collections import Counter from pathlib import Path -from typing import Dict, Tuple +from typing import Dict +from typing import Optional as TypingOptional +from typing import Tuple, Union import bw2io import numpy as np @@ -23,12 +26,14 @@ ) -def get_simapro_biosphere() -> Dict[str, str]: +def get_simapro_biosphere() -> Dict[str, Union[str, Dict[TypingOptional[str], str]]]: """Load the correspondence between ecoinvent and SimaPro biosphere flows. :return: Mapping from an ecoinvent biosphere flow name to its SimaPro - equivalent name. - :rtype: dict[str, str] + equivalent names. The mapping contains either a single string when the + flow is not regionalised, or a dictionary keyed by location codes for + flows that require regionalisation. + :rtype: dict[str, str | dict[str | None, str]] :raises FileNotFoundError: If the mapping file is missing from ``brightpath/data/export``. :raises json.JSONDecodeError: If the mapping file cannot be parsed. @@ -43,9 +48,33 @@ def get_simapro_biosphere() -> Dict[str, str]: ) with open(filepath, encoding="utf-8") as json_file: data = json.load(json_file) - dict_bio = {} - for d in data: - dict_bio[d[2]] = d[1] + + def _extract_location(simapro_name: str) -> TypingOptional[str]: + """Extract the regional suffix from a SimaPro biosphere flow name.""" + + base, separator, suffix = simapro_name.rpartition(",") + if not separator or not base: + return None + return suffix.strip() or None + + counts = Counter(entry[2] for entry in data) + dict_bio: Dict[str, Union[str, Dict[TypingOptional[str], str]]] = {} + + for _, simapro_name, bw_name in data: + if counts[bw_name] == 1: + dict_bio[bw_name] = simapro_name + continue + + location = _extract_location(simapro_name) + biosphere_entry = dict_bio.setdefault(bw_name, {}) + if not isinstance(biosphere_entry, dict): + biosphere_entry = {} + dict_bio[bw_name] = biosphere_entry + + if location is None: + biosphere_entry.setdefault(None, simapro_name) + else: + biosphere_entry[location] = simapro_name return dict_bio diff --git a/tests/test_biosphere_resolution.py b/tests/test_biosphere_resolution.py new file mode 100644 index 0000000..c268315 --- /dev/null +++ b/tests/test_biosphere_resolution.py @@ -0,0 +1,81 @@ +from brightpath.bwconverter import BrightwayConverter + + +def make_converter(): + """Create a converter with controllable biosphere mapping.""" + + converter = BrightwayConverter.__new__(BrightwayConverter) + converter.simapro_biosphere = {} + return converter + + +def test_resolve_returns_original_name_when_missing(): + converter = make_converter() + exchange = {"name": "Water, river", "location": "CH"} + + assert ( + converter._resolve_biosphere_flow_name(exchange, activity_location="GLO") + == "Water, river" + ) + + +def test_resolve_uses_direct_string_mapping(): + converter = make_converter() + converter.simapro_biosphere = {"Water": "Water, resource"} + exchange = {"name": "Water", "location": "CH"} + + assert ( + converter._resolve_biosphere_flow_name(exchange, activity_location=None) + == "Water, resource" + ) + + +def test_resolve_prefers_exact_exchange_location(): + converter = make_converter() + converter.simapro_biosphere = {"Water": {"CH": "Water, CH", "GLO": "Water, GLO"}} + exchange = {"name": "Water", "location": "CH"} + + assert ( + converter._resolve_biosphere_flow_name(exchange, activity_location="DE") + == "Water, CH" + ) + + +def test_resolve_supports_hierarchical_locations(): + converter = make_converter() + converter.simapro_biosphere = {"Water": {"CH": "Water, CH", "GLO": "Water, GLO"}} + exchange = {"name": "Water", "location": "CH-01"} + + assert ( + converter._resolve_biosphere_flow_name(exchange, activity_location=None) + == "Water, CH" + ) + + +def test_resolve_falls_back_to_activity_location(): + converter = make_converter() + converter.simapro_biosphere = {"Water": {"BR": "Water, BR", "GLO": "Water, GLO"}} + exchange = {"name": "Water"} + + assert ( + converter._resolve_biosphere_flow_name(exchange, activity_location="BR") + == "Water, BR" + ) + + +def test_resolve_defaults_to_global_and_none(): + converter = make_converter() + converter.simapro_biosphere = {"Water": {"GLO": "Water, GLO", None: "Water"}} + exchange = {"name": "Water"} + + assert ( + converter._resolve_biosphere_flow_name(exchange, activity_location=None) + == "Water, GLO" + ) + + converter.simapro_biosphere["Water"].pop("GLO") + + assert ( + converter._resolve_biosphere_flow_name(exchange, activity_location=None) + == "Water" + )