Write a Sienna investments portfolio from an expansion plan

docs/translation_mappings/translation-from-pypsa-to-sienna.md

@@ -11,7 +11,7 @@ Concretely, versus `main` (≈280 insertions / 294 deletions):
 - **Time series.** Replaced the `SingleTimeSeries`/`add_time_series!` framing with SiennaSchemas `TimeSeriesAssociation` records (integer `owner_id`); value arrays are **retained in HDF5** keyed by `time_series_uuid` (the schema's named external store). Standardised on the **per-unit shape + `scaling_factor_multiplier`** convention.
 - **Component extensions.** PyPSA fields with no SiennaSchemas home now go in a separate `extensions.json` sidecar, a document keyed by kind whose records are identified by `name`, replacing the PowerSystems.jl `ext` dict.
 - **References & ordering.** `get_component(...)` / "added to the `System`" became integer-`id` references with an id-assignment ordering (the lists themselves are unordered).
-- **Investments boundary.** Added that SiennaSchemas' `Investments/` namespace is a separate, parallel model (the home for capacity expansion); this document targets **Operations** (PowerSimulations.jl).
+- **Investments boundary.** Added that SiennaSchemas' `Investments/` namespace is a separate, parallel model (the home for capacity expansion); this document targets **Operations** (PowerSimulations.jl), and `pypsa-to-sienna-investments` translates the expansion problem.
 
 ---
 
@@ -79,7 +79,7 @@ This document targets the SiennaSchemas **Operations** namespace (the dispatchab
 - It adds policy `Requirements` (`CarbonCaps`, `CarbonTax`, `CapacityReserveMargin`, `EnergyShareRequirements`, …) and `Financials`.
 - It bridges to Operations through each technology's `power_systems_type` (the Operations type it realises into) and `TopologyMapping` (region → buses).
 
-**Implication for this translation.** PyPSA's capacity-expansion fields (`p_nom_extendable`, `capital_cost`, `p_nom_min`/`p_nom_max`, `build_year`, `lifetime`) belong to **Investments**, not extra fields on an Operations component. This document is the Operations translation, so it rates each component by the capacity an operations model may dispatch: `effective_p_nom` reads `p_nom_opt`, then `p_nom_min`, then `p_nom`, as the paragraph below states. Every other expansion parameter is dropped, or kept in the `ext` sidecar for the round trip.
+**Implication for this translation.** PyPSA's capacity-expansion fields (`p_nom_extendable`, `capital_cost`, `p_nom_min`/`p_nom_max`, `build_year`, `lifetime`) belong to **Investments**, not extra fields on an Operations component. This document is the Operations translation, so it rates each component by the capacity an operations model may dispatch: `effective_p_nom` reads `p_nom_opt`, then `p_nom_min`, then `p_nom`, as the paragraph below states. Every other expansion parameter is dropped, or kept in the `ext` sidecar for the round trip. The `pypsa-to-sienna-investments` pipeline translates the expansion problem itself, and writes a portfolio document beside the operations system. It is the second leg of `plexos-to-sienna-investments`, so [Translation from PLEXOS to a Sienna investments portfolio](translation-from-plexos-to-sienna-investments.md) states the source of each technology, requirement and attribute field the portfolio holds. The `From` column of that document names the PLEXOS property behind each field, not the PyPSA one.
 
 Throughout this document, **`effective_p_nom`** denotes the capacity an operations model may dispatch: `p_nom_opt` where an extendable component states one, `p_nom_min` where it states a capacity a build cannot take away, and `p_nom` otherwise. All capacity-derived fields — `base_power`, `active_power`, `active_power_limits`, `ramp_limits`, and the hydro energy budget scaling factor — use `effective_p_nom`. For a non-extendable component `effective_p_nom = p_nom`. A solve that builds none of an extendable component writes `p_nom_opt` 0, and that 0 is the capacity to translate: the plan rejected the build. A network no solve has touched leaves `p_nom_opt` out of the file altogether. An extendable component there is rated at the lower of its `p_nom_min` and its `p_nom`, because a `p_nom_min` above `p_nom` is a build the model forces rather than capacity the component already runs. An extendable component whose lower value is 0 is a candidate rather than a plant: the generator, renewable, hydro, storage and link mappings leave it out and record it as skipped.
 

interop/pipelines/plexos-to-sienna-investments.yaml

@@ -0,0 +1,13 @@
+source_framework: plexos
+destination_framework: sienna
+mappings:
+  - pipeline: derive-plexos-sienna-mappings
+compose:
+  - pipeline: plexos-to-pypsa
+    params:
+      emit_pypsa_network.output_path: network.nc
+      emit_extensions_json.output_path: extensions.json
+  - pipeline: pypsa-to-sienna-investments
+    params:
+      stage_pypsa_network_file.path: $plexos-to-pypsa.emit_pypsa_network.output_path
+      stage_pypsa_network_file.extensions_json_path: $plexos-to-pypsa.emit_extensions_json.output_path

interop/pipelines/pypsa-to-sienna-investments.yaml

@@ -0,0 +1,23 @@
+source_framework: pypsa
+destination_framework: sienna
+source:
+  name: stage_pypsa_network_file
+steps:
+  - name: pypsa_to_sienna_map_components
+  - name: pypsa_to_sienna_relate_components
+  - name: pypsa_to_sienna_investments_map_technologies
+validators:
+  - name: pypsa_bus_reference_integrity
+  - name: pypsa_generators
+  - name: pypsa_storage_units
+  - name: pypsa_loads
+  - name: pypsa_lines
+  - name: pypsa_links
+  - name: pypsa_unique_names
+sinks:
+  - name: emit_sienna_files
+    params:
+      output_system_json_file_path: outputs/system.json
+  - name: emit_sienna_portfolio
+    params:
+      output_path: outputs/portfolio.json

interop/plugins/shared/pypsa_sienna_investments_translations/__init__.py

@@ -0,0 +1,71 @@
+from ._associations import build_association_rows
+from ._carbon_caps import (
+    CARBON_CAP_SKIPS,
+    build_carbon_cap_translations,
+    build_carbon_caps_source_table,
+)
+from ._demand import LOAD_TYPE_COL, build_demand_translations
+from ._existing import (
+    TECHNOLOGY_NAME,
+    TECHNOLOGY_TYPE,
+    CandidateTechnology,
+    build_existing_devices_translations,
+    build_existing_fleet_source_table,
+    build_retirement_potential_translations,
+)
+from ._financials import build_financial_data_events, build_portfolio_financial_data
+from ._shared import (
+    FOM_CHARGE_COL,
+    FOM_CHARGE_DERIVATION,
+    FUEL_COL,
+    POWER_SYSTEMS_TYPE_COL,
+    PRIME_MOVER_COL,
+    REGION_COL,
+    TECHNICAL_LIFE_COL,
+    UNIT_SIZE_COL,
+    build_scope_skips,
+    yearly_fixed_charge,
+)
+from ._storage import (
+    STORAGE_SKIPS,
+    build_storage_technology_translations,
+    fill_storage_technology_defaults,
+)
+from ._supply import SUPPLY_SKIPS, build_supply_translations, fill_supply_defaults
+from ._topology import AREA_NAME, build_topology_mapping_translations, build_topology_source_table
+
+__all__ = [
+    "AREA_NAME",
+    "CARBON_CAP_SKIPS",
+    "FUEL_COL",
+    "LOAD_TYPE_COL",
+    "POWER_SYSTEMS_TYPE_COL",
+    "PRIME_MOVER_COL",
+    "REGION_COL",
+    "STORAGE_SKIPS",
+    "SUPPLY_SKIPS",
+    "FOM_CHARGE_COL",
+    "FOM_CHARGE_DERIVATION",
+    "yearly_fixed_charge",
+    "TECHNICAL_LIFE_COL",
+    "TECHNOLOGY_NAME",
+    "TECHNOLOGY_TYPE",
+    "UNIT_SIZE_COL",
+    "CandidateTechnology",
+    "build_association_rows",
+    "build_carbon_cap_translations",
+    "build_carbon_caps_source_table",
+    "build_demand_translations",
+    "build_existing_devices_translations",
+    "build_existing_fleet_source_table",
+    "build_financial_data_events",
+    "build_portfolio_financial_data",
+    "build_retirement_potential_translations",
+    "build_scope_skips",
+    "build_storage_technology_translations",
+    "build_supply_translations",
+    "build_topology_mapping_translations",
+    "build_topology_source_table",
+    "fill_storage_technology_defaults",
+    "fill_supply_defaults",
+]

interop/plugins/shared/pypsa_sienna_investments_translations/_associations.py

@@ -0,0 +1,37 @@
+"""The rows linking each supplemental attribute to the component it describes.
+
+The attributes travel in one flat array, so the association carries the type of the
+attribute as well as its id, and the type of the component as well as its name. The sink
+resolves the name to an id.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import polars as pl
+
+from interop.plugins.shared.sienna_investments_constants import (
+    SUPPLEMENTAL_ATTRIBUTE_ASSOCIATION_SCHEMA,
+    SiennaSupplementalAttribute,
+    SiennaSupplementalAttributeAssociationCol,
+)
+
+A = SiennaSupplementalAttributeAssociationCol
+
+
+def build_association_rows(
+    attribute_type: SiennaSupplementalAttribute,
+    attribute_ids: Sequence[int],
+    component_names: Sequence[str],
+    component_types: Sequence[str],
+) -> pl.DataFrame:
+    return pl.DataFrame(
+        {
+            A.COMPONENT_NAME: list(component_names),
+            A.COMPONENT_TYPE: list(component_types),
+            A.ATTRIBUTE_ID: list(attribute_ids),
+            A.ATTRIBUTE_TYPE: [str(attribute_type)] * len(attribute_ids),
+        },
+        schema=SUPPLEMENTAL_ATTRIBUTE_ASSOCIATION_SCHEMA,
+    )

interop/plugins/shared/pypsa_sienna_investments_translations/_carbon_caps.py

@@ -0,0 +1,227 @@
+"""Translation objects for an extensions-sidecar constraint -> Sienna CarbonCaps."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import partial
+from typing import Any
+
+import polars as pl
+
+from interop.core.extensions import (
+    ConstraintExtension,
+    ConstraintLimit,
+    ConstraintPeriod,
+    ConstraintSense,
+    ExtensionKind,
+)
+from interop.plugins.shared.pypsa_sienna_investments_translations._shared import (
+    PORTFOLIO_ID_NOTE,
+    investments_skip_report,
+)
+from interop.plugins.shared.pypsa_sienna_translations._shared import (
+    pypsa_source_field,
+    sienna_dest_field,
+)
+from interop.plugins.shared.sienna_constants import SIENNA_TYPE_ATTRIBUTE
+from interop.plugins.shared.sienna_investments_constants import (
+    SiennaCarbonCapsCol,
+    SiennaInvestmentsComponent,
+)
+from interop.plugins.shared.translation_runner import (
+    SkipRule,
+    Translation,
+    default_translation,
+    direct_translation,
+    row_position_id_translation,
+)
+from interop.ports.outbound.reporting import EventKind, TranslationEvent
+
+# Source-table columns read off a sidecar record, none of which is a schema field.
+CONSTRAINT_NAME = "name"
+CONSTRAINT_SENSE = "sense"
+LIMIT_PERIOD = "limit_period"
+LIMIT_VALUE = "limit_value"
+LIMIT_UNIT = "limit_unit"
+COVERS_MODEL = "covers_model"
+APPLIES_TO_PLAN = "applies_to_expansion_plan"
+
+CARBON_CAPS_SOURCE_SCHEMA: dict[str, pl.DataType | type[pl.DataType]] = {
+    CONSTRAINT_NAME: pl.Utf8,
+    CONSTRAINT_SENSE: pl.Utf8,
+    LIMIT_PERIOD: pl.Utf8,
+    LIMIT_VALUE: pl.Float64,
+    LIMIT_UNIT: pl.Utf8,
+    COVERS_MODEL: pl.Boolean,
+    APPLIES_TO_PLAN: pl.Boolean,
+}
+
+# The spans a cap is read from, most specific first: no other span bounds the whole run.
+_CAP_PERIODS: tuple[ConstraintPeriod, ...] = (ConstraintPeriod.YEAR, ConstraintPeriod.HORIZON)
+
+_source = partial(pypsa_source_field, ExtensionKind.CONSTRAINT)
+_dest = partial(sienna_dest_field, SiennaInvestmentsComponent.CARBON_CAPS)
+
+C = SiennaCarbonCapsCol
+
+_direct = partial(direct_translation, _source, _dest, name_col=CONSTRAINT_NAME)
+_default = partial(default_translation, _dest, name_col=CONSTRAINT_NAME)
+
+_constraint_skip = partial(
+    investments_skip_report,
+    component=ExtensionKind.CONSTRAINT,
+    name_col=CONSTRAINT_NAME,
+    counted_noun="constraint(s)",
+)
+
+SCOPED_CONSTRAINT_SKIP = _constraint_skip(
+    reason="weight a named subset of the model rather than all of it",
+    note=(
+        "CarbonCaps names no members and no region, so a cap written from this constraint "
+        "would hold the whole portfolio rather than the components the constraint names"
+    ),
+)
+
+WRONG_SENSE_SKIP = _constraint_skip(
+    reason="hold their weighted sum to something other than a ceiling",
+    note="CarbonCaps states an upper limit, and this constraint does not",
+    attribute_col=CONSTRAINT_SENSE,
+)
+
+NO_LIMIT_SKIP = _constraint_skip(
+    reason="state no yearly or horizon-wide right-hand side",
+    note="a cap with no limit bounds nothing, and no other span bounds the whole run",
+)
+
+NOT_IN_PLAN_SKIP = _constraint_skip(
+    reason="the expansion plan does not have to meet",
+    note=(
+        "the source states that the plan need not meet this constraint, so a cap written "
+        "from it would bound an expansion problem the model leaves free"
+    ),
+    attribute_col=APPLIES_TO_PLAN,
+)
+
+NON_FINITE_LIMIT_SKIP = _constraint_skip(
+    reason="state a right-hand side that is not a finite number",
+    note="max_mtons would be NaN or Infinity, which no JSON reader accepts as a number",
+    attribute_col=LIMIT_VALUE,
+)
+
+CARBON_CAP_SKIPS: tuple[SkipRule, ...] = (
+    # A source that states nothing about the plan leaves every constraint in it.
+    SkipRule(keep=pl.col(APPLIES_TO_PLAN).fill_null(value=True), report=NOT_IN_PLAN_SKIP),
+    SkipRule(keep=pl.col(CONSTRAINT_SENSE) == ConstraintSense.AT_MOST, report=WRONG_SENSE_SKIP),
+    SkipRule(keep=pl.col(LIMIT_VALUE).is_not_null(), report=NO_LIMIT_SKIP),
+    # is_finite answers null for a null limit, so this rule follows the one that drops those.
+    SkipRule(keep=pl.col(LIMIT_VALUE).is_finite(), report=NON_FINITE_LIMIT_SKIP),
+    SkipRule(keep=pl.col(COVERS_MODEL), report=SCOPED_CONSTRAINT_SKIP),
+)
+
+
+def build_carbon_caps_source_table(
+    records: Sequence[ConstraintExtension], model_components: set[tuple[str, str]]
+) -> pl.DataFrame:
+    """One row per sidecar constraint, with the limit a cap would read and its reach.
+
+    ``model_components`` is every component of the network a constraint could weight, each
+    named with the class it belongs to. A constraint naming all of them holds the whole
+    model; one naming fewer holds a subset. Two classes can hold an object of one name, so a
+    member counts only where its class matches as well, and a member whose class the network
+    states differently leaves the constraint short of the whole model.
+    """
+    rows = [_read_record(record, model_components) for record in records]
+    return pl.DataFrame(rows, schema=CARBON_CAPS_SOURCE_SCHEMA)
+
+
+def _read_record(
+    record: ConstraintExtension, model_components: set[tuple[str, str]]
+) -> dict[str, Any]:
+    limit = _choose_limit(record)
+    members = {(member.name, member.member_class) for member in record.members}
+    return {
+        CONSTRAINT_NAME: record.name,
+        CONSTRAINT_SENSE: None if record.sense is None else str(record.sense),
+        LIMIT_PERIOD: None if limit is None else str(limit.period),
+        LIMIT_VALUE: None if limit is None else limit.value,
+        LIMIT_UNIT: None if limit is None else limit.unit,
+        COVERS_MODEL: bool(model_components) and model_components <= members,
+        APPLIES_TO_PLAN: record.applies_to_expansion_plan,
+    }
+
+
+def _choose_limit(record: ConstraintExtension) -> ConstraintLimit | None:
+    for period in _CAP_PERIODS:
+        for limit in record.limits:
+            if limit.period == period:
+                return limit
+    return None
+
+
+CARBON_CAP_NAME = _direct(source_col=CONSTRAINT_NAME, dest_col=C.NAME)
+
+CARBON_CAP_AVAILABLE = _default(
+    dest_col=C.AVAILABLE,
+    value=True,
+    note="a constraint the sidecar carries is one the model applies",
+)
+
+CARBON_CAP_SIENNA_TYPE = Translation(
+    exprs=[],
+    make_events=lambda old, _: [
+        TranslationEvent(
+            kind=EventKind.VALUE_DERIVED,
+            sources=[_source(old[CONSTRAINT_NAME], CONSTRAINT_SENSE, old[CONSTRAINT_SENSE])],
+            destinations=[
+                _dest(
+                    old[CONSTRAINT_NAME],
+                    SIENNA_TYPE_ATTRIBUTE,
+                    SiennaInvestmentsComponent.CARBON_CAPS,
+                )
+            ],
+            derivation="a ceiling over every component of the model -> CarbonCaps",
+        )
+    ],
+)
+
+CARBON_CAP_MAX_MTONS = Translation(
+    exprs=[pl.col(LIMIT_VALUE).alias(C.MAX_MTONS)],
+    make_events=lambda old, new: [
+        TranslationEvent(
+            kind=EventKind.VALUE_DERIVED,
+            sources=[
+                _source(
+                    old[CONSTRAINT_NAME],
+                    f"{LIMIT_VALUE} ({old[LIMIT_PERIOD]})",
+                    old[LIMIT_VALUE],
+                    old[LIMIT_UNIT],
+                )
+            ],
+            destinations=[_dest(old[CONSTRAINT_NAME], C.MAX_MTONS, new[C.MAX_MTONS])],
+            derivation=f"the {old[LIMIT_PERIOD]} right-hand side of the constraint",
+            note=(
+                "max_mtons is read in million tonnes; the constraint states "
+                f"{old[LIMIT_UNIT]}, and no conversion is applied"
+                if old[LIMIT_UNIT]
+                else "max_mtons is read in million tonnes, and the constraint names no unit"
+            ),
+        )
+    ],
+)
+
+CARBON_CAP_TRANSLATIONS: list[Translation] = [
+    CARBON_CAP_NAME,
+    CARBON_CAP_AVAILABLE,
+    CARBON_CAP_SIENNA_TYPE,
+    CARBON_CAP_MAX_MTONS,
+]
+
+
+def build_carbon_cap_translations(start: int) -> list[Translation]:
+    """Every CarbonCaps translation, including the one the shared id counter decides."""
+    return [
+        row_position_id_translation(
+            _dest, dest_name_col=C.NAME, id_col=C.ID, note=PORTFOLIO_ID_NOTE, start=start
+        ),
+        *CARBON_CAP_TRANSLATIONS,
+    ]