flexmeasures.data.schemas.scheduling

Modules

`flexmeasures.data.schemas.scheduling.metadata`	These descriptions are used in documentation/features/scheduling.rst and in OpenAPI. If you need to use a new .rst directive, update make_openapi_compatible accordingly, so it shows up nicely in OpenAPI. For instance: - the X directive is converted to a <abbr title="Y">X</abbr> HTML tag. - any footnote references, such as [#quantity_field]_, are stripped (these are meant for .rst docs to explain field types, which in OpenAPI is redundant, given that each field is already documented as being of an explicit type).
`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.CommitmentSchema(*, 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)

class flexmeasures.data.schemas.scheduling.DBCommitmentSchema(*, 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)

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.

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: dict, original_data: dict): Convert price units to the same unit and scale if they can (incl. same currency).

check_prices(data: dict, original_data: dict, **kwargs)

Check assumptions about prices.

The flex-context must contain at most 1 consumption price and at most 1 production price field.
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’ and ‘asset’ 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.

class flexmeasures.data.schemas.scheduling.NoTimeSeriesSpecs(*, 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_time_series_specs(data: dict, **kwargs): Do not allow time series specs for the flex-context fields saved in the db.