Conversion to reference units

bw_simapro_csv/blocks/__init__.py

@@ -4,6 +4,7 @@
     "DatabaseInputParameters",
     "EmptyBlock",
     "GenericBiosphere",
+    "GenericUncertainBiosphere",
     "ImpactCategory",
     "LiteratureReference",
     "Method",
@@ -13,20 +14,21 @@
     "ProjectCalculatedParameters",
     "ProjectInputParameters",
     "Quantities",
+    "RemainingWaste",
+    "SeparatedWaste",
     "SimaProCSVBlock",
     "SystemDescription",
+    "TechnosphereEdges",
     "Units",
     "WasteScenario",
     "WasteTreatment",
-    "RemainingWaste",
-    "SeparatedWaste",
 )
 
 
 from .base import EmptyBlock, SimaProCSVBlock
 from .calculated_parameters import DatabaseCalculatedParameters, ProjectCalculatedParameters
 from .damage_category import DamageCategory
-from .generic_biosphere import GenericBiosphere
+from .generic_biosphere import GenericBiosphere, GenericUncertainBiosphere
 from .impact_category import ImpactCategory
 from .literature_reference import LiteratureReference
 from .method import Method
@@ -36,5 +38,6 @@
 from .products import Products
 from .quantities import Quantities
 from .system_description import SystemDescription
+from .technosphere_edges import TechnosphereEdges
 from .units import Units
 from .wastes import RemainingWaste, SeparatedWaste, WasteScenario, WasteTreatment

bw_simapro_csv/main.py

@@ -35,6 +35,7 @@
     prepare_formulas,
     substitute_in_formulas,
 )
+from .units import normalize_units
 from .utils import BeKindRewind
 
 
@@ -144,6 +145,7 @@ def __init__(
                 self.blocks.append(block)
 
         self.resolve_parameters()
+        normalize_units(self.blocks)
 
     def __iter__(self):
         return iter(self.blocks)

bw_simapro_csv/units.py

@@ -0,0 +1,63 @@
+from .blocks import (
+    GenericUncertainBiosphere,
+    Process,
+    Products,
+    SimaProCSVBlock,
+    TechnosphereEdges,
+    Units,
+    WasteTreatment,
+)
+
+HAS_UNITS = (
+    GenericUncertainBiosphere,
+    Products,
+    TechnosphereEdges,
+    WasteTreatment,
+)
+
+from loguru import logger
+
+from .uncertainty import recalculate_uncertainty_distribution
+
+missing_units = set()
+
+
+def normalize_units(blocks: list[SimaProCSVBlock]) -> list[SimaProCSVBlock]:
+    """Convert all values to the base unit.
+
+    Need to change:
+    * amount
+    * formula
+    * uncertainty distribution
+
+    """
+    unit_mapping = {
+        o["name"]: o
+        for block in filter(lambda x: isinstance(x, Units), blocks)
+        for o in block.parsed
+    }
+
+    for process_block in filter(lambda x: isinstance(x, Process), blocks):
+        for block in filter(lambda x: isinstance(x, HAS_UNITS), process_block.blocks.values()):
+            for obj in block.parsed:
+                try:
+                    mapping = unit_mapping[obj["unit"]]
+                    if mapping["reference unit name"] != obj["unit"]:
+                        logger.debug(
+                            "Changing units from {a} to {b} with conversion factor {c} on line {d}",
+                            a=obj["unit"],
+                            b=mapping["reference unit name"],
+                            c=mapping["conversion"],
+                            d=obj["line_no"],
+                        )
+                        obj["unit"] = mapping["reference unit name"]
+                        if "formula" in obj:
+                            obj["formula"] = f"({obj['formula']}) * {mapping['conversion']}"
+                        if "uncertainty type" in obj:
+                            recalculate_uncertainty_distribution(obj, mapping["conversion"])
+                        else:
+                            obj["amount"] *= mapping["conversion"]
+                except KeyError:
+                    if obj['unit'] not in missing_units:
+                        missing_units.add(obj['unit'])
+                        logger.critical("Unknown unit {unit} on line {line_no}", **obj)