Initial version of new SimaPro importer

Author: cmutel

bw2io/__init__.py

@@ -38,8 +38,9 @@
     "MultiOutputEcospold1Importer",
     "normalize_units",
     "restore_project_directory",
-    "SimaProCSVImporter",
-    "SimaProLCIACSVImporter",
+    # "SimaProCSVImporter",
+    # "SimaProLCIACSVImporter",
+    "SimaProBlockCSVImporter",
     "SingleOutputEcospold1Importer",
     "SingleOutputEcospold2Importer",
     "unlinked_data",
@@ -82,13 +83,14 @@
     Ecospold1LCIAImporter,
     ExcelImporter,
     ExcelLCIAImporter,
+    Exiobase3HybridImporter,
+    Exiobase3MonetaryImporter,
     MultiOutputEcospold1Importer,
-    SimaProCSVImporter,
-    SimaProLCIACSVImporter,
+    SimaProBlockCSVImporter,
+    # SimaProCSVImporter,
+    # SimaProLCIACSVImporter,
     SingleOutputEcospold1Importer,
     SingleOutputEcospold2Importer,
-    Exiobase3MonetaryImporter,
-    Exiobase3HybridImporter,
 )
 from .units import normalize_units
 from .unlinked_data import unlinked_data, UnlinkedData

bw2io/ecoinvent.py

@@ -2,7 +2,7 @@
 import zipfile
 from collections import defaultdict
 from pathlib import Path
-from typing import Any
+from typing import Any, Optional
 
 import bw2data as bd
 import ecoinvent_interface as ei
@@ -52,13 +52,14 @@ def pick_a_unit_label_already(obj: dict) -> str:
 def import_ecoinvent_release(
     version: str,
     system_model: str,
-    username: str | None = None,
-    password: str | None = None,
+    username: Optional[str] = None,
+    password: Optional[str] = None,
     lci: bool = True,
     lcia: bool = True,
-    biosphere_name: str | None = None,
+    biosphere_name: Optional[str] = None,
     biosphere_write_mode: str = "patch",
     importer_signal: Any = None,
+    lcia_prefix: Optional[str] = None,
 ) -> None:
     """
     Import an ecoinvent LCI and/or LCIA release.
@@ -109,6 +110,9 @@ def import_ecoinvent_release(
         How to handle an existing biosphere database. Must be either `replace` or `patch`
     importer_signal
         Used by the Activity Browser to provide feedback during the import
+    lcia_prefix
+        Extra element to add before impact category definitions, e.g. "foo" ->
+        `("foo", "global warming")`. Useful when multiple ecoinvent versions are being installed.
 
     Examples
     --------
@@ -308,7 +312,10 @@ def import_ecoinvent_release(
         substituted = set()
 
         for row in cfs:
-            impact_category = (row["method"], row["category"], row["indicator"])
+            if lcia_prefix:
+                impact_category = (lcia_prefix, row["method"], row["category"], row["indicator"])
+            else:
+                impact_category = (row["method"], row["category"], row["indicator"])
             if row[cf_col_label] is None:
                 continue
             try:

bw2io/export/excel.py

@@ -327,24 +327,24 @@ def write_headers(sheet, row):
     def write_row(sheet, row, data, exc=True):
         style = highlighted if ("input" not in data and exc) else None
         if exc:
-            sheet.write_string(row, 0, data.get("name", "(unknown)"), style)
+            sheet.write_string(row, 0, data.get("name") or "(unknown)", style)
             sheet.write_string(
-                row, 1, data.get("reference product", "(unknown)"), style
+                row, 1, data.get("reference product") or "(unknown)", style
             )
             try:
                 sheet.write_number(row, 2, float(data.get("amount")), style)
             except ValueError:
                 sheet.write_string(row, 2, "Unknown", style)
         else:
-            sheet.write_string(row, 0, data.get("name", "(unknown)"), bold)
-        sheet.write_string(row, 3, data.get("input", [""])[0], style)
-        sheet.write_string(row, 4, data.get("unit", "(unknown)"), style)
+            sheet.write_string(row, 0, data.get("name") or "(unknown)", bold)
+        sheet.write_string(row, 3, (data.get("input") or [""])[0], style)
+        sheet.write_string(row, 4, data.get("unit") or "(unknown)", style)
         sheet.write_string(
-            row, 5, ":".join(data.get("categories", ["(unknown)"])), style
+            row, 5, ":".join(data.get("categories") or ["(unknown)"]), style
         )
-        sheet.write_string(row, 6, data.get("location", "(unknown)"), style)
+        sheet.write_string(row, 6, data.get("location") or "(unknown)", style)
         if exc:
-            sheet.write_string(row, 7, data.get("type", "(unknown)"), style)
+            sheet.write_string(row, 7, data.get("type") or "(unknown)", style)
             sheet.write_boolean(row, 8, "input" in data, style)
 
     if only_unlinked and only_activity_names:
@@ -438,10 +438,10 @@ def write_headers(sheet, row):
             sheet.write_string(row, index, col, bold)
 
     def write_row(sheet, row, data):
-        sheet.write_string(row, 0, data.get("name", "(unknown)"))
+        sheet.write_string(row, 0, data.get("name") or "(unknown)")
         sheet.write_number(row, 1, data.get("amount", -1))
-        sheet.write_string(row, 2, data.get("unit", "(unknown)"))
-        sheet.write_string(row, 3, ":".join(data.get("categories", ["(unknown)"])))
+        sheet.write_string(row, 2, data.get("unit") or "(unknown)")
+        sheet.write_string(row, 3, ":".join(data.get("categories") or ["(unknown)"]))
         sheet.write_boolean(row, 4, "input" in data)
 
     safe_name = safe_filename(name, False)

bw2io/importers/__init__.py

@@ -11,8 +11,9 @@
 from .excel_lcia import CSVLCIAImporter, ExcelLCIAImporter
 from .exiobase3_hybrid import Exiobase3HybridImporter
 from .exiobase3_monetary import Exiobase3MonetaryImporter
-from .simapro_csv import SimaProCSVImporter
-from .simapro_lcia_csv import SimaProLCIACSVImporter
+# from .simapro_csv import SimaProCSVImporter
+# from .simapro_lcia_csv import SimaProLCIACSVImporter
+from .simapro_block_csv import SimaProBlockCSVImporter
 
 """
 This module provides classes for importing Life Cycle Impact Assessment (LCIA) data

bw2io/importers/base_lci.py

@@ -4,7 +4,7 @@
 import warnings
 from typing import Optional
 
-from bw2data import Database, config, databases, parameters
+from bw2data import Database, config, databases, parameters, labels
 from bw2data.parameters import (
     ActivityParameter,
     DatabaseParameter,
@@ -340,55 +340,44 @@ def match_database(
 
         self.apply_strategy(functools.partial(link_iterable_by_fields, **kwargs))
 
-    def create_new_biosphere(self, biosphere_name, relink=True):
-        """Create new biosphere database from biosphere flows in ``self.data``.
+    def create_new_biosphere(self, biosphere_name: str):
+        """Create new biosphere database from unlinked biosphere flows in ``self.data``"""
+        if biosphere_name in databases:
+            raise ValueError(f"{biosphere_name} database already exists")
 
-        Links all biosphere flows to new bio database if ``relink``."""
-        assert biosphere_name not in databases, "{} database already exists".format(
-            biosphere_name
-        )
+        def reformat(exc):
+            return exc | {"type": labels.biosphere_node_default, "exchanges": [], "database": biosphere_name, "code": activity_hash(exc)}
+
+        bio_data = {(flow["database"], flow["code"]): flow for flow in [
+            reformat(exc)
+            for ds in self.data
+            for exc in ds.get("exchanges", [])
+            if exc["type"] in labels.biosphere_edge_types
+            and not exc.get("input")
+        ]}
 
-        print("Creating new biosphere database: {}".format(biosphere_name))
+        if not bio_data:
+            print("Skipping biosphere database creation as all biosphere flows are linked")
+            return
+
+        print(f"Creating new biosphere database {biosphere_name} with {len(bio_data)} flows")
 
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
             new_bio = Database(biosphere_name)
             new_bio.register(
-                format=self.format, comment="New biosphere created by LCI import"
-            )
-
-        KEYS = {"name", "unit", "categories"}
-
-        def reformat(exc):
-            dct = {key: value for key, value in list(exc.items()) if key in KEYS}
-            dct.update(
-                type="emission",
-                exchanges=[],
-                database=biosphere_name,
-                code=activity_hash(dct),
+                format=self.format, comment=f"Database for unlinked biosphere flows from {self.db_name}"
             )
-            return dct
 
-        bio_data = [
-            reformat(exc)
-            for ds in self.data
-            for exc in ds.get("exchanges", [])
-            if exc["type"] == "biosphere"
-        ]
-
-        bio_data = {(ds["database"], ds["code"]): ds for ds in bio_data}
         new_bio.write(bio_data)
-
-        if relink:
-            self.apply_strategies(
-                [
-                    functools.partial(
-                        link_iterable_by_fields,
-                        other=list(bio_data.values()),
-                        relink=True,
-                    ),
-                ]
-            )
+        self.apply_strategies(
+            [
+                functools.partial(
+                    link_iterable_by_fields,
+                    other=list(bio_data.values()),
+                ),
+            ]
+        )
 
     def add_unlinked_flows_to_biosphere_database(
         self, biosphere_name=None, fields={"name", "unit", "categories"}