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.

flexmeasures.data.models.planning.devices.resolve_group_reference(group: Any | None) → tuple[str, int] | None

Return a normalized (“sensor”, id) or (“asset”, id) key for a “group” reference, or None.

The reference may be a {"sensor": ...}/{"asset": ...} dict (as produced by GroupReferenceSchema, used both by flex-model entries’ group field and by inflexible-device flex-context entries), or – for backwards compatibility – a raw sensor id/object.

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 devices from the flex-context, in order,
each commodity context’s own inflexible devices, in the order the commodity contexts are given.

Within each context, inflexible devices are enumerated per field in INFLEXIBLE_DEVICE_FIELDS order (the deprecated inflexible-device-sensors first, then inflexible-consumption, then inflexible-production), each field’s entries in the order they 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

_register_inflexible_devices(context: dict, commodity: str, index: int) → int

Register one (commodity) context’s inflexible devices, in canonical field order.

Entries of the newer sign-explicit fields are either plain Sensors or SensorReference objects (carrying source filters); the underlying sensor becomes the device’s power sensor either way.

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.

Members are the flexible devices and the inflexible devices (from the flex-context) that reference the group. Including the latter is what lets a group’s intermediate power constraint also account for inflexible (measured) load sitting behind the same node.

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.

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, consumption_is_positive: bool | None = None, sensor_reference: Any | None = None, group_reference: Any | None = None) → None

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

consumption_is_positive: bool | None = None: For inflexible devices: whether positive sensor values mean consumption, as determined by the flex-context field the device was listed under (inflexible-consumption → True, inflexible-production → False). None means the sign convention is read from the sensor’s consumption_is_positive attribute (deprecated inflexible-device-sensors).

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.

Flexible/stock/group entries carry their group inside flex_model["group"]; inflexible devices (which have no flex-model entry) carry it on group_reference instead.

group_reference: Any | None = None: For inflexible devices assigned to a group: the raw {"sensor": ...}/{"asset": ...} group reference (flexible devices carry theirs inside flex_model["group"] instead).

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

sensor_reference: Any | None = None: For inflexible devices given as a sensor reference with source filters: the SensorReference (a schema-layer object, hence untyped here); None otherwise.

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.