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

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 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_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.

Extension point (not yet implemented): CONVERTER_PORT (a commodity port of a multi-commodity converter).

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.

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) → None

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

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.