Bring the translator and pypsa_build up to the new table formats

src/ispypsa/cli/dodo.py

@@ -15,6 +15,7 @@
     read_csvs,
     write_csvs,
 )
+from ispypsa.feature_flags import FEATURE_FLAGS
 from ispypsa.iasr_table_caching import build_local_cache, list_cache_files
 from ispypsa.logging import configure_logging
 from ispypsa.plotting import (
@@ -483,15 +484,20 @@ def create_pypsa_inputs_for_capacity_expansion_model() -> None:
     ispypsa_tables = read_csvs(input_tables_dir)
     pypsa_tables = create_pypsa_friendly_inputs(config, ispypsa_tables)
 
-    # Create capacity expansion timeseries
-    pypsa_tables["snapshots"] = create_pypsa_friendly_timeseries_inputs(
-        config,
-        "capacity_expansion",
-        ispypsa_tables,
-        pypsa_tables["generators"],
-        parsed_trace_dir,
-        capacity_expansion_timeseries_location,
-    )
+    # FEATURE_FLAG_CLEANUP[use_new_table_format]: revisit once generator
+    # translation lands for the new format. Until then the new-format path
+    # has no generators, so no timeseries inputs are created and snapshots
+    # come from create_pypsa_friendly_inputs itself.
+    if not FEATURE_FLAGS["use_new_table_format"]:
+        # Create capacity expansion timeseries
+        pypsa_tables["snapshots"] = create_pypsa_friendly_timeseries_inputs(
+            config,
+            "capacity_expansion",
+            ispypsa_tables,
+            pypsa_tables["generators"],
+            parsed_trace_dir,
+            capacity_expansion_timeseries_location,
+        )
 
     write_csvs(pypsa_tables, pypsa_friendly_dir)
 

src/ispypsa/pypsa_build/build.py

@@ -7,7 +7,10 @@
     _add_buses_to_network,
 )
 from ispypsa.pypsa_build.carriers import _add_carriers_to_network
-from ispypsa.pypsa_build.custom_constraints import _add_custom_constraints
+from ispypsa.pypsa_build.custom_constraints import (
+    _add_custom_constraints,
+    _add_custom_constraints_with_temporal_scope,
+)
 from ispypsa.pypsa_build.generators import (
     _add_custom_constraint_generators_to_network,
     _add_generators_to_network,
@@ -70,13 +73,20 @@ def build_pypsa_network(
     )
 
     if "links" in pypsa_friendly_tables.keys():
-        _add_links_to_network(network, pypsa_friendly_tables["links"])
+        _add_links_to_network(
+            network,
+            pypsa_friendly_tables["links"],
+            pypsa_friendly_tables.get("link_timeslice_limits"),
+            pypsa_friendly_tables.get("timeslice_snapshots"),
+        )
 
-    _add_generators_to_network(
-        network,
-        pypsa_friendly_tables["generators"],
-        path_to_pypsa_friendly_timeseries_data,
-    )
+    # Absent under the new table format until generator translation lands.
+    if "generators" in pypsa_friendly_tables.keys():
+        _add_generators_to_network(
+            network,
+            pypsa_friendly_tables["generators"],
+            path_to_pypsa_friendly_timeseries_data,
+        )
 
     if "batteries" in pypsa_friendly_tables.keys():
         _add_batteries_to_network(network, pypsa_friendly_tables["batteries"])
@@ -91,11 +101,22 @@ def build_pypsa_network(
     # The underlying linopy model needs to get built so we can add custom constraints.
     network.optimize.create_model(multi_investment_periods=True)
 
+    # FEATURE_FLAG_CLEANUP[use_new_table_format]: the temporal-scope path
+    # becomes the only path once the old format is deleted; the column check
+    # distinguishes the formats until then.
     if "custom_constraints_rhs" in pypsa_friendly_tables:
-        _add_custom_constraints(
-            network,
-            pypsa_friendly_tables["custom_constraints_rhs"],
-            pypsa_friendly_tables["custom_constraints_lhs"],
-        )
+        if "investment_period" in pypsa_friendly_tables["custom_constraints_rhs"]:
+            _add_custom_constraints_with_temporal_scope(
+                network,
+                pypsa_friendly_tables["custom_constraints_rhs"],
+                pypsa_friendly_tables["custom_constraints_lhs"],
+                pypsa_friendly_tables["timeslice_snapshots"],
+            )
+        else:
+            _add_custom_constraints(
+                network,
+                pypsa_friendly_tables["custom_constraints_rhs"],
+                pypsa_friendly_tables["custom_constraints_lhs"],
+            )
 
     return network

src/ispypsa/pypsa_build/custom_constraints.py

@@ -5,6 +5,8 @@
 import pandas as pd
 import pypsa
 
+logger = logging.getLogger(__name__)
+
 
 def _get_variables(
     model: linopy.Model, component_name: str, component_type: str, attribute_type: str
@@ -115,3 +117,186 @@ def _add_custom_constraints(
             raise ValueError(
                 f"{row['constraint_type']} is not a valid constraint type."
             )
+
+
+def _add_custom_constraints_with_temporal_scope(
+    network: pypsa.Network,
+    custom_constraints_rhs: pd.DataFrame,
+    custom_constraints_lhs: pd.DataFrame,
+    timeslice_snapshots: pd.DataFrame,
+):
+    """Adds the new-format custom constraints to the `pypsa.Network`.
+
+    Each RHS row creates one linopy constraint scoped to the row's
+    investment_period and timeslice: time-indexed LHS variables are
+    restricted to the snapshots inside that scope. NaN investment_period or
+    timeslice means unrestricted in that dimension (e.g. the expansion-limit
+    constraints, which only involve p_nom variables, carry NaN in both).
+
+    LHS terms referencing components not in the model (e.g. generators while
+    generator translation for the new format is pending) are skipped with a
+    log line; constraint instances with no snapshots in scope or no terms in
+    the model are skipped silently — the translator logs the timeslices that
+    never apply.
+
+    Args:
+        network: The `pypsa.Network` object
+        custom_constraints_rhs: `pd.DataFrame` with columns constraint_name,
+            investment_period, timeslice, rhs, and constraint_type.
+        custom_constraints_lhs: `pd.DataFrame` with columns constraint_name,
+            investment_period, variable_name, component, attribute, and
+            coefficient. A NaN investment_period term applies to every
+            instance of its constraint.
+        timeslice_snapshots: `pd.DataFrame` mapping timeslice_ids to the
+            snapshots they are active at (columns timeslice_id,
+            investment_periods, snapshots).
+
+    Returns: None
+    """
+    timeslice_labels = _timeslice_constraint_labels(timeslice_snapshots)
+    for row in custom_constraints_rhs.itertuples():
+        snapshot_subset = _constraint_snapshot_subset(
+            row, timeslice_labels, network.snapshots
+        )
+        if snapshot_subset is not None and len(snapshot_subset) == 0:
+            continue
+        terms = _select_lhs_terms(custom_constraints_lhs, row)
+        expression = _build_constraint_expression(network.model, terms, snapshot_subset)
+        if expression is None:
+            continue
+        _add_constraint_for_rhs_row(network.model, expression, row)
+
+
+def _timeslice_constraint_labels(
+    timeslice_snapshots: pd.DataFrame,
+) -> dict[str, list[tuple]]:
+    """The (investment_period, snapshot) labels each timeslice is active at.
+
+    I/O Example:
+        timeslice_id=qld_peak_demand, investment_periods=2025,
+        snapshots=2025-01-13 12:00
+        -> {"qld_peak_demand": [(2025, Timestamp("2025-01-13 12:00"))]}
+    """
+    mapping = timeslice_snapshots.copy()
+    mapping["snapshots"] = pd.to_datetime(mapping["snapshots"])
+    return {
+        timeslice_id: list(zip(rows["investment_periods"], rows["snapshots"]))
+        for timeslice_id, rows in mapping.groupby("timeslice_id")
+    }
+
+
+def _constraint_snapshot_subset(
+    rhs_row, timeslice_labels: dict[str, list[tuple]], snapshots: pd.MultiIndex
+) -> list[tuple] | None:
+    """The snapshot labels an RHS row's constraint is restricted to, or None
+    when the constraint is unrestricted (NaN timeslice and investment_period).
+
+    I/O Example:
+        timeslice=qld_peak_demand, investment_period=2025,
+        timeslice_labels={"qld_peak_demand": [(2025, t1), (2030, t2)]}
+        -> [(2025, t1)]
+
+        timeslice=NaN, investment_period=2025 -> all 2025 snapshots
+        timeslice=NaN, investment_period=NaN  -> None
+    """
+    if not pd.isna(rhs_row.timeslice):
+        labels = timeslice_labels.get(rhs_row.timeslice, [])
+        if not pd.isna(rhs_row.investment_period):
+            period = int(rhs_row.investment_period)
+            labels = [label for label in labels if label[0] == period]
+        return labels
+    if not pd.isna(rhs_row.investment_period):
+        period = int(rhs_row.investment_period)
+        return list(snapshots[snapshots.get_level_values(0) == period])
+    return None
+
+
+def _select_lhs_terms(custom_constraints_lhs: pd.DataFrame, rhs_row) -> pd.DataFrame:
+    """The LHS terms belonging to an RHS row's constraint instance: terms for
+    the same constraint whose investment_period matches the row's or is NaN
+    (applies to every instance).
+
+    I/O Example:
+        lhs:
+            constraint_name  investment_period  variable_name
+            SWQLD1           2025               KINGASF1
+            SWQLD1           2030               KINGASF1
+            SWQLD1           ,                  SWQLD1_exp_2025
+
+        rhs_row (SWQLD1, investment_period=2025) selects rows 1 and 3.
+    """
+    terms = custom_constraints_lhs[
+        custom_constraints_lhs["constraint_name"] == rhs_row.constraint_name
+    ]
+    if pd.isna(rhs_row.investment_period):
+        return terms
+    return terms[
+        terms["investment_period"].isna()
+        | (terms["investment_period"] == rhs_row.investment_period)
+    ]
+
+
+def _build_constraint_expression(
+    model: linopy.Model, terms: pd.DataFrame, snapshot_subset: list[tuple] | None
+):
+    """Builds the linear expression sum(coefficient * variable) for a
+    constraint instance, restricting time-indexed variables to the snapshot
+    subset. Returns None when no term's variable is in the model."""
+    expression_terms = []
+    for term in terms.itertuples():
+        variable = _get_variable_with_snapshot_subset(
+            model, term.variable_name, term.component, term.attribute, snapshot_subset
+        )
+        if variable is not None:
+            expression_terms.append((term.coefficient, variable))
+    if not expression_terms:
+        return None
+    return model.linexpr(*expression_terms)
+
+
+def _get_variable_with_snapshot_subset(
+    model: linopy.Model,
+    component_name: str,
+    component_type: str,
+    attribute_type: str,
+    snapshot_subset: list[tuple] | None,
+):
+    """Retrieves a variable like _get_variables, restricting time-indexed
+    ('p') variables to the snapshot subset. Components not in the model are
+    skipped with a log line rather than raising — new-format constraints
+    legitimately reference generators and batteries before those components
+    are translated."""
+    try:
+        variable = _get_variables(model, component_name, component_type, attribute_type)
+    except KeyError:
+        logger.info(
+            f"{component_type} {component_name} not in model, "
+            f"custom constraint term skipped."
+        )
+        return None
+    if variable is not None and attribute_type == "p" and snapshot_subset is not None:
+        variable = variable.loc[snapshot_subset]
+    return variable
+
+
+def _add_constraint_for_rhs_row(model: linopy.Model, expression, rhs_row) -> None:
+    """Adds one constraint, named uniquely for the RHS row's temporal scope.
+
+    I/O Example:
+        (SWQLD1, 2025, qld_peak_demand) -> "SWQLD1_2025_qld_peak_demand"
+        (CQ-NQ_expansion_limit, NaN, NaN) -> "CQ-NQ_expansion_limit"
+    """
+    name_parts = [rhs_row.constraint_name]
+    if not pd.isna(rhs_row.investment_period):
+        name_parts.append(str(int(rhs_row.investment_period)))
+    if not pd.isna(rhs_row.timeslice):
+        name_parts.append(rhs_row.timeslice)
+    name = "_".join(name_parts)
+    if rhs_row.constraint_type == "<=":
+        model.add_constraints(expression <= rhs_row.rhs, name=name)
+    elif rhs_row.constraint_type == ">=":
+        model.add_constraints(expression >= rhs_row.rhs, name=name)
+    elif rhs_row.constraint_type == "==":
+        model.add_constraints(expression == rhs_row.rhs, name=name)
+    else:
+        raise ValueError(f"{rhs_row.constraint_type} is not a valid constraint type.")

src/ispypsa/pypsa_build/links.py

@@ -2,15 +2,92 @@
 import pypsa
 
 
-def _add_links_to_network(network: pypsa.Network, links: pd.DataFrame) -> None:
+def _add_links_to_network(
+    network: pypsa.Network,
+    links: pd.DataFrame,
+    link_timeslice_limits: pd.DataFrame | None = None,
+    timeslice_snapshots: pd.DataFrame | None = None,
+) -> None:
     """Adds the Links defined in a pypsa-friendly input table called `"links"` to the
     `pypsa.Network` object.
 
+    When the new-format per-timeslice limit tables are given, links with
+    timeslice-varying limits get per-snapshot p_max_pu / p_min_pu series
+    instead of their static values (see _build_link_pu_overrides).
+
     Args:
         network: The `pypsa.Network` object
         links: `pd.DataFrame` with `PyPSA` style `Link` attributes.
+        link_timeslice_limits: `pd.DataFrame` with per-timeslice per-unit
+            limits (columns name, attribute, timeslice, value), or None when
+            all link limits are static.
+        timeslice_snapshots: `pd.DataFrame` mapping timeslice_ids to the
+            snapshots they are active at (columns timeslice_id,
+            investment_periods, snapshots). Required when
+            link_timeslice_limits is given.
 
     Returns: None
     """
+    pu_overrides = _build_link_pu_overrides(
+        link_timeslice_limits, timeslice_snapshots, links, network.snapshots
+    )
     links["class_name"] = "Link"
-    links.apply(lambda row: network.add(**row.to_dict()), axis=1)
+    for _, row in links.iterrows():
+        network.add(**(row.to_dict() | pu_overrides.get(row["name"], {})))
+
+
+def _build_link_pu_overrides(
+    link_timeslice_limits: pd.DataFrame | None,
+    timeslice_snapshots: pd.DataFrame | None,
+    links: pd.DataFrame,
+    snapshots: pd.MultiIndex,
+) -> dict[str, dict[str, pd.Series]]:
+    """Expands each link's per-timeslice limits into per-snapshot series.
+
+    Each series starts at the link's static value (1.0 for p_max_pu, the
+    links table's p_min_pu for p_min_pu — both equivalent to the
+    winter_reference limit) and gets each timeslice's value at the snapshots
+    that timeslice is active. Timeslices with no snapshots simply leave the
+    static value in place — the translator has already logged them.
+
+    I/O Example:
+        link_timeslice_limits:
+            name            attribute  timeslice        value
+            CQ-NQ_existing  p_max_pu   qld_peak_demand  0.857
+
+        timeslice_snapshots: qld_peak_demand active at (2025, 2025-01-13 12:00)
+        snapshots: (2025, 2025-01-13 12:00), (2025, 2025-01-15 12:00)
+
+        returns:
+            {"CQ-NQ_existing": {"p_max_pu": series [0.857, 1.0]}}
+    """
+    if link_timeslice_limits is None or link_timeslice_limits.empty:
+        return {}
+    timeslice_labels = _timeslice_snapshot_labels(timeslice_snapshots)
+    static_p_min_pu = links.set_index("name")["p_min_pu"]
+    overrides = {}
+    for (name, attribute), rows in link_timeslice_limits.groupby(["name", "attribute"]):
+        static_value = 1.0 if attribute == "p_max_pu" else static_p_min_pu[name]
+        series = pd.Series(static_value, index=snapshots, dtype=float)
+        for row in rows.itertuples():
+            series.loc[timeslice_labels.get(row.timeslice, [])] = row.value
+        overrides.setdefault(name, {})[attribute] = series
+    return overrides
+
+
+def _timeslice_snapshot_labels(
+    timeslice_snapshots: pd.DataFrame,
+) -> dict[str, list[tuple]]:
+    """The (investment_period, snapshot) labels each timeslice is active at.
+
+    I/O Example:
+        timeslice_id=qld_peak_demand, investment_periods=2025,
+        snapshots=2025-01-13 12:00
+        -> {"qld_peak_demand": [(2025, Timestamp("2025-01-13 12:00"))]}
+    """
+    mapping = timeslice_snapshots.copy()
+    mapping["snapshots"] = pd.to_datetime(mapping["snapshots"])
+    return {
+        timeslice_id: list(zip(rows["investment_periods"], rows["snapshots"]))
+        for timeslice_id, rows in mapping.groupby("timeslice_id")
+    }