diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 6329fd9a4a..d077fc83f5 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -44,6 +44,7 @@ v3.0-34 | November XX, 2026 - ``GET /api/v3_0/assets//jobs`` now accepts ``page`` and ``per-page`` to return jobs newest first, with ``num-records`` and ``filtered-records`` counts. Paginated results can be sorted by enqueue time or queue with ``sort-by`` and ``sort-dir``. Requests without ``page`` retain the full list and previous response shape. - ``GET /api/v3_0/assets//jobs`` also lists the jobs of the assets below the asset, at any depth, as far as the caller may read them; pass ``include-child-assets=false`` to list only the asset's own jobs. Its entries include ``created_via`` provenance, and the automation that created a job only where the caller may read that automation. +- ``GET /api/v3_0/jobs/`` reports, for a finished reporting or scheduling job, how many beliefs it computed and how many it recorded, per output sensor. A reporting job's ``result`` is an object holding an ``outputs`` list, and a scheduling job's ``result`` gains the same ``outputs`` list, next to its unchanged ``num-beliefs`` (the total of ``num-computed-beliefs`` across ``outputs``). Each entry holds the ``sensor`` (its ID), ``num-computed-beliefs`` and ``num-recorded-beliefs``. The recorded count leaves out beliefs that were already on record (and NaN values), so a rerun that only repeats what is on record records 0 of the beliefs it computed. - ``GET /api/v3_0/assets//jobs`` renames its ``redis_connection_err`` field, which shipped in v1.0.0, to ``redis-connection-err``, as the automation endpoints spell it. It is read by FlexMeasures' own UI rather than by the FlexMeasures client. - ``GET /api/v3_0/assets`` now scopes a listing the way its parameters say. ``num-records`` respects the ``root`` and ``depth`` constraints, where it used to count every asset in the account and asset-type scope. And an explicit ``include_public=false`` is now honoured, also next to ``all_accessible`` or ``root``; leaving the parameter out keeps the behaviour it had, where ``all_accessible`` and ``root`` include public assets and a listing of a single account does not. - ``GET /api/v3_0/sensors//status`` now accepts an optional ``asset_id`` query parameter, naming the asset whose status page the sensor is reported on. It only affects the reported ``relation``, which used to claim that the sensor belongs to the asset shown, also for a sensor that is only listed because the asset's flex-context or graphs page refers to it. Reading the status of a sensor in the context of an asset requires read permission on that asset, too. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index d7aa3b9e51..592d6b35d6 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -191,7 +191,7 @@ Bugfixes * An ``AggregatorReporter`` input that holds more than one belief about an event, such as one read with ``most-recent-beliefs-only: false``, is now refused with a message naming the sensor and the event, where it used to fail with a ``KeyError`` [see `PR #2741 `_] * A ``PandasReporter`` with ``droplevels`` no longer fails with ``BeliefDataframe has more than one row per event`` when several data sources reported the same input event, as happens for a while when a job reporting on a rolling window gets a new data source; it keeps one belief per event, unless an input sets ``one_deterministic_belief_per_event`` to ``false`` [see `PR #2595 `_] -* The result of a reporting or scheduling job now counts the beliefs it saved rather than those it computed, so that a report's ``n_rows`` and a schedule's ``num-beliefs`` read 0 when a rerun only repeats what is on record [see `PR #2689 `_ and `PR #2571 `_] +* The results of reporting and scheduling jobs now list, per output sensor, the beliefs computed and those recorded, so that a rerun which only repeats what is on record reads as having recorded nothing, rather than as having computed nothing; a schedule's ``num-beliefs`` still totals the beliefs computed, now also on a CLI dry run [see `PR #2689 `_, `PR #2571 `_ and `PR #2822 `_] * A forecaster registered by a plugin no longer needs to save its forecasts itself, but can return them instead: ``flexmeasures add forecasts`` now saves them, where it used to drop them while reporting them as created [see `PR #2689 `_ and `PR #2571 `_] * Running the same forecast twice from the same start no longer fails on the beliefs it already recorded [see `PR #2688 `_] * Whether a forecaster takes a regressor's belief as realized now follows the knowledge horizon of the regressor's sensor, as in timely-beliefs, so that the past readings of an instantaneous regressor (such as a thermometer read on the hour) and day-ahead prices (beliefs about future periods, which count as realized once the auction has cleared) are no longer kept from the model [see `PR #2692 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index 80b38ecab0..d6f423253d 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -35,13 +35,14 @@ since v1.1.0 | November XX, 2026 * Add ``flexmeasures add report --as-job`` and the ``reporting`` worker queue for asynchronous one-off reports. * Add ``flexmeasures add forecasts --dry-run`` to compute a forecast and see what it would record, without saving anything to the database. It cannot be combined with ``--as-job``. * ``flexmeasures add report`` saves the reporter's config on its data source by default. As a report then records under a new source, anything referring to its old source by ID should be pointed at the new one after the first run. Pass ``--no-save-config`` to leave the config out and keep recording under the old source; ``--save-config`` is still accepted. -* ``flexmeasures add report`` saves all of a report's outputs or none of them, where each output used to be committed on its own, and says for each output how many beliefs it saved. +* ``flexmeasures add report`` saves all of a report's outputs or none of them, where each output used to be committed on its own, and says for each output how many beliefs it computed and how many of those it recorded. * ``flexmeasures add forecasts`` saves the forecasts of a forecaster that a plugin registers, which used to be dropped while the command reported them as created. * ``flexmeasures add forecasts`` no longer fails when it runs again from the same ``--from-date``, and says how many of the forecast beliefs it computed were new and saved. * ``flexmeasures add forecasts`` no longer keeps every trained model as a file; pass ``--model-save-dir`` to keep them. **Scheduling** +* ``flexmeasures add schedule`` says how many beliefs it computed and how many of those it recorded, so that a rerun which only repeats what is on record no longer reads as storing a new schedule. * Fix ``flexmeasures add schedule --dry-run`` saving a schedule when combined with ``--as-job``: the combination is now rejected, and a dry run reports the beliefs and events it would have saved. **Assets** diff --git a/flexmeasures/api/v3_0/jobs.py b/flexmeasures/api/v3_0/jobs.py index 524b33d745..40e8e84946 100644 --- a/flexmeasures/api/v3_0/jobs.py +++ b/flexmeasures/api/v3_0/jobs.py @@ -75,30 +75,28 @@ def get_job_status(self, job_id: str, **kwargs): get: summary: Get the status of a background job description: | - Look up a background job by its UUID and see whether it is - queued, running, finished, or failed. + Look up a background job by its UUID and see whether it is queued, running, finished, or failed. - The response includes a status message plus job metadata such - as the queue name, function name, timestamps, and the job - result when available. + The response includes a status message plus job metadata, + such as the queue name, function name, timestamps, and the job result when available. - Failed jobs also include traceback information when the worker - stored it with the job result. + Failed jobs also include traceback information when the worker stored it with the job result. - For a finished scheduling job, ``result`` is an object. For a - ``StorageScheduler`` job it holds soft state-of-charge constraint - analysis: ``unresolved`` lists constraints the scheduler could not - satisfy, and ``resolved`` lists constraints that were satisfied - with some margin. Each device entry's ``soc-minima``/``soc-maxima`` - value is a list, holding one entry per violated slot (for - ``unresolved``) or per met slot with its margin (for ``resolved``), - ordered chronologically. Both arrays are empty when the flex model - defines no ``soc-minima``/``soc-maxima``, or when a scheduler other - than ``StorageScheduler`` was used. The ``num-beliefs`` field holds - the total number of beliefs (scheduled values) saved to the database. - This is the only place constraint analysis is available — the sensor - schedule endpoint (``GET /api/v3_0/sensors//schedules/``) - returns power values only. + For a finished scheduling job, ``result`` is an object. + For a ``StorageScheduler`` job, it holds soft state-of-charge constraint analysis: + ``unresolved`` lists constraints the scheduler could not satisfy, + and ``resolved`` lists constraints that were satisfied with some margin. + Each device entry's ``soc-minima``/``soc-maxima`` value is a list, + holding one entry per violated slot (for ``unresolved``) or per met slot with its margin (for ``resolved``), ordered chronologically. + Both arrays are empty when the flex model defines no ``soc-minima``/``soc-maxima``, or when a scheduler other than ``StorageScheduler`` was used. + This is the only place constraint analysis is available: + the sensor schedule endpoint (``GET /api/v3_0/sensors//schedules/``) returns power values only. + The ``outputs`` field lists one entry per sensor the schedule records on: + the ``sensor`` (its ID), the number of beliefs (scheduled values) the scheduler computed for it (``num-computed-beliefs``), + and the number recorded in the database (``num-recorded-beliefs``), which leaves out beliefs that were already on record or had no value. + The ``num-beliefs`` field holds the total of ``num-computed-beliefs`` across ``outputs``. + + For a finished reporting job, ``result`` is an object holding the same ``outputs`` field, with one entry per output sensor of the report. security: - ApiKeyAuth: [] parameters: @@ -134,15 +132,14 @@ def get_job_status(self, job_id: str, **kwargs): description: Human-readable description of the job status. result: description: > - Return value of the job function, or null when not yet - available. For a finished scheduling job, this is an - object; a ``StorageScheduler`` job populates it with - ``unresolved``/``resolved`` soft state-of-charge - constraint analysis (empty arrays when the flex model - defines no ``soc-minima``/``soc-maxima``, or when a - scheduler other than ``StorageScheduler`` was used). - The ``num-beliefs`` field holds the total number of - beliefs (scheduled values) saved to the database. + Return value of the job function, or null when not yet available. + For a finished scheduling job, this is an object; + a ``StorageScheduler`` job populates it with ``unresolved``/``resolved`` soft state-of-charge constraint analysis + (empty arrays when the flex model defines no ``soc-minima``/``soc-maxima``, or when a scheduler other than ``StorageScheduler`` was used). + Its ``outputs`` field lists, per sensor the schedule records on, the ``sensor`` (its ID), + ``num-computed-beliefs`` and ``num-recorded-beliefs``, + and its ``num-beliefs`` field holds the total of ``num-computed-beliefs`` across ``outputs``. + For a finished reporting job, this is an object holding the same ``outputs`` field. nullable: true func-name: type: string @@ -223,6 +220,10 @@ def get_job_status(self, job_id: str, **kwargs): - datetime: "2024-01-01T10:15:00+00:00" violation: "180.0 kWh" resolved: [] + outputs: + - sensor: 5 + num-computed-beliefs: 96 + num-recorded-beliefs: 96 num-beliefs: 96 func-name: "flexmeasures.data.services.scheduling.create_schedule" origin: scheduling diff --git a/flexmeasures/api/v3_0/tests/test_jobs_api.py b/flexmeasures/api/v3_0/tests/test_jobs_api.py index 38cfdf2486..c8b32422e2 100644 --- a/flexmeasures/api/v3_0/tests/test_jobs_api.py +++ b/flexmeasures/api/v3_0/tests/test_jobs_api.py @@ -255,6 +255,10 @@ def test_get_job_status_finished( result = data["result"] assert result["unresolved"] == [] assert result["resolved"] == [] + assert set(result.keys()) == {"unresolved", "resolved", "outputs", "num-beliefs"} + assert result["outputs"] == [ + {"sensor": sensor.id, "num-computed-beliefs": 96, "num-recorded-beliefs": 96} + ] assert result["num-beliefs"] == 96 assert data["exc-info"] is None assert_legacy_job_status_fields(data) @@ -315,9 +319,17 @@ def test_get_job_status_finished_with_unresolved_soc_minima( result = data["result"] assert isinstance(result, dict) - assert set(result.keys()) == {"unresolved", "resolved", "num-beliefs"} + assert set(result.keys()) == { + "unresolved", + "resolved", + "outputs", + "num-beliefs", + } assert isinstance(result["num-beliefs"], int) assert result["num-beliefs"] > 0 + assert result["num-beliefs"] == sum( + output["num-computed-beliefs"] for output in result["outputs"] + ) asset_id = add_battery_assets["Test battery"].id unresolved_entry = next( diff --git a/flexmeasures/api/v3_0/tests/test_report_trigger_api_fresh_db.py b/flexmeasures/api/v3_0/tests/test_report_trigger_api_fresh_db.py index 917e30ec39..b57aa2ec5e 100644 --- a/flexmeasures/api/v3_0/tests/test_report_trigger_api_fresh_db.py +++ b/flexmeasures/api/v3_0/tests/test_report_trigger_api_fresh_db.py @@ -62,6 +62,22 @@ def test_trigger_report_worker_stores_report_data( # Adding the two hourly 0-9 input series yields 0, 2, ..., 18; two-hour mean resampling yields 1, 5, 9, 13, 17. assert stored_report.values.T.tolist() == [[1, 5, 9, 13, 17]] + # The job's result, as the API hands it out, lists the report's output with what it computed and what it saved. + with app.test_client() as client: + job_response = client.get( + url_for("JobAPI:get_job_status", uuid=response.json["job"]) + ) + assert job_response.status_code == 200 + result = job_response.json["result"] + assert set(result.keys()) == {"outputs"} + assert result["outputs"] == [ + { + "sensor": sensors["output"].id, + "num-computed-beliefs": 5, + "num-recorded-beliefs": 5, + } + ] + @pytest.mark.parametrize( "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True @@ -126,13 +142,17 @@ def test_trigger_report_worker_reports_nothing_saved_for_misaligned_inputs( job = app.queues["reporting"].fetch_job(response.json["job"]) assert job.get_status() == "finished" - # The count must exclude computed rows that persistence drops as NaN. - assert job.return_value() == [ - { - "sensor_id": hourly_output.id, - "n_rows": 0, - } - ] + # The saved count must exclude computed rows that persistence drops as NaN, while the computed count includes them: + # the two inputs do not align on source, so adding them gives a NaN for each of their 2 x 10 rows. + assert job.return_value() == { + "outputs": [ + { + "sensor": hourly_output.id, + "num-computed-beliefs": 20, + "num-recorded-beliefs": 0, + } + ] + } assert any( "produced no persistable values" in record.message for record in caplog.records ), "an empty report should be warned about" diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index bfe53da281..1b5808fe5e 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -2199,16 +2199,31 @@ def add_schedule( # noqa C901 **MsgStyle.SUCCESS, ) else: - make_schedule( + scheduling_result = make_schedule( asset_or_sensor=get_asset_or_sensor_ref(asset_or_sensor), dry_run=dry_run, **scheduling_kwargs, ) - if not dry_run: - click.secho("New schedule is stored.", **MsgStyle.SUCCESS) + n_computed = sum( + output["num-computed-beliefs"] for output in scheduling_result["outputs"] + ) + n_recorded = sum( + output["num-recorded-beliefs"] for output in scheduling_result["outputs"] + ) + if n_computed == 0: + click.secho( + "The scheduler computed no beliefs, so no new schedule was stored.", + **MsgStyle.WARN, + ) else: + if dry_run: + reason = ", as this was a dry run" + elif n_recorded < n_computed: + reason = ", as the rest was already on record" + else: + reason = "" click.secho( - "The schedule above was computed but not stored (because of --dry-run).", + f"Scheduling done: computed {n_computed}, recorded {n_recorded}{reason}.", **MsgStyle.SUCCESS, ) @@ -2504,19 +2519,26 @@ def add_report( # noqa: C901 # The report was saved above, unless running in dry mode, so say what that save stored. if not dry_run: n_saved = saved[i]["n_rows"] - if n_saved: + n_computed = saved[i]["n_computed"] + if n_saved and n_saved == n_computed: + click.secho( + f"Success. Recorded {pluralize('belief', n_saved, include_count=True)} of the report for sensor `{sensor}` in the database.", + **MsgStyle.SUCCESS, + ) + elif n_saved: click.secho( - f"Success. Saved {pluralize('belief', n_saved, include_count=True)} of the report for sensor `{sensor}` to the database.", + f"Success. Recorded {n_saved} of the {pluralize('belief', n_computed, include_count=True)} computed for the report for sensor `{sensor}` in the database:" + " the rest was already on record or had no value.", **MsgStyle.SUCCESS, ) elif count_persistable_values(data): click.secho( - f"The report for sensor `{sensor}` repeats beliefs already on record, so nothing new was saved.", + f"The report for sensor `{sensor}` repeats beliefs already on record, so nothing new was recorded (computed {n_computed}, recorded 0).", **MsgStyle.SUCCESS, ) else: click.secho( - f"Nothing was saved for sensor `{sensor}`, as the report holds no values to save." + f"Nothing was recorded for sensor `{sensor}`, as the report holds no values to record (computed {n_computed}, recorded 0)." " This can happen when its inputs do not align on source and belief time.", **MsgStyle.WARN, ) diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 8dd8172da3..5f0d44255a 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -1991,10 +1991,20 @@ def test_run_report_automation_with_two_outputs_saves_both( ) assert (stored_sub.values.T == [0, 0, 0, 0, 0]).all() - assert job.return_value() == [ - {"sensor_id": report_sensor_id, "n_rows": 5}, - {"sensor_id": report_sensor_2_id, "n_rows": 5}, - ] + assert job.return_value() == { + "outputs": [ + { + "sensor": report_sensor_id, + "num-computed-beliefs": 5, + "num-recorded-beliefs": 5, + }, + { + "sensor": report_sensor_2_id, + "num-computed-beliefs": 5, + "num-recorded-beliefs": 5, + }, + ] + } import pandas as pd @@ -2160,7 +2170,7 @@ def compute(self, parameters=None): StubReporter(), {}, permitted_output_sensor_ids=None ) save.assert_called_once() - assert saved == [{"sensor_id": sensor1_id, "n_rows": 0}] + assert saved == [{"sensor_id": sensor1_id, "n_computed": 0, "n_rows": 0}] def test_report_job_persistence_failure_rolls_back_all_outputs( @@ -2249,7 +2259,7 @@ def test_a_rerun_report_job_counts_what_it_saved_and_says_why_it_saved_nothing( from flexmeasures.data.models.time_series import Sensor from flexmeasures.data.services.reporting import run_report_job - report_sensor_id = setup_dummy_data_fresh_db[2] + *_, report_sensor_id, report_sensor_2_id = setup_dummy_data_fresh_db report_sensor = fresh_db.session.get(Sensor, report_sensor_id) runner = app.test_cli_runner() freeze_server_now(datetime(2023, 4, 10, 10, 0, 30, tzinfo=timezone.utc)) @@ -2266,13 +2276,24 @@ def test_a_rerun_report_job_counts_what_it_saved_and_says_why_it_saved_nothing( ) first = run_report_job(**job.kwargs) - assert all(result["n_rows"] > 0 for result in first), first + assert first == { + "outputs": [ + {"sensor": sensor_id, "num-computed-beliefs": 5, "num-recorded-beliefs": 5} + for sensor_id in (report_sensor_id, report_sensor_2_id) + ] + } info = mocker.spy(app.logger, "info") warning = mocker.spy(app.logger, "warning") second = run_report_job(**job.kwargs) - assert [result["n_rows"] for result in second] == [0] * len(first) + # The rerun computed as much as the first run did, which tells it apart from a run that computed nothing. + assert second == { + "outputs": [ + {"sensor": sensor_id, "num-computed-beliefs": 5, "num-recorded-beliefs": 0} + for sensor_id in (report_sensor_id, report_sensor_2_id) + ] + } assert any( "already on record" in call.args[0] for call in info.call_args_list ), info.call_args_list @@ -2319,7 +2340,7 @@ def compute(self, parameters=None): beliefs_before = count_beliefs() _, saved = compute_and_save_report(StubReporter(), {}) - assert saved == [{"sensor_id": sensor.id, "n_rows": 3}] + assert saved == [{"sensor_id": sensor.id, "n_computed": 3, "n_rows": 3}] assert count_beliefs() == beliefs_before + 3, "the results are staged" # A caller that decides against its transaction takes the report down with it. diff --git a/flexmeasures/cli/tests/test_data_add_fresh_db.py b/flexmeasures/cli/tests/test_data_add_fresh_db.py index a8e0759a16..2c2f605641 100644 --- a/flexmeasures/cli/tests/test_data_add_fresh_db.py +++ b/flexmeasures/cli/tests/test_data_add_fresh_db.py @@ -1,6 +1,7 @@ import json import logging import os +import re from datetime import datetime import pandas as pd @@ -1232,7 +1233,7 @@ def test_add_report_saves_beliefs(app, fresh_db, setup_dummy_data_fresh_db, tmp_ ), ) check_command_ran_without_error(result) - assert "Saved 5 beliefs of the report for sensor" in result.output + assert "Recorded 5 beliefs of the report for sensor" in result.output assert _count_beliefs(fresh_db, report_sensor_id) == beliefs_before + 5 report_sensor = fresh_db.session.get(Sensor, report_sensor_id) @@ -1303,8 +1304,11 @@ def test_add_report_says_when_a_rerun_saves_nothing_new( rerun = runner.invoke(add_report, cli_input) check_command_ran_without_error(rerun) - assert "repeats beliefs already on record, so nothing new was saved" in rerun.output - assert "Success. Saved" not in rerun.output + assert ( + "repeats beliefs already on record, so nothing new was recorded (computed 5, recorded 0)" + in rerun.output + ) + assert "Success. Recorded" not in rerun.output assert _count_beliefs(fresh_db, report_sensor_id) == recorded @@ -1490,7 +1494,7 @@ def test_add_process( # ProcessScheduler's make_schedule() call returns a dict (not the boolean # True), which used to be falsy enough to silently suppress this message; # confirm the message still appears. - assert "New schedule is stored." in result.output + assert "Scheduling done: computed" in result.output process_power_sensor = db.session.get(Sensor, process_power_sensor_id) schedule = process_power_sensor.search_beliefs() @@ -1525,18 +1529,59 @@ def test_add_schedule_dry_run_saves_no_beliefs( result = runner.invoke(add_schedule, cli_input + ["--dry-run"]) check_command_ran_without_error(result) - assert "Not saving schedule for sensor `power`" in result.output + assert "Not recording schedule for sensor `power`" in result.output assert "covering events from" in result.output - assert "computed but not stored (because of --dry-run)" in result.output - assert "New schedule is stored." not in result.output sensor = fresh_db.session.get(Sensor, process_power_sensor) assert len(sensor.search_beliefs()) == 0 + # A dry run still says how much it computed, which a normal run then records in full. + n_computed = int( + re.search( + r"Scheduling done: computed (\d+), recorded 0, as this was a dry run\.", + result.output, + ).group(1) + ) + assert n_computed > 0 # A normal run does record the schedule, so the dry run really skipped that step result = runner.invoke(add_schedule, cli_input) check_command_ran_without_error(result) - assert "New schedule is stored." in result.output - assert len(sensor.search_beliefs()) > 0 + assert ( + f"Scheduling done: computed {n_computed}, recorded {n_computed}.\n" + in result.output + ) + assert len(sensor.search_beliefs()) == n_computed + + # Running it again computes the same schedule, which is already on record, so nothing new is saved. + result = runner.invoke(add_schedule, cli_input) + check_command_ran_without_error(result) + assert ( + f"Scheduling done: computed {n_computed}, recorded 0, as the rest was already on record." + in result.output + ) + assert len(sensor.search_beliefs()) == n_computed + + +def test_add_schedule_says_when_the_scheduler_computed_nothing( + app, fresh_db, process_power_sensor, add_market_prices_fresh_db, monkeypatch +): + """A schedule of no beliefs is said to be one, rather than reported as a new schedule stored with 0 beliefs.""" + from flexmeasures.cli import data_add + from flexmeasures.data.services.scheduling_result import SchedulingJobResult + + monkeypatch.setattr( + data_add, "make_schedule", lambda **kwargs: SchedulingJobResult().to_dict() + ) + runner = app.test_cli_runner() + cli_input = to_flags(_process_schedule_cli_input(fresh_db, process_power_sensor)) + + result = runner.invoke(data_add.add_schedule, cli_input) + + check_command_ran_without_error(result) + assert ( + "The scheduler computed no beliefs, so no new schedule was stored." + in result.output + ) + assert "Scheduling done" not in result.output def test_add_schedule_rejects_dry_run_as_job(app, fresh_db, add_market_prices_fresh_db): diff --git a/flexmeasures/data/automations.py b/flexmeasures/data/automations.py index 3d15af6b58..83df740c9a 100644 --- a/flexmeasures/data/automations.py +++ b/flexmeasures/data/automations.py @@ -402,12 +402,18 @@ def check_execution_access(automation, sensors): @record_outcome_on_automation_run -def execute_automation_job(automation_id: int, data_source_id: int, parameters: dict): - """Compute and persist declared results with provenance in a plugin-enabled worker.""" +def execute_automation_job( + automation_id: int, data_source_id: int, parameters: dict +) -> dict: + """Compute and persist declared results with provenance in a plugin-enabled worker. + + The job's result lists its outputs as a report job's result does (see `run_report_job`). + """ from flexmeasures.data.models.automations import Automation from flexmeasures.data.services.generator_results import ( check_generator_results, describe_generator, + outputs_of_job_result, save_generator_results, ) @@ -433,4 +439,4 @@ def execute_automation_job(automation_id: int, data_source_id: int, parameters: ) saved = save_generator_results(results) db.session.commit() - return saved + return {"outputs": outputs_of_job_result(saved)} diff --git a/flexmeasures/data/services/generator_results.py b/flexmeasures/data/services/generator_results.py index 96065a6408..99760e66fb 100644 --- a/flexmeasures/data/services/generator_results.py +++ b/flexmeasures/data/services/generator_results.py @@ -136,8 +136,9 @@ def save_generator_results(results: list[dict]) -> list[dict]: A result may name events whose beliefs are saved even so, under "save_even_if_unchanged" (their event starts), as a report does where its belief has to supersede a belief of another source recorded since (see `AggregatorReporter`). - :returns: per result, the sensor id and the number of beliefs saved ("n_rows"), - which leaves out NaN values and beliefs that were already on record. + :returns: per result, the sensor id, the number of beliefs computed ("n_computed") and the number of beliefs saved ("n_rows"). + The computed count is every belief the generator returned, including NaN values and beliefs that were already on record. + The saved count leaves both of those out, so a rerun that only repeats what is on record saves 0 of the beliefs it computed. """ saved = [] with db.session.begin_nested(): @@ -153,7 +154,13 @@ def save_generator_results(results: list[dict]) -> list[dict]: data[is_forced], save_changed_beliefs_only=False ) n_saved = n_changed + n_forced - saved.append({"sensor_id": result["sensor"].id, "n_rows": n_saved}) + saved.append( + { + "sensor_id": result["sensor"].id, + "n_computed": len(data), + "n_rows": n_saved, + } + ) return saved @@ -171,3 +178,19 @@ def save_results_not_yet_saved(results: list[dict]) -> None: return for result, saved in zip(unsaved, save_generator_results(unsaved)): result["n_saved"] = saved["n_rows"] + + +def outputs_of_job_result(saved: list[dict]) -> list[dict]: + """Turn what ``save_generator_results`` says about each result into the "outputs" a job result reports, as API clients read them. + + Each output names its sensor (by ID), the number of beliefs computed ("num-computed-beliefs") and the number of beliefs recorded ("num-recorded-beliefs"), + counted as ``save_generator_results`` counts them. + """ + return [ + { + "sensor": entry["sensor_id"], + "num-computed-beliefs": entry["n_computed"], + "num-recorded-beliefs": entry["n_rows"], + } + for entry in saved + ] diff --git a/flexmeasures/data/services/reporting.py b/flexmeasures/data/services/reporting.py index c8016ceaff..ab4f23cf95 100644 --- a/flexmeasures/data/services/reporting.py +++ b/flexmeasures/data/services/reporting.py @@ -15,6 +15,7 @@ from flexmeasures.data.services.generator_results import ( check_generator_results, describe_generator, + outputs_of_job_result, save_generator_results, ) @@ -130,7 +131,8 @@ def compute_and_save_report( and without committing, which is left to the caller. :returns: the computed results, and per result a summary of what was saved: - the sensor id and the number of beliefs saved, which leaves out NaN values and beliefs that were already on record. + the sensor id, the number of beliefs computed ("n_computed"), which counts every belief the report returned, + and the number of beliefs saved ("n_rows"), which leaves out NaN values and beliefs that were already on record. """ results = compute_report( reporter, parameters, permitted_output_sensor_ids, automation_id @@ -139,9 +141,12 @@ def compute_and_save_report( @record_outcome_on_automation_run -def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: +def run_report_job(data_source_id: int, parameters: dict) -> dict: """Compute and store a report in a reporting worker. + The job's result lists, under "outputs", one entry per output sensor: + the sensor (by ID), the number of beliefs computed ("num-computed-beliefs") and the number of beliefs recorded ("num-recorded-beliefs"). + If the report was triggered by an automation, the end of the report window is recorded upon success, so the automation's next default window starts where this one ended. A failed report job therefore leaves no permanent gap in the reported periods. @@ -175,17 +180,21 @@ def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: ) db.session.commit() + # Say both counts, as saving none of what was computed is not the same as computing nothing. summary = ", ".join( - f"{result['n_rows']} values on sensor {result['sensor_id']}" for result in saved + f"sensor {result['sensor_id']}: computed {result['n_computed']}, recorded {result['n_rows']}" + for result in saved ) if any(result["n_rows"] for result in saved): current_app.logger.info( - "Report by %s ran successfully, producing %s.", source, summary + "Report by %s ran successfully, recording the values that were not yet on record (%s).", + source, + summary, ) elif any(count_persistable_values(result["data"]) for result in results): - # Saving nothing is not the same as computing nothing: these values were all on record already. + # Saving nothing is not the same as computing nothing: the values it computed were on record already, or NaN. current_app.logger.info( - "Report by %s ran successfully, but every value it computed was already on record (%s).", + "Report by %s ran successfully, but recorded nothing new, as every value it computed was already on record or had no value (%s).", source, summary, ) @@ -209,4 +218,4 @@ def run_report_job(data_source_id: int, parameters: dict) -> list[dict]: automation_id, now=datetime.fromisoformat(parameters["end"]) ) - return saved + return {"outputs": outputs_of_job_result(saved)} diff --git a/flexmeasures/data/services/scheduling.py b/flexmeasures/data/services/scheduling.py index ea372fe546..cabc9e16c7 100644 --- a/flexmeasures/data/services/scheduling.py +++ b/flexmeasures/data/services/scheduling.py @@ -44,6 +44,7 @@ from flexmeasures.data.services.generator_results import ( check_generator_results, describe_generator, + outputs_of_job_result, save_generator_results, ) from flexmeasures.utils.time_utils import server_now @@ -971,10 +972,11 @@ def make_schedule( # noqa: C901 **scheduler_kwargs: dict, ) -> dict: """ - This function computes a schedule. It returns a dict, empty on schedulers - that don't (yet) produce further analysis. If the scheduler produced soft - state-of-charge constraint analysis (see ``SchedulingJobResult``), the dict - instead holds that analysis under ``unresolved`` and ``resolved`` keys. + This function computes a schedule. It returns a dict (see ``SchedulingJobResult``), + which lists, per output sensor, the beliefs the schedule computed and those it recorded (under ``outputs``), + and the total of the beliefs computed (``num-beliefs``). + If the scheduler produced soft state-of-charge constraint analysis, + the dict also holds that analysis under ``unresolved`` and ``resolved`` keys, which are empty lists otherwise. It can be queued as a job (see create_scheduling_job). In that case, it will probably run on a different FlexMeasures node than where the job is created. @@ -1100,6 +1102,8 @@ def make_schedule( # noqa: C901 ) scheduling_result_dict: dict = SchedulingJobResult().to_dict() to_save = [] + # On a dry run, nothing is saved, so the outputs say what was computed, and that none of it was saved. + dry_run_outputs = [] for result in consumption_schedule: if result.get("name") == SCHEDULING_RESULT_KEY: scheduling_result_dict = result["data"].to_dict() @@ -1145,6 +1149,9 @@ def make_schedule( # noqa: C901 if not dry_run: to_save.append({"sensor": result["sensor"], "data": bdf}) else: + dry_run_outputs.append( + {"sensor_id": result["sensor"].id, "n_computed": len(bdf), "n_rows": 0} + ) # Report what would have been saved, in the same terms as a forecast dry run does: # the sensor, the number of beliefs, and the events they cover. event_range = ( @@ -1153,14 +1160,20 @@ def make_schedule( # noqa: C901 else "" ) print( - f"\nNot saving schedule for sensor `{bdf.sensor}` (ID {bdf.sensor.id}) to the database (because of --dry-run)," + f"\nNot recording schedule for sensor `{bdf.sensor}` (ID {bdf.sensor.id}) in the database (because of --dry-run)," f" but this is what I computed ({len(bdf)} beliefs{event_range}):\n{bdf}" ) # Saved the way every data generator's results are: all or nothing, and counting the beliefs actually saved, - # which leaves out beliefs that repeat what is already on record. On a dry run nothing is saved, so the count is 0. + # which leaves out beliefs that repeat what is already on record. + # Each output also counts the beliefs computed, so that saving none does not read as computing none. saved = save_generator_results(to_save) if to_save else [] - scheduling_result_dict["num-beliefs"] = sum(entry["n_rows"] for entry in saved) + outputs = outputs_of_job_result(saved + dry_run_outputs) + scheduling_result_dict["outputs"] = outputs + # num-beliefs is the total of the beliefs computed, also on a dry run. + scheduling_result_dict["num-beliefs"] = sum( + output["num-computed-beliefs"] for output in outputs + ) if not dry_run: scheduler.persist_flex_model() diff --git a/flexmeasures/data/services/scheduling_result.py b/flexmeasures/data/services/scheduling_result.py index 4448e0cc81..75af51c9da 100644 --- a/flexmeasures/data/services/scheduling_result.py +++ b/flexmeasures/data/services/scheduling_result.py @@ -13,14 +13,18 @@ class SchedulingJobResult: The sensor schedule endpoint (``GET /api/v3_0/sensors//schedules/``) returns power values only and does not include constraint analysis. **Structure:** - Results contain three top-level fields: + Results contain four top-level fields: - ``unresolved``: List of soft constraints that the scheduler could not satisfy - Each entry is a dict with ``"asset"`` field (asset ID) and constraint-type keys (``"soc-minima"``, ``"soc-maxima"``) - Each constraint-type key holds a list of dicts, one per violated slot (chronologically ordered): ``{"datetime": ISO 8601 UTC, "violation": "X kWh"}`` - ``resolved``: List of soft constraints that were satisfied with available headroom - Each entry is a dict with ``"asset"`` field and constraint-type keys - Each constraint-type key holds a list of dicts, one per met slot (chronologically ordered): ``{"datetime": ISO 8601 UTC, "margin": "X kWh"}`` - - ``num-beliefs``: Total number of beliefs (scheduled values) saved to the database + - ``outputs``: List of the sensors the schedule records on, one entry per sensor + - Each entry is a dict with ``"sensor"`` field (sensor ID), ``"num-computed-beliefs"`` (the beliefs the scheduler computed for it), + and ``"num-recorded-beliefs"`` (those recorded, which leaves out beliefs that repeat what is already on record or have no value). + - On a dry run, nothing is recorded, so each entry's ``"num-recorded-beliefs"`` is 0. + - ``num-beliefs``: Total number of beliefs (scheduled values) computed, i.e. the total of ``"num-computed-beliefs"`` across ``outputs``. **Important:** ``soc-targets`` (hard constraints) are never included since they are strictly enforced by the scheduler. Only hard constraint failures cause job failure. @@ -45,6 +49,9 @@ class SchedulingJobResult: ], } ], + "outputs": [ + {"sensor": 5, "num-computed-beliefs": 96, "num-recorded-beliefs": 96}, + ], "num-beliefs": 96 } @@ -53,6 +60,7 @@ class SchedulingJobResult: unresolved: list = field(default_factory=list) resolved: list = field(default_factory=list) + outputs: list = field(default_factory=list) num_beliefs: int = 0 def to_dict(self) -> dict: @@ -60,6 +68,7 @@ def to_dict(self) -> dict: return { "unresolved": self.unresolved, "resolved": self.resolved, + "outputs": self.outputs, "num-beliefs": self.num_beliefs, } @@ -69,5 +78,6 @@ def from_dict(cls, d: dict) -> "SchedulingJobResult": return cls( unresolved=d.get("unresolved", []), resolved=d.get("resolved", []), + outputs=d.get("outputs", []), num_beliefs=d.get("num-beliefs", 0), ) diff --git a/flexmeasures/data/tests/test_generator_results_fresh_db.py b/flexmeasures/data/tests/test_generator_results_fresh_db.py index 5bd622d655..226c138c5a 100644 --- a/flexmeasures/data/tests/test_generator_results_fresh_db.py +++ b/flexmeasures/data/tests/test_generator_results_fresh_db.py @@ -158,12 +158,30 @@ def fail_on_second_save(data, **kwargs): def test_saving_results_counts_what_was_saved_and_repeats_nothing( fresh_db, two_outputs ): - """Each result says how many beliefs it saved, and saving the same results again saves none of them.""" + """Each result says how many beliefs it computed and how many it saved, and saving the same results again saves none of them. + + The computed count stays the same on the second save, which tells a rerun that repeats what is on record from one that computed nothing. + """ first = save_generator_results(two_outputs) fresh_db.session.commit() again = save_generator_results(two_outputs) fresh_db.session.commit() + assert [entry["n_computed"] for entry in first] == [3, 3] assert [entry["n_rows"] for entry in first] == [3, 3] + assert [entry["n_computed"] for entry in again] == [3, 3] assert [entry["n_rows"] for entry in again] == [0, 0] assert _beliefs_in_session(fresh_db) == 6 + + +def test_a_value_left_out_as_nan_counts_as_computed_but_not_as_saved( + fresh_db, two_outputs +): + """A NaN value the generator returned is counted as computed, but it is not saved, so it is not counted as saved.""" + data = two_outputs[0]["data"] + data.loc[data.index[0], "event_value"] = float("nan") + + saved = save_generator_results(two_outputs) + + assert [entry["n_computed"] for entry in saved] == [3, 3] + assert [entry["n_rows"] for entry in saved] == [2, 3] diff --git a/flexmeasures/data/tests/test_plugin_automations_fresh_db.py b/flexmeasures/data/tests/test_plugin_automations_fresh_db.py index 4b9463f56d..14f546c647 100644 --- a/flexmeasures/data/tests/test_plugin_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_plugin_automations_fresh_db.py @@ -209,7 +209,7 @@ def test_plugin_job_keeps_its_result_by_its_queue( def test_plugin_job_reports_rows_saved( fresh_db, app, ingestion_plugin, ingestion_assets ): - """A rerun whose beliefs are unchanged reports that it saved nothing.""" + """A rerun whose beliefs are unchanged reports that it saved nothing, while it computed as much as the first run.""" root, sensors = ingestion_assets automation = make_automation(fresh_db, root, sensors[0], parameters={"value": 3.0}) n_rows = [] @@ -222,8 +222,24 @@ def test_plugin_job_reports_rows_saved( assert job.is_finished, job.exc_info n_rows.append(job.return_value()) assert n_rows == [ - [{"sensor_id": sensors[0].id, "n_rows": 1}], - [{"sensor_id": sensors[0].id, "n_rows": 0}], + { + "outputs": [ + { + "sensor": sensors[0].id, + "num-computed-beliefs": 1, + "num-recorded-beliefs": 1, + } + ] + }, + { + "outputs": [ + { + "sensor": sensors[0].id, + "num-computed-beliefs": 1, + "num-recorded-beliefs": 0, + } + ] + }, ] diff --git a/flexmeasures/data/tests/test_scheduling_jobs.py b/flexmeasures/data/tests/test_scheduling_jobs.py index 74391c7fd1..ad28d906f8 100644 --- a/flexmeasures/data/tests/test_scheduling_jobs.py +++ b/flexmeasures/data/tests/test_scheduling_jobs.py @@ -22,6 +22,7 @@ create_scheduling_job, load_custom_scheduler, handle_scheduling_exception, + make_schedule, ) from flexmeasures.utils.unit_utils import ur from flexmeasures.utils.calculations import integrate_time_series @@ -425,6 +426,74 @@ def test_save_state_of_charge( ) +def test_schedule_result_lists_per_output_what_was_computed_and_saved( + fresh_db, app, smart_building +): + """A schedule's result lists, per output sensor, the beliefs computed and those recorded, and num-beliefs totals those computed. + + A heat buffer with a state-of-charge sensor records on two sensors: its power and its state of charge. + A dry run records none of what it computes, a first run records all of it, + and a rerun computes the same schedule, which is already on record, so it records none of it. + num-beliefs is the total of the beliefs computed, on each of these runs. + """ + assets, sensors, soc_sensors = smart_building + power_sensor = sensors["Test Heat Buffer"] + soc_sensor = soc_sensors["Test Heat Buffer"] + kwargs = dict( + asset_or_sensor={"id": power_sensor.id, "class": "Sensor"}, + start=pd.Timestamp("2015-01-03").tz_localize("Europe/Amsterdam"), + end=pd.Timestamp("2015-01-04").tz_localize("Europe/Amsterdam"), + belief_time=pd.Timestamp("2015-01-02").tz_localize("Europe/Amsterdam"), + resolution=timedelta(minutes=15), + scheduler_specs={ + "module": "flexmeasures.data.models.planning.storage", + "class": "StorageScheduler", + }, + flex_model={ + "power-capacity": "10kW", + "soc-at-start": "0kWh", + "soc-unit": "kWh", + "soc-min": 0.0, + "soc-max": "100kWh", + "soc-usage": ["10kW"], + "soc-gain": ["1kW"], + "state-of-charge": {"sensor": soc_sensor.id}, + }, + flex_context={ + "consumption-price": "100 EUR/MWh", + "production-price": "0 EUR/MWh", + "site-production-capacity": "1MW", + "site-consumption-capacity": "1MW", + }, + ) + + dry_run = make_schedule(dry_run=True, **kwargs) + first = make_schedule(**kwargs) + rerun = make_schedule(**kwargs) + + def counts(result: dict) -> dict: + return { + output["sensor"]: ( + output["num-computed-beliefs"], + output["num-recorded-beliefs"], + ) + for output in result["outputs"] + } + + computed = {sensor: n for sensor, (n, _) in counts(first).items()} + assert set(computed) == {power_sensor.id, soc_sensor.id} + assert all(n > 0 for n in computed.values()) + assert counts(dry_run) == {sensor: (n, 0) for sensor, n in computed.items()} + assert counts(first) == {sensor: (n, n) for sensor, n in computed.items()} + assert counts(rerun) == {sensor: (n, 0) for sensor, n in computed.items()} + total = sum(computed.values()) + assert [dry_run["num-beliefs"], first["num-beliefs"], rerun["num-beliefs"]] == [ + total, + total, + total, + ] + + def test_save_state_of_charge_percent_sensor( fresh_db, app, diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index dce443126e..751538e77c 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -5341,7 +5341,7 @@ "/api/v3_0/jobs/{uuid}": { "get": { "summary": "Get the status of a background job", - "description": "Look up a background job by its UUID and see whether it is\nqueued, running, finished, or failed.\n\nThe response includes a status message plus job metadata such\nas the queue name, function name, timestamps, and the job\nresult when available.\n\nFailed jobs also include traceback information when the worker\nstored it with the job result.\n\nFor a finished scheduling job, result is an object. For a\nStorageScheduler job it holds soft state-of-charge constraint\nanalysis: unresolved lists constraints the scheduler could not\nsatisfy, and resolved lists constraints that were satisfied\nwith some margin. Each device entry's soc-minima/soc-maxima\nvalue is a list, holding one entry per violated slot (for\nunresolved) or per met slot with its margin (for resolved),\nordered chronologically. Both arrays are empty when the flex model\ndefines no soc-minima/soc-maxima, or when a scheduler other\nthan StorageScheduler was used. The num-beliefs field holds\nthe total number of beliefs (scheduled values) saved to the database.\nThis is the only place constraint analysis is available \u2014 the sensor\nschedule endpoint (GET /api/v3_0/sensors//schedules/)\nreturns power values only.\n", + "description": "Look up a background job by its UUID and see whether it is queued, running, finished, or failed.\n\nThe response includes a status message plus job metadata,\nsuch as the queue name, function name, timestamps, and the job result when available.\n\nFailed jobs also include traceback information when the worker stored it with the job result.\n\nFor a finished scheduling job, result is an object.\nFor a StorageScheduler job, it holds soft state-of-charge constraint analysis:\nunresolved lists constraints the scheduler could not satisfy,\nand resolved lists constraints that were satisfied with some margin.\nEach device entry's soc-minima/soc-maxima value is a list,\nholding one entry per violated slot (for unresolved) or per met slot with its margin (for resolved), ordered chronologically.\nBoth arrays are empty when the flex model defines no soc-minima/soc-maxima, or when a scheduler other than StorageScheduler was used.\nThis is the only place constraint analysis is available:\nthe sensor schedule endpoint (GET /api/v3_0/sensors//schedules/) returns power values only.\nThe outputs field lists one entry per sensor the schedule records on:\nthe sensor (its ID), the number of beliefs (scheduled values) the scheduler computed for it (num-computed-beliefs),\nand the number recorded in the database (num-recorded-beliefs), which leaves out beliefs that were already on record or had no value.\nThe num-beliefs field holds the total of num-computed-beliefs across outputs.\n\nFor a finished reporting job, result is an object holding the same outputs field, with one entry per output sensor of the report.\n", "security": [ { "ApiKeyAuth": [] @@ -5386,7 +5386,7 @@ "description": "Human-readable description of the job status." }, "result": { - "description": "Return value of the job function, or null when not yet available. For a finished scheduling job, this is an object; a StorageScheduler job populates it with unresolved/resolved soft state-of-charge constraint analysis (empty arrays when the flex model defines no soc-minima/soc-maxima, or when a scheduler other than StorageScheduler was used). The num-beliefs field holds the total number of beliefs (scheduled values) saved to the database.\n", + "description": "Return value of the job function, or null when not yet available. For a finished scheduling job, this is an object; a StorageScheduler job populates it with unresolved/resolved soft state-of-charge constraint analysis (empty arrays when the flex model defines no soc-minima/soc-maxima, or when a scheduler other than StorageScheduler was used). Its outputs field lists, per sensor the schedule records on, the sensor (its ID), num-computed-beliefs and num-recorded-beliefs, and its num-beliefs field holds the total of num-computed-beliefs across outputs. For a finished reporting job, this is an object holding the same outputs field.\n", "nullable": true }, "func-name": { @@ -5491,6 +5491,13 @@ } ], "resolved": [], + "outputs": [ + { + "sensor": 5, + "num-computed-beliefs": 96, + "num-recorded-beliefs": 96 + } + ], "num-beliefs": 96 }, "func-name": "flexmeasures.data.services.scheduling.create_schedule",