diff --git a/CONTEXT.md b/CONTEXT.md
index d31e5acc0..d175f07f7 100644
--- a/CONTEXT.md
+++ b/CONTEXT.md
@@ -57,6 +57,16 @@ A named set of functional unit(s) and LCIA method(s) used to run LCA / multi-LCA
A Life Cycle Impact Assessment method (characterization factors for elementary flows). In Brightway, methods are keyed tuples; AB exposes them in impact-category UI.
+### AB impact-category file (AB LCIA format)
+
+Activity Browser’s multi–impact-category interchange for import/export: characterization factors plus per–impact-category unit and description. Method keys and elementary-flow identities use `::` (variable-length Brightway tuples / categories). Excel uses sheets `CFs` and `Impact categories`; CSV uses a sibling pair `*.cfs.csv` + `*.metadata.csv`. Distinct from ecoinvent’s LCIA implementation workbook (fixed three-part names; separate name/compartment/subcompartment columns) and from the **bw2io impact-category file**.
+_Avoid_: Indicators sheet (when meaning AB’s impact-category metadata table), AB ecoinvent format
+
+### bw2io impact-category file (bw2io LCIA format)
+
+bw2io’s Excel/CSV LCIA CF template: **one impact category per CF file/sheet** (`name`, `categories` with `::`, `amount`, optional uncertainty). AB may add a `metadata` sheet (xlsx) or `metadata.csv` sidecar for method/unit/description/`filename`; stock bw2io only needs the CF table.
+_Avoid_: one-shot, bw2io native (as a product name), AB impact-category file
+
### Characterization factor (CF)
A factor that converts an elementary flow amount into an impact-category score for a given method.
@@ -120,4 +130,6 @@ Extensibility mechanism for third-party AB features. **Architecture TBD** — do
| “table of processes” | database / activity |
| “flow” without kind | intermediate (technosphere) or elementary (biosphere) flow; exchanges is another synonym |
| “impact method” only | LCIA method / impact category (as used in UI) |
+| “Indicators” (AB LCIA metadata sheet/file) | Impact categories (Excel sheet) / `.metadata.csv` (AB CSV sidecar) |
+| “one-shot” / “bw2io native” (LCIA file) | bw2io impact-category file |
| “global app settings file” ad hoc | `app.settings` |
diff --git a/activity_browser/app/actions/__init__.py b/activity_browser/app/actions/__init__.py
index b5808ccf3..0f0e735a9 100644
--- a/activity_browser/app/actions/__init__.py
+++ b/activity_browser/app/actions/__init__.py
@@ -6,9 +6,9 @@
from .activity.activity_modify import ActivityModify
from .activity.activity_new_process import ActivityNewProcess
from .activity.activity_new_product import ActivityNewProduct
-from .activity.new_elementary_flow import NewElementaryFlow
-from .activity.edit_elementary_flow import EditElementaryFlow
-from .activity.delete_elementary_flow import DeleteElementaryFlow
+from .activity.elementary_flow_new import NewElementaryFlow
+from .activity.elementary_flow_edit import EditElementaryFlow
+from .activity.elementary_flow_delete import DeleteElementaryFlow
from .activity.activity_open import ActivityOpen
from .activity.activity_relink import ActivityRelink
from .activity.activity_sdf_to_clipboard import ActivitySDFToClipboard
@@ -59,8 +59,12 @@
from .method.method_meta_modify import MethodMetaModify
from .method.method_new import MethodNew
-from .method.importer.method_importer_ecoinvent import MethodImporterEcoinvent
-from .method.importer.method_importer_bw2io import MethodImporterBW2IO
+from .method.method_import_ecoinvent import MethodImportEcoinvent
+from .method.method_import_ab import MethodImportAB
+from .method.method_import_bw2io import MethodImportBW2IO
+from .method.method_export_ab import MethodExportAB
+from .method.method_export_bw2io import MethodExportBW2IO
+from .method.method_get_template import MethodGetTemplate
from .method.cf_uncertainty_modify import CFUncertaintyModify
from .method.cf_amount_modify import CFAmountModify
diff --git a/activity_browser/app/actions/activity/delete_elementary_flow.py b/activity_browser/app/actions/activity/elementary_flow_delete.py
similarity index 100%
rename from activity_browser/app/actions/activity/delete_elementary_flow.py
rename to activity_browser/app/actions/activity/elementary_flow_delete.py
diff --git a/activity_browser/app/actions/activity/edit_elementary_flow.py b/activity_browser/app/actions/activity/elementary_flow_edit.py
similarity index 96%
rename from activity_browser/app/actions/activity/edit_elementary_flow.py
rename to activity_browser/app/actions/activity/elementary_flow_edit.py
index db5b60a88..c7ad8e352 100644
--- a/activity_browser/app/actions/activity/edit_elementary_flow.py
+++ b/activity_browser/app/actions/activity/elementary_flow_edit.py
@@ -3,7 +3,7 @@
from qtpy import QtWidgets
from activity_browser import app
-from activity_browser.app.actions.activity.new_elementary_flow import ElementaryFlowDialog
+from activity_browser.app.actions.activity.elementary_flow_new import ElementaryFlowDialog
from activity_browser.app.actions.base import ABAction, exception_dialogs
from activity_browser.bwutils.commontasks import (
get_writable_databases,
diff --git a/activity_browser/app/actions/activity/new_elementary_flow.py b/activity_browser/app/actions/activity/elementary_flow_new.py
similarity index 100%
rename from activity_browser/app/actions/activity/new_elementary_flow.py
rename to activity_browser/app/actions/activity/elementary_flow_new.py
diff --git a/activity_browser/app/actions/database/database_import_from_ecoinvent.py b/activity_browser/app/actions/database/database_import_from_ecoinvent.py
index 0e45431e8..66fe3c62d 100644
--- a/activity_browser/app/actions/database/database_import_from_ecoinvent.py
+++ b/activity_browser/app/actions/database/database_import_from_ecoinvent.py
@@ -16,7 +16,7 @@
from activity_browser.ui import widgets, icons
from activity_browser.app.actions.base import ABAction, exception_dialogs
from activity_browser.bwutils.io.ecoinvent_importer import Ecoinvent7zImporter
-from activity_browser.bwutils.io.ecoinvent_lcia_importer import EcoinventLCIAImporter
+from activity_browser.bwutils.impact_categories import EcoinventLCIAImporter
from activity_browser.mod.bw2io.migrations import ab_create_core_migrations
from activity_browser.ui.core import threading
diff --git a/activity_browser/app/actions/method/importer/method_importer_bw2io.py b/activity_browser/app/actions/method/importer/method_importer_bw2io.py
deleted file mode 100644
index 4d6dfc8b0..000000000
--- a/activity_browser/app/actions/method/importer/method_importer_bw2io.py
+++ /dev/null
@@ -1,59 +0,0 @@
-import os.path
-from loguru import logger
-
-from qtpy.QtCore import Signal, SignalInstance
-
-from activity_browser import app
-from activity_browser.app.actions.base import exception_dialogs
-from activity_browser.ui import icons, widgets
-from activity_browser.bwutils.io.ecoinvent_lcia_importer import EcoinventLCIAImporter
-from activity_browser.ui.core import threading
-
-from .method_importer_ecoinvent import ExtractExcelThread, MethodImporterEcoinvent
-
-
-
-
-class MethodImporterBW2IO(MethodImporterEcoinvent):
- """ABAction to import ecoinvent methods shipped with BW2IO"""
-
- icon = icons.qicons.import_db
- text = "Import from bw2io..."
- tool_tip = "Import methods that come shipped with BW2IO"
-
- @classmethod
- @exception_dialogs
- def run(cls):
- # initialize the import thread, setting needed attributes
- extract_thread = ExtractMethodsThread(app.application)
- extract_thread.loaded.connect(cls.write_database)
-
- # show progress dialog for importing the excel
- progress_dialog = widgets.ABProgressDialog.get_connected_dialog("Importing Database")
- extract_thread.finished.connect(progress_dialog.deleteLater)
-
- extract_thread.start()
-
-
-class ExtractMethodsThread(threading.ABThread):
- loaded: SignalInstance = Signal(EcoinventLCIAImporter)
-
- def run_safely(self):
- import zipfile
- import json
- from bw2io.data import dirpath
-
- fp = os.path.join(dirpath, "lcia", "lcia_39_ecoinvent.zip")
-
- with zipfile.ZipFile(fp, mode="r") as archive:
- data = json.load(archive.open("data.json"))
-
- for method in data:
- method['name'] = tuple(method['name'])
- for obj in method['exchanges']:
- del obj['input']
-
- ei = EcoinventLCIAImporter("lcia_39_ecoinvent.zip")
- ei.data = data
- self.loaded.emit(ei)
-
diff --git a/activity_browser/app/actions/method/method_export_ab.py b/activity_browser/app/actions/method/method_export_ab.py
new file mode 100644
index 000000000..65c76df05
--- /dev/null
+++ b/activity_browser/app/actions/method/method_export_ab.py
@@ -0,0 +1,121 @@
+"""Export impact categories to an AB LCIA file (.xlsx/.csv)."""
+from __future__ import annotations
+
+from typing import List, Optional, Sequence
+
+from loguru import logger
+from qtpy import QtWidgets
+from qtpy.QtCore import Signal, SignalInstance
+
+from activity_browser import app
+from activity_browser.app import application
+from activity_browser.app.actions.base import ABAction, exception_dialogs
+from activity_browser.app.panes.impact_categories import resolve_methods_for_export
+from activity_browser.bwutils.impact_categories import (
+ CancelledError,
+ export_methods_ab_csv_pair,
+ export_methods_ab_xlsx,
+)
+from activity_browser.ui.core import threading
+from activity_browser.app.dialogs import run_thread_with_progress
+
+
+class MethodExportAB(ABAction):
+ """Export selected (or all) impact categories to an AB LCIA Excel workbook."""
+
+ icon = application.style().standardIcon(QtWidgets.QStyle.SP_DialogSaveButton)
+ text = "To AB LCIA file (.xlsx/.csv)…"
+ tool_tip = "Export impact categories to Activity Browser spreadsheet format"
+
+ @classmethod
+ @exception_dialogs
+ def run(cls, method_names: Optional[List[tuple]] = None):
+ method_names = resolve_methods_for_export(method_names)
+ if not method_names:
+ return
+
+ path, selected_filter = QtWidgets.QFileDialog.getSaveFileName(
+ parent=app.main_window,
+ caption="Export impact categories (AB impact-category file)",
+ directory="ab-impact-categories.xlsx",
+ filter="Excel spreadsheet (*.xlsx);;CSV pair (*.cfs.csv);; All files (*.*)",
+ )
+ if not path:
+ return
+
+ as_csv = "CSV" in selected_filter or path.lower().endswith(".cfs.csv")
+ if as_csv and path.lower().endswith(".xlsx"):
+ path = path[:-5]
+ elif not as_csv and not path.lower().endswith(".xlsx"):
+ path = path + ".xlsx"
+
+ export_ab_with_progress(method_names, path, as_csv=as_csv)
+
+
+class ExportABThread(threading.ABThread):
+ done: SignalInstance = Signal(str)
+ failed: SignalInstance = Signal(str)
+
+ method_names: list
+ path: str
+ as_csv: bool
+
+ def run_safely(self):
+ try:
+ if self.ab_cancel_requested():
+ return
+ cancel = lambda: self.ab_cancel_requested()
+ if self.as_csv:
+ cfs_path, ic_path = export_methods_ab_csv_pair(
+ self.method_names, self.path, cancel_check=cancel
+ )
+ message = f"{cfs_path}\n{ic_path}"
+ else:
+ export_methods_ab_xlsx(
+ self.method_names, self.path, cancel_check=cancel
+ )
+ message = self.path
+ except CancelledError:
+ self.request_ab_cancel()
+ return
+ except Exception as exc:
+ self.failed.emit(str(exc))
+ return
+ if self.ab_cancel_requested():
+ return
+ self.done.emit(message)
+
+
+def export_ab_with_progress(
+ method_names: Sequence[tuple],
+ path: str,
+ *,
+ as_csv: bool,
+) -> None:
+ thread = ExportABThread(app.application)
+ thread.method_names = list(method_names)
+ thread.path = path
+ thread.as_csv = as_csv
+
+ def done(message: str):
+ logger.info(
+ f"Exported {len(method_names)} impact categories to {message.replace(chr(10), ' and ')}"
+ )
+ QtWidgets.QMessageBox.information(
+ app.main_window,
+ "Export complete",
+ f"Exported {len(method_names)} impact categories to:\n{message}",
+ )
+
+ def failed(message: str):
+ QtWidgets.QMessageBox.warning(app.main_window, "Export impact categories", message)
+
+ thread.done.connect(done)
+ thread.failed.connect(failed)
+ run_thread_with_progress(
+ "Exporting impact categories",
+ thread,
+ on_cancelled=lambda: QtWidgets.QMessageBox.information(
+ app.main_window, "Export cancelled", "Export cancelled."
+ ),
+ )
diff --git a/activity_browser/app/actions/method/method_export_bw2io.py b/activity_browser/app/actions/method/method_export_bw2io.py
new file mode 100644
index 000000000..7470f23b9
--- /dev/null
+++ b/activity_browser/app/actions/method/method_export_bw2io.py
@@ -0,0 +1,131 @@
+"""Export impact categories as bw2io LCIA files."""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Sequence
+
+from qtpy import QtWidgets
+from qtpy.QtCore import Signal, SignalInstance
+
+from activity_browser import app
+from activity_browser.app import application
+from activity_browser.app.actions.base import ABAction, exception_dialogs
+from activity_browser.app.panes.impact_categories import resolve_methods_for_export
+from activity_browser.bwutils.impact_categories import (
+ CancelledError,
+ method_name_to_filename_stem,
+ raise_if_cancelled,
+)
+from activity_browser.bwutils.impact_categories.bw2io_lcia_file import (
+ export_method_bw2io_xlsx,
+ export_methods_bw2io_csv_batch,
+)
+from activity_browser.ui.core import threading
+from activity_browser.app.dialogs import run_thread_with_progress
+
+
+class MethodExportBW2IO(ABAction):
+ icon = application.style().standardIcon(QtWidgets.QStyle.SP_DialogSaveButton)
+ text = "To bw2io LCIA file (.xlsx/.csv)…"
+ tool_tip = "Export impact categories as bw2io LCIA files"
+
+ @classmethod
+ @exception_dialogs
+ def run(cls, method_names: list[tuple] | None = None):
+ method_names = resolve_methods_for_export(method_names)
+ if not method_names:
+ return
+
+ fmt, ok = QtWidgets.QInputDialog.getItem(
+ app.main_window,
+ "bw2io export format",
+ "Format:",
+ ["Excel (.xlsx, one file per impact category)", "CSV (folder + metadata.csv)"],
+ 0,
+ False,
+ )
+ if not ok:
+ return
+
+ directory = QtWidgets.QFileDialog.getExistingDirectory(
+ app.main_window,
+ "Select folder for bw2io Excel files"
+ if fmt.startswith("Excel")
+ else "Select folder for bw2io CSV files",
+ )
+ if not directory:
+ return
+
+ export_bw2io_with_progress(
+ method_names, directory, as_excel=fmt.startswith("Excel")
+ )
+
+
+class ExportBW2IOThread(threading.ABThread):
+ done: SignalInstance = Signal(str)
+ failed: SignalInstance = Signal(str)
+
+ method_names: list
+ directory: str
+ as_excel: bool
+
+ def run_safely(self):
+ try:
+ directory = Path(self.directory)
+ cancel = lambda: self.ab_cancel_requested()
+ if self.as_excel:
+ import tqdm
+
+ for name in tqdm.tqdm(
+ self.method_names,
+ desc="Exporting bw2io Excel",
+ total=len(self.method_names),
+ ):
+ raise_if_cancelled(cancel)
+ path = directory / f"{method_name_to_filename_stem(name)}.xlsx"
+ export_method_bw2io_xlsx(name, path)
+ message = (
+ f"Wrote {len(self.method_names)} Excel file(s) to:\n{directory}"
+ )
+ else:
+ written = export_methods_bw2io_csv_batch(
+ self.method_names, directory, cancel_check=cancel
+ )
+ message = f"Wrote {len(written)} file(s) to:\n{directory}"
+ except CancelledError:
+ self.request_ab_cancel()
+ return
+ except Exception as exc:
+ self.failed.emit(str(exc))
+ return
+ if self.ab_cancel_requested():
+ return
+ self.done.emit(message)
+
+
+def export_bw2io_with_progress(
+ method_names: Sequence[tuple],
+ directory: str,
+ *,
+ as_excel: bool,
+) -> None:
+ thread = ExportBW2IOThread(app.application)
+ thread.method_names = list(method_names)
+ thread.directory = directory
+ thread.as_excel = as_excel
+
+ def done(message: str):
+ QtWidgets.QMessageBox.information(app.main_window, "Export complete", message)
+
+ def failed(message: str):
+ QtWidgets.QMessageBox.warning(app.main_window, "Export impact categories", message)
+
+ thread.done.connect(done)
+ thread.failed.connect(failed)
+ run_thread_with_progress(
+ "Exporting impact categories",
+ thread,
+ on_cancelled=lambda: QtWidgets.QMessageBox.information(
+ app.main_window, "Export cancelled", "Export cancelled."
+ ),
+ )
diff --git a/activity_browser/app/actions/method/method_get_template.py b/activity_browser/app/actions/method/method_get_template.py
new file mode 100644
index 000000000..1356bd3c8
--- /dev/null
+++ b/activity_browser/app/actions/method/method_get_template.py
@@ -0,0 +1,65 @@
+"""Copy impact-category spreadsheet templates for the user."""
+from pathlib import Path
+
+from qtpy import QtWidgets
+
+from activity_browser import app
+from activity_browser.app.actions.base import ABAction, exception_dialogs
+from activity_browser.bwutils.impact_categories.templates import (
+ TEMPLATE_LABELS,
+ copy_impact_category_template,
+)
+from activity_browser.ui.icons import qicons
+
+
+class MethodGetTemplate(ABAction):
+ icon = qicons.import_db
+ text = "Get template…"
+ tool_tip = "Save an impact-category import/export starter template"
+
+ @classmethod
+ @exception_dialogs
+ def run(cls):
+ kind, ok = QtWidgets.QInputDialog.getItem(
+ app.main_window,
+ "Get impact-category template",
+ "Choose a format:\n\n"
+ "AB impact-category file = multi–impact-category (recommended default).\n"
+ "bw2io impact-category file = one impact category per CF file.",
+ list(TEMPLATE_LABELS.values()),
+ 0,
+ False,
+ )
+ if not ok or not kind:
+ return
+ # map label back to key
+ label_to_kind = {v: k for k, v in TEMPLATE_LABELS.items()}
+ key = label_to_kind[kind]
+
+ if key.endswith("csv"):
+ path = QtWidgets.QFileDialog.getExistingDirectory(
+ app.main_window,
+ "Select folder for CSV template files",
+ )
+ if not path:
+ return
+ stem = "ab-lcia" if key.startswith("ab") else "bw2io-lcia"
+ written = copy_impact_category_template(key, Path(path) / stem)
+ else:
+ suggested = "ab-lcia.xlsx" if key.startswith("ab") else "bw2io-lcia.xlsx"
+ path, _ = QtWidgets.QFileDialog.getSaveFileName(
+ app.main_window,
+ "Save impact-category template",
+ suggested,
+ "Excel spreadsheet (*.xlsx);; All files (*.*)",
+ )
+ if not path:
+ return
+ written = copy_impact_category_template(key, Path(path))
+
+ names = "\n".join(str(p) for p in written)
+ QtWidgets.QMessageBox.information(
+ app.main_window,
+ "Template saved",
+ f"Wrote template file(s):\n{names}",
+ )
diff --git a/activity_browser/app/actions/method/method_import_ab.py b/activity_browser/app/actions/method/method_import_ab.py
new file mode 100644
index 000000000..477150b99
--- /dev/null
+++ b/activity_browser/app/actions/method/method_import_ab.py
@@ -0,0 +1,478 @@
+"""Import impact categories from an AB LCIA file (.xlsx/.csv)."""
+from __future__ import annotations
+
+import csv
+from enum import Enum
+from typing import Callable, Optional
+
+from loguru import logger
+from qtpy import QtCore, QtWidgets
+from qtpy.QtCore import Signal, SignalInstance
+
+from activity_browser import app
+from activity_browser.app.actions.base import ABAction, exception_dialogs
+from activity_browser.bwutils.impact_categories import (
+ ABLCIAImporter,
+ CancelledError,
+ ConflictMode,
+ ab_csv_sibling_path,
+ apply_name_conflicts,
+ drop_unlinked_exchanges,
+ exchange_link_counts,
+ join_tuple_path,
+ load_ab_csv_pair,
+ load_ab_xlsx,
+ split_tuple_path,
+ unlinked_exchanges,
+)
+from activity_browser.mod import bw2data as bd
+from activity_browser.ui import widgets
+from activity_browser.ui.core import threading
+from activity_browser.app.dialogs import run_thread_with_progress
+from activity_browser.ui.icons import qicons
+
+
+class MethodImportAB(ABAction):
+ """Import impact categories from an AB LCIA Excel workbook or CSV pair."""
+
+ icon = qicons.import_db
+ text = "From AB LCIA file (.xlsx/.csv)…"
+ tool_tip = "Import impact categories from Activity Browser spreadsheet format"
+
+ @classmethod
+ @exception_dialogs
+ def run(cls):
+ path, _ = QtWidgets.QFileDialog.getOpenFileName(
+ parent=app.main_window,
+ caption="Import impact categories (AB impact-category file)",
+ filter=(
+ "AB LCIA (*.xlsx *.cfs.csv *.metadata.csv);;"
+ "Excel spreadsheet (*.xlsx);;"
+ "AB CSV (*.cfs.csv *.metadata.csv);;"
+ "All files (*.*)"
+ ),
+ )
+ if not path:
+ return
+
+ other = None
+ if not path.lower().endswith(".xlsx"):
+ sibling = ab_csv_sibling_path(path)
+ if sibling is None or not sibling.is_file():
+ other, _ = QtWidgets.QFileDialog.getOpenFileName(
+ parent=app.main_window,
+ caption="Select matching AB CSV sibling file",
+ filter="AB CSV (*.cfs.csv *.metadata.csv);; All files (*.*)",
+ )
+ if not other:
+ return
+
+ def after_load(data: list):
+ if not data:
+ QtWidgets.QMessageBox.warning(
+ app.main_window,
+ "Import impact categories",
+ "No impact categories found in the selected file.",
+ )
+ return
+
+ setup = MultiImportSetupDialog(data, parent=app.main_window)
+ if setup.exec_() != QtWidgets.QDialog.Accepted:
+ return
+
+ finalize_lcia_import(
+ setup.prepared_data,
+ biosphere_name=setup.biosphere_name,
+ overwrite=setup.conflict_mode == ConflictMode.OVERWRITE,
+ )
+
+ load_ab_file_with_progress(path, other_path=other, on_loaded=after_load)
+
+
+class UnlinkedDecision(str, Enum):
+ CANCEL = "cancel"
+ EXPORT = "export"
+ DROP = "drop"
+ CONTINUE = "continue"
+
+
+def export_unmatched_cfs(unmatched: list[dict], parent=None) -> None:
+ path, _ = QtWidgets.QFileDialog.getSaveFileName(
+ parent or app.main_window,
+ "Export unmatched characterization factors",
+ "unmatched-cfs.csv",
+ "CSV (*.csv);; All files (*.*)",
+ )
+ if not path:
+ return
+ if not path.lower().endswith(".csv"):
+ path = path + ".csv"
+ fieldnames = ["method", "name", "categories", "amount"]
+ with open(path, "w", newline="", encoding="utf-8") as handle:
+ writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore")
+ writer.writeheader()
+ for row in unmatched:
+ writer.writerow(
+ {
+ "method": "::".join(map(str, row.get("method", ()))),
+ "name": row.get("name", ""),
+ "categories": "::".join(map(str, row.get("categories", ()))),
+ "amount": row.get("amount", ""),
+ }
+ )
+ QtWidgets.QMessageBox.information(
+ parent or app.main_window,
+ "Unmatched CFs exported",
+ f"Wrote {len(unmatched)} rows to:\n{path}",
+ )
+
+
+def ask_unlinked_cfs(linked: int, unlinked: int, parent=None) -> UnlinkedDecision:
+ box = QtWidgets.QMessageBox(parent or app.main_window)
+ box.setWindowTitle("Characterization factor linking")
+ box.setIcon(QtWidgets.QMessageBox.Warning)
+ box.setText(f"Linked: {linked} | Unlinked: {unlinked}")
+ box.setInformativeText(
+ "Cancel import, export the unmatched list, or drop unlinked CFs and continue?"
+ )
+ cancel = box.addButton("Cancel", QtWidgets.QMessageBox.RejectRole)
+ export_btn = box.addButton("Export unmatched…", QtWidgets.QMessageBox.ActionRole)
+ drop = box.addButton("Drop unlinked", QtWidgets.QMessageBox.DestructiveRole)
+ box.setDefaultButton(cancel)
+ box.exec_()
+ clicked = box.clickedButton()
+ if clicked is export_btn:
+ return UnlinkedDecision.EXPORT
+ if clicked is drop:
+ confirm = QtWidgets.QMessageBox.question(
+ parent or app.main_window,
+ "Drop unlinked CFs",
+ "Drop all unlinked characterization factors and write the rest?",
+ QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
+ QtWidgets.QMessageBox.No,
+ )
+ return (
+ UnlinkedDecision.DROP
+ if confirm == QtWidgets.QMessageBox.Yes
+ else UnlinkedDecision.CANCEL
+ )
+ return UnlinkedDecision.CANCEL
+
+
+class MultiImportSetupDialog(QtWidgets.QDialog):
+ """Biosphere + conflict policy for multi–impact-category imports."""
+
+ biosphere_name: str
+ conflict_mode: ConflictMode
+ prepared_data: list
+
+ def __init__(self, data: list[dict], parent=None):
+ super().__init__(parent)
+ self._data = data
+ self.setWindowTitle("Import impact categories")
+
+ self.db_chooser = widgets.ABComboBox.get_database_combobox(self)
+ default_bio = bd.config.biosphere
+ idx = self.db_chooser.findText(default_bio)
+ if idx >= 0:
+ self.db_chooser.setCurrentIndex(idx)
+
+ self.conflict_skip = QtWidgets.QRadioButton("Skip existing impact categories")
+ self.conflict_overwrite = QtWidgets.QRadioButton("Overwrite existing")
+ self.conflict_rename = QtWidgets.QRadioButton("Rename conflicts with prefix")
+ self.conflict_skip.setChecked(True)
+ self.prefix_edit = QtWidgets.QLineEdit()
+ self.prefix_edit.setPlaceholderText("Namespace prefix (e.g. Import 2026)")
+ self.prefix_edit.setEnabled(False)
+ self.conflict_rename.toggled.connect(self.prefix_edit.setEnabled)
+
+ existing = set(bd.methods)
+ conflicts = [ds for ds in data if tuple(ds["name"]) in existing]
+ self._conflict_table: Optional[QtWidgets.QTableWidget] = None
+ if conflicts:
+ self._conflict_table = QtWidgets.QTableWidget(len(conflicts), 2)
+ self._conflict_table.setHorizontalHeaderLabels(
+ ["Existing name", "Import as (:: editable)"]
+ )
+ self._conflict_table.horizontalHeader().setStretchLastSection(True)
+ for row, ds in enumerate(conflicts):
+ original = join_tuple_path(ds["name"])
+ item0 = QtWidgets.QTableWidgetItem(original)
+ item0.setFlags(item0.flags() & ~QtCore.Qt.ItemIsEditable)
+ item0.setData(QtCore.Qt.UserRole, tuple(ds["name"]))
+ self._conflict_table.setItem(row, 0, item0)
+ self._conflict_table.setItem(
+ row, 1, QtWidgets.QTableWidgetItem(original)
+ )
+
+ info = QtWidgets.QLabel(
+ f"File contains {len(data)} impact categories"
+ + (f" ({len(conflicts)} name conflicts)." if conflicts else ".")
+ )
+ info.setWordWrap(True)
+
+ buttons = QtWidgets.QDialogButtonBox(
+ QtWidgets.QDialogButtonBox.Ok | QtWidgets.QDialogButtonBox.Cancel
+ )
+ buttons.accepted.connect(self.accept)
+ buttons.rejected.connect(self.reject)
+
+ layout = QtWidgets.QVBoxLayout(self)
+ layout.addWidget(info)
+ layout.addWidget(QtWidgets.QLabel("Biosphere database:"))
+ layout.addWidget(self.db_chooser)
+ layout.addWidget(QtWidgets.QLabel("If names already exist:"))
+ layout.addWidget(self.conflict_skip)
+ layout.addWidget(self.conflict_overwrite)
+ layout.addWidget(self.conflict_rename)
+ layout.addWidget(self.prefix_edit)
+ if self._conflict_table is not None:
+ layout.addWidget(
+ QtWidgets.QLabel(
+ "Per-conflict rename (optional; overrides bulk policy for edited rows):"
+ )
+ )
+ layout.addWidget(self._conflict_table)
+ layout.addWidget(buttons)
+
+ def _table_renames(self) -> dict[tuple, tuple]:
+ renames: dict[tuple, tuple] = {}
+ if self._conflict_table is None:
+ return renames
+ for row in range(self._conflict_table.rowCount()):
+ original = self._conflict_table.item(row, 0).data(QtCore.Qt.UserRole)
+ target_text = (self._conflict_table.item(row, 1).text() or "").strip()
+ target = split_tuple_path(target_text)
+ if target and target != original:
+ renames[tuple(original)] = target
+ return renames
+
+ def accept(self):
+ if self.conflict_overwrite.isChecked():
+ confirm = QtWidgets.QMessageBox.question(
+ self,
+ "Overwrite impact categories",
+ "Overwrite all conflicting impact categories?",
+ QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
+ QtWidgets.QMessageBox.No,
+ )
+ if confirm != QtWidgets.QMessageBox.Yes:
+ return
+ mode = ConflictMode.OVERWRITE
+ prefix = None
+ elif self.conflict_rename.isChecked():
+ mode = ConflictMode.RENAME_PREFIX
+ prefix = self.prefix_edit.text().strip()
+ if not prefix:
+ QtWidgets.QMessageBox.warning(
+ self,
+ "Rename conflicts",
+ "Please enter a namespace prefix.",
+ )
+ return
+ else:
+ mode = ConflictMode.SKIP
+ prefix = None
+
+ self.biosphere_name = self.db_chooser.currentText()
+ self.conflict_mode = mode
+ self.prepared_data = apply_name_conflicts(
+ self._data,
+ set(bd.methods),
+ mode=mode,
+ prefix=prefix,
+ renames=self._table_renames(),
+ )
+ if not self.prepared_data:
+ QtWidgets.QMessageBox.information(
+ self,
+ "Import impact categories",
+ "Nothing left to import after applying the conflict policy.",
+ )
+ return
+ super().accept()
+
+
+def _cancel_check(thread: threading.ABThread) -> bool:
+ return thread.ab_cancel_requested()
+
+
+def notify_import_cancelled(title: str = "Cancelled") -> None:
+ QtWidgets.QMessageBox.information(
+ app.main_window,
+ title,
+ "Operation cancelled. No impact categories were written to the project.",
+ )
+
+
+class LoadABFileThread(threading.ABThread):
+ loaded: SignalInstance = Signal(object)
+ failed: SignalInstance = Signal(str)
+
+ path: str
+ other_path: Optional[str] = None
+
+ def run_safely(self):
+ path = self.path
+ try:
+ if self.ab_cancel_requested():
+ return
+ if path.lower().endswith(".xlsx"):
+ data = load_ab_xlsx(path)
+ else:
+ data = load_ab_csv_pair(path, other_path=self.other_path)
+ except (ValueError, FileNotFoundError) as exc:
+ self.failed.emit(str(exc))
+ return
+ if self.ab_cancel_requested():
+ return
+ self.loaded.emit(data)
+
+
+class LinkLCIAThread(threading.ABThread):
+ linked: SignalInstance = Signal(object)
+
+ data: list
+ biosphere_name: str
+
+ def run_safely(self):
+ try:
+ importer = ABLCIAImporter(self.data, biosphere=self.biosphere_name)
+ importer.apply_strategies(cancel_check=lambda: _cancel_check(self))
+ except CancelledError:
+ self.request_ab_cancel()
+ return
+ if self.ab_cancel_requested():
+ return
+ self.linked.emit(importer)
+
+
+class WriteLCIAThread(threading.ABThread):
+ written: SignalInstance = Signal(int)
+ failed: SignalInstance = Signal(str)
+
+ importer: ABLCIAImporter
+ overwrite: bool
+ biosphere_name: str
+
+ def run_safely(self):
+ try:
+ self.importer.write_methods(
+ overwrite=self.overwrite,
+ verbose=False,
+ cancel_check=lambda: _cancel_check(self),
+ )
+ except CancelledError:
+ self.request_ab_cancel()
+ return
+ except ValueError as exc:
+ self.failed.emit(str(exc))
+ return
+ if self.ab_cancel_requested():
+ return
+ logger.info(
+ f"Imported {len(self.importer.data)} impact categories "
+ f"(biosphere={self.biosphere_name})"
+ )
+ self.written.emit(len(self.importer.data))
+
+
+def load_ab_file_with_progress(
+ path: str,
+ *,
+ other_path: Optional[str] = None,
+ on_loaded: Callable[[list], None],
+) -> None:
+ thread = LoadABFileThread(app.application)
+ thread.path = path
+ thread.other_path = other_path
+
+ def _fail(message: str):
+ QtWidgets.QMessageBox.warning(
+ app.main_window, "Import impact categories", message
+ )
+
+ thread.failed.connect(_fail)
+ thread.loaded.connect(on_loaded)
+ run_thread_with_progress(
+ "Loading impact categories",
+ thread,
+ on_cancelled=lambda: notify_import_cancelled("Import cancelled"),
+ )
+
+
+def finalize_lcia_import(
+ data: list[dict],
+ *,
+ biosphere_name: str,
+ overwrite: bool,
+ parent=None,
+) -> None:
+ """Apply strategies (with progress), gate on unlinked CFs, then write."""
+ parent = parent or app.main_window
+ link_thread = LinkLCIAThread(app.application)
+ link_thread.data = data
+ link_thread.biosphere_name = biosphere_name
+
+ def after_link(importer: ABLCIAImporter):
+ linked, unlinked = exchange_link_counts(importer.data)
+ unmatched = unlinked_exchanges(importer.data)
+
+ if unlinked:
+ decision = ask_unlinked_cfs(linked, unlinked, parent=parent)
+ if decision == UnlinkedDecision.CANCEL:
+ return
+ if decision == UnlinkedDecision.EXPORT:
+ export_unmatched_cfs(unmatched, parent=parent)
+ return
+ if decision == UnlinkedDecision.DROP:
+ importer.data = drop_unlinked_exchanges(importer.data)
+ else:
+ QtWidgets.QMessageBox.information(
+ parent,
+ "Ready to import",
+ f"Linked: {linked} | Unlinked: 0\n\n"
+ f"Writing {len(importer.data)} impact categories.",
+ )
+
+ write_thread = WriteLCIAThread(app.application)
+ write_thread.importer = importer
+ write_thread.overwrite = overwrite
+ write_thread.biosphere_name = biosphere_name
+
+ def after_write(count: int):
+ QtWidgets.QMessageBox.information(
+ parent,
+ "Import complete",
+ f"Imported {count} impact categories.",
+ )
+
+ def write_failed(message: str):
+ QtWidgets.QMessageBox.warning(parent, "Import impact categories", message)
+
+ write_thread.written.connect(after_write)
+ write_thread.failed.connect(write_failed)
+
+ def write_cancelled():
+ if overwrite:
+ QtWidgets.QMessageBox.information(
+ parent,
+ "Import cancelled",
+ "Import cancelled. Newly created impact categories from this run "
+ "were removed. Any categories already overwritten may remain updated.",
+ )
+ else:
+ notify_import_cancelled("Import cancelled")
+
+ run_thread_with_progress(
+ "Writing impact categories",
+ write_thread,
+ on_cancelled=write_cancelled,
+ )
+
+ link_thread.linked.connect(after_link)
+ run_thread_with_progress(
+ "Linking characterization factors",
+ link_thread,
+ on_cancelled=lambda: notify_import_cancelled("Import cancelled"),
+ )
diff --git a/activity_browser/app/actions/method/method_import_bw2io.py b/activity_browser/app/actions/method/method_import_bw2io.py
new file mode 100644
index 000000000..fc0b5117b
--- /dev/null
+++ b/activity_browser/app/actions/method/method_import_bw2io.py
@@ -0,0 +1,456 @@
+"""Import impact categories from bw2io LCIA files."""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Callable, Sequence
+
+from qtpy import QtCore, QtWidgets
+from qtpy.QtCore import Signal, SignalInstance
+
+from activity_browser import app
+from activity_browser.app.actions.base import ABAction, exception_dialogs
+from activity_browser.app.actions.method.method_import_ab import (
+ MultiImportSetupDialog,
+ finalize_lcia_import,
+ notify_import_cancelled,
+)
+from activity_browser.bwutils.impact_categories import (
+ ConflictMode,
+ join_tuple_path,
+ split_tuple_path,
+)
+from activity_browser.bwutils.impact_categories.bw2io_lcia_file import (
+ load_bw2io_lcia_file,
+ read_bw2io_metadata_csv,
+ read_bw2io_metadata_xlsx,
+)
+from activity_browser.mod import bw2data as bd
+from activity_browser.ui import widgets
+from activity_browser.ui.core import threading
+from activity_browser.app.dialogs import run_thread_with_progress
+from activity_browser.ui.icons import qicons
+
+
+class MethodImportBW2IO(ABAction):
+ """Import one or more bw2io impact-category files."""
+
+ icon = qicons.import_db
+ text = "From bw2io LCIA file (.xlsx/.csv)…"
+ tool_tip = "Import one or more bw2io LCIA Excel or CSV files"
+
+ @classmethod
+ @exception_dialogs
+ def run(cls):
+ paths, _ = QtWidgets.QFileDialog.getOpenFileNames(
+ parent=app.main_window,
+ caption="Import bw2io LCIA file(s)",
+ filter="LCIA (*.xlsx *.csv);;Excel (*.xlsx);;CSV (*.csv);;All files (*.*)",
+ )
+ if not paths:
+ return
+
+ path_objs = [Path(p) for p in paths]
+ if len(path_objs) == 1:
+ cls._import_single(path_objs[0])
+ else:
+ cls._import_many(path_objs)
+
+ @classmethod
+ def _import_single(cls, path: Path):
+ prefill = _prefill_for_bw2io_path(path)
+ dialog = BW2IOMetadataDialog(prefill, parent=app.main_window)
+ if dialog.exec_() != QtWidgets.QDialog.Accepted:
+ return
+
+ name = split_tuple_path(dialog.method_path)
+ if not name:
+ QtWidgets.QMessageBox.warning(
+ app.main_window,
+ "Import bw2io LCIA",
+ "Method name is required (use :: between parts).",
+ )
+ return
+
+ overwrite = False
+ if name in bd.methods:
+ conflict = BW2IOLciaFileConflictDialog(name, parent=app.main_window)
+ if conflict.exec_() != QtWidgets.QDialog.Accepted:
+ return
+ name = conflict.result_name
+ overwrite = conflict.overwrite
+
+ bio = BiospherePickDialog(parent=app.main_window)
+ if bio.exec_() != QtWidgets.QDialog.Accepted:
+ return
+
+ def after_load(data: list):
+ finalize_lcia_import(
+ data,
+ biosphere_name=bio.biosphere_name,
+ overwrite=overwrite,
+ )
+
+ load_bw2io_file_with_progress(
+ path,
+ name=name,
+ unit=dialog.unit,
+ description=dialog.description,
+ on_loaded=after_load,
+ )
+
+ @classmethod
+ def _import_many(cls, paths: list[Path]):
+ rows = []
+ for path in paths:
+ prefill = _prefill_for_bw2io_path(path)
+ rows.append(
+ {
+ "path": path,
+ "method": prefill.get("method") or "",
+ "unit": prefill.get("unit") or "",
+ "description": prefill.get("description") or "",
+ }
+ )
+
+ review = BW2IOBatchMetadataDialog(rows, parent=app.main_window)
+ if review.exec_() != QtWidgets.QDialog.Accepted:
+ return
+
+ def after_load(data: list):
+ if not data:
+ QtWidgets.QMessageBox.warning(
+ app.main_window,
+ "Import bw2io LCIA",
+ "No impact categories found in the selected files.",
+ )
+ return
+ setup = MultiImportSetupDialog(data, parent=app.main_window)
+ if setup.exec_() != QtWidgets.QDialog.Accepted:
+ return
+ finalize_lcia_import(
+ setup.prepared_data,
+ biosphere_name=setup.biosphere_name,
+ overwrite=setup.conflict_mode == ConflictMode.OVERWRITE,
+ )
+
+ load_bw2io_files_with_progress(review.rows, on_loaded=after_load)
+
+
+class BW2IOLciaFileConflictDialog(QtWidgets.QDialog):
+ """Overwrite / edit name / cancel when a bw2io impact-category file name already exists."""
+
+ result_name: tuple
+ overwrite: bool
+
+ def __init__(self, name: tuple, parent=None):
+ super().__init__(parent)
+ self._original = tuple(name)
+ self.setWindowTitle("Impact category already exists")
+ self.name_edit = QtWidgets.QLineEdit(join_tuple_path(name))
+ info = QtWidgets.QLabel(
+ f"{join_tuple_path(name)} already exists in this project."
+ )
+ info.setWordWrap(True)
+ buttons = QtWidgets.QDialogButtonBox()
+ self.overwrite_btn = buttons.addButton(
+ "Overwrite", QtWidgets.QDialogButtonBox.AcceptRole
+ )
+ self.use_name_btn = buttons.addButton(
+ "Use edited name", QtWidgets.QDialogButtonBox.AcceptRole
+ )
+ buttons.addButton(QtWidgets.QDialogButtonBox.Cancel)
+ buttons.rejected.connect(self.reject)
+ self.overwrite_btn.clicked.connect(self._accept_overwrite)
+ self.use_name_btn.clicked.connect(self._accept_edit)
+
+ layout = QtWidgets.QVBoxLayout(self)
+ layout.addWidget(info)
+ layout.addWidget(QtWidgets.QLabel("Name (:: parts):"))
+ layout.addWidget(self.name_edit)
+ layout.addWidget(buttons)
+
+ def _accept_overwrite(self):
+ confirm = QtWidgets.QMessageBox.question(
+ self,
+ "Overwrite impact category",
+ f"Overwrite {join_tuple_path(self._original)}?",
+ QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
+ QtWidgets.QMessageBox.No,
+ )
+ if confirm != QtWidgets.QMessageBox.Yes:
+ return
+ self.result_name = self._original
+ self.overwrite = True
+ self.accept()
+
+ def _accept_edit(self):
+ name = split_tuple_path(self.name_edit.text().strip())
+ if not name:
+ QtWidgets.QMessageBox.warning(self, "Edit name", "Name cannot be empty.")
+ return
+ if name in bd.methods and name != self._original:
+ QtWidgets.QMessageBox.warning(
+ self,
+ "Edit name",
+ "That name already exists. Choose another or overwrite the original.",
+ )
+ return
+ self.result_name = name
+ self.overwrite = name == self._original and name in bd.methods
+ if self.overwrite:
+ self._accept_overwrite()
+ return
+ self.accept()
+
+
+class BiospherePickDialog(QtWidgets.QDialog):
+ biosphere_name: str
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ self.setWindowTitle("Choose biosphere database")
+ self.db_chooser = widgets.ABComboBox.get_database_combobox(self)
+ default_bio = bd.config.biosphere
+ idx = self.db_chooser.findText(default_bio)
+ if idx >= 0:
+ self.db_chooser.setCurrentIndex(idx)
+ buttons = QtWidgets.QDialogButtonBox(
+ QtWidgets.QDialogButtonBox.Ok | QtWidgets.QDialogButtonBox.Cancel
+ )
+ buttons.accepted.connect(self.accept)
+ buttons.rejected.connect(self.reject)
+ layout = QtWidgets.QVBoxLayout(self)
+ layout.addWidget(self.db_chooser)
+ layout.addWidget(buttons)
+
+ def accept(self):
+ self.biosphere_name = self.db_chooser.currentText()
+ super().accept()
+
+
+class BW2IOMetadataDialog(QtWidgets.QDialog):
+ method_path: str
+ unit: str
+ description: str
+
+ def __init__(self, prefill: dict, parent=None):
+ super().__init__(parent)
+ self.setWindowTitle("bw2io impact category metadata")
+ self.method_edit = QtWidgets.QLineEdit(prefill.get("method") or "")
+ self.method_edit.setPlaceholderText("My method::climate change::GWP100")
+ self.unit_edit = QtWidgets.QLineEdit(prefill.get("unit") or "")
+ self.description_edit = QtWidgets.QPlainTextEdit(
+ prefill.get("description") or ""
+ )
+ buttons = QtWidgets.QDialogButtonBox(
+ QtWidgets.QDialogButtonBox.Ok | QtWidgets.QDialogButtonBox.Cancel
+ )
+ buttons.accepted.connect(self.accept)
+ buttons.rejected.connect(self.reject)
+ layout = QtWidgets.QFormLayout(self)
+ layout.addRow("Method (:: parts):", self.method_edit)
+ layout.addRow("Unit:", self.unit_edit)
+ layout.addRow("Description:", self.description_edit)
+ layout.addRow(buttons)
+
+ def accept(self):
+ self.method_path = self.method_edit.text().strip()
+ self.unit = self.unit_edit.text().strip()
+ self.description = self.description_edit.toPlainText().strip()
+ super().accept()
+
+
+class BW2IOBatchMetadataDialog(QtWidgets.QDialog):
+ """Edit method / unit / description for several bw2io impact-category files."""
+
+ rows: list[dict]
+
+ def __init__(self, rows: list[dict], parent=None):
+ super().__init__(parent)
+ self._rows = [dict(r) for r in rows]
+ self.setWindowTitle("bw2io impact category metadata")
+ self.resize(820, min(120 + 28 * len(rows), 520))
+
+ self.table = QtWidgets.QTableWidget(len(rows), 4)
+ self.table.setHorizontalHeaderLabels(
+ ["File", "Method (:: parts)", "Unit", "Description"]
+ )
+ self.table.horizontalHeader().setStretchLastSection(True)
+ self.table.horizontalHeader().setSectionResizeMode(
+ 0, QtWidgets.QHeaderView.ResizeToContents
+ )
+ for i, row in enumerate(self._rows):
+ file_item = QtWidgets.QTableWidgetItem(Path(row["path"]).name)
+ file_item.setFlags(file_item.flags() & ~QtCore.Qt.ItemIsEditable)
+ self.table.setItem(i, 0, file_item)
+ self.table.setItem(i, 1, QtWidgets.QTableWidgetItem(row.get("method") or ""))
+ self.table.setItem(i, 2, QtWidgets.QTableWidgetItem(row.get("unit") or ""))
+ self.table.setItem(
+ i, 3, QtWidgets.QTableWidgetItem(row.get("description") or "")
+ )
+
+ info = QtWidgets.QLabel(
+ f"Review metadata for {len(rows)} bw2io file(s). "
+ "Method names use :: between Brightway parts."
+ )
+ info.setWordWrap(True)
+ buttons = QtWidgets.QDialogButtonBox(
+ QtWidgets.QDialogButtonBox.Ok | QtWidgets.QDialogButtonBox.Cancel
+ )
+ buttons.accepted.connect(self.accept)
+ buttons.rejected.connect(self.reject)
+ layout = QtWidgets.QVBoxLayout(self)
+ layout.addWidget(info)
+ layout.addWidget(self.table)
+ layout.addWidget(buttons)
+
+ def accept(self):
+ seen: set[tuple] = set()
+ out: list[dict] = []
+ for i, row in enumerate(self._rows):
+ method_text = (self.table.item(i, 1).text() or "").strip()
+ name = split_tuple_path(method_text)
+ if not name:
+ QtWidgets.QMessageBox.warning(
+ self,
+ "bw2io metadata",
+ f"Row {i + 1} ({Path(row['path']).name}): method name is required.",
+ )
+ return
+ if name in seen:
+ QtWidgets.QMessageBox.warning(
+ self,
+ "bw2io metadata",
+ f"Duplicate method name in the selection:\n{join_tuple_path(name)}",
+ )
+ return
+ seen.add(name)
+ out.append(
+ {
+ "path": row["path"],
+ "name": name,
+ "unit": (self.table.item(i, 2).text() or "").strip(),
+ "description": (self.table.item(i, 3).text() or "").strip(),
+ }
+ )
+ self.rows = out
+ super().accept()
+
+
+def _prefill_for_bw2io_path(path: Path) -> dict[str, str]:
+ prefill = {"method": "", "unit": "", "description": ""}
+ if path.suffix.lower() in {".xlsx", ".xls"}:
+ prefill.update(read_bw2io_metadata_xlsx(path) or {})
+ else:
+ sibling = path.with_name("metadata.csv")
+ row = read_bw2io_metadata_csv(sibling, cf_filename=path.name)
+ if row:
+ prefill.update(row)
+ return prefill
+
+
+class LoadBW2IOFileThread(threading.ABThread):
+ loaded: SignalInstance = Signal(object)
+ failed: SignalInstance = Signal(str)
+
+ path: Path
+ name: tuple
+ unit: str
+ description: str
+
+ def run_safely(self):
+ try:
+ if self.ab_cancel_requested():
+ return
+ data = load_bw2io_lcia_file(
+ self.path,
+ name=self.name,
+ unit=self.unit,
+ description=self.description,
+ )
+ except (ValueError, FileNotFoundError) as exc:
+ self.failed.emit(str(exc))
+ return
+ if self.ab_cancel_requested():
+ return
+ self.loaded.emit(data)
+
+
+class LoadBW2IOFilesThread(threading.ABThread):
+ """Load several bw2io impact-category files into one importer-shaped list."""
+
+ loaded: SignalInstance = Signal(object)
+ failed: SignalInstance = Signal(str)
+
+ specs: list
+
+ def run_safely(self):
+ import tqdm
+
+ combined: list = []
+ try:
+ for spec in tqdm.tqdm(
+ self.specs, desc="Loading bw2io files", total=len(self.specs)
+ ):
+ if self.ab_cancel_requested():
+ return
+ combined.extend(
+ load_bw2io_lcia_file(
+ Path(spec["path"]),
+ name=tuple(spec["name"]),
+ unit=spec.get("unit") or "",
+ description=spec.get("description") or "",
+ )
+ )
+ except (ValueError, FileNotFoundError) as exc:
+ self.failed.emit(str(exc))
+ return
+ if self.ab_cancel_requested():
+ return
+ self.loaded.emit(combined)
+
+
+def load_bw2io_file_with_progress(
+ path: Path,
+ *,
+ name: tuple,
+ unit: str,
+ description: str,
+ on_loaded: Callable[[list], None],
+) -> None:
+ thread = LoadBW2IOFileThread(app.application)
+ thread.path = path
+ thread.name = name
+ thread.unit = unit
+ thread.description = description
+
+ def _fail(message: str):
+ QtWidgets.QMessageBox.warning(app.main_window, "Import bw2io LCIA", message)
+
+ thread.failed.connect(_fail)
+ thread.loaded.connect(on_loaded)
+ run_thread_with_progress(
+ "Loading impact category file",
+ thread,
+ on_cancelled=lambda: notify_import_cancelled("Import cancelled"),
+ )
+
+
+def load_bw2io_files_with_progress(
+ specs: Sequence[dict],
+ *,
+ on_loaded: Callable[[list], None],
+) -> None:
+ thread = LoadBW2IOFilesThread(app.application)
+ thread.specs = list(specs)
+
+ def _fail(message: str):
+ QtWidgets.QMessageBox.warning(app.main_window, "Import bw2io LCIA", message)
+
+ thread.failed.connect(_fail)
+ thread.loaded.connect(on_loaded)
+ run_thread_with_progress(
+ "Loading impact category files",
+ thread,
+ on_cancelled=lambda: notify_import_cancelled("Import cancelled"),
+ )
diff --git a/activity_browser/app/actions/method/importer/method_importer_ecoinvent.py b/activity_browser/app/actions/method/method_import_ecoinvent.py
similarity index 84%
rename from activity_browser/app/actions/method/importer/method_importer_ecoinvent.py
rename to activity_browser/app/actions/method/method_import_ecoinvent.py
index 2300f0d80..a7b5afc1a 100644
--- a/activity_browser/app/actions/method/importer/method_importer_ecoinvent.py
+++ b/activity_browser/app/actions/method/method_import_ecoinvent.py
@@ -7,18 +7,19 @@
from activity_browser.mod import bw2data as bd
from activity_browser.app.actions.base import ABAction, exception_dialogs
from activity_browser.ui import icons, widgets
-from activity_browser.bwutils.io.ecoinvent_lcia_importer import EcoinventLCIAImporter
+from activity_browser.ui.dialogs import ABProgressDialog
+from activity_browser.bwutils.impact_categories import EcoinventLCIAImporter
from activity_browser.ui.core import threading
-class MethodImporterEcoinvent(ABAction):
+class MethodImportEcoinvent(ABAction):
"""ABAction to import methods from ecoinvent"""
icon = icons.qicons.import_db
- text = "Import from ecoinvent excel..."
- tool_tip = "Import methods from ecoinvent excel format"
+ text = "From ecoinvent Excel…"
+ tool_tip = "Import impact categories from an ecoinvent LCIA Implementation Excel workbook"
@classmethod
@exception_dialogs
@@ -38,7 +39,7 @@ def run(cls):
extract_thread.loaded.connect(cls.write_database)
# show progress dialog for importing the excel
- progress_dialog = widgets.ABProgressDialog.get_connected_dialog("Importing Database")
+ progress_dialog = ABProgressDialog.get_connected_dialog("Importing Database")
extract_thread.finished.connect(progress_dialog.deleteLater)
extract_thread.start()
@@ -57,7 +58,7 @@ def write_database(importer: EcoinventLCIAImporter):
importer_thread.prepend = import_dialog.prepend
# setup a progress dialog
- progress_dialog = widgets.ABProgressDialog.get_connected_dialog("Importing Impact Categories")
+ progress_dialog = ABProgressDialog.get_connected_dialog("Importing Impact Categories")
importer_thread.finished.connect(progress_dialog.deleteLater)
progress_dialog.show()
@@ -75,7 +76,10 @@ def __init__(self, importer: EcoinventLCIAImporter, parent=None):
self.setWindowTitle("Import methods from ecoinvent Excel")
self.db_chooser = widgets.ABComboBox.get_database_combobox(self)
- self.button_comp = composites.HorizontalButtonsComposite("Cancel", "*OK")
+ self.buttons = QtWidgets.QDialogButtonBox(
+ QtWidgets.QDialogButtonBox.Ok | QtWidgets.QDialogButtonBox.Cancel
+ )
+ self.ok_button = self.buttons.button(QtWidgets.QDialogButtonBox.Ok)
self.info = QtWidgets.QLabel()
self.info.setWordWrap(True)
@@ -88,8 +92,8 @@ def __init__(self, importer: EcoinventLCIAImporter, parent=None):
self.prepend_textbox.textChanged.connect(self.check_overwrite)
# Connect the necessary signals
- self.button_comp["OK"].clicked.connect(self.accept)
- self.button_comp["Cancel"].clicked.connect(self.reject)
+ self.buttons.accepted.connect(self.accept)
+ self.buttons.rejected.connect(self.reject)
# Create final layout
layout = QtWidgets.QVBoxLayout()
@@ -98,7 +102,7 @@ def __init__(self, importer: EcoinventLCIAImporter, parent=None):
layout.addWidget(self.prepend_label)
layout.addWidget(self.prepend_textbox)
layout.addWidget(self.info)
- layout.addWidget(self.button_comp)
+ layout.addWidget(self.buttons)
# Set the dialog layout
self.setLayout(layout)
@@ -127,7 +131,7 @@ def check_overwrite(self, prepend=None) -> int:
def validate(self):
"""Validate the user input and enable the OK button if all is clear"""
valid = True
- self.button_comp["OK"].setEnabled(valid)
+ self.ok_button.setEnabled(valid)
def accept(self):
"""Correctly set the dialog's attributes for further use in the action"""
diff --git a/activity_browser/app/dialogs/__init__.py b/activity_browser/app/dialogs/__init__.py
index 5a337f238..def17ce93 100644
--- a/activity_browser/app/dialogs/__init__.py
+++ b/activity_browser/app/dialogs/__init__.py
@@ -1,3 +1,11 @@
from .import_preview_dialog import ImportPreviewDialog
from .node_select_dialog import NodeSelectDialog
from .database_select_dialog import DatabaseSelectDialog
+from .thread_progress import run_thread_with_progress
+
+__all__ = [
+ "DatabaseSelectDialog",
+ "ImportPreviewDialog",
+ "NodeSelectDialog",
+ "run_thread_with_progress",
+]
diff --git a/activity_browser/app/dialogs/thread_progress.py b/activity_browser/app/dialogs/thread_progress.py
new file mode 100644
index 000000000..ba7f2f5c6
--- /dev/null
+++ b/activity_browser/app/dialogs/thread_progress.py
@@ -0,0 +1,38 @@
+"""App-layer sticky-cancel progress starter for ABThread jobs."""
+from __future__ import annotations
+
+from activity_browser.ui.dialogs import ABProgressDialog
+
+
+def run_thread_with_progress(title: str, thread, *, on_cancelled=None) -> None:
+ """
+ Show a cancellable ABProgressDialog for an ABThread and start it.
+
+ Sticky cancel: closing after success must not look like a user cancel.
+ ``on_cancelled`` runs only if the thread reported ``ab_cancel_requested``.
+ """
+ progress = ABProgressDialog.get_connected_dialog(title, cancellable=True)
+ thread.connect_progress_dialog(progress)
+
+ def request_cancel():
+ thread.request_ab_cancel()
+ progress.mark_cancelled()
+ progress.setLabelText("Cancelling…")
+
+ progress.canceled.connect(request_cancel)
+
+ def cleanup():
+ was_cancelled = thread.ab_cancel_requested()
+ try:
+ progress.canceled.disconnect(request_cancel)
+ except (RuntimeError, TypeError):
+ pass
+ progress.detach()
+ progress.close()
+ progress.deleteLater()
+ if was_cancelled and on_cancelled is not None:
+ on_cancelled()
+
+ thread.finished.connect(cleanup)
+ progress.show()
+ thread.start()
diff --git a/activity_browser/app/menu_bar.py b/activity_browser/app/menu_bar.py
index 9d8297d00..a4ee90f7f 100644
--- a/activity_browser/app/menu_bar.py
+++ b/activity_browser/app/menu_bar.py
@@ -89,7 +89,7 @@ def __init__(self, parent=None) -> None:
class ImpactCategoriesMenu(QtWidgets.QMenu):
- """Impact category (LCIA method) import."""
+ """Impact category (LCIA method) import/export."""
def __init__(self, parent=None) -> None:
super().__init__(parent)
@@ -97,7 +97,12 @@ def __init__(self, parent=None) -> None:
self.setTitle("&Impact categories")
self.import_menu = ImportICMenu(self)
+ self.export_menu = ExportICMenu(self)
+ self.get_template_action = app.actions.MethodGetTemplate.get_QAction(parent=self)
+
self.addMenu(self.import_menu)
+ self.addMenu(self.export_menu)
+ self.addAction(self.get_template_action)
class ProjectNewMenu(QtWidgets.QMenu):
@@ -312,8 +317,21 @@ def __init__(self, parent=None) -> None:
self.setTitle("Import")
self.setIcon(qicons.import_db)
- self.import_from_ei_excel_action = app.actions.MethodImporterEcoinvent.get_QAction(parent=self)
- self.import_from_bw2io_action = app.actions.MethodImporterBW2IO.get_QAction(parent=self)
+ self.import_from_ab_action = app.actions.MethodImportAB.get_QAction(parent=self)
+ self.import_from_bw2io_file_action = app.actions.MethodImportBW2IO.get_QAction(parent=self)
+ self.import_from_ei_excel_action = app.actions.MethodImportEcoinvent.get_QAction(parent=self)
+ self.addAction(self.import_from_ab_action)
+ self.addAction(self.import_from_bw2io_file_action)
self.addAction(self.import_from_ei_excel_action)
- self.addAction(self.import_from_bw2io_action)
+
+
+class ExportICMenu(QtWidgets.QMenu):
+ def __init__(self, parent=None) -> None:
+ super().__init__(parent=parent)
+ self.setTitle("Export")
+
+ self.export_to_ab_action = app.actions.MethodExportAB.get_QAction(parent=self)
+ self.export_to_bw2io_action = app.actions.MethodExportBW2IO.get_QAction(parent=self)
+ self.addAction(self.export_to_ab_action)
+ self.addAction(self.export_to_bw2io_action)
diff --git a/activity_browser/app/panes/impact_categories.py b/activity_browser/app/panes/impact_categories.py
index f909a8581..b314534b4 100644
--- a/activity_browser/app/panes/impact_categories.py
+++ b/activity_browser/app/panes/impact_categories.py
@@ -4,10 +4,66 @@
import bw2data as bd
import pandas as pd
+from typing import List, Optional
+
from activity_browser import app, app
from activity_browser.ui import widgets, core, delegates
+def live_impact_category_selection() -> Optional[List[tuple]]:
+ """
+ Return selected impact-category names from the Impact categories pane.
+
+ On any failure (no main window, pane missing, etc.) return ``None`` so
+ callers can fall back to the empty-selection / export-all prompt.
+ """
+ try:
+ window = app.main_window
+ if window is None:
+ return None
+ for pane in window.panes():
+ if isinstance(pane, ImpactCategoriesPane):
+ selected = pane.view.selected_impact_categories
+ return list(selected) if selected else None
+ return None
+ except Exception:
+ return None
+
+
+def resolve_methods_for_export(
+ method_names: Optional[List[tuple]] = None,
+) -> Optional[List[tuple]]:
+ """
+ Resolve which impact categories to export.
+
+ If ``method_names`` is None, use the live pane selection. When nothing is
+ selected, ask whether to export all. Returns ``None`` if the user cancels
+ or the project has no methods.
+ """
+ if method_names is None:
+ method_names = live_impact_category_selection()
+ if not method_names:
+ choice = QtWidgets.QMessageBox.question(
+ app.main_window,
+ "Export impact categories",
+ "No impact categories are selected.\n\n"
+ "Export all impact categories in this project?",
+ QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No,
+ QtWidgets.QMessageBox.No,
+ )
+ if choice != QtWidgets.QMessageBox.Yes:
+ return None
+ method_names = list(bd.methods)
+ if not method_names:
+ QtWidgets.QMessageBox.information(
+ app.main_window,
+ "Export impact categories",
+ "There are no impact categories to export.",
+ )
+ return None
+ return list(method_names)
+
+
class ImpactCategoriesPane(widgets.ABAbstractPane):
title = "Impact Categories"
unique = True
@@ -51,16 +107,19 @@ def sync(self):
self.model.set_dataframe(df, group=["_method_name"])
def build_df(self):
- df = pd.DataFrame(bd.methods.values())
- df["_method_name"] = bd.methods.keys()
-
- df["name"] = df["_method_name"].apply(lambda x: x[-1])
-
cols = ["name", "unit", "num_cfs", "_method_name"]
-
- if df.empty:
+ if not bd.methods:
return pd.DataFrame(columns=cols)
+ df = pd.DataFrame(list(bd.methods.values()))
+ df["_method_name"] = list(bd.methods.keys())
+ df["name"] = df["_method_name"].apply(lambda x: x[-1] if x else "")
+ if "unit" not in df.columns:
+ df["unit"] = ""
+ if "num_cfs" not in df.columns:
+ df["num_cfs"] = 0
+ else:
+ df["num_cfs"] = df["num_cfs"].fillna(0)
return df[cols]
@@ -90,6 +149,25 @@ class ContextMenu(widgets.ABMenu):
text="Rename impact category",
enable=len(p.selected_impact_categories) == 1
),
+ lambda m: m.addSeparator(),
+ lambda m, p: m.addMenu(ImpactCategoriesView.ExportContextMenu(parent=p)),
+ ]
+
+ class ExportContextMenu(widgets.ABMenu):
+ menuSetup = [
+ lambda m: m.setTitle("Export"),
+ lambda m, p: m.add(
+ app.actions.MethodExportAB,
+ p.selected_impact_categories,
+ text="To AB LCIA file (.xlsx/.csv)…",
+ enable=len(p.selected_impact_categories) > 0,
+ ),
+ lambda m, p: m.add(
+ app.actions.MethodExportBW2IO,
+ p.selected_impact_categories,
+ text="To bw2io LCIA file (.xlsx/.csv)…",
+ enable=len(p.selected_impact_categories) > 0,
+ ),
]
@property
diff --git a/activity_browser/bwutils/impact_categories/__init__.py b/activity_browser/bwutils/impact_categories/__init__.py
new file mode 100644
index 000000000..7687e283a
--- /dev/null
+++ b/activity_browser/bwutils/impact_categories/__init__.py
@@ -0,0 +1,71 @@
+"""Impact category (LCIA method) interchange helpers."""
+
+from .ab_lcia_file import (
+ CFS_COLUMNS,
+ CFS_SHEET,
+ CFS_SUFFIX,
+ ABLCIAImporter,
+ IC_COLUMNS,
+ IMPACT_CATEGORIES_SHEET,
+ META_SUFFIX,
+ ab_csv_sibling_path,
+ export_methods_ab_csv_pair,
+ export_methods_ab_xlsx,
+ import_ab_methods,
+ load_ab_csv_pair,
+ load_ab_xlsx,
+ resolve_ab_csv_pair,
+)
+from .bw2io_lcia_file import (
+ load_bw2io_lcia_file,
+ method_name_to_filename_stem,
+)
+from .common import (
+ UNCERTAINTY_FIELDS,
+ CancelledError,
+ ConflictMode,
+ ImportStats,
+ activity_for_cf_key,
+ apply_name_conflicts,
+ cf_amount_and_uncertainty,
+ drop_unlinked_exchanges,
+ exchange_link_counts,
+ join_tuple_path,
+ raise_if_cancelled,
+ split_tuple_path,
+ unlinked_exchanges,
+)
+from .ecoinvent_lcia import EcoinventLCIAImporter
+
+__all__ = [
+ "ABLCIAImporter",
+ "CFS_COLUMNS",
+ "CFS_SHEET",
+ "CFS_SUFFIX",
+ "CancelledError",
+ "ConflictMode",
+ "EcoinventLCIAImporter",
+ "IC_COLUMNS",
+ "IMPACT_CATEGORIES_SHEET",
+ "ImportStats",
+ "META_SUFFIX",
+ "UNCERTAINTY_FIELDS",
+ "ab_csv_sibling_path",
+ "activity_for_cf_key",
+ "apply_name_conflicts",
+ "cf_amount_and_uncertainty",
+ "drop_unlinked_exchanges",
+ "exchange_link_counts",
+ "export_methods_ab_csv_pair",
+ "export_methods_ab_xlsx",
+ "import_ab_methods",
+ "join_tuple_path",
+ "load_ab_csv_pair",
+ "load_ab_xlsx",
+ "load_bw2io_lcia_file",
+ "method_name_to_filename_stem",
+ "raise_if_cancelled",
+ "resolve_ab_csv_pair",
+ "split_tuple_path",
+ "unlinked_exchanges",
+]
diff --git a/activity_browser/bwutils/impact_categories/ab_lcia_file.py b/activity_browser/bwutils/impact_categories/ab_lcia_file.py
new file mode 100644
index 000000000..66c59ea98
--- /dev/null
+++ b/activity_browser/bwutils/impact_categories/ab_lcia_file.py
@@ -0,0 +1,373 @@
+"""AB impact-category file load/export and shared prepared-dataset importer.
+
+``ABLCIAImporter`` links and writes prepared LCIA datasets (AB-shaped
+``list[dict]``). AB and bw2io LCIA file loaders both feed it.
+"""
+from __future__ import annotations
+
+import functools
+from pathlib import Path
+from typing import Any, Iterable, Sequence
+
+import bw2data as bd
+import pandas as pd
+import tqdm
+from bw2data import Database, Method, config, methods
+from bw2io.importers.base_lcia import LCIAImporter
+from bw2io.strategies import (
+ convert_uncertainty_types_to_integers,
+ drop_falsey_uncertainty_fields_but_keep_zeros,
+ drop_unspecified_subcategories,
+ link_iterable_by_fields,
+ set_biosphere_type,
+)
+
+from .common import (
+ UNCERTAINTY_FIELDS,
+ CancelledError,
+ ConflictMode,
+ ImportStats,
+ activity_for_cf_key,
+ apply_name_conflicts,
+ cell_str,
+ cf_amount_and_uncertainty,
+ drop_unlinked_exchanges,
+ join_tuple_path,
+ raise_if_cancelled,
+ split_tuple_path,
+ uncertainty_from_series,
+ unlinked_exchanges,
+)
+
+CFS_SHEET = "CFs"
+IMPACT_CATEGORIES_SHEET = "Impact categories"
+CFS_COLUMNS = ("method", "flow", "amount", *UNCERTAINTY_FIELDS)
+IC_COLUMNS = ("method", "unit", "description")
+CFS_SUFFIX = ".cfs.csv"
+META_SUFFIX = ".metadata.csv"
+
+
+def _flow_path_for_activity(act) -> str:
+ cats = tuple(act.get("categories") or ())
+ return join_tuple_path((act.get("name", ""), *cats))
+
+
+def _ab_csv_directory_and_stem(base: Path) -> tuple[Path, str]:
+ name, lower = base.name, base.name.lower()
+ for suffix in (CFS_SUFFIX, META_SUFFIX):
+ if lower.endswith(suffix):
+ return base.parent, name[: -len(suffix)]
+ return base.parent, base.name
+
+
+def methods_to_ab_records(
+ method_names: Iterable[tuple],
+ *,
+ cancel_check=None,
+) -> tuple[pd.DataFrame, pd.DataFrame]:
+ """Build CFs and Impact categories dataframes for the given method keys."""
+ ic_rows: list[dict] = []
+ cf_rows: list[dict] = []
+ names = [tuple(name) for name in method_names]
+ for name in tqdm.tqdm(names, desc="Reading impact categories", total=len(names)):
+ raise_if_cancelled(cancel_check)
+ meta = methods.get(name) or {}
+ method_path = join_tuple_path(name)
+ ic_rows.append(
+ {
+ "method": method_path,
+ "unit": meta.get("unit") or "",
+ "description": meta.get("description") or "",
+ }
+ )
+ for key, cf_data in Method(name).load():
+ row = {
+ "method": method_path,
+ "flow": _flow_path_for_activity(activity_for_cf_key(key)),
+ **cf_amount_and_uncertainty(cf_data),
+ }
+ for field in UNCERTAINTY_FIELDS:
+ row.setdefault(field, None)
+ cf_rows.append(row)
+ raise_if_cancelled(cancel_check)
+ return (
+ pd.DataFrame(cf_rows, columns=list(CFS_COLUMNS)),
+ pd.DataFrame(ic_rows, columns=list(IC_COLUMNS)),
+ )
+
+
+def export_methods_ab_xlsx(
+ method_names: Sequence[tuple],
+ path: str | Path,
+ *,
+ cancel_check=None,
+) -> Path:
+ path = Path(path)
+ cfs, ics = methods_to_ab_records(method_names, cancel_check=cancel_check)
+ raise_if_cancelled(cancel_check)
+ with pd.ExcelWriter(path, engine="openpyxl") as writer:
+ cfs.to_excel(writer, sheet_name=CFS_SHEET, index=False)
+ ics.to_excel(writer, sheet_name=IMPACT_CATEGORIES_SHEET, index=False)
+ return path
+
+
+def ab_csv_sibling_path(path: str | Path) -> Path | None:
+ """Return the expected sibling path for an AB CSV pair member, or None."""
+ path = Path(path)
+ name, lower = path.name, path.name.lower()
+ if lower.endswith(CFS_SUFFIX):
+ return path.with_name(name[: -len(CFS_SUFFIX)] + META_SUFFIX)
+ if lower.endswith(META_SUFFIX):
+ return path.with_name(name[: -len(META_SUFFIX)] + CFS_SUFFIX)
+ return None
+
+
+def resolve_ab_csv_pair(
+ path: str | Path,
+ *,
+ other_path: str | Path | None = None,
+) -> tuple[Path, Path]:
+ """Return (cfs_path, metadata_path) for an AB CSV import."""
+ path = Path(path)
+ lower = path.name.lower()
+ if lower.endswith(CFS_SUFFIX):
+ cfs_path, ic_path = path, Path(other_path) if other_path else ab_csv_sibling_path(path)
+ elif lower.endswith(META_SUFFIX):
+ ic_path, cfs_path = path, Path(other_path) if other_path else ab_csv_sibling_path(path)
+ else:
+ raise ValueError("AB CSV import expects a '.cfs.csv' or '.metadata.csv' file")
+ if cfs_path is None or ic_path is None:
+ raise FileNotFoundError("Matching AB CSV sibling file not found")
+ if not cfs_path.is_file():
+ raise FileNotFoundError(f"CF file not found: {cfs_path}")
+ if not ic_path.is_file():
+ raise FileNotFoundError(f"Metadata file not found: {ic_path}")
+ return cfs_path, ic_path
+
+
+def export_methods_ab_csv_pair(
+ method_names: Sequence[tuple],
+ base_path: str | Path,
+ *,
+ cancel_check=None,
+) -> tuple[Path, Path]:
+ """Write AB CSV pair. ``base_path`` is a directory + stem (no required suffix)."""
+ directory, stem = _ab_csv_directory_and_stem(Path(base_path))
+ cfs_path = directory / f"{stem}{CFS_SUFFIX}"
+ ic_path = directory / f"{stem}{META_SUFFIX}"
+ cfs, ics = methods_to_ab_records(method_names, cancel_check=cancel_check)
+ raise_if_cancelled(cancel_check)
+ cfs.to_csv(cfs_path, index=False)
+ ics.to_csv(ic_path, index=False)
+ return cfs_path, ic_path
+
+
+def _exchange_from_cf_row(row: pd.Series) -> dict:
+ parts = split_tuple_path(row["flow"])
+ if not parts:
+ raise ValueError("CF row missing flow path")
+ return {
+ "name": parts[0],
+ "categories": tuple(parts[1:]),
+ "amount": float(row["amount"]),
+ **uncertainty_from_series(row),
+ }
+
+
+def _method_dataset(name: tuple, unit: str, description: str, filename: str, exchanges: list) -> dict:
+ return {
+ "name": name,
+ "unit": unit,
+ "description": description,
+ "filename": filename,
+ "exchanges": exchanges,
+ }
+
+
+def _records_from_frames(cfs: pd.DataFrame, ics: pd.DataFrame, filename: str) -> list[dict]:
+ for col in ("method", "flow", "amount"):
+ if col not in cfs.columns:
+ raise ValueError(f"AB LCIA file missing CF column '{col}'")
+ for col in ("method", "unit", "description"):
+ if col not in ics.columns:
+ raise ValueError(f"AB LCIA file missing Impact categories column '{col}'")
+
+ meta_by_method: dict[tuple, dict[str, str]] = {}
+ for _, row in ics.iterrows():
+ key = split_tuple_path(row["method"])
+ if key:
+ meta_by_method[key] = {
+ "unit": cell_str(row.get("unit")),
+ "description": cell_str(row.get("description")),
+ }
+
+ grouped: dict[tuple, list] = {}
+ for _, row in cfs.iterrows():
+ if pd.isna(row.get("amount")):
+ continue
+ key = split_tuple_path(row["method"])
+ if key:
+ grouped.setdefault(key, []).append(_exchange_from_cf_row(row))
+
+ empty = {"unit": "", "description": ""}
+ data = []
+ for key, exchanges in grouped.items():
+ meta = meta_by_method.pop(key, empty)
+ data.append(
+ _method_dataset(key, meta["unit"], meta["description"], filename, exchanges)
+ )
+ for key, meta in meta_by_method.items():
+ data.append(_method_dataset(key, meta["unit"], meta["description"], filename, []))
+ return data
+
+
+def load_ab_xlsx(path: str | Path) -> list[dict]:
+ """Parse an AB impact-category workbook into LCIAImporter-shaped datasets."""
+ path = Path(path)
+ return _records_from_frames(
+ pd.read_excel(path, sheet_name=CFS_SHEET),
+ pd.read_excel(path, sheet_name=IMPACT_CATEGORIES_SHEET),
+ path.name,
+ )
+
+
+def load_ab_csv_pair(
+ path: str | Path,
+ *,
+ other_path: str | Path | None = None,
+) -> list[dict]:
+ """Parse an AB CSV sibling pair into LCIAImporter-shaped datasets."""
+ cfs_path, ic_path = resolve_ab_csv_pair(path, other_path=other_path)
+ return _records_from_frames(
+ pd.read_csv(cfs_path, comment="#"),
+ pd.read_csv(ic_path, comment="#"),
+ cfs_path.name,
+ )
+
+
+def _reformat_cfs_with_uncertainty(exchanges: list[dict]) -> list[tuple]:
+ rows = []
+ for obj in exchanges:
+ if "input" not in obj:
+ continue
+ data = {"amount": obj["amount"]}
+ for field in UNCERTAINTY_FIELDS:
+ if field in obj and obj[field] is not None:
+ data[field] = obj[field]
+ rows.append((obj["input"], data if len(data) > 1 else obj["amount"]))
+ return rows
+
+
+class ABLCIAImporter(LCIAImporter):
+ """
+ Shared write/link path for prepared LCIA datasets (bw2io ``LCIAImporter``).
+
+ Accepts AB-shaped ``list[dict]`` from AB or bw2io LCIA file loaders.
+ Preserves CF uncertainty on write.
+ """
+
+ def __init__(self, data: list[dict], biosphere: str | None = None):
+ self.applied_strategies = []
+ self.filepath = "(ab-lcia)"
+ self.biosphere_name = biosphere or config.biosphere
+ if self.biosphere_name not in bd.databases:
+ raise ValueError(f"Can't find biosphere database {self.biosphere_name}")
+ self.data = data
+ self.strategies = [
+ set_biosphere_type,
+ drop_unspecified_subcategories,
+ functools.partial(
+ link_iterable_by_fields,
+ other=Database(self.biosphere_name),
+ fields=("name", "categories"),
+ ),
+ drop_falsey_uncertainty_fields_but_keep_zeros,
+ convert_uncertainty_types_to_integers,
+ ]
+
+ def _reformat_cfs(self, ds):
+ return _reformat_cfs_with_uncertainty(ds)
+
+ def apply_strategies(self, strategies=None, verbose=False, cancel_check=None):
+ for strategy in tqdm.tqdm(
+ strategies if strategies is not None else self.strategies,
+ desc="Applying strategies",
+ ):
+ raise_if_cancelled(cancel_check)
+ self.apply_strategy(strategy, verbose=verbose)
+ raise_if_cancelled(cancel_check)
+
+ def write_methods(self, overwrite=False, verbose=True, cancel_check=None):
+ num_methods, num_cfs, num_unlinked = self.statistics(False)
+ if num_unlinked:
+ raise ValueError(
+ f"Can't write unlinked methods ({num_unlinked} unlinked cfs)"
+ )
+ prepared = []
+ for ds in tqdm.tqdm(self.data, desc="Preparing impact categories"):
+ raise_if_cancelled(cancel_check)
+ prepared.append((ds, self._reformat_cfs(ds["exchanges"])))
+
+ raise_if_cancelled(cancel_check)
+ written_names: list[tuple] = []
+ preexisting = set(methods)
+ try:
+ for ds, cfs in tqdm.tqdm(prepared, desc="Writing impact categories"):
+ raise_if_cancelled(cancel_check)
+ name = tuple(ds["name"])
+ if name in methods:
+ if not overwrite:
+ raise ValueError(
+ f"Method {name} already exists. Use overwrite=True"
+ )
+ del methods[name]
+ method = Method(name)
+ method.register(
+ description=ds.get("description") or "",
+ filename=ds.get("filename") or "",
+ unit=ds.get("unit") or "",
+ num_cfs=len(cfs),
+ )
+ method.write(cfs)
+ written_names.append(name)
+ except CancelledError:
+ for name in written_names:
+ if name not in preexisting:
+ methods.pop(name, None)
+ raise
+ if verbose:
+ print(
+ f"Wrote {num_methods} LCIA methods with {num_cfs} characterization factors"
+ )
+
+
+def import_ab_methods(
+ data: list[dict],
+ *,
+ biosphere_name: str,
+ conflict_mode: ConflictMode = ConflictMode.SKIP,
+ prefix: str | None = None,
+ renames: dict[tuple, tuple] | None = None,
+ drop_unlinked: bool = False,
+) -> ImportStats:
+ """Link and write prepared method datasets into the current project."""
+ prepared = apply_name_conflicts(
+ data,
+ set(methods),
+ mode=conflict_mode,
+ prefix=prefix,
+ renames=renames,
+ )
+ skipped = len(data) - len(prepared)
+ importer = ABLCIAImporter(prepared, biosphere=biosphere_name)
+ importer.apply_strategies()
+ unlinked = len(unlinked_exchanges(importer.data))
+ if unlinked and not drop_unlinked:
+ return ImportStats(written=0, skipped=skipped, unlinked=unlinked)
+ if drop_unlinked:
+ importer.data = drop_unlinked_exchanges(importer.data)
+ unlinked = 0
+ importer.write_methods(
+ overwrite=conflict_mode == ConflictMode.OVERWRITE, verbose=False
+ )
+ return ImportStats(written=len(importer.data), skipped=skipped, unlinked=unlinked)
diff --git a/activity_browser/bwutils/impact_categories/bw2io_lcia_file.py b/activity_browser/bwutils/impact_categories/bw2io_lcia_file.py
new file mode 100644
index 000000000..1de1d4554
--- /dev/null
+++ b/activity_browser/bwutils/impact_categories/bw2io_lcia_file.py
@@ -0,0 +1,182 @@
+"""bw2io impact-category file load/export (CF table + AB metadata helpers)."""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Sequence
+
+import bw2data as bd
+import pandas as pd
+import tqdm
+
+from .common import (
+ UNCERTAINTY_FIELDS,
+ activity_for_cf_key,
+ cell_str,
+ cf_amount_and_uncertainty,
+ join_tuple_path,
+ raise_if_cancelled,
+ split_tuple_path,
+ uncertainty_from_series,
+)
+
+BW2IO_CF_COLUMNS = ("name", "categories", "amount", *UNCERTAINTY_FIELDS)
+BW2IO_META_COLUMNS = ("filename", "method", "unit", "description")
+_FORBIDDEN_FILENAME_CHARS = '<>:"/\\|?*'
+
+
+def method_name_to_filename_stem(name: tuple) -> str:
+ """
+ Build a cross-platform filename stem from a Brightway method key.
+
+ Tuple parts are joined with ``__`` (``::`` is illegal on Windows). Remaining
+ forbidden characters are replaced with ``-``.
+ """
+ parts: list[str] = []
+ for part in name:
+ text = str(part).strip()
+ for ch in _FORBIDDEN_FILENAME_CHARS:
+ text = text.replace(ch, "-")
+ text = "".join(c for c in text if ord(c) >= 32).rstrip(" .")
+ parts.append(text or "part")
+ stem = "__".join(parts) if parts else "method"
+ return (stem[:200].rstrip(" .") if len(stem) > 200 else stem) or "method"
+
+
+def _method_meta(name: tuple, *, filename: str = "") -> dict[str, str]:
+ meta = bd.methods.get(name) or {}
+ return {
+ "filename": filename,
+ "method": join_tuple_path(name),
+ "unit": meta.get("unit") or "",
+ "description": meta.get("description") or "",
+ }
+
+
+def _cf_frame_for_method(name: tuple) -> pd.DataFrame:
+ rows = []
+ for key, cf_data in bd.Method(name).load():
+ act = activity_for_cf_key(key)
+ cats = tuple(act.get("categories") or ())
+ row = {
+ "name": act.get("name", ""),
+ "categories": join_tuple_path(cats) if cats else "",
+ **cf_amount_and_uncertainty(cf_data),
+ }
+ for field in UNCERTAINTY_FIELDS:
+ row.setdefault(field, None)
+ rows.append(row)
+ return pd.DataFrame(rows, columns=list(BW2IO_CF_COLUMNS))
+
+
+def export_method_bw2io_xlsx(name: tuple, path: str | Path) -> Path:
+ path = Path(path)
+ with pd.ExcelWriter(path, engine="openpyxl") as writer:
+ _cf_frame_for_method(name).to_excel(writer, sheet_name="CFs", index=False)
+ pd.DataFrame(
+ [_method_meta(name, filename=path.name)],
+ columns=list(BW2IO_META_COLUMNS),
+ ).to_excel(writer, sheet_name="metadata", index=False)
+ return path
+
+
+def export_methods_bw2io_csv_batch(
+ method_names: Sequence[tuple],
+ directory: str | Path,
+ *,
+ metadata_name: str = "metadata.csv",
+ cancel_check=None,
+) -> list[Path]:
+ directory = Path(directory)
+ directory.mkdir(parents=True, exist_ok=True)
+ written: list[Path] = []
+ meta_rows = []
+ for name in tqdm.tqdm(method_names, desc="Exporting bw2io CSV"):
+ raise_if_cancelled(cancel_check)
+ cf_name = f"{method_name_to_filename_stem(name)}.csv"
+ cf_path = directory / cf_name
+ _cf_frame_for_method(name).to_csv(cf_path, index=False)
+ written.append(cf_path)
+ meta_rows.append(_method_meta(name, filename=cf_name))
+ raise_if_cancelled(cancel_check)
+ meta_path = directory / metadata_name
+ pd.DataFrame(meta_rows, columns=list(BW2IO_META_COLUMNS)).to_csv(
+ meta_path, index=False
+ )
+ written.append(meta_path)
+ return written
+
+
+def _row_to_meta(row: pd.Series) -> dict[str, str]:
+ return {key: cell_str(row.get(key)) for key in BW2IO_META_COLUMNS}
+
+
+def read_bw2io_metadata_xlsx(path: str | Path) -> dict[str, str] | None:
+ try:
+ meta = pd.read_excel(path, sheet_name="metadata")
+ except ValueError:
+ return None
+ return None if meta.empty else _row_to_meta(meta.iloc[0])
+
+
+def read_bw2io_metadata_csv(
+ metadata_path: str | Path,
+ *,
+ cf_filename: str | None = None,
+) -> dict[str, str] | None:
+ """
+ Prefill metadata for a CF CSV.
+
+ Match ``filename`` to ``cf_filename`` when that column exists; if there is
+ exactly one row, use it; otherwise return ``None``.
+ """
+ path = Path(metadata_path)
+ if not path.is_file():
+ return None
+ meta = pd.read_csv(path, comment="#")
+ if meta.empty:
+ return None
+ if cf_filename and "filename" in meta.columns:
+ match = meta[meta["filename"].astype(str) == str(cf_filename)]
+ if len(match) >= 1:
+ return _row_to_meta(match.iloc[0])
+ return None
+ return _row_to_meta(meta.iloc[0]) if len(meta) == 1 else None
+
+
+def load_bw2io_lcia_file(
+ path: str | Path,
+ *,
+ name: tuple,
+ unit: str,
+ description: str,
+) -> list[dict]:
+ """Parse one bw2io CF table into ABLCIAImporter-shaped data (first sheet only for xlsx)."""
+ path = Path(path)
+ if path.suffix.lower() in {".xlsx", ".xls"}:
+ cfs = pd.read_excel(path, sheet_name=0)
+ else:
+ cfs = pd.read_csv(path, comment="#")
+ if "name" not in cfs.columns or "amount" not in cfs.columns:
+ raise ValueError("bw2io LCIA file must include 'name' and 'amount' columns")
+
+ exchanges = []
+ for _, row in cfs.iterrows():
+ if pd.isna(row.get("amount")):
+ continue
+ exchanges.append(
+ {
+ "name": str(row["name"]),
+ "categories": split_tuple_path(cell_str(row.get("categories"))),
+ "amount": float(row["amount"]),
+ **uncertainty_from_series(row),
+ }
+ )
+ return [
+ {
+ "name": tuple(name),
+ "unit": unit or "",
+ "description": description or "",
+ "filename": path.name,
+ "exchanges": exchanges,
+ }
+ ]
diff --git a/activity_browser/bwutils/impact_categories/common.py b/activity_browser/bwutils/impact_categories/common.py
new file mode 100644
index 000000000..213ae6f50
--- /dev/null
+++ b/activity_browser/bwutils/impact_categories/common.py
@@ -0,0 +1,140 @@
+"""Shared helpers for impact-category (LCIA) file interchange."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Mapping, Sequence
+
+import bw2data as bd
+import pandas as pd
+
+from activity_browser.bwutils.uncertainty import UNCERTAINTY_FIELDS
+
+
+class ConflictMode(str, Enum):
+ SKIP = "skip"
+ OVERWRITE = "overwrite"
+ RENAME_PREFIX = "rename_prefix"
+
+
+@dataclass
+class ImportStats:
+ written: int = 0
+ skipped: int = 0
+ unlinked: int = 0
+
+
+class CancelledError(Exception):
+ """Raised when the user cancels a long-running LCIA file job."""
+
+
+def raise_if_cancelled(check) -> None:
+ if check and check():
+ raise CancelledError()
+
+
+def join_tuple_path(parts: Sequence[Any]) -> str:
+ return "::".join(str(p) for p in parts)
+
+
+def split_tuple_path(value: str | None) -> tuple[str, ...]:
+ if value is None or (isinstance(value, float) and pd.isna(value)):
+ return ()
+ text = str(value).strip()
+ return tuple(p for p in text.split("::") if p) if text else ()
+
+
+def cell_str(value: Any) -> str:
+ """Coerce a spreadsheet cell to ``str``; blank for missing/NaN."""
+ if value is None or (isinstance(value, float) and pd.isna(value)):
+ return ""
+ return str(value)
+
+
+def apply_name_conflicts(
+ data: list[dict],
+ existing: set[tuple],
+ *,
+ mode: ConflictMode,
+ prefix: str | None = None,
+ renames: dict[tuple, tuple] | None = None,
+) -> list[dict]:
+ """Return a new method list after applying conflict policy (pure transform)."""
+ renames = renames or {}
+ result: list[dict] = []
+ for ds in data:
+ original = tuple(ds["name"])
+ name = tuple(renames.get(original, original))
+ if name != original:
+ ds = {**ds, "name": name}
+ if name not in existing:
+ result.append(ds)
+ elif mode == ConflictMode.SKIP:
+ continue
+ elif mode == ConflictMode.OVERWRITE:
+ result.append(ds)
+ elif mode == ConflictMode.RENAME_PREFIX:
+ if not prefix:
+ raise ValueError("prefix is required for RENAME_PREFIX")
+ result.append({**ds, "name": (prefix, *name)})
+ else:
+ raise ValueError(f"Unknown conflict mode: {mode}")
+ return result
+
+
+def cf_amount_and_uncertainty(cf_data: Any) -> dict[str, Any]:
+ if not isinstance(cf_data, dict):
+ return {"amount": float(cf_data)}
+ row = {"amount": float(cf_data.get("amount", 0))}
+ for field in UNCERTAINTY_FIELDS:
+ if field in cf_data and cf_data[field] is not None:
+ row[field] = cf_data[field]
+ return row
+
+
+def uncertainty_from_series(row: Mapping[str, Any]) -> dict[str, Any]:
+ """Typed uncertainty fields from a spreadsheet row (skip missing/NaN)."""
+ out: dict[str, Any] = {}
+ for field in UNCERTAINTY_FIELDS:
+ if field not in row:
+ continue
+ val = row[field]
+ if val is None or (isinstance(val, float) and pd.isna(val)):
+ continue
+ if field == "uncertainty type":
+ out[field] = int(val)
+ elif field == "negative":
+ out[field] = bool(val)
+ else:
+ out[field] = float(val)
+ return out
+
+
+def activity_for_cf_key(key: Any):
+ try:
+ return bd.get_activity(key)
+ except Exception:
+ return bd.get_node(id=key)
+
+
+def unlinked_exchanges(data: list[dict]) -> list[dict]:
+ return [
+ {"method": ds["name"], **exc}
+ for ds in data
+ for exc in ds.get("exchanges", [])
+ if "input" not in exc
+ ]
+
+
+def exchange_link_counts(data: list[dict]) -> tuple[int, int]:
+ """Return ``(linked_cf_count, unlinked_cf_count)`` after strategies."""
+ total = sum(len(ds.get("exchanges", [])) for ds in data)
+ unlinked = len(unlinked_exchanges(data))
+ return total - unlinked, unlinked
+
+
+def drop_unlinked_exchanges(data: list[dict]) -> list[dict]:
+ return [
+ {**ds, "exchanges": [e for e in ds.get("exchanges", []) if "input" in e]}
+ for ds in data
+ ]
diff --git a/activity_browser/bwutils/impact_categories/ecoinvent_lcia.py b/activity_browser/bwutils/impact_categories/ecoinvent_lcia.py
new file mode 100644
index 000000000..c79e7c8f9
--- /dev/null
+++ b/activity_browser/bwutils/impact_categories/ecoinvent_lcia.py
@@ -0,0 +1,156 @@
+"""Ecoinvent LCIA Implementation Excel import (vendor multi-method workbook)."""
+from __future__ import annotations
+
+import functools
+import warnings
+from numbers import Number
+
+import tqdm
+from bw2data import Database, Method, config, methods
+from bw2io.importers.base_lcia import LCIAImporter
+from bw2io.strategies import (
+ drop_unspecified_subcategories,
+ link_iterable_by_fields,
+ normalize_units,
+ rationalize_method_names,
+ set_biosphere_type,
+)
+from openpyxl import load_workbook
+
+
+class EcoinventLCIAImporter(LCIAImporter):
+ """Import ecoinvent-compatible LCIA Implementation Excel workbooks."""
+
+ def __init__(self, filepath, biosphere=None):
+ self.strategies = []
+ self.applied_strategies = []
+ self.filepath = filepath
+ self.biosphere_name = biosphere
+ if self.biosphere_name:
+ self.set_biosphere(self.biosphere_name)
+
+ @classmethod
+ def setup_with_ei_excel(cls, file: str, biosphere_database: str | None = None):
+ importer = cls(file, biosphere_database)
+ importer.set_biosphere(biosphere_database or config.biosphere)
+ importer.cf_data, importer.units = convert_lcia_methods_data(file)
+ importer.separate_methods()
+ return importer
+
+ def set_biosphere(self, biosphere_database: str, *, relink: bool = False):
+ kwargs = {"other": Database(biosphere_database), "fields": ("name", "categories")}
+ if relink:
+ kwargs["relink"] = True
+ self.strategies = [
+ normalize_units,
+ set_biosphere_type,
+ drop_unspecified_subcategories,
+ functools.partial(link_iterable_by_fields, **kwargs),
+ ]
+
+ def add_rationalize_method_names_strategy(self):
+ self.strategies.append(rationalize_method_names)
+
+ def separate_methods(self):
+ """Split flat CF rows into distinct method datasets."""
+ missing = {line["method"] for line in self.cf_data if line["method"] not in self.units}
+ if missing:
+ warnings.warn(
+ "Missing units for following: "
+ + " | ".join(sorted(str(m) for m in missing))
+ )
+
+ by_method: dict[tuple, dict] = {}
+ for line in self.cf_data:
+ if line is None:
+ continue
+ assert isinstance(line["amount"], Number)
+ name = line["method"]
+ if name not in by_method:
+ by_method[name] = {
+ "filename": self.filepath,
+ "unit": self.units.get(name, ""),
+ "name": name,
+ "description": "",
+ "exchanges": [],
+ }
+ by_method[name]["exchanges"].append(
+ {
+ "name": line["name"],
+ "categories": line["categories"],
+ "amount": line["amount"],
+ }
+ )
+ self.data = list(by_method.values())
+
+ def apply_strategies(self, strategies=None, verbose=False):
+ for strategy in tqdm.tqdm(
+ strategies or self.strategies, desc="Applying strategies"
+ ):
+ self.apply_strategy(strategy, verbose=verbose)
+
+ def prepend_methods(self, prepend: str):
+ if not prepend:
+ return
+ for method in tqdm.tqdm(self.data, desc=f"Prepending {prepend} to ICs"):
+ method["name"] = (prepend, *method["name"])
+
+ def write_methods(self, overwrite=False, verbose=True):
+ num_methods, num_cfs, num_unlinked = self.statistics(False)
+ if num_unlinked:
+ raise ValueError(f"Can't write unlinked methods ({num_unlinked} unlinked cfs)")
+ for ds in tqdm.tqdm(self.data, desc="Writing impact categories"):
+ name = ds["name"]
+ if name in methods:
+ if not overwrite:
+ raise ValueError(
+ f"Method {name} already exists. Use overwrite=True"
+ )
+ del methods[name]
+ method = Method(name)
+ cfs = self._reformat_cfs(ds["exchanges"])
+ method.register(
+ description=ds["description"],
+ filename=ds["filename"],
+ unit=ds["unit"],
+ num_cfs=len(cfs),
+ )
+ method.write(cfs)
+ if verbose:
+ print(
+ f"Wrote {num_methods} LCIA methods with {num_cfs} characterization factors"
+ )
+
+
+def convert_lcia_methods_data(filename: str):
+ wb = load_workbook(filename, read_only=True)
+
+ cf_data = []
+ sheet = wb["CFs"]
+ for rowidx, row in tqdm.tqdm(
+ enumerate(sheet.rows), total=sheet.max_row, desc="Processing CFs"
+ ):
+ if not rowidx:
+ continue
+ data = [cell.value for _, cell in zip(range(8), row)]
+ if isinstance(data[-1], Number):
+ cf_data.append(
+ {
+ "method": tuple(data[:3]),
+ "name": data[3],
+ "categories": tuple(data[4:6]),
+ "amount": data[6],
+ }
+ )
+
+ units = {}
+ sheet = wb["Indicators"]
+ for rowidx, row in tqdm.tqdm(
+ enumerate(sheet.rows), total=sheet.max_row, desc="Processing indicators"
+ ):
+ if not rowidx:
+ continue
+ data = [cell.value for _, cell in zip(range(4), row)]
+ units[tuple(data[:3])] = data[3]
+
+ return cf_data, units
diff --git a/activity_browser/bwutils/impact_categories/templates.py b/activity_browser/bwutils/impact_categories/templates.py
new file mode 100644
index 000000000..ffcfef8bd
--- /dev/null
+++ b/activity_browser/bwutils/impact_categories/templates.py
@@ -0,0 +1,86 @@
+"""Resolve and copy impact-category interchange templates."""
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+TEMPLATES_DIR = Path(__file__).resolve().parents[2] / "templates" / "impact-categories"
+
+# kind -> files relative to TEMPLATES_DIR (csv kinds are pairs)
+TEMPLATE_FILES = {
+ "ab-xlsx": ("ab-lcia.xlsx",),
+ "ab-csv": ("ab-lcia.cfs.csv", "ab-lcia.metadata.csv"),
+ "bw2io-xlsx": ("bw2io-lcia.xlsx",),
+ "bw2io-csv": ("bw2io-lcia.csv", "bw2io-lcia.metadata.csv"),
+}
+
+TEMPLATE_LABELS = {
+ "ab-xlsx": "AB impact-category file (Excel) — multi–IC; recommended",
+ "ab-csv": "AB impact-category file (CSV pair) — multi–IC",
+ "bw2io-xlsx": "bw2io impact-category file (Excel) — one IC per file",
+ "bw2io-csv": "bw2io impact-category file (CSV) — one IC per file + metadata sidecar",
+}
+
+
+def template_paths(kind: str) -> list[Path]:
+ if kind not in TEMPLATE_FILES:
+ raise ValueError(f"Unknown impact-category template kind: {kind}")
+ paths = [TEMPLATES_DIR / name for name in TEMPLATE_FILES[kind]]
+ missing = [p for p in paths if not p.is_file()]
+ if missing:
+ raise FileNotFoundError(f"Template file(s) not found: {missing}")
+ return paths
+
+
+def copy_impact_category_template(kind: str, destination: Path) -> list[Path]:
+ """
+ Copy template file(s) for ``kind`` to ``destination``.
+
+ For single-file kinds, ``destination`` is the target file path.
+ For CSV pairs, ``destination`` is a directory or a base stem path; both
+ siblings are written beside/into it.
+ """
+ sources = template_paths(kind)
+ destination = Path(destination)
+ written: list[Path] = []
+
+ if len(sources) == 1:
+ dest = destination
+ if dest.is_dir():
+ dest = dest / sources[0].name
+ dest.parent.mkdir(parents=True, exist_ok=True)
+ shutil.copy2(sources[0], dest)
+ written.append(dest)
+ return written
+
+ # CSV pair
+ if destination.suffix:
+ # treat as stem path (possibly with .csv)
+ directory = destination.parent
+ stem = destination.name
+ for suffix in (".cfs.csv", ".metadata.csv", ".impact-categories.csv", ".csv"):
+ if stem.lower().endswith(suffix):
+ stem = stem[: -len(suffix)]
+ break
+ else:
+ stem = destination.stem
+ else:
+ directory = destination if destination.suffix == "" else destination.parent
+ if destination.exists() and destination.is_dir():
+ directory = destination
+ stem = sources[0].name.split(".")[0]
+ else:
+ directory = destination.parent
+ stem = destination.name or sources[0].name.split(".")[0]
+
+ directory.mkdir(parents=True, exist_ok=True)
+ for source in sources:
+ # preserve meaningful suffixes after the shared stem prefix of the packaged name
+ # e.g. ab-lcia.cfs.csv -> {stem}.cfs.csv
+ name = source.name
+ packaged_stem = name.split(".")[0]
+ remainder = name[len(packaged_stem) :] # includes leading dots/suffixes
+ dest = directory / f"{stem}{remainder}"
+ shutil.copy2(source, dest)
+ written.append(dest)
+ return written
diff --git a/activity_browser/bwutils/io/ecoinvent_lcia_importer.py b/activity_browser/bwutils/io/ecoinvent_lcia_importer.py
deleted file mode 100644
index a067fbba8..000000000
--- a/activity_browser/bwutils/io/ecoinvent_lcia_importer.py
+++ /dev/null
@@ -1,183 +0,0 @@
-import functools
-import warnings
-import tqdm
-from numbers import Number
-
-from bw2data import Database, config, methods, Method
-from openpyxl import load_workbook
-
-from bw2io.strategies import (
- drop_unspecified_subcategories,
- link_iterable_by_fields,
- normalize_units,
- rationalize_method_names,
- set_biosphere_type,
-)
-from bw2io.importers.base_lcia import LCIAImporter
-
-
-class EcoinventLCIAImporter(LCIAImporter):
- """
- A class for importing ecoinvent-compatible LCIA methods
-
- """
- def __init__(self, filepath, biosphere=None):
- self.strategies = []
- self.applied_strategies = []
- self.filepath = filepath
- self.biosphere_name = biosphere
-
- if self.biosphere_name:
- self.set_biosphere(self.biosphere_name)
-
- @classmethod
- def setup_with_ei_excel(cls, file: str, biosphere_database: str | None = None):
- """Initialize an instance of EcoinventLCIAImporter.
-
- Defines strategies in ``__init__`` because ``config.biosphere`` is dynamic.
- """
- importer = cls(file, biosphere_database)
- importer.strategies = [
- normalize_units,
- set_biosphere_type,
- drop_unspecified_subcategories,
- functools.partial(
- link_iterable_by_fields,
- other=Database(biosphere_database or config.biosphere),
- fields=("name", "categories"),
- ),
- ]
- importer.cf_data, importer.units = convert_lcia_methods_data(file)
- importer.separate_methods()
- return importer
-
- def set_biosphere(self, biosphere_database: str):
- self.strategies = [
- normalize_units,
- set_biosphere_type,
- drop_unspecified_subcategories,
- functools.partial(
- link_iterable_by_fields,
- other=Database(biosphere_database),
- fields=("name", "categories"),
- relink=True,
- ),
- ]
-
- def add_rationalize_method_names_strategy(self):
- self.strategies.append(rationalize_method_names)
-
- def separate_methods(self):
- """Separate the list of CFs into distinct methods"""
- methods = {obj["method"] for obj in self.cf_data}
-
- self.data = {}
-
- missing = set()
-
- for line in self.cf_data:
- if line["method"] not in self.units:
- missing.add(line["method"])
-
- if missing:
- _ = lambda x: sorted([str(y) for y in x])
- warnings.warn("Missing units for following:" + " | ".join(_(missing)))
-
- for line in self.cf_data:
- assert isinstance(line["amount"], Number)
-
- if line["method"] not in self.data:
- self.data[line["method"]] = {
- "filename": self.filepath,
- "unit": self.units.get(line["method"], ""),
- "name": line["method"],
- "description": "",
- "exchanges": [],
- }
-
- self.data[line["method"]]["exchanges"].append(
- {
- "name": line["name"],
- "categories": line["categories"],
- "amount": line["amount"],
- }
- )
-
- self.data = list(self.data.values())
-
- def apply_strategies(self, strategies=None, verbose=False):
- strategies = strategies or self.strategies
- for strategy in tqdm.tqdm(strategies, desc="Applying strategies", total=len(strategies)):
- self.apply_strategy(strategy, verbose=verbose)
-
- def prepend_methods(self, prepend: str):
- if not prepend:
- return
- for method in tqdm.tqdm(self.data, desc=f"Prepending {prepend} to ICs"):
- method["name"] = tuple([prepend, *method["name"]])
-
- def write_methods(self, overwrite=False, verbose=True):
- num_methods, num_cfs, num_unlinked = self.statistics(False)
- if num_unlinked:
- raise ValueError(
- ("Can't write unlinked methods ({} unlinked cfs)").format(num_unlinked)
- )
- for ds in tqdm.tqdm(self.data, total=len(self.data), desc="Processing CF's"):
- if ds["name"] in methods:
- if overwrite:
- del methods[ds["name"]]
- else:
- raise ValueError(
- (
- "Method {} already exists. Use "
- "``overwrite=True`` to overwrite existing methods"
- ).format(ds["name"])
- )
- method = Method(ds["name"])
- method.register(
- description=ds["description"],
- filename=ds["filename"],
- unit=ds["unit"],
- )
- method.write(self._reformat_cfs(ds["exchanges"]))
- if verbose:
- print(
- "Wrote {} LCIA methods with {} characterization factors".format(
- num_methods, num_cfs
- )
- )
-
-
-def convert_lcia_methods_data(filename: str):
- sheet = load_workbook(filename, read_only=True)["CFs"]
-
- def process_row(row):
- data = [cell.value for i, cell in zip(range(8), row)]
- if not isinstance(data[-1], Number):
- return None
- else:
- return {
- "method": tuple(data[:3]),
- "name": data[3],
- "categories": tuple(data[4:6]),
- "amount": data[6],
- }
-
- cf_data = []
- for rowidx, row in tqdm.tqdm(enumerate(sheet.rows), total=sheet.max_row, desc="Processing CF's"):
- if rowidx:
- cf_data.append(process_row(row))
-
- sheet = load_workbook(filename, read_only=True)["Indicators"]
-
- def process_unit_row(row):
- data = [cell.value for i, cell in zip(range(4), row)]
- return tuple(data[:3]), data[3]
-
- units = {}
- for rowidx, row in tqdm.tqdm(enumerate(sheet.rows), total=sheet.max_row, desc="Processing indicators"):
- if rowidx:
- key, value = process_unit_row(row)
- units[key] = value
-
- return cf_data, units
diff --git a/activity_browser/bwutils/uncertainty.py b/activity_browser/bwutils/uncertainty.py
index 9af92b245..794891a72 100644
--- a/activity_browser/bwutils/uncertainty.py
+++ b/activity_browser/bwutils/uncertainty.py
@@ -37,6 +37,9 @@
"negative": False,
}
+# Ordered field names for spreadsheet columns / CF serialization (keys of EMPTY_UNCERTAINTY).
+UNCERTAINTY_FIELDS = tuple(EMPTY_UNCERTAINTY)
+
# Fields that may be left empty; ``stats_arrays`` supplies defaults or ignores them.
OPTIONAL_UNCERTAINTY_FIELDS = {
sa.BetaUncertainty.id: frozenset({"minimum", "maximum"}),
diff --git a/activity_browser/templates/README.md b/activity_browser/templates/README.md
index d07e987fc..c74352487 100644
--- a/activity_browser/templates/README.md
+++ b/activity_browser/templates/README.md
@@ -7,8 +7,9 @@ Bundled spreadsheet templates shipped with Activity Browser (top-level package f
- **`scenarios/`** — scenario difference / parameter-scenario workbooks for calculation-setup Scenario mode
- `flow-scenarios.xlsx` / `.csv` — empty flow scenario (SDF) headers; xlsx includes a `README` sheet
- `parameter-scenarios.xlsx` / `.csv` — empty parameter scenario headers; xlsx includes a `README` sheet
-
-Future additions may include Brightway Excel database examples under additional subfolders (e.g. `databases/`).
+- **`impact-categories/`** — LCIA / impact-category interchange starters
+ - `ab-lcia.xlsx` / `ab-lcia.cfs.csv` + `ab-lcia.metadata.csv` — **AB impact-category file** (multi–impact-category; recommended default)
+ - `bw2io-lcia.xlsx` / `bw2io-lcia.csv` + `bw2io-lcia.metadata.csv` — **bw2io impact-category file** (one impact category per CF file)
## Usage
@@ -22,8 +23,8 @@ templates = Path(activity_browser.__file__).resolve().parent / "templates"
flow = templates / "scenarios" / "flow-scenarios.xlsx"
```
-Excel workbooks: **data sheet first**, then **`README`**.
-CSV files: header row, blank rows, then notes on lines starting with `#` (ignored on import).
+Excel workbooks: **data sheet first**, then **`README`** (and metadata sheets where applicable).
+CSV files: header row, then notes on lines starting with `#` (ignored on import).
Scenario import comments (Excel and CSV):
@@ -32,7 +33,10 @@ Scenario import comments (Excel and CSV):
**Get template → flow-scenarios** always copies the empty starter file (does not generate from project parameters).
+Impact categories: **Impact categories → Get template…** in the application menu (AB impact-category file is the default choice).
+
## Maintenance
- Keep column headers aligned with `SUPERSTRUCTURE` / parameter-scenario import expectations in `bwutils/superstructure`.
+- Keep impact-category templates aligned with `bwutils.impact_categories`.
- When adding new `.xlsx` / `.csv` templates, include them in `MANIFEST.in` so they ship in wheels/sdists.
diff --git a/activity_browser/templates/impact-categories/ab-lcia.cfs.csv b/activity_browser/templates/impact-categories/ab-lcia.cfs.csv
new file mode 100644
index 000000000..e8caeffe4
--- /dev/null
+++ b/activity_browser/templates/impact-categories/ab-lcia.cfs.csv
@@ -0,0 +1,3 @@
+method,flow,amount,uncertainty type,loc,scale,shape,minimum,maximum,negative
+# method uses :: for Brightway name parts; flow is name::category::...
+# Optional uncertainty columns: uncertainty type, loc, scale, shape, minimum, maximum, negative
diff --git a/activity_browser/templates/impact-categories/ab-lcia.metadata.csv b/activity_browser/templates/impact-categories/ab-lcia.metadata.csv
new file mode 100644
index 000000000..2176ce964
--- /dev/null
+++ b/activity_browser/templates/impact-categories/ab-lcia.metadata.csv
@@ -0,0 +1,2 @@
+method,unit,description
+# One row per impact category; method must match CFs.method values
diff --git a/activity_browser/templates/impact-categories/ab-lcia.xlsx b/activity_browser/templates/impact-categories/ab-lcia.xlsx
new file mode 100644
index 000000000..7810b409e
Binary files /dev/null and b/activity_browser/templates/impact-categories/ab-lcia.xlsx differ
diff --git a/activity_browser/templates/impact-categories/bw2io-lcia.csv b/activity_browser/templates/impact-categories/bw2io-lcia.csv
new file mode 100644
index 000000000..26dd3a0cc
--- /dev/null
+++ b/activity_browser/templates/impact-categories/bw2io-lcia.csv
@@ -0,0 +1,2 @@
+name,categories,amount,uncertainty type,loc,scale,shape,minimum,maximum,negative
+# bw2io one-shot CF table; categories use ::
diff --git a/activity_browser/templates/impact-categories/bw2io-lcia.metadata.csv b/activity_browser/templates/impact-categories/bw2io-lcia.metadata.csv
new file mode 100644
index 000000000..f4d33e204
--- /dev/null
+++ b/activity_browser/templates/impact-categories/bw2io-lcia.metadata.csv
@@ -0,0 +1,3 @@
+filename,method,unit,description
+# Shared metadata sidecar for a batch of bw2io CSV CF files
+# filename matches the sibling CF CSV name; method uses :: for Brightway name parts
diff --git a/activity_browser/templates/impact-categories/bw2io-lcia.xlsx b/activity_browser/templates/impact-categories/bw2io-lcia.xlsx
new file mode 100644
index 000000000..6786b9d08
Binary files /dev/null and b/activity_browser/templates/impact-categories/bw2io-lcia.xlsx differ
diff --git a/activity_browser/ui/core/threading.py b/activity_browser/ui/core/threading.py
index 7cea5ebad..f50238d2b 100644
--- a/activity_browser/ui/core/threading.py
+++ b/activity_browser/ui/core/threading.py
@@ -42,7 +42,10 @@ def run(self):
raise e
qt_tqdm.updated.disconnect(self._emit_status)
- self.status.emit(100, "Complete")
+ if not self.isInterruptionRequested() and not getattr(
+ self, "_ab_cancel_requested", False
+ ):
+ self.status.emit(100, "Complete")
def _emit_status(self, progress: int, message: str):
if progress == 100:
@@ -53,11 +56,26 @@ def _emit_status(self, progress: int, message: str):
def run_safely(self, *args, **kwargs):
raise NotImplementedError
+ def request_ab_cancel(self):
+ """Sticky cancel for long-running jobs (survives progress-dialog resets)."""
+ self._ab_cancel_requested = True
+ self.requestInterruption()
+
+ def ab_cancel_requested(self) -> bool:
+ return bool(
+ getattr(self, "_ab_cancel_requested", False)
+ or self.isInterruptionRequested()
+ )
+
def connect_progress_dialog(self, progress_dialog: QtWidgets.QProgressDialog):
"""
Connects the status signal to a progress dialog.
"""
def slot(progress, message):
+ if getattr(progress_dialog, "ab_cancelled", False) or (
+ hasattr(progress_dialog, "wasCanceled") and progress_dialog.wasCanceled()
+ ):
+ return
if progress == -1:
progress_dialog.setLabelText(message)
progress_dialog.setRange(0, 0)
diff --git a/activity_browser/ui/dialogs/progress_dialog.py b/activity_browser/ui/dialogs/progress_dialog.py
index 5dec1df83..ff0ea3169 100644
--- a/activity_browser/ui/dialogs/progress_dialog.py
+++ b/activity_browser/ui/dialogs/progress_dialog.py
@@ -7,20 +7,77 @@
class ABProgressDialog(QProgressDialog):
@classmethod
- def get_connected_dialog(cls, title: str) -> "ABProgressDialog":
+ def get_connected_dialog(
+ cls, title: str, *, cancellable: bool = False
+ ) -> "ABProgressDialog":
from activity_browser.app import application
-
+
dialog = cls(application.main_window)
dialog.setWindowTitle(title)
dialog.setLabelText("Initializing")
+ dialog.setRange(0, 100)
dialog.setAutoReset(False)
- dialog.setCancelButton(None)
+ dialog.setAutoClose(False)
+ dialog.setMinimumDuration(0)
+ dialog._ab_cancelled = False
+ dialog._updates_disconnected = False
+ if cancellable:
+ dialog.setCancelButtonText("Cancel")
+ else:
+ dialog.setCancelButton(None)
- qt_tqdm.updated.connect(dialog._receive_update)
- qt_pyprind.updated.connect(dialog._receive_update)
+ # qt_tqdm emits (percent: int, desc: str); qt_pyprind emits (title: str, percent)
+ qt_tqdm.updated.connect(dialog._receive_tqdm_update)
+ qt_pyprind.updated.connect(dialog._receive_pyprind_update)
+ dialog.canceled.connect(dialog._on_canceled)
return dialog
- def _receive_update(self, title: str, value: int):
- self.setLabelText(title)
- self.setValue(value)
+ def _on_canceled(self):
+ self.mark_cancelled()
+
+ def mark_cancelled(self) -> None:
+ """Sticky cancel flag + stop progress updates (safe after dialog close)."""
+ self._ab_cancelled = True
+ self.disconnect_progress_updates()
+
+ def disconnect_progress_updates(self):
+ """Idempotent: safe to call more than once (close() may re-enter via canceled)."""
+ if getattr(self, "_updates_disconnected", False):
+ return
+ self._updates_disconnected = True
+ for signal, slot in (
+ (qt_tqdm.updated, self._receive_tqdm_update),
+ (qt_pyprind.updated, self._receive_pyprind_update),
+ ):
+ try:
+ signal.disconnect(slot)
+ except (RuntimeError, TypeError):
+ pass
+
+ def detach(self):
+ """Disconnect all external slots before closing after a finished job."""
+ try:
+ self.canceled.disconnect(self._on_canceled)
+ except (RuntimeError, TypeError):
+ pass
+ self.disconnect_progress_updates()
+
+ @property
+ def ab_cancelled(self) -> bool:
+ return bool(getattr(self, "_ab_cancelled", False) or self.wasCanceled())
+
+ def _receive_tqdm_update(self, value: int, title: str):
+ # Calling setValue after Cancel can re-show / clear canceled state in Qt.
+ if self.ab_cancelled or getattr(self, "_updates_disconnected", False):
+ return
+ self.setRange(0, 100)
+ self.setLabelText(title or "Working...")
+ self.setValue(int(value))
+
+ def _receive_pyprind_update(self, title: str, value: float):
+ if self.ab_cancelled or getattr(self, "_updates_disconnected", False):
+ return
+ self.setRange(0, 100)
+ self.setLabelText(title or "Working...")
+ self.setValue(int(value))
diff --git a/docs/advanced-topics/impact-category-interchange.md b/docs/advanced-topics/impact-category-interchange.md
new file mode 100644
index 000000000..5a6742571
--- /dev/null
+++ b/docs/advanced-topics/impact-category-interchange.md
@@ -0,0 +1,56 @@
+# Impact category import and export
+
+Activity Browser supports three spreadsheet interchange styles for **impact categories** (Brightway LCIA methods).
+
+## Formats
+
+### AB impact-category file (recommended default)
+
+Multi–impact-category file with self-describing metadata and variable-length method names.
+
+- **Excel:** sheets `CFs` and `Impact categories` (templates also include `README`)
+- **CSV pair:** `*.cfs.csv` and `*.metadata.csv`
+
+| Sheet / file | Columns |
+|---|---|
+| CFs / `*.cfs.csv` | `method`, `flow`, `amount`, optional uncertainty fields |
+| Impact categories (xlsx) / `*.metadata.csv` | `method`, `unit`, `description` |
+
+- `method` and `flow` use `::` (e.g. `My method::climate change::GWP100`, `Ammonia::air::unspecified`)
+- Uncertainty columns: `uncertainty type`, `loc`, `scale`, `shape`, `minimum`, `maximum`, `negative`
+
+Menu: **Impact categories → Import → From AB LCIA file (.xlsx/.csv)…** / **Export → To AB LCIA file (.xlsx/.csv)…**
+
+### bw2io impact-category file
+
+Compatible with bw2io’s Excel/CSV LCIA CF table (`name`, `categories` with `::`, `amount`, optional uncertainty): **one impact category per CF file**.
+
+- **Excel:** CF sheet + AB `metadata` sheet (`filename`, `method`, `unit`, `description`) used to prefill the import dialog
+- **CSV:** one CF file per impact category + shared `metadata.csv` sidecar (`filename` matches the CF file name when several rows exist)
+- Exported file names join method-tuple parts with `__` (not `::`, which is illegal on Windows) and replace other forbidden characters with `-`
+
+Stock bw2io only needs the CF table; AB reads metadata when present and always shows a confirmation dialog. You can multi-select several Excel/CSV files in one import; shared CSV `metadata.csv` rows are matched by `filename`.
+
+Menu: **Impact categories → Import → From bw2io LCIA file (.xlsx/.csv)…** / **Export → To bw2io LCIA file (.xlsx/.csv)…**
+
+### ecoinvent LCIA Implementation Excel
+
+Vendor multi-method workbook (`CFs` + `units` / `Indicators`). Method names are three parts (`method` / `category` / `indicator`); flows use name / compartment / subcompartment columns. Here **Indicators** is ecoinvent’s sheet name (not AB’s metadata table).
+
+Menu: **Impact categories → Import → From ecoinvent Excel…**
+
+## Templates
+
+**Impact categories → Get template…** copies starters from `activity_browser/templates/impact-categories/`. AB impact-category file is listed first (default).
+
+## Import behaviour (AB and bw2io file imports)
+
+- Choose the biosphere database used for linking
+- Multi–impact-category conflicts: skip existing / overwrite (with confirm) / rename with a namespace prefix; optional per-conflict rename table
+- Single bw2io impact-category file name conflict: overwrite (with confirm), edit name, or cancel
+- Unlinked characterization factors: dialog shows linked and unlinked counts; cancel, export unmatched list, or drop unlinked (no automatic biosphere creation)
+- Long-running load, link, write, and export steps show a progress dialog with **Cancel**. Cancel stops the job. For import write, methods **created** in that run are rolled back so they do not remain in the project. **Overwrite runs are different:** if cancel happens after an existing impact category was already removed/replaced, that category may stay deleted or partially updated (the UI warns about this when overwrite is chosen). Full overwrite undo is out of scope for v1.
+
+## Export behaviour
+
+Menu export uses the current Impact categories pane selection. If nothing is selected (or the pane cannot be resolved), you are asked whether to export all impact categories. Export also shows a progress dialog while reading methods and writing files.
\ No newline at end of file
diff --git a/docs/agents/impact-category-interchange.md b/docs/agents/impact-category-interchange.md
new file mode 100644
index 000000000..badaee31f
--- /dev/null
+++ b/docs/agents/impact-category-interchange.md
@@ -0,0 +1,13 @@
+# Impact-category interchange (agent notes)
+
+Canonical user docs: [`docs/advanced-topics/impact-category-interchange.md`](../advanced-topics/impact-category-interchange.md).
+
+Glossary in root `CONTEXT.md`: **AB impact-category file**, **bw2io impact-category file**. Prefer “impact category” / “LCIA method”; avoid “Indicators” for AB metadata and avoid “one-shot” / “bw2io native”.
+
+Implementation seam: `activity_browser.bwutils.impact_categories` — `ab_lcia_file` / `bw2io_lcia_file` / `ecoinvent_lcia` / `common`; `ABLCIAImporter` links and writes prepared datasets (fed by AB or bw2io loaders).
+
+UI: one ABAction per file under `app.actions.method` (`method_import_*`, `method_export_*`, `method_get_template`). Action-specific dialogs and worker threads live in those same modules (repo convention). Shared sticky-cancel progress starter is `app.dialogs.run_thread_with_progress` (`ABProgressDialog` remains in `ui.dialogs`). Menu export selection is `live_impact_category_selection()` / `resolve_methods_for_export()` on the Impact categories pane.
+
+AB CSV pair suffixes: `.cfs.csv` + `.metadata.csv` (Excel sheet name remains **Impact categories**). Do not use “Indicators” for AB metadata.
+
+Templates: `activity_browser/templates/impact-categories/`.
diff --git a/docs/user-interface/panes/impact-categories.md b/docs/user-interface/panes/impact-categories.md
index e1f4076a9..38717af2d 100644
--- a/docs/user-interface/panes/impact-categories.md
+++ b/docs/user-interface/panes/impact-categories.md
@@ -20,4 +20,7 @@ The Impact Categories view displays a list of all impact categories in your proj
## Actions
### Open Impact Category
-Open a database in the Database Product Pane by double-clicking the entry.
+Open an impact category details page by double-clicking the entry (or use the context menu).
+
+### Import / export / templates
+Use the **Impact categories** top menu to import, export, or get spreadsheet templates. Formats (AB impact-category file, bw2io impact-category file, ecoinvent) are described in [Impact category import and export](../../advanced-topics/impact-category-interchange.md).
diff --git a/tests/test_ab_lcia_interchange.py b/tests/test_ab_lcia_interchange.py
new file mode 100644
index 000000000..0bec53f9f
--- /dev/null
+++ b/tests/test_ab_lcia_interchange.py
@@ -0,0 +1,289 @@
+"""Fast AB impact-category Excel interchange tests (bwutils seam)."""
+from pathlib import Path
+
+import bw2data as bd
+import pytest
+from bw2data.tests import bw2test
+from stats_arrays import LognormalUncertainty
+
+from activity_browser.bwutils.impact_categories import (
+ ConflictMode,
+ apply_name_conflicts,
+ export_methods_ab_xlsx,
+ import_ab_methods,
+ join_tuple_path,
+ load_ab_xlsx,
+ split_tuple_path,
+)
+from fixtures.basic import DATABASE
+from fixtures.bw_helpers import write_functional_database
+
+
+def test_join_split_tuple_path():
+ assert join_tuple_path(("My", "climate", "GWP")) == "My::climate::GWP"
+ assert split_tuple_path("My::climate::GWP") == ("My", "climate", "GWP")
+ assert split_tuple_path("single") == ("single",)
+
+
+def test_apply_name_conflicts_rename_prefix():
+ data = [
+ {"name": ("IPCC", "GWP"), "unit": "kg", "description": "", "exchanges": []},
+ {"name": ("new",), "unit": "kg", "description": "", "exchanges": []},
+ ]
+ out = apply_name_conflicts(
+ data,
+ {("IPCC", "GWP")},
+ mode=ConflictMode.RENAME_PREFIX,
+ prefix="Import",
+ )
+ names = {ds["name"] for ds in out}
+ assert ("Import", "IPCC", "GWP") in names
+ assert ("new",) in names
+
+
+def test_apply_name_conflicts_skip():
+ data = [
+ {"name": ("IPCC", "GWP"), "unit": "kg", "description": "", "exchanges": []},
+ {"name": ("new",), "unit": "kg", "description": "", "exchanges": []},
+ ]
+ out = apply_name_conflicts(data, {("IPCC", "GWP")}, mode=ConflictMode.SKIP)
+ assert [ds["name"] for ds in out] == [("new",)]
+
+
+def test_apply_name_conflicts_per_row_renames():
+ data = [
+ {"name": ("IPCC", "GWP"), "unit": "kg", "description": "", "exchanges": []},
+ {"name": ("kept",), "unit": "kg", "description": "", "exchanges": []},
+ ]
+ out = apply_name_conflicts(
+ data,
+ {("IPCC", "GWP")},
+ mode=ConflictMode.SKIP,
+ renames={("IPCC", "GWP"): ("IPCC", "GWP", "imported")},
+ )
+ assert [ds["name"] for ds in out] == [("IPCC", "GWP", "imported"), ("kept",)]
+
+
+def test_ab_csv_sibling_resolution(tmp_path: Path):
+ from activity_browser.bwutils.impact_categories import (
+ ab_csv_sibling_path,
+ resolve_ab_csv_pair,
+ )
+
+ cfs = tmp_path / "demo.cfs.csv"
+ ics = tmp_path / "demo.metadata.csv"
+ cfs.write_text("method,flow,amount\n", encoding="utf-8")
+ ics.write_text("method,unit,description\n", encoding="utf-8")
+ assert ab_csv_sibling_path(cfs) == ics
+ assert resolve_ab_csv_pair(cfs) == (cfs, ics)
+ assert resolve_ab_csv_pair(ics) == (cfs, ics)
+
+
+def test_bw2io_metadata_csv_filename_match(tmp_path: Path):
+ from activity_browser.bwutils.impact_categories.bw2io_lcia_file import (
+ read_bw2io_metadata_csv,
+ )
+
+ meta = tmp_path / "metadata.csv"
+ meta.write_text(
+ "filename,method,unit,description\n"
+ "a.csv,Method::A,kg,desc A\n"
+ "b.csv,Method::B,t,desc B\n",
+ encoding="utf-8",
+ )
+ row = read_bw2io_metadata_csv(meta, cf_filename="b.csv")
+ assert row["method"] == "Method::B"
+ assert row["unit"] == "t"
+ assert read_bw2io_metadata_csv(meta, cf_filename="missing.csv") is None
+
+
+def test_bw2io_metadata_csv_single_row_fallback(tmp_path: Path):
+ from activity_browser.bwutils.impact_categories.bw2io_lcia_file import (
+ read_bw2io_metadata_csv,
+ )
+
+ meta = tmp_path / "metadata.csv"
+ meta.write_text(
+ "method,unit,description\none::method,kg,only\n",
+ encoding="utf-8",
+ )
+ row = read_bw2io_metadata_csv(meta, cf_filename="anything.csv")
+ assert row["method"] == "one::method"
+
+
+def test_method_name_to_filename_stem_is_cross_platform():
+ from activity_browser.bwutils.impact_categories import method_name_to_filename_stem
+
+ stem = method_name_to_filename_stem(("IPCC", "climate change", "GWP100"))
+ assert stem == "IPCC__climate change__GWP100"
+ assert ":" not in stem
+ assert "/" not in stem
+ assert "\\" not in stem
+ dirty = method_name_to_filename_stem(("a:b", "c/d", "e|f"))
+ assert ":" not in dirty
+ assert "/" not in dirty
+ assert "|" not in dirty
+
+
+@bw2test
+def test_ab_csv_round_trip(tmp_path: Path):
+ from activity_browser.bwutils.impact_categories import (
+ export_methods_ab_csv_pair,
+ load_ab_csv_pair,
+ )
+
+ write_functional_database("basic", DATABASE, process=True)
+ name = ("climate", "gwp")
+ method = bd.Method(name)
+ method.register(unit="kg", description="GWP demo")
+ method.write([(("basic", "elementary"), 1.5)], process=True)
+ bd.methods.flush()
+
+ export_methods_ab_csv_pair([name], tmp_path / "demo")
+ loaded = load_ab_csv_pair(tmp_path / "demo.cfs.csv")
+ assert len(loaded) == 1
+ assert loaded[0]["name"] == name
+ assert loaded[0]["exchanges"][0]["amount"] == 1.5
+
+
+@bw2test
+def test_bw2io_xlsx_round_trip(tmp_path: Path):
+ from activity_browser.bwutils.impact_categories.bw2io_lcia_file import (
+ export_method_bw2io_xlsx,
+ load_bw2io_lcia_file,
+ read_bw2io_metadata_xlsx,
+ )
+
+ write_functional_database("basic", DATABASE, process=True)
+ name = ("climate", "gwp")
+ method = bd.Method(name)
+ method.register(unit="kg", description="GWP demo")
+ method.write([(("basic", "elementary"), 3.0)], process=True)
+ bd.methods.flush()
+
+ out = tmp_path / "one.xlsx"
+ export_method_bw2io_xlsx(name, out)
+ meta = read_bw2io_metadata_xlsx(out)
+ assert meta["method"] == "climate::gwp"
+ assert meta["filename"] == "one.xlsx"
+ data = load_bw2io_lcia_file(
+ out, name=name, unit=meta["unit"], description=meta["description"]
+ )
+ assert data[0]["exchanges"][0]["amount"] == 3.0
+ assert data[0]["exchanges"][0]["categories"] == ("air",)
+
+
+@bw2test
+def test_ab_xlsx_round_trip_preserves_uncertainty(tmp_path: Path):
+ write_functional_database("basic", DATABASE, process=True)
+ cfs = [
+ (
+ ("basic", "elementary"),
+ {
+ "amount": 2.5,
+ "uncertainty type": LognormalUncertainty.id,
+ "loc": 0.9,
+ "scale": 0.2,
+ "negative": False,
+ },
+ )
+ ]
+ name = ("climate", "gwp")
+ method = bd.Method(name)
+ method.register(unit="kg", description="GWP demo")
+ method.write(cfs, process=True)
+ bd.methods.flush()
+
+ out = tmp_path / "ab-lcia.xlsx"
+ export_methods_ab_xlsx([name], out)
+
+ loaded = load_ab_xlsx(out)
+ assert len(loaded) == 1
+ ds = loaded[0]
+ assert ds["name"] == name
+ assert ds["unit"] == "kg"
+ assert ds["description"] == "GWP demo"
+ assert len(ds["exchanges"]) == 1
+ exc = ds["exchanges"][0]
+ assert exc["name"] == "elementary"
+ assert exc["categories"] == ("air",)
+ assert exc["amount"] == 2.5
+ assert exc["uncertainty type"] == LognormalUncertainty.id
+ assert exc["loc"] == pytest.approx(0.9)
+ assert exc["scale"] == pytest.approx(0.2)
+
+ del bd.methods[name]
+ bd.methods.flush()
+ stats = import_ab_methods(
+ loaded,
+ biosphere_name="basic",
+ conflict_mode=ConflictMode.OVERWRITE,
+ )
+ assert stats.written == 1
+ assert stats.unlinked == 0
+ written = list(bd.Method(name).load())
+ assert len(written) == 1
+ payload = written[0][1]
+ assert isinstance(payload, dict)
+ assert payload["amount"] == 2.5
+ assert payload["uncertainty type"] == LognormalUncertainty.id
+ assert bd.methods[name].get("description") == "GWP demo"
+
+
+@bw2test
+def test_import_ab_methods_blocks_on_unlinked():
+ write_functional_database("basic", DATABASE, process=True)
+ data = [
+ {
+ "name": ("climate", "gwp"),
+ "unit": "kg",
+ "description": "",
+ "filename": "test.xlsx",
+ "exchanges": [
+ {
+ "name": "does-not-exist",
+ "categories": ("air",),
+ "amount": 1.0,
+ }
+ ],
+ }
+ ]
+ stats = import_ab_methods(data, biosphere_name="basic")
+ assert stats.written == 0
+ assert stats.unlinked == 1
+ assert ("climate", "gwp") not in bd.methods
+
+
+@bw2test
+def test_import_ab_methods_drop_unlinked_writes_linked_only():
+ write_functional_database("basic", DATABASE, process=True)
+ data = [
+ {
+ "name": ("climate", "gwp"),
+ "unit": "kg",
+ "description": "",
+ "filename": "test.xlsx",
+ "exchanges": [
+ {
+ "name": "elementary",
+ "categories": ("air",),
+ "amount": 2.0,
+ },
+ {
+ "name": "does-not-exist",
+ "categories": ("air",),
+ "amount": 9.0,
+ },
+ ],
+ }
+ ]
+ stats = import_ab_methods(
+ data, biosphere_name="basic", drop_unlinked=True
+ )
+ assert stats.written == 1
+ assert stats.unlinked == 0
+ cfs = list(bd.Method(("climate", "gwp")).load())
+ assert len(cfs) == 1
+ assert cfs[0][1] == 2.0 or (
+ isinstance(cfs[0][1], dict) and cfs[0][1]["amount"] == 2.0
+ )
diff --git a/tests/test_activity_edit_elementary_flow.py b/tests/test_activity_edit_elementary_flow.py
index ac48791cd..e77d14224 100644
--- a/tests/test_activity_edit_elementary_flow.py
+++ b/tests/test_activity_edit_elementary_flow.py
@@ -3,7 +3,7 @@
from qtpy import QtWidgets
from activity_browser import app
-from activity_browser.app.actions.activity import edit_elementary_flow as edit_mod
+from activity_browser.app.actions.activity import elementary_flow_edit as edit_mod
from activity_browser.bwutils.elementary_flows import create_elementary_flow
from activity_browser.bwutils.commontasks import is_node_biosphere
diff --git a/tests/test_activity_new_elementary_flow.py b/tests/test_activity_new_elementary_flow.py
index ec2f2ebc7..130f5e331 100644
--- a/tests/test_activity_new_elementary_flow.py
+++ b/tests/test_activity_new_elementary_flow.py
@@ -3,7 +3,7 @@
from qtpy import QtWidgets
from activity_browser import app
-from activity_browser.app.actions.activity import new_elementary_flow as mod
+from activity_browser.app.actions.activity import elementary_flow_new as mod
from activity_browser.bwutils.commontasks import is_node_biosphere
diff --git a/tests/test_impact_category_templates.py b/tests/test_impact_category_templates.py
new file mode 100644
index 000000000..a6209712e
--- /dev/null
+++ b/tests/test_impact_category_templates.py
@@ -0,0 +1,28 @@
+"""Impact-category template path/copy smoke tests."""
+from pathlib import Path
+
+from activity_browser.bwutils.impact_categories.templates import (
+ TEMPLATE_FILES,
+ copy_impact_category_template,
+ template_paths,
+)
+
+
+def test_impact_category_template_paths_exist():
+ for kind in TEMPLATE_FILES:
+ paths = template_paths(kind)
+ assert paths
+ assert all(p.is_file() for p in paths)
+
+
+def test_copy_ab_xlsx_template(tmp_path: Path):
+ dest = tmp_path / "out.xlsx"
+ written = copy_impact_category_template("ab-xlsx", dest)
+ assert len(written) == 1
+ assert written[0].is_file()
+
+
+def test_copy_ab_csv_template_pair(tmp_path: Path):
+ written = copy_impact_category_template("ab-csv", tmp_path / "my-lcia")
+ assert len(written) == 2
+ assert all(p.is_file() for p in written)
diff --git a/tests/test_method_import_progress_dialog.py b/tests/test_method_import_progress_dialog.py
new file mode 100644
index 000000000..29bcb6b22
--- /dev/null
+++ b/tests/test_method_import_progress_dialog.py
@@ -0,0 +1,28 @@
+"""Source-level smoke checks for method import progress wiring (no app startup)."""
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+METHOD = ROOT / "activity_browser" / "app" / "actions" / "method"
+APP_DIALOGS = ROOT / "activity_browser" / "app" / "dialogs"
+
+
+def test_method_import_ecoinvent_uses_ui_progress_dialog():
+ src = (METHOD / "method_import_ecoinvent.py").read_text(encoding="utf-8")
+ assert "from activity_browser.ui.dialogs import ABProgressDialog" in src
+ assert "widgets.ABProgressDialog" not in src
+ assert "composites" not in src
+
+
+def test_method_file_actions_use_run_thread_with_progress():
+ progress_src = (APP_DIALOGS / "thread_progress.py").read_text(encoding="utf-8")
+ assert "def run_thread_with_progress(" in progress_src
+
+ for name in (
+ "method_import_ab.py",
+ "method_import_bw2io.py",
+ "method_export_ab.py",
+ "method_export_bw2io.py",
+ ):
+ src = (METHOD / name).read_text(encoding="utf-8")
+ assert "run_thread_with_progress" in src, name
+ assert "from activity_browser.app.dialogs import run_thread_with_progress" in src, name