fluxopt.constraints.status

Status (on/off) constraint helpers.

Module-level functions that add binary status tracking constraints to a linopy Model. Used by FlowSystem to build status features.

Functions:

Name	Description
compute_previous_duration	Compute consecutive duration of target_state at end of previous_status.
add_duration_tracking	Add consecutive duration tracking for a binary state variable.
add_switch_transitions	Add startup/shutdown transition constraints.

compute_previous_duration

compute_previous_duration(
    previous_status: DataArray, target_state: int, dt: DataArray | float
) -> float

Compute consecutive duration of target_state at end of previous_status.

Walks backward through previous_status counting timesteps that match the target state, then multiplies by timestep duration.

Parameters:

Name	Type	Description	Default
previous_status	DataArray	Previous status values (time dimension).	required
target_state	int	1 for active (uptime), 0 for inactive (downtime).	required
dt	DataArray | float	Duration per timestep (scalar or DataArray).	required

Returns:

Type	Description
float	Total duration in target state at end of previous period.

Source code in src/fluxopt/constraints/status.py

def compute_previous_duration(
    previous_status: xr.DataArray,
    target_state: int,
    dt: xr.DataArray | float,
) -> float:
    """Compute consecutive duration of target_state at end of previous_status.

    Walks backward through previous_status counting timesteps that match
    the target state, then multiplies by timestep duration.

    Args:
        previous_status: Previous status values (time dimension).
        target_state: 1 for active (uptime), 0 for inactive (downtime).
        dt: Duration per timestep (scalar or DataArray).

    Returns:
        Total duration in target state at end of previous period.
    """
    values = previous_status.values
    count = 0
    for v in reversed(values):
        if (target_state == 1 and v > 0) or (target_state == 0 and v == 0):
            count += 1
        else:
            break

    if isinstance(dt, xr.DataArray):
        return float(dt.values[-count:].sum()) if count > 0 else 0.0
    return dt * count

add_duration_tracking

add_duration_tracking(
    m: Model,
    state: Variable | LinearExpression,
    dt: DataArray,
    *,
    name: str,
    element_dim: str = 'flow',
    dim: str = 'time',
    minimum: DataArray | None = None,
    maximum: DataArray | None = None,
    previous: DataArray | None = None,
) -> Variable

Add consecutive duration tracking for a binary state variable.

Uses Big-M formulation to track how long each element has been continuously in the given state.

Parameters:

Name	Type	Description	Default
m	Model	Linopy model to add constraints to.	required
state	Variable | LinearExpression	Binary state variable with (element_dim, time) dims.	required
dt	DataArray	Timestep durations (time,).	required
name	str	Base name for created variables and constraints.	required
element_dim	str	Element dimension name in state.	'flow'
dim	str	Temporal dimension name.	'time'
minimum	DataArray | None	Minimum duration per element. NaN = no constraint.	None
maximum	DataArray | None	Maximum duration per element. NaN = no constraint.	None
previous	DataArray | None	Previous duration per element. NaN = no previous.	None

Returns:

Type	Description
Variable	Duration variable with same dims as state.

Source code in src/fluxopt/constraints/status.py

def add_duration_tracking(
    m: Model,
    state: Variable | LinearExpression,
    dt: xr.DataArray,
    *,
    name: str,
    element_dim: str = 'flow',
    dim: str = 'time',
    minimum: xr.DataArray | None = None,
    maximum: xr.DataArray | None = None,
    previous: xr.DataArray | None = None,
) -> Variable:
    """Add consecutive duration tracking for a binary state variable.

    Uses Big-M formulation to track how long each element has been
    continuously in the given state.

    Args:
        m: Linopy model to add constraints to.
        state: Binary state variable with (element_dim, time) dims.
        dt: Timestep durations (time,).
        name: Base name for created variables and constraints.
        element_dim: Element dimension name in state.
        dim: Temporal dimension name.
        minimum: Minimum duration per element. NaN = no constraint.
        maximum: Maximum duration per element. NaN = no constraint.
        previous: Previous duration per element. NaN = no previous.

    Returns:
        Duration variable with same dims as state.
    """
    element_ids: xr.DataArray = state.coords[element_dim]

    # Big-M per element: total horizon + any previous carryover
    mega = dt.sum(dim)
    if previous is not None:
        mega = mega + previous.fillna(0)

    # Variable upper bound: use maximum where provided, else mega
    upper: xr.DataArray = maximum.where(maximum.notnull(), mega) if maximum is not None else mega

    coords = [state.coords[element_dim], state.coords[dim]]
    duration = m.add_variables(lower=0, upper=upper, coords=coords, name=name)

    # duration[e,t] <= state[e,t] * M[e]
    m.add_constraints(duration <= state * mega, name=f'{name}|ub')

    # Forward: duration[e,t+1] <= duration[e,t] + dt[t]
    m.add_constraints(
        duration.isel({dim: slice(1, None)}) <= duration.isel({dim: slice(None, -1)}) + dt.isel({dim: slice(None, -1)}),
        name=f'{name}|fwd',
    )

    # Backward: duration[e,t+1] >= duration[e,t] + dt[t] + (state[e,t+1] - 1) * M[e]
    m.add_constraints(
        duration.isel({dim: slice(1, None)})
        >= duration.isel({dim: slice(None, -1)})
        + dt.isel({dim: slice(None, -1)})
        + (state.isel({dim: slice(1, None)}) - 1) * mega,
        name=f'{name}|bwd',
    )

    # Initial constraints for elements with previous duration
    if previous is not None:
        has_prev = previous.notnull()
        if has_prev.any():
            prev_ids = list(element_ids.values[has_prev.values])
            _add_initial_constraints(
                m,
                state,
                duration,
                dt,
                previous=previous.sel({element_dim: prev_ids}),
                minimum=minimum.sel({element_dim: prev_ids}) if minimum is not None else None,
                name=name,
                dim=dim,
                element_dim=element_dim,
            )

    # Minimum duration: duration[t] >= min * (state[t] - state[t+1])
    if minimum is not None:
        _add_minimum_constraints(m, state, duration, minimum, name, dim, element_dim)

    return duration

add_switch_transitions

add_switch_transitions(
    m: Model,
    status: Variable,
    startup: Variable,
    shutdown: Variable,
    *,
    name: str,
    element_dim: str = 'flow',
    dim: str = 'time',
    previous_state: DataArray | None = None,
) -> None

Add startup/shutdown transition constraints.

Links status changes to startup and shutdown indicator variables: startup[t] - shutdown[t] == status[t] - status[t-1].

Parameters:

Name	Type	Description	Default
m	Model	Linopy model.	required
status	Variable	Binary on/off variable.	required
startup	Variable	Binary startup indicator variable.	required
shutdown	Variable	Binary shutdown indicator variable.	required
name	str	Base name for constraints.	required
element_dim	str	Element dimension name in status.	'flow'
dim	str	Temporal dimension name.	'time'
previous_state	DataArray | None	Previous on/off per element (pre-filtered, no NaN).	None

Source code in src/fluxopt/constraints/status.py

def add_switch_transitions(
    m: Model,
    status: Variable,
    startup: Variable,
    shutdown: Variable,
    *,
    name: str,
    element_dim: str = 'flow',
    dim: str = 'time',
    previous_state: xr.DataArray | None = None,
) -> None:
    """Add startup/shutdown transition constraints.

    Links status changes to startup and shutdown indicator variables:
    ``startup[t] - shutdown[t] == status[t] - status[t-1]``.

    Args:
        m: Linopy model.
        status: Binary on/off variable.
        startup: Binary startup indicator variable.
        shutdown: Binary shutdown indicator variable.
        name: Base name for constraints.
        element_dim: Element dimension name in status.
        dim: Temporal dimension name.
        previous_state: Previous on/off per element (pre-filtered, no NaN).
    """
    # Transition for t > 0
    m.add_constraints(
        startup.isel({dim: slice(1, None)}) - shutdown.isel({dim: slice(1, None)})
        == status.isel({dim: slice(1, None)}) - status.isel({dim: slice(None, -1)}),
        name=f'{name}|transition',
    )

    # Initial transition from previous state
    if previous_state is not None:
        ids = list(previous_state.coords[element_dim].values)
        m.add_constraints(
            startup.sel({element_dim: ids}).isel({dim: 0}) - shutdown.sel({element_dim: ids}).isel({dim: 0})
            == status.sel({element_dim: ids}).isel({dim: 0}) - previous_state,
            name=f'{name}|init',
        )