In Scenario mode, alternative values for flows or parameters can " + "can be defined. Each scenario can thus consider different flow or parameter values defined in two types of scenario files.
" + "A flow-scenario file (scenario difference file) identifies flows (left-side part)" + "(e.g. inputs from one to another activity) and contains scenario values for each flow (right-side part). " + "In an empty file, you can start by adding rows via Copy for scenario file on processes or flows, then paste " + "into a template from Template....
" + "A parameter-scenario file varies Brightway parameters across scenarios." + "The columns are Name and Group (mandatory to identify parameters), then optional default values (as in the database), plus scenario columns. " + "When you load it, AB converts it into flow scenarios for calculation.
" + "Use Add scenarios... more than once. " + "Combine scenarios builds the product of scenario names across files (parameter and flow scenarios can be mixed); " + "Extend scenarios aligns files on shared scenario names.
" + "Download an empty flow or parameter starter (.xlsx or .csv). "
+ "If the project has parameters, the parameter template is filled with "
+ "Name / Group / default and empty example scenario columns. "
+ "Lines or columns starting with # are ignored on import "
+ "(useful for your notes).
Writes the currently loaded, merged flow-scenario table to a file.
" + ) + text.setOpenExternalLinks(False) + + scroll = QtWidgets.QScrollArea(self) + scroll.setWidgetResizable(True) + scroll.setWidget(text) + + buttons = QtWidgets.QDialogButtonBox(QtWidgets.QDialogButtonBox.Ok, parent=self) + buttons.accepted.connect(self.accept) + + layout = QtWidgets.QVBoxLayout(self) + layout.addWidget(scroll) + layout.addWidget(buttons) + diff --git a/activity_browser/bwutils/commontasks.py b/activity_browser/bwutils/commontasks.py index 8aa9be077..48d28815a 100644 --- a/activity_browser/bwutils/commontasks.py +++ b/activity_browser/bwutils/commontasks.py @@ -286,18 +286,33 @@ def biosphere_node_types() -> frozenset[str]: def is_node_product_or_waste(node: tuple | int | bd.Node) -> bool: return is_node_product(node) or is_node_waste(node) + +def _sqlite_functional_production_amount(node: bd.Node) -> float | None: + """Production amount on a sqlite process+reference-product node, if any.""" + productions = list(node.production()) + if not productions: + return None + return productions[0]["amount"] + + def is_node_product(node: tuple | int | bd.Node) -> bool: node = refresh_node(node) raw_type = node._document.type - if raw_type in ["product", "processwithreferenceproduct"]: + if raw_type == "product": return True + if raw_type == "processwithreferenceproduct": + amount = _sqlite_functional_production_amount(node) + # Missing production → keep legacy product behavior + return amount is None or amount >= 0 + if raw_type == "process" and len(node.upstream(kinds=["production"])): return True return False + def is_node_waste(node: tuple | int | bd.Node) -> bool: node = refresh_node(node) raw_type = node._document.type @@ -305,6 +320,10 @@ def is_node_waste(node: tuple | int | bd.Node) -> bool: if raw_type == "waste": return True + if raw_type == "processwithreferenceproduct": + amount = _sqlite_functional_production_amount(node) + return amount is not None and amount < 0 + return False @@ -313,6 +332,21 @@ def is_node_biosphere(node: tuple | int | bd.Node) -> bool: node = refresh_node(node) return node._document.type in biosphere_node_types() + +def get_exchange_type(activity_key: tuple, output: bool = False) -> str | None: + """ + Exchange type (and optional leading ``-`` for negative amount) when dropping + *activity_key* onto an activity's Output (*output*=True) or Input table. + """ + if is_node_product(activity_key): + return "substitution" if output else "technosphere" + if is_node_waste(activity_key): + return "-technosphere" if output else "-substitution" + if is_node_biosphere(activity_key): + return "biosphere" + return None + + def is_node_process(node: tuple | int | bd.Node) -> bool: node = refresh_node(node) raw_type = node._document.type diff --git a/activity_browser/bwutils/superstructure/convert_parameter_to_flow_scenarios.py b/activity_browser/bwutils/superstructure/convert_parameter_to_flow_scenarios.py index 0944fffde..4c4dcebed 100644 --- a/activity_browser/bwutils/superstructure/convert_parameter_to_flow_scenarios.py +++ b/activity_browser/bwutils/superstructure/convert_parameter_to_flow_scenarios.py @@ -183,34 +183,74 @@ def recalculate_activity_parameters( return StaticParameters.prune_result_data(data) +def _exchange_formula(exc: ExchangeDataset) -> str: + """Return a stripped formula from exchange data, if any.""" + formula = "" + if isinstance(getattr(exc, "data", None), dict): + formula = str((exc.data or {}).get("formula") or "").strip() + if not formula: + formula = str(getattr(exc, "formula", "") or "").strip() + return formula + + def exchange_formula_rows_for_selected_groups( selected_groups: set[str], ) -> list[tuple[str, int, str, str | None]]: - """Collect formula-bearing exchanges for selected output databases/groups.""" + """Collect formula-bearing exchanges for selected parameter groups. + + ``selected_groups`` comes from the parameter-scenario ``Group`` column and may + contain activity-parameter group ids, database names, and/or ``project``. + + Activity groups are resolved via ``ParameterizedExchange`` (same source as + ``ParameterManager``). Database-named groups still scan formula exchanges + whose ``output_database`` matches. + """ + from bw2data.parameters import ActivityParameter, ParameterizedExchange + rows = [] + seen: set[int] = set() activity_groups = activity_group_by_output_key() - query = ExchangeDataset.select().where( - ExchangeDataset.output_database << list(selected_groups) - ) - for exc in query: - formula = "" - if isinstance(getattr(exc, "data", None), dict): - formula = str((exc.data or {}).get("formula") or "").strip() - if not formula: - formula = str(getattr(exc, "formula", "") or "").strip() - if formula: + known_act_groups = { + str(ap.group) + for ap in ActivityParameter.select(ActivityParameter.group).distinct() + } + selected_act_groups = selected_groups & known_act_groups + selected_db_groups = selected_groups - known_act_groups - {"project"} + + def _append(exc: ExchangeDataset, formula: str, activity_group: str | None) -> None: + eid = int(exc.id) + if eid in seen or not formula: + return + seen.add(eid) + rows.append( + ( + str(exc.output_database), + eid, + formula, + activity_group, + 0 if str(exc.input_database) == str(exc.output_database) else 1, + ) + ) + + if selected_act_groups: + for pe in ParameterizedExchange.select().where( + ParameterizedExchange.group << list(selected_act_groups) + ): + exc = ExchangeDataset.get_by_id(pe.exchange) + formula = str(pe.formula or "").strip() or _exchange_formula(exc) + _append(exc, formula, str(pe.group)) + + if selected_db_groups: + query = ExchangeDataset.select().where( + ExchangeDataset.output_database << list(selected_db_groups) + ) + for exc in query: + formula = _exchange_formula(exc) activity_group = activity_groups.get( (str(exc.output_database), str(exc.output_code)) ) - rows.append( - ( - str(exc.output_database), - int(exc.id), - formula, - activity_group, - 0 if str(exc.input_database) == str(exc.output_database) else 1, - ) - ) + _append(exc, formula, activity_group) + rows.sort(key=lambda x: (x[0], x[4], x[2], x[1])) return [(g, eid, f, act_group) for g, eid, f, act_group, _prio in rows] diff --git a/activity_browser/bwutils/superstructure/excel.py b/activity_browser/bwutils/superstructure/excel.py index 9f2c33856..ab0ae1b61 100644 --- a/activity_browser/bwutils/superstructure/excel.py +++ b/activity_browser/bwutils/superstructure/excel.py @@ -67,10 +67,9 @@ def import_from_excel( The default index chosen represents the second sheet (first after the 'information' sheet). - Any '*' character used at the start of a row or will cause that row - to be excluded from the import. - A '#' character at the start of a column will cause that column to be - excluded from the import. + A '#' character at the start of a row causes that row to be excluded from + the import. A '#' character at the start of a column name causes that + column to be excluded from the import. 'usecols' is used to exclude specific columns from the excel document. 'comment' is used to exclude specific rows from the excel document. @@ -84,7 +83,7 @@ def import_from_excel( sheet_name=import_sheet, header=header_idx, usecols=valid_cols, - comment="*", + comment="#", na_values="", keep_default_na=False, engine="openpyxl", diff --git a/activity_browser/bwutils/superstructure/file_imports.py b/activity_browser/bwutils/superstructure/file_imports.py index 72c0fc93f..4d81142a8 100644 --- a/activity_browser/bwutils/superstructure/file_imports.py +++ b/activity_browser/bwutils/superstructure/file_imports.py @@ -227,6 +227,7 @@ def read_file(path: Optional[Union[str, Path]], **kwargs): compression="infer", sep=separator, index_col=False, + comment="#", converters={"from key": ast.literal_eval, "to key": ast.literal_eval}, ) # Scenario headers typed as numbers (e.g. 2025) must be strings. diff --git a/activity_browser/bwutils/superstructure/scenario_templates.py b/activity_browser/bwutils/superstructure/scenario_templates.py new file mode 100644 index 000000000..127955482 --- /dev/null +++ b/activity_browser/bwutils/superstructure/scenario_templates.py @@ -0,0 +1,62 @@ +"""Resolve and write calculation-setup scenario starter templates.""" +from __future__ import annotations + +import shutil +from pathlib import Path + +import pandas as pd + +from activity_browser.bwutils.utils import Parameters + +TEMPLATES_DIR = Path(__file__).resolve().parents[2] / "templates" / "scenarios" + +EXAMPLE_SCENARIO_COLS = ("S1 example", "S2 example", "S3 example") + + +def scenario_template_path(kind: str, fmt: str) -> Path: + """Return packaged starter path for ``kind`` in ``{'parameter','flow'}`` and ``fmt`` in ``{'xlsx','csv'}``.""" + if kind not in {"parameter", "flow"}: + raise ValueError(f"Unknown template kind: {kind}") + if fmt not in {"xlsx", "csv"}: + raise ValueError(f"Unknown template format: {fmt}") + name = "parameter-scenarios" if kind == "parameter" else "flow-scenarios" + path = TEMPLATES_DIR / f"{name}.{fmt}" + if not path.is_file(): + raise FileNotFoundError(f"Scenario template not found: {path}") + return path + + +def project_has_parameters() -> bool: + return bool(Parameters.from_bw_parameters()) + + +def parameter_template_dataframe() -> pd.DataFrame: + """Project parameters as a parameter-scenario table with empty example scenario columns.""" + data = [p[:3] for p in Parameters.from_bw_parameters()] + df = pd.DataFrame(data, columns=["Name", "Group", "default"]) + for col in EXAMPLE_SCENARIO_COLS: + df[col] = pd.NA + return df + + +def write_parameter_template(path: Path, df: pd.DataFrame | None = None) -> None: + """Write a parameter-scenario template to ``path`` (.xlsx or .csv).""" + path = Path(path) + table = parameter_template_dataframe() if df is None else df + if path.suffix.lower() == ".csv": + table.to_csv(path, index=False, sep=";") + else: + if path.suffix.lower() not in {".xlsx", ".xls"}: + path = path.with_suffix(".xlsx") + table.to_excel(path, index=False) + + +def copy_scenario_template(kind: str, fmt: str, destination: Path) -> Path: + """Copy a packaged empty starter to ``destination`` (suffix forced to match ``fmt``).""" + source = scenario_template_path(kind, fmt) + destination = Path(destination) + if destination.suffix.lower() != f".{fmt}": + destination = destination.with_suffix(f".{fmt}") + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + return destination diff --git a/activity_browser/templates/README.md b/activity_browser/templates/README.md new file mode 100644 index 000000000..a21902d06 --- /dev/null +++ b/activity_browser/templates/README.md @@ -0,0 +1,35 @@ +# templates + +Bundled spreadsheet templates shipped with Activity Browser (top-level package folder so they are easy to find). + +## Layout + +- **`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/`). + +## Usage + +Package path: `activity_browser/templates/`. Resolve at runtime via: + +```python +from pathlib import Path +import activity_browser + +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). + +Rows or columns whose first cell / header starts with `#` are ignored by scenario import (Excel and CSV). + +**Get template → flow-scenarios** always copies the empty starter file (does not generate from project parameters). + +## Maintenance + +- Keep column headers aligned with `SUPERSTRUCTURE` / parameter-scenario import expectations in `bwutils/superstructure`. +- When adding new `.xlsx` / `.csv` templates, include them in `MANIFEST.in` so they ship in wheels/sdists. diff --git a/activity_browser/templates/scenarios/flow-scenarios.csv b/activity_browser/templates/scenarios/flow-scenarios.csv new file mode 100644 index 000000000..78d1d19f0 --- /dev/null +++ b/activity_browser/templates/scenarios/flow-scenarios.csv @@ -0,0 +1,10 @@ +from activity name;from reference product;from location;from categories;from database;from key;to activity name;to reference product;to location;to categories;to database;to key;flow type;S1 example;S2 example;S3 example +;;;;;;;;;;;;;;; +;;;;;;;;;;;;;;; +# Flow scenario template (CSV);;;;;;;;;;;;;;; +;;;;;;;;;;;;;;; +# Rows or columns whose first cell / header starts with # are ignored when Activity Browser loads the file. Use # to leave notes in the file.;;;;;;;;;;;;;;; +"# How to fill: in Activity Browser, select flows/products, right-click ""Copy for scenario file"", paste under the header row, then enter scenario amounts.";;;;;;;;;;;;;;; +"# Empty scenario cells mean ""use the default inventory amount"".";;;;;;;;;;;;;;; +# Rename S1/S2/S3 example headers to your scenario names as needed.;;;;;;;;;;;;;;; +# Load via calculation setup Scenario mode, Add scenarios...;;;;;;;;;;;;;;; diff --git a/activity_browser/templates/scenarios/flow-scenarios.xlsx b/activity_browser/templates/scenarios/flow-scenarios.xlsx new file mode 100644 index 000000000..ccf709a80 Binary files /dev/null and b/activity_browser/templates/scenarios/flow-scenarios.xlsx differ diff --git a/activity_browser/templates/scenarios/parameter-scenarios.csv b/activity_browser/templates/scenarios/parameter-scenarios.csv new file mode 100644 index 000000000..39cfaf8c1 --- /dev/null +++ b/activity_browser/templates/scenarios/parameter-scenarios.csv @@ -0,0 +1,9 @@ +Name;Group;default;S1 example;S2 example;S3 example +;;;;; +;;;;; +# Parameter scenario template (CSV);;;;; +;;;;; +# Rows or columns whose first cell / header starts with # are ignored when Activity Browser loads the file. Use # to leave notes in the file.;;;;; +"# Columns: Name and Group must match parameters in your project; default is optional; add scenario columns with numeric values.";;;;; +"# Prefer ""Get template"" from the calculation setup Scenarios tools to start from your project parameters when available.";;;;; +# Load via calculation setup Scenario mode, Add scenarios... (AB converts parameter scenarios to flow scenarios for LCA).;;;;; diff --git a/activity_browser/templates/scenarios/parameter-scenarios.xlsx b/activity_browser/templates/scenarios/parameter-scenarios.xlsx new file mode 100644 index 000000000..fe4a405c3 Binary files /dev/null and b/activity_browser/templates/scenarios/parameter-scenarios.xlsx differ diff --git a/tests/fixtures/product_waste_links.py b/tests/fixtures/product_waste_links.py new file mode 100644 index 000000000..9324e8c4e --- /dev/null +++ b/tests/fixtures/product_waste_links.py @@ -0,0 +1,44 @@ +"""Shared expectations for product/waste drop-link LCA cases.""" + +from __future__ import annotations + +# Own CO2 + linked P(10)/W(5) contribution +EXPECTED_SCORES = { + "A": 6.0, # production; add waste treatment W + "Aw": -4.0, # production; substitute waste treatment W + "B": 7.0, # waste treatment; add waste treatment W + "Bw": -3.0, # waste treatment; substitute waste treatment W + "C": 11.0, # production; consume product p + "Cw": -9.0, # production; substitute production of p + "D": 12.0, # waste treatment; consume product p + "Dw": -8.0, # waste treatment; substitute production of p +} + +# (host_code, dragged_code, drop_on_output) +# drop_on_output=True means Output table; False means Input table +LINK_SPECS = ( + ("A", "W", True), # add waste treatment + ("Aw", "W", False), # substitute waste treatment + ("B", "W", True), + ("Bw", "W", False), + ("C", "P", False), # consume product + ("Cw", "P", True), # substitute production + ("D", "P", False), + ("Dw", "P", True), +) + +CO2_OWN = { + "P": 10.0, + "W": 5.0, + "A": 1.0, + "Aw": 1.0, + "B": 2.0, + "Bw": 2.0, + "C": 1.0, + "Cw": 1.0, + "D": 2.0, + "Dw": 2.0, +} + +# Hosts that are waste treatments (functional input / negative production) +WASTE_HOSTS = frozenset({"W", "B", "Bw", "D", "Dw"}) diff --git a/tests/test_parameter_to_flow_scenarios.py b/tests/test_parameter_to_flow_scenarios.py new file mode 100644 index 000000000..0bfc16d41 --- /dev/null +++ b/tests/test_parameter_to_flow_scenarios.py @@ -0,0 +1,136 @@ +"""Parameter scenarios (project / database / activity) → flow scenarios → SuperstructureMLCA.""" + +from __future__ import annotations + +import pandas as pd +import pytest +from bw2data.parameters import ActivityParameter +from bw2data.tests import bw2test + +import bw2data as bd + +from activity_browser.bwutils.superstructure.manager import SuperstructureManager +from activity_browser.bwutils.superstructure.mlca import SuperstructureMLCA +from activity_browser.bwutils.superstructure.utils import parameters_to_sdf +from fixtures.bw_helpers import ( + register_parameter_setup, + write_calculation_setup, + write_functional_database, + write_method, +) +from fixtures.database_roundtrip import functional_data, parameter_setup + +DB_NAME = "fuel_elec" + + +@pytest.fixture +@bw2test +def fuel_elec_with_all_parameter_levels(): + """Elec consumes fuel; fuel emits CO₂. + + Exchange formula: ``share * db_mult * proj_mult`` so each parameter level + can scale the technosphere amount independently. + """ + bio = bd.config.biosphere + bd.Database(bio).register() + bd.Database(bio).write( + { + (bio, "co2"): { + "name": "Carbon dioxide", + "code": "co2", + "unit": "kg", + "type": "emission", + "categories": ("air",), + "database": bio, + "exchanges": [], + } + } + ) + + data = functional_data(DB_NAME, parameters=False) + data[(DB_NAME, "elec_proc")]["exchanges"].append( + {"input": (DB_NAME, "fuel_prod"), "type": "technosphere", "amount": 0.5} + ) + data[(DB_NAME, "fuel_proc")]["exchanges"].append( + {"input": (bio, "co2"), "type": "biosphere", "amount": 1.0} + ) + write_functional_database(DB_NAME, data, process=True) + + bd.parameters.new_project_parameters( + [{"name": "proj_mult", "amount": 1.0, "formula": ""}] + ) + bd.parameters.new_database_parameters( + [{"name": "db_mult", "amount": 1.0, "formula": ""}], + DB_NAME, + ) + setup = parameter_setup(DB_NAME, process_code="elec_proc") + setup["parameterized_exchanges"][0]["formula"] = "share * db_mult * proj_mult" + register_parameter_setup(DB_NAME, setup) + + write_method("GWP", [((bio, "co2"), 1.0)]) + write_calculation_setup( + "cs1", + {"inv": [{(DB_NAME, "elec_prod"): 1}], "ia": [("GWP",)]}, + ) + return DB_NAME + + +def test_project_database_activity_parameter_scenarios_convert_and_run_mlca( + fuel_elec_with_all_parameter_levels, +): + act_group = str(ActivityParameter.get(ActivityParameter.name == "share").group) + param_scenarios = pd.DataFrame( + [ + { + "Name": "proj_mult", + "Group": "project", + "default": 1.0, + "baseline": 1.0, + "project_high": 2.0, + "database_high": 1.0, + "activity_high": 1.0, + }, + { + "Name": "db_mult", + "Group": DB_NAME, + "default": 1.0, + "baseline": 1.0, + "project_high": 1.0, + "database_high": 2.0, + "activity_high": 1.0, + }, + { + "Name": "share", + "Group": act_group, + "default": 0.5, + "baseline": 0.5, + "project_high": 0.5, + "database_high": 0.5, + "activity_high": 1.0, + }, + ] + ) + + flow_df = parameters_to_sdf(param_scenarios) + # Formula share * db_mult * proj_mult + assert float(flow_df["baseline"].iloc[0]) == pytest.approx(0.5) + assert float(flow_df["project_high"].iloc[0]) == pytest.approx(1.0) + assert float(flow_df["database_high"].iloc[0]) == pytest.approx(1.0) + assert float(flow_df["activity_high"].iloc[0]) == pytest.approx(1.0) + + # Same post-processing the CS scenario UI applies before calculate. + flow_df = SuperstructureManager(flow_df).combined_data() + mlca = SuperstructureMLCA("cs1", flow_df) + mlca.calculate() + scores = mlca.lca_scores_to_dataframe() + baseline = float(scores.iloc[0][("GWP", "baseline")]) + assert baseline > 0 + assert float(scores.iloc[0][("GWP", "project_high")]) == pytest.approx( + 2 * baseline + ) + assert float(scores.iloc[0][("GWP", "database_high")]) == pytest.approx( + 2 * baseline + ) + assert float(scores.iloc[0][("GWP", "activity_high")]) == pytest.approx( + 2 * baseline + ) diff --git a/tests/test_product_waste_links.py b/tests/test_product_waste_links.py new file mode 100644 index 000000000..e7b7b39c0 --- /dev/null +++ b/tests/test_product_waste_links.py @@ -0,0 +1,199 @@ +""" +Product vs waste classification and drop-link LCA impacts (sqlite + functional_sqlite). + +One ``@bw2test`` project per backend: classification, ``get_exchange_type``, and the +eight LCA scores share a single inventory build. +""" + +from __future__ import annotations + +import bw2calc as bc +import bw2data as bd +import bw_functional as bf +import pytest +from bw2data.tests import bw2test + +from activity_browser.bwutils.commontasks import ( + get_exchange_type, + is_node_product, + is_node_waste, +) +from fixtures.bw_helpers import write_functional_database, write_method +from fixtures.product_waste_links import ( + CO2_OWN, + EXPECTED_SCORES, + LINK_SPECS, + WASTE_HOSTS, +) + +DB = "pw" +METHOD = ("pw_co2",) +CO2 = (DB, "CO2") + +SQLITE_DROP_CASES = ( + ("W", True, "technosphere", -1.0), + ("W", False, "substitution", -1.0), + ("P", False, "technosphere", 1.0), + ("P", True, "substitution", 1.0), +) +FUNCTIONAL_DROP_CASES = ( + ("w", True, "technosphere", -1.0), + ("w", False, "substitution", -1.0), + ("p", False, "technosphere", 1.0), + ("p", True, "substitution", 1.0), +) + + +def _parse_drop_exchange(dragged_key: tuple, output: bool) -> tuple[str, float]: + """Map ``get_exchange_type`` to (type, amount) as ``ExchangesTab.dropEvent`` does.""" + exc_type = get_exchange_type(dragged_key, output=output) + assert exc_type is not None + if exc_type.startswith("-"): + return exc_type[1:], -1.0 + return exc_type, 1.0 + + +def _add_drop_link(host_key: tuple, dragged_key: tuple, output: bool) -> None: + exc_type, amount = _parse_drop_exchange(dragged_key, output) + host = bd.get_activity(host_key) + host.new_exchange(input=dragged_key, type=exc_type, amount=amount).save() + + +def _fu_demand(fu_key: tuple) -> dict: + if is_node_waste(fu_key): + return {fu_key: -1.0} + return {fu_key: 1.0} + + +def _score(fu_key: tuple) -> float: + lca = bc.LCA(_fu_demand(fu_key), METHOD) + lca.lci() + lca.lcia() + return float(lca.score) + + +def _sqlite_base_nodes() -> dict: + nodes = { + CO2: { + "name": "Carbon dioxide", + "type": "emission", + "unit": "kg", + "categories": ("air",), + "exchanges": [], + }, + } + for code, co2 in CO2_OWN.items(): + is_waste = code in WASTE_HOSTS + nodes[(DB, code)] = { + "name": code, + "type": "processwithreferenceproduct", + "unit": "kg", + "location": "GLO", + "reference product": code.lower(), + "exchanges": [ + { + "input": (DB, code), + "type": "production", + "amount": -1.0 if is_waste else 1.0, + }, + {"input": CO2, "type": "biosphere", "amount": co2}, + ], + } + return nodes + + +def _functional_base_nodes() -> dict: + nodes = { + CO2: { + "name": "Carbon dioxide", + "code": "CO2", + "type": "emission", + "unit": "kg", + "categories": ("air",), + "exchanges": [], + }, + } + for code, co2 in CO2_OWN.items(): + is_waste = code in WASTE_HOSTS + flow_code = code.lower() + nodes[(DB, flow_code)] = { + "name": flow_code, + "type": "waste" if is_waste else "product", + "unit": "kg", + "location": "GLO", + "processor": (DB, code), + "exchanges": [], + } + nodes[(DB, code)] = { + "name": code, + "type": "process", + "location": "GLO", + "exchanges": [ + { + "input": (DB, flow_code), + "type": "production", + "amount": -1.0 if is_waste else 1.0, + }, + {"input": CO2, "type": "biosphere", "amount": co2}, + ], + } + return nodes + + +def _assert_drop_cases(cases: tuple) -> None: + for drag_code, output, expected_type, expected_amount in cases: + got_type, got_amount = _parse_drop_exchange((DB, drag_code), output) + assert got_type == expected_type, drag_code + assert got_amount == expected_amount, drag_code + + +@bw2test +def test_sqlite_product_waste_links(): + # Missing production → product (before writing the full inventory) + bd.Database(DB).write( + { + (DB, "X"): { + "name": "X", + "type": "processwithreferenceproduct", + "unit": "kg", + "location": "GLO", + "reference product": "x", + "exchanges": [], + } + } + ) + assert is_node_product((DB, "X")) + assert not is_node_waste((DB, "X")) + + bd.Database(DB).write(_sqlite_base_nodes()) + + assert is_node_product((DB, "P")) and not is_node_waste((DB, "P")) + assert is_node_waste((DB, "W")) and not is_node_product((DB, "W")) + assert is_node_product((DB, "A")) + assert is_node_waste((DB, "B")) + _assert_drop_cases(SQLITE_DROP_CASES) + + for host, dragged, on_output in LINK_SPECS: + _add_drop_link((DB, host), (DB, dragged), on_output) + bd.Database(DB).process() + write_method(METHOD[0], [(CO2, 1.0)], process=True) + + for host_code, expected in EXPECTED_SCORES.items(): + assert _score((DB, host_code)) == pytest.approx(expected), host_code + + +@bw2test +def test_functional_product_waste_links(): + write_functional_database(DB, _functional_base_nodes(), process=True) + + assert is_node_product((DB, "p")) and not is_node_waste((DB, "p")) + assert is_node_waste((DB, "w")) and not is_node_product((DB, "w")) + _assert_drop_cases(FUNCTIONAL_DROP_CASES) + + for host, dragged, on_output in LINK_SPECS: + _add_drop_link((DB, host), (DB, dragged.lower()), on_output) + bf.FunctionalSQLiteDatabase(DB).process() + write_method(METHOD[0], [(CO2, 1.0)], process=True) + + for host_code, expected in EXPECTED_SCORES.items(): + assert _score((DB, host_code.lower())) == pytest.approx(expected), host_code diff --git a/tests/test_scenario_templates.py b/tests/test_scenario_templates.py new file mode 100644 index 000000000..944ae3d51 --- /dev/null +++ b/tests/test_scenario_templates.py @@ -0,0 +1,28 @@ +"""Scenario starter template helpers (no Qt).""" +from pathlib import Path + +import pandas as pd +from bw2data.tests import bw2test + +from activity_browser.bwutils.superstructure import scenario_templates as scen_tpl + + +def test_scenario_template_paths_exist(): + for kind in ("parameter", "flow"): + for fmt in ("xlsx", "csv"): + path = scen_tpl.scenario_template_path(kind, fmt) + assert path.is_file(), path + + +def test_copy_scenario_template(tmp_path: Path): + dest = tmp_path / "my-flow.xlsx" + out = scen_tpl.copy_scenario_template("flow", "xlsx", dest) + assert out.is_file() + assert out.suffix == ".xlsx" + + +@bw2test +def test_parameter_template_dataframe_has_example_columns(): + df = scen_tpl.parameter_template_dataframe() + assert list(df.columns)[:3] == ["Name", "Group", "default"] + assert list(df.columns)[3:] == list(scen_tpl.EXAMPLE_SCENARIO_COLS)