diff --git a/CHANGELOG.md b/CHANGELOG.md index 31d82acc..d9b6d9ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,43 @@ All notable changes to this project are documented in this file. +## [2.4.9.2] + +### Added +- Added `NewDatabase.write_scenario_array_db_to_brightway` for modern + Brightway. It writes one union database and one compressed `bw_processing` + ZIP containing synchronized technosphere and biosphere arrays, ordered as + `original` followed by the generated scenarios. +- Added `bw_processing >= 1.0` to the Brightway 2.5 optional dependencies. + +### Changed +- Reused the superstructure preparation pipeline for scenario-array export, + preserving existing superstructure CSV, Excel, and Feather behavior while + applying the same loading, validation, duplicate aggregation, reporting, and + cleanup to both export paths. +- Stored only changing matrix coordinates in deterministic sequential arrays; + the written base database retains the `original` state when the ZIP is not + loaded. + +### Fixed +- Included matching production rows when changed self-consumption exchanges + are netted into technosphere diagonals, preventing invalid or singular + scenario matrices. + +### Documentation +- Added a complete three-pathway IMAGE 2050 example for the + `ecoinvent-3.12-cutoff` project to the examples notebook, loading guide, + and README, including multi-activity scoring, base-database validation, + synchronized scenario selection, and wraparound behavior. + +### Tests +- Added unit and orchestration coverage for scenario ordering, values, indices, + technosphere flips, biosphere placement, structural zeros, validation errors, + atomic ZIP output, and production/self-consumption netting. +- Validated the export end to end with IMAGE SSP1-L, SSP2-M, and SSP3-H for + 2050, checking three activities against the original database and confirming + deterministic scenario selection and wraparound. + ## [2.4.9.1] ### Added diff --git a/README.md b/README.md index eb00e79b..485572c1 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,30 @@ How to use it? The best way is to follow [the examples from the Jupyter Notebook](https://github.com/polca/premise/blob/master/examples/examples.ipynb). +Sequential scenario arrays in Brightway 2.5 +-------------------------------------------- + +With modern Brightway (`bw2data >= 4`), one export call can write a union +database plus a compressed, deterministic scenario-array ZIP: + +```python +array_path = ndb.write_scenario_array_db_to_brightway( + name="scenario-ensemble", +) +``` + +The initial array state is `original`; successive `next(lca)` calls select each +generated scenario in `ndb.scenarios` order and then wrap to `original`. +Append `array_path` after the database datapackages passed to `bc.LCA` so that +the joint technosphere and biosphere arrays override the base database. The ZIP +is project-specific and provides deterministic scenario enumeration, not +probability-weighted Monte Carlo sampling. + +See the [complete three-IMAGE-scenario notebook example](https://github.com/polca/premise/blob/master/examples/examples.ipynb) +and the [Brightway loading guide](https://premise.readthedocs.io/en/latest/load.html#sequential-scenario-arrays-modern-brightway) +for database creation, exact activity matching, multi-activity LCIA, and +wraparound/base-database checks. + ## Disclaimer on the Use of IAM-Based Scenarios in Premise It is essential to recognize the nature and diff --git a/docs/load.rst b/docs/load.rst index e4955f5c..ae5d9912 100644 --- a/docs/load.rst +++ b/docs/load.rst @@ -68,6 +68,197 @@ Finally, you can also give a name to the superstructure database: Superstructure databases can only be used by Activity-Browser at the moment. +Sequential scenario arrays (modern Brightway) +********************************************** + +With modern Brightway (``bw2data >= 4``), *premise* can write one union +database and one compressed ``bw_processing`` ZIP for deterministic scenario +enumeration. The package contains joint technosphere and biosphere arrays in +this order: ``original`` first, followed by the generated scenarios in +``ndb.scenarios`` order. + +This export is useful when the same functional unit must be evaluated across +many *premise* scenarios without writing one Brightway database per scenario. +It is separate from the Activity Browser superstructure workflow: no scenario +difference CSV is created. + +Requirements and output +^^^^^^^^^^^^^^^^^^^^^^^ + +The export requires all of the following: + +* modern Brightway (``bw2data >= 4``) and ``bw_processing >= 1.0``; +* an active Brightway project containing the source and configured biosphere + databases; +* at least two generated scenarios with unique labels; and +* a completed scenario transformation, normally ``ndb.update()``. + +Calling ``write_scenario_array_db_to_brightway`` writes the union database to +the active project before resolving the Brightway IDs used by the array ZIP. +The return value is the absolute path to the ZIP. If ``filepath`` is omitted, +the package is written to +``export/scenario arrays/scenario_array_.zip``. When supplied, +``filepath`` is the complete destination filename and must end in ``.zip``; +missing parent directories are created and an existing ZIP is replaced +atomically. + +Complete three-scenario IMAGE example +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +The following example uses the ``ecoinvent-3.12-cutoff`` project and applies +all sector updates to three IMAGE pathways for 2050. Set ``PREMISE_KEY`` in the +environment before running it. The source and biosphere database names must +match the names registered in your project. + +.. code-block:: python + + import os + + import bw2data as bd + + from premise import NewDatabase + + PROJECT = "ecoinvent-3.12-cutoff" + SOURCE_DATABASE = "ecoinvent-3.12-cutoff" + BIOSPHERE_DATABASE = "ecoinvent-3.12-biosphere" + DATABASE_NAME = "premise-image-2050-three-pathways-array" + SCENARIOS = [ + {"model": "image", "pathway": "SSP1-L", "year": 2050}, + {"model": "image", "pathway": "SSP2-M", "year": 2050}, + {"model": "image", "pathway": "SSP3-H", "year": 2050}, + ] + + bd.projects.set_current(PROJECT) + missing = [ + name + for name in (SOURCE_DATABASE, BIOSPHERE_DATABASE) + if name not in bd.databases + ] + if missing: + raise ValueError(f"Missing Brightway databases: {missing}") + + ndb = NewDatabase( + scenarios=SCENARIOS, + source_db=SOURCE_DATABASE, + source_version="3.12", + source_type="brightway", + system_model="cutoff", + biosphere_name=BIOSPHERE_DATABASE, + key=os.environ["PREMISE_KEY"], + ) + ndb.update() # apply all sector transformations + + array_path = ndb.write_scenario_array_db_to_brightway( + name=DATABASE_NAME, + ) + print(array_path) + +The one export call writes both ``DATABASE_NAME`` to the active Brightway +project and the returned array ZIP. Do not call ``write_db_to_brightway`` first +for the same database name. + +Calculate and enumerate scores +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Select activities by their name, reference product, and location, and require +exactly one match. This example evaluates one kilowatt-hour of Swiss +low-voltage electricity with the ecoinvent 3.12 EF v3.1 GWP100 method. + +.. code-block:: python + + import bw2calc as bc + import numpy as np + + from premise.utils import create_scenario_list + + method = ( + "ecoinvent-3.12", + "EF v3.1", + "climate change", + "global warming potential (GWP100)", + ) + if method not in bd.methods: + raise ValueError(f"Missing LCIA method: {method}") + + database = bd.Database(DATABASE_NAME) + matches = [ + activity + for activity in database + if activity.get("name") == "market for electricity, low voltage" + and activity.get("reference product") == "electricity, low voltage" + and activity.get("location") == "CH" + ] + if len(matches) != 1: + raise ValueError(f"Expected one matching activity; found {len(matches)}") + functional_unit = matches[0] + + demand, data_objs, remapping = bd.prepare_lca_inputs( + {functional_unit: 1}, + method=method, + ) + + lca = bc.LCA( + demand, + data_objs=[*data_objs, array_path], + remapping_dicts=remapping, + use_arrays=True, + use_distributions=False, + ) + + lca.lci() + lca.lcia() + scores = [lca.score] # original + + for _ in ndb.scenarios: + next(lca) + scores.append(lca.score) + + labels = ["original", *create_scenario_list(ndb.scenarios)] + print(dict(zip(labels, scores))) + + # The next selection after the final scenario wraps to original. + next(lca) + assert np.isclose(lca.score, scores[0]) + +The returned ZIP must be appended *after* the database datapackages in +``data_objs`` so its values override the changing coordinates in the base +database. The initial LCA represents ``original``. Each ``next(lca)`` advances +the technosphere and biosphere arrays together to the next complete scenario; +the selection after the final scenario wraps to ``original``. + +To confirm that the written database remains at its original values without +the overlay, create a second LCA with the unmodified ``data_objs``: + +.. code-block:: python + + base_lca = bc.LCA( + demand, + data_objs=data_objs, + remapping_dicts=remapping, + use_arrays=False, + use_distributions=False, + ) + base_lca.lci() + base_lca.lcia() + assert np.isclose(base_lca.score, scores[0]) + +Only coordinates that vary across ``original`` and the generated scenarios are +stored in the ZIP. Existing exchange uncertainty distributions are not copied +to the scenario arrays, and this version does not combine array selection with +parameter uncertainty, weights, seeds, or random sampling. + +.. warning:: + + The ZIP is tied to the active Brightway project and to the IDs assigned to + the written database. Regenerate it after moving, deleting, or rewriting + that database. These arrays enumerate scenarios deterministically; they are + not probability-weighted Monte Carlo samples. + +The repository's `examples notebook`_ includes the complete three-activity +workflow and reference scores from an ecoinvent 3.12 validation run. + +.. _examples notebook: https://github.com/polca/premise/blob/master/examples/examples.ipynb + As sparse matrices ------------------ diff --git a/examples/examples.ipynb b/examples/examples.ipynb index 960c1cc6..65adf2b5 100644 --- a/examples/examples.ipynb +++ b/examples/examples.ipynb @@ -760,6 +760,313 @@ "![Example superstructure](example_superDB.png)" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### Sequential scenario arrays for Brightway 2.5\n", + "\n", + "This tutorial is for users who want to compare the same functional unit across several `premise` scenarios without writing one Brightway database per scenario.\n", + "\n", + "**Prerequisites**\n", + "\n", + "- `premise[bw25]` with `bw2data >= 4` and `bw_processing >= 1.0`;\n", + "- a licensed ecoinvent 3.12 cutoff database and its biosphere registered in the `ecoinvent-3.12-cutoff` project;\n", + "- the ecoinvent 3.12 EF v3.1 LCIA methods; and\n", + "- a valid IAM data key in the `PREMISE_KEY` environment variable.\n", + "\n", + "By the end, you will have applied every sector update to three IMAGE 2050 pathways, written one union database and one compressed array ZIP, scored three activities, and verified both the original state and sequential wraparound. A full build is computationally intensive; keep the existing `ndb` object if you want to try more functional units.\n", + "\n", + "The array export is a modern-Brightway alternative to the Activity Browser scenario-difference file above. It does **not** create a scenario-difference CSV." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### 1. Configure and validate the project\n", + "\n", + "Use exact database and LCIA method names. The three scenario dictionaries define the array-column order after `original`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import os\n", + "\n", + "import bw2calc as bc\n", + "import bw2data as bd\n", + "import numpy as np\n", + "import pandas as pd\n", + "\n", + "from premise import NewDatabase\n", + "from premise.utils import create_scenario_list\n", + "\n", + "PROJECT = \"ecoinvent-3.12-cutoff\"\n", + "SOURCE_DATABASE = \"ecoinvent-3.12-cutoff\"\n", + "BIOSPHERE_DATABASE = \"ecoinvent-3.12-biosphere\"\n", + "DATABASE_NAME = \"premise-image-2050-three-pathways-array\"\n", + "SCENARIOS = [\n", + " {\"model\": \"image\", \"pathway\": \"SSP1-L\", \"year\": 2050},\n", + " {\"model\": \"image\", \"pathway\": \"SSP2-M\", \"year\": 2050},\n", + " {\"model\": \"image\", \"pathway\": \"SSP3-H\", \"year\": 2050},\n", + "]\n", + "METHOD = (\n", + " \"ecoinvent-3.12\",\n", + " \"EF v3.1\",\n", + " \"climate change\",\n", + " \"global warming potential (GWP100)\",\n", + ")\n", + "\n", + "if \"PREMISE_KEY\" not in os.environ:\n", + " raise RuntimeError(\"Set PREMISE_KEY before running this tutorial.\")\n", + "\n", + "bd.projects.set_current(PROJECT)\n", + "missing_databases = [\n", + " name\n", + " for name in (SOURCE_DATABASE, BIOSPHERE_DATABASE)\n", + " if name not in bd.databases\n", + "]\n", + "if missing_databases:\n", + " raise ValueError(f\"Missing Brightway databases: {missing_databases}\")\n", + "if METHOD not in bd.methods:\n", + " raise ValueError(f\"Missing LCIA method: {METHOD}\")\n", + "\n", + "scenario_labels = [\"original\", *create_scenario_list(SCENARIOS)]\n", + "scenario_labels" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### 2. Build all three scenarios\n", + "\n", + "`ndb.update()` applies every available sector transformation. Use the same source database, system model, and biosphere that are registered in the active project. Never hard-code or publish the IAM key." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "ndb = NewDatabase(\n", + " scenarios=[scenario.copy() for scenario in SCENARIOS],\n", + " source_db=SOURCE_DATABASE,\n", + " source_version=\"3.12\",\n", + " source_type=\"brightway\",\n", + " system_model=\"cutoff\",\n", + " biosphere_name=BIOSPHERE_DATABASE,\n", + " key=os.environ[\"PREMISE_KEY\"],\n", + ")\n", + "ndb.update()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### 3. Write the database and array ZIP\n", + "\n", + "This one call writes `DATABASE_NAME` to the active Brightway project and returns the absolute path to the compressed array package. With no `filepath`, the default is `export/scenario arrays/scenario_array_.zip`. If you provide `filepath`, it is the complete destination path and must end in `.zip`; missing parent directories are created and an existing ZIP is replaced atomically.\n", + "\n", + "At least two generated scenarios and unique scenario labels are required. The base database represents `original`, and the ZIP stores only coordinates that change in at least one scenario." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "array_path = ndb.write_scenario_array_db_to_brightway(\n", + " name=DATABASE_NAME,\n", + ")\n", + "print(f\"Database: {DATABASE_NAME}\")\n", + "print(f\"Scenario arrays: {array_path}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### 4. Match activities exactly and calculate scores\n", + "\n", + "Brightway databases can contain similar activities in several locations. Match the activity name, reference product, and location together, and fail unless there is exactly one result.\n", + "\n", + "The returned ZIP must be appended **after** `data_objs` so that its values override the changing coordinates in the written database. `use_arrays=True` enables column selection; `use_distributions=False` keeps this example deterministic." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "def find_activity(database, *, name, product, location):\n", + " matches = [\n", + " activity\n", + " for activity in database\n", + " if activity.get(\"name\") == name\n", + " and activity.get(\"reference product\") == product\n", + " and activity.get(\"location\") == location\n", + " ]\n", + " if len(matches) != 1:\n", + " keys = [activity.key for activity in matches[:20]]\n", + " raise ValueError(\n", + " f\"Expected one activity for {name!r}, {product!r}, {location!r}; \"\n", + " f\"found {len(matches)}: {keys}\"\n", + " )\n", + " return matches[0]\n", + "\n", + "\n", + "def calculate_scenario_scores(activity):\n", + " demand, data_objs, remapping = bd.prepare_lca_inputs(\n", + " {activity: 1},\n", + " method=METHOD,\n", + " )\n", + " lca = bc.LCA(\n", + " demand,\n", + " data_objs=[*data_objs, array_path],\n", + " remapping_dicts=remapping,\n", + " use_arrays=True,\n", + " use_distributions=False,\n", + " )\n", + " lca.lci()\n", + " lca.lcia()\n", + " scores = [float(lca.score)] # original\n", + " for _ in ndb.scenarios:\n", + " next(lca)\n", + " scores.append(float(lca.score))\n", + "\n", + " next(lca)\n", + " if not np.isclose(lca.score, scores[0], rtol=1e-6, atol=1e-9):\n", + " raise AssertionError(\"Scenario selection did not wrap to original.\")\n", + "\n", + " base_lca = bc.LCA(\n", + " demand,\n", + " data_objs=data_objs,\n", + " remapping_dicts=remapping,\n", + " use_arrays=False,\n", + " use_distributions=False,\n", + " )\n", + " base_lca.lci()\n", + " base_lca.lcia()\n", + " if not np.isclose(base_lca.score, scores[0], rtol=1e-6, atol=1e-9):\n", + " raise AssertionError(\"The original array state differs from the base database.\")\n", + "\n", + " return scores" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "ACTIVITIES = [\n", + " {\n", + " \"name\": \"market for electricity, low voltage\",\n", + " \"product\": \"electricity, low voltage\",\n", + " \"location\": \"CH\",\n", + " },\n", + " {\n", + " \"name\": \"market for diesel, low-sulfur\",\n", + " \"product\": \"diesel, low-sulfur\",\n", + " \"location\": \"CH\",\n", + " },\n", + " {\n", + " \"name\": \"market for steel, low-alloyed\",\n", + " \"product\": \"steel, low-alloyed\",\n", + " \"location\": \"GLO\",\n", + " },\n", + "]\n", + "\n", + "database = bd.Database(DATABASE_NAME)\n", + "rows = []\n", + "for selector in ACTIVITIES:\n", + " activity = find_activity(database, **selector)\n", + " for label, score in zip(scenario_labels, calculate_scenario_scores(activity)):\n", + " rows.append(\n", + " {\n", + " \"activity\": activity.get(\"name\"),\n", + " \"reference product\": activity.get(\"reference product\"),\n", + " \"location\": activity.get(\"location\"),\n", + " \"activity unit\": activity.get(\"unit\"),\n", + " \"scenario\": label,\n", + " \"score\": score,\n", + " }\n", + " )\n", + "\n", + "scores = pd.DataFrame(rows)\n", + "scores.pivot(\n", + " index=[\"activity\", \"reference product\", \"location\", \"activity unit\"],\n", + " columns=\"scenario\",\n", + " values=\"score\",\n", + ").reindex(columns=scenario_labels)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### Reference result\n", + "\n", + "The table below records one end-to-end validation with `premise` 2.4.9.2, ecoinvent 3.12 cutoff, and the EF v3.1 GWP100 method. Scores are in kg CO2-eq per activity unit and are rounded for readability; they can change when `premise`, IAM files, inventories, or LCIA methods change.\n", + "\n", + "| Functional unit | original | IMAGE SSP1-L 2050 | IMAGE SSP2-M 2050 | IMAGE SSP3-H 2050 |\n", + "|---|---:|---:|---:|---:|\n", + "| Swiss low-voltage electricity, 1 kWh | 0.031382 | 0.037540 | 0.037347 | 0.117412 |\n", + "| Swiss low-sulfur diesel, 1 kg | 0.970115 | 0.787928 | 0.815387 | 0.827007 |\n", + "| Global low-alloyed steel, 1 kg | 2.027076 | 1.266054 | 1.549705 | 1.667573 |\n", + "\n", + "The checks in `calculate_scenario_scores` also establish two important properties: loading without the ZIP leaves the written database at `original`, and one more `next(lca)` after the last scenario returns to `original`. Technosphere and biosphere arrays always advance together." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### Lifecycle, interpretation, and common pitfalls\n", + "\n", + "- The ZIP contains Brightway node IDs from the active project. Regenerate it after moving, deleting, or rewriting the database; it is not a portable inventory archive.\n", + "- Append the ZIP last. If it is omitted or placed before an overlapping database datapackage, the scenario values will not override the base coordinates as intended.\n", + "- Scenario selection is deterministic enumeration, not probability-weighted Monte Carlo. This version does not expose random sampling, weights, seeds, or combined parameter uncertainty.\n", + "- Keep technosphere and biosphere selection joint by using the single returned package. Do not advance matrix resources independently.\n", + "- Existing exchange uncertainty metadata is outside the scenario-array package.\n", + "\n", + "For the public API contract and a shorter single-activity example, see the [Brightway loading guide](https://premise.readthedocs.io/en/latest/load.html#sequential-scenario-arrays-modern-brightway)." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### Exercise: score another activity without rebuilding\n", + "\n", + "Choose another activity from `DATABASE_NAME`, identify it by the same three fields, and pass it to `calculate_scenario_scores`. Reuse `array_path` and the written database; scenario generation and export do not need to run again." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Answer scaffold: replace these three fields with one unique activity.\n", + "exercise_selector = {\n", + " \"name\": \"market for electricity, low voltage\",\n", + " \"product\": \"electricity, low voltage\",\n", + " \"location\": \"CH\",\n", + "}\n", + "exercise_activity = find_activity(database, **exercise_selector)\n", + "dict(zip(scenario_labels, calculate_scenario_scores(exercise_activity)))" + ] + }, { "cell_type": "markdown", "metadata": {}, diff --git a/premise/export.py b/premise/export.py index 40cc1eb4..45c70f92 100644 --- a/premise/export.py +++ b/premise/export.py @@ -748,6 +748,10 @@ def generate_scenario_difference_file( ) inds_std = sparse.argwhere((m[..., 1:] == m[..., 0, None]).all(axis=-1).T == False) + inds_std = _include_production_rows_for_changing_self_consumption( + indices=inds_std, + acts_ind=acts_ind, + ) for i in inds_std: c_name, c_ref, c_cat, c_loc, c_unit, _ = acts_ind[i[0]] @@ -889,7 +893,7 @@ def _resolve_superstructure_flow_type(flow_types: pd.Series) -> str: if len(unique_flow_types) == 1: return unique_flow_types.pop() - if unique_flow_types == {"production", "technosphere"}: + if _is_production_consumption_flow_type_set(unique_flow_types): # A self-loop technosphere exchange nets against the diagonal # production exchange and should stay represented as a production row. return "production" @@ -899,6 +903,44 @@ def _resolve_superstructure_flow_type(flow_types: pd.Series) -> str: ) +def _is_production_consumption_flow_type_set(flow_types) -> bool: + """Return whether flow types describe production plus one negative input.""" + + flow_types = set(flow_types) + return ( + len(flow_types) == 2 + and "production" in flow_types + and bool(flow_types & {"technosphere", "generic consumption"}) + ) + + +def _include_production_rows_for_changing_self_consumption( + indices, acts_ind: dict +) -> list[tuple[int, int]]: + """Include unchanged production rows needed to net changing diagonal inputs.""" + + coordinates = [tuple(map(int, index)) for index in indices] + coordinate_set = set(coordinates) + negative_edge_types = {"technosphere", "generic consumption"} + + for consumer_index, supplier_index in coordinates: + consumer = acts_ind[consumer_index] + supplier = acts_ind[supplier_index] + same_activity = all( + supplier[position] == consumer[position] for position in (0, 1, 3, 4) + ) + production_coordinate = (consumer_index, consumer_index) + if ( + supplier[-1] in negative_edge_types + and same_activity + and production_coordinate not in coordinate_set + ): + coordinates.append(production_coordinate) + coordinate_set.add(production_coordinate) + + return coordinates + + def _net_superstructure_scenario_values( df: pd.DataFrame, scenario_columns: list[str], group_columns: list[str] ) -> pd.DataFrame: @@ -917,13 +959,9 @@ def _net_superstructure_scenario_values( ) invalid_flow_type_sets = flow_type_sets[ - ~flow_type_sets["flow type set"].isin( - [ - frozenset({"biosphere"}), - frozenset({"technosphere"}), - frozenset({"production"}), - frozenset({"production", "technosphere"}), - ] + ~flow_type_sets["flow type set"].map( + lambda flow_type_set: len(flow_type_set) == 1 + or _is_production_consumption_flow_type_set(flow_type_set) ) ] if not invalid_flow_type_sets.empty: @@ -939,13 +977,17 @@ def _net_superstructure_scenario_values( ) signed_df = df.merge(flow_type_sets, on=group_columns, how="left") - mixed_self_loops = signed_df["flow type set"] == frozenset( - {"production", "technosphere"} + mixed_self_loops = signed_df["flow type set"].map( + _is_production_consumption_flow_type_set ) signed_df.loc[ - mixed_self_loops & (signed_df["flow type"] == "technosphere"), scenario_columns + mixed_self_loops + & signed_df["flow type"].isin({"technosphere", "generic consumption"}), + scenario_columns, ] = signed_df.loc[ - mixed_self_loops & (signed_df["flow type"] == "technosphere"), scenario_columns + mixed_self_loops + & signed_df["flow type"].isin({"technosphere", "generic consumption"}), + scenario_columns, ].mul( -1 ) @@ -1031,33 +1073,16 @@ def generate_superstructure_db( :return: a superstructure database """ - print("Building superstructure database...") - - # create the dataframe - df, new_db, _ = generate_scenario_difference_file( + new_db, df = _build_superstructure_db( origin_db=origin_db, scenarios=scenarios, db_name=db_name, biosphere_name=biosphere_name, - version=version, scenario_list=scenario_list, + version=version, + preserve_original_column=preserve_original_column, ) - # remove unneeded columns "to unit" - df = df.drop(columns=["to unit"]) - - # rename column "from unit" to "unit" - df = df.rename(columns={"from unit": "unit"}) - - # remove the column `original` - if not preserve_original_column: - df = df.drop(columns=["original"]) - else: - scenario_list = ["original"] + scenario_list - - if "unit" in df.columns: - df = df.drop(columns=["unit"]) - if filepath is not None: filepath = Path(filepath) else: @@ -1066,20 +1091,6 @@ def generate_superstructure_db( if not os.path.exists(filepath): os.makedirs(filepath) - df, exact_duplicates, duplicate_collisions = ( - _aggregate_duplicate_superstructure_rows( - df=df, - scenario_columns=scenario_list, - ) - ) - - if exact_duplicates: - print(f"Dropped {exact_duplicates} exact duplicate(s).") - if duplicate_collisions: - print( - f"Collapsed {duplicate_collisions} overlapping row(s) by netting scenario values." - ) - # if df is longer than the row limit of Excel, # the export to Excel is not an option if len(df) > 1048576: @@ -1105,6 +1116,56 @@ def generate_superstructure_db( return new_db +def _build_superstructure_db( + origin_db, + scenarios, + db_name, + biosphere_name, + version, + scenario_list, + preserve_original_column: bool = True, +) -> tuple[List[dict], pd.DataFrame]: + """Build a union database and finalized scenario dataframe without writing files.""" + + print("Building superstructure database...") + + df, new_db, _ = generate_scenario_difference_file( + origin_db=origin_db, + scenarios=scenarios, + db_name=db_name, + biosphere_name=biosphere_name, + version=version, + scenario_list=scenario_list, + ) + + df = df.drop(columns=["to unit"]) + df = df.rename(columns={"from unit": "unit"}) + if "unit" in df.columns: + df = df.drop(columns=["unit"]) + + if preserve_original_column: + scenario_columns = ["original", *scenario_list] + else: + df = df.drop(columns=["original"]) + scenario_columns = scenario_list + + df, exact_duplicates, duplicate_collisions = ( + _aggregate_duplicate_superstructure_rows( + df=df, + scenario_columns=scenario_columns, + ) + ) + + if exact_duplicates: + print(f"Dropped {exact_duplicates} exact duplicate(s).") + if duplicate_collisions: + print( + f"Collapsed {duplicate_collisions} overlapping row(s) by netting scenario values." + ) + + return new_db, df + + def check_geographical_linking(scenario, original_database): # geo = Geomap(scenario["model"]) diff --git a/premise/new_database.py b/premise/new_database.py index 77244ae9..8d725047 100644 --- a/premise/new_database.py +++ b/premise/new_database.py @@ -30,6 +30,7 @@ from .final_energy import _update_final_energy from .export import ( Export, + _build_superstructure_db, _prepare_database, build_datapackage, generate_scenario_factor_file, @@ -50,6 +51,10 @@ from .metals import _update_metals from .mining import _update_mining from .report import generate_change_report, generate_summary_report +from .scenario_array import ( + _load_scenario_array_dependencies, + _write_scenario_array_datapackage, +) from .steel import _update_steel from .transport import _update_vehicles from .utils import ( @@ -1274,13 +1279,126 @@ def write_superstructure_db_to_brightway( :return: filepath of the "scenarios difference file" """ - if len(self.scenarios) < 2: + self._prepare_superstructure_export( + name=name, + filepath=filepath, + file_format=file_format, + preserve_original_column=preserve_original_column, + ) + + write_brightway_database( + data=self.database, + name=name, + fast=True, + check_internal=False, + ) + + self._finalize_superstructure_export() + + def write_scenario_array_db_to_brightway( + self, + name: str = f"scenario_array_db_{datetime.now():%d-%m-%Y}", + filepath: str | Path | None = None, + ) -> Path: + """Write a union database and deterministic Brightway scenario arrays. + + ``filepath`` is the complete destination ZIP path. The returned package + is project-specific because its matrix indices refer to IDs assigned + when ``name`` is written in the active Brightway project. + """ + + version = bw2data.__version__ + major_version = ( + int(version[0]) + if isinstance(version, (tuple, list)) + else Version(str(version)).major + ) + if major_version < 4: + raise NotImplementedError( + "Scenario-array export requires modern Brightway (bw2data >= 4)." + ) + + self._validate_superstructure_export_prerequisites() + + destination = Path(filepath).expanduser() if filepath is not None else None + if destination is not None and destination.suffix.lower() != ".zip": raise ValueError( - "At least two scenarios are needed to" - "create a super-structure database." + "Scenario-array filepath must be the complete destination path " + "with a '.zip' suffix." ) - check_presence_biosphere_database(self.biosphere_name) + scenario_labels = create_scenario_list(self.scenarios) + duplicates = sorted( + {label for label in scenario_labels if scenario_labels.count(label) > 1} + ) + if duplicates: + raise ValueError( + "Scenario labels must be unique for scenario-array export. " + f"Duplicate label(s): {duplicates}." + ) + + dependencies = _load_scenario_array_dependencies() + bw_processing = dependencies[0] + if destination is None: + sanitized_name = bw_processing.clean_datapackage_name(name) or "database" + destination = ( + Path.cwd() + / "export" + / "scenario arrays" + / f"scenario_array_{sanitized_name}.zip" + ) + + scenario_labels, dataframe = self._prepare_superstructure_export( + name=name, + scenario_array=True, + prerequisites_validated=True, + ) + + write_brightway_database( + data=self.database, + name=name, + fast=True, + check_internal=False, + ) + + ordered_labels = ["original", *scenario_labels] + project_name = getattr(bw2data.projects, "current", None) + metadata = { + "database_name": name, + "brightway_project": project_name, + "source_database": getattr(self, "source", None), + "ecoinvent_version": self.version, + "premise_version": ".".join(map(str, __version__)), + "scenario_count": len(ordered_labels), + "scenario_labels": ordered_labels, + } + destination = _write_scenario_array_datapackage( + dataframe=dataframe, + scenario_labels=ordered_labels, + filepath=destination, + name=name, + metadata=metadata, + dependencies=dependencies, + ) + + self._finalize_superstructure_export() + return destination + + def _prepare_superstructure_export( + self, + *, + name: str, + filepath: str | Path | None = None, + file_format: str = "csv", + preserve_original_column: bool = False, + scenario_array: bool = False, + prerequisites_validated: bool = False, + ) -> tuple[list[str], object]: + """Run the common preparation path for both superstructure exporters.""" + + if not prerequisites_validated: + self._validate_superstructure_export_prerequisites() + original_database = self._load_original_database() for scenario in self.scenarios: @@ -1304,19 +1422,29 @@ def write_superstructure_db_to_brightway( "The database is not ready for export: MAJOR anomalies found. Check the change report." ) - list_scenarios = create_scenario_list(self.scenarios) - - self.database = generate_superstructure_db( - origin_db=original_database, - scenarios=self.scenarios, - db_name=name, - biosphere_name=self.biosphere_name, - filepath=filepath, - version=self.version, - file_format=file_format, - scenario_list=list_scenarios, - preserve_original_column=preserve_original_column, - ) + scenario_labels = create_scenario_list(self.scenarios) + dataframe = None + if scenario_array: + self.database, dataframe = _build_superstructure_db( + origin_db=original_database, + scenarios=self.scenarios, + db_name=name, + biosphere_name=self.biosphere_name, + version=self.version, + scenario_list=scenario_labels, + ) + else: + self.database = generate_superstructure_db( + origin_db=original_database, + scenarios=self.scenarios, + db_name=name, + biosphere_name=self.biosphere_name, + filepath=filepath, + version=self.version, + file_format=file_format, + scenario_list=scenario_labels, + preserve_original_column=preserve_original_column, + ) tmp_scenario = self.scenarios[0].copy() tmp_scenario["database"] = self.database @@ -1338,12 +1466,21 @@ def write_superstructure_db_to_brightway( version=self.version, ) - write_brightway_database( - data=self.database, - name=name, - fast=True, - check_internal=False, - ) + return scenario_labels, dataframe + + def _validate_superstructure_export_prerequisites(self) -> None: + """Validate prerequisites shared by both superstructure exporters.""" + + if len(self.scenarios) < 2: + raise ValueError( + "At least two scenarios are needed to " + "create a super-structure database." + ) + + check_presence_biosphere_database(self.biosphere_name) + + def _finalize_superstructure_export(self) -> None: + """Generate reports and release scenario export state once.""" if self.generate_reports: # generate scenario report diff --git a/premise/scenario_array.py b/premise/scenario_array.py new file mode 100644 index 00000000..7168874a --- /dev/null +++ b/premise/scenario_array.py @@ -0,0 +1,257 @@ +"""Build sequential Brightway scenario-array datapackages.""" + +import os +import tempfile +import zipfile +from pathlib import Path +from typing import Callable + +import numpy as np +import pandas as pd + + +def _load_scenario_array_dependencies(): + """Import modern-Brightway-only dependencies lazily.""" + + try: + import bw_processing + from bw2data import get_id + from bw2data.configuration import labels + from fsspec.implementations.zip import ZipFileSystem + except ImportError as error: + raise ImportError( + "Scenario-array export requires the modern Brightway dependencies. " + "Install premise with the 'bw25' optional dependencies." + ) from error + + return bw_processing, get_id, labels, ZipFileSystem + + +def _missing_key(key) -> bool: + return key is None or (isinstance(key, float) and np.isnan(key)) + + +def _exchange_identity(row: pd.Series) -> str: + return ( + f"{row.get('from activity name')!r} ({row.get('from key')!r}) -> " + f"{row.get('to activity name')!r} ({row.get('to key')!r}), " + f"flow type {row.get('flow type')!r}" + ) + + +def _resolve_key( + *, + key, + role: str, + row: pd.Series, + get_id: Callable, + database_name: str, + project_name: str, +) -> int: + """Resolve one Brightway key and add export context to lookup failures.""" + + try: + if _missing_key(key): + raise KeyError(f"Missing {role} key") + return get_id(key) + except Exception as error: + raise KeyError( + f"Could not resolve the {role} key {key!r} for exchange " + f"{_exchange_identity(row)} after writing database {database_name!r} " + f"in Brightway project {project_name!r}." + ) from error + + +def _scenario_dataframe_to_arrays( + *, + dataframe: pd.DataFrame, + scenario_labels: list[str], + get_id: Callable, + indices_dtype: np.dtype, + biosphere_edge_types, + technosphere_negative_edge_types, + technosphere_positive_edge_types, + database_name: str, + project_name: str, +) -> dict[str, dict[str, np.ndarray]]: + """Convert finalized scenario rows to Brightway matrix array resources.""" + + missing_columns = [ + column for column in scenario_labels if column not in dataframe.columns + ] + if missing_columns: + raise ValueError( + f"Scenario dataframe is missing value columns: {missing_columns}." + ) + + values = dataframe.loc[:, scenario_labels].to_numpy(dtype=np.float64) + if not np.isfinite(values).all(): + raise ValueError("Scenario arrays cannot contain non-finite values.") + + varying = ~np.all(values == values[:, :1], axis=1) + dataframe = dataframe.loc[varying].reset_index(drop=True) + values = values[varying] + + if dataframe.empty: + raise ValueError( + "Cannot export scenario arrays because no exchanges change across scenarios." + ) + + biosphere_edge_types = set(biosphere_edge_types) + negative_edge_types = set(technosphere_negative_edge_types) + positive_edge_types = set(technosphere_positive_edge_types) + supported_edge_types = ( + biosphere_edge_types | negative_edge_types | positive_edge_types + ) + + unsupported = sorted( + { + flow_type + for flow_type in dataframe["flow type"] + if flow_type not in supported_edge_types + }, + key=lambda value: str(value), + ) + if unsupported: + raise ValueError( + f"Unsupported Brightway flow type(s) in scenario arrays: {unsupported}." + ) + + rows = np.empty(len(dataframe), dtype=np.int64) + cols = np.empty(len(dataframe), dtype=np.int64) + for position, (_, row) in enumerate(dataframe.iterrows()): + rows[position] = _resolve_key( + key=row["from key"], + role="from", + row=row, + get_id=get_id, + database_name=database_name, + project_name=project_name, + ) + cols[position] = _resolve_key( + key=row["to key"], + role="to", + row=row, + get_id=get_id, + database_name=database_name, + project_name=project_name, + ) + + resources = {} + flow_types = dataframe["flow type"].to_numpy() + matrix_masks = { + "biosphere_matrix": np.fromiter( + (flow_type in biosphere_edge_types for flow_type in flow_types), + dtype=bool, + count=len(flow_types), + ), + "technosphere_matrix": np.fromiter( + ( + flow_type in negative_edge_types or flow_type in positive_edge_types + for flow_type in flow_types + ), + dtype=bool, + count=len(flow_types), + ), + } + + for matrix, mask in matrix_masks.items(): + if not mask.any(): + continue + + indices = np.empty(int(mask.sum()), dtype=indices_dtype) + indices["row"] = rows[mask] + indices["col"] = cols[mask] + resource = { + "data_array": np.ascontiguousarray(values[mask], dtype=np.float64), + "indices_array": indices, + } + if matrix == "technosphere_matrix": + resource["flip_array"] = np.fromiter( + (flow_type in negative_edge_types for flow_type in flow_types[mask]), + dtype=bool, + count=int(mask.sum()), + ) + resources[matrix] = resource + + if not resources: + raise ValueError( + "Cannot export scenario arrays because no supported matrix coordinates change." + ) + + return resources + + +def _write_scenario_array_datapackage( + *, + dataframe: pd.DataFrame, + scenario_labels: list[str], + filepath: Path, + name: str, + metadata: dict, + dependencies=None, +) -> Path: + """Write a compressed datapackage to a temporary file and atomically replace it.""" + + bw_processing, get_id, labels, ZipFileSystem = ( + dependencies or _load_scenario_array_dependencies() + ) + resources = _scenario_dataframe_to_arrays( + dataframe=dataframe, + scenario_labels=scenario_labels, + get_id=get_id, + indices_dtype=bw_processing.INDICES_DTYPE, + biosphere_edge_types=labels.biosphere_edge_types, + technosphere_negative_edge_types=labels.technosphere_negative_edge_types, + technosphere_positive_edge_types=labels.technosphere_positive_edge_types, + database_name=name, + project_name=metadata["brightway_project"], + ) + + filepath = filepath.expanduser().resolve() + filepath.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{filepath.stem}-", + suffix=".zip", + dir=filepath.parent, + ) + os.close(descriptor) + temporary_path = Path(temporary_name) + filesystem = None + + try: + filesystem = ZipFileSystem( + str(temporary_path), mode="w", compression=zipfile.ZIP_DEFLATED + ) + datapackage_name = ( + bw_processing.clean_datapackage_name(name) or "scenario_array" + ) + datapackage = bw_processing.create_datapackage( + fs=filesystem, + name=datapackage_name, + metadata=metadata, + sequential=True, + sum_intra_duplicates=False, + sum_inter_duplicates=False, + ) + + for matrix, resource in resources.items(): + datapackage.add_persistent_array( + matrix=matrix, + name=bw_processing.clean_datapackage_name( + f"{name} {matrix.replace('_', ' ')}" + ), + **resource, + ) + + datapackage.finalize_serialization() + filesystem = None + os.replace(temporary_path, filepath) + except Exception: + if filesystem is not None: + filesystem.close() + raise + finally: + temporary_path.unlink(missing_ok=True) + + return filepath diff --git a/pyproject.toml b/pyproject.toml index 567a8227..9130e3de 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ bw25 = [ "bw2calc >=2.0.1", "bw2data >=4.3", "bw2io >=0.9.4", + "bw_processing>=1.0", "bottleneck", "ecoinvent_interface", "constructive-geometries>=0.9.5", diff --git a/tests/test_export.py b/tests/test_export.py index 64a17ebc..f7d5d1f6 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -1,11 +1,16 @@ from types import SimpleNamespace +import numpy as np import pandas as pd import pytest from premise.clean_datasets import remove_uncertainty from premise.export import * -from premise.export import _aggregate_duplicate_superstructure_rows +from premise.export import ( + _aggregate_duplicate_superstructure_rows, + _build_superstructure_db, + _include_production_rows_for_changing_self_consumption, +) def test_simapro_units(): @@ -214,7 +219,10 @@ def test_aggregate_duplicate_superstructure_rows_sums_biosphere_collisions(): assert aggregated.loc[0, "scenario a"] == pytest.approx(0.6) -def test_aggregate_duplicate_superstructure_rows_nets_production_and_technosphere(): +@pytest.mark.parametrize("consumption_type", ["technosphere", "generic consumption"]) +def test_aggregate_duplicate_superstructure_rows_nets_production_and_technosphere( + consumption_type, +): df = pd.DataFrame( [ { @@ -231,7 +239,7 @@ def test_aggregate_duplicate_superstructure_rows_nets_production_and_technospher "to key": ("super-db", "act-1"), "from activity name": "self supplier", "to activity name": "self supplier", - "flow type": "technosphere", + "flow type": consumption_type, "original": 0.01, "scenario a": 0.4, }, @@ -253,6 +261,73 @@ def test_aggregate_duplicate_superstructure_rows_nets_production_and_technospher assert aggregated.loc[0, "scenario a"] == pytest.approx(0.6) +def test_changing_self_consumption_includes_unchanged_production_coordinate(): + acts_ind = { + 0: ("market", "product", None, "GLO", "kilogram", "production"), + 1: ("market", "product", None, "GLO", "kilogram", "technosphere"), + 2: ("supplier", "product", None, "GLO", "kilogram", "technosphere"), + } + + result = _include_production_rows_for_changing_self_consumption( + indices=np.array([[0, 1], [0, 2]]), + acts_ind=acts_ind, + ) + + assert result == [(0, 1), (0, 2), (0, 0)] + + +def test_superstructure_builder_preserves_legacy_drop_before_aggregation(monkeypatch): + dataframe = pd.DataFrame( + [ + { + "from key": ("biosphere3", "bio-1"), + "to key": ("super-db", "act-1"), + "flow type": "biosphere", + "from unit": "kilogram", + "to unit": "kilogram", + "original": 1.0, + "scenario a": 5.0, + }, + { + "from key": ("biosphere3", "bio-1"), + "to key": ("super-db", "act-1"), + "flow type": "biosphere", + "from unit": "kilogram", + "to unit": "kilogram", + "original": 2.0, + "scenario a": 5.0, + }, + ] + ) + monkeypatch.setattr( + "premise.export.generate_scenario_difference_file", + lambda **kwargs: (dataframe.copy(), [{"name": "dummy"}], []), + ) + + _, legacy_dataframe = _build_superstructure_db( + origin_db=[], + scenarios=[], + db_name="super-db", + biosphere_name="biosphere3", + version="3.12", + scenario_list=["scenario a"], + preserve_original_column=False, + ) + _, array_dataframe = _build_superstructure_db( + origin_db=[], + scenarios=[], + db_name="super-db", + biosphere_name="biosphere3", + version="3.12", + scenario_list=["scenario a"], + ) + + assert "original" not in legacy_dataframe + assert legacy_dataframe.loc[0, "scenario a"] == pytest.approx(5) + assert array_dataframe.loc[0, "original"] == pytest.approx(3) + assert array_dataframe.loc[0, "scenario a"] == pytest.approx(10) + + def test_generate_superstructure_db_aggregates_duplicate_key_pairs( monkeypatch, tmp_path ): @@ -342,6 +417,23 @@ def test_generate_superstructure_db_aggregates_duplicate_key_pairs( lambda **kwargs: (df.copy(), [{"name": "dummy"}], []), ) + union_database, scenario_dataframe = _build_superstructure_db( + origin_db=[], + scenarios=[], + db_name="super-db", + biosphere_name="biosphere3", + version="3.12", + scenario_list=["scenario a"], + ) + + assert union_database == [{"name": "dummy"}] + assert ["original", "scenario a"] == [ + column + for column in scenario_dataframe.columns + if column in {"original", "scenario a"} + ] + assert len(scenario_dataframe) == 2 + generate_superstructure_db( origin_db=[], scenarios=[], diff --git a/tests/test_new_database.py b/tests/test_new_database.py index fcd3c70d..e0fd6684 100644 --- a/tests/test_new_database.py +++ b/tests/test_new_database.py @@ -579,6 +579,196 @@ def fake_write_brightway_database(data, name, fast=False, check_internal=True): assert captured["pickles_deleted"] == 1 +def _scenario_array_test_object(): + obj = object.__new__(NewDatabase) + obj.biosphere_name = "test-biosphere" + obj.version = "3.12" + obj.source = "source-db" + obj.generate_reports = False + obj.scenarios = [ + {"model": "image", "pathway": "SSP2-Base", "year": 2030}, + {"model": "image", "pathway": "SSP2-Base", "year": 2040}, + ] + return obj + + +def test_write_scenario_array_rejects_legacy_brightway_before_writing(monkeypatch): + obj = _scenario_array_test_object() + monkeypatch.setattr(new_database_module.bw2data, "__version__", (3, 6, 6)) + monkeypatch.setattr( + new_database_module, + "write_brightway_database", + lambda **kwargs: (_ for _ in ()).throw( + AssertionError("legacy validation must happen before writing") + ), + ) + monkeypatch.setattr( + new_database_module, + "_load_scenario_array_dependencies", + lambda: (_ for _ in ()).throw( + AssertionError("legacy validation must happen before dependency loading") + ), + ) + + with pytest.raises(NotImplementedError, match="bw2data >= 4"): + obj.write_scenario_array_db_to_brightway() + + +def test_write_scenario_array_requires_two_generated_scenarios(monkeypatch): + obj = _scenario_array_test_object() + obj.scenarios = obj.scenarios[:1] + monkeypatch.setattr(new_database_module.bw2data, "__version__", "4.5.3") + + with pytest.raises(ValueError, match="At least two scenarios"): + obj.write_scenario_array_db_to_brightway() + + +def test_write_scenario_array_requires_registered_biosphere(monkeypatch): + obj = _scenario_array_test_object() + monkeypatch.setattr(new_database_module.bw2data, "__version__", "4.5.3") + monkeypatch.setattr(new_database_module.bw2data, "databases", {}) + + with pytest.raises( + ValueError, match="Brightway export requires a biosphere database" + ): + obj.write_scenario_array_db_to_brightway() + + +def test_write_scenario_array_requires_unique_labels(monkeypatch): + obj = _scenario_array_test_object() + obj.scenarios[1]["year"] = 2030 + monkeypatch.setattr(new_database_module.bw2data, "__version__", "4.5.3") + monkeypatch.setattr( + new_database_module, + "check_presence_biosphere_database", + lambda _: None, + ) + + with pytest.raises(ValueError, match="Duplicate label"): + obj.write_scenario_array_db_to_brightway() + + +def test_write_scenario_array_requires_zip_destination(monkeypatch): + obj = _scenario_array_test_object() + monkeypatch.setattr(new_database_module.bw2data, "__version__", "4.5.3") + monkeypatch.setattr( + new_database_module, + "check_presence_biosphere_database", + lambda _: None, + ) + + with pytest.raises(ValueError, match="'.zip' suffix"): + obj.write_scenario_array_db_to_brightway(filepath="arrays.csv") + + +def test_write_scenario_array_writes_database_then_package_and_finalizes_once( + monkeypatch, tmp_path +): + obj = _scenario_array_test_object() + obj.generate_reports = True + prepared_database = [{"name": "prepared database"}] + dataframe = object() + events = [] + + class DummyBwProcessing: + @staticmethod + def clean_datapackage_name(name): + return name.replace(" ", "_") + + def fake_prepare(**kwargs): + events.append(("prepare", kwargs)) + obj.database = prepared_database + return ["scenario-a", "scenario-b"], dataframe + + def fake_write_database(**kwargs): + events.append(("database", kwargs)) + + def fake_write_package(**kwargs): + events.append(("package", kwargs)) + assert events[-2][0] == "database" + return kwargs["filepath"].resolve() + + monkeypatch.setattr(new_database_module.bw2data, "__version__", "4.5.3") + monkeypatch.setattr( + new_database_module.bw2data, + "projects", + types.SimpleNamespace(current="scenario-project"), + ) + monkeypatch.setattr( + new_database_module, + "check_presence_biosphere_database", + lambda _: None, + ) + monkeypatch.setattr( + new_database_module, + "_load_scenario_array_dependencies", + lambda: (DummyBwProcessing, object(), object(), object()), + ) + monkeypatch.setattr( + new_database_module, + "write_brightway_database", + fake_write_database, + ) + monkeypatch.setattr( + new_database_module, + "_write_scenario_array_datapackage", + fake_write_package, + ) + monkeypatch.setattr( + new_database_module, + "end_of_process", + lambda scenario: events.append(("end", scenario)), + ) + monkeypatch.setattr( + new_database_module, + "delete_all_pickles", + lambda: events.append(("delete", None)), + ) + obj._prepare_superstructure_export = fake_prepare + obj.generate_scenario_report = lambda: events.append(("scenario report", None)) + obj.generate_change_report = lambda: events.append(("change report", None)) + + destination = tmp_path / "scenario arrays.zip" + result = obj.write_scenario_array_db_to_brightway( + name="scenario-db", filepath=destination + ) + + assert result == destination.resolve() + assert [event[0] for event in events] == [ + "prepare", + "database", + "package", + "scenario report", + "change report", + "end", + "end", + "delete", + ] + database_call = events[1][1] + assert database_call == { + "data": prepared_database, + "name": "scenario-db", + "fast": True, + "check_internal": False, + } + package_call = events[2][1] + assert package_call["dataframe"] is dataframe + assert package_call["scenario_labels"] == [ + "original", + "scenario-a", + "scenario-b", + ] + assert package_call["metadata"] == { + "database_name": "scenario-db", + "brightway_project": "scenario-project", + "source_database": "source-db", + "ecoinvent_version": "3.12", + "premise_version": "2.4.9.1", + "scenario_count": 3, + "scenario_labels": ["original", "scenario-a", "scenario-b"], + } + + def test_pathways_datapackage_does_not_prevalidate_biosphere_database(monkeypatch): captured = {} diff --git a/tests/test_scenario_array.py b/tests/test_scenario_array.py new file mode 100644 index 00000000..233f6f73 --- /dev/null +++ b/tests/test_scenario_array.py @@ -0,0 +1,329 @@ +from types import SimpleNamespace +from zipfile import ZIP_DEFLATED, ZipFile + +import numpy as np +import pandas as pd +import pytest + +from premise.scenario_array import ( + _load_scenario_array_dependencies, + _scenario_dataframe_to_arrays, + _write_scenario_array_datapackage, +) + +INDICES_DTYPE = np.dtype([("row", np.int64), ("col", np.int64)]) +SCENARIO_LABELS = ["original", "scenario one", "scenario two"] + + +def scenario_dataframe(): + return pd.DataFrame( + [ + { + "from activity name": "CO2", + "from key": ("biosphere", "co2"), + "to activity name": "supplier", + "to key": ("scenario-db", "supplier"), + "flow type": "biosphere", + "original": 1, + "scenario one": 10, + "scenario two": 100, + }, + { + "from activity name": "supplier", + "from key": ("scenario-db", "supplier"), + "to activity name": "consumer", + "to key": ("scenario-db", "consumer"), + "flow type": "technosphere", + "original": 1, + "scenario one": 2, + "scenario two": 0, + }, + { + "from activity name": "consumer", + "from key": ("scenario-db", "consumer"), + "to activity name": "consumer", + "to key": ("scenario-db", "consumer"), + "flow type": "production", + "original": 1, + "scenario one": 0.6, + "scenario two": 1, + }, + { + "from activity name": "substitute", + "from key": ("scenario-db", "substitute"), + "to activity name": "consumer", + "to key": ("scenario-db", "consumer"), + "flow type": "substitution", + "original": 0, + "scenario one": 0.2, + "scenario two": 0.3, + }, + { + "from activity name": "constant", + "from key": ("scenario-db", "constant"), + "to activity name": "consumer", + "to key": ("scenario-db", "consumer"), + "flow type": "technosphere", + "original": 5, + "scenario one": 5, + "scenario two": 5, + }, + ] + ) + + +def convert(dataframe): + ids = { + ("biosphere", "co2"): 1, + ("scenario-db", "supplier"): 2, + ("scenario-db", "consumer"): 3, + ("scenario-db", "substitute"): 4, + ("scenario-db", "constant"): 5, + } + return _scenario_dataframe_to_arrays( + dataframe=dataframe, + scenario_labels=SCENARIO_LABELS, + get_id=ids.__getitem__, + indices_dtype=INDICES_DTYPE, + biosphere_edge_types={"biosphere"}, + technosphere_negative_edge_types={"technosphere", "generic consumption"}, + technosphere_positive_edge_types={ + "production", + "generic production", + "substitution", + }, + database_name="scenario-db", + project_name="scenario-project", + ) + + +def test_scenario_dataframe_to_arrays_preserves_columns_dtypes_flips_and_zeros(): + resources = convert(scenario_dataframe()) + + biosphere = resources["biosphere_matrix"] + assert biosphere["data_array"].dtype == np.float64 + assert biosphere["data_array"].tolist() == [[1.0, 10.0, 100.0]] + assert biosphere["indices_array"].dtype == INDICES_DTYPE + assert biosphere["indices_array"].tolist() == [(1, 2)] + assert "flip_array" not in biosphere + + technosphere = resources["technosphere_matrix"] + assert technosphere["data_array"].dtype == np.float64 + assert technosphere["data_array"].tolist() == [ + [1.0, 2.0, 0.0], + [1.0, 0.6, 1.0], + [0.0, 0.2, 0.3], + ] + assert technosphere["indices_array"].tolist() == [(2, 3), (3, 3), (4, 3)] + assert technosphere["flip_array"].tolist() == [True, False, False] + + +def test_scenario_dataframe_to_arrays_omits_empty_matrix_resource(): + dataframe = scenario_dataframe() + dataframe = dataframe[dataframe["flow type"] != "biosphere"] + + resources = convert(dataframe) + + assert set(resources) == {"technosphere_matrix"} + + +def test_scenario_dataframe_to_arrays_rejects_no_changes(): + dataframe = scenario_dataframe().iloc[[-1]] + + with pytest.raises(ValueError, match="no exchanges change"): + convert(dataframe) + + +def test_scenario_dataframe_to_arrays_rejects_unsupported_flow_type(): + dataframe = scenario_dataframe().iloc[[0]].copy() + dataframe["flow type"] = "unsupported" + + with pytest.raises(ValueError, match="Unsupported Brightway flow type"): + convert(dataframe) + + +def test_scenario_dataframe_to_arrays_wraps_unresolved_keys_with_context(): + dataframe = scenario_dataframe().iloc[[0]].copy() + dataframe.at[dataframe.index[0], "from key"] = ("biosphere", "missing") + + with pytest.raises(KeyError) as error: + convert(dataframe) + + message = str(error.value) + assert "('biosphere', 'missing')" in message + assert "'CO2'" in message + assert "scenario-db" in message + assert "scenario-project" in message + + +def test_write_scenario_array_datapackage_is_compressed_and_replaces_atomically( + tmp_path, +): + bw_processing = pytest.importorskip("bw_processing") + from fsspec.implementations.zip import ZipFileSystem + + ids = { + ("biosphere", "co2"): 1, + ("scenario-db", "supplier"): 2, + ("scenario-db", "consumer"): 3, + ("scenario-db", "substitute"): 4, + ("scenario-db", "constant"): 5, + } + labels = SimpleNamespace( + biosphere_edge_types={"biosphere"}, + technosphere_negative_edge_types={"technosphere", "generic consumption"}, + technosphere_positive_edge_types={ + "production", + "generic production", + "substitution", + }, + ) + destination = tmp_path / "nested" / "arrays.zip" + destination.parent.mkdir() + destination.write_bytes(b"old artifact") + + result = _write_scenario_array_datapackage( + dataframe=scenario_dataframe(), + scenario_labels=SCENARIO_LABELS, + filepath=destination, + name="scenario-db", + metadata={"brightway_project": "scenario-project", "scenario_count": 3}, + dependencies=(bw_processing, ids.__getitem__, labels, ZipFileSystem), + ) + + assert result == destination.resolve() + with ZipFile(result) as archive: + assert archive.testzip() is None + assert all(item.compress_type == ZIP_DEFLATED for item in archive.infolist()) + assert not any( + path.name.startswith(".arrays-") for path in result.parent.iterdir() + ) + + +def test_scenario_array_datapackage_advances_matrices_together_and_wraps(tmp_path): + bd = pytest.importorskip("bw2data") + bc = pytest.importorskip("bw2calc") + pytest.importorskip("bw_processing") + if int(bd.__version__[0]) < 4: + pytest.skip("Scenario-array integration requires modern Brightway") + + from bw2data.tests import bw2test + + @bw2test + def run_in_temporary_brightway_project(): + biosphere = bd.Database("biosphere") + biosphere.write( + { + ("biosphere", "co2"): { + "name": "CO2", + "unit": "kilogram", + "categories": ("air",), + "type": "emission", + } + } + ) + database = bd.Database("scenario-db") + database.write( + { + ("scenario-db", "supplier"): { + "name": "supplier", + "reference product": "product", + "unit": "kilogram", + "location": "GLO", + "exchanges": [ + { + "input": ("scenario-db", "supplier"), + "amount": 1, + "type": "production", + }, + { + "input": ("biosphere", "co2"), + "amount": 1, + "type": "biosphere", + }, + ], + }, + ("scenario-db", "consumer"): { + "name": "consumer", + "reference product": "service", + "unit": "unit", + "location": "GLO", + "exchanges": [ + { + "input": ("scenario-db", "consumer"), + "amount": 1, + "type": "production", + }, + { + "input": ("scenario-db", "supplier"), + "amount": 1, + "type": "technosphere", + }, + ], + }, + } + ) + method = ("scenario-array-test", "GWP") + bd.Method(method).write([(("biosphere", "co2"), 1)]) + dataframe = pd.DataFrame( + [ + { + "from activity name": "supplier", + "from key": ("scenario-db", "supplier"), + "to activity name": "consumer", + "to key": ("scenario-db", "consumer"), + "flow type": "technosphere", + "original": 1, + "scenario one": 2, + "scenario two": 3, + }, + { + "from activity name": "CO2", + "from key": ("biosphere", "co2"), + "to activity name": "supplier", + "to key": ("scenario-db", "supplier"), + "flow type": "biosphere", + "original": 1, + "scenario one": 10, + "scenario two": 100, + }, + ] + ) + array_path = _write_scenario_array_datapackage( + dataframe=dataframe, + scenario_labels=SCENARIO_LABELS, + filepath=tmp_path / "integration-arrays.zip", + name="scenario-db", + metadata={"brightway_project": bd.projects.current}, + dependencies=_load_scenario_array_dependencies(), + ) + + demand, data_objs, remapping = bd.prepare_lca_inputs( + {database.get("consumer"): 1}, method=method + ) + lca = bc.LCA( + demand, + data_objs=[*data_objs, array_path], + remapping_dicts=remapping, + use_arrays=True, + use_distributions=False, + ) + lca.lci() + lca.lcia() + scores = [lca.score] + for _ in range(3): + next(lca) + scores.append(lca.score) + + base_lca = bc.LCA( + demand, + data_objs=data_objs, + remapping_dicts=remapping, + ) + base_lca.lci() + base_lca.lcia() + + assert scores == pytest.approx([1, 20, 300, 1]) + assert base_lca.score == pytest.approx(1) + + run_in_temporary_brightway_project()