Add regression tests for regional biosphere resolution

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"]),

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
 

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"
+    )