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 an occurrence claim before its non-transactional queueing attempt.

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.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[DueAutomation]: Return the newest unhandled occurrence 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_occurrence(automation: Automation, now: datetime) → datetime: Return the latest canonical occurrence for an automation through now.

flexmeasures.data.services.automations.run_automation(automation: Automation) → dict[str, Any] | None

Queue the jobs for one run of an automation.

Returns:: the data generator’s return value, e.g. {“job_id”: <uuid>, “n_jobs”: <int>} for forecasting jobs.

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)

An automation together with the canonical occurrence it should handle.