deprecate, rename and check config entries

Snakefile

@@ -10,7 +10,14 @@ from snakemake.utils import min_version
 
 min_version("8.11")
 
-from scripts._helpers import path_provider, copy_default_files, get_scenarios, get_rdir
+from scripts._helpers import (
+    path_provider,
+    copy_default_files,
+    get_scenarios,
+    get_rdir,
+    check_deprecated_config,
+    check_invalid_config,
+)
 
 
 copy_default_files(workflow)
@@ -20,6 +27,9 @@ configfile: "config/config.default.yaml"
 configfile: "config/config.yaml"
 
 
+check_deprecated_config(config, "config/deprecations.yaml")
+check_invalid_config(config, "config/config.default.yaml")
+
 run = config["run"]
 scenarios = get_scenarios(run)
 RDIR = get_rdir(run)

config/deprecations.yaml

@@ -0,0 +1,8 @@
+# SPDX-FileCopyrightText: Contributors to PyPSA-Eur <https://github.com/pypsa/pypsa-eur>
+#
+# SPDX-License-Identifier: MIT
+
+# Format: list of deprecation entries with:
+# - old_entry: Dot-separated path to deprecated entry (e.g. "electricity:co2_limit")
+# - new_entry: [optional] New location path for renamed entries
+# - message: [optional] Custom warning message

doc/release_notes.rst

@@ -11,7 +11,10 @@ Release Notes
 Upcoming Release
 ================
 
-- ...
+- Add scheme to deprecate config entries. Deprecated entries with eventual renamings are now listed in `config/deprecations.yaml` and will be removed in future releases. Snakemake will raise a warning if deprecated entries are used. (https://github.com/PyPSA/pypsa-eur/pull/1514)
+
+- Snakemake now raises a warning if the user specifies config entries that are not present in the default config. (https://github.com/PyPSA/pypsa-eur/pull/1514)
+
 
 
 PyPSA-Eur v2025.01.0 (24th January 2025)

scripts/_helpers.py

@@ -6,10 +6,12 @@
 import copy
 import hashlib
 import logging
+import operator
 import os
 import re
 import time
-from functools import partial, wraps
+import warnings
+from functools import partial, reduce, wraps
 from os.path import exists
 from pathlib import Path
 from shutil import copyfile
@@ -29,6 +31,18 @@
 REGION_COLS = ["geometry", "name", "x", "y", "country"]
 
 
+class DeprecationConfigWarning(Warning):
+    """Warning for use of deprecated configuration entries."""
+
+    pass
+
+
+class InvalidConfigWarning(Warning):
+    """Warning for use of invalid/unsupported configuration entries."""
+
+    pass
+
+
 def copy_default_files(workflow):
     default_files = {
         "config/config.default.yaml": "config/config.yaml",
@@ -162,6 +176,100 @@ def path_provider(dir, rdir, shared_resources, exclude_from_shared):
     )
 
 
+def _check_scenarios(config: dict, check_fn: Callable, fn_kwargs: dict) -> None:
+    """Helper function to check configuration in scenario files"""
+    scenarios = config.get("run", {}).get("scenarios", {})
+    if scenarios.get("enable"):
+        with open(scenarios["file"]) as f:
+            scenario_config = yaml.safe_load(f)
+
+        for run in scenario_config:
+            # Disable recursive scenario checking to avoid infinite loops
+            fn_kwargs["check_scenarios"] = False
+            check_fn(scenario_config[run], **fn_kwargs)
+
+
+def check_deprecated_config(
+    config: dict, deprecations_file: str, check_scenarios: bool = True
+) -> None:
+    """Check config against deprecations and warn users"""
+
+    with open(deprecations_file) as f:
+        deprecations = yaml.safe_load(f)
+
+    def get_by_path(root, path):
+        try:
+            return reduce(operator.getitem, path.split(":"), root)
+        except KeyError:
+            return None
+
+    def set_by_path(root, path, value):
+        keys = path.split(":")
+        for key in keys[:-1]:
+            root = root.setdefault(key, {})
+        root[keys[-1]] = value
+
+    if deprecations is None:
+        return
+
+    for entry in deprecations:
+        old_entry = entry["old_entry"]
+        current_value = get_by_path(config, old_entry)
+
+        if current_value is not None:
+            msg = f"Config entry '{old_entry}' is deprecated. "
+
+            if "new_entry" in entry:  # Rename case
+                new_entry = entry["new_entry"]
+                msg += f"Use '{new_entry}' instead."
+
+                if get_by_path(config, new_entry) is not None:
+                    msg += " Both keys present - remove deprecated entry."
+                else:
+                    set_by_path(config, new_entry, current_value)
+            else:  # Removal case
+                msg += "This entry is no longer used and should be removed."
+
+            if "message" in entry:
+                msg += f" Note: {entry['message']}"
+
+            warnings.warn(msg, DeprecationConfigWarning)
+
+    if check_scenarios:
+        _check_scenarios(
+            config, check_deprecated_config, {"deprecations_file": deprecations_file}
+        )
+
+
+def check_invalid_config(
+    config: dict, config_default_fn: str, check_scenarios: bool = True
+) -> None:
+    """Check if config contains entries that are not supported by the default config"""
+
+    with open(config_default_fn) as f:
+        config_default = yaml.safe_load(f)
+
+    def check_keys(config, config_default, path=""):
+        for key in config.keys():
+            nested_path = f"{path}:{key}" if path else key
+            if key not in config_default:
+                warnings.warn(
+                    f"Config entry '{nested_path}' is not supported in {config_default_fn}.",
+                    InvalidConfigWarning,
+                )
+            elif isinstance(config[key], dict):
+                # Only recurse if the key exists in both configs and is a dict in both
+                if isinstance(config_default[key], dict):
+                    check_keys(config[key], config_default[key], nested_path)
+
+    check_keys(config, config_default)
+
+    if check_scenarios:
+        _check_scenarios(
+            config, check_invalid_config, {"config_default_fn": config_default_fn}
+        )
+
+
 def get_opt(opts, expr, flags=None):
     """
     Return the first option matching the regular expression.

test/test_config_checks.py

@@ -0,0 +1,185 @@
+# SPDX-FileCopyrightText: Contributors to PyPSA-Eur <https://github.com/pypsa/pypsa-eur>
+#
+# SPDX-License-Identifier: MIT
+import warnings
+from io import StringIO
+from unittest.mock import patch
+
+from scripts._helpers import (
+    DeprecationConfigWarning,
+    InvalidConfigWarning,
+    check_deprecated_config,
+    check_invalid_config,
+)
+
+SAMPLE_DEPRECATIONS = """
+- old_entry: "old:key"
+  new_entry: "new:key"
+
+- old_entry: "removed:key"
+  message: "This key is obsolete and should be deleted"
+
+- old_entry: "example:old_key"
+  new_entry: "example:new_key"
+  message: "Custom warning message"
+
+- old_entry: "clustering:deprecated_option"
+  message: "Custom warning message for deprecated clustering option"
+"""
+
+SAMPLE_SCENARIOS = """
+scenario_1:
+    example:
+        old_key: "test_value"
+    invalid_section:
+        some_key: "value"
+
+scenario_2:
+    clustering:
+        invalid_option: "bad"
+        deprecated_option: "old_value"
+"""
+
+
+def test_config_deprecations():
+    test_config = {
+        "old": {"key": "legacy_value"},
+        "removed": {"key": "dangerous_value"},
+        "example": {"old_key": "original_value", "new_key": "existing_value"},
+        "unrelated": {"data": "untouched"},
+    }
+
+    with warnings.catch_warnings(record=True) as captured_warnings:
+        with patch("builtins.open", return_value=StringIO(SAMPLE_DEPRECATIONS)):
+            check_deprecated_config(test_config, "dummy_path.yaml")
+
+    # Verify warnings
+    assert len(captured_warnings) == 3
+
+    warning_messages = [str(w.message) for w in captured_warnings]
+
+    # Check basic rename warning
+    assert any(
+        "'old:key' is deprecated. Use 'new:key' instead" in msg
+        for msg in warning_messages
+    )
+
+    # Check removal warning with custom message
+    assert any("obsolete and should be deleted" in msg for msg in warning_messages)
+
+    # Check custom message and conflict warning
+    assert any("Custom warning message" in msg for msg in warning_messages)
+    assert any(
+        "Both keys present - remove deprecated entry" in msg for msg in warning_messages
+    )
+
+    # Verify warning types
+    assert all(
+        isinstance(w.message, DeprecationConfigWarning) for w in captured_warnings
+    )
+
+    # Verify config updates
+    assert test_config["new"]["key"] == "legacy_value"  # Renamed value
+    assert (
+        test_config["example"]["new_key"] == "existing_value"
+    )  # Existing value preserved
+    assert "key" in test_config["removed"]  # Removed key not deleted (just warned)
+    assert test_config["unrelated"] == {"data": "untouched"}  # Unrelated data unchanged
+
+
+def test_config_invalid_entries():
+    test_config = {
+        "valid_section": {"nested_valid": "ok"},
+        "invalid_section": {"bad_key": "value"},
+        "clustering": {"invalid_option": "bad"},
+    }
+
+    default_config = """
+valid_section:
+    nested_valid: default
+    other_valid: default
+clustering:
+    temporal:
+        resolution: 1
+    """
+
+    with warnings.catch_warnings(record=True) as captured_warnings:
+        with patch("builtins.open", return_value=StringIO(default_config)):
+            check_invalid_config(test_config, "dummy_default.yaml")
+
+    warning_messages = [str(w.message) for w in captured_warnings]
+
+    # Check warning for invalid top-level section
+    assert any(
+        "Config entry 'invalid_section' is not supported" in msg
+        for msg in warning_messages
+    )
+
+    # Check warning for invalid nested option
+    assert any(
+        "Config entry 'clustering:invalid_option' is not supported" in msg
+        for msg in warning_messages
+    )
+
+    # Verify warning types
+    assert all(isinstance(w.message, InvalidConfigWarning) for w in captured_warnings)
+
+
+def test_config_scenario_checks():
+    """Test that configuration checks are performed on scenario files"""
+    config = {
+        "run": {
+            "scenarios": {
+                "enable": True,
+                "file": "scenarios.yaml",
+            }
+        }
+    }
+
+    # Setup mock files
+    mock_files = {
+        "deprecations.yaml": SAMPLE_DEPRECATIONS,
+        "config.default.yaml": """
+            run:
+              scenarios:
+                  enable: true
+                  file: scenarios.yaml
+            example:
+                new_key: "default"
+            clustering:
+                temporal:
+                    resolution: 1
+            """,
+        "scenarios.yaml": SAMPLE_SCENARIOS,
+    }
+
+    def mock_open(filename, *args, **kwargs):
+        return StringIO(mock_files[filename.split("/")[-1]])
+
+    with warnings.catch_warnings(record=True) as captured_warnings:
+        with patch("builtins.open", mock_open):
+            # Check both deprecated and invalid entries in scenarios
+            check_deprecated_config(config, "deprecations.yaml")
+            check_invalid_config(config, "config.default.yaml")
+
+    warning_messages = [str(w.message) for w in captured_warnings]
+
+    # Verify warnings from scenario_1
+    assert any("'example:old_key' is deprecated" in msg for msg in warning_messages)
+    assert any("'invalid_section' is not supported" in msg for msg in warning_messages)
+
+    # Verify warnings from scenario_2
+    assert any(
+        "'clustering:invalid_option' is not supported" in msg
+        for msg in warning_messages
+    )
+
+    # Verify warning types
+    deprecation_warnings = [
+        w for w in captured_warnings if isinstance(w.message, DeprecationConfigWarning)
+    ]
+    invalid_warnings = [
+        w for w in captured_warnings if isinstance(w.message, InvalidConfigWarning)
+    ]
+    assert len(deprecation_warnings) == 2
+    assert len(invalid_warnings) == 4