flexmeasures.data.services.automations
Logic for running automations (see also the CLI command flexmeasures jobs run-automations).
Functions
- 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]
Collect sensors on which the prepared schedule trigger records results.
- 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 sensor objects and references from a nested scheduling structure.
- 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_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.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.