flexmeasures.api.v3_0.sensors

Functions

flexmeasures.api.v3_0.sensors.regressors_loader(config: dict | None) → dict[str, list[Sensor]]

Extract regressor sensors from the forecasting config for permission checking.

Parameters:: config – Deserialized forecasting config (output of TrainPredictPipelineConfigSchema), which already contains resolved Sensor objects for regressor fields.
Returns:: Mapping of request field name to referenced Sensor objects, or an empty dict if no config or no regressors are specified.

flexmeasures.api.v3_0.sensors.sensor_search_term_filter(term: str)

Classes

Schema for the request body of the DELETE /sensors/<id>/annotations endpoint.

validate_time_window(data, **kwargs): Validate that end is not before start.

Schema for the request body of the DELETE /sensors/<id>/data endpoint.

Validates that end is not before start, and for non-instantaneous sensors, that end is at least one sensor resolution later than start. The sensor must be provided via schema context (context["sensor"]). until is still accepted in place of end, as it was the name until v1.1.0.

validate_time_window(data, **kwargs): Validate start/end consistency against each other and the sensor resolution.

class flexmeasures.api.v3_0.sensors.SensorAPI

delete(id: int, sensor: Sensor): Delete a sensor given its identifier. .. :quickref: Sensors; Delete a sensor

delete_annotations(id: int, sensor: Sensor)

delete_data(id: int, sensor: Sensor)

fetch_one(id, sensor: Sensor)

get_data(id: int, **sensor_data_description: dict)

get_forecast(id: int, uuid: str, sensor: Sensor, job_id: str)

get_schedule(sensor: Sensor, job_id: str, duration: timedelta, unit: str | None = None, sign_convention: str = 'consumption-positive', **kwargs)

get_stats(id, sensor: Sensor, event_start_time: str, event_end_time: str, sort_keys: bool, fresh: bool)

get_status(id, sensor, asset=None)

index(account: Account | None = None, asset: GenericAsset | None = None, include_consultancy_clients: bool = False, include_public_assets: bool = False, page: int | None = None, per_page: int | None = None, filter: list[str] | None = None, unit: str | None = None)

patch(sensor_data: dict, id: int, sensor: Sensor)

post(sensor_data: dict)

post_annotation(annotation: Annotation, id: int, sensor: Sensor)

post_data(id: int, sensor: Sensor, sensor_data: dict)

trigger_forecast(id: int, **params)

trigger_schedule(sensor: Sensor, start_of_schedule: datetime, duration: timedelta, resolution: timedelta | None = None, belief_time: datetime | None = None, flex_model: dict | None = None, flex_context: dict | None = None, force_new_job_creation: bool | None = False, **kwargs)

upload_data(sensor: Sensor, uploaded_files: list, filenames: list[str], unit: str | None = None, belief_time_measured_instantly: bool | None = None, prior: datetime | None = None, **kwargs)

class flexmeasures.api.v3_0.sensors.SensorKwargsSchema(*, 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)

class flexmeasures.api.v3_0.sensors.TriggerScheduleKwargsSchema(*, 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)