Duck-typing products for the exchanges tab

Author: mrvisscher

activity_browser/bwutils/__init__.py

@@ -8,7 +8,7 @@
 from .commontasks import cleanup_deleted_bw_projects as cleanup
 from .commontasks import (refresh_node, refresh_node_or_none, refresh_parameter, refresh_edge, refresh_edge_or_none,
                           parameters_in_scope, exchanges_to_sdf, database_is_locked, database_is_legacy, projects_by_last_opened,
-                          node_group)
+                          node_group, is_node_product, is_node_biosphere)
 from .metadata import AB_metadata
 from .montecarlo import MonteCarloLCA
 from .multilca import MLCA, Contributions

activity_browser/bwutils/commontasks.py

@@ -195,6 +195,28 @@ def get_activity_name(key, str_length=22):
     return ",".join(key.get("name", "").split(",")[:3])[:str_length]
 
 
+def is_node_product(node: tuple | int | bd.Node) -> bool:
+    node = refresh_node(node)
+    raw_type = node._document.type
+
+    if raw_type in ["product", "processwithreferenceproduct"]:
+        return True
+
+    if raw_type == "process" and len(node.production()):
+        return True
+
+    return False
+
+
+def is_node_biosphere(node: tuple | int | bd.Node) -> bool:
+    node = refresh_node(node)
+    raw_type = node._document.type
+
+    if raw_type in ["natural resource", "emission", "inventory indicator", "economic", "social"]:
+        return True
+    return False
+
+
 def refresh_node(node: tuple | int | bd.Node) -> bd.Node:
     if isinstance(node, bd.Node):
         node = bd.get_node(id=node.id)

activity_browser/layouts/pages/activity_details/exchanges_tab.py

@@ -228,16 +228,21 @@ def dropEvent(self, event):
         exchanges = {"technosphere": set(), "biosphere": set()}
 
         for key in keys:
-            act = bd.get_node(key=key)
-            if act["type"] not in EXCHANGE_MAP:
-                continue
-            exc_type = EXCHANGE_MAP[act["type"]]
-            exchanges[exc_type].add(act.key)
+            if exc_type := get_exchange_type(key):
+                exchanges[exc_type].add(key)
 
         # Run the action for new exchanges
         for exc_type, keys in exchanges.items():
             actions.ExchangeNew.run(keys, self.activity.key, exc_type)
 
+def get_exchange_type(activity_key: tuple) -> str | None:
+    if bwutils.is_node_product(activity_key):
+        return "technosphere"
+    elif bwutils.is_node_biosphere(activity_key):
+        return "biosphere"
+    return None
+
+
 
 class DropOverlay(QtWidgets.QWidget):
     def __init__(self, parent=None):