flexmeasures.data.services.sensors

Functions

flexmeasures.data.services.sensors.build_asset_jobs_data(asset: GenericAsset) → list[dict]: Get all jobs data for an asset Returns a list of dictionaries, each containing the following keys: - job_id: id of a job - queue: job queue (scheduling or forecasting) - asset_or_sensor_type: type of an asset that is linked to the job (asset or sensor) - asset_id: id of sensor or asset - status: job status (e.g finished, failed, etc) - err: job error (equals to None when there was no error for a job) - enqueued_at: time when the job was enqueued - metadata_hash: hash of job metadata (internal field)

flexmeasures.data.services.sensors.get_asset_sensors_metadata(asset: GenericAsset, now: datetime | None = None) → list[dict]

Get the metadata of sensors for a given asset and its children.

Parameters:

asset – Asset to get the sensors for.
now – Datetime representing now, used to get the status of the sensors.

Returns:

A list of dictionaries, each representing a sensor’s metadata.

flexmeasures.data.services.sensors.get_sensor_stats(sensor: Sensor, event_start_time: str, event_end_time: str, sort_keys: bool = True) → dict: Get stats for a sensor

flexmeasures.data.services.sensors.get_sensors(account: Account | list[Account] | None, include_public_assets: bool = False, sensor_id_allowlist: list[int] | None = None, sensor_name_allowlist: list[str] | None = None) → list[Sensor]

Return a list of Sensor objects that belong to the given account, and/or public sensors.

Parameters:

account – select only sensors from this account (or list of accounts)
include_public_assets – if True, include sensors that belong to a public asset
sensor_id_allowlist – optionally, allow only sensors whose id is in this list
sensor_name_allowlist – optionally, allow only sensors whose name is in this list

flexmeasures.data.services.sensors.get_staleness_start_times(sensor: Sensor, staleness_search: dict, now: datetime) → dict[str, timedelta] | None

Get staleness start times for a given sensor by source. Also add whether there has any relevant data (for forecasters and schedulers this is future data). For scheduler and forecaster sources staleness start is latest event start time.

For other sources staleness start is the knowledge time of the sensor’s most recent event. This knowledge time represents when you could have known about the event (specifically, when you could have formed an ex-post belief about it).

flexmeasures.data.services.sensors.get_stalenesses(sensor: Sensor, staleness_search: dict, now: datetime) → dict[str, timedelta] | None

Get the staleness of the sensor split by source.

The staleness is defined relative to the knowledge time of the most recent event, rather than to its belief time. Basically, that means that we don’t really care when the data arrived, as long as the available data is about what we should be able to know by now.

Parameters:

sensor – The sensor to compute the staleness for.
staleness_search – Deserialized keyword arguments to TimedBelief.search.
now – Datetime representing now, used both to mask future beliefs, and to measures staleness against.

flexmeasures.data.services.sensors.get_status_specs(sensor: Sensor) → dict: Get status specs from a given sensor.

flexmeasures.data.services.sensors.get_statuses(sensor: Sensor, now: datetime, status_specs: dict | None = None) → list[dict]: Get the status of the sensor by source type. Main part of result here is a stale value, which is True if the sensor is stale, False otherwise. Other values are just context information for the stale value.

flexmeasures.data.services.sensors.serialize_sensor_status_data(sensor: Sensor) → list[dict]

Serialize the status of a sensor belonging to an asset.

Parameters:: sensor – Sensor to get the status of
Returns:: A list of dictionaries, each representing the statuses of the sensor - one status per data source type that stored data on that sensor