diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index d06b8c1784..a8232b7038 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -52,6 +52,7 @@ create_automation, delete_automation as remove_automation, describe_cronstr, + get_asset_automations_job_stats, get_automation_job_stats, update_automation, ) @@ -1237,10 +1238,12 @@ def get_automations(self, id: int, asset: GenericAsset): get: summary: Get all automations defined on an asset. description: | - The response will be a list of automations: recurring tasks (for now, computing forecasts) - defined on the asset. Each entry shows the automation's ID, when it was created, - its type, name, activation status, and its recurrence, both as a cron string - and described in natural language. + The response will be a list of automations: recurring tasks (computing forecasts, + schedules or reports) defined on the asset. Each entry shows the automation's ID, + when it was created, its type, name, activation status, its recurrence (both as a + cron string and described in natural language), and counts of recently created + jobs per job status (note that jobs in Redis have a limited TTL, so not all past + jobs are counted). security: - ApiKeyAuth: [] parameters: @@ -1268,6 +1271,9 @@ def get_automations(self, id: int, asset: GenericAsset): cronstr: "0 6 * * *" recurrence_description: "At 06:00" active: true + job_stats: + finished: 3 + redis_connection_err: null 400: description: INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS 401: @@ -1279,14 +1285,24 @@ def get_automations(self, id: int, asset: GenericAsset): tags: - Assets """ + redis_connection_err = None + try: + job_stats = get_asset_automations_job_stats(asset) + except NoRedisConfigured as e: + job_stats = {} + redis_connection_err = e.args[0] automations_data = [] for automation in asset.automations: automation_data = automation_schema.dump(automation) automation_data["recurrence_description"] = describe_cronstr( automation.cronstr ) + automation_data["job_stats"] = job_stats.get(automation.id, {}) automations_data.append(automation_data) - return {"automations": automations_data}, 200 + return { + "automations": automations_data, + "redis_connection_err": redis_connection_err, + }, 200 @route("//automations/", methods=["GET"]) @use_kwargs( diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 3c1ea8ce52..e1866f009d 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -82,6 +82,7 @@ def test_get_automations( assert day_ahead["recurrence_description"] == "At 06:00" assert day_ahead["active"] is True assert day_ahead["created_at"] is not None + assert day_ahead["job_stats"] == {} # this automation has not queued any jobs # generator and parameters are not listed assert "generator_id" not in day_ahead assert "generator" not in day_ahead diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index 59b99e60c1..061db63c62 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -228,16 +228,14 @@ def _relevant_sensor_ids(automation: Automation, parameter_values: list) -> set[ return sensor_ids -def get_automation_job_stats(automation: Automation) -> dict[str, int]: - """Count the jobs created by this automation, per job status. +def _job_cache_refs(automation: Automation) -> set[tuple[int, str, str]]: + """The job cache entries where this automation's jobs may live. - Note that jobs in Redis have a limited TTL, so this only counts fairly recent jobs. + Forecasting and reporting jobs are cached under their target/output sensor(s), + which may belong to a different asset than the automation's own asset. + Scheduling jobs are cached under the asset (multi-device wrap-up jobs) + and under individual sensors (per-device jobs). """ - from flask import current_app - - # Determine the job cache entries to scan. Forecasting and reporting jobs - # are cached under their target/output sensor(s), which may belong to a - # different asset than the automation's own asset. parameters = automation.parameters or {} if automation.type == "schedules": # Scheduling jobs are cached under the asset (multi-device wrap-up jobs) @@ -251,9 +249,9 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: if isinstance(entry, dict) ], ) - cache_refs = [(automation.asset_id, "scheduling", "asset")] + [ + return {(automation.asset_id, "scheduling", "asset")} | { (sensor_id, "scheduling", "sensor") for sensor_id in sensor_ids - ] + } elif automation.type == "reports": sensor_ids = _relevant_sensor_ids( automation, @@ -263,27 +261,64 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: if isinstance(output, dict) ], ) - cache_refs = [(sensor_id, "reporting", "sensor") for sensor_id in sensor_ids] + return {(sensor_id, "reporting", "sensor") for sensor_id in sensor_ids} else: sensor_ids = _relevant_sensor_ids( automation, [parameters.get("sensor"), parameters.get("sensor-to-save")], ) - cache_refs = [(sensor_id, "forecasting", "sensor") for sensor_id in sensor_ids] + return {(sensor_id, "forecasting", "sensor") for sensor_id in sensor_ids} + - counts: dict[str, int] = {} +def _count_automation_jobs( + cache_refs: set[tuple[int, str, str]], automation_ids: set[int] +) -> dict[int, dict[str, int]]: + """Count jobs per automation and job status, in a single pass over the given job cache entries.""" + from flask import current_app + + counts: dict[int, dict[str, int]] = { + automation_id: {} for automation_id in automation_ids + } seen_job_ids: set[str] = set() for entity_id, queue, asset_or_sensor_type in cache_refs: for job in current_app.job_cache.get(entity_id, queue, asset_or_sensor_type): if job.id in seen_job_ids: continue seen_job_ids.add(job.id) - if job.meta.get("trigger", {}).get("automation_id") == automation.id: + automation_id = job.meta.get("trigger", {}).get("automation_id") + if automation_id in counts: status = str(job.get_status().value) - counts[status] = counts.get(status, 0) + 1 + counts[automation_id][status] = counts[automation_id].get(status, 0) + 1 return counts +def 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. + """ + return _count_automation_jobs(_job_cache_refs(automation), {automation.id})[ + automation.id + ] + + +def get_asset_automations_job_stats(asset) -> dict[int, dict[str, int]]: + """Count the jobs created by each of the asset's automations, per job status, + in a single pass over the relevant job cache entries. + + Note that jobs in Redis have a limited TTL, so this only counts fairly recent jobs. + """ + automations = asset.automations + if not automations: + return {} + cache_refs: set[tuple[int, str, str]] = set() + for automation in automations: + cache_refs |= _job_cache_refs(automation) + return _count_automation_jobs( + cache_refs, {automation.id for automation in automations} + ) + + def _prepare_forecast_automation( asset, parameters: dict, generator_class: str | None, config: dict | None, source ) -> tuple[int, list[str]]: diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 7cf3ac3669..5a0f196d51 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -3298,7 +3298,7 @@ "/api/v3_0/assets/{id}/automations": { "get": { "summary": "Get all automations defined on an asset.", - "description": "The response will be a list of automations: recurring tasks (for now, computing forecasts)\ndefined on the asset. Each entry shows the automation's ID, when it was created,\nits type, name, activation status, and its recurrence, both as a cron string\nand described in natural language.\n", + "description": "The response will be a list of automations: recurring tasks (computing forecasts,\nschedules or reports) defined on the asset. Each entry shows the automation's ID,\nwhen it was created, its type, name, activation status, its recurrence (both as a\ncron string and described in natural language), and counts of recently created\njobs per job status (note that jobs in Redis have a limited TTL, so not all past\njobs are counted).\n", "security": [ { "ApiKeyAuth": [] @@ -3333,9 +3333,13 @@ "name": "Day-ahead PV forecasts", "cronstr": "0 6 *", "recurrence_description": "At 06:00", - "active": true + "active": true, + "job_stats": { + "finished": 3 + } } - ] + ], + "redis_connection_err": null } } } diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 62ba6d73f7..06135c9962 100644 --- a/flexmeasures/ui/templates/assets/asset_automations.html +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -87,6 +87,35 @@
New automation for {{ asset.name }}
+ + + {% endif %}