flexmeasures.data.models.planning.devices

Typed device tracking for schedulers.

Multi-device flex-models describe several kinds of entries (schedulable devices, stock-only entries carrying SoC parameters for a shared stock, group entries carrying constraints on the aggregate power of a set of member devices, and — in the future — converter ports). Historically, each scheduling feature re-derived an entry’s kind from the raw dicts and kept parallel lists aligned by integer position, which is where several alignment bugs crept in.

This module classifies every entry exactly once, right after flex-config deserialization, into a DeviceInventory: the single source of truth for

which entries are schedulable devices, and their canonical solver indices,
which entries only carry SoC parameters for a shared stock (stock-only entries),
which entries are group entries, and which devices belong to each group,
the inflexible devices from the flex-context, in canonical solver order, and
the stock groups (devices sharing a state-of-charge sensor).

The raw (deserialized) flex-model dicts are kept as-is on each FlexDevice, so downstream code and new flex-model fields need no dataclass changes.

Module Attributes

Functions

flexmeasures.data.models.planning.devices.group_key_label(group_key: tuple[str, int]) → str: Return a human-readable label for a group key, for use in error messages.

flexmeasures.data.models.planning.devices.resolve_group_key(flex_model: dict | None) → tuple[str, int] | None: Return a normalized (“sensor”, id) or (“asset”, id) key for the group a flex-model entry’s “group” field references, or None if it has none.

Classes

class flexmeasures.data.models.planning.devices.DeviceInventory(entries: list[~flexmeasures.data.models.planning.devices.FlexDevice] = <factory>, devices: list[~flexmeasures.data.models.planning.devices.FlexDevice] = <factory>, inflexible_devices: list[~flexmeasures.data.models.planning.devices.FlexDevice] = <factory>, stock_entries: dict[int, dict] = <factory>, group_entries: dict[tuple[str, int], dict] = <factory>, referenced_group_keys: set[tuple[str, int]] = <factory>, is_single_sensor_mode: bool = False)

All devices of a scheduling problem, classified once, in canonical solver order.

The canonical device enumeration is:

flexible devices (flex-model entries with role DEVICE), in flex-model order,
top-level (electricity) inflexible-device-sensors from the flex-context, in order,
each commodity context’s own inflexible-device-sensors, in the order the commodity contexts are given.

This is the one enumeration both _prepare() and the result mapping rely on, so they cannot drift apart.

__init__(entries: list[~flexmeasures.data.models.planning.devices.FlexDevice] = <factory>, devices: list[~flexmeasures.data.models.planning.devices.FlexDevice] = <factory>, inflexible_devices: list[~flexmeasures.data.models.planning.devices.FlexDevice] = <factory>, stock_entries: dict[int, dict] = <factory>, group_entries: dict[tuple[str, int], dict] = <factory>, referenced_group_keys: set[tuple[str, int]] = <factory>, is_single_sensor_mode: bool = False) → None

_coupling_reference_port(members: list[tuple[int, float]]) → FlexDevice

Return a coupling group’s reference port: the port with |coefficient| == 1.

The reference port is the driving variable of the group; it carries the group’s coupling-min and the power-capacity that bounds the group’s marginal level.

Raises:: ValueError – When no member has a unit coefficient.

property assets: list[GenericAsset | None]: The flexible devices’ assets, by device index.

by_index(d: int) → FlexDevice: Return the device with canonical solver index d (flexible or inflexible).

by_sensor_id(sensor_id: int) → list[FlexDevice]: Return the flexible devices whose power sensor has the given id.

property commodity_to_devices: dict[str, list[int]]: Map each commodity to its device indices, in canonical solver order.

property coupling_bases: dict[str, list[tuple[int, float]]]

Map each unit-committed coupling group to its ports’ (device index, signed base) pairs (in MW).

Restricted to the groups in coupling_uc; suitable for passing to device_scheduler(coupling_bases=...) alongside coupling_groups.

property coupling_groups: dict[str, list[tuple[int, float]]]

Map each coupling-group name to its ports’ (device index, signed coefficient) pairs.

Devices sharing a coupling name are the commodity ports of one converter (e.g. a CHP unit’s gas input, heat output and electricity output). The optimization model introduces a decision variable alpha per group per time step, and constrains every port by P[d] == coeff_d * alpha. The coefficient signs follow the internal convention (see _resolve_coupling_coefficient()): positive for inputs, negative for outputs. The result is suitable for passing to device_scheduler(coupling_groups=...); it is empty when no device defines a coupling field.

property coupling_uc: dict[str, tuple[float, float]]

Map each unit-committed coupling group to its (min, max) marginal-level bounds (in MW).

A coupling group is unit-committed when its reference port declares a coupling-min, or any of its ports declares a non-zero coupling-base. The minimum comes from the reference port’s coupling-min (0 when only a base is given); the maximum from the reference port’s power-capacity. Empty for purely proportional coupling (leaving the problem an LP).

Raises:: ValueError – When a unit-committed group lacks a resolvable power-capacity on its reference port.

property device_flex_models: list[dict]: The flexible devices’ flex-model entries, by device index.

devices: list[FlexDevice]: The schedulable devices; devices[d].index == d.

entries: list[FlexDevice]: All flex-model entries, in their original order (including stock-only entries).

classmethod from_flex_config(flex_model: list[dict] | dict, flex_context: dict | None = None, sensor: Sensor | None = None) → DeviceInventory

Classify a deserialized flex-model (and flex-context) into an inventory.

Parameters:

flex_model – The deserialized flex-model: a dict (single-sensor mode, in which case sensor is the device’s power sensor) or a list of entry dicts (multi-device mode).
flex_context – The deserialized flex-context, used for the inflexible devices (top-level and per commodity context).
sensor – The scheduler’s target sensor (single-sensor mode only).

group_entries: dict[tuple[str, int], dict]: Group entries (raw flex-model dicts) per group key: (“sensor”, id) or (“asset”, id).

property group_to_devices: dict[tuple[str, int], list[int]]

Map each group key to the indices of the (leaf) member devices of that group.

Membership is resolved transitively: a group entry may itself belong to another group (via its own “group” field), in which case its member devices count as members of the outer group, too.

Raises:: ValueError – When group entries reference each other cyclically.

inflexible_devices: list[FlexDevice]: The inflexible devices from the flex-context, with indices following the devices.

property inflexible_sensors: list[Sensor]

The inflexible devices’ power sensors, in canonical solver order.

Inflexible devices are constructed from the flex-context’s sensors, so their power sensor is always set.

property num_flexible: int: The number of schedulable (flexible) devices.

property num_scheduled: int: The number of devices in the optimization problem (flexible + inflexible).

property power_sensors: list[Sensor | None]: The flexible devices’ power sensors, by device index.

referenced_group_keys: set[tuple[str, int]]: The group keys referenced by entries’ “group” fields (used to detect dangling references).

stock_constraint_device(stock_key: int) → int | None

Return the device that applies the given stock’s SoC constraints.

Hard SoC constraints (and the device-indexed remains of softened ones) land on the first device of the stock group, whose stock represents the group’s stock in the solver. Use stock_groups to look up all member devices instead.

stock_entries: dict[int, dict]: SoC parameters per stock key. Keys are shared with stock_groups.

property stock_groups: dict[int, list[int]]

Map each stock key to the indices of the devices drawing from that stock.

Devices sharing a state-of-charge sensor are grouped together; devices without one form singleton groups under their synthetic (negative) stock key.

stock_params(stock_key: int) → dict | None: Return the flex-model entry holding the SoC parameters of the given stock.

class flexmeasures.data.models.planning.devices.DeviceRole(*values)

The role a flex-model (or flex-context) entry plays in the scheduling problem.

Converter ports (the commodity ports of a multi-commodity converter, such as a CHP unit) are DEVICE entries carrying a coupling field; see DeviceInventory.coupling_groups. GROUP entries constrain the aggregate power of a set of member devices.

DEVICE = 'device': A schedulable flexible device (usually with a power sensor).

GROUP = 'group': An entry constraining the aggregate power of a set of member devices; not itself scheduled.

INFLEXIBLE = 'inflexible': An inflexible device from the flex-context (scheduled with a fixed profile).

STOCK_ONLY = 'stock-only': An entry carrying SoC parameters for a shared stock; not itself scheduled.

class flexmeasures.data.models.planning.devices.FlexDevice(role: DeviceRole, index: int | None, flex_model: dict | None, power_sensor: Sensor | None, asset: GenericAsset | None, commodity: str = 'electricity', stock_key: int | None = None, coupling: str | None = None, coupling_coefficient: float = 1.0, coupling_base: float = 0.0, coupling_min: float | None = None, coupling_max: float | None = None)

A single classified flex-model (or flex-context) entry.

Note that flex_model references the original deserialized dict (it is not a copy); code that mutates per-device parameters should work on copies.

__init__(role: DeviceRole, index: int | None, flex_model: dict | None, power_sensor: Sensor | None, asset: GenericAsset | None, commodity: str = 'electricity', stock_key: int | None = None, coupling: str | None = None, coupling_coefficient: float = 1.0, coupling_base: float = 0.0, coupling_min: float | None = None, coupling_max: float | None = None) → None

asset: GenericAsset | None: The device’s asset, resolved from the power sensor or the entry’s “asset” key.

coupling: str | None = None: Name of the coupling group this device belongs to (converter ports of one converter share a coupling name). None for uncoupled devices.

coupling_base: float = 0.0: Signed per-port no-load base (in MW), gated by the group’s on/off binary when the group is unit-committed. Follows the same sign convention as the coefficient; 0.0 when no coupling-base is given.

coupling_coefficient: float = 1.0: Signed internal coupling coefficient: positive for input (consuming) ports, negative for output (producing) ports. Meaningless (1.0) for uncoupled devices.

coupling_max: float | None = None: The reference port’s power capacity (in MW), used as the group’s maximum marginal level. None when not resolvable.

coupling_min: float | None = None: Group minimum marginal level (in MW), declared on the reference port (|coefficient| == 1). None when not set.

flex_model: dict | None: The deserialized flex-model entry (with underscore keys); None for inflexible devices.

property group_key: tuple[str, int] | None: The key of the group this entry belongs to (via its “group” field), if any.

index: int | None: Canonical solver device index; None for stock-only entries.

power_sensor: Sensor | None: The device’s power sensor, resolved from the entry’s top-level “sensor” key, else from a nested consumption/production output reference. None for entries that reference no power sensor at all (e.g. asset-only entries).

stock_key: int | None = None: Key of the stock this device draws from: the id of its state-of-charge sensor, or a unique negative synthetic key for devices without one. None for inflexible devices.