From acf39772ccfde002872eb596ace2776f503fbe13 Mon Sep 17 00:00:00 2001 From: Romain Sacchi Date: Fri, 7 Nov 2025 17:43:29 +0100 Subject: [PATCH 1/2] Add tests for regional biosphere name resolution --- brightpath/bwconverter.py | 44 +++++++++++++-- brightpath/utils.py | 41 +++++++++++--- tests/test_biosphere_resolution.py | 89 ++++++++++++++++++++++++++++++ 3 files changed, 163 insertions(+), 11 deletions(-) create mode 100644 tests/test_biosphere_resolution.py diff --git a/brightpath/bwconverter.py b/brightpath/bwconverter.py index 1a6d43b..9369919 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,34 @@ 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 +462,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 +507,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..d8c815a 100644 --- a/brightpath/utils.py +++ b/brightpath/utils.py @@ -2,8 +2,9 @@ import json import logging import re +from collections import Counter from pathlib import Path -from typing import Dict, Tuple +from typing import Dict, Optional as TypingOptional, Tuple, Union import bw2io import numpy as np @@ -23,12 +24,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 +46,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..8a3363d --- /dev/null +++ b/tests/test_biosphere_resolution.py @@ -0,0 +1,89 @@ +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" + ) From 37079cc051cbfe6321813f044b170bd34d114d55 Mon Sep 17 00:00:00 2001 From: romainsacchi <22473067+romainsacchi@users.noreply.github.com> Date: Fri, 7 Nov 2025 16:43:46 +0000 Subject: [PATCH 2/2] style: auto-format with black + isort --- brightpath/bwconverter.py | 4 +++- brightpath/utils.py | 4 +++- tests/test_biosphere_resolution.py | 16 ++++------------ 3 files changed, 10 insertions(+), 14 deletions(-) diff --git a/brightpath/bwconverter.py b/brightpath/bwconverter.py index 9369919..90c0cbd 100644 --- a/brightpath/bwconverter.py +++ b/brightpath/bwconverter.py @@ -148,7 +148,9 @@ def _resolve_biosphere_flow_name( if exchange_location: candidates.append(exchange_location) if "-" in exchange_location: - candidates.extend([part for part in exchange_location.split("-") if part]) + 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]) diff --git a/brightpath/utils.py b/brightpath/utils.py index d8c815a..c041fa0 100644 --- a/brightpath/utils.py +++ b/brightpath/utils.py @@ -4,7 +4,9 @@ import re from collections import Counter from pathlib import Path -from typing import Dict, Optional as TypingOptional, Tuple, Union +from typing import Dict +from typing import Optional as TypingOptional +from typing import Tuple, Union import bw2io import numpy as np diff --git a/tests/test_biosphere_resolution.py b/tests/test_biosphere_resolution.py index 8a3363d..c268315 100644 --- a/tests/test_biosphere_resolution.py +++ b/tests/test_biosphere_resolution.py @@ -32,9 +32,7 @@ def test_resolve_uses_direct_string_mapping(): def test_resolve_prefers_exact_exchange_location(): converter = make_converter() - converter.simapro_biosphere = { - "Water": {"CH": "Water, CH", "GLO": "Water, GLO"} - } + converter.simapro_biosphere = {"Water": {"CH": "Water, CH", "GLO": "Water, GLO"}} exchange = {"name": "Water", "location": "CH"} assert ( @@ -45,9 +43,7 @@ def test_resolve_prefers_exact_exchange_location(): def test_resolve_supports_hierarchical_locations(): converter = make_converter() - converter.simapro_biosphere = { - "Water": {"CH": "Water, CH", "GLO": "Water, GLO"} - } + converter.simapro_biosphere = {"Water": {"CH": "Water, CH", "GLO": "Water, GLO"}} exchange = {"name": "Water", "location": "CH-01"} assert ( @@ -58,9 +54,7 @@ def test_resolve_supports_hierarchical_locations(): def test_resolve_falls_back_to_activity_location(): converter = make_converter() - converter.simapro_biosphere = { - "Water": {"BR": "Water, BR", "GLO": "Water, GLO"} - } + converter.simapro_biosphere = {"Water": {"BR": "Water, BR", "GLO": "Water, GLO"}} exchange = {"name": "Water"} assert ( @@ -71,9 +65,7 @@ def test_resolve_falls_back_to_activity_location(): def test_resolve_defaults_to_global_and_none(): converter = make_converter() - converter.simapro_biosphere = { - "Water": {"GLO": "Water, GLO", None: "Water"} - } + converter.simapro_biosphere = {"Water": {"GLO": "Water, GLO", None: "Water"}} exchange = {"name": "Water"} assert (