Enable duplication to functional-sqlite

Author: mrvisscher

activity_browser/actions/activity/activity_duplicate_to_db.py

@@ -3,6 +3,7 @@
 from qtpy import QtWidgets
 
 import bw2data as bd
+import bw_functional as bf
 
 from activity_browser import application
 from activity_browser.bwutils import refresh_node
@@ -18,40 +19,44 @@ class ActivityDuplicateToDB(ABAction):
 
     @classmethod
     @exception_dialogs
-    def run(cls, nodes: List[tuple | int | bd.Node], to_db: str = None):
+    def run(cls, nodes: List[tuple | int | bd.Node], to_db_name: str = None):
         nodes = [refresh_node(node) for node in nodes]
-
         dbs = {node.get("database") for node in nodes}
+        from_db_name = next(iter(dbs))
+        from_db_backend = bd.databases[from_db_name]["backend"]
+
         if not len(dbs) == 1:
             raise ValueError("All selected activities must be from the same database.")
-        from_db = next(iter(dbs))
 
-        if to_db and not cls.confirm_db(to_db):
-            return
+        if any([isinstance(node, bf.Product) for node in nodes]):
+            raise ValueError("Products cannot be duplicated to another database. Duplicate the parent process instead.")
 
-        to_db = to_db or cls.request_db(nodes, backend=bd.databases[from_db]["backend"])
+        if to_db_name:
+            if not cls.confirm_db(to_db_name):
+                return
+        else:
+            to_db_name = cls.request_db(from_db_name)
 
-        if not to_db:
-            return
+        to_db_backend = bd.databases[to_db_name]["backend"]
 
-        new_nodes = []
-
-        # otherwise move all supplied nodes to the db by copying them
-        for node in nodes:
-            new_node = node.copy(database=to_db)
-            new_nodes.append(new_node)
+        if from_db_backend == to_db_backend:
+            new_nodes = cls.duplicate_simple(nodes, to_db_name)
+        elif from_db_backend == "sqlite" and to_db_backend == "functional_sqlite":
+            new_nodes = cls.duplicate_sqlite_to_functional_sqlite(nodes, to_db_name)
+        elif from_db_backend == "functional_sqlite" and to_db_backend == "sqlite":
+            new_nodes = cls.duplicate_functional_sqlite_to_sqlite(nodes, to_db_name)
+        else:
+            raise NotImplementedError(f"Moving from {from_db_backend} to {to_db_backend} is not supported.")
 
         ActivityOpen.run(new_nodes)
 
     @staticmethod
-    def request_db(nodes: list[bd.Node], backend: str) -> str | None:
-        # get valid databases (not the original database, locked databases, or databases with a different backend)
-        origin_db = next(iter(nodes)).get("database")
+    def request_db(from_db_name: str) -> str | None:
+        # get valid databases (not the original database, or locked databases)
         target_dbs = [
-            db for db, meta in bd.databases.items() if
-            db != origin_db
-            and meta.get("read_only") is not True
-            and meta.get("backend") == backend
+            db_name for db_name, meta in bd.databases.items() if
+            db_name != from_db_name
+            and meta.get("read_only", True) is not True
         ]
 
         # return if there are no valid databases to duplicate to
@@ -76,10 +81,54 @@ def request_db(nodes: list[bd.Node], backend: str) -> str | None:
         return target_db if ok else None
 
     @staticmethod
-    def confirm_db(to_db: str):
+    def confirm_db(to_db_name: str):
         user_choice = QtWidgets.QMessageBox.question(
             application.main_window,
             "Move to new database",
-            f"Move to {to_db} and open as new tab?",
+            f"Move to {to_db_name} and open as new tab?",
         )
         return user_choice == user_choice.Yes
+
+    @staticmethod
+    def duplicate_simple(nodes: list[bd.Node], to_db_name: str) -> list[bd.Node]:
+        new_nodes = []
+
+        # move all supplied nodes to the db by copying them
+        for node in nodes:
+            new_node = node.copy(database=to_db_name)
+            new_nodes.append(new_node)
+
+        return new_nodes
+
+    @staticmethod
+    def duplicate_sqlite_to_functional_sqlite(nodes: list[bd.Node], to_db_name: str) -> list[bd.Node]:
+        from bw_functional.convert import SQLiteToFunctionalSQLite
+        new_nodes = []
+
+        for node in nodes:
+            dataset = node.as_dict()
+
+            dataset.pop("id", None)
+            dataset.pop("key", None)
+
+            dataset["exchanges"] = [exc.as_dict() for exc in node.exchanges()]
+            dataset["database"] = to_db_name  # because we didn't copy the dict this will also be reflected in node.key
+
+            new_datasets = SQLiteToFunctionalSQLite.convert_process(node.key, dataset, False)
+            new_exchanges = [x for ds in new_datasets.values() for x in ds.pop("exchanges", [])]
+
+            for key, new_dataset in new_datasets.items():
+                new_node = bd.Node(**new_dataset)
+                new_node.save()
+                new_nodes.append(new_node)
+
+            for exc in new_exchanges:
+                exc["output"] = (to_db_name, exc["output"][1])  # relink output to new db
+                new_exc = bd.Edge(**exc)
+                new_exc.save()
+
+        return new_nodes
+
+    @staticmethod
+    def duplicate_functional_sqlite_to_sqlite(nodes: list[bd.Node], to_db_name: str) -> list[bd.Node]:
+        raise NotImplementedError("Duplicating from functional_sqlite to sqlite is not yet implemented.")

activity_browser/layouts/panes/database_products.py

@@ -107,7 +107,7 @@ def build_df(self) -> pd.DataFrame:
         df = AB_metadata.get_database_metadata(self.database.name, cols)
 
         processors = set(df["processor"].dropna().unique())
-        df = df.drop(processors)
+        df = df.drop(processors, errors="ignore")
 
         if not df.properties.isna().all():
             props_df = df[df.properties.notna()]