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"]]