fluxopt.types

Type Aliases:

Name	Description
Variate	Any input that varies over a subset of the model's variate dims (time,

Classes:

Name	Description
ProfileRef	Reference to a time-series stored outside the model definition.

Functions:

Name	Description
variate_out_of_range	The first value outside [low, high], or None if all of them are in.
as_dataarray	Convert a Variate to a DataArray aligned to given coordinates.
normalize_timesteps	Normalize user-provided timesteps to an internal time index.
compute_dt	Compute dt (hours) for each timestep as a DataArray.

Variate

Variate = (
    float
    | int
    | list[float]
    | ndarray
    | Series
    | DataFrame
    | DataArray
    | ProfileRef
)

Any input that varies over a subset of the model's variate dims (time, optionally period, eventually scenario).

• Scalar: broadcast to all variate dims.
• 1-D (list/ndarray): matched to a coord by length (must be unambiguous).
• 1-D (pd.Series): index name selects the dim if set; else matched by length.
• 2-D (pd.DataFrame): index.name and columns.name must match target dims.
• n-D (xr.DataArray): dims must be a subset of the target; coords must match exactly.

Per-field reach (which dims a particular field can vary over) is documented on the field itself; as_dataarray enforces that user input only uses dims the caller declared in coords.

ProfileRef pydantic-model

Bases: BaseModel

Reference to a time-series stored outside the model definition.

A serializable stand-in for an inline Variate array: the profile lives in a data file / dataset and is named here, so structural definitions round-trip to YAML/JSON without inlining 8760-point series. Resolve it to a :class:xr.DataArray with :meth:resolve before building the model.

Show JSON schema:

{
  "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
  "properties": {
    "dataset": {
      "title": "Dataset",
      "type": "string"
    },
    "variable": {
      "title": "Variable",
      "type": "string"
    }
  },
  "required": [
    "dataset",
    "variable"
  ],
  "title": "ProfileRef",
  "type": "object"
}

Config:

• frozen: True

Fields:

• dataset (str)
• variable (str)

dataset pydantic-field

dataset: str

Id of the dataset holding the profile (a key into profiles).

variable pydantic-field

variable: str

Variable / column name within dataset.

resolve

resolve(profiles: Mapping[str, Dataset | Mapping[str, DataArray]]) -> DataArray

Look up the referenced series in profiles.

Parameters:

Name	Type	Description	Default
profiles	Mapping[str, Dataset | Mapping[str, DataArray]]	Mapping from dataset id to a dataset (or mapping) that contains variable.	required

Raises:

Type	Description
KeyError	If dataset or variable is absent from profiles.

Source code in src/fluxopt/types.py

def resolve(self, profiles: Mapping[str, xr.Dataset | Mapping[str, xr.DataArray]]) -> xr.DataArray:
    """Look up the referenced series in *profiles*.

    Args:
        profiles: Mapping from dataset id to a dataset (or mapping) that
            contains ``variable``.

    Raises:
        KeyError: If *dataset* or *variable* is absent from *profiles*.
    """
    if self.dataset not in profiles:
        raise KeyError(f'ProfileRef dataset {self.dataset!r} not in profiles {sorted(profiles)}')
    ds = profiles[self.dataset]
    try:
        return xr.DataArray(ds[self.variable])
    except KeyError as exc:
        raise KeyError(f'ProfileRef variable {self.variable!r} not in dataset {self.dataset!r}') from exc

variate_out_of_range

variate_out_of_range(
    value: Variate, *, low: float, high: float, low_open: bool = False
) -> float | None

The first value outside [low, high], or None if all of them are in.

A :data:Variate is a scalar, a series, or a :class:ProfileRef that names numbers living somewhere else. The first two can be checked where they are written, which is what this is for; a ProfileRef cannot, because its values arrive when profiles are resolved — so it reads as in range here and is checked again once it is real (docs/design/validation-layers.md).

Parameters:

Name	Type	Description	Default
value	Variate	The declared value.	required
low	float	Lower bound.	required
high	float	Upper bound.	required
low_open	bool	Whether low itself is excluded.	False

Returns:

Type	Description
float | None	An offending value, or None.

Source code in src/fluxopt/types.py

def variate_out_of_range(
    value: Variate,
    *,
    low: float,
    high: float,
    low_open: bool = False,
) -> float | None:
    """The first value outside ``[low, high]``, or None if all of them are in.

    A :data:`Variate` is a scalar, a series, or a :class:`ProfileRef` that
    names numbers living somewhere else. The first two can be checked where
    they are written, which is what this is for; a ``ProfileRef`` cannot,
    because its values arrive when profiles are resolved — so it reads as in
    range here and is checked again once it is real
    (``docs/design/validation-layers.md``).

    Args:
        value: The declared value.
        low: Lower bound.
        high: Upper bound.
        low_open: Whether *low* itself is excluded.

    Returns:
        An offending value, or None.
    """
    if isinstance(value, ProfileRef):
        return None
    values = np.atleast_1d(np.asarray(value, dtype=float))
    below = values <= low if low_open else values < low
    bad = values[below | (values > high)]
    return float(bad[0]) if bad.size else None

as_dataarray

as_dataarray(
    value: Variate,
    coords: Mapping[str, Any],
    *,
    name: str = 'value',
    broadcast: bool = True,
) -> DataArray

Convert a Variate to a DataArray aligned to given coordinates.

Pipeline: convert → validate dims → validate coord values → broadcast.

See :data:Variate for accepted inputs. Pandas inputs (Series, DataFrame) follow the same convention as linopy.as_dataarray: the axis name attribute selects the corresponding target dim. For ndarray/list, the dim is selected by length (must be unambiguous). For DataArray, dims must be a subset of coords and coord values must match exactly — alignment errors are surfaced loudly, not silently masked.

Parameters:

Name	Type	Description	Default
value	Variate	Scalar, list, ndarray, Series, DataFrame, or DataArray.	required
coords	Mapping[str, Any]	Target coordinates, e.g. {"time": idx, "period": pidx}. Used both as the reach declaration and as alignment targets.	required
name	str	Name for the resulting DataArray.	'value'
broadcast	bool	Expand result to span all dimensions in coords.	True

Source code in src/fluxopt/types.py

def as_dataarray(
    value: Variate,
    coords: Mapping[str, Any],
    *,
    name: str = 'value',
    broadcast: bool = True,
) -> xr.DataArray:
    """Convert a Variate to a DataArray aligned to given coordinates.

    Pipeline: ``convert → validate dims → validate coord values → broadcast``.

    See :data:`Variate` for accepted inputs. Pandas inputs (``Series``,
    ``DataFrame``) follow the same convention as ``linopy.as_dataarray``: the
    axis ``name`` attribute selects the corresponding target dim. For
    ``ndarray``/``list``, the dim is selected by length (must be unambiguous).
    For ``DataArray``, dims must be a subset of *coords* and coord values must
    match exactly — alignment errors are surfaced loudly, not silently masked.

    Args:
        value: Scalar, list, ndarray, Series, DataFrame, or DataArray.
        coords: Target coordinates, e.g. ``{"time": idx, "period": pidx}``.
            Used both as the reach declaration and as alignment targets.
        name: Name for the resulting DataArray.
        broadcast: Expand result to span all dimensions in *coords*.
    """
    if isinstance(value, ProfileRef):
        raise ValueError(
            f'Unresolved ProfileRef {value!r}: resolve it to an array via ProfileRef.resolve(profiles) '
            f'before building the model.'
        )

    coord_idx = {k: v if isinstance(v, pd.Index) else pd.Index(v) for k, v in coords.items()}

    # --- scalar: 0-dim unless broadcast ---
    if isinstance(value, (int, float)):
        if not broadcast:
            return xr.DataArray(float(value), name=name)
        shape = tuple(len(v) for v in coord_idx.values())
        return xr.DataArray(
            np.full(shape, float(value)),
            dims=list(coord_idx),
            coords=coord_idx,
            name=name,
        )

    # --- 1) Convert to DataArray ---
    da: xr.DataArray
    if isinstance(value, xr.DataArray):
        da = value
    elif isinstance(value, (pd.Series, pd.DataFrame)):
        # Mirror linopy: pandas axes already carry coords; use axis.name as dim.
        # Fall back to length-matching only when no axis is named.
        named = [a.name for a in value.axes if a.name is not None]
        if len(named) == value.ndim:
            da = xr.DataArray(value)
        elif value.ndim == 1 and not named:
            return _from_unnamed_1d(np.asarray(value.values, dtype=float), coord_idx, name, broadcast)
        else:
            raise ValueError(
                f'{type(value).__name__} requires axis.name set on every axis '
                f'(got {[a.name for a in value.axes]!r}). '
                f"Set e.g. df.index.name='time', df.columns.name='period'."
            )
    elif isinstance(value, np.ndarray):
        if value.ndim != 1:
            raise ValueError(
                f'np.ndarray must be 1-D (got ndim={value.ndim}); pass an xr.DataArray '
                f'or pd.DataFrame with named axes for higher-dim inputs.'
            )
        return _from_unnamed_1d(value, coord_idx, name, broadcast)
    elif isinstance(value, list):
        return _from_unnamed_1d(np.asarray(value, dtype=float), coord_idx, name, broadcast)
    else:
        raise TypeError(f'Unsupported Variate type: {type(value)}')

    # --- 2) Validate dims are a subset of the target ---
    foreign = [str(d) for d in da.dims if d not in coord_idx]
    if foreign:
        raise ValueError(
            f'{type(value).__name__} has dims {foreign} not in target coords {list(coord_idx)}. '
            f'Rename before calling as_dataarray().'
        )

    # --- 3) Validate coord values match exactly (close the alignment gap) ---
    for d in da.dims:
        dim_name = str(d)
        if d in da.coords and not pd.Index(da.coords[d].values).equals(coord_idx[dim_name]):
            raise ValueError(
                f'Coord mismatch on dim {dim_name!r}: input coord does not equal target. '
                f"Use the same index as the model's {dim_name}."
            )

    da = da.rename(name)
    if broadcast:
        for dim, idx in coord_idx.items():
            if dim not in da.dims:
                da = da.expand_dims({dim: idx})
        da = da.transpose(*coord_idx)
    return da

normalize_timesteps

normalize_timesteps(timesteps: Timesteps) -> TimeIndex

Normalize user-provided timesteps to an internal time index.

Parameters:

Name	Type	Description	Default
timesteps	Timesteps	Datetime objects, integers, or a DatetimeIndex.	required

Returns:

Type	Description
TimeIndex	A datetime index for datetime inputs, or an integer index for integer inputs.

Raises:

Type	Description
ValueError	If timesteps are not strictly monotonically increasing.

Source code in src/fluxopt/types.py

def normalize_timesteps(timesteps: Timesteps) -> TimeIndex:
    """Normalize user-provided timesteps to an internal time index.

    Args:
        timesteps: Datetime objects, integers, or a DatetimeIndex.

    Returns:
        A datetime index for datetime inputs, or an integer index for integer inputs.

    Raises:
        ValueError: If timesteps are not strictly monotonically increasing.
    """
    if len(timesteps) == 0:
        raise ValueError('Timesteps must not be empty')

    if isinstance(timesteps, pd.DatetimeIndex):
        idx: TimeIndex = timesteps
    elif isinstance(timesteps, pd.Index):
        if isinstance(timesteps, pd.RangeIndex) or pd.api.types.is_integer_dtype(timesteps.dtype):
            idx = timesteps
        elif pd.api.types.is_datetime64_any_dtype(timesteps.dtype):
            idx = pd.DatetimeIndex(timesteps)
        else:
            raise TypeError(f'Unsupported pd.Index dtype: {timesteps.dtype}. Use datetime or integer index.')
    elif not isinstance(timesteps, list):
        raise TypeError(f'Unsupported Timesteps type: {type(timesteps)}')
    elif isinstance(timesteps[0], datetime):
        idx = pd.DatetimeIndex(timesteps)
    elif type(timesteps[0]) is int:
        idx = pd.Index(timesteps)
        if not pd.api.types.is_integer_dtype(idx.dtype):
            raise TypeError('Integer timesteps contain non-integer values')
    else:
        raise TypeError(f'Unsupported timestep element type: {type(timesteps[0])}. Use datetime or int.')

    if len(idx) > 1 and not idx.is_monotonic_increasing:
        raise ValueError('Timesteps must be strictly monotonically increasing')
    if not idx.is_unique:
        raise ValueError('Timesteps contain duplicates')
    return idx

compute_dt

compute_dt(timesteps: TimeIndex, dt: float | list[float] | None) -> DataArray

Compute dt (hours) for each timestep as a DataArray.

When dt is None, auto-derives from timesteps: - Datetime: consecutive differences in hours; first = second (forward-looking). - Integer: 1.0 for all. - Single timestep: 1.0.

Parameters:

Name	Type	Description	Default
timesteps	TimeIndex	Time index.	required
dt	float | list[float] | None	Override timestep duration. Validated against timesteps length.	required

Source code in src/fluxopt/types.py

def compute_dt(timesteps: TimeIndex, dt: float | list[float] | None) -> xr.DataArray:
    """Compute dt (hours) for each timestep as a DataArray.

    When dt is None, auto-derives from timesteps:
    - Datetime: consecutive differences in hours; first = second (forward-looking).
    - Integer: 1.0 for all.
    - Single timestep: 1.0.

    Args:
        timesteps: Time index.
        dt: Override timestep duration. Validated against timesteps length.
    """
    n = len(timesteps)

    if dt is not None:
        if isinstance(dt, (int, float)):
            values = np.full(n, float(dt))
        elif isinstance(dt, list):
            if len(dt) != n:
                raise ValueError(f'dt length {len(dt)} does not match timesteps length {n}')
            values = np.array(dt, dtype=float)
        else:
            raise TypeError(f'Unsupported dt type: {type(dt)}')
        return xr.DataArray(values, dims=['time'], coords={'time': timesteps}, name='dt')

    # Auto-derive
    if n <= 1:
        return xr.DataArray(np.ones(n), dims=['time'], coords={'time': timesteps}, name='dt')

    if not isinstance(timesteps, pd.DatetimeIndex):
        # Integer timesteps: default to 1.0
        return xr.DataArray(np.ones(n), dims=['time'], coords={'time': timesteps}, name='dt')

    # Datetime: derive from diff in hours
    diffs = np.diff(timesteps.values) / np.timedelta64(1, 'h')
    dt_values = np.empty(n)
    dt_values[0] = diffs[0]
    dt_values[1:] = diffs
    return xr.DataArray(dt_values, dims=['time'], coords={'time': timesteps}, name='dt')