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_run(due_automation: DueAutomation, owner: str | None = None, lease: timedelta = datetime.timedelta(seconds=600)) → ClaimedAutomationRun | None

Atomically claim a newly due occurrence and create its durable run.

flexmeasures.data.services.automations.claim_existing_automation_run(run: AutomationRun, owner: str | None = None, lease: timedelta = datetime.timedelta(seconds=600)) → ClaimedAutomationRun | None

Claim a durable automation run whose dispatch is unfinished and unclaimed.

A run is only up for grabs once no other runner holds a live claim on it, because the dispatch state turns to ‘partially_queued’ while the owning runner is still queueing the rest of its jobs. A runner which fails releases its own claim, so its run is immediately retryable.

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.dispatch_automation_run(claimed_run: ClaimedAutomationRun) → dict[str, Any]

Dispatch an already claimed automation run and record its attempt outcome.

flexmeasures.data.services.automations.ensure_automation_run_job_intents(run_id: int, job_specs: list[dict[str, Any]]) → list[AutomationRunJob]

Persist immutable logical job intents before any Redis enqueue.

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_automation_run_stats(automation: Automation) → dict[str, Any]

Summarize durable automation runs for API and UI status displays.

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_dispatchable_automation_runs(now: datetime | None = None, owner: str | None = None) → list[ClaimedAutomationRun]

Claim new due occurrences and resumable durable runs for dispatch.

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.mark_automation_job_queued(run_id: int, logical_job_key: str, rq_job_id: str) → None

Mark one logical job intent as queued in Redis.

flexmeasures.data.services.automations.mark_automation_run_dispatch_failed(run_id: int, attempt: AutomationRunAttempt | None, error: BaseException) → None

Record a failed dispatch attempt and release the claim, so the run stays retryable.

The failure may have come from the database itself, so roll back first to get a usable session, then re-read the run and the attempt through it.

flexmeasures.data.services.automations.mark_automation_run_dispatch_queued(run_id: int, attempt: AutomationRunAttempt | None = None) → None

Mark an automation run as fully queued and release its dispatch claim.

flexmeasures.data.services.automations.reconcile_automation_job_intent(intent: AutomationRunJob) → bool

Return whether Redis already has the deterministic job for an intent.

flexmeasures.data.services.automations.record_automation_job_failed(run_id: int | None, logical_job_key: str | None, error: BaseException) → None

Record that a worker failed an automation-created job.

The job may well have failed on the database itself, which leaves the session in an aborted transaction where every further statement is refused. Roll back first, so that the failure is still recorded. The job’s own uncommitted work is lost either way, since it is failing.

flexmeasures.data.services.automations.record_automation_job_started(run_id: int | None, logical_job_key: str | None) → None

Record that a worker started an automation-created job.

flexmeasures.data.services.automations.record_automation_job_succeeded(run_id: int | None, logical_job_key: str | None) → None

Record that a worker finished an automation-created job successfully.

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.

The sensors are derived from the data generator, configured with the automation’s own parameters. Raises AutomationSensorsUnknown if that cannot be done, e.g. because the 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.run_automation(automation: Automation, automation_run: AutomationRun | None = None) → 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.serialize_automation_run(run: AutomationRun) → dict[str, Any]

Return operator-facing durable status for one automation run.

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.ClaimedAutomationRun(run: AutomationRun, attempt: AutomationRunAttempt)

An automation run and the attempt which currently owns its dispatch.

__init__(run: AutomationRun, attempt: AutomationRunAttempt) → None

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.

__init__(automation: Automation, scheduled_at: datetime, expected_cursor: datetime | None, expected_cronstr: str, expected_timezone: str) → None

Exceptions

exception flexmeasures.data.services.automations.AutomationRunClaimError

Raised when an automation occurrence cannot be claimed.

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.