flexmeasures.data.schemas.scheduling

Modules

flexmeasures.data.schemas.scheduling.process
flexmeasures.data.schemas.scheduling.storage

Classes

class flexmeasures.data.schemas.scheduling.AssetTriggerSchema(*, only: types.StrSequenceOrSet | None = None, exclude: types.StrSequenceOrSet = (), many: bool | None = None, context: dict | None = None, load_only: types.StrSequenceOrSet = (), dump_only: types.StrSequenceOrSet = (), partial: bool | types.StrSequenceOrSet | None = None, unknown: str | None = None)

{

“start”: “2025-01-21T15:00+01”, “flex-model”: [

{

“sensor”: 1, “soc-at-start”: “10 kWh”

}, {

“sensor”: 2, “soc-at-start”: “20 kWh”

},

]

}

check_flex_model_sensors(data, **kwargs)

Verify that the flex-model’s sensors live under the asset for which a schedule is triggered.

class flexmeasures.data.schemas.scheduling.DBFlexContextSchema(*, only: types.StrSequenceOrSet | None = None, exclude: types.StrSequenceOrSet = (), many: bool | None = None, context: dict | None = None, load_only: types.StrSequenceOrSet = (), dump_only: types.StrSequenceOrSet = (), partial: bool | types.StrSequenceOrSet | None = None, unknown: str | None = None)

_forbid_fixed_prices(data: dict, **kwargs)

Do not allow fixed consumption price or fixed production price in the flex-context fields saved in the db.

This is a temporary restriction as future iterations will allow fixed prices on these fields as well.

_validate_field(data: dict, field_type: str, field: str, unit_validator)

Validate fields based on type and unit validator.

_validate_inflexible_device_sensors(data: dict)

Validate inflexible device sensors.

_validate_power_fields(data: dict)

Validate power fields.

_validate_price_fields(data: dict)

Validate price fields.

forbid_time_series_specs(data: dict, **kwargs)

Do not allow time series specs for the flex-context 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.FlexContextSchema(*, only: types.StrSequenceOrSet | None = None, exclude: types.StrSequenceOrSet = (), many: bool | None = None, context: dict | None = None, load_only: types.StrSequenceOrSet = (), dump_only: types.StrSequenceOrSet = (), partial: bool | types.StrSequenceOrSet | None = None, unknown: str | None = None)

This schema defines fields that provide context to the portfolio to be optimized.

static _to_currency_per_mwh(price_unit: str) → str

Convert a price unit to a base currency used to express that price per MWh.

>>> FlexContextSchema()._to_currency_per_mwh("EUR/MWh")
'EUR'
>>> FlexContextSchema()._to_currency_per_mwh("EUR/kWh")
'EUR'

_try_to_convert_price_units(data)

Convert price units to the same unit and scale if they can (incl. same currency).

check_prices(data: dict, **kwargs)

Check assumptions about prices.

1. The flex-context must contain at most 1 consumption price and at most 1 production price field.

2. All prices must share the same currency.

set_default_breach_prices(data: dict, fields: list[str], price: Quantity)

Fill in default breach prices.

This relies on _try_to_convert_price_units to run first, setting a shared currency unit.

class flexmeasures.data.schemas.scheduling.MultiSensorFlexModelSchema(*, only: types.StrSequenceOrSet | None = None, exclude: types.StrSequenceOrSet = (), many: bool | None = None, context: dict | None = None, load_only: types.StrSequenceOrSet = (), dump_only: types.StrSequenceOrSet = (), partial: bool | types.StrSequenceOrSet | None = None, unknown: str | None = None)

This schema is agnostic to the underlying type of flex-model, which is governed by the chosen Scheduler instead. Therefore, the underlying type of flex-model is not deserialized.

So:

{

“sensor”: 1, “soc-at-start”: “10 kWh”

}

becomes:

{

“sensor”: <Sensor 1>, “sensor_flex_model”: {

“soc-at-start”: “10 kWh”

}

}

unwrap_envelope(data, **kwargs)

Any field other than ‘sensor’ becomes part of the sensor’s flex-model.

wrap_with_envelope(data, **kwargs)

Any field in the ‘sensor-flex-model’ field becomes a main field.