Add check for waste treatment paradigm alignment

bw_simapro_csv/blocks/process.py

@@ -1,6 +1,7 @@
 from bw2parameters import Interpreter, ParameterSet
 
 from ..constants import CONTEXT_MAPPING, MAGIC
+from ..errors import WasteModelMismatch
 from ..parameters import (
     FormulaSubstitutor,
     add_prefix_to_uppercase_input_parameters,
@@ -182,3 +183,25 @@ def supplement_biosphere_edges(self, blocks: list[SimaProCSVBlock]) -> None:
                         edge["comment"] += MAGIC + partner["comment"]
                     else:
                         edge["comment"] = partner["comment"]
+
+    def check_waste_production_model_consistency(self):
+        """Check to make sure that our understanding of SimaPro waste treatment aligns with the data."""
+        if "Waste treatment" in self.blocks and self.blocks["Waste treatment"].parsed:
+            if "Products" in self.blocks and self.blocks["Products"].parsed:
+                raise WasteModelMismatch(
+                    "We don't know how to parse a process with {} waste treatment inputs and {} products".format(
+                        len(self.blocks["Waste treatment"].parsed),
+                        len(self.blocks["Products"].parsed),
+                    )
+                )
+            elif self.parsed["metadata"]["Category type"] != "waste treatment":
+                raise WasteModelMismatch(
+                    "Expected waste treatment processes to have category type `waste treatment`; instead got `{}`".format(
+                        self.parsed["metadata"]["Category type"]
+                    )
+                )
+        elif "Products" in self.blocks and self.blocks["Products"].parsed:
+            if self.parsed["metadata"]["Category type"] == "waste treatment":
+                raise WasteModelMismatch(
+                    "Expected processes with `Products` blocks not have category type `waste treatment`"
+                )

bw_simapro_csv/errors.py

@@ -1,2 +1,8 @@
+class WasteModelMismatch(Exception):
+    """Our understanding of SimaPro waste models doesn't agree with this data"""
+
+    pass
+
+
 class IndeterminateBlockEnd(Exception):
     pass

bw_simapro_csv/main.py

@@ -280,4 +280,5 @@ def resolve_parameters(self) -> None:
 
         for block in filter(lambda b: isinstance(b, Process), self):
             block.resolve_local_parameters(global_params=global_params, substitutes=substitutes)
+            block.check_waste_production_model_consistency()
             block.supplement_biosphere_edges(blocks=self.blocks)

tests/test_process.py

@@ -1,4 +1,7 @@
+import pytest
+
 from bw_simapro_csv.blocks import GenericBiosphere, Process
+from bw_simapro_csv.errors import WasteModelMismatch
 
 
 def test_supplement_biosphere_edges():
@@ -51,3 +54,71 @@ def __init__(self):
 
     p.supplement_biosphere_edges([b])
     assert p.blocks["Emissions to air"].parsed == expected
+
+
+def test_check_waste_production_model_consistency_products():
+    class O:
+        pass
+
+    class P(Process):
+        def __init__(self):
+            pass
+
+    wt = O()
+    wt.category = "Waste treatment"
+    wt.parsed = [1]
+
+    pr = O()
+    pr.category = "Products"
+    pr.parsed = []
+
+    p = P()
+    p.blocks = {"Products": pr, "Waste treatment": wt}
+    p.parsed = {"metadata": {"Category type": "waste treatment"}}
+    assert p.check_waste_production_model_consistency() is None
+
+    wt.parsed = [1, 2, 3]
+    pr.parsed = [2, 3, 4, 5]
+
+    with pytest.raises(WasteModelMismatch) as exc_info:
+        p.check_waste_production_model_consistency()
+
+    assert "3 waste treatment inputs and 4 products" in str(exc_info.value)
+
+
+def test_check_waste_production_model_consistency_category_type():
+    class O:
+        pass
+
+    class P(Process):
+        def __init__(self):
+            pass
+
+    wt = O()
+    wt.category = "Waste treatment"
+    wt.parsed = [1]
+
+    p = P()
+    p.blocks = {"Waste treatment": wt}
+    p.parsed = {"metadata": {"Category type": "waste treatment"}}
+    assert p.check_waste_production_model_consistency() is None
+
+    p.parsed = {"metadata": {"Category type": "w00t"}}
+
+    with pytest.raises(WasteModelMismatch) as exc_info:
+        p.check_waste_production_model_consistency()
+
+    assert "`waste treatment`; instead got `w00t`" in str(exc_info.value)
+
+    pr = O()
+    pr.category = "Products"
+    pr.parsed = [1]
+
+    wt.parsed = []
+
+    p.blocks = {"Waste treatment": wt, "Products": pr}
+    p.parsed = {"metadata": {"Category type": "waste treatment"}}
+    with pytest.raises(WasteModelMismatch) as exc_info:
+        p.check_waste_production_model_consistency()
+
+    assert "Expected processes with `Products` blocks" in str(exc_info.value)