diff --git a/CHANGELOG.md b/CHANGELOG.md index 5402b40e..d2f5d584 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 Note, that the semantics of metadata imports have changed, see the Changed section below. - Added missing `export_metadata()`, `export_entity_metadata()` and `export_parameter_value_metadata()` functions to `export_functions` module. +- A new experimental module `spinedb_api.scenario_recipes` contains functions to support complex scenario generation + from sets of alternatives. ### Changed diff --git a/spinedb_api/filters/alternative_filter.py b/spinedb_api/filters/alternative_filter.py index 02c2e4e9..63aaf20a 100644 --- a/spinedb_api/filters/alternative_filter.py +++ b/spinedb_api/filters/alternative_filter.py @@ -21,8 +21,6 @@ from ..exception import SpineDBAPIError from .query_utils import filter_by_active_elements -__all__ = ("alternative_filter_config",) - ALTERNATIVE_FILTER_TYPE = "alternative_filter" ALTERNATIVE_FILTER_SHORTHAND_TAG = "alternatives" @@ -31,6 +29,8 @@ def apply_alternative_filter_to_parameter_value_sq(db_map, alternatives): """ Replaces parameter value subquery properties in ``db_map`` such that they return only values of given alternatives. + :meta private: + Args: db_map (DatabaseMapping): a database map to alter alternatives (Iterable of str or int, optional): alternative names or ids; @@ -69,6 +69,8 @@ def alternative_filter_from_dict(db_map, config): """ Applies alternative filter to given database map. + :meta private: + Args: db_map (DatabaseMapping): target database map config (dict): alternative filter configuration @@ -80,6 +82,8 @@ def alternative_filter_config_to_shorthand(config): """ Makes a shorthand string from alternative filter configuration. + :meta private: + Args: config (dict): alternative filter configuration @@ -96,6 +100,8 @@ def alternative_names_from_dict(config): """ Returns alternatives' names from filter config. + :meta private: + Args: config (dict): alternative filter configuration @@ -111,6 +117,8 @@ def alternative_filter_shorthand_to_config(shorthand): """ Makes configuration dictionary out of a shorthand string. + :meta private: + Args: shorthand (str): a shorthand string diff --git a/spinedb_api/filters/renamer.py b/spinedb_api/filters/renamer.py index ff064728..b97e3409 100644 --- a/spinedb_api/filters/renamer.py +++ b/spinedb_api/filters/renamer.py @@ -17,8 +17,6 @@ from functools import partial from sqlalchemy import case -__all__ = ("entity_class_renamer_config", "parameter_renamer_config") - ENTITY_CLASS_RENAMER_TYPE = "entity_class_renamer" ENTITY_CLASS_RENAMER_SHORTHAND_TAG = "entity_class_rename" PARAMETER_RENAMER_TYPE = "parameter_renamer" @@ -29,6 +27,8 @@ def apply_renaming_to_entity_class_sq(db_map, name_map): """ Applies renaming to entity class subquery. + :meta private: + Args: db_map (DatabaseMapping): a database map name_map (dict): a map from old name to new name @@ -55,6 +55,8 @@ def entity_class_renamer_from_dict(db_map, config): """ Applies entity class renamer manipulator to given database map. + :meta private: + Args: db_map (DatabaseMapping): target database map config (dict): renamer configuration @@ -66,6 +68,8 @@ def entity_class_renamer_config_to_shorthand(config): """ Makes a shorthand string from renamer configuration. + :meta private: + Args: config (dict): renamer configuration @@ -82,6 +86,8 @@ def entity_class_renamer_shorthand_to_config(shorthand): """ Makes configuration dictionary out of a shorthand string. + :meta private: + Args: shorthand (str): a shorthand string @@ -99,6 +105,8 @@ def apply_renaming_to_parameter_definition_sq(db_map, name_map): """ Applies renaming to parameter definition subquery. + :meta private: + Args: db_map (DatabaseMapping): a database map name_map (dict): a map from old name to new name @@ -124,6 +132,8 @@ def parameter_renamer_from_dict(db_map, config): """ Applies parameter renamer manipulator to given database map. + :meta private: + Args: db_map (DatabaseMapping): target database map config (dict): renamer configuration @@ -135,6 +145,8 @@ def parameter_renamer_config_to_shorthand(config): """ Makes a shorthand string from renamer configuration. + :meta private: + Args: config (dict): renamer configuration @@ -152,6 +164,8 @@ def parameter_renamer_shorthand_to_config(shorthand): """ Makes configuration dictionary out of a shorthand string. + :meta private: + Args: shorthand (str): a shorthand string diff --git a/spinedb_api/filters/scenario_filter.py b/spinedb_api/filters/scenario_filter.py index b278e5ea..77b5160c 100644 --- a/spinedb_api/filters/scenario_filter.py +++ b/spinedb_api/filters/scenario_filter.py @@ -9,7 +9,7 @@ # Public License for more details. You should have received a copy of the GNU Lesser General Public License along with # this program. If not, see . ###################################################################################################################### -""" This module provides the scenario filter. """ +"""This module provides the scenario filter.""" from functools import partial from sqlalchemy import and_, case, desc, func, or_ @@ -17,8 +17,6 @@ from ..exception import SpineDBAPIError from .query_utils import filter_by_active_elements -__all__ = ("scenario_filter_config",) - SCENARIO_FILTER_TYPE = "scenario_filter" SCENARIO_SHORTHAND_TAG = "scenario" @@ -27,6 +25,8 @@ def apply_scenario_filter_to_subqueries(db_map, scenario): """ Replaces affected subqueries in ``db_map`` such that they return only values of given scenario. + :meta private: + Args: db_map (DatabaseMapping): a database map to alter scenario (str or int): scenario name or id @@ -65,6 +65,8 @@ def scenario_filter_from_dict(db_map, config): """ Applies scenario filter to given database map. + :meta private: + Args: db_map (DatabaseMapping): target database map config (dict): scenario filter configuration @@ -76,6 +78,8 @@ def scenario_name_from_dict(config): """ Returns scenario's name from filter config. + :meta private: + Args: config (dict): scenario filter configuration @@ -91,6 +95,8 @@ def scenario_filter_config_to_shorthand(config): """ Makes a shorthand string from scenario filter configuration. + :meta private: + Args: config (dict): scenario filter configuration @@ -104,6 +110,8 @@ def scenario_filter_shorthand_to_config(shorthand): """ Makes configuration dictionary out of a shorthand string. + :meta private: + Args: shorthand (str): a shorthand string diff --git a/spinedb_api/filters/value_transformer.py b/spinedb_api/filters/value_transformer.py index ec120e8b..b10a1c93 100644 --- a/spinedb_api/filters/value_transformer.py +++ b/spinedb_api/filters/value_transformer.py @@ -31,13 +31,13 @@ VALUE_TRANSFORMER_TYPE = "value_transformer" VALUE_TRANSFORMER_SHORTHAND_TAG = "value_transform" -__all__ = ("value_transformer_config",) - def apply_value_transform_to_parameter_value_sq(db_map, instructions): """ Applies renaming to parameter definition subquery. + :meta private: + Args: db_map (DatabaseMapping): a database map instructions (dict): mapping from entity class name to mapping from parameter name to list of @@ -73,6 +73,8 @@ def value_transformer_from_dict(db_map, config): """ Applies value transformer manipulator to given database map. + :meta private: + Args: db_map (DatabaseMapping): target database map config (dict): transformer configuration @@ -84,6 +86,8 @@ def value_transformer_config_to_shorthand(config): """ Makes a shorthand string from transformer configuration. + :meta private: + Args: config (dict): transformer configuration @@ -113,6 +117,8 @@ def value_transformer_shorthand_to_config(shorthand): Args: shorthand (str): a shorthand string + :meta private: + Returns: dict: value transformer configuration """ diff --git a/spinedb_api/scenario_recipes.py b/spinedb_api/scenario_recipes.py new file mode 100644 index 00000000..f1cea857 --- /dev/null +++ b/spinedb_api/scenario_recipes.py @@ -0,0 +1,159 @@ +###################################################################################################################### +# Copyright (C) 2017-2022 Spine project consortium +# Copyright Spine Database API contributors +# This file is part of Spine Database API. +# Spine Toolbox is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General +# Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) +# any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; +# without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General +# Public License for more details. You should have received a copy of the GNU Lesser General Public License along with +# this program. If not, see . +###################################################################################################################### + +""" +This module contains functions and methods to support scenario generation. + +.. warning:: + + This API is experimental. + +Let's build some scenarios! +Imagine we have five alternatives: ``Base``, ``high_fuel``, ``low_fuel``, ``high_co2`` and ``low_co2``. +Each scenario should use ``Base`` as the lowest rank alternative. +On top of that, we want to build useful combinations of the other alternatives. +First, we combine ``Base`` with all other alternatives. + +.. code-block:: python + + import spinedb_api as api + import spinedb_api.scenario_recipes as recipes + + url = "" + + with api.DatabaseMapping(url) as db_map: + base_alternative = db_map.alternative(name="Base") + for alternative_name in ("high_fuel", "low_fuel", "high_co2", "low_co2"): + scenario_name = alternative_name + alternative = db_map.alternative(name=alternative_name) + recipes.create_with_alternatives([base_alternative, alternative], scenario_name) + +Now, we have four scenarios, ``high_fuel``, ``low_fuel``, ``high_co2`` and ``low_co2``. +Each scenario has ``Base`` alternative, and one of the other alternatives corresponding to scenario's name. + +Let's create another set of scenarios by duplicating the existing ones +and adding ``high_co2`` and ``low_co2`` alternatives to the scenarios that deal with fuel prices. + +.. code-block:: python + + import spinedb_api as api + import spinedb_api.scenario_recipes as recipes + + url = "" + + with api.DatabaseMapping(url) as db_map: + for fuel in ("high_fuel", "low_fuel"): + for co2 in ("high_co2", "low_co2"): + scenario_name = f"{fuel}+{co2}" + base_scenario = db_map.scenario(name=fuel) + new_scenario = recipes.duplicate_scenario(base_scenario, scenario_name) + co2_alternative = db_map.alternative(name=co2) + recipes.with_alternative(new_scenario, co2_alternative) + +After running the script above, our database contains four new scenarios: +``high_fuel+high_co2``, ``high_fuel+low_co2``, ``low_fuel+high_co2`` and ``low_fuel+low_co2``. + +The combinatoric iterators of the `itertools module`_ in Python's standard library are also useful +when generating scenarios based on existing alternatives. +Let's assume we have another set of alternatives which we want to combine in as many ways as possible +to generate more scenarios. +Again, all scenarios should have ``Base`` alternative as lowest rank alternative. +The available alternatives are: ``coal``, ``coal_chp``, ``wind`` and ``antimatter``. + +.. _itertools module: https://docs.python.org/3.14/library/itertools.html + +.. code-block:: python + + import itertools + import spinedb_api as api + import spinedb_api.scenario_recipes as recipes + + url = "" + + with api.DatabaseMapping(url) as db_map: + base_alternative = db_map.alternative(name="Base") + sector_alternatives = [db_map.alternative(name=name) for name in ("coal", "coal_chp", "wind", "anti_matter")] + for n_sectors in range(1, len(sector_alternatives) + 1): + for sector_set in itertools.combinations(sector_alternatives, n_sectors): + scenario_name = "+".join(alternative["name"] for alternative in sector_set) + full_alternatives = (base_alternative,) + sector_set + recipes.create_with_alternatives(full_alternatives, scenario_name) + +``itertools.combinations()`` creates us all sensible combinations of alternatives which result in 15 scenarios: +``coal``, ``coal_chp``, ``wind``, ``antimatter``, +``coal+coal_chp``, ``coal+wind``, ``coal+antimatter``, ``coal_chp+wind``, ``coal_chp+antimatter``, ``wind+antimatter``, +``coal+coal_chp+wind``, ``coal+coal_chp+antimatter``, ``coal+wind+antimatter``, ``coal_chp+wind+antimatter`` +and ``coal+coal_chp+wind+antimatter`` +""" +from collections.abc import Iterable, Sequence +from spinedb_api import SpineDBAPIError +from spinedb_api.db_mapping_base import PublicItem + + +def create_with_alternatives(alternatives: Sequence[PublicItem], scenario_name: str) -> PublicItem: + """Creates a new scenario with given alternatives.""" + if not alternatives: + raise SpineDBAPIError("no alternatives given") + db_map = alternatives[0].db_map + new_scenario = db_map.add_scenario(name=scenario_name) + scenario_id = new_scenario["id"] + for rank, alternative in enumerate(alternatives): + db_map.add_scenario_alternative(scenario_id=scenario_id, alternative_id=alternative["id"], rank=rank) + return new_scenario + + +def duplicate_scenario(scenario: PublicItem, duplicate_name: str) -> PublicItem: + """Duplicates a scenario and its scenario alternatives.""" + db_map = scenario.db_map + new_scenario = db_map.add_scenario(name=duplicate_name, description=scenario["description"]) + new_scenario_id = new_scenario["id"] + for scenario_alternative in db_map.find_scenario_alternatives(scenario_id=scenario["id"]): + db_map.add_scenario_alternative( + scenario_id=new_scenario_id, + alternative_id=scenario_alternative["alternative_id"], + rank=scenario_alternative["rank"], + ) + return new_scenario + + +def with_alternative(scenario: PublicItem, alternative: PublicItem) -> None: + """Adds an alternative to an existing scenario. + + The alternative will be added as the highest ranking alternative. + """ + alternative_id = alternative["id"] + db_map = scenario.db_map + scenario_id = scenario["id"] + existing_scenario_alternatives = scenario.db_map.find_scenario_alternatives(scenario_id=scenario_id) + if existing_scenario_alternatives: + last_rank = max(scenario_alternative["rank"] for scenario_alternative in existing_scenario_alternatives) + else: + last_rank = -1 + db_map.add_scenario_alternative(scenario_id=scenario_id, alternative_id=alternative_id, rank=last_rank + 1) + + +def with_alternatives(scenario: PublicItem, alternatives: Iterable[PublicItem]) -> None: + """Adds given alternatives to an existing scenario. + + The alternatives will be added as the highest ranking alternatives. + """ + db_map = scenario.db_map + scenario_id = scenario["id"] + existing_scenario_alternatives = scenario.db_map.find_scenario_alternatives(scenario_id=scenario_id) + if existing_scenario_alternatives: + last_rank = max(scenario_alternative["rank"] for scenario_alternative in existing_scenario_alternatives) + else: + last_rank = -1 + for i, alternative in enumerate(alternatives): + db_map.add_scenario_alternative( + scenario_id=scenario_id, alternative_id=alternative["id"], rank=last_rank + i + 1 + ) diff --git a/tests/test_scenario_recipes.py b/tests/test_scenario_recipes.py new file mode 100644 index 00000000..970d3294 --- /dev/null +++ b/tests/test_scenario_recipes.py @@ -0,0 +1,96 @@ +###################################################################################################################### +# Copyright (C) 2017-2022 Spine project consortium +# Copyright Spine Database API contributors +# This file is part of Spine Database API. +# Spine Toolbox is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General +# Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) +# any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; +# without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General +# Public License for more details. You should have received a copy of the GNU Lesser General Public License along with +# this program. If not, see . +###################################################################################################################### +import pytest +from spinedb_api import DatabaseMapping, SpineDBAPIError +import spinedb_api.scenario_recipes as recipes + + +@pytest.fixture +def db_map(): + with DatabaseMapping("sqlite://", create=True) as db_map: + yield db_map + + +class TestDuplicateScenario: + def test_duplicate_scenario_without_alternatives(self, db_map): + original = db_map.add_scenario(name="Scenario", description="Original scenario") + duplicate = recipes.duplicate_scenario(original, "Duplicate scenario") + assert duplicate["name"] == "Duplicate scenario" + assert duplicate["description"] == "Original scenario" + assert db_map.find_scenario_alternatives(scenario_id=duplicate["id"]) == [] + + def test_duplicate_scenario_with_alternatives(self, db_map): + original = db_map.add_scenario(name="Scenario", description="Original scenario") + alternative_1 = db_map.add_alternative(name="Alternative 1") + alternative_2 = db_map.add_alternative(name="Alternative 2") + db_map.add_scenario_alternative(scenario_id=original["id"], alternative_id=alternative_1["id"], rank=0) + db_map.add_scenario_alternative(scenario_id=original["id"], alternative_id=alternative_2["id"], rank=1) + duplicate = recipes.duplicate_scenario(original, "Duplicate scenario") + assert duplicate["name"] == "Duplicate scenario" + assert duplicate["description"] == "Original scenario" + assert duplicate["alternative_name_list"] == ["Alternative 1", "Alternative 2"] + + +class TestCreateWithAlternatives: + def test_no_alternatives_raises(self): + with pytest.raises(SpineDBAPIError, match="^no alternatives given$"): + recipes.create_with_alternatives([], "Scenario 1") + + def test_alternatives_get_added_in_order(self, db_map): + alternative_1 = db_map.add_alternative(name="Alternative 1") + alternative_2 = db_map.add_alternative(name="Alternative 2") + alternative_3 = db_map.add_alternative(name="Alternative 3") + scenario = recipes.create_with_alternatives([alternative_2, alternative_3, alternative_1], "Scenario") + assert scenario["name"] == "Scenario" + assert scenario["description"] is None + assert scenario["alternative_id_list"] == [alternative_2["id"], alternative_3["id"], alternative_1["id"]] + + +class TestWithAlternative: + def test_add_alternative_to_empty_scenario(self, db_map): + scenario = db_map.add_scenario(name="Scenario") + alternative = db_map.add_alternative(name="Alternative") + recipes.with_alternative(scenario, alternative) + assert scenario["alternative_name_list"] == ["Alternative"] + + def test_added_alternative_has_the_highest_rank(self, db_map): + scenario = db_map.add_scenario(name="Scenario") + alternative_1 = db_map.add_alternative(name="Alternative 1") + db_map.add_scenario_alternative(scenario_id=scenario["id"], alternative_id=alternative_1["id"], rank=1) + alternative_2 = db_map.add_alternative(name="Alternative 2") + db_map.add_scenario_alternative(scenario_id=scenario["id"], alternative_id=alternative_2["id"], rank=23) + alternative_3 = db_map.add_alternative(name="Alternative 3") + recipes.with_alternative(scenario, alternative_3) + assert scenario["alternative_name_list"] == ["Alternative 1", "Alternative 2", "Alternative 3"] + + def test_adding_alternative_that_is_already_in_scenario(self, db_map): + scenario = db_map.add_scenario(name="Scenario") + alternative_1 = db_map.add_alternative(name="Alternative 1") + db_map.add_scenario_alternative(scenario_id=scenario["id"], alternative_id=alternative_1["id"], rank=1) + with pytest.raises(SpineDBAPIError, match="^there's already a scenario_alternative with"): + recipes.with_alternative(scenario, alternative_1) + + +class TestWithAlternatives: + def test_empty_alternatives(self, db_map): + scenario = db_map.add_scenario(name="Scenario") + recipes.with_alternatives(scenario, []) + assert scenario["alternative_name_list"] == [] + + def test_alternatives_added_to_highest_ranks(self, db_map): + scenario = db_map.add_scenario(name="Scenario") + alternative_1 = db_map.add_alternative(name="Alternative 1") + recipes.with_alternative(scenario, alternative_1) + alternative_2 = db_map.add_alternative(name="Alternative 2") + alternative_3 = db_map.add_alternative(name="Alternative 3") + recipes.with_alternatives(scenario, (alternative_2, alternative_3)) + assert scenario["alternative_id_list"] == [alternative_1["id"], alternative_2["id"], alternative_3["id"]]