Compatibility with simapro multifunctionality

Author: cmutel

bw2io/__init__.py

@@ -50,20 +50,12 @@
 
 __version__ = "0.9.DEV28"
 
-from .chemidplus import ChemIDPlus
-from .package import BW2Package
-from .export import (
-    DatabaseToGEXF,
-    DatabaseSelectionToGEXF,
-    keyword_to_gephi_graph,
-    lci_matrices_to_excel,
-    lci_matrices_to_matlab,
-)
 from .backup import (
     backup_data_directory,
     backup_project_directory,
     restore_project_directory,
 )
+from .chemidplus import ChemIDPlus
 from .data import (
     add_ecoinvent_33_biosphere_flows,
     add_ecoinvent_34_biosphere_flows,
@@ -76,7 +68,13 @@
     get_csv_example_filepath,
     get_xlsx_example_filepath,
 )
-from .migrations import migrations, Migration, create_core_migrations
+from .export import (
+    DatabaseSelectionToGEXF,
+    DatabaseToGEXF,
+    keyword_to_gephi_graph,
+    lci_matrices_to_excel,
+    lci_matrices_to_matlab,
+)
 from .importers import (
     CSVImporter,
     CSVLCIAImporter,
@@ -92,10 +90,12 @@
     SingleOutputEcospold1Importer,
     SingleOutputEcospold2Importer,
 )
+from .migrations import Migration, create_core_migrations, migrations
+from .package import BW2Package
+from .remote import install_project
 from .units import normalize_units
-from .unlinked_data import unlinked_data, UnlinkedData
+from .unlinked_data import UnlinkedData, unlinked_data
 from .utils import activity_hash, es2_activity_hash, load_json_data_file
-from .remote import install_project
 
 try:
     from .ecoinvent import import_ecoinvent_release
@@ -128,9 +128,10 @@ def create_default_lcia_methods(
     overwrite=False, rationalize_method_names=False, shortcut=True
 ):
     if shortcut:
-        import zipfile
         import json
+        import zipfile
         from pathlib import Path
+
         from .importers.base_lcia import LCIAImporter
 
         fp = Path(__file__).parent.resolve() / "data" / "lcia" / "lcia_39_ecoinvent.zip"
@@ -174,13 +175,14 @@ def useeio20(name="USEEIO-2.0", collapse_products=False, prune=False):
         print(f"{name} already present")
         return
 
-    from .importers.json_ld import JSONLDImporter
-    from .importers.json_ld_lcia import JSONLDLCIAImporter
-    from .strategies import remove_useeio_products, remove_random_exchanges
-    from .download_utils import download_with_progressbar
-    from pathlib import Path
     import tempfile
     import zipfile
+    from pathlib import Path
+
+    from .download_utils import download_with_progressbar
+    from .importers.json_ld import JSONLDImporter
+    from .importers.json_ld_lcia import JSONLDLCIAImporter
+    from .strategies import remove_random_exchanges, remove_useeio_products
 
     with tempfile.TemporaryDirectory() as td:
         dp = Path(td)
@@ -218,10 +220,11 @@ def exiobase_monetary(
     name=None,
     ignore_small_balancing_corrections=True,
 ):
-    from .download_utils import download_with_progressbar
     import tempfile
     from pathlib import Path
 
+    from .download_utils import download_with_progressbar
+
     mapping = {
         (3, 8, 2): {
             "url": "https://zenodo.org/record/5589597/files/IOT_{year}_{system}.zip?download=1",

bw2io/ecoinvent.py

@@ -313,7 +313,12 @@ def import_ecoinvent_release(
 
         for row in cfs:
             if lcia_prefix:
-                impact_category = (lcia_prefix, row["method"], row["category"], row["indicator"])
+                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:

bw2io/extractors/ecospold1.py

@@ -3,7 +3,7 @@
 import os
 from io import StringIO
 from pathlib import Path
-from typing import Any, Union, Optional
+from typing import Any, Optional, Union
 
 import numpy as np
 import pyecospold

bw2io/importers/__init__.py

@@ -11,9 +11,9 @@
 from .excel_lcia import CSVLCIAImporter, ExcelLCIAImporter
 from .exiobase3_hybrid import Exiobase3HybridImporter
 from .exiobase3_monetary import Exiobase3MonetaryImporter
+from .simapro_block_csv import SimaProBlockCSVImporter
 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

@@ -2,9 +2,10 @@
 import functools
 import itertools
 import warnings
-from typing import Optional
+from typing import Optional, Tuple
 
-from bw2data import Database, config, databases, parameters, labels
+from bw2data import Database, config, databases, labels, parameters
+from bw2data.data_store import ProcessedDataStore
 from bw2data.parameters import (
     ActivityParameter,
     DatabaseParameter,
@@ -56,7 +57,11 @@ def __init__(self, db_name):
     def all_linked(self):
         return self.statistics()[2] == 0
 
-    def statistics(self, print_stats=True):
+    @property
+    def needs_multifunctional_database(self):
+        return any(ds.get("type") == "multifunctional" for ds in self.data)
+
+    def statistics(self, print_stats: bool = True) -> Tuple[int, int, int, int]:
         num_datasets = len(self.data)
         num_exchanges = sum([len(ds.get("exchanges", [])) for ds in self.data])
         num_unlinked = len(
@@ -65,8 +70,12 @@ def statistics(self, print_stats=True):
                 for ds in self.data
                 for exc in ds.get("exchanges", [])
                 if not exc.get("input")
+                and not (ds.get("type") == "multifunctional" and exc.get("functional"))
             ]
         )
+        num_multifunctional = sum(
+            1 for ds in self.data if ds.get("type") == "multifunctional"
+        )
         if print_stats:
             unique_unlinked = collections.defaultdict(set)
             for ds in self.data:
@@ -75,19 +84,28 @@ def statistics(self, print_stats=True):
             unique_unlinked = sorted(
                 [(k, len(v)) for k, v in list(unique_unlinked.items())]
             )
-
-            print(
-                (
-                    "{} datasets\n{} exchanges\n{} unlinked exchanges\n  "
-                    + "\n  ".join(
-                        [
-                            "Type {}: {} unique unlinked exchanges".format(*o)
-                            for o in unique_unlinked
-                        ]
-                    )
-                ).format(num_datasets, num_exchanges, num_unlinked)
+            uu = "\n\t".join(
+                [
+                    "Type {}: {} unique unlinked exchanges".format(*o)
+                    for o in unique_unlinked
+                ]
             )
-        return num_datasets, num_exchanges, num_unlinked
+
+            if num_multifunctional:
+                print(
+                    f"""{num_datasets} datasets, including {num_multifunctional} multifunctional datasets
+\t{num_exchanges} exchanges
+\t{num_unlinked} unlinked exchanges ({len(unique_unlinked)} unique)
+\t{uu}"""
+                )
+            else:
+                print(
+                    f"""{num_datasets} datasets
+\t{num_exchanges} exchanges
+\t{num_unlinked} unlinked exchanges ({len(unique_unlinked)} unique)
+\t{uu}"""
+                )
+        return num_datasets, num_exchanges, num_unlinked, num_multifunctional
 
     def write_project_parameters(self, data=None, delete_existing=True):
         """Write global parameters to ``ProjectParameter`` database table.
@@ -201,6 +219,16 @@ def _write_activity_parameters(self, activity_parameters):
             for key in keys:
                 parameters.add_exchanges_to_group(group, key)
 
+    def database_class(
+        self, db_name: str, requested_backend: str = "sqlite"
+    ) -> ProcessedDataStore:
+        from multifunctional import MultifunctionalDatabase
+
+        if self.needs_multifunctional_database:
+            return MultifunctionalDatabase(db_name)
+        else:
+            return Database(db_name, backend=requested_backend)
+
     def write_database(
         self,
         data: Optional[dict] = None,
@@ -260,7 +288,7 @@ def write_database(
 
         if db_name in databases:
             # TODO: Raise error if unlinked exchanges?
-            db = Database(db_name)
+            db = self.database_class(db_name)
             if delete_existing:
                 existing = {}
             else:
@@ -269,9 +297,14 @@ def write_database(
             existing = {}
             if "format" not in self.metadata:
                 self.metadata["format"] = self.format
+            if (
+                self.needs_multifunctional_database
+                and "default_allocation" not in self.metadata
+            ):
+                self.metadata["default_allocation"] = "manual_allocation"
             with warnings.catch_warnings():
                 warnings.simplefilter("ignore")
-                db = Database(db_name, backend=backend)
+                db = self.database_class(db_name)
                 db.register(**self.metadata)
 
         self.write_database_parameters(activate_parameters, delete_existing)
@@ -346,27 +379,39 @@ def create_new_biosphere(self, biosphere_name: str):
             raise ValueError(f"{biosphere_name} database already exists")
 
         def reformat(exc):
-            return exc | {"type": labels.biosphere_node_default, "exchanges": [], "database": biosphere_name, "code": activity_hash(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")
-        ]}
+        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")
+            ]
+        }
 
         if not bio_data:
-            print("Skipping biosphere database creation as all biosphere flows are linked")
+            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")
+        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=f"Database for unlinked biosphere flows from {self.db_name}"
+                format=self.format,
+                comment=f"Database for unlinked biosphere flows from {self.db_name}",
             )
 
         new_bio.write(bio_data)