Fix #15

multifunctional/allocation.py

@@ -51,6 +51,7 @@ def generic_allocation(
     if not total:
         raise ZeroDivisionError("Sum of allocation factors is zero")
 
+    act["mf_allocation_run_uuid"] = uuid4().hex
     processes = [act]
 
     for original_exc in filter(lambda x: x.get("functional"), act.get("exchanges", [])):
@@ -146,6 +147,9 @@ def generic_allocation(
 
         processes.append(allocated_process)
 
+    # Useful for other functions like purging expired links in future
+    act["mf_was_once_allocated"] = True
+
     return processes
 
 

multifunctional/edge_classes.py

@@ -19,9 +19,6 @@ def __setitem__(self, key, value):
 
 
 class ReadOnlyExchanges(Exchanges):
-    def delete(self):
-        raise NotImplementedError("Exchanges are read-only")
-
     def __iter__(self):
         for obj in self._get_queryset():
             yield ReadOnlyExchange(obj)

multifunctional/errors.py

@@ -1,2 +1,8 @@
 class NoAllocationNeeded:
     pass
+
+
+class MultipleFunctionalExchangesWithSameInput(Exception):
+    """Multiple functional links to same input product is not allowed."""
+
+    pass

multifunctional/node_classes.py

@@ -9,6 +9,7 @@
 from .errors import NoAllocationNeeded
 from .utils import (
     product_as_process_name,
+    purge_expired_linked_readonly_processes,
     set_correct_process_type,
     update_datasets_from_allocation_results,
 )
@@ -32,12 +33,8 @@ class MaybeMultifunctionalProcess(BaseMultifunctionalNode):
     Sets flag on save if multifunctional."""
 
     def save(self):
-        if self.multifunctional:
-            self._data["type"] = "multifunctional"
-        elif not self._data.get("type"):
-            # TBD: This should use bw2data.utils.set_correct_process_type
-            # but that wants datasets as dicts with exchanges
-            self._data["type"] = labels.process_node_default
+        set_correct_process_type(self)
+        purge_expired_linked_readonly_processes(self)
         super().save()
 
     def __str__(self):
@@ -52,6 +49,8 @@ def allocate(
         if self.get("skip_allocation"):
             return NoAllocationNeeded
         if not self.multifunctional:
+            # Call save because we don't know if the process type should be changed
+            self.save()
             return NoAllocationNeeded
 
         from . import allocation_strategies
@@ -121,11 +120,6 @@ def new_edge(self, **kwargs):
             "This node is read only. Update the corresponding multifunctional process."
         )
 
-    def delete(self):
-        raise NotImplementedError(
-            "This node is read only. Update the corresponding multifunctional process."
-        )
-
     def exchanges(self, exchanges_class=None):
         if exchanges_class is not None:
             warnings.warn("`exchanges_class` argument ignored; must be `ReadOnlyExchanges`")

multifunctional/utils.py

@@ -1,16 +1,17 @@
+from collections import Counter
 from pprint import pformat
 from typing import Dict, List
 
-from bw2data import get_node
-from bw2data.backends import Exchange
+from bw2data import get_node, labels
+from bw2data.backends import Exchange, Node
 from bw2data.backends.schema import ExchangeDataset
 from bw2data.errors import UnknownObject
 from loguru import logger
 
+from multifunctional.errors import MultipleFunctionalExchangesWithSameInput
 
-def allocation_before_writing(
-    data: Dict[tuple, dict], strategy_label: str
-) -> Dict[tuple, dict]:
+
+def allocation_before_writing(data: Dict[tuple, dict], strategy_label: str) -> Dict[tuple, dict]:
     """Utility to perform allocation on datasets and expand `data` with allocated processes."""
     from . import allocation_strategies
 
@@ -62,7 +63,7 @@ def add_exchange_input_if_missing(data: dict) -> dict:
 
 
 def update_datasets_from_allocation_results(data: List[dict]) -> None:
-    """Given data from allocation, create or update datasets as needed from `data`."""
+    """Given data from allocation, create, update, or delete datasets as needed from `data`."""
     from .node_classes import ReadOnlyProcessWithReferenceProduct
 
     for ds in data:
@@ -98,3 +99,121 @@ def product_as_process_name(data: List[dict]) -> None:
         functional_excs = [exc for exc in ds["exchanges"] if exc.get("functional")]
         if len(functional_excs) == 1 and functional_excs[0].get("name"):
             ds["name"] = functional_excs[0]["name"]
+
+
+def set_correct_process_type(dataset: Node) -> Node:
+    """
+    Change the `type` for an LCI process under certain conditions.
+
+    Only will make changes if the following conditions are met:
+
+    * `type` is `multifunctional` but the dataset is no longer multifunctional ->
+        set to either `process` or `processwithreferenceproduct`
+    * `type` is `None` or missing -> set to either `process` or `processwithreferenceproduct`
+    * `type` is `process` but the dataset also includes an exchange which points to the same node
+        -> `processwithreferenceproduct`
+
+    """
+    if dataset.get("type") not in (
+        labels.chimaera_node_default,
+        labels.process_node_default,
+        "multifunctional",
+        None,
+    ):
+        pass
+    elif dataset.multifunctional:
+        dataset["type"] = "multifunctional"
+    elif any(exc.input == exc.output for exc in dataset.exchanges()):
+        if dataset["type"] == "multifunctional":
+            logger.debug(
+                "Changed %s (%s) type from `multifunctional` to `%s`",
+                dataset.get("name"),
+                dataset.id,
+                labels.chimaera_node_default,
+            )
+        dataset["type"] = labels.chimaera_node_default
+    elif any(exc.get("functional") for exc in dataset.exchanges()):
+        if dataset["type"] == "multifunctional":
+            logger.debug(
+                "Changed %s (%s) type from `multifunctional` to `%s`",
+                dataset.get("name"),
+                dataset.id,
+                labels.process_node_default,
+            )
+        dataset["type"] = labels.process_node_default
+    elif (
+        # No production edges -> implicit self production -> chimaera
+        not any(
+            exc.get("type") in labels.technosphere_positive_edge_types
+            for exc in dataset.exchanges()
+        )
+    ):
+        dataset["type"] = labels.chimaera_node_default
+    elif not dataset.get("type"):
+        dataset["type"] = labels.process_node_default
+    else:
+        # No conditions for setting or changing type occurred
+        pass
+
+    return dataset
+
+
+def purge_expired_linked_readonly_processes(dataset: Node) -> None:
+    from .database import MultifunctionalDatabase
+
+    if not dataset.get("mf_was_once_allocated"):
+        return
+
+    if dataset["type"] == "multifunctional":
+        # Can have some readonly allocated processes which refer to non-functional edges
+        for ds in MultifunctionalDatabase(dataset["database"]):
+            if (
+                ds["type"] in ("readonly_process",)
+                and ds.get("mf_parent_key") == dataset.key
+                and ds["mf_allocation_run_uuid"] != dataset["mf_allocation_run_uuid"]
+            ):
+                ds.delete()
+
+        for exc in dataset.exchanges():
+            try:
+                exc.input
+            except UnknownObject:
+                exc.input = dataset
+                exc.save()
+                logger.debug(
+                    "Edge to deleted readonly process redirected to parent process: %s",
+                    exc,
+                )
+
+    else:
+        # Process or chimaera process with one functional edge
+        # Make sure that single functional edge is not referring to obsolete readonly process
+        functional_edges = [exc for exc in dataset.exchanges() if exc.get("functional")]
+        if not len(functional_edges) < 2:
+            raise ValueError(
+                f"Process marked monofunctional with type {dataset['type']} but has {len(functional_edges)} functional edges"
+            )
+        edge = functional_edges[0]
+        if edge.input["type"] in (
+            "readonly_process",
+        ):  # TBD https://github.com/brightway-lca/multifunctional/issues/23
+            # This node should be deleted; have to change to chimaera process with self-input
+            logger.debug(
+                "Edge to expired readonly process %s redirected to parent process %s",
+                edge.input,
+                dataset,
+            )
+            edge.input = dataset
+            edge.save()
+            if dataset["type"] != labels.chimaera_node_default:
+                logger.debug(
+                    "Change node type to chimaera: %s (%s)",
+                    dataset,
+                    dataset.id,
+                )
+                dataset["type"] = labels.chimaera_node_default
+
+        # Obsolete readonly processes
+        for ds in MultifunctionalDatabase(dataset["database"]):
+            if ds["type"] in ("readonly_process",) and ds.get("mf_parent_key") == dataset.key:
+                ds.delete()

tests/conftest.py

@@ -7,6 +7,7 @@
 from fixtures.internal_linking import DATA as INTERNAL_LINKING_DATA
 from fixtures.product_properties import DATA as PP_DATA
 from fixtures.products import DATA as PRODUCT_DATA
+from fixtures.many_products import DATA as MANY_PRODUCTS_DATA
 
 from multifunctional import MultifunctionalDatabase, allocation_before_writing
 
@@ -34,6 +35,15 @@ def products():
     return db
 
 
+@pytest.fixture
+@bw2test
+def many_products():
+    db = MultifunctionalDatabase("products")
+    db.write(deepcopy(MANY_PRODUCTS_DATA), process=False)
+    db.metadata["dirty"] = True
+    return db
+
+
 @pytest.fixture
 @bw2test
 def errors():

tests/fixtures/many_products.py

@@ -0,0 +1,71 @@
+DATA = {
+    ("products", "a"): {
+        "name": "flow - a",
+        "code": "a",
+        "unit": "kg",
+        "type": "emission",
+        "categories": ("air",),
+    },
+    ("products", "p1"): {
+        "type": "product",
+        "name": "first product",
+        "unit": "kg",
+        "exchanges": [],
+    },
+    ("products", "p2"): {
+        "type": "product",
+        "name": "first product",
+        "unit": "kg",
+        "exchanges": [],
+    },
+    ("products", "p3"): {
+        "type": "product",
+        "name": "first product",
+        "unit": "kg",
+        "exchanges": [],
+    },
+    ("products", "1"): {
+        "name": "process - 1",
+        "code": "1",
+        "location": "first",
+        "type": "multifunctional",
+        "exchanges": [
+            {
+                "functional": True,
+                "type": "production",
+                "input": ("products", "p1"),
+                "amount": 4,
+                "properties": {
+                    "price": 7,
+                    "mass": 6,
+                },
+            },
+            {
+                "functional": True,
+                "type": "production",
+                "input": ("products", "p2"),
+                "amount": 4,
+                "properties": {
+                    "price": 7,
+                    "mass": 6,
+                },
+            },
+            {
+                "functional": True,
+                "type": "production",
+                "input": ("products", "p3"),
+                "amount": 4,
+                "properties": {
+                    "price": 7,
+                    "mass": 6,
+                },
+            },
+            {
+                "type": "biosphere",
+                "name": "flow - a",
+                "amount": 10,
+                "input": ("products", "a"),
+            },
+        ],
+    },
+}

tests/test_internal_linking_zero_allocation.py

@@ -80,6 +80,7 @@ def test_allocation_sets_code_for_zero_allocation_products_in_multifunctional_pr
                 },
             ],
             "type": "multifunctional",
+            "mf_was_once_allocated": True,
             "mf_strategy_label": "property allocation by 'manual_allocation'",
             "name": "(unknown)",
             "location": None,
@@ -153,4 +154,8 @@ def test_allocation_sets_code_for_zero_allocation_products_in_multifunctional_pr
             "database": "db",
         },
     ]
-    assert allocation_strategies["manual_allocation"](given) == expected
+    result = allocation_strategies["manual_allocation"](given)
+    for node in result:
+        if "mf_allocation_run_uuid" in node:
+            del node["mf_allocation_run_uuid"]
+    assert result == expected

tests/test_node_creation.py

@@ -34,7 +34,7 @@ def test_node_creation_default_label():
         assert node["name"] == "foo"
         assert node["database"] == "test database"
         assert node["code"]
-        assert node["type"] == bd.labels.process_node_default
+        assert node["type"] == bd.labels.chimaera_node_default
 
 
 @bw2test

tests/test_read_only_nodes.py

@@ -10,10 +10,6 @@ def test_read_only_node(basic):
     node = sorted(basic, key=lambda x: (x["name"], x.get("reference product", "")))[2]
     assert isinstance(node, mf.ReadOnlyProcessWithReferenceProduct)
 
-    with pytest.raises(NotImplementedError) as info:
-        node.delete()
-    assert "This node is read only" in info.value.args[0]
-
     with pytest.raises(NotImplementedError) as info:
         node.copy()
     assert "This node is read only" in info.value.args[0]
@@ -42,10 +38,6 @@ def test_read_only_exchanges(basic):
             exc.save()
         assert "Read-only exchange" in info.value.args[0]
 
-        with pytest.raises(NotImplementedError) as info:
-            exc.delete()
-        assert "Read-only exchange" in info.value.args[0]
-
         with pytest.raises(NotImplementedError) as info:
             exc["foo"] = "bar"
         assert "Read-only exchange" in info.value.args[0]
@@ -58,10 +50,6 @@ def test_read_only_exchanges(basic):
         #     exc.output = node
         # assert 'Read-only exchange' in info.value.args[0]
 
-    with pytest.raises(NotImplementedError) as info:
-        node.exchanges().delete()
-    assert "Exchanges are read-only" in info.value.args[0]
-
 
 def test_read_only_parent(basic):
     basic.metadata["default_allocation"] = "mass"