flexmeasures.data.schemas.scheduling.storage

Functions

flexmeasures.data.schemas.scheduling.storage.validate_inflexible_flex_model_entry(data: dict, original_data: dict)

Validate a flex-model entry that declares an inflexible device.

An inflexible device is declared by a single inflexible-consumption or inflexible-production sensor reference. Such an entry must not declare both signs, must not use a sensor whose explicit consumption_is_positive attribute contradicts the field’s sign convention, and must not also carry schedulable-device fields (so it is unambiguously classified as an inflexible device). The last check inspects the original (hyphenated) input keys against a whitelist of keys allowed alongside an inflexible declaration, so it works for both flex-model schemas, stays complete as device fields are added, and ignores load-default fills.

Classes

class flexmeasures.data.schemas.scheduling.storage.DBStorageFlexModelSchema(*args, **kwargs)

Schema for flex-models stored in the db. Supports fixed quantities and sensor references, while disallowing time series specs.

__init__(*args, **kwargs)

_validate_array_fields(data: dict): Validate power array fields.

_validate_energy_fields(data: dict): Validate energy fields.

_validate_field(data: dict, field: str, unit_validator: Callable): Validate fields based on type and unit validator.

_validate_power_fields(data: dict): Validate power fields.

forbid_time_series_specs(data: dict, **kwargs): Do not allow time series specs for the flex-model fields saved in the db.

validate_fields_unit(data: dict, **kwargs): Check that each field value has a valid unit.

class flexmeasures.data.schemas.scheduling.storage.EfficiencyField(*args, **kwargs)

Field that deserializes to a Quantity with % units. Fixed values must be greater than 0% and less than or equal to 100%.

Examples:

>>> ef = EfficiencyField()
>>> ef.deserialize(0.9)
<Quantity(90.0, 'percent')>
>>> ef.deserialize("90%")
<Quantity(90, 'percent')>
>>> ef.deserialize("0%")
Traceback (most recent call last):
...
marshmallow.exceptions.ValidationError: ['Must be greater than 0 % and less than or equal to 100 %.']

__init__(*args, **kwargs)

One operation mode of a device, in the sense of the S2 standard.

A device with operation modes can only run within one of the declared modes’ power ranges at any given time. Each range is given with an explicit sign convention: consumption-range (non-negative, positive means consumption) and/or production-range (non-negative, positive means production). A mode may use either or both; using both forms a single band through zero (so both must then start at 0). The S2 standard fixes one sign convention for power (positive means consumption), whereas FM leaves it to the user; an S2 power-range therefore maps onto these fields by sign: its non-negative part corresponds to the FM consumption-range, negative S2 power values (production) correspond to the FM production-range (with their sign flipped to non-negative), and an S2 range spanning zero maps to a combination of both. A device that can only be off or run at exactly 883.7 W of consumption declares:

[{“consumption-range”: [“0 W”, “0 W”]}, {“consumption-range”: [“883.7 W”, “883.7 W”]}]

fields: dict[str, Field]: Dictionary mapping field_names -> Field objects

static signed_band(mode: dict) → tuple[float, float]

Convert one deserialized operation mode into a signed (min, max) power band in MW (positive is consumption), as used by the device scheduler.

The consumption-range maps to the positive side and the production-range to the negative side; combining both (each validated to start at 0) yields one band through zero [-production_max, +consumption_max].

>>> schema = OperationModeSchema()
>>> OperationModeSchema.signed_band(schema.load({"consumption-range": ["500 kW", "2 MW"]}))
(0.5, 2.0)
>>> OperationModeSchema.signed_band(schema.load({"production-range": ["500 kW", "1 MW"]}))
(-1.0, -0.5)
>>> OperationModeSchema.signed_band(schema.load(
...     {"consumption-range": ["0 MW", "2 MW"], "production-range": ["0 MW", "1 MW"]}
... ))
(-1.0, 2.0)

class flexmeasures.data.schemas.scheduling.storage.SoCTarget

class flexmeasures.data.schemas.scheduling.storage.StorageFlexModelSchema(start: datetime, sensor: Sensor | None, *args, default_soc_unit: str | None = None, **kwargs)

This schema lists fields we require when scheduling storage assets. Some fields are not required, as they might live on the Sensor.attributes. You can use StorageScheduler.deserialize_flex_config to get that filled in.

__init__(start: datetime, sensor: Sensor | None, *args, default_soc_unit: str | None = None, **kwargs): Pass the schedule’s start, so we can use it to validate soc-target datetimes.

check_redundant_efficiencies(data: dict, **kwargs)

Check that none of the following cases occurs:

flex-model contains both a round-trip efficiency and a charging efficiency
flex-model contains both a round-trip efficiency and a discharging efficiency
flex-model contains a round-trip efficiency, a charging efficiency and a discharging efficiency

Raise:: ValidationError

fields: dict[str, Field]: Dictionary mapping field_names -> Field objects

post_load_sequence(data: dict, **kwargs) → dict: Perform some checks and corrections after we loaded.

validate_coupling_direction_is_unambiguous(data: dict, **kwargs)

A coupled device must have an inferable flow direction.

The flow direction is inferred from which directional capacity is given: a device with (only) a consumption-capacity is an input (consuming) device, and a device with (only) a production-capacity is an output (producing) device. The unspecified direction is assumed to be zero, mirroring how a missing directional site capacity defaults to zero, so the user does not need to set the opposite direction to a fixed 0 (though doing so still works).

The direction is ambiguous only when both directions are active (each side either flows itself, or is marked active by a fixed zero on the opposite side), or when neither is (both missing). Such flex-models are rejected.