flexmeasures.data.services.automations
Logic for running automations (see also the CLI command flexmeasures jobs run-automations).
Functions
- flexmeasures.data.services.automations.check_sensor_access(input_sensors: list[Sensor], output_sensors: list[Sensor]) None
Require access to the sensors that an automation would read from and write to.
Reading a sensor’s data requires read access to it, and recording data on a sensor requires the same permission as recording data through the API (create-children).
- flexmeasures.data.services.automations.claim_due_automation(due_automation: DueAutomation) bool
Persist a run claim if its scheduling configuration is unchanged.
- flexmeasures.data.services.automations.collect_schedule_output_sensors(message: dict) list[Sensor]
The sensors that scheduling with this trigger message would record data on.
That is the power sensor of each device in the flex-model, plus any sensor named by a field that defines where generated data goes (see OUTPUT_SENSOR_FIELDS), both per device and, for the aggregates, in the flex-context.
- flexmeasures.data.services.automations.collect_sensors(value: Any, sensors: dict[int, Sensor] | None = None, only_under_output_field: bool = False, _under_output_field: bool = False) list[Sensor]
Collect the sensors referenced anywhere in a (possibly nested) structure.
Both deserialized sensors and the sensor references that survive deserialization as raw data (e.g. the flex-context and each device’s flex-model, which schedulers deserialize themselves) are picked up.
- Parameters:
only_under_output_field – only collect the sensors that are referenced under one of the OUTPUT_SENSOR_FIELDS, at any depth.
- flexmeasures.data.services.automations.create_automation(asset, name: str, cronstr: str, timezone: str | None = None, automation_type: str = 'forecasts', active: bool = True, parameters: dict | None = None, forecaster_class: str = 'TrainPredictPipeline', config: dict | None = None, source=None, origin: str = 'API', check_permissions: bool = False) tuple[Automation, list[str]]
Create an automation (not committed yet), validating its parameters by type.
For forecasts, the forecaster config is stored on a data source. An audit log record is added to the asset.
- Parameters:
check_permissions – whether to require that the current user may read the sensors that the automation reads from, and record data on the sensors it writes to. Set this for automations created by a user (through the API or the UI); the CLI runs without a user, and is trusted.
- Raises:
marshmallow.ValidationError – if the parameters are invalid.
ValueError – if the forecaster cannot be set up.
werkzeug.exceptions.Forbidden – if a sensor is not accessible to the user.
- Returns:
the automation and a list of warnings.
- flexmeasures.data.services.automations.delete_automation(automation: Automation, origin: str = 'API')
Delete an automation (not committed yet), recording it in the asset’s audit log.
- flexmeasures.data.services.automations.describe_cronstr(cronstr: str) str
Describe a cron string in natural language, e.g. “At 06:00”.
Explicitly renders times in 24-hour format, as cron-descriptor otherwise picks a format based on the system locale.
- flexmeasures.data.services.automations.floor_to_minute(dt: datetime) datetime
Floor a timezone-aware datetime to a UTC minute.
- flexmeasures.data.services.automations.floor_to_resolution(dt: datetime, resolution: timedelta) datetime
Floor an aware datetime to a fixed resolution without losing its DST fold.
- flexmeasures.data.services.automations.get_automation_job_stats(automation: Automation) dict[str, int]
Count the jobs created by this automation, per job status.
Note that jobs in Redis have a limited TTL, so this only counts fairly recent jobs.
- flexmeasures.data.services.automations.get_automation_sensors(automation: Automation) dict[str, list[Sensor]]
Look up which sensors an automation reads from and writes to on each run, for display purposes.
Automations whose sensors cannot be worked out report no sensors, so that one broken automation does not keep a page or an API response from rendering. Do not use this to decide whether something is permitted, as “no sensors” then reads as “nothing to check”: call resolve_automation_sensors instead and let its error propagate.
- flexmeasures.data.services.automations.get_automations_feeding_sensor(sensor: Sensor) list[Automation]
Find the automations that write data to the given sensor.
Only automations on the sensor’s own asset or on one of its ancestors are considered, as an automation may only write to its asset’s subtree (see validate_forecast_output_scope). Working out the output sensors requires setting up each candidate’s data generator, so this keeps the work proportional to the number of automations that could feed this sensor.
Note that this does not filter by permission: callers showing these to a user should check read access on each automation (e.g. with user_can_read).
- flexmeasures.data.services.automations.get_automations_involving_sensor(sensor: Sensor) list[Automation]
Find the automations that read from or write to the given sensor.
Unlike get_automations_feeding_sensor, this considers every automation, because a regressor may live anywhere in the tree, not just on the sensor’s asset or one of its ancestors. That makes this proportional to the number of automations, so keep it out of hot paths; it is meant for rare, interactive checks, such as warning before a sensor is deleted.
- flexmeasures.data.services.automations.get_due_automations(now: datetime | None = None) list[DueAutomation]
Return the newest unhandled run for each active automation.
- flexmeasures.data.services.automations.get_forecast_output_sensor(parameters: dict[str, Any]) Sensor
Resolve the sensor on which a forecast automation registers beliefs.
- flexmeasures.data.services.automations.get_latest_scheduled_run(automation: Automation, now: datetime) datetime
Return the latest canonical run for an automation through
now.
- flexmeasures.data.services.automations.prepare_schedule_trigger_message(parameters: dict, asset_id: int) dict
Complete stored schedule parameters into a message for the AssetTriggerSchema.
The asset id is injected, and the (required) schedule start defaults to now, floored to the message’s resolution (if given, otherwise to the minute), so recurring automations produce fresh schedules on each run.
- flexmeasures.data.services.automations.resolve_automation_sensors(automation: Automation) dict[str, list[Sensor]]
Work out which sensors an automation reads from and writes to on each run.
Forecast sensors are derived from the data generator, while schedule sensors are derived from the same prepared trigger message used to queue the scheduling job. Raises AutomationSensorsUnknown if that cannot be done, e.g. because a forecast automation has no data generator, because its generator is not registered in this FlexMeasures instance, or because its parameters no longer load (say, after a sensor was deleted). Use this wherever the answer decides whether something is permitted; use get_automation_sensors for display.
- flexmeasures.data.services.automations.resolve_data_generator_sensors(data_generator, deserialized_parameters: dict) dict[str, list[Sensor]]
Ask a data generator which sensors it would read from and write to, given these parameters.
A data generator derives this from its own config and parameters, so it also picks up a regressor that filters on sources, which is a sensor reference rather than a plain sensor. Work out the answer here rather than in each caller, so that displaying the sensors involved and checking access to them can never disagree about what they are.
- flexmeasures.data.services.automations.resolve_schedule_automation_sensors(parameters: dict, asset_id: int) dict[str, list[Sensor]]
Resolve the sensors declared by a prepared schedule trigger.
- flexmeasures.data.services.automations.run_automation(automation: Automation) dict[str, Any] | None
Queue the jobs for one run of an automation.
- Returns:
a dict like {“job_id”: <uuid>, “n_jobs”: <int>}.
- flexmeasures.data.services.automations.update_automation(automation: Automation, name: str | None = None, cronstr: str | None = None, timezone: str | None = None, active: bool | None = None, origin: str = 'API') list[str]
Update an automation’s name, cron string, timezone and/or activation status (not committed yet).
Anything that changes which runs are due, namely the recurrence, the timezone and reactivation, also rebases the cursor, so that runs from before the change are not caught up on. An audit log record is added to the asset.
- Returns:
a list of (human-readable) changes; empty if nothing changed.
- flexmeasures.data.services.automations.validate_forecast_output_scope(asset_id: int, output_sensor: Sensor) None
Require forecast output on the automation asset or a descendant.
Classes
- class flexmeasures.data.services.automations.DueAutomation(automation: Automation, scheduled_at: datetime, expected_cursor: datetime | None, expected_cronstr: str, expected_timezone: str)
An automation together with the canonical run it should handle.
Exceptions
- exception flexmeasures.data.services.automations.AutomationSensorsUnknown
Raised when the sensors an automation involves cannot be worked out.
Callers that decide whether something is allowed must let this propagate rather than treat it as “no sensors”, because an automation with no known sensors would otherwise pass every check on the sensors it involves.