feat(piecewise): partial active gate support via active_gate helper (#796)

doc/api.rst

@@ -480,6 +480,7 @@ Construction helpers
    piecewise.breakpoints
    piecewise.segments
    piecewise.Slopes
+   active_gate
 
 PiecewiseFormulation
 --------------------

doc/piecewise-linear-constraints.rst

@@ -565,6 +565,22 @@ manual gating.
    variable: combined with the ``y ≤ 0`` constraint from deactivation,
    this forces ``y = 0`` automatically.
 
+Partial gates
+^^^^^^^^^^^^^
+
+``active`` must cover the formulation's full coordinate; a gate defined
+over only a subset (or with masked entries) is rejected. Pad it with
+:func:`~linopy.active_gate`, which leaves missing entries always-active
+(``fill_value=1``) or off (``0``) — handy when one formulation mixes
+committable and non-committable units sharing a single ``status``:
+
+.. code-block:: python
+
+    gate = linopy.active_gate(status, {"unit": units})
+    m.add_piecewise_formulation(
+        (power, [30, 60, 100]), (fuel, [40, 90, 170]), active=gate
+    )
+
 Auto-broadcasting
 ~~~~~~~~~~~~~~~~~
 

doc/release_notes.rst

@@ -20,6 +20,7 @@ Upcoming Version
 *Other*
 
 * ``add_variables(binary=True, ...)`` now accepts ``lower``/``upper`` bounds, as long as they are 0 or 1. Previously binary bounds could only be set via the ``.lower``/``.upper`` setters after creation. (https://github.com/PyPSA/linopy/issues/776)
+* New ``linopy.active_gate`` helper pads a partial ``active`` gate for ``add_piecewise_formulation`` to full coverage (missing/masked entries become always-active, or off with ``fill_value=0``). Partial gates that were previously zeroed silently are now rejected with a clear error. (https://github.com/PyPSA/linopy/issues/796)
 
 **Deprecations**
 

linopy/__init__.py

@@ -12,6 +12,7 @@
 # Note: For intercepting multiplications between xarray dataarrays, Variables and Expressions
 # we need to extend their __mul__ functions with a quick special case
 import linopy.monkey_patch_xarray  # noqa: F401
+from linopy._active_gate import active_gate
 from linopy.alignment import align
 from linopy.config import options
 from linopy.constants import (
@@ -67,6 +68,7 @@
     "SolverFeature",
     "Variable",
     "Variables",
+    "active_gate",
     "align",
     "available_solvers",
     "licensed_solvers",

linopy/_active_gate.py

@@ -0,0 +1,73 @@
+"""
+Legacy helper for padding a partial piecewise ``active`` gate.
+
+This module is a temporary stopgap.  Under the planned v1 arithmetic
+semantics (#717) the bare idiom ``active.reindex(coords).fillna(fill_value)``
+is correct on its own, so :func:`active_gate` is expected to be deprecated
+and this file removed once v1 lands.  Keeping it isolated makes that
+removal a single-file delete.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Any
+
+from linopy.piecewise import _to_linexpr, _warn_evolving_api
+
+if TYPE_CHECKING:
+    from linopy.expressions import LinearExpression
+    from linopy.types import LinExprLike
+
+
+def active_gate(
+    active: LinExprLike,
+    coords: Mapping[Any, Any],
+    fill_value: float = 1,
+) -> LinearExpression:
+    r"""
+    Pad a partial ``active`` gate to full coverage for piecewise gating.
+
+    Reindexes ``active`` to ``coords`` and fills missing/masked entries with
+    ``fill_value`` (``1`` = always active, ``0`` = always off), so a gate
+    defined over only a subset of :meth:`~linopy.Model.add_piecewise_formulation`'s
+    coordinate does not force the uncovered entries to zero. Equivalent to the
+    v1 idiom ``active.reindex(coords).fillna(fill_value)`` but correct under
+    legacy too (see the module docstring).
+
+    .. code-block:: python
+
+        gate = active_gate(status, {"component": components})
+        m.add_piecewise_formulation((power, xs), (fuel, ys), active=gate)
+
+    Parameters
+    ----------
+    active : Variable or LinearExpression
+        The (possibly partial) gate expression.
+    coords : mapping of dim to labels
+        Reindex target, passed straight to ``reindex``; unlisted dims
+        broadcast.
+    fill_value : float, default 1
+        Value for missing/masked entries (``1`` = on, ``0`` = off).
+
+    Returns
+    -------
+    LinearExpression
+        The padded gate, suitable to pass as ``active=``.
+
+    Warns
+    -----
+    EvolvingAPIWarning
+        Part of the evolving piecewise API; may be refined.
+    """
+    _warn_evolving_api(
+        "active_gate",
+        "piecewise: active_gate is a new API; its signature and the way it "
+        "resolves missing/masked entries may be refined in minor releases.  "
+        "It is primarily a legacy stopgap and may be removed once legacy "
+        "semantics are dropped.  This warning fires once per session; "
+        "silence with "
+        '`warnings.filterwarnings("ignore", category=linopy.EvolvingAPIWarning)`.',
+    )
+    gate = _to_linexpr(active).reindex(coords)
+    return gate.where(gate.has_terms, fill_value)

linopy/piecewise.py

@@ -64,7 +64,7 @@
 # most once per process.  Without dedup, a single model build emits the
 # verbose warning hundreds of times and drowns out other output.
 _EvolvingApiKey: TypeAlias = Literal[
-    "tangent_lines", "add_piecewise_formulation", "Slopes"
+    "tangent_lines", "add_piecewise_formulation", "Slopes", "active_gate"
 ]
 _emitted_evolving_warnings: set[_EvolvingApiKey] = set()
 
@@ -838,6 +838,35 @@ def tangent_lines(
 # ---------------------------------------------------------------------------
 
 
+def _validate_active_coverage(active: LinearExpression, reference: DataArray) -> None:
+    """
+    Reject an ``active`` gate that does not cover the formulation.
+
+    Entries where ``active`` is missing (a strict subset) or masked would
+    otherwise be gated as if ``active=0`` and forced to zero.  Such gates
+    must be padded to full coverage (e.g. via :func:`active_gate`) before
+    use.  Dimensions absent from ``active`` broadcast and are not checked.
+    """
+    skip = {BREAKPOINT_DIM, SEGMENT_DIM} | set(HELPER_DIMS)
+    indexers = {
+        d: reference.indexes[d]
+        for d in active.coord_dims
+        if d in reference.indexes and d not in skip
+    }
+    aligned = active.reindex(indexers) if indexers else active
+    term_dims = [d for d in aligned.vars.dims if d not in aligned.coord_dims]
+    dangling = ((aligned.vars < 0) & aligned.coeffs.notnull()).any(term_dims)
+    covered = aligned.has_terms | (aligned.const.notnull() & ~dangling)
+    if not bool(covered.all()):
+        raise ValueError(
+            "`active` is not defined over the full coordinate of the "
+            "piecewise formulation; it has missing or masked entries that "
+            "would be gated to zero. Pad it to full coverage first, e.g. "
+            "`active=linopy.active_gate(active, {dim: labels})` (missing "
+            "entries become always-active), or pass a fully-defined `active`."
+        )
+
+
 def _validate_breakpoint_shapes(bp_a: DataArray, bp_b: DataArray) -> bool:
     """
     Validate that two breakpoint arrays have compatible shapes.
@@ -1118,6 +1147,10 @@ def add_piecewise_formulation(
         ``active=0``, all auxiliary variables are forced to zero.
         Not supported with ``method="lp"``.
 
+        ``active`` must cover the formulation's full coordinate; a partial
+        gate (subset or masked) is rejected. Pad it with
+        :func:`~linopy.active_gate` to leave missing entries ungated.
+
         With all-equality tuples (the default), the output is then pinned
         to ``0``.  With a bounded tuple (``"<="`` / ``">="``), deactivation
         only pushes the signed bound to ``0`` (the output is ≤ 0 or ≥ 0
@@ -1286,6 +1319,8 @@ def add_piecewise_formulation(
             link_coords.append(f"_pwl_{i}")
 
     active_expr = _to_linexpr(active) if active is not None else None
+    if active_expr is not None:
+        _validate_active_coverage(active_expr, bp_list[0])
 
     if signed_idx is None:
         inputs = _PwlInputs(

test/test_piecewise_active_gate.py

@@ -0,0 +1,203 @@
+"""
+Tests for the partial-``active`` gate and the ``active_gate`` helper (#796).
+
+Kept in a dedicated module because ``active_gate`` is a temporary legacy
+stopgap (see ``linopy/_active_gate.py``): once the v1 arithmetic semantics
+(#717) land, the helper is expected to be deprecated and these tests removed
+or rewritten against the bare ``reindex().fillna()`` idiom.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any, Literal, TypeAlias
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from linopy import Model, active_gate, available_solvers, segments
+from linopy.solver_capabilities import (
+    SolverFeature,
+    get_available_solvers_with_feature,
+)
+
+Method: TypeAlias = Literal["sos2", "incremental", "lp", "auto"]
+GateBuilder: TypeAlias = Callable[[Model], Any]
+
+_any_solvers = [
+    s for s in ["highs", "gurobi", "glpk", "cplex"] if s in available_solvers
+]
+_sos2_solvers = get_available_solvers_with_feature(
+    SolverFeature.SOS_CONSTRAINTS, available_solvers
+)
+
+
+# ``active`` is meaningful only for the committable subset {a, c}; "b" stays
+# ungated. The partial-gate shapes below all leave "b" as the gap.
+_PWL_GENS = pd.Index(["a", "b", "c"], name="gen")
+_COMMITTABLE = pd.Index(["a", "c"], name="gen")
+
+
+def _subset_gate(m: Model) -> Any:
+    """``active`` indexed over a strict subset of the formulation's dim."""
+    return m.add_variables(binary=True, coords=[_COMMITTABLE], name="u")
+
+
+def _masked_gate(m: Model) -> Any:
+    """``active`` over the full dim but masked where it does not apply."""
+    mask = pd.Series([True, False, True], index=_PWL_GENS)
+    return m.add_variables(binary=True, coords=[_PWL_GENS], name="u", mask=mask)
+
+
+_PARTIAL_GATES = [
+    pytest.param(_subset_gate, id="strict-subset"),
+    pytest.param(_masked_gate, id="masked"),
+]
+
+
+def _full_gate(m: Model) -> Any:
+    return m.add_variables(binary=True, coords=[_PWL_GENS], name="u")
+
+
+def _scalar_gate(m: Model) -> Any:
+    return m.add_variables(binary=True, name="u")
+
+
+def _padded(make: GateBuilder, fill_value: float = 1) -> GateBuilder:
+    return lambda m: active_gate(make(m), {"gen": _PWL_GENS}, fill_value)
+
+
+# (builder, should_raise): raw partial gates are rejected; padded/full/scalar ok.
+_COVERAGE_CASES = [
+    pytest.param(_subset_gate, True, id="strict-subset-raises"),
+    pytest.param(_masked_gate, True, id="masked-raises"),
+    pytest.param(_padded(_subset_gate), False, id="padded-subset-ok"),
+    pytest.param(_padded(_masked_gate), False, id="padded-masked-ok"),
+    pytest.param(_padded(_subset_gate, 0), False, id="padded-off-ok"),
+    pytest.param(_full_gate, False, id="full-ok"),
+    pytest.param(_scalar_gate, False, id="scalar-ok"),
+]
+
+
+def _solve_partial_gate(
+    solver_name: str,
+    make_active: GateBuilder,
+    *,
+    method: Method,
+    disjunctive: bool = False,
+) -> None:
+    """Pad a partial gate, force the committable units off, demand "b" runs."""
+    m = Model()
+    x = m.add_variables(lower=0, upper=100, coords=[_PWL_GENS], name="x")
+    y = m.add_variables(lower=0, coords=[_PWL_GENS], name="y")
+    u = make_active(m)
+    gate = active_gate(u, {"gen": _PWL_GENS})
+    if disjunctive:
+        m.add_piecewise_formulation(
+            (x, segments([[0.0, 50.0], [50.0, 100.0]])),
+            (y, segments([[0.0, 10.0], [10.0, 50.0]])),
+            active=gate,
+        )
+    else:
+        m.add_piecewise_formulation(
+            (x, [0, 50, 100]), (y, [0, 10, 50]), active=gate, method=method
+        )
+    m.add_constraints(u <= 0, name="force_off")
+    m.add_constraints(x.sel(gen="b") >= 50, name="demand")
+    m.add_objective(y.sum(), sense="min")
+    status, _ = m.solve(solver_name=solver_name)
+    assert status == "ok"
+    np.testing.assert_allclose(float(x.solution.sel(gen="a")), 0, atol=1e-4)
+    np.testing.assert_allclose(float(x.solution.sel(gen="c")), 0, atol=1e-4)
+    np.testing.assert_allclose(float(x.solution.sel(gen="b")), 50, atol=1e-4)
+    np.testing.assert_allclose(float(y.solution.sel(gen="b")), 10, atol=1e-4)
+
+
+class TestActiveGateHelper:
+    """``active_gate`` pads a partial gate; gaps -> ``fill_value``."""
+
+    @pytest.mark.parametrize("fill_value", [1, 0])
+    @pytest.mark.parametrize(
+        "make_active",
+        [*_PARTIAL_GATES, pytest.param(lambda m: 2 * _subset_gate(m), id="linexpr")],
+    )
+    def test_fills_gap(self, make_active: GateBuilder, fill_value: float) -> None:
+        gate = active_gate(make_active(Model()), {"gen": _PWL_GENS}, fill_value)
+        assert gate.const.sel(gen="b").item() == fill_value
+        assert bool((gate.vars.sel(gen="b") < 0).all())
+        assert bool((gate.vars.sel(gen="a") >= 0).any())
+
+
+class TestPartialActiveValidation:
+    """``add_piecewise_formulation`` rejects an under-defined ``active`` (#796)."""
+
+    @pytest.mark.parametrize("make_active, should_raise", _COVERAGE_CASES)
+    def test_coverage(self, make_active: GateBuilder, should_raise: bool) -> None:
+        m = Model()
+        x = m.add_variables(lower=0, upper=100, coords=[_PWL_GENS], name="x")
+        y = m.add_variables(lower=0, coords=[_PWL_GENS], name="y")
+
+        def build() -> None:
+            m.add_piecewise_formulation(
+                (x, [0, 50, 100]),
+                (y, [0, 10, 50]),
+                active=make_active(m),
+                method="incremental",
+            )
+
+        if should_raise:
+            with pytest.raises(ValueError, match="active_gate"):
+                build()
+        else:
+            build()
+
+    def test_lower_dimensional_active_broadcasts(self) -> None:
+        """A gate missing an entire dim broadcasts and must not be rejected."""
+        ts = pd.Index([0, 1], name="t")
+        m = Model()
+        x = m.add_variables(lower=0, upper=100, coords=[_PWL_GENS, ts], name="x")
+        y = m.add_variables(lower=0, coords=[_PWL_GENS, ts], name="y")
+        u = m.add_variables(binary=True, coords=[_PWL_GENS], name="u")
+        m.add_piecewise_formulation(
+            (x, [0, 50, 100]), (y, [0, 10, 50]), active=u, method="incremental"
+        )
+
+
+@pytest.mark.skipif(len(_any_solvers) == 0, reason="No solver available")
+class TestSolverPartialActiveGate:
+    """End-to-end: a padded partial gate leaves ungated units free (#796)."""
+
+    @pytest.fixture(params=_any_solvers)
+    def solver_name(self, request: pytest.FixtureRequest) -> str:
+        return request.param
+
+    @pytest.mark.parametrize("make_active", _PARTIAL_GATES)
+    def test_incremental(self, solver_name: str, make_active: GateBuilder) -> None:
+        _solve_partial_gate(solver_name, make_active, method="incremental")
+
+
+@pytest.mark.skipif(len(_sos2_solvers) == 0, reason="No SOS2-capable solver")
+class TestSolverPartialActiveGateSOS2:
+    @pytest.fixture(params=_sos2_solvers)
+    def solver_name(self, request: pytest.FixtureRequest) -> str:
+        return request.param
+
+    @pytest.mark.parametrize("make_active", _PARTIAL_GATES)
+    @pytest.mark.parametrize(
+        "method, disjunctive",
+        [
+            pytest.param("sos2", False, id="sos2"),
+            pytest.param("auto", True, id="disjunctive"),
+        ],
+    )
+    def test_solves(
+        self,
+        solver_name: str,
+        make_active: GateBuilder,
+        method: Method,
+        disjunctive: bool,
+    ) -> None:
+        _solve_partial_gate(
+            solver_name, make_active, method=method, disjunctive=disjunctive
+        )