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, forecaster_class: str = 'TrainPredictPipeline', 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, the forecaster 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 forecaster 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_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_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.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.