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

{

“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

normalize_flex_context_format(data, **kwargs)

Normalize flex_context to always be a dict.

Accepts both: - Single commodity dict: {“commodity”: “electricity”, …} - List of commodity dicts: [{“commodity”: “electricity”, …}, {“commodity”: “heat”, …}] - MultiDict with multiple ‘flex-context’ entries (when JSON list is parsed by webargs)

If a list is provided, it is wrapped under the ‘commodities’ field. If a dict is provided, it is kept as-is. This ensures downstream code always sees a dict structure.

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

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

__init__(*args, **kwargs)

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

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

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

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

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.

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

validate_commodity_contexts_shared_currency(commodity_contexts: list[dict], **kwargs)

Validate that all prices across commodity contexts share the same currency.

Each commodity context already computed its own normalized shared_currency_unit (a base-unit currency string, e.g. “EUR”) via the inherited _try_to_convert_price_units schema-level validator. We simply compare those.

validate_commodity_contexts_unique(commodity_contexts: list[dict], **kwargs)

Validate that each commodity is listed at most once.

_get_commodity_contexts (storage.py) builds a dict keyed by commodity, so duplicate entries would otherwise silently overwrite each other.

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

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.

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.

class flexmeasures.data.schemas.scheduling.ScheduleSignConvention

Named constants for the three sign-convention modes of the get_schedule endpoint.

Variables:

CONSUMPTION_POSITIVE – Always return schedules with consumption as positive values and production as negative values. This is the default and matches the view a consumer has of their device.
PRODUCTION_POSITIVE – Always return schedules with production as positive values and consumption as negative values. This matches the view a producer (or generator) has of their device.
WYSIWYG – Return the raw values from the database without any sign inversion, regardless of the sensor’s consumption_is_positive attribute. Useful when you want to see exactly what was stored.

Shared schema for fields common across commodities in flex-context and commodity-context.

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