diff --git a/documentation/changelog.rst b/documentation/changelog.rst index ec545fced8..c37aa75130 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -57,6 +57,7 @@ New features * In the UI, the full record of the data source selected on a sensor page can be inspected, backed by a new API endpoint (``[GET] /sources/(id)``) [see `PR #2290 `_] * Automations can also compute schedules on a recurring basis (``flexmeasures add automation --type schedules``), with the schedule start defaulting to each run's time [see `PR #2293 `_] * Automations can be created, edited and deleted in the UI and through new API endpoints (``[POST|PATCH|DELETE] /assets/(id)/automations``), by organisation admins and consultants, with their recurrence expressed in a selectable IANA timezone, and only involving sensors they can access themselves (read access to the sensors an automation reads, and permission to record data on the sensors it writes to) [see `PR #2294 `_] +* Reports can run as background jobs (``flexmeasures add report --as-job``, processed by workers of the new ``reporting`` queue) and be computed on a recurring basis by automations, with a rolling report window expressed as Pandas offsets or defaulting to the last cron period [see `PR #2297 `_] * ``flexmeasures show data-sources`` now shows which organisation a data source belongs to, and can list the sensors holding data recorded by a given source [see `PR #2401 `_] * The flex-context can now define multiple commodities, each specifying their own prices and grid capacities [see `PR #1946 `_, `PR #2172 `_, `PR #2235 `_, `PR #2271 `_, `PR #2355 `_ and `PR #2380 `_] * Support multiple feeders to a shared storage [see `PR #2001 `_, `PR #2321 `_, `PR #2322 `_, `PR #2325 `_ and `PR #2431 `_] diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index a796c9d4fa..e2c4a1283a 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -12,8 +12,9 @@ since v1.0.0 | August 11, 2026 * Add ``flexmeasures add plan``, ``flexmeasures show plans`` and ``flexmeasures edit plan``, to manage the rate limits and quotas which apply to the accounts on a plan. * Add ``flexmeasures edit secret`` to store an encrypted secret on an account or asset. * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset. -* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, computing forecasts or schedules). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. -* Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. Runs missed while the runner was down are caught up once, with several missed forecast runs coalesced into the latest useful forecast, and a run at a skipped or repeated daylight-saving-time hour happens exactly once. +* Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset, computing forecasts, schedules or reports). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. +* Add an ``--as-job`` flag to ``flexmeasures add report``, to queue a reporting job (processed by workers of the new ``reporting`` queue) instead of computing directly. +* Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. Runs missed while the runner was down are caught up once, with several missed runs coalesced into the latest useful one, and a run at a skipped or repeated daylight-saving-time hour happens exactly once. * ``flexmeasures delete sensor`` now warns which automations read from or write to a sensor before it is deleted, as an automation refers to its sensors by ID and would fail on its next run. * ``flexmeasures show data-sources`` now shows the account a data source belongs to, and lists the sensors holding data recorded by a single source with ``--show-sensors``. diff --git a/documentation/cli/commands.rst b/documentation/cli/commands.rst index 7d81a1e309..205b9c41d6 100644 --- a/documentation/cli/commands.rst +++ b/documentation/cli/commands.rst @@ -41,7 +41,7 @@ of which some are referred to in this documentation. ``flexmeasures add annotation`` Add annotation to accounts, assets and/or sensors. ``flexmeasures add toy-account`` Create a toy account, for tutorials and trying things. ``flexmeasures add report`` Create a report. -``flexmeasures add automation`` Add an automation: a recurring task (computing forecasts or schedules) on an asset, with its own cron timezone. +``flexmeasures add automation`` Add an automation: a recurring task (computing forecasts, schedules or reports) on an asset, with its own cron timezone. ================================================= ======================================= diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst index 7a441226e7..4dd8d30ded 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -3,65 +3,61 @@ Automations ============ -An **automation** is a recurring task defined on an asset. -For now, an automation computes forecasts; automating schedules and reports is planned. +Hosts and users often want the three main FlexMeasures features — :ref:`forecasting`, :ref:`scheduling` and :ref:`reporting` — to run on a recurring basis, across larger numbers of sites. +*Automations* make that a first-class concept: an automation is a recurring task defined on an asset, and each time it runs, it queues jobs. -On each run, the automation queues jobs (so make sure a worker is processing the ``forecasting`` queue, see :ref:`redis-queue`). -The parameters of the task were stored when the automation was created, and validated with the same schema that the CLI and API use. -Timing parameters are resolved on each run — for instance, the forecast start defaults to the time the automation runs, so each run produces fresh forecasts. +An automation consists of: -Creating an automation ----------------------- +- a **type**: ``forecasts``, ``schedules`` or ``reports``; +- a **recurrence**: a cron string (e.g. ``"0 6 * * *"`` for daily at 6 AM), interpreted in the automation's own IANA timezone; +- a **data generator** (for forecasts and reports): the forecaster or reporter class and its configuration, stored on a data source. + The data source stays the same across runs, so all results the automation produces attribute to one steady source; +- **parameters**: what to compute on each run, validated by the same schema the CLI and API use for one-off runs. + Timing parameters are resolved freshly on each run, so a recurring automation always computes fresh periods + (see the type-specific sections below for the exact rules); +- an **activation status**: only active automations run. -Here is how you create an automation in the CLI, asking for daily (at 6 AM) forecasts of sensor 12: +Managing automations +-------------------- -.. code-block:: bash - - flexmeasures add automation --asset 3 --name "Daily PV forecasts" --type forecasts \ - --cron "0 6 * * *" --timezone Europe/Amsterdam --sensor 12 - -``--type`` says what the automation computes, and defaults to ``forecasts``. -The remaining options are the ones the task itself needs: a forecast automation accepts everything `flexmeasures add forecast` accepts, such as ``--forecaster`` to pick the forecaster and ``--config`` to configure it (see :ref:`forecasting`). -The forecaster and its configuration are stored on a data source, so you can also pass ``--source`` to reuse the data source of an existing forecaster, in which case ``--forecaster`` and ``--config`` (and the individual configuration options) are not needed — the data source already determines them. -That data source is required while the automation exists, so it cannot be deleted until the automation is removed. - -The recurrence is defined by a standard five-field cron string (minute, hour, day of month, month, and day of week), which defaults to ``"0 0 * * *"`` (daily at midnight). -It is interpreted in the automation's IANA timezone. -If ``--timezone`` is omitted, the current ``FLEXMEASURES_TIMEZONE`` value is copied to the automation. -Changing that configuration later does not change existing automations. -Cron aliases and optional seconds or year fields are not supported. +Automations can be managed in three ways: -Automations are active by default (use ``--inactive`` to create them in deactivated state). -Use ``flexmeasures edit automation`` to rename, re-schedule (``--cron``), change the timezone, activate or deactivate an automation, and ``flexmeasures delete automation`` to remove one. -These changes are recorded in the asset's audit log. +- **CLI**: ``flexmeasures add automation``, ``flexmeasures edit automation`` (name, cron string, timezone and activation status) and ``flexmeasures delete automation``. +- **API**: list and inspect with ``[GET] /assets/(id)/automations`` and ``[GET] /assets/(id)/automations/(automation_id)``; + create, update and delete with ``[POST|PATCH|DELETE]`` on the same paths (see the `API documentation <../api/v3_0.html>`_). +- **UI**: each asset has an *Automations* page (in the breadcrumbs dropdown), with a tab per automation type. + It lists each automation's recurrence and recent job counts, and lets you create, edit, (de)activate and delete automations. -For forecast automations, the sensor on which forecasts are saved (``sensor-to-save``, falling back to ``sensor``) must belong to the automation's asset or one of its descendants. -This relationship is checked both when the automation is created and immediately before each run. +Creating, updating and deleting automations requires account admin or consultant rights, and is recorded in the asset's audit log. Running automations -------------------- +-------------------- -For automations to actually run, let a cron job execute the following command once per minute: +An automation is due whenever its cron string matches the current minute in its configured timezone. To actually run due automations, let a cron job execute the following command once per minute: .. code-block:: bash * * * * * flexmeasures jobs run-automations -Each due automation then queues its jobs. +Each due automation then queues its jobs — so make sure workers are processing the relevant queues (``forecasting``, ``scheduling`` and/or ``reporting``, see :ref:`redis-queue`). +Each scheduled run receives at most one automatic queueing attempt, so the command is safe to run more than once within a minute. +If the process crashes, or queueing fails after creating some jobs, that run is not retried automatically, because a retry could duplicate partial work. + If the runner misses runs, because it was down or overloaded, it catches up when it resumes: it queues only the latest missed run of each automation, rather than replaying stale ones. -Timing parameters that default to the run time are resolved when that catch-up run is queued, so it produces a current forecast. +Timing parameters that default to the run time are resolved when that catch-up run is queued, so it produces a current result. -Each scheduled run receives at most one automatic queueing attempt. -If the process crashes, or queueing fails after creating some jobs, that run is not retried automatically, because a retry could duplicate partial work. +Jobs record how they were created (via the CLI, the API or an automation), which is shown in the *Created Via* column +of the jobs table on the asset's status page, where recent jobs are listed. -The jobs record how they were created, which is shown on the asset's status page (UI), where recent jobs are listed. +Automating each feature +----------------------- -Viewing automations -------------------- +The parameters stored on an automation follow the same schemas as one-off CLI/API calls, with type-specific rules for resolving timing on each run: -Automations defined on an asset can be viewed on the asset's *Automations* page in the UI, and listed with the API endpoint `[GET] /assets/(id)/automations <../api/v3_0.html#get--api-v3_0-assets-id-automations>`_. -An automation's details show the sensors it reads from and writes to, linking to each sensor's page. -Conversely, a sensor's page lists the automations that write data to it. +- :ref:`automating_forecasts` — forecast parameters; the forecast start defaults to the run time. +- :ref:`automating_schedules` — a schedule trigger message; omit ``start`` to schedule from the run time. +- :ref:`automating_reports` — report parameters; use ``start-offset``/``end-offset`` (Pandas offsets) for a rolling window, + or omit timing fields to report on the period since the last successfully covered report window. .. _automation_cursor: diff --git a/documentation/features/forecasting.rst b/documentation/features/forecasting.rst index 32991bcd58..11e03eae45 100644 --- a/documentation/features/forecasting.rst +++ b/documentation/features/forecasting.rst @@ -219,6 +219,20 @@ Usage: Automating forecasts -------------------- -Instead of asking for forecasts one at a time, you can set up an *automation*: a recurring task defined on an asset, which queues forecasting jobs on a cron schedule. -See :ref:`automations`. -Schedules can be automated in the same way — see :ref:`automating_schedules`. +Instead of asking for forecasts one at a time, you can set up an *automation*: a recurring task defined on an asset (see :ref:`automations` for the full concept, including how to manage and run automations). +On each run, the automation queues forecasting jobs (so make sure a worker is processing the ``forecasting`` queue, see :ref:`redis-queue`). +When the automation was created, its forecast parameters (see above) were stored, and validated with the same schema that the CLI and API use. +Timing parameters are resolved on each run — for instance, the forecast start defaults to the time the automation runs, so each run produces fresh forecasts. +The sensor on which forecasts are saved (``sensor-to-save``, falling back to ``sensor``) must belong to the automation's asset or one of its descendants. +This relationship is checked both when the automation is created and immediately before each run. + +Here is how you create a forecast automation in the CLI, asking for daily (at 6 AM) forecasts of sensor 12: + +.. code-block:: bash + + flexmeasures add automation --asset 3 --name "Daily PV forecasts" --type forecasts \ + --cron "0 6 * * *" --timezone Europe/Amsterdam --sensor 12 + +A forecast automation accepts everything ``flexmeasures add forecast`` accepts, such as ``--forecaster`` to pick the forecaster and ``--config`` to configure it. +The forecaster and its configuration are stored on a data source, so you can also pass ``--source`` to reuse the data source of an existing forecaster, in which case ``--forecaster`` and ``--config`` (and the individual configuration options) are not needed — the data source already determines them. +That data source is required while the automation exists, so it cannot be deleted until the automation is removed. diff --git a/documentation/features/reporting.rst b/documentation/features/reporting.rst index 9576b035bd..15764192d9 100644 --- a/documentation/features/reporting.rst +++ b/documentation/features/reporting.rst @@ -122,4 +122,29 @@ The input sensor stores the power/energy flow, and the output sensor will store Here, the ``ProfitOrLossReporter`` used as source (with Id 6) is the one we configured above. With the offsets, we control the timing ― we indicate that we want the new report to encompass the day of tomorrow (see Pandas offset strings). -The report sensor will now store all costs which we know will be made tomorrow by the schedule. \ No newline at end of file +The report sensor will now store all costs which we know will be made tomorrow by the schedule. + +.. _automating_reports: + +Automating reports +-------------------- + +Reports can be queued as background jobs (add ``--as-job`` to ``flexmeasures add report``, and let a worker process the ``reporting`` queue, see :ref:`redis-queue`), +and computed on a recurring basis by an *automation* defined on the asset (see :ref:`automations` for the full concept, including how to manage and run automations). + +The reporter and its configuration are stored on a data source (steady across runs, so all report results attribute to the same source), +while the report parameters are stored on the automation itself and their timing is resolved freshly on each run: + +- Use ``start-offset`` and/or ``end-offset`` fields (comma-separated Pandas offsets, like the CLI options above) for a rolling window relative to the claimed cron occurrence, + in the timezone of the first output sensor. For instance, ``"start-offset": "-1D,DB"`` with ``"end-offset": "DB"`` reports on the whole previous day. +- Omit timing fields entirely to report from the end of the latest successfully completed report window through the claimed cron occurrence. + When no completed window is known, such as on the first run, the start falls back to the previous cron occurrence in the automation's timezone. + The completion marker only moves forward, so concurrent reporting workers that finish out of order cannot reopen an already covered period. +- Absolute ``start``/``end`` fields are also accepted, but draw a warning, as each run would then compute the same period. + +For example, this automation computes a report over each past day, every morning at 1 AM: + +.. code-block:: bash + + flexmeasures add automation --asset 3 --name "Daily aggregation report" --cron "0 1 * * *" --type reports \ + --reporter PandasReporter --config reporter-config.yml --parameters report-parameters.yml diff --git a/documentation/host/queues.rst b/documentation/host/queues.rst index 5454061615..337afc75a0 100644 --- a/documentation/host/queues.rst +++ b/documentation/host/queues.rst @@ -24,7 +24,7 @@ Here is how to run one worker for each kind of job (in separate terminals): .. code-block:: bash - $ flexmeasures jobs run-worker --name our-only-worker --queue forecasting|scheduling|ingestion + $ flexmeasures jobs run-worker --name our-only-worker --queue forecasting|scheduling|ingestion|reporting Running multiple workers in parallel might be a great idea. diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index f3be3ae146..7465beadd4 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -1661,7 +1661,7 @@ def post_automation(self, id: int, asset: GenericAsset): automation_type=automation_data["type"], active=automation_data["active"], parameters=automation_data["parameters"], - forecaster_class=automation_data["forecaster"], + generator_class=automation_data["generator"], config=automation_data["config"], origin="API", check_permissions=True, diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 854bd88954..e9c3c9a856 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -216,6 +216,170 @@ def test_post_automation( fresh_db.session.flush() +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_automation_with_foreign_sensor( + app, + db, + setup_accounts, + add_battery_assets, + requesting_user, +): + """Referencing a sensor outside the caller's reach is forbidden.""" + from datetime import timedelta + + from flexmeasures.data.models.generic_assets import GenericAsset + from flexmeasures.data.models.time_series import Sensor + + battery = add_battery_assets["Test battery"] + foreign_asset = GenericAsset( + name="Foreign asset", + generic_asset_type=battery.generic_asset_type, + owner=setup_accounts["Dummy"], + ) + foreign_sensor = Sensor( + "foreign power", + generic_asset=foreign_asset, + event_resolution=timedelta(minutes=15), + unit="MW", + ) + db.session.add(foreign_sensor) + db.session.flush() + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Sneaky forecasts", + "cronstr": "0 6 * * *", + "type": "forecasts", + "parameters": {"sensor": foreign_sensor.id}, + }, + ) + assert response.status_code == 403 + assert ( + db.session.execute( + select(Automation).filter_by(name="Sneaky forecasts") + ).scalar_one_or_none() + is None + ) + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_report_automation_with_foreign_config_sensor( + app, + db, + setup_accounts, + add_battery_assets, + requesting_user, +): + """Reporter configuration may not read a sensor outside the caller's reach.""" + from flexmeasures.data.models.generic_assets import GenericAsset + + battery = add_battery_assets["Test battery"] + foreign_asset = GenericAsset( + name="Foreign price asset", + generic_asset_type=battery.generic_asset_type, + owner=setup_accounts["Dummy"], + ) + foreign_price_sensor = Sensor( + "private foreign price", + generic_asset=foreign_asset, + event_resolution=timedelta(hours=1), + unit="EUR/MWh", + ) + report_sensor = Sensor( + "profit report", + generic_asset=battery, + event_resolution=timedelta(hours=1), + unit="EUR", + ) + db.session.add_all([foreign_price_sensor, report_sensor]) + db.session.flush() + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Cross-organisation profit report", + "cronstr": "0 1 * * *", + "type": "reports", + "generator": "ProfitOrLossReporter", + "config": { + "consumption_price_sensor": foreign_price_sensor.id, + }, + "parameters": { + "input": [{"sensor": battery.sensors[0].id}], + "output": [{"sensor": report_sensor.id}], + }, + }, + ) + + assert response.status_code == 403 + assert foreign_price_sensor.name not in response.text + assert ( + db.session.execute( + select(Automation).filter_by(name="Cross-organisation profit report") + ).scalar_one_or_none() + is None + ) + db.session.commit() + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True +) +def test_post_report_automation_rejects_output_outside_asset_subtree( + app, + db, + add_battery_assets, + requesting_user, +): + """Report output must stay on the automation asset or a descendant.""" + battery = add_battery_assets["Test battery"] + sibling_battery = add_battery_assets["Test small battery"] + report_sensor = Sensor( + "sibling report output", + generic_asset=sibling_battery, + event_resolution=timedelta(hours=1), + unit="MW", + ) + db.session.add(report_sensor) + db.session.flush() + + with app.test_client() as client: + response = client.post( + url_for("AssetAPI:post_automation", id=battery.id), + json={ + "name": "Misplaced report output", + "cronstr": "0 1 * * *", + "type": "reports", + "generator": "PandasReporter", + "config": { + "required_input": [{"name": "flow"}], + "required_output": [{"name": "copied_flow"}], + "transformations": [ + { + "df_input": "flow", + "df_output": "copied_flow", + "method": "copy", + } + ], + }, + "parameters": { + "input": [{"name": "flow", "sensor": battery.sensors[0].id}], + "output": [{"name": "copied_flow", "sensor": report_sensor.id}], + }, + }, + ) + + assert response.status_code == 422 + assert "must belong to asset" in response.text + db.session.commit() + + @pytest.mark.parametrize( "requesting_user", ["test_prosumer_user_2@seita.nl"], indirect=True ) diff --git a/flexmeasures/app.py b/flexmeasures/app.py index c95f9e73e0..05023fd3b2 100644 --- a/flexmeasures/app.py +++ b/flexmeasures/app.py @@ -124,7 +124,11 @@ def create( # noqa C901 name="ingestion", default_timeout=get_job_timeout("ingestion", app.config, app.logger), ), - # reporting=Queue(connection=redis_conn, name="reporting"), + reporting=Queue( + connection=redis_conn, + name="reporting", + default_timeout=get_job_timeout("reporting", app.config, app.logger), + ), # labelling=Queue(connection=redis_conn, name="labelling"), # alerting=Queue(connection=redis_conn, name="alerting"), ) diff --git a/flexmeasures/cli/data_add.py b/flexmeasures/cli/data_add.py index 39f26cd572..c2cf22b215 100755 --- a/flexmeasures/cli/data_add.py +++ b/flexmeasures/cli/data_add.py @@ -1456,7 +1456,8 @@ def _assemble_forecaster_config_and_parameters( config = _load_yaml_mapping(config_file, "--config") for field_name, field in TrainPredictPipelineConfigSchema._declared_fields.items(): field_value = kwargs.pop(field_name, None) - if field_value is not None: + # skip unset options (click passes None, or an empty tuple for multiple-value options) + if field_value is not None and field_value != (): if field_name in { "future_regressors", "past_regressors", @@ -1505,8 +1506,8 @@ def _assemble_forecaster_config_and_parameters( if kebab_key not in parameters: parameters[kebab_key] = v - # Drop None values - parameters = {k: v for k, v in parameters.items() if v is not None} + # Drop unset values + parameters = {k: v for k, v in parameters.items() if v is not None and v != ()} return config, parameters @@ -1729,20 +1730,28 @@ def add_forecast( # noqa: C901 " Defaults to TrainPredictPipeline. Use the command `flexmeasures show forecasters` to list all the available forecasters." " Cannot be combined with --source, which already determines the forecaster.", ) +@click.option( + "--reporter", + "reporter_class", + required=False, + type=click.STRING, + help="Reporter class registered in flexmeasures.data.models.reporting or in an available flexmeasures plugin (only used for --type reports)." + " Use the command `flexmeasures show reporters` to list all the available reporters.", +) @click.option( "--source", "source", required=False, type=DataSourceIdField(), - help="DataSource ID of the `Forecaster`. The forecaster class and its configuration are read from" - " the data source's data generator attributes, so --forecaster and --config are not needed (or allowed) with it.", + help="DataSource ID of the data generator (`Forecaster` or `Reporter`). The generator class and its configuration are read from" + " the data source's attributes, so --forecaster/--reporter and --config are not needed (or allowed) with it.", ) @click.option( "--config", "config_file", required=False, type=click.File("r"), - help="Path to the JSON or YAML file with the configuration of the forecaster." + help="Path to the JSON or YAML file with the configuration of the forecaster or reporter." " Cannot be combined with --source, which already determines the configuration.", ) @click.option( @@ -1751,7 +1760,8 @@ def add_forecast( # noqa: C901 required=False, type=click.File("r"), help="Path to the JSON or YAML file with the parameters used on each run of the automation:" - " forecast parameters for --type forecasts, or a schedule trigger message for --type schedules.", + " forecast parameters for --type forecasts, a schedule trigger message for --type schedules," + " or report parameters for --type reports.", ) @add_cli_options_from_schema( ForecasterParametersSchema(), hidden=True, force_optional=True @@ -1767,13 +1777,14 @@ def add_automation( automation_type: str, inactive: bool = False, forecaster_class: str | None = None, + reporter_class: str | None = None, source: DataSource | None = None, config_file: TextIOBase | None = None, parameters_file: TextIOBase | None = None, **kwargs, ): """ - Add an automation: a recurring task (computing forecasts or schedules) on an asset. + Add an automation: a recurring task (computing forecasts, schedules or reports) on an asset. \b Examples @@ -1782,12 +1793,18 @@ def add_automation( --parameters forecast-parameters.yml flexmeasures add automation --asset 3 --name "Hourly schedules" --cron "0 * * * *" --type schedules --parameters trigger-message.yml + flexmeasures add automation --asset 3 --name "Daily self-consumption report" + --cron "0 1 * * *" --type reports --reporter PandasReporter + --config reporter-config.yml --parameters report-parameters.yml - For forecasts, the forecaster configuration is stored on a data source, and - the forecast parameters are validated and stored on the automation itself. + For forecasts and reports, the data generator configuration is stored on a + data source, and the parameters are validated and stored on the automation itself. For schedules, the parameters form a schedule trigger message (as accepted by the [POST] /assets/(id)/schedules/trigger API endpoint, without the asset id); omit its "start" field to schedule from the run time on each run. + For reports, use "start-offset"/"end-offset" (comma-separated Pandas offsets, + applied to the run time) for a rolling report window, or omit timing fields + entirely to report on the last cron period. Each time the automation runs, jobs are queued (see `flexmeasures jobs run-automations`). Alternatively, pass an existing data source (--source) to reuse the forecaster @@ -1835,7 +1852,9 @@ def add_automation( automation_type=automation_type, active=not inactive, parameters=parameters, - forecaster_class=forecaster_class, + generator_class=( + reporter_class if automation_type == "reports" else forecaster_class + ), config=config, source=source, origin="CLI", @@ -2145,6 +2164,12 @@ def add_schedule( # noqa C901 is_flag=True, help="Add this flag to save the `config` in the attributes of the DataSource for future reference.", ) +@click.option( + "--as-job", + is_flag=True, + help="Whether to queue a reporting job instead of computing directly. " + "To process the job, run a worker (on any computer, but configured to the same databases) to process the 'reporting' queue. Defaults to False.", +) def add_report( # noqa: C901 reporter_class: str, source: DataSource | None = None, @@ -2161,11 +2186,26 @@ def add_report( # noqa: C901 edit_parameters: bool = False, save_config: bool = False, timezone: str | None = None, + as_job: bool = False, ): """ Create a new report using the Reporter class and save the results to the database or export them as CSV or Excel file. """ + if as_job and (dry_run or output_file_pattern): + click.secho( + "The --as-job flag cannot be combined with --dry-run or --output-file:" + " the job saves the report to the database only.", + **MsgStyle.ERROR, + ) + raise click.Abort() + if as_job and not save_config: + # the worker rebuilds the reporter from its data source, so the config must be stored there + click.secho( + "Saving the reporter config to its data source (required for --as-job).", + **MsgStyle.WARN, + ) + save_config = True config = dict() @@ -2268,6 +2308,16 @@ def add_report( # noqa: C901 if ("resolution" not in parameters) and (resolution is not None): parameters["resolution"] = pd.Timedelta(resolution).isoformat() + reporter.set_job_trigger("CLI") + + if as_job: + returns = reporter.compute(as_job=True, parameters=parameters) + click.secho( + f"Created reporting job {returns['job_id']} (the report will be saved to the database once processed).", + **MsgStyle.SUCCESS, + ) + return + click.echo("Report computation is running...") # compute the report diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index 195fed1fb2..02afd78d13 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -106,11 +106,15 @@ def run_automations(): ) continue try: - returns = run_automation(automation) - n_jobs = returns.get("n_jobs") if returns else 0 - queue_name = {"forecasts": "forecasting", "schedules": "scheduling"}.get( - automation.type, automation.type + returns = run_automation( + automation, scheduled_at=due_automation.scheduled_at ) + n_jobs = returns.get("n_jobs") if returns else 0 + queue_name = { + "forecasts": "forecasting", + "schedules": "scheduling", + "reports": "reporting", + }.get(automation.type, automation.type) click.secho( f"Automation {automation.id} ('{automation.name}') queued {n_jobs} {queue_name} job(s) for asset {automation.asset_id}.", **MsgStyle.SUCCESS, @@ -431,7 +435,7 @@ def inspect_job(job_id: str): "--queue", default=None, required=True, - help="State which queue(s) to work on (using '|' as separator), e.g. 'forecasting', 'scheduling', 'ingestion' or 'forecasting|scheduling'.", + help="State which queue(s) to work on (using '|' as separator), e.g. 'forecasting', 'scheduling', 'ingestion', 'reporting' or 'forecasting|scheduling'.", ) @click.option( "--name", @@ -450,7 +454,7 @@ def inspect_job(job_id: str): ) def run_worker(queue: str, name: str | None, with_scheduler: bool): """ - Start a worker process for forecasting, scheduling and/or ingestion jobs. + Start a worker process for forecasting, scheduling, ingestion and/or reporting jobs. We use the app context to find out which redis queues to use. """ diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 874d273ac4..0dab845fd9 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -2,6 +2,7 @@ import json import pytest +import yaml import pytz from types import SimpleNamespace @@ -1004,6 +1005,263 @@ class FakeJob: assert calls["kwargs"]["end"] - start == timedelta(hours=12) +def test_prepare_report_parameters(app): + """Report start/end resolve per run: from Pandas offsets, or defaulting to the last cron period.""" + import pandas as pd + + from flexmeasures.data.services.automations import prepare_report_parameters + from flexmeasures.utils.time_utils import get_timezone + + now = pd.Timestamp("2026-07-11T14:00:00+02:00") + # without an output sensor, offsets resolve in the platform timezone + local_now = now.tz_convert(get_timezone()) + + # default: the last cron period (hourly cron -> the previous hour) + message = prepare_report_parameters({}, "0 * * * *", now=now) + assert pd.Timestamp(message["start"]) == now - pd.Timedelta(hours=1) + assert pd.Timestamp(message["end"]) == now + + # The fallback cron period is interpreted in the automation timezone and + # ends at the claimed run rather than at a delayed runner's wall time. + scheduled_at = datetime(2026, 1, 1, 16, 0, tzinfo=timezone.utc) + message = prepare_report_parameters( + {}, + "0 1 * * *", + now=datetime(2026, 1, 2, 0, 30, tzinfo=timezone.utc), + cron_timezone="Asia/Seoul", + scheduled_at=scheduled_at, + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp("2025-12-31T16:00:00+00:00") + assert pd.Timestamp(message["end"]) == pd.Timestamp(scheduled_at) + + # A cron run in Amsterdam's spring gap is canonicalized to 03:00, + # while its report starts at the prior day's real 02:30 run. + spring_run = datetime(2026, 3, 29, 1, 0, tzinfo=timezone.utc) + message = prepare_report_parameters( + {}, + "30 2 * * *", + cron_timezone="Europe/Amsterdam", + scheduled_at=spring_run, + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp("2026-03-28T01:30:00+00:00") + assert pd.Timestamp(message["end"]) == pd.Timestamp(spring_run) + + # with a known actual last run, the window starts there instead + app.redis_connection.set("automation-last-run:1234", "2026-07-11T09:30:00+02:00") + try: + message = prepare_report_parameters( + {}, "0 * * * *", now=now, automation_id=1234 + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp( + "2026-07-11T09:30:00+02:00" + ) + assert pd.Timestamp(message["end"]) == now + # an unknown automation id still falls back to the last cron period + message = prepare_report_parameters( + {}, "0 * * * *", now=now, automation_id=5678 + ) + assert pd.Timestamp(message["start"]) == now - pd.Timedelta(hours=1) + finally: + app.redis_connection.delete("automation-last-run:1234") + + # offsets applied to the run time; "DB" floors to the day begin + message = prepare_report_parameters( + {"start-offset": "-1D,DB", "end-offset": "DB"}, "0 1 * * *", now=now + ) + assert ( + pd.Timestamp(message["start"]) == (local_now - pd.Timedelta(days=1)).normalize() + ) + assert pd.Timestamp(message["end"]) == local_now.normalize() + assert "start-offset" not in message and "end-offset" not in message + + # absolute datetimes pass through untouched + message = prepare_report_parameters( + {"start": "2026-01-01T00:00:00+01:00", "end": "2026-01-02T00:00:00+01:00"}, + "0 1 * * *", + now=now, + ) + assert pd.Timestamp(message["start"]) == pd.Timestamp("2026-01-01T00:00:00+01:00") + assert pd.Timestamp(message["end"]) == pd.Timestamp("2026-01-02T00:00:00+01:00") + + +def test_report_coverage_cannot_move_backwards(app, clean_redis): + """An older report finishing later may not reopen already covered periods.""" + from flexmeasures.data.services.automations import ( + get_automation_last_run, + record_automation_run, + ) + + later_end = datetime(2026, 1, 3, tzinfo=timezone.utc) + older_end = datetime(2026, 1, 2, tzinfo=timezone.utc) + + assert record_automation_run(42, later_end) is True + assert record_automation_run(42, older_end) is False + assert get_automation_last_run(42) == later_end + + +def _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + parameters_extra=None, + asset_id=1, +): + """CLI input for a report automation using a simple PandasReporter aggregation.""" + reporter_config = dict( + required_input=[{"name": "sensor_1"}, {"name": "sensor_2"}], + required_output=[{"name": "df_agg"}], + transformations=[ + dict( + df_input="sensor_1", + method="add", + args=["@sensor_2"], + df_output="df_agg", + ), + dict(method="resample_events", args=["2h"]), + ], + ) + parameters = dict( + input=[ + dict(name="sensor_1", sensor=sensor1_id), + dict(name="sensor_2", sensor=sensor2_id), + ], + output=[dict(name="df_agg", sensor=report_sensor_id)], + **(parameters_extra or {}), + ) + config_file = tmp_path / "reporter_config.yml" + config_file.write_text(yaml.dump(reporter_config)) + parameters_file = tmp_path / "parameters.yml" + parameters_file.write_text(yaml.dump(parameters)) + return [ + "--asset", str(asset_id), + "--name", "Aggregation report", + "--cron", "0 1 * * *", + "--type", "reports", + "--reporter", "PandasReporter", + "--config", str(config_file), + "--parameters", str(parameters_file), + ] # fmt: skip + + +def test_add_report_automation(app, fresh_db, setup_dummy_data, tmp_path): + """Create a reports automation; the reporter config lands on a data source.""" + from flexmeasures.cli.data_add import add_automation + + sensor1_id, sensor2_id, report_sensor_id, _ = setup_dummy_data + from flexmeasures.data.models.time_series import Sensor + + report_sensor = fresh_db.session.get(Sensor, report_sensor_id) + runner = app.test_cli_runner() + result = runner.invoke( + add_automation, + _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + parameters_extra={"start-offset": "-1D,DB", "end-offset": "DB"}, + asset_id=report_sensor.generic_asset_id, + ), + ) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute(select(Automation)).scalar_one() + assert automation.type == "reports" + assert automation.generator is not None + assert automation.generator.model == "PandasReporter" + assert automation.parameters["start-offset"] == "-1D,DB" + + # a reports automation without a reporter is rejected + result = runner.invoke( + add_automation, + [ + "--asset", "1", + "--name", "No reporter", + "--cron", "0 1 * * *", + "--type", "reports", + ], + ) # fmt: skip + assert result.exit_code != 0 + assert "reporter is required" in result.output + + # invalid time offsets are rejected (they would otherwise be silently skipped at run time) + result = runner.invoke( + add_automation, + _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + parameters_extra={ + "start-offset": "P1D,DB" + }, # ISO duration, not a Pandas offset + ), + ) + assert result.exit_code != 0 + assert "Invalid start-offset" in result.output + + +def test_run_report_automation(app, fresh_db, setup_dummy_data, clean_redis, tmp_path): + """A due reports automation queues a reporting job; a worker computes and saves the report.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_automations + from flexmeasures.data.models.time_series import Sensor + from flexmeasures.utils.job_utils import work_on_rq + + sensor1_id, sensor2_id, report_sensor_id, _ = setup_dummy_data + report_sensor = fresh_db.session.get(Sensor, report_sensor_id) + runner = app.test_cli_runner() + cli_input = _report_automation_cli_input( + tmp_path, + sensor1_id, + sensor2_id, + report_sensor_id, + # the dummy data lives in April 2023, so use an absolute reporting window + parameters_extra={ + "start": "2023-04-10T00:00:00+00:00", + "end": "2023-04-10T10:00:00+00:00", + }, + asset_id=report_sensor.generic_asset_id, + ) + cli_input[cli_input.index("0 1 * * *")] = "* * * * *" # due every minute + result = runner.invoke(add_automation, cli_input) + assert "Successfully created" in result.output, result.output + automation = fresh_db.session.execute(select(Automation)).scalar_one() + + result = runner.invoke(run_automations) + assert result.exit_code == 0, result.output + assert "queued 1 reporting job(s)" in result.output, result.output + + # the queued job recorded how it was created + jobs = app.queues["reporting"].jobs + assert len(jobs) == 1 + assert jobs[0].meta["trigger"] == { + "origin": "automation", + "automation_id": automation.id, + } + + # the covered-until anchor is only recorded once the job succeeds + assert not app.redis_connection.get(f"automation-last-run:{automation.id}") + + # process the job and check the report got saved + work_on_rq(app.queues["reporting"]) + report_sensor = fresh_db.session.get(Sensor, report_sensor_id) + stored_report = report_sensor.search_beliefs( + event_starts_after="2023-04-10T00:00:00+00:00", + event_ends_before="2023-04-10T10:00:00+00:00", + ) + assert (stored_report.values.T == [1, 2 + 3, 4 + 5, 6 + 7, 8 + 9]).all() + + # the successful job recorded the end of the report window as covered + import pandas as pd + + covered_until = app.redis_connection.get(f"automation-last-run:{automation.id}") + assert covered_until is not None + assert pd.Timestamp(covered_until.decode()) == pd.Timestamp( + "2023-04-10T10:00:00+00:00" + ) + + def test_run_automations(app, fresh_db, setup_dummy_data, clean_redis): """Active automations due this minute queue forecasting jobs (with trigger meta data); inactive ones do not. @@ -1043,6 +1301,9 @@ def test_run_automations(app, fresh_db, setup_dummy_data, clean_redis): and job.meta["trigger"]["automation_id"] in automation_ids for job in jobs ) + # the run got recorded (used e.g. to anchor default report windows) + for automation in automations: + assert app.redis_connection.get(f"automation-last-run:{automation.id}") # running again within the same minute does not queue jobs twice n_jobs = len(jobs) result = runner.invoke(run_automations) @@ -1123,7 +1384,7 @@ def test_failed_automation_attempt_is_not_retried(app, clean_redis, mocker): ) mocker.patch("flexmeasures.cli.jobs.claim_due_automation", return_value=True) - def queue_then_fail(_automation): + def queue_then_fail(_automation, **_kwargs): app.queues["forecasting"].enqueue("flexmeasures.utils.time_utils.server_now") raise RuntimeError("failed after queueing") diff --git a/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py b/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py index 9afd0fd28e..374db91662 100644 --- a/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py +++ b/flexmeasures/data/migrations/versions/c63896a97a8e_merge_the_automation_timezone_and_.py @@ -1,6 +1,6 @@ """merge the automation timezone and schedule generator migrations -Two migrations branched off the same revision: one adding an automation's timezone and scheduling cursor, +Two migrations branched off the same revision: one adding an automation's timezone and cursor, the other allowing a schedule automation to exist without a data generator. They touch different columns, so this merge only rejoins them and has nothing of its own to do. diff --git a/flexmeasures/data/migrations/versions/d2a4f6b8c901_require_generators_for_report_automations.py b/flexmeasures/data/migrations/versions/d2a4f6b8c901_require_generators_for_report_automations.py new file mode 100644 index 0000000000..9726544e7d --- /dev/null +++ b/flexmeasures/data/migrations/versions/d2a4f6b8c901_require_generators_for_report_automations.py @@ -0,0 +1,33 @@ +"""Require generators for forecast and report automations. + +Revision ID: d2a4f6b8c901 +Revises: c63896a97a8e +Create Date: 2026-08-12 02:20:00.000000 + +""" + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "d2a4f6b8c901" +down_revision = "c63896a97a8e" +branch_labels = None +depends_on = None + + +def upgrade(): + op.drop_constraint("forecast_generator", "automation", type_="check") + op.create_check_constraint( + "automation_generator", + "automation", + "type NOT IN ('forecasts', 'reports') OR generator_id IS NOT NULL", + ) + + +def downgrade(): + op.drop_constraint("automation_generator", "automation", type_="check") + op.create_check_constraint( + "forecast_generator", + "automation", + "type != 'forecasts' OR generator_id IS NOT NULL", + ) diff --git a/flexmeasures/data/models/automations.py b/flexmeasures/data/models/automations.py index 11ed0b48bf..3d2515f6a8 100644 --- a/flexmeasures/data/models/automations.py +++ b/flexmeasures/data/models/automations.py @@ -41,12 +41,12 @@ class Automation(db.Model, AuthModelMixin): __tablename__ = "automation" __table_args__ = ( db.CheckConstraint( - "type != 'forecasts' OR generator_id IS NOT NULL", - name="forecast_generator", + "type NOT IN ('forecasts', 'reports') OR generator_id IS NOT NULL", + name="automation_generator", ), ) - SUPPORTED_TYPES = ["forecasts", "schedules"] # later also "reports" + SUPPORTED_TYPES = ["forecasts", "schedules", "reports"] id = db.Column(db.Integer, autoincrement=True, primary_key=True) created_at = db.Column( diff --git a/flexmeasures/data/models/reporting/__init__.py b/flexmeasures/data/models/reporting/__init__.py index 56ffb265fa..d9152a54c5 100644 --- a/flexmeasures/data/models/reporting/__init__.py +++ b/flexmeasures/data/models/reporting/__init__.py @@ -19,14 +19,45 @@ class Reporter(DataGenerator): _parameters_schema = ReporterParametersSchema() _config_schema = ReporterConfigSchema() - def _compute(self, check_output_resolution=True, **kwargs) -> list[dict[str, Any]]: + @property + def input_sensors(self) -> list: + """The sensors from which the report reads its input data.""" + parameters = self._parameters or {} + return self._resolve_sensors( + [ + input_description.get("sensor") + for input_description in parameters.get("input", []) + ] + ) + + @property + def output_sensors(self) -> list: + """The sensors on which the report records its results.""" + parameters = self._parameters or {} + return self._resolve_sensors( + [ + output_description.get("sensor") + for output_description in parameters.get("output", []) + ] + ) + + def _compute( + self, check_output_resolution=True, as_job: bool = False, **kwargs + ) -> list[dict[str, Any]] | dict[str, Any]: """This method triggers the creation of a new report. The same object can generate multiple reports with different start, end, resolution and belief_time values. :param check_output_resolution: If True, checks each output for whether the event_resolution matches that of the sensor it is supposed to be recorded on. + :param as_job: If True, a job to compute (and save) the report is queued instead, + and a dict like {"job_id": , "n_jobs": 1} is returned. """ + if as_job: + from flexmeasures.data.services.reporting import create_reporting_job + + job = create_reporting_job(self) + return {"job_id": job.id, "n_jobs": 1} results = self._compute_report(**kwargs) diff --git a/flexmeasures/data/models/reporting/profit.py b/flexmeasures/data/models/reporting/profit.py index 3503bdcc69..6b3f550f6a 100644 --- a/flexmeasures/data/models/reporting/profit.py +++ b/flexmeasures/data/models/reporting/profit.py @@ -47,6 +47,15 @@ class ProfitOrLossReporter(Reporter): weights: dict method: str + @property + def input_sensors(self) -> list: + """The flow input and price sensors read to compute profit or loss.""" + return self._resolve_sensors( + super().input_sensors, + self._config.get("consumption_price_sensor"), + self._config.get("production_price_sensor"), + ) + def _compute_report( self, start: datetime, diff --git a/flexmeasures/data/schemas/automations.py b/flexmeasures/data/schemas/automations.py index 94481b74cf..e42279f323 100644 --- a/flexmeasures/data/schemas/automations.py +++ b/flexmeasures/data/schemas/automations.py @@ -87,15 +87,20 @@ class AutomationCreationSchema(Schema): ) active = fields.Bool(load_default=True) parameters = fields.Dict(keys=fields.Str(), load_default=dict) - forecaster = fields.Str( - load_default="TrainPredictPipeline", - metadata={"description": "Forecaster class (only used for type 'forecasts')."}, + generator = fields.Str( + load_default=None, + allow_none=True, + metadata={ + "description": "Data generator class, e.g. a forecaster (defaults to TrainPredictPipeline)" + " or a reporter (required for type 'reports', e.g. PandasReporter)." + " Not used for type 'schedules'." + }, ) config = fields.Dict( keys=fields.Str(), load_default=dict, metadata={ - "description": "Forecaster configuration (only used for type 'forecasts')." + "description": "Data generator configuration (only used for types 'forecasts' and 'reports')." }, ) diff --git a/flexmeasures/data/schemas/reporting/__init__.py b/flexmeasures/data/schemas/reporting/__init__.py index 02e580ae42..1b72571198 100644 --- a/flexmeasures/data/schemas/reporting/__init__.py +++ b/flexmeasures/data/schemas/reporting/__init__.py @@ -29,7 +29,9 @@ class ReporterParametersSchema(Schema): validate=validate.Length(min=1), ) - output = fields.List(fields.Nested(Output()), validate=validate.Length(min=1)) + output = fields.List( + fields.Nested(Output()), required=True, validate=validate.Length(min=1) + ) start = AwareDateTimeField(required=True) end = AwareDateTimeField(required=True) diff --git a/flexmeasures/data/schemas/reporting/aggregation.py b/flexmeasures/data/schemas/reporting/aggregation.py index 165e646fb5..e3c37c55d5 100644 --- a/flexmeasures/data/schemas/reporting/aggregation.py +++ b/flexmeasures/data/schemas/reporting/aggregation.py @@ -56,5 +56,5 @@ class AggregatorParametersSchema(ReporterParametersSchema): # redefining output to restrict the output length to 1 output = fields.List( - fields.Nested(Output()), validate=validate.Length(min=1, max=1) + fields.Nested(Output()), required=True, validate=validate.Length(min=1, max=1) ) diff --git a/flexmeasures/data/schemas/reporting/profit.py b/flexmeasures/data/schemas/reporting/profit.py index 2cae5069b9..8444082ba9 100644 --- a/flexmeasures/data/schemas/reporting/profit.py +++ b/flexmeasures/data/schemas/reporting/profit.py @@ -95,7 +95,9 @@ class ProfitOrLossReporterParametersSchema(ReporterParametersSchema): """ # redefining output to restrict the input length to 1 - input = fields.List(fields.Nested(Input()), validate=validate.Length(min=1, max=1)) + input = fields.List( + fields.Nested(Input()), required=True, validate=validate.Length(min=1, max=1) + ) @validates("input") def validate_input_measures_power_energy(self, value, **kwargs): diff --git a/flexmeasures/data/schemas/tests/test_reporting.py b/flexmeasures/data/schemas/tests/test_reporting.py index 337f8884d3..f01f7c9b78 100644 --- a/flexmeasures/data/schemas/tests/test_reporting.py +++ b/flexmeasures/data/schemas/tests/test_reporting.py @@ -199,6 +199,22 @@ def test_profit_reporter_config_schema(config, is_valid, db, app, setup_dummy_se }, True, ), + ( # missing required input + { + "output": [{"sensor": 3}], + "start": start, + "end": end, + }, + False, + ), + ( # missing required output + { + "input": [{"sensor": 4}], + "start": start, + "end": end, + }, + False, + ), ( # wrong output unit { "input": [{"sensor": 4}], # unit: MW diff --git a/flexmeasures/data/scripts/data_gen.py b/flexmeasures/data/scripts/data_gen.py index 047e95039d..73efdedcd9 100644 --- a/flexmeasures/data/scripts/data_gen.py +++ b/flexmeasures/data/scripts/data_gen.py @@ -468,6 +468,7 @@ def depopulate_prognoses( if not sensor: num_forecasting_jobs_deleted = app.queues["forecasting"].empty() num_scheduling_jobs_deleted = app.queues["scheduling"].empty() + num_reporting_jobs_deleted = app.queues["reporting"].empty() # Clear all forecasts (data with positive horizon) query = delete(TimedBelief).filter(TimedBelief.belief_horizon > timedelta(hours=0)) @@ -480,6 +481,7 @@ def depopulate_prognoses( if not sensor: click.echo("Deleted %d Forecast Jobs" % num_forecasting_jobs_deleted) click.echo("Deleted %d Schedule Jobs" % num_scheduling_jobs_deleted) + click.echo("Deleted %d Report Jobs" % num_reporting_jobs_deleted) click.echo("Deleted %d forecasts (ex-ante beliefs)" % num_forecasts_deleted) diff --git a/flexmeasures/data/services/automations.py b/flexmeasures/data/services/automations.py index e33fe047ae..71555a495b 100644 --- a/flexmeasures/data/services/automations.py +++ b/flexmeasures/data/services/automations.py @@ -14,6 +14,8 @@ from croniter import croniter from croniter.croniter import CroniterError import isodate +import pandas as pd +import pytz from isodate.isoerror import ISO8601Error from flask import current_app from marshmallow import ValidationError @@ -21,7 +23,7 @@ from werkzeug.exceptions import Forbidden -from flexmeasures import Forecaster +from flexmeasures import Forecaster, Reporter from flexmeasures.auth.policy import check_access from flexmeasures.data import db from flexmeasures.data.models.automations import ( @@ -33,7 +35,7 @@ asset_and_ancestor_ids, asset_is_in_subtree, ) -from flexmeasures.utils.time_utils import server_now +from flexmeasures.utils.time_utils import apply_offset_chain, get_timezone, server_now @dataclass(frozen=True) @@ -443,8 +445,8 @@ def resolve_schedule_automation_sensors( def resolve_automation_sensors(automation: Automation) -> dict[str, list[Sensor]]: """Work out which sensors an automation reads from and writes to on each run. - Forecast sensors are derived from the data generator, while schedule sensors are - derived from the same prepared trigger message used to queue the scheduling job. + Forecast and report sensors are derived from the data generator, while schedule sensors + are derived from the same prepared trigger message used to queue the scheduling job. Raises `AutomationSensorsUnknown` if that cannot be done, e.g. because a forecast 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). @@ -465,9 +467,17 @@ def resolve_automation_sensors(automation: Automation) -> dict[str, list[Sensor] ) try: data_generator = automation.generator.data_generator + parameters = dict(automation.parameters or {}) + if automation.type == "reports": + parameters = prepare_report_parameters( + parameters, + automation.cronstr, + automation_id=automation.id, + cron_timezone=automation.timezone, + ) return resolve_data_generator_sensors( data_generator, - data_generator._parameters_schema.load(dict(automation.parameters or {})), + data_generator._parameters_schema.load(parameters), ) except (NotImplementedError, ValidationError) as e: raise AutomationSensorsUnknown( @@ -495,7 +505,7 @@ def get_automations_feeding_sensor(sensor: Sensor) -> list[Automation]: 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 + (see `validate_automation_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. @@ -547,6 +557,194 @@ def prepare_schedule_trigger_message(parameters: dict, asset_id: int) -> dict: return message +def validate_offset_chain(offset_chain: str): + """Raise a ValueError on any offset that apply_offset_chain would silently skip. + + Valid offsets are Pandas offset strings, plus "DB" (day begin) and "HB" (hour begin). + """ + from pandas.tseries.frequencies import to_offset + + for offset in str(offset_chain).split(","): + offset = offset.strip() + if offset.lower() in ("db", "hb"): + continue + try: + to_offset(offset) + except ValueError: + raise ValueError( + f"'{offset}' is not a valid Pandas offset string (nor 'DB'/'HB')." + ) + + +def _last_run_redis_key(automation_id: int) -> str: + return f"automation-last-run:{automation_id}" + + +def record_automation_run(automation_id: int, now: datetime | None = None) -> bool: + """Remember (in Redis) until when this automation's work is covered. + + For forecasts and schedules automations, this is the (enqueue) run time. + For reports automations, the reporting job records the end of the report window + instead, upon success (see run_report_job), so a failed report job does not + create a permanent gap in the reported periods. + """ + from redis.exceptions import WatchError + + if now is None: + now = server_now() + candidate = floor_to_minute(now) + key = _last_run_redis_key(automation_id) + connection = current_app.redis_connection + while True: + with connection.pipeline() as pipeline: + try: + pipeline.watch(key) + value = pipeline.get(key) + if value: + if isinstance(value, bytes): + value = value.decode() + try: + current = floor_to_minute(datetime.fromisoformat(value)) + except ValueError: + current = None + if current is not None and current >= candidate: + pipeline.unwatch() + return False + pipeline.multi() + pipeline.set(key, candidate.isoformat()) + pipeline.execute() + return True + except WatchError: + # Another worker updated the coverage after our read. Re-read it + # and only advance from the new value. + continue + + +def get_automation_last_run(automation_id: int) -> datetime | None: + """Until when this automation's work is covered, if known (the record lives in Redis).""" + from flask import current_app + + value = current_app.redis_connection.get(_last_run_redis_key(automation_id)) + if not value: + return None + if isinstance(value, bytes): + value = value.decode() + try: + return datetime.fromisoformat(value) + except ValueError: + return None + + +def prepare_report_parameters( + parameters: dict, + cronstr: str, + now: datetime | None = None, + automation_id: int | None = None, + cron_timezone: str | None = None, + scheduled_at: datetime | None = None, +) -> dict: + """Complete stored report parameters into a message for the ReporterParametersSchema. + + The (required) start and end of the report are resolved on each run: + + - "start-offset" and "end-offset" fields hold comma-separated Pandas offsets + (e.g. "-1D,DB" for the start of the previous day), applied to the run time + (or to the given absolute start/end), in the timezone of the first output sensor. + - Without offsets or absolutes, the window runs since the end of the automation's + last (successfully) covered window, falling back to the last cron period (from + the previous cron fire time until the run time) when none is known (e.g. on the + first run). + """ + message = dict(parameters) + if scheduled_at is None: + scheduled_at = now if now is not None else server_now() + scheduled_at = floor_to_minute(scheduled_at) + + # Compute the run time in the timezone local to the first output sensor + # (matching `flexmeasures add report`), falling back to the platform timezone. + tz = get_timezone() + outputs = message.get("output") or [] + if ( + outputs + and isinstance(outputs[0], dict) + and outputs[0].get("sensor") is not None + ): + from flexmeasures.data.models.time_series import Sensor + + try: + output_sensor = db.session.get(Sensor, int(outputs[0]["sensor"])) + except (TypeError, ValueError): + output_sensor = None + if output_sensor is not None: + tz = pytz.timezone(output_sensor.timezone) + now = scheduled_at.astimezone(tz) + + start_offset = message.pop("start-offset", None) + end_offset = message.pop("end-offset", None) + start = pd.Timestamp(message["start"]) if "start" in message else None + end = pd.Timestamp(message["end"]) if "end" in message else None + + # Apply offsets to the given absolute datetime, or to the run time + if start_offset is not None: + start = apply_offset_chain( + start if start is not None else pd.Timestamp(now), start_offset + ) + if end_offset is not None: + end = apply_offset_chain( + end if end is not None else pd.Timestamp(now), end_offset + ) + + # Default to the window since the last covered window's end, falling back to + # the last cron period (from the previous cron fire time until the run time) + if start is None: + last_run = ( + get_automation_last_run(automation_id) + if automation_id is not None + else None + ) + if last_run is not None: + start = last_run + else: + cron_tz = ( + ZoneInfo(cron_timezone) + if cron_timezone is not None + else ZoneInfo(str(get_timezone())) + ) + nominal_scheduled_at = _as_nominal_wall_time( + scheduled_at.astimezone(cron_tz) + ) + previous_nominal = croniter(cronstr, nominal_scheduled_at).get_prev( + datetime + ) + start = _canonical_run_time(previous_nominal, cron_tz) + # A skipped wall time can canonicalize to the first valid instant after + # the gap, which may be the current run. Step back once more so + # the first report still covers a non-empty cron period. + if start >= scheduled_at: + previous_nominal = croniter(cronstr, previous_nominal).get_prev( + datetime + ) + start = _canonical_run_time(previous_nominal, cron_tz) + if end is None: + end = now + + message["start"] = pd.Timestamp(start).isoformat() + message["end"] = pd.Timestamp(end).isoformat() + return message + + +def _relevant_sensor_ids(automation: Automation, parameter_values: list) -> set[int]: + """The asset's sensor ids, plus any (castable) sensor ids among the given parameter values.""" + sensor_ids = {sensor.id for sensor in automation.asset.sensors} + for value in parameter_values: + if value is not None: + try: + sensor_ids.add(int(value)) + except (TypeError, ValueError): + pass + return sensor_ids + + def get_automations_involving_sensor(sensor: Sensor) -> list[Automation]: """Find the automations that read from or write to the given sensor. @@ -572,27 +770,40 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: Note that jobs in Redis have a limited TTL, so this only counts fairly recent jobs. """ - # Determine the job cache entries to scan. + # 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) - # and under individual sensors (per-device jobs). - assets = [automation.asset, *automation.asset.offspring] + # and under individual device sensors (per-device jobs), which may belong + # to child assets rather than the automation's own (site) asset. + sensor_ids = _relevant_sensor_ids( + automation, + [ + entry.get("sensor") + for entry in parameters.get("flex-model", []) or [] + if isinstance(entry, dict) + ], + ) cache_refs = [(automation.asset_id, "scheduling", "asset")] + [ - (sensor.id, "scheduling", "sensor") - for asset in assets - for sensor in asset.sensors + (sensor_id, "scheduling", "sensor") for sensor_id in sensor_ids ] + elif automation.type == "reports": + sensor_ids = _relevant_sensor_ids( + automation, + [ + output.get("sensor") + for output in parameters.get("output", []) or [] + if isinstance(output, dict) + ], + ) + cache_refs = [(sensor_id, "reporting", "sensor") for sensor_id in sensor_ids] else: - # Forecasting jobs are cached under the forecast target sensor(s), - # which may belong to a different asset than the automation's own asset. - sensor_ids = {sensor.id for sensor in automation.asset.sensors} - for key in ("sensor", "sensor-to-save"): - value = (automation.parameters or {}).get(key) - if value is not None: - try: - sensor_ids.add(int(value)) - except (TypeError, ValueError): - pass + 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] counts: dict[str, int] = {} @@ -608,6 +819,101 @@ def get_automation_job_stats(automation: Automation) -> dict[str, int]: return counts +def _prepare_forecast_automation( + asset, parameters: dict, generator_class: str | None, config: dict | None, source +) -> tuple[int, list[str]]: + """Validate forecast automation parameters and set up the forecaster's data source.""" + from flexmeasures.data.models.time_series import Sensor + from flexmeasures.data.schemas.forecasting.pipeline import ( + ForecasterParametersSchema, + ) + from flexmeasures.data.services.data_sources import get_data_generator + + warnings = [] + deserialized_parameters = ForecasterParametersSchema().load(parameters) + sensor = deserialized_parameters.get("sensor") + if isinstance(sensor, Sensor) and sensor.generic_asset_id != asset.id: + warnings.append( + f"The sensor to forecast ({sensor.id}) does not belong to asset {asset.id}." + ) + forecaster = get_data_generator( + source=source, + model=generator_class or "TrainPredictPipeline", + config=config or {}, + save_config=True, + data_generator_type=Forecaster, + ) + if forecaster is None: + raise ValueError(f"Could not set up forecaster '{generator_class}'.") + generator = ( + forecaster.data_source + ) # looks up or creates the data source storing the forecaster config + db.session.flush() + return generator.id, warnings + + +def _prepare_schedule_automation(asset, parameters: dict) -> tuple[None, list[str]]: + """Validate schedule automation parameters (the scheduler's data source is resolved at job time).""" + from flexmeasures.data.schemas.scheduling import AssetTriggerSchema + + warnings = [] + AssetTriggerSchema().load(prepare_schedule_trigger_message(parameters, asset.id)) + if "start" in parameters: + warnings.append( + "The schedule 'start' is fixed, so each run will compute the same period." + " Omit 'start' to schedule from the run time instead." + ) + return None, warnings + + +def _prepare_report_automation( + parameters: dict, + cronstr: str, + generator_class: str | None, + config: dict | None, + source, +) -> tuple[Reporter, dict, list[str]]: + """Validate report automation parameters without creating a data source.""" + from marshmallow import ValidationError + + from flexmeasures.data.services.data_sources import get_data_generator + + warnings = [] + if generator_class is None and source is None: + raise ValidationError( + "A reporter is required for report automations (e.g. PandasReporter)." + ) + for offset_field in ("start-offset", "end-offset"): + if offset_field in parameters: + try: + validate_offset_chain(parameters[offset_field]) + except ValueError as e: + raise ValidationError(f"Invalid {offset_field}: {e}") + reporter = get_data_generator( + source=source, + model=generator_class, + config=config or {}, + save_config=True, + data_generator_type=Reporter, + ) + if reporter is None: + raise ValueError(f"Could not set up reporter '{generator_class}'.") + # Validate with the chosen reporter's own parameters schema, + # which may extend the base ReporterParametersSchema. + deserialized_parameters = reporter._parameters_schema.load( + prepare_report_parameters(parameters, cronstr) + ) + if ( + "start" in parameters or "end" in parameters + ) and "start-offset" not in parameters: + warnings.append( + "The report period is (partly) fixed, so each run may compute the same period." + " Use 'start-offset'/'end-offset' (Pandas offsets applied to the run time)," + " or omit timing fields to report on the period since the last run instead." + ) + return reporter, deserialized_parameters, warnings + + def create_automation( asset, name: str, @@ -616,7 +922,7 @@ def create_automation( automation_type: str = "forecasts", active: bool = True, parameters: dict | None = None, - forecaster_class: str = "TrainPredictPipeline", + generator_class: str | None = None, config: dict | None = None, source=None, origin: str = "API", @@ -624,7 +930,7 @@ def create_automation( ) -> 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. + For forecasts and reports, the data generator config is stored on a data source. An audit log record is added to the asset. :param check_permissions: whether to require that the current user may read the @@ -633,27 +939,25 @@ def create_automation( created by a user (through the API or the UI); the CLI runs without a user, and is trusted. :raises marshmallow.ValidationError: if the parameters are invalid. - :raises ValueError: if the forecaster cannot be set up. + :raises ValueError: if the data generator cannot be set up. :raises werkzeug.exceptions.Forbidden: if a sensor is not accessible to the user. :returns: the automation and a list of warnings. """ from marshmallow import ValidationError from flexmeasures.data.models.audit_log import AssetAuditLog - from flexmeasures.data.models.time_series import Sensor parameters = parameters or {} warnings: list[str] = [] generator_id = None - forecaster = None + data_generator = None input_sensors: list[Sensor] = [] output_sensors: list[Sensor] = [] - forecast_output_sensor: Sensor | None = None if automation_type == "forecasts": + from flexmeasures.data.services.data_sources import get_data_generator from flexmeasures.data.schemas.forecasting.pipeline import ( ForecasterParametersSchema, ) - from flexmeasures.data.services.data_sources import get_data_generator deserialized_parameters = ForecasterParametersSchema().load(parameters) sensor = deserialized_parameters.get("sensor") @@ -663,13 +967,14 @@ def create_automation( ) forecaster = get_data_generator( source=source, - model=forecaster_class, + model=generator_class or "TrainPredictPipeline", config=config or {}, save_config=True, data_generator_type=Forecaster, ) if forecaster is None: - raise ValueError(f"Could not set up forecaster '{forecaster_class}'.") + raise ValueError(f"Could not set up forecaster '{generator_class}'.") + data_generator = forecaster # A forecast reads the history of the sensor to forecast, plus its regressors, # and records the forecast on the sensor to save to (the same sensor by default). @@ -679,7 +984,6 @@ def create_automation( ) input_sensors = forecast_sensors["input_sensors"] output_sensors = forecast_sensors["output_sensors"] - forecast_output_sensor = output_sensors[0] if output_sensors else None elif automation_type == "schedules": # A schedule is recorded on the sensors that the scheduler returns its results # for, and reads whatever other sensors the flex-model and flex-context refer to @@ -692,6 +996,16 @@ def create_automation( "The schedule 'start' is fixed, so each run will compute the same period." " Omit 'start' to schedule from the run time instead." ) + elif automation_type == "reports": + reporter, deserialized_parameters, warnings = _prepare_report_automation( + parameters, cronstr, generator_class, config, source + ) + data_generator = reporter + report_sensors = resolve_data_generator_sensors( + reporter, deserialized_parameters + ) + input_sensors = report_sensors["input_sensors"] + output_sensors = report_sensors["output_sensors"] else: raise ValidationError( f"Automation type '{automation_type}' is not supported (supported types: {Automation.SUPPORTED_TYPES})." @@ -702,13 +1016,14 @@ def create_automation( # Only once the sensors are known to be the user's to involve do we say anything about them, # so that this does not reveal where a sensor sits to someone who may not read it. - if forecast_output_sensor is not None: - validate_forecast_output_scope(asset.id, forecast_output_sensor) + if automation_type in ("forecasts", "reports"): + for output_sensor in output_sensors: + validate_automation_output_scope(asset.id, output_sensor, automation_type) - if forecaster is not None: - # Look up or create the data source storing the forecaster config only now that the automation is going ahead, + if data_generator is not None: + # Look up or create the data source storing the generator config only now that the automation is going ahead, # so that a refused request leaves nothing behind, whatever the caller does with the session afterwards. - generator = forecaster.data_source + generator = data_generator.data_source db.session.flush() generator_id = generator.id @@ -812,27 +1127,40 @@ def get_forecast_output_sensor(parameters: dict[str, Any]) -> Sensor: return sensor -def validate_forecast_output_scope(asset_id: int, output_sensor: Sensor) -> None: - """Require forecast output on the automation asset or a descendant.""" +def validate_automation_output_scope( + asset_id: int, output_sensor: Sensor, automation_type: str +) -> None: + """Require generated output on the automation asset or a descendant.""" if not asset_is_in_subtree(asset_id, output_sensor.generic_asset_id): raise ValueError( - f"Forecast automation output sensor {output_sensor.id} must belong to asset " + f"{automation_type.capitalize()} automation output sensor {output_sensor.id} must belong to asset " f"{asset_id} or one of its descendants." ) -def run_automation(automation: Automation) -> dict[str, Any] | None: +def run_automation( + automation: Automation, scheduled_at: datetime | None = None +) -> dict[str, Any] | None: """Queue the jobs for one run of an automation. :returns: a dict like {"job_id": , "n_jobs": }. """ + now = server_now() if automation.type == "forecasts": - return _run_forecast_automation(automation) + returns = _run_forecast_automation(automation) elif automation.type == "schedules": - return _run_schedule_automation(automation) - raise NotImplementedError( - f"Automations of type '{automation.type}' cannot be run yet." - ) + returns = _run_schedule_automation(automation) + elif automation.type == "reports": + # NB the reporting job itself records the end of the report window upon + # success (see run_report_job), so failed jobs do not create gaps in the + # reported periods. + return _run_report_automation(automation, now=now, scheduled_at=scheduled_at) + else: + raise NotImplementedError( + f"Automations of type '{automation.type}' cannot be run yet." + ) + record_automation_run(automation.id, now=now) + return returns def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: @@ -848,13 +1176,51 @@ def _run_forecast_automation(automation: Automation) -> dict[str, Any] | None: f"Data source {automation.generator_id} of automation {automation.id} does not store a Forecaster." ) output_sensor = get_forecast_output_sensor(automation.parameters or {}) - validate_forecast_output_scope(automation.asset_id, output_sensor) + validate_automation_output_scope( + automation.asset_id, output_sensor, automation.type + ) # Wipe any parameter state the copy inherited from a previous run. forecaster._parameters = None forecaster.set_job_trigger("automation", automation_id=automation.id) return forecaster.compute(as_job=True, parameters=dict(automation.parameters)) +def _run_report_automation( + automation: Automation, + now: datetime | None = None, + scheduled_at: datetime | None = None, +) -> dict[str, Any] | None: + if automation.generator is None: + raise ValueError( + f"Automation {automation.id} has no data generator to run (generator_id is not set)." + ) + reporter = automation.generator.data_generator + if not isinstance(reporter, Reporter): + raise ValueError( + f"Data source {automation.generator_id} of automation {automation.id} does not store a Reporter." + ) + parameters = prepare_report_parameters( + dict(automation.parameters), + automation.cronstr, + now=now, + automation_id=automation.id, + cron_timezone=automation.timezone, + scheduled_at=scheduled_at, + ) + report_sensors = resolve_data_generator_sensors( + reporter, reporter._parameters_schema.load(parameters) + ) + for output_sensor in report_sensors["output_sensors"]: + validate_automation_output_scope( + automation.asset_id, output_sensor, automation.type + ) + # The data generator instance is cached on the data source, which may be shared + # by several automations, so wipe any parameter state from a previous run. + reporter._parameters = None + reporter.set_job_trigger("automation", automation_id=automation.id) + return reporter.compute(as_job=True, parameters=parameters) + + def _run_schedule_automation(automation: Automation) -> dict[str, Any]: from flexmeasures.data.schemas.scheduling import AssetTriggerSchema from flexmeasures.data.services.scheduling import ( diff --git a/flexmeasures/data/services/reporting.py b/flexmeasures/data/services/reporting.py new file mode 100644 index 0000000000..b784136ede --- /dev/null +++ b/flexmeasures/data/services/reporting.py @@ -0,0 +1,121 @@ +""" +Logic for queueing and running reporting jobs. +""" + +from __future__ import annotations + +from datetime import timedelta +from typing import TYPE_CHECKING + +from flask import current_app +from rq.job import Job + +from flexmeasures.data import db +from flexmeasures.data.utils import save_to_db + +if TYPE_CHECKING: + from flexmeasures.data.models.reporting import Reporter + + +def create_reporting_job(reporter: "Reporter", queue: str = "reporting") -> Job: + """Queue a job that computes a report and saves the results to the database. + + The reporter's (loaded) parameters are re-serialized into the job, + and the reporter itself travels as its data source ID (which stores its config), + so the job is fully deserializable by the worker. + """ + # Ensure the data source ID is available in the database when the job runs. + reporter._data_source = db.session.merge(reporter.data_source) + db.session.flush() + data_source_id = reporter._data_source.id + db.session.commit() + + parameters = reporter._parameters_schema.dump(reporter._parameters) + output_sensor_ids = [ + output["sensor"] for output in parameters.get("output", []) or [] + ] + + # job metadata for tracking (datetimes as ISO strings, + # a workaround for https://github.com/Parallels/rq-dashboard/issues/510) + job_metadata = { + "data_source_info": {"id": data_source_id}, + "start": parameters.get("start"), + "end": parameters.get("end"), + "sensor_id": output_sensor_ids[0] if output_sensor_ids else None, + } + if reporter._job_trigger: + job_metadata["trigger"] = reporter._job_trigger + + job = Job.create( + run_report_job, + kwargs=dict( + data_source_id=data_source_id, + parameters=parameters, + automation_id=(reporter._job_trigger or {}).get("automation_id"), + ), + connection=current_app.queues[queue].connection, + ttl=int( + current_app.config.get( + "FLEXMEASURES_JOB_TTL", timedelta(-1) + ).total_seconds() + ), + result_ttl=int( + current_app.config.get( + "FLEXMEASURES_PLANNING_TTL", timedelta(-1) + ).total_seconds() + ), # NB job.cleanup docs says a negative number of seconds means persisting forever + meta=job_metadata, + timeout=60 * 60, # 1 hour + ) + current_app.queues[queue].enqueue_job(job) + for sensor_id in output_sensor_ids: + current_app.job_cache.add( + sensor_id, + job_id=job.id, + queue=queue, + asset_or_sensor_type="sensor", + ) + return job + + +def run_report_job( + data_source_id: int, parameters: dict, automation_id: int | None = None +) -> list[dict]: + """Compute a report (with the data generator stored on the given data source) + and save the results to the database. + + This function is meant to be run by a worker processing the reporting queue. + 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 there. + """ + from flexmeasures.data.models.data_sources import DataSource + from flexmeasures.data.models.reporting import Reporter + + source = db.session.get(DataSource, data_source_id) + if source is None: + raise ValueError(f"Data source {data_source_id} no longer exists.") + reporter = source.data_generator + if not isinstance(reporter, Reporter): + raise ValueError(f"Data source {data_source_id} does not store a Reporter.") + # The data generator instance is cached on the data source, which may be shared + # (e.g. within a long-lived worker process), so wipe any previous parameter state. + reporter._parameters = None + results = reporter.compute(parameters=parameters) + for result in results: + save_to_db(result["data"]) + db.session.commit() + + if automation_id is not None and parameters.get("end"): + from datetime import datetime + + from flexmeasures.data.services.automations import record_automation_run + + record_automation_run( + automation_id, now=datetime.fromisoformat(parameters["end"]) + ) + + # return a light summary (the report data itself is stored in the database) + return [ + {"sensor_id": result["sensor"].id, "n_rows": len(result["data"])} + for result in results + ] diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py index 07846f69a5..952decb311 100644 --- a/flexmeasures/data/services/sensors.py +++ b/flexmeasures/data/services/sensors.py @@ -800,6 +800,15 @@ def build_asset_jobs_data( current_app.job_cache.get(sensor.id, "forecasting", "sensor"), ) ) + jobs.append( + ( + "reporting", + "sensor", + sensor.id, + sensor.name, + current_app.job_cache.get(sensor.id, "reporting", "sensor"), + ) + ) jobs_data = list() # Building the actual return list - we also unpack lists of jobs, each to its own entry, and we add error info diff --git a/flexmeasures/data/services/utils.py b/flexmeasures/data/services/utils.py index 62dc2066c9..e6b15ec435 100644 --- a/flexmeasures/data/services/utils.py +++ b/flexmeasures/data/services/utils.py @@ -286,6 +286,9 @@ def wrapper(*args, **kwargs): "force_new_job_creation", False ) + # provenance meta data (how the job got created) must not affect job identity + kwargs_for_hash.pop("trigger", None) + # creating a hash from args and kwargs_for_hash args_hash = f"{queue}:{func.__name__}:{hash_function_arguments(args, kwargs_for_hash)}" diff --git a/flexmeasures/data/tests/test_automations_fresh_db.py b/flexmeasures/data/tests/test_automations_fresh_db.py index fc54751e0c..5b90b48364 100644 --- a/flexmeasures/data/tests/test_automations_fresh_db.py +++ b/flexmeasures/data/tests/test_automations_fresh_db.py @@ -74,6 +74,21 @@ def test_automation_requires_generator(fresh_db, automation_with_generator): fresh_db.session.commit() +def test_report_automation_requires_generator(fresh_db, automation_with_generator): + forecast_automation, _ = automation_with_generator + report_automation = Automation( + asset=forecast_automation.asset, + type="reports", + name="generator-free report", + cronstr="0 1 * * *", + parameters={}, + ) + fresh_db.session.add(report_automation) + + with pytest.raises(IntegrityError): + fresh_db.session.commit() + + def test_schedule_automation_does_not_require_generator( fresh_db, automation_with_generator ): @@ -130,6 +145,11 @@ def test_run_schedule_automation( "automation_id": automation.id, } + # Trigger provenance must not affect job identity: the same schedule request + # from another origin deduplicates onto the same job through the job cache. + returns_2 = run_automation(automation) + assert returns_2["job_id"] == returns["job_id"] + @pytest.mark.parametrize("sequential", (False, True)) def test_run_minimal_schedule_automation_with_stored_flex_config( diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 63cfce523c..c717dfc54a 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -6422,7 +6422,8 @@ "default": "forecasts", "enum": [ "forecasts", - "schedules" + "schedules", + "reports" ] }, "name": { @@ -6450,14 +6451,17 @@ "type": "object", "additionalProperties": {} }, - "forecaster": { - "type": "string", - "default": "TrainPredictPipeline", - "description": "Forecaster class (only used for type 'forecasts')." + "generator": { + "type": [ + "string", + "null" + ], + "default": null, + "description": "Data generator class, e.g. a forecaster (defaults to TrainPredictPipeline) or a reporter (required for type 'reports', e.g. PandasReporter). Not used for type 'schedules'." }, "config": { "type": "object", - "description": "Forecaster configuration (only used for type 'forecasts').", + "description": "Data generator configuration (only used for types 'forecasts' and 'reports').", "additionalProperties": {} } }, diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 11d7c08afa..9d1b9e6639 100644 --- a/flexmeasures/ui/templates/assets/asset_automations.html +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -15,7 +15,7 @@

Automations of {{ asset.name }} @@ -55,6 +55,7 @@

New automation for {{ asset.name }}
@@ -67,11 +68,22 @@
New automation for {{ asset.name }}
Choose the IANA timezone in which this recurrence should follow the local clock. The asset timezone is selected by default.
+
+ + +
+ Forecaster class (defaults to TrainPredictPipeline) or reporter class (required for type reports, e.g. PandasReporter). Not used for schedules. +
+
+
+ + +
- Forecast parameters (for type forecasts) or a schedule trigger message (for type schedules). + Forecast parameters (for type forecasts), a schedule trigger message (for type schedules), or report parameters (for type reports).
@@ -135,7 +147,7 @@
Edit automation
@@ -149,6 +161,11 @@
Edit automation
+
+
+
+
+
@@ -339,7 +356,7 @@
Recently created jobs
method: "GET", success: function (res) { res.automations.forEach(automation => automationsById.set(automation.id, automation)); - for (const automationType of ["forecasts", "schedules"]) { + for (const automationType of ["forecasts", "schedules", "reports"]) { makeAutomationsTable( automationType, res.automations.filter(automation => automation.type === automationType), @@ -347,7 +364,8 @@
Recently created jobs
} }, error: function (xhr) { - for (const automationType of ["forecasts", "schedules"]) { + console.error("Error fetching automations:", xhr); + for (const automationType of ["forecasts", "schedules", "reports"]) { makeAutomationsTable(automationType, []); $(`#automationsTable-${automationType}`).hide(); } @@ -433,6 +451,16 @@
Recently created jobs
return; } } + let config = {}; + const configText = $("#automationConfig").val().trim(); + if (configText) { + try { + config = JSON.parse(configText); + } catch (e) { + $("#newAutomationErr").removeClass("d-none").text("The data generator config is not valid JSON."); + return; + } + } $.ajax({ url: `/api/v3_0/assets/${assetId}/automations`, method: "POST", @@ -443,6 +471,8 @@
Recently created jobs
cronstr: $("#automationCron").val(), timezone: $("#automationTimezone").val(), active: $("#automationActive").is(":checked"), + generator: $("#automationGenerator").val().trim() || null, + config: config, parameters: parameters, }), success: () => location.reload(), diff --git a/flexmeasures/ui/tests/test_asset_crud.py b/flexmeasures/ui/tests/test_asset_crud.py index 9340e48d01..79c41b663b 100644 --- a/flexmeasures/ui/tests/test_asset_crud.py +++ b/flexmeasures/ui/tests/test_asset_crud.py @@ -72,8 +72,10 @@ def test_asset_page(db, client, setup_assets, as_prosumer_user1, view): assert "Automations of".encode() in asset_page.data assert "Forecasts".encode() in asset_page.data assert "Schedules".encode() in asset_page.data + assert "Reports".encode() in asset_page.data assert b'id="automationsTable-forecasts"' in asset_page.data assert b'id="automationsTable-schedules"' in asset_page.data + assert b'id="automationsTable-reports"' in asset_page.data assert b"automation.type === automationType" in asset_page.data assert b"No ${automationType} automations" in asset_page.data assert b'id="automations_err"' in asset_page.data