cba.py

# SPDX-FileCopyrightText: Contributors to Open-TYNDP <https://github.com/open-energy-transition/open-tyndp>
#
# SPDX-License-Identifier: MIT

"""
Cost-benefit analysis configuration.

See docs in https://open-tyndp.readthedocs.io/en/latest/configuration.html#cba
"""

from typing import Any, Literal

from pydantic import BaseModel, Field, field_validator

from scripts.lib.validation.config._base import ConfigModel
from scripts.lib.validation.config.solving import _SolvingOptionsConfig


def _validate_solving_options_keys(options: dict[str, Any]) -> dict[str, Any]:
    """
    Reject misspelled CBA solving option names while accepting solving config options.

    This checks that all keys in the provided options dict are valid solving option names,
    while allowing any valid option name from the top-level `solving.options` section.

    This is used to validate both the top-level `cba.solving.options` and the
    `cba.msv_extraction.solving.options`, which have the same valid option names as
    the top-level `solving.options`.
    """
    valid_options = set(_SolvingOptionsConfig.model_fields)
    unknown_options = sorted(set(options) - valid_options)
    if unknown_options:
        valid = ", ".join(sorted(valid_options))
        unknown = ", ".join(unknown_options)
        raise ValueError(
            f"Unknown CBA solving option(s): {unknown}. Valid options are: {valid}"
        )
    return options


class _CbaStorageConfig(ConfigModel):
    """Configuration for `cba.storage` settings."""

    cyclic_carriers: list[str] = Field(
        default_factory=lambda: ["battery", "home battery"],
        description="Carriers that should remain cyclic (short-term storage). All other store and storage unit carriers automatically receive marginal storage value and have cyclicity disabled.",
    )
    soc_boundary_carriers: list[str] = Field(
        default_factory=lambda: ["hydro-reservoir"],
        description="Storage unit carriers for which the state of charge is pinned at the boundaries between rolling horizon windows, using values pre-computed from the perfect foresight (full-year) optimisation.",
    )


class _CbaMsvSolvingConfig(ConfigModel):
    """Configuration for `cba.msv_extraction.solving` settings."""

    options: dict[str, Any] = Field(
        default_factory=dict,
        description="Solving option overrides for MSV extraction. Uses the same option names as the top-level `solving.options` section.",
    )
    solver: dict[str, str] = Field(
        default_factory=lambda: {"name": "highs", "options": "highs-simplex"},
        description="Solver configuration for MSV extraction.",
    )
    solver_options: dict[str, dict[str, Any]] = Field(
        default_factory=dict,
        description="Solver-specific options for MSV extraction.",
    )

    @field_validator("options")
    @classmethod
    def check_options(cls, options: dict[str, Any]) -> dict[str, Any]:
        """Validate CBA MSV solving option names."""
        return _validate_solving_options_keys(options)


class _CbaMsvExtractionConfig(ConfigModel):
    """Configuration for `cba.msv_extraction` settings."""

    resolution: bool | str = Field(
        default=False,
        description="Temporal resolution for extraction solve. False uses native resolution, or a string like '24H', '48H' for faster solve.",
    )
    resample_method: Literal["ffill", "interpolate"] = Field(
        default="ffill",
        description="Method for resampling marginal storage value to target network resolution.",
    )
    solving: _CbaMsvSolvingConfig = Field(
        default_factory=_CbaMsvSolvingConfig,
        description="Solver configuration overrides for the MSV extraction solve.",
    )


class _CbaSolvingConfig(ConfigModel):
    """Configuration for `cba.solving` settings."""

    options: dict[str, Any] = Field(
        default_factory=lambda: {
            "load_shedding": {"enable": True},
            "io_api": "direct",
        },
        description="Solving option overrides for rolling horizon CBA dispatch. Uses the same option names as the top-level `solving.options`.",
    )
    horizon: int = Field(
        168,
        description="Number of snapshots to consider in each rolling-horizon window for CBA project solves.",
    )
    overlap: int = Field(
        1,
        description="Number of snapshots to overlap between rolling-horizon windows for CBA project solves.",
    )
    solver: dict[str, str] = Field(
        default_factory=lambda: {"name": "highs", "options": "highs-simplex"},
        description="Solver configuration.",
    )
    solver_options: dict[str, dict[str, Any]] = Field(
        default_factory=dict,
        description="Solver-specific options.",
    )

    @field_validator("options")
    @classmethod
    def check_options(cls, options: dict[str, Any]) -> dict[str, Any]:
        """Validate CBA rolling-horizon solving option names."""
        return _validate_solving_options_keys(options)


class _CbaSbToCbaConfig(ConfigModel):
    """Configuration for using pre-solved SB networks in the CBA workflow."""

    use_presolved: bool = Field(
        False,
        description="If true, use a pre-solved SB network from an external archive instead of running the SB workflow.",
    )
    sb_version: str = Field(
        "latest",
        description="Version of open_tyndp_prelim to use for pre-solved SB network input in CBA. Use 'latest' or a supported version from data/versions.csv.",
    )


class CbaConfig(BaseModel):
    """Configuration for top level `cba` (cost-benefit analysis) settings."""

    hurdle_costs: float = Field(
        0.01,
        description="Marginal cost for transmission lines in cost-benefit analysis networks (EUR/MWh).",
    )
    co2_societal_cost: dict[int, dict[str, float]] = Field(
        default_factory=dict,
        description="Dictionary mapping planning horizons to societal costs of CO2 emissions (EUR/t) for 'low', 'central', and 'high' scenarios.",
    )
    planning_horizons: list[int] = Field(
        default_factory=list,
        description="List of planning horizons for which to run cost-benefit analysis.",
    )
    cba_scenario_input: _CbaSbToCbaConfig = Field(
        default_factory=_CbaSbToCbaConfig,
        description="Settings for using pre-solved SB networks as inputs to the CBA workflow.",
    )
    projects: list[str] = Field(
        default_factory=list,
        description="List of project identifiers to evaluate (e.g., 't1-t35').",
    )
    area: Literal["tyndp", "entso-e", "eu27"] = Field(
        default="tyndp",
        description="Geographical area for cost-benefit analysis. Options include 'tyndp', 'entso-e', and 'eu27'.",
    )
    remove_noisy_costs: bool = Field(
        default=True,
        description="If true, use original pre-noise capital and marginal costs for CBA indicators.",
    )
    negative_toot_capacity: Literal["zero", "break"] = Field(
        default="zero",
        description="How to handle TOOT project removal when removing project capacity would make an existing interconnector capacity negative. 'zero' clamps the resulting capacity to zero and continues; 'break' raises an error.",
    )
    storage: _CbaStorageConfig = Field(
        default_factory=_CbaStorageConfig,
        description="Storage configuration for the cost-benefit analysis workflow.",
    )
    constrain_dsr: bool = Field(
        True,
        description="Toggles specific dispatch constraints for DSR units within the CBA workflow.",
    )
    msv_extraction: _CbaMsvExtractionConfig = Field(
        default_factory=_CbaMsvExtractionConfig,
        description="Marginal storage value extraction settings for seasonal storage dispatch.",
    )
    solving: _CbaSolvingConfig = Field(
        default_factory=_CbaSolvingConfig,
        description="Configuration for rolling horizon network optimization. Uses the same structure as the top-level `solving` section.",
    )