flexmeasures.data.services.automations
Logic for running automations (see also the CLI command flexmeasures jobs run-automations).
Functions
- flexmeasures.data.services.automations.create_automation(asset, name: str, cronstr: str, automation_type: str = 'forecasts', active: bool = True, parameters: dict | None = None, generator_class: str | None = None, config: dict | None = None, source=None, origin: str = 'API') tuple[Automation, list[str]]
Create an automation (not committed yet), validating its parameters by type.
For forecasts and reports, the data generator config is stored on a data source. An audit log record is added to the asset.
- Raises:
marshmallow.ValidationError – if the parameters are invalid.
ValueError – if the data generator cannot be set up.
- 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 datetime to the minute, in the FLEXMEASURES_TIMEZONE.
- 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_last_run(automation_id: int) datetime | None
Until when this automation’s work is covered, if known (the record lives in Redis).
- flexmeasures.data.services.automations.get_due_automations(now: datetime | None = None) list[Automation]
Return active automations whose cron string matches the given (or current) minute.
Cron strings are interpreted in the FLEXMEASURES_TIMEZONE.
- flexmeasures.data.services.automations.prepare_report_parameters(parameters: dict, cronstr: str, now: datetime | None = None, automation_id: int | None = None) dict
Complete stored report parameters into a message for the ReporterParametersSchema.
The (required) start and end of the report are resolved on each run:
“start-offset” and “end-offset” fields hold comma-separated Pandas offsets (e.g. “-1D,DB” for the start of the previous day), applied to the run time (or to the given absolute start/end), in the timezone of the first output sensor.
Without offsets or absolutes, the window runs since the end of the automation’s last (successfully) covered window, falling back to the last cron period (from the previous cron fire time until the run time) when none is known (e.g. on the first run).
- 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.record_automation_run(automation_id: int, now: datetime | None = None)
Remember (in Redis) until when this automation’s work is covered.
For forecasts and schedules automations, this is the (enqueue) run time. For reports automations, the reporting job records the end of the report window instead, upon success (see run_report_job), so a failed report job does not create a permanent gap in the reported periods.
- 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, active: bool | None = None, origin: str = 'API') list[str]
Update an automation’s name, cron string and/or activation status (not committed yet).
An audit log record is added to the asset.
- Returns:
a list of (human-readable) changes; empty if nothing changed.