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: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | 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.

fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

class flexmeasures.data.schemas.scheduling.CommitmentSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | None = None)
fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

class flexmeasures.data.schemas.scheduling.DBCommitmentSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | None = None)
fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

class flexmeasures.data.schemas.scheduling.DBFlexContextSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | 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.

fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

validate_fields_unit(data: dict, **kwargs)

Check that each field value has a valid unit.

class flexmeasures.data.schemas.scheduling.FlexContextSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | 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.

  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.

fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

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.GetScheduleSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | None = None)
fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

class flexmeasures.data.schemas.scheduling.MultiSensorFlexModelSchema(*, only: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | 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”

}

}

fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

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: Sequence[str] | AbstractSet[str] | None = None, exclude: Sequence[str] | AbstractSet[str] = (), many: bool | None = None, load_only: Sequence[str] | AbstractSet[str] = (), dump_only: Sequence[str] | AbstractSet[str] = (), partial: bool | Sequence[str] | AbstractSet[str] | None = None, unknown: Literal['exclude', 'include', 'raise'] | None = None)
fields: dict[str, Field]

Dictionary mapping field_names -> Field objects

forbid_time_series_specs(data: dict, **kwargs)

Do not allow time series specs for the flex-context fields saved in the db.