From ed0dce4305a4da5e9a6251af1aacb30178256f25 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 01:57:17 +0100 Subject: [PATCH 01/15] feat(client): add job failure and timeout exceptions Background jobs polled via the generic jobs API can end in a failed terminal state or never reach one at all. Give both cases a dedicated exception so callers can tell them apart from transport errors. Signed-off-by: Mohamed Belhsan Hmida --- src/flexmeasures_client/exceptions.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/flexmeasures_client/exceptions.py b/src/flexmeasures_client/exceptions.py index 96beb4d9..312ef197 100644 --- a/src/flexmeasures_client/exceptions.py +++ b/src/flexmeasures_client/exceptions.py @@ -44,3 +44,19 @@ class InsufficientServerVersionError(Exception): def __init__(self, message): self.message = message super().__init__(self.message) + + +class JobFailedError(Exception): + """Raised when a background job ends in a non-successful terminal state""" + + def __init__(self, message): + self.message = message + super().__init__(self.message) + + +class JobTimeoutError(Exception): + """Raised when a background job does not reach a terminal state in time""" + + def __init__(self, message): + self.message = message + super().__init__(self.message) From 0d0f387efe822d4e3869c1a0cd9771eda32bf09f Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 01:58:25 +0100 Subject: [PATCH 02/15] feat(client): poll background jobs via the generic jobs API Add get_job_status() for a single lookup of GET /api/v3_0/jobs/, and wait_for_job() to poll one until it reaches a terminal state, with exponential backoff capped at max_polling_interval and a total timeout budget. Failed jobs raise JobFailedError carrying the server's message and traceback, so callers get something actionable instead of a bare status. Signed-off-by: Mohamed Belhsan Hmida --- src/flexmeasures_client/client.py | 111 ++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/src/flexmeasures_client/client.py b/src/flexmeasures_client/client.py index 4d552278..a8b3fc42 100644 --- a/src/flexmeasures_client/client.py +++ b/src/flexmeasures_client/client.py @@ -6,6 +6,7 @@ import os import re import socket +import time import warnings from dataclasses import dataclass, field from datetime import datetime, timedelta @@ -24,6 +25,8 @@ EmailValidationError, EmptyPasswordError, InsufficientServerVersionError, + JobFailedError, + JobTimeoutError, WrongAPIVersionError, WrongHostError, ) @@ -41,6 +44,14 @@ POLLING_INTERVAL = 10.0 # seconds API_VERSIONS_LIST = ["v3_0"] +JOB_POLLING_INTERVAL = 2.0 # seconds, first wait between job status polls +JOB_POLLING_MAX_INTERVAL = 30.0 # seconds, cap on the backing-off wait +JOB_POLLING_TIMEOUT = 600.0 # seconds, total budget for a job to finish + +# Terminal job states, as reported by GET /api/v3_0/jobs/ +JOB_STATUS_FINISHED = "FINISHED" +JOB_STATUS_UNSUCCESSFUL = frozenset({"FAILED", "STOPPED", "CANCELED"}) + def _parse_json_field(data: dict, field_name: str) -> None: """Parse a JSON string field in-place if it exists and is a string.""" @@ -114,6 +125,18 @@ def convert_units( return values +def _describe_failed_job(job_id: str, job: dict) -> str: + """Build an error message for a job that ended in a non-successful state.""" + msg = f"Job {job_id} ended with status {job['status']}." + job_message = job.get("message") + if job_message: + msg += f" {job_message}" + exc_info = job.get("exc-info") or job.get("exc_info") + if exc_info: + msg += f"\nServer traceback:\n{exc_info}" + return msg + + @dataclass class FlexMeasuresClient: """Main class for connecting to the FlexMeasures API.""" @@ -1701,6 +1724,94 @@ async def trigger_and_get_forecast( forecast_id=forecast_id, ) + async def get_job_status(self, job_id: str) -> dict: + """Get the status of a background job. + + :param job_id: UUID of the job, as returned by a trigger endpoint. + + :returns: job status as a dictionary, for example: + { + 'status': 'FINISHED', + 'message': 'Report job finished.', + 'result': None, + 'origin': 'flexmeasures:reporting', + 'func-name': 'flexmeasures.data.services.reporting.run_report_job', + 'enqueued-at': '2026-08-18T09:00:00+00:00', + 'started-at': '2026-08-18T09:00:01+00:00', + 'ended-at': '2026-08-18T09:00:04+00:00', + 'exc-info': None, + } + + The ``status`` field is one of QUEUED, STARTED, FINISHED, FAILED, + DEFERRED, SCHEDULED, STOPPED or CANCELED. + + This function raises a ValueError when an unhandled status code is returned. + """ + response, status = await self.request( + uri=f"jobs/{job_id}", + method="GET", + ) + check_for_status(status, 200) + + if not isinstance(response, dict): + raise ContentTypeError( + f"Expected a dictionary, but got {type(response)}", + ) + + if not isinstance(response.get("status"), str): + raise ContentTypeError( + f"Expected a job status string, but got {type(response.get('status'))}", + ) + return response + + async def wait_for_job( + self, + job_id: str, + timeout: float = JOB_POLLING_TIMEOUT, + polling_interval: float = JOB_POLLING_INTERVAL, + max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, + ) -> dict: + """Poll a background job until it reaches a terminal state. + + Waits are backed off exponentially, from ``polling_interval`` up to + ``max_polling_interval``, so that short jobs are picked up quickly + without hammering the server on long ones. + + :param job_id: UUID of the job, as returned by a trigger endpoint. + :param timeout: total number of seconds to wait for the job to finish. + :param polling_interval: seconds to wait before the first status poll. + :param max_polling_interval: upper bound on the wait between polls. + + :returns: the final job status dictionary (see :func:`get_job_status`). + + :raises JobFailedError: if the job ended as FAILED, STOPPED or CANCELED. + :raises JobTimeoutError: if the job did not finish within ``timeout``. + """ + deadline = time.monotonic() + timeout + interval = polling_interval + job = await self.get_job_status(job_id) + while True: + status = job["status"] + if status == JOB_STATUS_FINISHED: + self.logger.info(f"Job {job_id} finished.") + return job + if status in JOB_STATUS_UNSUCCESSFUL: + raise JobFailedError(_describe_failed_job(job_id, job)) + + remaining = deadline - time.monotonic() + if remaining <= 0: + raise JobTimeoutError( + f"Job {job_id} did not finish within {timeout} seconds. " + f"Last known status: {status}." + ) + self.logger.debug( + f"Job {job_id} has status {status}. " + f"Checking again in {interval} seconds..." + ) + await asyncio.sleep(min(interval, remaining)) + interval = min(interval * 2, max_polling_interval) + job = await self.get_job_status(job_id) + @staticmethod def create_storage_flex_model( soc_unit: str, From 2c6c3b21f7c5cecceff078587541b951a39cbabc Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 01:59:01 +0100 Subject: [PATCH 03/15] feat(client): trigger one-off reports over the asset API Add trigger_report(), posting to POST /assets//reports/trigger and returning the queued job UUID, plus trigger_and_await_report() which chains it with wait_for_job(). The endpoint lands in FlexMeasures v1.1.0, so a 404 from an older server is reported as a version problem pointing at the CLI fallback. Signed-off-by: Mohamed Belhsan Hmida --- src/flexmeasures_client/client.py | 98 +++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/src/flexmeasures_client/client.py b/src/flexmeasures_client/client.py index a8b3fc42..3fef0612 100644 --- a/src/flexmeasures_client/client.py +++ b/src/flexmeasures_client/client.py @@ -52,6 +52,9 @@ JOB_STATUS_FINISHED = "FINISHED" JOB_STATUS_UNSUCCESSFUL = frozenset({"FAILED", "STOPPED", "CANCELED"}) +# The asset report trigger endpoint landed in FlexMeasures v1.1.0 +REPORT_TRIGGER_MIN_SERVER_VERSION = "1.1.0" + def _parse_json_field(data: dict, field_name: str) -> None: """Parse a JSON string field in-place if it exists and is a string.""" @@ -1812,6 +1815,101 @@ async def wait_for_job( interval = min(interval * 2, max_polling_interval) job = await self.get_job_status(job_id) + async def trigger_report( + self, + asset_id: int, + reporter: str, + parameters: dict, + config: dict | None = None, + ) -> str: + """Trigger a one-off reporting job for the given asset. + + The report runs on the server's ``reporting`` queue, so the server + needs a worker listening on that queue: + + flexmeasures jobs run-worker --queue reporting + + The caller needs to be able to read every input and configuration + sensor, and to record data on every output sensor. Each output sensor + must belong to the given asset or one of its descendants. + + :param asset_id: ID of the asset to report on. Output sensors must sit + in this asset's subtree. + :param reporter: name of the reporter class, e.g. "PandasReporter". + :param parameters: reporter parameters, holding at least the ``input`` + and ``output`` sensors and the ``start`` and ``end`` + of the reporting period. + :param config: reporter configuration. Defaults to an empty config. + + :returns: job UUID (string), to be passed to :func:`wait_for_job`. + + This function raises a ValueError when an unhandled status code is returned. + """ + json_payload: dict[str, Any] = { + "reporter": reporter, + "parameters": parameters, + } + if config is not None: + json_payload["config"] = config + + response, status = await self.request( + uri=f"assets/{asset_id}/reports/trigger", + json_payload=json_payload, + method="POST", + minimum_server_version=REPORT_TRIGGER_MIN_SERVER_VERSION, + minimum_server_version_msg=( + "Reports can only be triggered over the API from this version on. " + "On older servers, use the `flexmeasures add report` CLI command." + ), + ) + check_for_status(status, 202) + + if not isinstance(response, dict): + raise ContentTypeError( + f"Expected a dictionary, but got {type(response)}", + ) + + if not isinstance(response.get("job"), str): + raise ContentTypeError( + f"Expected a job ID, but got {type(response.get('job'))}", + ) + job_id = response["job"] + self.logger.info(f"Report triggered successfully. Job ID: {job_id}") + return job_id + + async def trigger_and_await_report( + self, + asset_id: int, + reporter: str, + parameters: dict, + config: dict | None = None, + timeout: float = JOB_POLLING_TIMEOUT, + polling_interval: float = JOB_POLLING_INTERVAL, + max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, + ) -> dict: + """Trigger a one-off reporting job and wait for it to finish. + + Reports write their result to their output sensors rather than + returning it, so use :func:`get_sensor_data` to read the values back. + + :returns: the final job status dictionary (see :func:`get_job_status`). + + :raises JobFailedError: if the report job ended as FAILED, STOPPED or CANCELED. + :raises JobTimeoutError: if the report job did not finish within ``timeout``. + """ + job_id = await self.trigger_report( + asset_id=asset_id, + reporter=reporter, + parameters=parameters, + config=config, + ) + return await self.wait_for_job( + job_id=job_id, + timeout=timeout, + polling_interval=polling_interval, + max_polling_interval=max_polling_interval, + ) + @staticmethod def create_storage_flex_model( soc_unit: str, From 5da84f0cf07c5787156d0a570c69cfdd17ac10ec Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 01:59:50 +0100 Subject: [PATCH 04/15] test(client): cover report triggering and job polling Assert the exact request bodies sent to the trigger endpoint, that polling walks through QUEUED/DEFERRED/STARTED before returning a finished job, and that failed, stopped, canceled and never-finishing jobs each raise the right error. Signed-off-by: Mohamed Belhsan Hmida --- tests/client/test_report.py | 261 ++++++++++++++++++++++++++++++++++++ 1 file changed, 261 insertions(+) create mode 100644 tests/client/test_report.py diff --git a/tests/client/test_report.py b/tests/client/test_report.py new file mode 100644 index 00000000..a2ec740f --- /dev/null +++ b/tests/client/test_report.py @@ -0,0 +1,261 @@ +import pytest +from aioresponses import aioresponses +from yarl import URL + +from flexmeasures_client.client import FlexMeasuresClient +from flexmeasures_client.exceptions import ( + ContentTypeError, + JobFailedError, + JobTimeoutError, +) + +ASSET_ID = 3 +JOB_ID = "364bfd06-c1fa-430b-8d25-8f5a547651fb" +TRIGGER_URL = f"http://localhost:5000/api/v3_0/assets/{ASSET_ID}/reports/trigger" +JOB_URL = f"http://localhost:5000/api/v3_0/jobs/{JOB_ID}" + +PARAMETERS = { + "input": [{"name": "pv", "sensor": 1}], + "output": [{"name": "self-consumption", "sensor": 2}], + "start": "2026-08-18T00:00:00+02:00", + "end": "2026-08-19T00:00:00+02:00", +} +ACCEPTED_PAYLOAD = { + "status": "ACCEPTED", + "message": "Request has been accepted for processing.", + "job": JOB_ID, + "job-url": f"/api/v3_0/jobs/{JOB_ID}", +} + + +def make_client() -> FlexMeasuresClient: + client = FlexMeasuresClient(email="test@test.test", password="test") + client.access_token = "test-token" + return client + + +def job_payload(status: str, **extra) -> dict: + payload = { + "status": status, + "message": f"Job is {status.lower()}.", + "result": None, + "origin": "flexmeasures:reporting", + "exc-info": None, + } + payload.update(extra) + return payload + + +@pytest.mark.asyncio +async def test_trigger_report() -> None: + """Test triggering a report without an explicit reporter config.""" + with aioresponses() as m: + client = make_client() + m.post(TRIGGER_URL, status=202, payload=ACCEPTED_PAYLOAD) + + job_id = await client.trigger_report( + asset_id=ASSET_ID, + reporter="PandasReporter", + parameters=PARAMETERS, + ) + + assert job_id == JOB_ID + + m.assert_called_once_with( + TRIGGER_URL, + method="POST", + headers={ + "Content-Type": "application/json", + "Authorization": "test-token", + }, + json={"reporter": "PandasReporter", "parameters": PARAMETERS}, + params=None, + ssl=False, + allow_redirects=False, + ) + + await client.close() + + +@pytest.mark.asyncio +async def test_trigger_report_with_config() -> None: + """Test that a reporter config is passed on as its own field.""" + config = {"required_input": [{"name": "pv"}]} + with aioresponses() as m: + client = make_client() + m.post(TRIGGER_URL, status=202, payload=ACCEPTED_PAYLOAD) + + await client.trigger_report( + asset_id=ASSET_ID, + reporter="PandasReporter", + parameters=PARAMETERS, + config=config, + ) + + m.assert_called_once_with( + TRIGGER_URL, + method="POST", + headers={ + "Content-Type": "application/json", + "Authorization": "test-token", + }, + json={ + "reporter": "PandasReporter", + "parameters": PARAMETERS, + "config": config, + }, + params=None, + ssl=False, + allow_redirects=False, + ) + + await client.close() + + +@pytest.mark.asyncio +async def test_trigger_report_without_job_id() -> None: + """Test that a malformed trigger response is reported clearly.""" + with aioresponses() as m: + client = make_client() + m.post(TRIGGER_URL, status=202, payload={"status": "ACCEPTED"}) + + with pytest.raises(ContentTypeError): + await client.trigger_report( + asset_id=ASSET_ID, + reporter="PandasReporter", + parameters=PARAMETERS, + ) + + await client.close() + + +@pytest.mark.asyncio +async def test_get_job_status() -> None: + """Test a single job status lookup.""" + with aioresponses() as m: + client = make_client() + m.get(JOB_URL, status=200, payload=job_payload("STARTED")) + + job = await client.get_job_status(JOB_ID) + + assert job["status"] == "STARTED" + m.assert_called_once_with( + JOB_URL, + method="GET", + headers={ + "Content-Type": "application/json", + "Authorization": "test-token", + }, + json=None, + params=None, + ssl=False, + allow_redirects=False, + ) + + await client.close() + + +@pytest.mark.asyncio +async def test_get_job_status_without_status() -> None: + """Test that a malformed job status response is reported clearly.""" + with aioresponses() as m: + client = make_client() + m.get(JOB_URL, status=200, payload={"message": "Hi."}) + + with pytest.raises(ContentTypeError): + await client.get_job_status(JOB_ID) + + await client.close() + + +@pytest.mark.asyncio +async def test_wait_for_job_polls_until_finished() -> None: + """Test that polling continues through the non-terminal states.""" + with aioresponses() as m: + client = make_client() + m.get(JOB_URL, status=200, payload=job_payload("QUEUED")) + m.get(JOB_URL, status=200, payload=job_payload("DEFERRED")) + m.get(JOB_URL, status=200, payload=job_payload("STARTED")) + m.get(JOB_URL, status=200, payload=job_payload("FINISHED")) + + job = await client.wait_for_job(JOB_ID, polling_interval=0.01) + + assert job["status"] == "FINISHED" + assert len(m.requests[("GET", URL(JOB_URL))]) == 4 + + await client.close() + + +@pytest.mark.asyncio +async def test_wait_for_job_raises_on_failed_job() -> None: + """Test that a failed job surfaces the server message and traceback.""" + with aioresponses() as m: + client = make_client() + m.get( + JOB_URL, + status=200, + payload=job_payload( + "FAILED", + message="Report job failed.", + **{"exc-info": "Traceback: KeyError: 'pv'"}, + ), + ) + + with pytest.raises(JobFailedError) as exc_info: + await client.wait_for_job(JOB_ID, polling_interval=0.01) + + message = str(exc_info.value) + assert "FAILED" in message + assert "Report job failed." in message + assert "KeyError: 'pv'" in message + + await client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["STOPPED", "CANCELED"]) +async def test_wait_for_job_raises_on_unsuccessful_job(status: str) -> None: + """Test that jobs stopped or canceled by an operator also raise.""" + with aioresponses() as m: + client = make_client() + m.get(JOB_URL, status=200, payload=job_payload(status)) + + with pytest.raises(JobFailedError): + await client.wait_for_job(JOB_ID, polling_interval=0.01) + + await client.close() + + +@pytest.mark.asyncio +async def test_wait_for_job_times_out() -> None: + """Test that a job that never finishes hits the timeout budget.""" + with aioresponses() as m: + client = make_client() + m.get(JOB_URL, status=200, payload=job_payload("QUEUED"), repeat=True) + + with pytest.raises(JobTimeoutError) as exc_info: + await client.wait_for_job(JOB_ID, timeout=0.05, polling_interval=0.01) + + assert "QUEUED" in str(exc_info.value) + + await client.close() + + +@pytest.mark.asyncio +async def test_trigger_and_await_report() -> None: + """Test triggering a report and waiting for the queued job in one call.""" + with aioresponses() as m: + client = make_client() + m.post(TRIGGER_URL, status=202, payload=ACCEPTED_PAYLOAD) + m.get(JOB_URL, status=200, payload=job_payload("FINISHED")) + + job = await client.trigger_and_await_report( + asset_id=ASSET_ID, + reporter="PandasReporter", + parameters=PARAMETERS, + polling_interval=0.01, + ) + + assert job["status"] == "FINISHED" + + await client.close() From c5e5dcdc487add82f2b3a8ab5dccc10ba7357e6c Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:01:12 +0100 Subject: [PATCH 05/15] refactor(hems): drive reports over the API instead of the CLI Replace run_report_cmd() and its subprocess call with an async run_report() that triggers the report on the asset owning its output sensors and waits for the job. Parameters are now built in memory rather than written to configs/*_reporter_param.json, and configs are read on the client side and posted along with the request. Neither the CLI nor a bind-mount of configs/ into the server is needed anymore, so cli_command_prefix() and the FLEXMEASURES_CLI_CMD/FLEXMEASURES_CLI_CONFIG_DIR handling are gone. Signed-off-by: Mohamed Belhsan Hmida --- examples/HEMS/utils/reporter_utils.py | 181 +++++++++++++------------- 1 file changed, 92 insertions(+), 89 deletions(-) diff --git a/examples/HEMS/utils/reporter_utils.py b/examples/HEMS/utils/reporter_utils.py index 105a35b8..e154b264 100644 --- a/examples/HEMS/utils/reporter_utils.py +++ b/examples/HEMS/utils/reporter_utils.py @@ -1,63 +1,66 @@ import json import os -import shlex -import subprocess from pathlib import Path -BASE_DIR = Path(__file__).parent.parent - +from flexmeasures_client import FlexMeasuresClient +from flexmeasures_client.exceptions import ( + ContentTypeError, + InsufficientServerVersionError, + JobFailedError, + JobTimeoutError, +) -def cli_command_prefix() -> list[str]: - """ - The command used to invoke the FlexMeasures CLI, split into argv tokens. +BASE_DIR = Path(__file__).parent.parent - Defaults to a plain ``flexmeasures`` on PATH, which requires the CLI to be - installed locally and configured to talk to the same database as the - server the client is scripting against (see the "Running the Tutorial" - instructions in docs/HEMS.rst). +# Errors that mean "this report did not happen", as opposed to a bug in the example. +REPORT_ERRORS = ( + JobFailedError, + JobTimeoutError, + InsufficientServerVersionError, + ContentTypeError, + ConnectionError, + ValueError, +) - Override via the ``FLEXMEASURES_CLI_CMD`` environment variable to run the - CLI elsewhere, e.g. inside a Docker Compose service: - FLEXMEASURES_CLI_CMD="docker compose -f /path/to/docker-compose.yml exec -T server flexmeasures" +def load_reporter_config(reporter_type: str) -> dict: """ - return shlex.split(os.environ.get("FLEXMEASURES_CLI_CMD", "flexmeasures")) + Load a reporter configuration from the `configs/` directory. + Configurations are static, hand-written data, kept as JSON for readability: -def _cli_config_path(local_path: str) -> str: - """ - Translate a local config/parameter file path to the path the CLI process - will see it at, when that process runs somewhere other than this host - (e.g. inside a container with the ``configs/`` directory bind-mounted - elsewhere). Controlled via ``FLEXMEASURES_CLI_CONFIG_DIR``; a no-op if unset. + configs/{reporter_type}_reporter_config.json + + They are read here on the client side and posted to the API as part of the + report request, so the server never needs access to these files. """ - remote_dir = os.environ.get("FLEXMEASURES_CLI_CONFIG_DIR") - if not remote_dir: - return local_path - return os.path.join(remote_dir, os.path.basename(local_path)) + full_path = os.path.join(BASE_DIR, f"configs/{reporter_type}_reporter_config.json") + with open(full_path) as f: + return json.load(f) -def fill_reporter_params( +def build_reporter_parameters( input_sensors: list[dict], output_sensors: list[dict] | dict, start: str, end: str, reporter_type: str, -): +) -> dict: """ - Fill reporter parameters and save them to a JSON configuration file. + Build the reporter parameters for a single report. - The file is saved inside the `configs/` directory, with the name - derived from the given `reporter_type`, e.g.: configs/{reporter_type}_config.json + :param input_sensors: list of single-entry dicts, mapping the input name the + reporter config expects to a sensor ID. + :param output_sensors: list of sensors to write to. The aggregate reporter + takes a single sensor instead, written without a name. """ - if reporter_type == "aggregate": - # For the aggregate reporter, output_sensors is a single sensor ID + # For the aggregate reporter, output_sensors is a single sensor output = [{"sensor": output_sensors["id"]}] else: output = [{"name": s["name"], "sensor": s["id"]} for s in output_sensors] - params = { + return { "input": [ { "name": name, @@ -74,65 +77,65 @@ def fill_reporter_params( "check_output_resolution": False, } - # overwrite the file (creates it if not exists) - file_path = f"configs/{reporter_type}_reporter_param.json" - full_path = os.path.join(BASE_DIR, file_path) - with open(full_path, "w") as f: - json.dump(params, f, indent=4) +def asset_id_for_outputs(output_sensors: list[dict] | dict) -> int: + """ + The asset to trigger a report against: the one owning its output sensors. -def run_report_cmd(reporter_map: dict, start: str, end: str) -> bool: + The API requires every output sensor to sit in the subtree of the asset in + the URL, so a report on site sensors is triggered against that site, and a + report on community sensors against the community. + """ + sensors = [output_sensors] if isinstance(output_sensors, dict) else output_sensors + asset_ids = {sensor["generic_asset_id"] for sensor in sensors} + if len(asset_ids) != 1: + raise ValueError( + f"Expected all output sensors to belong to one asset, but got {asset_ids}. " + "Split this into one report per asset." + ) + return asset_ids.pop() + + +async def run_report( + client: FlexMeasuresClient, + reporter: str, + reporter_type: str, + input_sensors: list[dict], + output_sensors: list[dict] | dict, + start: str, + end: str, +) -> bool: """ - Run the FlexMeasures CLI command to generate a report for a given reporter. - - This function expects the reporter configuration and parameter files - to already exist in the `configs/` directory, following the naming pattern - created by `fill_reporter_params()`: - - configs/{reporter_name}_reporter_config.json - configs/{reporter_name}_reporter_param.json - - Args: - reporter_map (dict): A dictionary describing the reporter to run. - Must contain: - - "name": str → name of the reporter, used in file paths - - "reporter": str → FlexMeasures reporter class name - Example: - reporter_map = { - "name": "aggregate", - "reporter": "AggregatorReporter", - } - - start (str): Start time of the report period (ISO 8601 format). - end (str): End time of the report period (ISO 8601 format). + Run a single report through the API and wait for its job to finish. + + Requires the server to run a worker on the `reporting` queue: + + flexmeasures jobs run-worker --queue reporting + + :param reporter: FlexMeasures reporter class name, e.g. "PandasReporter". + :param reporter_type: name of the reporter in this example, used to find its + configuration file, e.g. "self-consumption". + + :returns: True if the report finished, False if it failed or timed out. """ - config_path = os.path.join( - BASE_DIR, f"configs/{reporter_map['name']}_reporter_config.json" + asset_id = asset_id_for_outputs(output_sensors) + parameters = build_reporter_parameters( + input_sensors=input_sensors, + output_sensors=output_sensors, + start=start, + end=end, + reporter_type=reporter_type, ) - param_path = os.path.join( - BASE_DIR, f"configs/{reporter_map['name']}_reporter_param.json" - ) - cmd = [ - *cli_command_prefix(), - "add", - "report", - "--reporter", - reporter_map["reporter"], - "--config", - _cli_config_path(config_path), - "--parameters", - _cli_config_path(param_path), - "--start", - start, - "--end", - end, - ] - - print(f"Running: {' '.join(cmd)}") - result = subprocess.run(cmd, capture_output=True, text=True, timeout=3000) - if result.returncode == 0: - print(f"{reporter_map['name']} reporters generated successfully") - return True - else: - print(f"{reporter_map['name']} reporter generation failed: {result.stderr}") + print(f"Running {reporter_type} report on asset {asset_id}...") + try: + await client.trigger_and_await_report( + asset_id=asset_id, + reporter=reporter, + parameters=parameters, + config=load_reporter_config(reporter_type), + ) + except REPORT_ERRORS as exception: + print(f"{reporter_type} report failed: {exception}") return False + print(f"{reporter_type} report generated successfully") + return True From cd799f9b1c6fd65cb974b59d72935a743f76cf96 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:01:36 +0100 Subject: [PATCH 06/15] refactor(hems): trigger site reports through the API Drop the 'which flexmeasures' probe that silently skipped report generation, and call run_report() per site instead of writing parameter files and shelling out. Also carry the per-site outcome into the return value, which previously reflected the last site only. Signed-off-by: Mohamed Belhsan Hmida --- examples/HEMS/reporters.py | 58 +++++++++++++++----------------------- 1 file changed, 23 insertions(+), 35 deletions(-) diff --git a/examples/HEMS/reporters.py b/examples/HEMS/reporters.py index 958e387e..500c7fd5 100644 --- a/examples/HEMS/reporters.py +++ b/examples/HEMS/reporters.py @@ -1,5 +1,3 @@ -import subprocess - from const import ( SCHEDULING_END, SCHEDULING_START, @@ -11,11 +9,7 @@ pv_name, ) from utils.asset_utils import find_sensors_by_asset -from utils.reporter_utils import ( - cli_command_prefix, - fill_reporter_params, - run_report_cmd, -) +from utils.reporter_utils import run_report from flexmeasures_client import FlexMeasuresClient @@ -23,17 +17,15 @@ async def create_reports( client: FlexMeasuresClient, community_name: str, site_names: list[str] ): - """Generate reports using FlexMeasures CLI.""" - print("Generating reports...") + """Generate reports through the FlexMeasures API. - # Check if the configured FlexMeasures CLI command is available - # (only meaningful to check the first token, e.g. "flexmeasures" or "docker") - check_cmd = ["which", cli_command_prefix()[0]] - check_result = subprocess.run(check_cmd, capture_output=True, text=True) + Each site gets a self-consumption report and a total energy costs report, + triggered against that site's asset. The latter reads the aggregate power + written by the aggregate reporter, so it has to run after those. + """ + print("Generating reports...") - if check_result.returncode != 0: - print("FlexMeasures CLI not found. Skipping report generation.") - return False + all_reports_succeeded = True for i, site_name in enumerate(site_names, start=1): # Find all required sensors @@ -60,8 +52,11 @@ async def create_reports( client, sensor_mappings, top_level_asset_name=community_name ) - # Prepare parameters for self-consumption reporter - fill_reporter_params( + # Run SelfConsumptionReporter + self_consumption_result = await run_report( + client=client, + reporter="PandasReporter", + reporter_type="self-consumption", input_sensors=[ {"production": sensors["electricity-production"]["id"]}, {"pv-power": sensors["pv-power"]["id"]}, @@ -77,11 +72,13 @@ async def create_reports( ], start=SCHEDULING_START, end=SCHEDULING_END, - reporter_type="self-consumption", ) - # Prepare parameters for the total energy costs reporter - fill_reporter_params( + # Run TotalEnergyCostsReporter + total_energy_costs_result = await run_report( + client=client, + reporter="PandasReporter", + reporter_type="total-energy-costs", input_sensors=[ {"aggregate-power": sensors["electricity-aggregate"]["id"]}, {"consumption-production-price": sensors["electricity-price"]["id"]}, @@ -93,21 +90,12 @@ async def create_reports( ], start=SCHEDULING_START, end=SCHEDULING_END, - reporter_type="total-energy-costs", ) - # Run SelfConsumptionReporter - self_consumption_result = run_report_cmd( - reporter_map={"name": "self-consumption", "reporter": "PandasReporter"}, - start=SCHEDULING_START, - end=SCHEDULING_END, - ) - - # Run TotalEnergyCostsReporter - total_energy_costs_result = run_report_cmd( - reporter_map={"name": "total-energy-costs", "reporter": "PandasReporter"}, - start=SCHEDULING_START, - end=SCHEDULING_END, + all_reports_succeeded = ( + self_consumption_result + and total_energy_costs_result + and all_reports_succeeded ) - return self_consumption_result and total_energy_costs_result + return all_reports_succeeded From 2177b35a32eac4a53fd9aaa8375da094d571daab Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:02:48 +0100 Subject: [PATCH 07/15] fix(hems): let callers name the asset to report against Sensors nested in an asset listing are dumped with only id and name, so deriving the asset from an output sensor fails for anything read that way. Add an asset_id override to run_report() and a clear error when derivation is not possible. Signed-off-by: Mohamed Belhsan Hmida --- examples/HEMS/utils/reporter_utils.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/examples/HEMS/utils/reporter_utils.py b/examples/HEMS/utils/reporter_utils.py index e154b264..c1ca7805 100644 --- a/examples/HEMS/utils/reporter_utils.py +++ b/examples/HEMS/utils/reporter_utils.py @@ -87,6 +87,13 @@ def asset_id_for_outputs(output_sensors: list[dict] | dict) -> int: report on community sensors against the community. """ sensors = [output_sensors] if isinstance(output_sensors, dict) else output_sensors + if any("generic_asset_id" not in sensor for sensor in sensors): + # Sensors nested in an asset listing are dumped with only id and name, + # so their asset has to be passed to run_report() explicitly. + raise ValueError( + "Output sensors carry no generic_asset_id. " + "Pass asset_id to run_report() instead." + ) asset_ids = {sensor["generic_asset_id"] for sensor in sensors} if len(asset_ids) != 1: raise ValueError( @@ -104,6 +111,7 @@ async def run_report( output_sensors: list[dict] | dict, start: str, end: str, + asset_id: int | None = None, ) -> bool: """ Run a single report through the API and wait for its job to finish. @@ -115,10 +123,14 @@ async def run_report( :param reporter: FlexMeasures reporter class name, e.g. "PandasReporter". :param reporter_type: name of the reporter in this example, used to find its configuration file, e.g. "self-consumption". + :param asset_id: asset to trigger the report against. Defaults to the asset + owning the output sensors, which needs those sensors to + carry a generic_asset_id. :returns: True if the report finished, False if it failed or timed out. """ - asset_id = asset_id_for_outputs(output_sensors) + if asset_id is None: + asset_id = asset_id_for_outputs(output_sensors) parameters = build_reporter_parameters( input_sensors=input_sensors, output_sensors=output_sensors, From b9ad30f5dafd90c7080c86fd273bf27c150628c9 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:03:01 +0100 Subject: [PATCH 08/15] refactor(hems): trigger aggregate reports through the API Make run_community_aggregate async and route both the per-site and the community aggregate through run_report(). Awaiting each job in turn keeps the ordering the community report depends on, and the community report is now skipped outright when a site aggregate did not finish, rather than silently aggregating stale data. Signed-off-by: Mohamed Belhsan Hmida --- examples/HEMS/scheduling.py | 54 ++++++++++++++++++++++--------------- 1 file changed, 33 insertions(+), 21 deletions(-) diff --git a/examples/HEMS/scheduling.py b/examples/HEMS/scheduling.py index d77c5b86..f361673e 100644 --- a/examples/HEMS/scheduling.py +++ b/examples/HEMS/scheduling.py @@ -30,7 +30,7 @@ calculate_ev_soc_targets_and_constraints, simulate_random_trip, ) -from utils.reporter_utils import fill_reporter_params, run_report_cmd +from utils.reporter_utils import run_report from utils.scheduling_utils import create_dynamic_storage_flex_model from flexmeasures_client.client import FlexMeasuresClient @@ -208,7 +208,8 @@ async def run_scheduling_simulation( next_current_soc_dict[site_name]["heating"] = heating_next_current_soc # Run reporter to log community site aggregate power consumption each scheduling step - run_community_aggregate( + await run_community_aggregate( + client=client, sensors=sensors, current_time=current_time, step_end_time=step_end_time, @@ -906,22 +907,32 @@ async def map_site_sensors( return sensors -def run_community_aggregate( +async def run_community_aggregate( + client: FlexMeasuresClient, sensors: dict, current_time: pd.Timestamp, step_end_time: pd.Timestamp, community_asset: dict, site_names: list[str], -): +) -> bool: + """Aggregate power per site, then aggregate those into community power. + + The community report reads the sites' aggregate sensors, so the site + reports have to finish first. Awaiting each job in turn keeps that order. + """ community_power_sensor = None for x in community_asset["sensors"]: if x["name"] == "power": community_power_sensor = x break - # Run each site's aggregate reporter + + # Run each site's aggregate reporter, against that site's asset + all_aggregates_succeeded = True for index, site_name in enumerate(site_names, start=1): - # Fill reporter parameters for each site - fill_reporter_params( + site_aggregate_succeeded = await run_report( + client=client, + reporter="AggregatorReporter", + reporter_type="aggregate", input_sensors=[ {"pv": sensors[f"pv-power-{index}"]["id"]}, {"consumption": sensors[f"building-consumption-{index}"]["id"]}, @@ -933,16 +944,21 @@ def run_community_aggregate( output_sensors=sensors[f"electricity-aggregate-{index}"], start=current_time.isoformat(), end=step_end_time.isoformat(), - reporter_type="aggregate", ) - # Run AggregatorReporter - run_report_cmd( - reporter_map={"name": "aggregate", "reporter": "AggregatorReporter"}, - start=current_time.isoformat(), - end=step_end_time.isoformat(), + all_aggregates_succeeded = site_aggregate_succeeded and all_aggregates_succeeded + + if not all_aggregates_succeeded: + print( + "Skipping the community aggregate report, " + "because not every site aggregate is available." ) + return False - fill_reporter_params( + # Run the community aggregate reporter, against the community asset + return await run_report( + client=client, + reporter="AggregatorReporter", + reporter_type="aggregate", input_sensors=[ {f"aggregate-{index}": sensors[f"electricity-aggregate-{index}"]["id"]} for index, _ in enumerate(site_names, start=1) @@ -950,11 +966,7 @@ def run_community_aggregate( output_sensors=community_power_sensor, start=current_time.isoformat(), end=step_end_time.isoformat(), - reporter_type="aggregate", - ) - # Run AggregatorReporter - run_report_cmd( - reporter_map={"name": "aggregate", "reporter": "AggregatorReporter"}, - start=current_time.isoformat(), - end=step_end_time.isoformat(), + # community_asset comes from an asset listing, whose nested sensors + # carry no generic_asset_id, so name the asset explicitly + asset_id=community_asset["id"], ) From a4fc453026d2e82a43d41caf0d09053b8e0250d2 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:03:36 +0100 Subject: [PATCH 09/15] docs(hems): describe the API-based reporting setup Note the v1.1.0 server requirement, add reporting to the worker queues, and replace the CLI workaround section, which no longer applies now that report configuration is posted rather than read from disk by the server. Signed-off-by: Mohamed Belhsan Hmida --- docs/HEMS.rst | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/docs/HEMS.rst b/docs/HEMS.rst index ed24a9af..abea7288 100644 --- a/docs/HEMS.rst +++ b/docs/HEMS.rst @@ -15,7 +15,7 @@ This is the resulting dashboard: :align: center | -.. note:: The tutorial still uses the CLI for reporting. In future versions, we might make reporting available via the API, as well. +.. note:: The tutorial talks to FlexMeasures over the API only, including for reporting. That requires a FlexMeasures server of version 1.1.0 or above, and a worker listening on the ``reporting`` queue. Set up your environment @@ -63,11 +63,11 @@ Open three terminals. In the first terminal, run the server: flexmeasures run -In the second terminal, run a flexmeasures worker that listens to both the scheduling and forecasting queues: +In the second terminal, run a flexmeasures worker that listens to the scheduling, forecasting and reporting queues: .. code-block:: bash - flexmeasures jobs run-worker --queue "forecasting|scheduling" + flexmeasures jobs run-worker --queue "forecasting|scheduling|reporting" Note: you can run the same command in two terminals (2 workers), to speed up the computation! @@ -79,7 +79,4 @@ In the third terminal, run the client script using the `/examples/HEMS` folder a python3 HEMS_setup.py .. note:: - Report generation (see :ref:`hems-tutorial` note above) shells out to a ``flexmeasures`` CLI process, which by default is expected on ``PATH`` and configured against the same database as the server. If your FlexMeasures server runs elsewhere (e.g. inside a Docker Compose service), point report generation at it instead via two environment variables: - - - ``FLEXMEASURES_CLI_CMD``: the command used to invoke the CLI, e.g. ``"docker compose exec -T server flexmeasures"``. - - ``FLEXMEASURES_CLI_CONFIG_DIR``: the directory the CLI process sees the ``examples/HEMS/configs/`` files at, if different from their local path (e.g. because that directory is bind-mounted into a container at a different path). + Reports are triggered over the API, so the client script needs nothing beyond its API credentials: no local CLI, no database access, and no bind-mount of ``examples/HEMS/configs/`` into the server. Those configuration files are read by the client and posted along with each report request. The server does need a worker on the ``reporting`` queue, as above, or reports will stay queued until the client gives up on them. From 3f0b0733363bec8d57f152988a8c04af0e35e835 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:03:46 +0100 Subject: [PATCH 10/15] chore: stop ignoring generated reporter parameter files Reporter parameters are built in memory now, so nothing writes configs/*_reporter_param.json anymore. Signed-off-by: Mohamed Belhsan Hmida --- .gitignore | 3 --- 1 file changed, 3 deletions(-) diff --git a/.gitignore b/.gitignore index 81a97133..85d2d889 100644 --- a/.gitignore +++ b/.gitignore @@ -56,6 +56,3 @@ venv/* log HEMS data full *.pkl - -# Reporter parameters json files -*reporter_param.json From cdfc5ed352d644d2ff5679e81d9bbcce9962a441 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:04:40 +0100 Subject: [PATCH 11/15] docs: document the reporting client methods Mirror the forecasting page: endpoints and server requirement, which asset a report is triggered against, the polling knobs, and what JobFailedError and JobTimeoutError each mean. Signed-off-by: Mohamed Belhsan Hmida --- docs/index.rst | 1 + docs/reporting.rst | 151 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 152 insertions(+) create mode 100644 docs/reporting.rst diff --git a/docs/index.rst b/docs/index.rst index 05174429..c8a70ca6 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -31,6 +31,7 @@ Contents Overview Forecasting + Reporting Contributions & Help License Authors diff --git a/docs/reporting.rst b/docs/reporting.rst new file mode 100644 index 00000000..51911118 --- /dev/null +++ b/docs/reporting.rst @@ -0,0 +1,151 @@ +.. _reporting: + +Reporting +========= + +The FlexMeasures Client supports the report trigger endpoint introduced in +FlexMeasures v1.1.0: + +- ``POST /assets//reports/trigger`` — queue a one-off reporting job +- ``GET /jobs/`` — poll a background job + +These are exposed through four client methods: + +- :meth:`trigger_report` — trigger and return the job UUID +- :meth:`get_job_status` — look a job up once +- :meth:`wait_for_job` — poll a job until it reaches a terminal state +- :meth:`trigger_and_await_report` — convenience wrapper for triggering and waiting + +.. note:: + + These endpoints require a FlexMeasures server of version **1.1.0** or above, + running a worker on the ``reporting`` queue:: + + flexmeasures jobs run-worker --queue reporting + + +Basic example +------------- + +Compute a report over yesterday, writing its result to a sensor: + +.. code-block:: python + + import asyncio + from flexmeasures_client import FlexMeasuresClient + + async def main(): + client = FlexMeasuresClient( + host="localhost:5000", + ssl=False, + email="user@example.com", + password="password", + ) + + await client.trigger_and_await_report( + asset_id=3, + reporter="AggregatorReporter", + config={"method": "sum"}, + parameters={ + "input": [{"sensor": 1}, {"sensor": 2}], + "output": [{"sensor": 4}], + "start": "2026-08-17T00:00:00+02:00", + "end": "2026-08-18T00:00:00+02:00", + }, + ) + + # Reports write to their output sensors, so read the result back + values = await client.get_sensor_data( + sensor_id=4, + start="2026-08-17T00:00:00+02:00", + duration="P1D", + unit="MW", + resolution="PT15M", + ) + print(values) + + await client.close() + + asyncio.run(main()) + + +Which asset to trigger against +------------------------------ + +Every output sensor has to belong to the asset in the URL or one of its +descendants, and the caller needs to be able to read the input sensors and to +record data on the output sensors. So a report writing to sensors of one site +is triggered against that site, and a report aggregating several sites into a +community sensor is triggered against the community. + + +Step-by-step usage +------------------- + +Trigger and wait separately to handle the job UUID yourself: + +.. code-block:: python + + # Step 1 – enqueue the reporting job + job_id = await client.trigger_report( + asset_id=3, + reporter="AggregatorReporter", + config={"method": "sum"}, + parameters=parameters, + ) + print(f"Job queued: {job_id}") + + # Step 2 – look it up whenever you like + job = await client.get_job_status(job_id) + print(job["status"]) # QUEUED, STARTED, FINISHED, FAILED, ... + + # Step 3 – or block until it reaches a terminal state + job = await client.wait_for_job(job_id) + + +Polling behaviour +----------------- + +``wait_for_job`` polls ``GET /jobs/`` until the job reaches a terminal +state. Waits back off exponentially, so short jobs are picked up quickly +without hammering the server on long ones: + +- ``polling_interval`` (default 2 s) — wait before the first poll +- ``max_polling_interval`` (default 30 s) — cap on the backing-off wait +- ``timeout`` (default 600 s) — total budget for the job to finish + +.. code-block:: python + + job = await client.wait_for_job( + job_id, + timeout=3600.0, # allow an hour for a heavy report + max_polling_interval=60.0, + ) + + +Error handling +-------------- + +Jobs that end badly raise rather than returning a status: + +- :class:`JobFailedError` — the job ended as ``FAILED``, ``STOPPED`` or + ``CANCELED``. The message carries the server's own message and, where the + worker stored one, its traceback. +- :class:`JobTimeoutError` — the job did not finish within ``timeout``. The + message names the last status seen, which distinguishes a report stuck in + ``QUEUED`` (no worker on the ``reporting`` queue) from one still ``STARTED``. + +.. code-block:: python + + from flexmeasures_client.exceptions import JobFailedError, JobTimeoutError + + try: + await client.trigger_and_await_report(...) + except JobFailedError as exception: + print(f"The report did not compute: {exception}") + except JobTimeoutError as exception: + print(f"The report is taking too long: {exception}") + +A request the server rejects outright — unknown reporter, malformed +parameters, an output sensor outside the asset's subtree — raises ``ValueError`` +from :meth:`trigger_report`, before any job is queued. From 63da8be6eeb43104e4d12ce85985877fdf407525 Mon Sep 17 00:00:00 2001 From: Mohamed Belhsan Hmida Date: Tue, 18 Aug 2026 02:07:06 +0100 Subject: [PATCH 12/15] test(hems): cover the example's reporting helpers Assert that reports are scoped to the asset owning their output sensors, that mixed-asset and asset-less outputs are refused, that parameters and configs come out as the reporters expect, and that no HEMS module shells out anymore. Signed-off-by: Mohamed Belhsan Hmida --- tests/examples/test_hems_reporting.py | 166 ++++++++++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 tests/examples/test_hems_reporting.py diff --git a/tests/examples/test_hems_reporting.py b/tests/examples/test_hems_reporting.py new file mode 100644 index 00000000..5f6a6c10 --- /dev/null +++ b/tests/examples/test_hems_reporting.py @@ -0,0 +1,166 @@ +"""Tests for the HEMS example's reporting helpers.""" + +import sys +from pathlib import Path + +import pytest + +HEMS_DIR = Path(__file__).parents[2] / "examples" / "HEMS" +if str(HEMS_DIR) not in sys.path: + sys.path.insert(0, str(HEMS_DIR)) + +from utils import reporter_utils # noqa: E402 +from utils.reporter_utils import ( # noqa: E402 + asset_id_for_outputs, + build_reporter_parameters, + load_reporter_config, +) + + +def test_reports_do_not_shell_out() -> None: + """Test that report generation no longer invokes the CLI.""" + source = (HEMS_DIR / "utils" / "reporter_utils.py").read_text() + assert "subprocess" not in source + assert "add report" not in source + for module in ("reporters.py", "scheduling.py"): + assert "subprocess" not in (HEMS_DIR / module).read_text() + + +@pytest.mark.parametrize( + "reporter_type", ["aggregate", "self-consumption", "total-energy-costs"] +) +def test_load_reporter_config(reporter_type: str) -> None: + """Test that every reporter used by the example has a loadable config.""" + config = load_reporter_config(reporter_type) + assert isinstance(config, dict) + assert config + + +def test_build_reporter_parameters() -> None: + """Test the parameters built for a multi-output report.""" + parameters = build_reporter_parameters( + input_sensors=[{"pv-power": 1}, {"building-consumption": 2}], + output_sensors=[ + {"name": "self-consumption", "id": 3}, + {"name": "daily-share-of-self-consumption", "id": 4}, + ], + start="2026-08-17T00:00:00+02:00", + end="2026-08-18T00:00:00+02:00", + reporter_type="self-consumption", + ) + + assert parameters["input"] == [ + { + "name": "pv-power", + "sensor": 1, + "exclude_source_types": ["scheduler", "forecaster"], + }, + { + "name": "building-consumption", + "sensor": 2, + "exclude_source_types": ["scheduler", "forecaster"], + }, + ] + assert parameters["output"] == [ + {"name": "self-consumption", "sensor": 3}, + {"name": "daily-share-of-self-consumption", "sensor": 4}, + ] + assert parameters["start"] == "2026-08-17T00:00:00+02:00" + assert parameters["end"] == "2026-08-18T00:00:00+02:00" + + +def test_build_reporter_parameters_for_aggregate() -> None: + """Test that the aggregate reporter writes to a single, unnamed output.""" + parameters = build_reporter_parameters( + input_sensors=[{"pv": 1}], + output_sensors={"name": "electricity-aggregate", "id": 9}, + start="2026-08-17T00:00:00+02:00", + end="2026-08-18T00:00:00+02:00", + reporter_type="aggregate", + ) + + assert parameters["output"] == [{"sensor": 9}] + + +def test_asset_id_for_outputs() -> None: + """Test that a report is scoped to the asset owning its output sensors.""" + assert ( + asset_id_for_outputs( + [ + {"id": 3, "generic_asset_id": 7}, + {"id": 4, "generic_asset_id": 7}, + ] + ) + == 7 + ) + assert asset_id_for_outputs({"id": 9, "generic_asset_id": 2}) == 2 + + +def test_asset_id_for_outputs_rejects_mixed_assets() -> None: + """Test that outputs spread over several assets are refused. + + The API requires every output to sit in the subtree of the asset in the + URL, so such a report has to be split up. + """ + with pytest.raises(ValueError, match="one asset"): + asset_id_for_outputs( + [ + {"id": 3, "generic_asset_id": 7}, + {"id": 4, "generic_asset_id": 8}, + ] + ) + + +def test_asset_id_for_outputs_rejects_sensors_without_asset() -> None: + """Test the guard for sensors dumped as part of an asset listing.""" + with pytest.raises(ValueError, match="generic_asset_id"): + asset_id_for_outputs({"id": 9, "name": "power"}) + + +@pytest.mark.asyncio +async def test_run_report_reports_failures() -> None: + """Test that a failing report is reported rather than raised.""" + + class FailingClient: + async def trigger_and_await_report(self, **kwargs): + raise reporter_utils.JobFailedError("Job ended with status FAILED.") + + succeeded = await reporter_utils.run_report( + client=FailingClient(), + reporter="AggregatorReporter", + reporter_type="aggregate", + input_sensors=[{"pv": 1}], + output_sensors={"id": 9, "generic_asset_id": 2}, + start="2026-08-17T00:00:00+02:00", + end="2026-08-18T00:00:00+02:00", + ) + + assert succeeded is False + + +@pytest.mark.asyncio +async def test_run_report_triggers_against_the_owning_asset() -> None: + """Test what run_report() sends for a site report.""" + calls = [] + + class RecordingClient: + async def trigger_and_await_report(self, **kwargs): + calls.append(kwargs) + return {"status": "FINISHED"} + + succeeded = await reporter_utils.run_report( + client=RecordingClient(), + reporter="AggregatorReporter", + reporter_type="aggregate", + input_sensors=[{"pv": 1}, {"consumption": 2}], + output_sensors={"id": 9, "generic_asset_id": 2}, + start="2026-08-17T00:00:00+02:00", + end="2026-08-18T00:00:00+02:00", + ) + + assert succeeded is True + assert len(calls) == 1 + assert calls[0]["asset_id"] == 2 + assert calls[0]["reporter"] == "AggregatorReporter" + assert calls[0]["config"] == load_reporter_config("aggregate") + assert calls[0]["parameters"]["output"] == [{"sensor": 9}] From 910bb0f47654df0a5bac5c3b252fa0ba0b274c10 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicolas=20H=C3=B6ning?= Date: Fri, 4 Sep 2026 16:12:15 +0200 Subject: [PATCH 13/15] also document scheduling with its own .rst, test/parse Python code in these docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Nicolas Höning --- docs/forecasting.rst | 6 +- docs/index.rst | 1 + docs/scheduling.rst | 240 ++++++++++++++++++++ tests/client/test_documentation_examples.py | 130 +++++++++++ 4 files changed, 374 insertions(+), 3 deletions(-) create mode 100644 docs/scheduling.rst create mode 100644 tests/client/test_documentation_examples.py diff --git a/docs/forecasting.rst b/docs/forecasting.rst index 5bc3dacd..48cd347d 100644 --- a/docs/forecasting.rst +++ b/docs/forecasting.rst @@ -129,10 +129,10 @@ Polling behaviour ----------------- ``get_forecast`` polls the server with a ``GET`` request and returns when the -server responds with HTTP 200. The polling respects the same client-level -settings as scheduling: +server responds with HTTP 200. Polling uses exponential backoff and respects +the same client-level settings as scheduling: -- ``polling_interval`` (default 10 s) — time between retries +- ``polling_interval`` (default 10 s) — initial wait between retries - ``polling_timeout`` (default 200 s) — maximum total wait time - ``max_polling_steps`` (default 10) — maximum number of poll attempts diff --git a/docs/index.rst b/docs/index.rst index c8a70ca6..be36ba56 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -31,6 +31,7 @@ Contents Overview Forecasting + Scheduling Reporting Contributions & Help License diff --git a/docs/scheduling.rst b/docs/scheduling.rst new file mode 100644 index 00000000..9cd09c22 --- /dev/null +++ b/docs/scheduling.rst @@ -0,0 +1,240 @@ +.. _scheduling: + +Scheduling +========== + +The FlexMeasures Client supports the scheduling API endpoints: + +- ``POST /assets//schedules/trigger`` — queue a scheduling job +- ``GET /sensors//schedules/`` — poll for a sensor's schedule + +These are exposed through three client methods: + +- :meth:`trigger_schedule` — trigger and return the schedule UUID +- :meth:`get_schedule` — poll until a sensor's schedule is ready +- :meth:`trigger_and_get_schedule` — convenience wrapper for both + +.. note:: + + The asset trigger endpoint requires FlexMeasures **v0.27.0** or above and + a worker listening on the ``scheduling`` queue:: + + flexmeasures jobs run-worker --queue scheduling + + +Basic example +------------- + +Schedule a single storage device against an electricity price. ``sensor_id`` +is the device's power sensor; the client resolves the asset that owns it and +adds the sensor to the flex model sent to the asset endpoint: + +.. code-block:: python + + import asyncio + from flexmeasures_client import FlexMeasuresClient + + async def main(): + client = FlexMeasuresClient( + host="localhost:5000", + ssl=False, + email="user@example.com", + password="password", + ) + + schedule = await client.trigger_and_get_schedule( + sensor_id=8, + start="2026-09-05T08:00:00+02:00", + duration="PT12H", + flex_context={ + "consumption-price": {"sensor": 7}, + }, + flex_model={ + "soc-unit": "kWh", + "soc-at-start": 50, + "soc-min": 10, + "soc-max": 100, + "power-capacity": "20 kW", + "soc-targets": [ + { + "value": 80, + "datetime": "2026-09-05T18:00:00+02:00", + }, + ], + }, + unit="kW", + ) + print(schedule) + # e.g. {"values": [...], "start": "...", "duration": "PT12H", "unit": "kW"} + + await client.close() + + asyncio.run(main()) + + +Scheduling multiple devices +--------------------------- + +Pass the ID of the common parent asset and one flex-model entry per power +sensor to optimize several devices together. Every referenced sensor must +belong to that asset or one of its descendants: + +.. code-block:: python + + schedules = await client.trigger_and_get_schedule( + asset_id=3, + start="2026-09-05T08:00:00+02:00", + duration="PT12H", + flex_context={ + "consumption-price": {"sensor": 7}, + }, + flex_model=[ + { + "sensor": 8, + "soc-unit": "kWh", + "soc-at-start": 50, + "soc-min": 10, + "soc-max": 100, + "power-capacity": "20 kW", + }, + { + "sensor": 9, + "consumption-capacity": "0 kW", + "production-capacity": {"sensor": 9}, + }, + ], + unit="kW", + ) + + for schedule in schedules: + print(schedule["sensor"], schedule["values"]) + +The convenience method returns one schedule dictionary for ``sensor_id`` and +a list of dictionaries for ``asset_id``. Each item in the latter is tagged +with its ``sensor`` ID. + + +Using stored flexibility +------------------------ + +Flex context and flex models may already be configured on the server's asset +tree. The ``flex_context`` and ``flex_model`` arguments can be omitted to use +that configuration, or supplied to add to or override it for one request. + +When using ``asset_id``, :meth:`trigger_and_get_schedule` needs a ``flex_model`` +list to know which sensor schedules to retrieve. If you rely entirely on +stored flex models, trigger first and then retrieve the known power sensors +explicitly, as shown below. + + +Step-by-step usage +------------------ + +Trigger and retrieve separately to keep the schedule UUID or to fetch several +output sensors, such as both power and state of charge: + +.. code-block:: python + + # Step 1 – enqueue one joint scheduling job + schedule_id = await client.trigger_schedule( + asset_id=3, + start="2026-09-05T08:00:00+02:00", + duration="PT12H", + flex_model=[{"sensor": 8}, {"sensor": 9}], + ) + print(f"Job queued: {schedule_id}") + + # Step 2 – retrieve a result for each relevant sensor + battery_power = await client.get_schedule( + sensor_id=8, + schedule_id=schedule_id, + duration="PT12H", + unit="kW", + ) + battery_soc = await client.get_schedule( + sensor_id=10, + schedule_id=schedule_id, + duration="PT12H", + unit="kWh", + ) + + +Belief time and repeated scheduling +----------------------------------- + +Set ``prior`` to restrict the scheduler to sensor data recorded before that +time. Supplying it also asks the server to create a new job rather than reuse +a cached schedule for the same window: + +.. code-block:: python + + schedule = await client.trigger_and_get_schedule( + sensor_id=8, + start="2026-09-05T08:00:00+02:00", + duration="PT12H", + prior="2026-09-05T07:55:00+02:00", + ) + +This is useful for rolling or simulated scheduling, where each run represents +a new information horizon. + + +Selecting a custom scheduler +---------------------------- + +Pass ``scheduler`` together with ``asset_id`` to select a server-side custom +scheduler: + +.. code-block:: python + + schedule_id = await client.trigger_schedule( + asset_id=3, + start="2026-09-05T08:00:00+02:00", + duration="PT12H", + scheduler="MyCustomScheduler", + ) + +The client implements this by updating the asset's ``custom-scheduler`` +attribute before triggering. This is a persistent asset update, not merely a +field on this scheduling request. + + +Units +----- + +Pass ``unit`` to :meth:`get_schedule` or :meth:`trigger_and_get_schedule` to +request the desired output unit. FlexMeasures v0.32.0 and newer perform this +conversion server-side. For older servers, the client converts between +``W``, ``kW`` and ``MW`` locally; other client-side conversions raise +``NotImplementedError``. + + +Polling and errors +------------------ + +``get_schedule`` polls the sensor schedule endpoint until the result is ready. +Polling uses exponential backoff and is controlled by these client settings: + +- ``polling_interval`` (default 10 s) — initial wait between attempts +- ``polling_timeout`` (default 200 s) — maximum total wait +- ``max_polling_steps`` (default 10) — maximum number of attempts + +Override them when constructing the client: + +.. code-block:: python + + client = FlexMeasuresClient( + ..., + polling_interval=5.0, + polling_timeout=300.0, + max_polling_steps=12, + ) + +A scheduling job rejected or reported as failed by the server raises +``ValueError`` with the server's message. Connection and polling timeouts +raise ``ConnectionError``. + +Schedule, forecast and report triggers share the server's computation rate +limit. A server running simulations should use ``FLEXMEASURES_MODE = "play"``; +production deployments can configure the server-wide limit or assign the +account an appropriate plan. diff --git a/tests/client/test_documentation_examples.py b/tests/client/test_documentation_examples.py new file mode 100644 index 00000000..ce0dbb6c --- /dev/null +++ b/tests/client/test_documentation_examples.py @@ -0,0 +1,130 @@ +"""Compile API-guided Python blocks and verify the examples' client calls run without error.""" + +from __future__ import annotations + +import ast +import inspect +import re +from pathlib import Path + +import pytest + +import flexmeasures_client + +DOCS_DIR = Path(__file__).parents[2] / "docs" +GUIDES = ("forecasting.rst", "scheduling.rst", "reporting.rst") +REAL_CLIENT = flexmeasures_client.FlexMeasuresClient + + +def python_blocks(path: Path) -> list[str]: + """Extract Python code blocks while removing their directive indentation.""" + matches = re.findall( + r"\.\. code-block:: python\n" + r"(?:\n|[ \t]+:[^\n]*\n)*\n" + r"((?:(?: ).*(?:\n|$)|\n)+)", + path.read_text(), + ) + return [ + "\n".join( + line[4:] if line.startswith(" ") else line + for line in match.rstrip().splitlines() + ) + for match in matches + ] + + +@pytest.mark.parametrize("guide", GUIDES) +def test_python_blocks_compile(guide: str) -> None: + """Catch invalid Python in both complete examples and shorter snippets.""" + blocks = python_blocks(DOCS_DIR / guide) + assert blocks + for block_number, source in enumerate(blocks, start=1): + compile( + source, + f"{guide}:code-block-{block_number}", + "exec", + flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT, + ) + + +class ExampleClient: + """Small API stand-in used to execute each guide's complete basic example.""" + + instances: list[ExampleClient] = [] + + def __init__(self, **connection): + inspect.signature(REAL_CLIENT).bind(**connection) + self.connection = connection + self.calls: list[tuple[str, dict]] = [] + self.closed = False + self.instances.append(self) + + async def trigger_and_get_forecast(self, **kwargs) -> dict: + inspect.signature(REAL_CLIENT.trigger_and_get_forecast).bind(None, **kwargs) + self.calls.append(("trigger_and_get_forecast", kwargs)) + return { + "values": [1.2, 1.5, 1.8], + "start": "2026-09-05T08:00:00+02:00", + "duration": "PT24H", + "unit": "kW", + } + + async def trigger_and_get_schedule(self, **kwargs) -> dict: + inspect.signature(REAL_CLIENT.trigger_and_get_schedule).bind(None, **kwargs) + self.calls.append(("trigger_and_get_schedule", kwargs)) + return { + "values": [2.0, -1.0], + "start": "2026-09-05T08:00:00+02:00", + "duration": "PT12H", + "unit": "kW", + } + + async def trigger_and_await_report(self, **kwargs) -> dict: + inspect.signature(REAL_CLIENT.trigger_and_await_report).bind(None, **kwargs) + self.calls.append(("trigger_and_await_report", kwargs)) + return {"status": "FINISHED"} + + async def get_sensor_data(self, **kwargs) -> dict: + inspect.signature(REAL_CLIENT.get_sensor_data).bind(None, **kwargs) + self.calls.append(("get_sensor_data", kwargs)) + return { + "values": [3.0], + "start": "2026-08-17T00:00:00+02:00", + "duration": "P1D", + "unit": "MW", + } + + async def close(self) -> None: + self.closed = True + + +@pytest.mark.parametrize( + ("guide", "expected_methods"), + ( + ("forecasting.rst", ["trigger_and_get_forecast"]), + ("scheduling.rst", ["trigger_and_get_schedule"]), + ( + "reporting.rst", + ["trigger_and_await_report", "get_sensor_data"], + ), + ), +) +def test_basic_example_runs( + monkeypatch: pytest.MonkeyPatch, guide: str, expected_methods: list[str] +) -> None: + """Execute the complete first Python example without requiring a server.""" + ExampleClient.instances = [] + monkeypatch.setattr(flexmeasures_client, "FlexMeasuresClient", ExampleClient) + + source = python_blocks(DOCS_DIR / guide)[0] + exec(compile(source, guide, "exec"), {"__name__": "__documentation_example__"}) + + client = ExampleClient.instances[-1] + assert [method for method, _ in client.calls] == expected_methods + assert client.connection == { + "host": "localhost:5000", + "ssl": False, + "email": "user@example.com", + "password": "password", + } + assert client.closed From a786990e83a1530a70a23f0ac50bb562ed4e29c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicolas=20H=C3=B6ning?= Date: Sat, 5 Sep 2026 01:56:13 +0200 Subject: [PATCH 14/15] refactor get_job_status() / wait_for_job() loop across forecasting and scheduling; update code in Readme and also test it syntax-wise MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Nicolas Höning --- README.rst | 61 +++++++------ docs/forecasting.rst | 47 +++++++--- docs/reporting.rst | 7 +- docs/scheduling.rst | 49 ++++++++--- src/flexmeasures_client/client.py | 84 +++++++++++++++--- src/flexmeasures_client/response_handling.py | 5 +- tests/client/test_documentation_examples.py | 15 ++-- tests/client/test_forecast.py | 90 ++++++++++++++++++++ tests/client/test_report.py | 31 +++++-- tests/client/test_schedule.py | 54 ++++++++++++ 10 files changed, 370 insertions(+), 73 deletions(-) diff --git a/README.rst b/README.rst index caee6f5a..895f1121 100644 --- a/README.rst +++ b/README.rst @@ -110,7 +110,7 @@ Post a measurement from a sensor: .. code-block:: python await client.post_sensor_data( - sensor_id=, # integer + sensor_id=1, start="2023-03-26T10:00+02:00", # ISO datetime duration="PT6H", # ISO duration values=[1, 2, 3, 4], # list @@ -162,38 +162,41 @@ Scheduling With FlexMeasures a schedule can be requested to optimize at what time the flexible assets can be activated to optimize for price of energy or emissions. -The calculation of the schedule can take some time depending on the complexity of the calculations. A polling function is used to check if a schedule is available after triggering the schedule. +The calculation of a schedule can take some time. On FlexMeasures v0.33.0 and +newer, the convenience method waits on the generic job-status endpoint before +retrieving the schedule values. It falls back to result-endpoint polling on +older servers. Trigger and retrieve a schedule for multiple devices: .. code-block:: python - schedule = await flexmeasures_client.trigger_and_get_schedule( - asset_id=, # the asset ID (int) of the asset that all relevant power sensors belong to (or live under, in case of a tree-like asset structure) - start="2023-03-26T10:00+02:00", # ISO datetime + schedules = await client.trigger_and_get_schedule( + asset_id=3, + start="2026-09-05T08:00+02:00", duration="PT12H", # ISO duration flex_context={ - "consumption-price": {"sensor": }, # int + "consumption-price": {"sensor": 7}, }, - flex-model=[ + flex_model=[ # Example flex-model for an electric truck at a regular Charge Point { - "sensor": , # int + "sensor": 8, "power-capacity": "22 kVA", "production-capacity": "0 kW", "soc-at-start": "50 kWh", "soc-max": "400 kWh", "soc-min": "20 kWh", "soc-targets": [ - {"value": "100 kWh", "datetime": "2023-03-03T11:00+02:00"}, + {"value": "100 kWh", "datetime": "2026-09-05T18:00+02:00"}, ], }, # Example flex-model for curtailable solar panels { - "sensor": , # int + "sensor": 9, "power-capacity": "20 kVA", "consumption-capacity": "0 kW", - "production-capacity": {"sensor": }, # int + "production-capacity": {"sensor": 9}, }, ], ) @@ -203,19 +206,19 @@ Alternatively, use a single-device flex-model (no list) and move the device's po .. code-block:: python - schedule = await flexmeasures_client.trigger_and_get_schedule( - sensor_id=, # int - start="2023-03-26T10:00+02:00", # ISO datetime + schedule = await client.trigger_and_get_schedule( + sensor_id=8, + start="2026-09-05T08:00+02:00", duration="PT12H", # ISO duration flex_context={ - "consumption-price": {"sensor": }, # int + "consumption-price": {"sensor": 7}, }, - flex-model={ + flex_model={ "soc-at-start": "50 kWh", "soc-max": "400 kWh", "soc-min": "20 kWh", "soc-targets": [ - {"value": "100 kWh", "datetime": "2023-03-03T11:00+02:00"}, + {"value": "100 kWh", "datetime": "2026-09-05T18:00+02:00"}, ], }, ) @@ -226,22 +229,26 @@ Trigger a schedule: .. code-block:: python - schedule_uuid = await flexmeasures_client.trigger_schedule( + schedule_uuid = await client.trigger_schedule( **kwargs, # same kwargs as previous example ) The ``trigger_schedule`` method returns a ``schedule_uuid``. -This can be used to retrieve the schedule, using: +On FlexMeasures v0.33.0 and newer, wait for the job once before retrieving one +or more sensor results: .. code-block:: python - schedule = await flexmeasures_client.get_schedule( - sensor_id=, # int - schedule_id="", # uuid + await client.wait_for_job(schedule_uuid) + + schedule = await client.get_schedule( + sensor_id=8, + schedule_id=schedule_uuid, duration="PT45M", # ISO duration ) -The client will re-try until the schedule is available or the ``MAX_POLLING_STEPS`` of ``10`` is reached. +For the complete scheduling API, including multi-device results, job timeouts, +and compatibility with older servers, see :doc:`scheduling`. Forecasting @@ -252,13 +259,15 @@ Trigger a forecast for a sensor and wait for the result: .. code-block:: python forecast = await client.trigger_and_get_forecast( - sensor_id=, # int + sensor_id=1, duration="PT24H", # ISO duration – how far ahead to forecast ) # Returns e.g. {"values": [1.2, 1.5, ...], "start": "...", "duration": "PT24H", "unit": "kW"} -The client polls until the forecasting job is complete. For more advanced options -(training window, regressors, forecast frequency, etc.) see :doc:`forecasting`. +On FlexMeasures v0.33.0 and newer, the client polls the generic job endpoint +until the forecasting job is complete, then retrieves its values. For more +advanced options (training window, regressors, forecast frequency, etc.) see +:doc:`forecasting`. Development diff --git a/docs/forecasting.rst b/docs/forecasting.rst index 48cd347d..acca2704 100644 --- a/docs/forecasting.rst +++ b/docs/forecasting.rst @@ -7,17 +7,19 @@ The FlexMeasures Client supports the forecasting API endpoints introduced in FlexMeasures v0.31.0: - ``POST /sensors//forecasts/trigger`` — queue a forecasting job -- ``GET /sensors//forecasts/`` — poll for results +- ``GET /jobs/`` — inspect the job (v0.33.0+) +- ``GET /sensors//forecasts/`` — retrieve the result These are exposed through three client methods: - :meth:`trigger_forecast` — trigger and return the job UUID -- :meth:`get_forecast` — poll until results are ready -- :meth:`trigger_and_get_forecast` — convenience wrapper for both +- :meth:`get_forecast` — retrieve results, with legacy result polling when needed +- :meth:`trigger_and_get_forecast` — trigger, wait, and retrieve .. note:: - These endpoints require a FlexMeasures server of version **0.31.0** or above. + Forecasting requires a FlexMeasures server of version **0.31.0** or above. + The generic job status endpoint is available from **v0.33.0**. Basic example @@ -117,7 +119,10 @@ Trigger and retrieve separately to handle the job UUID yourself: ) print(f"Job queued: {forecast_id}") - # Step 2 – poll until the job finishes + # Step 2 – wait for the job itself to finish (FlexMeasures v0.33.0+) + await client.wait_for_job(forecast_id) + + # Step 3 – retrieve the forecast values forecast = await client.get_forecast( sensor_id=1, forecast_id=forecast_id, @@ -125,18 +130,38 @@ Trigger and retrieve separately to handle the job UUID yourself: print(forecast) -Polling behaviour ------------------ +Waiting and legacy polling +-------------------------- + +On FlexMeasures v0.33.0 and newer, ``trigger_and_get_forecast`` waits through +``GET /jobs/`` and fetches the forecast values only after the job has +finished. Its job wait uses exponential backoff and accepts the same options +as :meth:`wait_for_job`: + +- ``polling_interval`` (default 2 s) — delay before a repeated status check +- ``max_polling_interval`` (default 30 s) — maximum delay between checks +- ``timeout`` (default 600 s) — total job wait budget + +For example: + +.. code-block:: python + + forecast = await client.trigger_and_get_forecast( + sensor_id=1, + duration="PT24H", + timeout=1800.0, + max_polling_interval=60.0, + ) -``get_forecast`` polls the server with a ``GET`` request and returns when the -server responds with HTTP 200. Polling uses exponential backoff and respects -the same client-level settings as scheduling: +``get_forecast`` still polls the result endpoint when it is called directly, +and ``trigger_and_get_forecast`` retains that behaviour for servers older than +v0.33.0. This legacy polling is controlled by client-level settings: - ``polling_interval`` (default 10 s) — initial wait between retries - ``polling_timeout`` (default 200 s) — maximum total wait time - ``max_polling_steps`` (default 10) — maximum number of poll attempts -Override them at client construction time: +Configure those settings at client construction time: .. code-block:: python diff --git a/docs/reporting.rst b/docs/reporting.rst index 51911118..403f2c06 100644 --- a/docs/reporting.rst +++ b/docs/reporting.rst @@ -110,7 +110,7 @@ Polling behaviour state. Waits back off exponentially, so short jobs are picked up quickly without hammering the server on long ones: -- ``polling_interval`` (default 2 s) — wait before the first poll +- ``polling_interval`` (default 2 s) — delay before a repeated status check - ``max_polling_interval`` (default 30 s) — cap on the backing-off wait - ``timeout`` (default 600 s) — total budget for the job to finish @@ -149,3 +149,8 @@ Jobs that end badly raise rather than returning a status: A request the server rejects outright — unknown reporter, malformed parameters, an output sensor outside the asset's subtree — raises ``ValueError`` from :meth:`trigger_report`, before any job is queued. + +The job endpoint uses HTTP 202 for jobs still in progress and HTTP 422 for a +failed job. :meth:`get_job_status` returns both responses as status dictionaries +after one request; :meth:`wait_for_job` is responsible for polling and turning +unsuccessful terminal states into :class:`JobFailedError`. diff --git a/docs/scheduling.rst b/docs/scheduling.rst index 9cd09c22..62191ccb 100644 --- a/docs/scheduling.rst +++ b/docs/scheduling.rst @@ -6,17 +6,20 @@ Scheduling The FlexMeasures Client supports the scheduling API endpoints: - ``POST /assets//schedules/trigger`` — queue a scheduling job -- ``GET /sensors//schedules/`` — poll for a sensor's schedule +- ``GET /jobs/`` — inspect the job (v0.33.0+) +- ``GET /sensors//schedules/`` — retrieve a sensor's schedule These are exposed through three client methods: - :meth:`trigger_schedule` — trigger and return the schedule UUID -- :meth:`get_schedule` — poll until a sensor's schedule is ready -- :meth:`trigger_and_get_schedule` — convenience wrapper for both +- :meth:`get_schedule` — retrieve one sensor's result, with legacy polling +- :meth:`trigger_and_get_schedule` — trigger, wait, and retrieve .. note:: The asset trigger endpoint requires FlexMeasures **v0.27.0** or above and + the generic job status endpoint is available from **v0.33.0**. Scheduling + also needs a worker listening on the ``scheduling`` queue:: flexmeasures jobs run-worker --queue scheduling @@ -144,7 +147,10 @@ output sensors, such as both power and state of charge: ) print(f"Job queued: {schedule_id}") - # Step 2 – retrieve a result for each relevant sensor + # Step 2 – wait once for the joint job (FlexMeasures v0.33.0+) + await client.wait_for_job(schedule_id) + + # Step 3 – retrieve a result for each relevant sensor battery_power = await client.get_schedule( sensor_id=8, schedule_id=schedule_id, @@ -209,11 +215,30 @@ conversion server-side. For older servers, the client converts between ``NotImplementedError``. -Polling and errors ------------------- +Waiting, legacy polling, and errors +----------------------------------- + +On FlexMeasures v0.33.0 and newer, ``trigger_and_get_schedule`` waits through +``GET /jobs/`` once, then retrieves the requested sensor result or +results. Its job wait is controlled by method arguments: + +- ``polling_interval`` (default 2 s) — delay before a repeated status check +- ``max_polling_interval`` (default 30 s) — maximum delay between checks +- ``timeout`` (default 600 s) — total job wait budget + +.. code-block:: python + + schedule = await client.trigger_and_get_schedule( + sensor_id=8, + start="2026-09-05T08:00:00+02:00", + duration="PT12H", + timeout=1800.0, + max_polling_interval=60.0, + ) -``get_schedule`` polls the sensor schedule endpoint until the result is ready. -Polling uses exponential backoff and is controlled by these client settings: +``get_schedule`` still polls the sensor result endpoint when called directly, +and ``trigger_and_get_schedule`` retains that behaviour for servers older than +v0.33.0. This legacy polling is controlled by client settings: - ``polling_interval`` (default 10 s) — initial wait between attempts - ``polling_timeout`` (default 200 s) — maximum total wait @@ -230,9 +255,11 @@ Override them when constructing the client: max_polling_steps=12, ) -A scheduling job rejected or reported as failed by the server raises -``ValueError`` with the server's message. Connection and polling timeouts -raise ``ConnectionError``. +With the jobs API, a job that ends as ``FAILED``, ``STOPPED``, or ``CANCELED`` +raises :class:`JobFailedError`; exceeding the job wait budget raises +:class:`JobTimeoutError`. A trigger rejected before a job is queued raises +``ValueError``. Direct and legacy result polling can raise ``ValueError`` or +``ConnectionError``. Schedule, forecast and report triggers share the server's computation rate limit. A server running simulations should use ``FLEXMEASURES_MODE = "play"``; diff --git a/src/flexmeasures_client/client.py b/src/flexmeasures_client/client.py index 78ffd87e..4bc71060 100644 --- a/src/flexmeasures_client/client.py +++ b/src/flexmeasures_client/client.py @@ -52,6 +52,9 @@ JOB_STATUS_FINISHED = "FINISHED" JOB_STATUS_UNSUCCESSFUL = frozenset({"FAILED", "STOPPED", "CANCELED"}) +# The generic job status endpoint landed in FlexMeasures v0.33.0 +JOB_STATUS_MIN_SERVER_VERSION = "0.33.0" + # The asset report trigger endpoint landed in FlexMeasures v1.1.0 REPORT_TRIGGER_MIN_SERVER_VERSION = "1.1.0" @@ -253,6 +256,7 @@ async def request( include_auth: bool = True, minimum_server_version: str | None = None, minimum_server_version_msg: str | None = None, + pass_through_statuses: frozenset[int] = frozenset(), ) -> tuple[dict | list, int]: """Send a request to FlexMeasures. @@ -264,6 +268,10 @@ async def request( Fails if: - the server response indicated a status code of 400 or higher - the client polling timed out (as indicated by the client's self.polling_timeout) + + ``pass_through_statuses`` is for callers that interpret particular HTTP + statuses themselves. Such responses are returned without generic + polling or error handling. """ # noqa: E501 url = self.build_url(uri, path=path) @@ -292,11 +300,12 @@ async def request( json_payload=json_payload, polling_step=polling_step, reauth_once=reauth_once, + pass_through_statuses=pass_through_statuses, ) if ( response.status < 300 - and polling_step == previous_polling_step - ): + or response.status in pass_through_statuses + ) and polling_step == previous_polling_step: break except asyncio.TimeoutError: sleep_interval = self.polling_interval * (2**polling_step) @@ -341,6 +350,7 @@ async def request_once( json_payload: dict | None = None, polling_step: int = 0, reauth_once: bool = True, + pass_through_statuses: frozenset[int] = frozenset(), ) -> tuple[ClientResponse, int, bool, URL]: url_msg = f"url: {url}" json_msg = f"payload: {json_payload}" @@ -390,10 +400,23 @@ async def request_once( self.server_version = header_version polling_step, reauth_once, url = await check_response( - self, response, polling_step, reauth_once, url, method=method + self, + response, + polling_step, + reauth_once, + url, + method=method, + pass_through_statuses=pass_through_statuses, ) return response, polling_step, reauth_once, url + def _supports_job_status_api(self) -> bool: + """Return whether the connected server exposes the generic jobs API.""" + return _server_version_at_least( + self.server_version, + JOB_STATUS_MIN_SERVER_VERSION, + ) + def ensure_session(self): """If there is no session, start one.""" if self.session is None: @@ -1003,9 +1026,16 @@ async def trigger_and_get_schedule( prior: datetime | None = None, scheduler: str | None = None, unit: str | None = None, + timeout: float = JOB_POLLING_TIMEOUT, + polling_interval: float = JOB_POLLING_INTERVAL, + max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, ) -> dict | list[dict]: """Trigger a schedule and then fetch it. + On FlexMeasures v0.33.0 and newer, wait for the scheduling job through + the generic jobs endpoint before fetching sensor results. Older servers + retain result-endpoint polling. + To schedule a single flexible device, use the sensor ID of its power sensor. To schedule a collection of flexible devices, use the asset ID of the asset on which the power sensors are registered. @@ -1014,6 +1044,9 @@ async def trigger_and_get_schedule( :param unit: desired unit for the schedule values (e.g. "W", "kW", "MW"). Passed through to get_schedule(); see its docstring for details. + :param timeout: total seconds to wait through the jobs endpoint. + :param polling_interval: seconds before the first repeated status lookup. + :param max_polling_interval: maximum delay between status lookups. :returns: For a single device, returns the schedule as a dictionary. For example: { @@ -1037,6 +1070,14 @@ async def trigger_and_get_schedule( scheduler=scheduler, ) + if self._supports_job_status_api(): + await self.wait_for_job( + job_id=schedule_id, + timeout=timeout, + polling_interval=polling_interval, + max_polling_interval=max_polling_interval, + ) + if sensor_id is not None: # Get the schedule for a single device return await self.get_schedule( @@ -1584,11 +1625,11 @@ async def trigger_schedule( f"Expected a dictionary, but got {type(response)}", ) - if not isinstance(response.get("schedule"), str): + schedule_id = response.get("job") or response.get("schedule") + if not isinstance(schedule_id, str): raise ContentTypeError( - f"Expected a schedule ID, but got {type(response.get('schedule'))}", + f"Expected a schedule job ID, but got {type(schedule_id)}", ) - schedule_id = response["schedule"] self.logger.info(f"Schedule triggered successfully. Schedule ID: {schedule_id}") return schedule_id @@ -1684,11 +1725,11 @@ async def trigger_forecast( f"Expected a dictionary, but got {type(response)}", ) - if not isinstance(response.get("forecast"), str): + forecast_id = response.get("job") or response.get("forecast") + if not isinstance(forecast_id, str): raise ContentTypeError( - f"Expected a forecast ID, but got {type(response.get('forecast'))}", + f"Expected a forecast job ID, but got {type(forecast_id)}", ) - forecast_id = response["forecast"] self.logger.info(f"Forecast triggered successfully. Forecast ID: {forecast_id}") return forecast_id @@ -1739,9 +1780,20 @@ async def trigger_and_get_forecast( max_forecast_horizon: str | timedelta | None = None, forecast_frequency: str | timedelta | None = None, probabilistic: bool | None = None, + timeout: float = JOB_POLLING_TIMEOUT, + polling_interval: float = JOB_POLLING_INTERVAL, + max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, ) -> dict: """Trigger a forecasting job and then fetch the result. + On FlexMeasures v0.33.0 and newer, wait for the forecasting job through + the generic jobs endpoint before fetching its values. Older servers + retain result-endpoint polling. + + :param timeout: total seconds to wait through the jobs endpoint. + :param polling_interval: seconds before the first repeated status lookup. + :param max_polling_interval: maximum delay between status lookups. + :returns: forecast as dictionary, for example: { 'values': [1.2, 1.5, 1.4, 0.8], @@ -1768,6 +1820,13 @@ async def trigger_and_get_forecast( forecast_frequency=forecast_frequency, probabilistic=probabilistic, ) + if self._supports_job_status_api(): + await self.wait_for_job( + job_id=forecast_id, + timeout=timeout, + polling_interval=polling_interval, + max_polling_interval=max_polling_interval, + ) return await self.get_forecast( sensor_id=sensor_id, forecast_id=forecast_id, @@ -1794,13 +1853,18 @@ async def get_job_status(self, job_id: str) -> dict: The ``status`` field is one of QUEUED, STARTED, FINISHED, FAILED, DEFERRED, SCHEDULED, STOPPED or CANCELED. + This method performs exactly one HTTP request. Pending (HTTP 202) and + failed (HTTP 422) job responses are returned for the caller to inspect. + This function raises a ValueError when an unhandled status code is returned. """ response, status = await self.request( uri=f"jobs/{job_id}", method="GET", + pass_through_statuses=frozenset({202, 422}), ) - check_for_status(status, 200) + if status not in {200, 202, 422}: + raise ValueError(f"Request failed with status code {status}") if not isinstance(response, dict): raise ContentTypeError( diff --git a/src/flexmeasures_client/response_handling.py b/src/flexmeasures_client/response_handling.py index 32803bf3..936cccc7 100644 --- a/src/flexmeasures_client/response_handling.py +++ b/src/flexmeasures_client/response_handling.py @@ -22,6 +22,7 @@ async def check_response( reauth_once: bool, url: URL, method: str = "GET", + pass_through_statuses: frozenset[int] = frozenset(), ) -> tuple[int, bool, URL]: """ <300: passes @@ -42,7 +43,9 @@ async def check_response( if payload is None: payload = {} headers = response.headers - if status == 202 and method.upper() == "GET": + if status in pass_through_statuses: + pass + elif status == 202 and method.upper() == "GET": sleep_interval = self.polling_interval * (2**polling_step) job_status = payload.get("status") message = "Server accepted the request but the result is not ready yet." diff --git a/tests/client/test_documentation_examples.py b/tests/client/test_documentation_examples.py index ce0dbb6c..cad45250 100644 --- a/tests/client/test_documentation_examples.py +++ b/tests/client/test_documentation_examples.py @@ -1,4 +1,4 @@ -"""Compile API-guided Python blocks and verify the examples' client calls run without error.""" +"""Check documented Python syntax plus basic-guide calls, arguments, and cleanup.""" from __future__ import annotations @@ -12,7 +12,12 @@ import flexmeasures_client DOCS_DIR = Path(__file__).parents[2] / "docs" +PROJECT_DIR = DOCS_DIR.parent GUIDES = ("forecasting.rst", "scheduling.rst", "reporting.rst") +PYTHON_DOCUMENTS = ( + *(DOCS_DIR / guide for guide in GUIDES), + PROJECT_DIR / "README.rst", +) REAL_CLIENT = flexmeasures_client.FlexMeasuresClient @@ -33,15 +38,15 @@ def python_blocks(path: Path) -> list[str]: ] -@pytest.mark.parametrize("guide", GUIDES) -def test_python_blocks_compile(guide: str) -> None: +@pytest.mark.parametrize("path", PYTHON_DOCUMENTS, ids=lambda path: path.name) +def test_python_blocks_compile(path: Path) -> None: """Catch invalid Python in both complete examples and shorter snippets.""" - blocks = python_blocks(DOCS_DIR / guide) + blocks = python_blocks(path) assert blocks for block_number, source in enumerate(blocks, start=1): compile( source, - f"{guide}:code-block-{block_number}", + f"{path.name}:code-block-{block_number}", "exec", flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT, ) diff --git a/tests/client/test_forecast.py b/tests/client/test_forecast.py index 081dcc37..92dbd5c2 100644 --- a/tests/client/test_forecast.py +++ b/tests/client/test_forecast.py @@ -1,7 +1,9 @@ import pytest from aioresponses import aioresponses +from yarl import URL from flexmeasures_client.client import FlexMeasuresClient +from flexmeasures_client.exceptions import JobFailedError @pytest.mark.asyncio @@ -261,3 +263,91 @@ async def test_trigger_and_get_forecast() -> None: assert forecast["unit"] == "kW" await flexmeasures_client.close() + + +@pytest.mark.asyncio +async def test_trigger_and_get_forecast_waits_via_jobs_api() -> None: + """Modern servers are polled through /jobs before results are fetched.""" + sensor_id = 1 + forecast_id = "test-forecast-uuid" + trigger_url = ( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/forecasts/trigger" + ) + job_url = f"http://localhost:5000/api/v3_0/jobs/{forecast_id}" + result_url = ( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/forecasts/{forecast_id}" + ) + + with aioresponses() as m: + client = FlexMeasuresClient( + email="test@test.test", + password="test", + access_token="test-token", + ) + m.post( + trigger_url, + status=202, + payload={"job": forecast_id, "status": "ACCEPTED"}, + headers={"FlexMeasures-Version": "0.33.0"}, + ) + m.get(job_url, status=202, payload={"status": "STARTED"}) + m.get(job_url, status=200, payload={"status": "FINISHED"}) + m.get( + result_url, + status=200, + payload={ + "values": [1.2, 1.5], + "start": "2025-01-05T00:00:00+00:00", + "duration": "PT2H", + "unit": "kW", + }, + ) + + forecast = await client.trigger_and_get_forecast( + sensor_id=sensor_id, + duration="PT2H", + polling_interval=0, + ) + + assert forecast["values"] == [1.2, 1.5] + assert len(m.requests[("GET", URL(job_url))]) == 2 + assert len(m.requests[("GET", URL(result_url))]) == 1 + await client.close() + + +@pytest.mark.asyncio +async def test_trigger_and_get_forecast_surfaces_job_failure() -> None: + """A failed modern forecasting job raises before fetching result data.""" + sensor_id = 1 + forecast_id = "failed-forecast-uuid" + trigger_url = ( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/forecasts/trigger" + ) + job_url = f"http://localhost:5000/api/v3_0/jobs/{forecast_id}" + + with aioresponses() as m: + client = FlexMeasuresClient( + email="test@test.test", + password="test", + access_token="test-token", + ) + m.post( + trigger_url, + status=202, + payload={"job": forecast_id, "status": "ACCEPTED"}, + headers={"FlexMeasures-Version": "0.33.0"}, + ) + m.get( + job_url, + status=422, + payload={"status": "FAILED", "message": "Training data is incomplete."}, + ) + + with pytest.raises(JobFailedError, match="Training data is incomplete"): + await client.trigger_and_get_forecast( + sensor_id=sensor_id, + duration="PT2H", + polling_interval=0, + ) + + await client.close() diff --git a/tests/client/test_report.py b/tests/client/test_report.py index a2ec740f..bc91d1d7 100644 --- a/tests/client/test_report.py +++ b/tests/client/test_report.py @@ -131,10 +131,10 @@ async def test_trigger_report_without_job_id() -> None: @pytest.mark.asyncio async def test_get_job_status() -> None: - """Test a single job status lookup.""" + """A pending job's HTTP 202 response is returned after one lookup.""" with aioresponses() as m: client = make_client() - m.get(JOB_URL, status=200, payload=job_payload("STARTED")) + m.get(JOB_URL, status=202, payload=job_payload("STARTED")) job = await client.get_job_status(JOB_ID) @@ -155,6 +155,21 @@ async def test_get_job_status() -> None: await client.close() +@pytest.mark.asyncio +async def test_get_job_status_returns_failed_job() -> None: + """A failed job's HTTP 422 response remains available to the caller.""" + with aioresponses() as m: + client = make_client() + m.get(JOB_URL, status=422, payload=job_payload("FAILED")) + + job = await client.get_job_status(JOB_ID) + + assert job["status"] == "FAILED" + assert len(m.requests[("GET", URL(JOB_URL))]) == 1 + + await client.close() + + @pytest.mark.asyncio async def test_get_job_status_without_status() -> None: """Test that a malformed job status response is reported clearly.""" @@ -173,9 +188,9 @@ async def test_wait_for_job_polls_until_finished() -> None: """Test that polling continues through the non-terminal states.""" with aioresponses() as m: client = make_client() - m.get(JOB_URL, status=200, payload=job_payload("QUEUED")) - m.get(JOB_URL, status=200, payload=job_payload("DEFERRED")) - m.get(JOB_URL, status=200, payload=job_payload("STARTED")) + m.get(JOB_URL, status=202, payload=job_payload("QUEUED")) + m.get(JOB_URL, status=202, payload=job_payload("DEFERRED")) + m.get(JOB_URL, status=202, payload=job_payload("STARTED")) m.get(JOB_URL, status=200, payload=job_payload("FINISHED")) job = await client.wait_for_job(JOB_ID, polling_interval=0.01) @@ -193,7 +208,7 @@ async def test_wait_for_job_raises_on_failed_job() -> None: client = make_client() m.get( JOB_URL, - status=200, + status=422, payload=job_payload( "FAILED", message="Report job failed.", @@ -218,7 +233,7 @@ async def test_wait_for_job_raises_on_unsuccessful_job(status: str) -> None: """Test that jobs stopped or canceled by an operator also raise.""" with aioresponses() as m: client = make_client() - m.get(JOB_URL, status=200, payload=job_payload(status)) + m.get(JOB_URL, status=202, payload=job_payload(status)) with pytest.raises(JobFailedError): await client.wait_for_job(JOB_ID, polling_interval=0.01) @@ -231,7 +246,7 @@ async def test_wait_for_job_times_out() -> None: """Test that a job that never finishes hits the timeout budget.""" with aioresponses() as m: client = make_client() - m.get(JOB_URL, status=200, payload=job_payload("QUEUED"), repeat=True) + m.get(JOB_URL, status=202, payload=job_payload("QUEUED"), repeat=True) with pytest.raises(JobTimeoutError) as exc_info: await client.wait_for_job(JOB_ID, timeout=0.05, polling_interval=0.01) diff --git a/tests/client/test_schedule.py b/tests/client/test_schedule.py index 7a3eb31a..ff6b76be 100644 --- a/tests/client/test_schedule.py +++ b/tests/client/test_schedule.py @@ -6,6 +6,7 @@ import pytest from aioresponses import aioresponses +from yarl import URL from flexmeasures_client.client import ContentTypeError, FlexMeasuresClient @@ -488,6 +489,59 @@ async def test_trigger_and_get_schedule_asset_id_flex_model_list(): await client.close() +@pytest.mark.asyncio +async def test_trigger_and_get_schedule_waits_once_via_jobs_api(): + """A modern multi-device schedule waits once, then fetches each result.""" + schedule_id = "sched-uuid" + trigger_url = "http://localhost:5000/api/v3_0/assets/1/schedules/trigger" + job_url = f"http://localhost:5000/api/v3_0/jobs/{schedule_id}" + result_urls = [ + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/schedules/" + f"{schedule_id}?duration=P0DT0H45M0S" + for sensor_id in (10, 11) + ] + + with aioresponses() as m: + client = FlexMeasuresClient( + email="test@test.test", + password="test", + access_token="test-token", + ) + m.post( + trigger_url, + status=202, + payload={"job": schedule_id, "status": "ACCEPTED"}, + headers={"FlexMeasures-Version": "0.33.0"}, + ) + m.get(job_url, status=202, payload={"status": "STARTED"}) + m.get(job_url, status=200, payload={"status": "FINISHED"}) + for sensor_id, result_url in zip((10, 11), result_urls): + m.get( + result_url, + status=200, + payload={ + "values": [float(sensor_id)], + "start": "2023-01-01T00:00:00+00:00", + "duration": "PT45M", + "unit": "MW", + }, + ) + + schedules = await client.trigger_and_get_schedule( + asset_id=1, + start="2023-01-01T00:00:00+00:00", + duration="PT45M", + flex_model=[{"sensor": 10}, {"sensor": 11}], + polling_interval=0, + ) + + assert [schedule["sensor"] for schedule in schedules] == [10, 11] + assert len(m.requests[("GET", URL(job_url))]) == 2 + for result_url in result_urls: + assert len(m.requests[("GET", URL(result_url))]) == 1 + await client.close() + + @pytest.mark.asyncio async def test_trigger_and_get_schedule_asset_id_no_flex_model(): """asset_id with flex_model=None returns [].""" From e2ec466b92ff9ef9baec24e3a4e0800dc6876e48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicolas=20H=C3=B6ning?= Date: Sun, 6 Sep 2026 00:31:40 +0200 Subject: [PATCH 15/15] consolidate/clarify parameters for polling of requests and jobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Nicolas Höning --- docs/forecasting.rst | 40 ++++-- docs/reporting.rst | 23 +++- docs/scheduling.rst | 40 ++++-- examples/HEMS/utils/asset_utils.py | 40 +----- src/flexmeasures_client/client.py | 132 +++++++++++++++---- src/flexmeasures_client/response_handling.py | 6 +- src/flexmeasures_client/utils.py | 37 ++++++ tests/client/test_forecast.py | 12 +- tests/client/test_init.py | 71 +++++++++- tests/client/test_report.py | 46 +++++++ tests/client/test_schedule.py | 18 +-- 11 files changed, 345 insertions(+), 120 deletions(-) create mode 100644 src/flexmeasures_client/utils.py diff --git a/docs/forecasting.rst b/docs/forecasting.rst index acca2704..742d430d 100644 --- a/docs/forecasting.rst +++ b/docs/forecasting.rst @@ -135,31 +135,47 @@ Waiting and legacy polling On FlexMeasures v0.33.0 and newer, ``trigger_and_get_forecast`` waits through ``GET /jobs/`` and fetches the forecast values only after the job has -finished. Its job wait uses exponential backoff and accepts the same options -as :meth:`wait_for_job`: +finished. Its job wait uses exponential backoff and the following client +defaults: -- ``polling_interval`` (default 2 s) — delay before a repeated status check -- ``max_polling_interval`` (default 30 s) — maximum delay between checks -- ``timeout`` (default 600 s) — total job wait budget +- ``job_polling_interval`` (default 2 s) — delay before a repeated status check +- ``job_polling_max_interval`` (default 30 s) — maximum delay between checks +- ``job_polling_timeout`` (default 600 s) — total job wait budget + +Set these defaults when constructing the client. The convenience method's +``polling_interval``, ``max_polling_interval``, and ``timeout`` arguments can +override them for one forecast: For example: .. code-block:: python + client = FlexMeasuresClient( + ..., + job_polling_interval=5.0, + job_polling_timeout=1800.0, + ) + forecast = await client.trigger_and_get_forecast( sensor_id=1, duration="PT24H", - timeout=1800.0, max_polling_interval=60.0, ) ``get_forecast`` still polls the result endpoint when it is called directly, and ``trigger_and_get_forecast`` retains that behaviour for servers older than -v0.33.0. This legacy polling is controlled by client-level settings: +v0.33.0. This legacy polling uses the client's general HTTP request settings. +Those settings also apply to authentication, API discovery, asset and sensor +operations, data transfer, trigger calls, result retrieval, and each individual +job-status lookup: + +- ``request_timeout`` (default 40 s) — timeout for one HTTP attempt +- ``request_retry_interval`` (default 10 s) — initial wait between retries +- ``request_retry_timeout`` (default 200 s) — total request/retry budget +- ``max_request_attempts`` (default 10) — maximum attempts in that loop -- ``polling_interval`` (default 10 s) — initial wait between retries -- ``polling_timeout`` (default 200 s) — maximum total wait time -- ``max_polling_steps`` (default 10) — maximum number of poll attempts +They do not control the cadence or total lifetime of a background-job wait; +the ``job_polling_*`` settings above do that. Configure those settings at client construction time: @@ -167,6 +183,6 @@ Configure those settings at client construction time: client = FlexMeasuresClient( ..., - polling_interval=5.0, # check every 5 seconds - polling_timeout=300.0, # wait up to 5 minutes + request_retry_interval=5.0, # retry after 5 seconds + request_retry_timeout=300.0, # allow retries for up to 5 minutes ) diff --git a/docs/reporting.rst b/docs/reporting.rst index 403f2c06..173f2f16 100644 --- a/docs/reporting.rst +++ b/docs/reporting.rst @@ -110,18 +110,33 @@ Polling behaviour state. Waits back off exponentially, so short jobs are picked up quickly without hammering the server on long ones: -- ``polling_interval`` (default 2 s) — delay before a repeated status check -- ``max_polling_interval`` (default 30 s) — cap on the backing-off wait -- ``timeout`` (default 600 s) — total budget for the job to finish +- ``job_polling_interval`` (default 2 s) — delay before a repeated status check +- ``job_polling_max_interval`` (default 30 s) — cap on the backing-off wait +- ``job_polling_timeout`` (default 600 s) — total job wait budget + +Configure these defaults on the client. Calls to ``wait_for_job`` and +``trigger_and_await_report`` can override them with ``polling_interval``, +``max_polling_interval``, and ``timeout``: .. code-block:: python + client = FlexMeasuresClient( + ..., + job_polling_interval=5.0, + job_polling_timeout=3600.0, + ) + job = await client.wait_for_job( job_id, - timeout=3600.0, # allow an hour for a heavy report max_polling_interval=60.0, ) +Each job-status lookup is itself an HTTP request, so ``request_timeout``, +``request_retry_interval``, ``request_retry_timeout``, and +``max_request_attempts`` still govern that individual lookup and any transient +retries. They do not determine when the next job-status lookup happens or the +total job-wait budget; the ``job_polling_*`` settings do. + Error handling -------------- diff --git a/docs/scheduling.rst b/docs/scheduling.rst index 62191ccb..a192d6aa 100644 --- a/docs/scheduling.rst +++ b/docs/scheduling.rst @@ -220,29 +220,45 @@ Waiting, legacy polling, and errors On FlexMeasures v0.33.0 and newer, ``trigger_and_get_schedule`` waits through ``GET /jobs/`` once, then retrieves the requested sensor result or -results. Its job wait is controlled by method arguments: +results. Its job wait is controlled by the following client defaults: -- ``polling_interval`` (default 2 s) — delay before a repeated status check -- ``max_polling_interval`` (default 30 s) — maximum delay between checks -- ``timeout`` (default 600 s) — total job wait budget +- ``job_polling_interval`` (default 2 s) — delay before a repeated status check +- ``job_polling_max_interval`` (default 30 s) — maximum delay between checks +- ``job_polling_timeout`` (default 600 s) — total job wait budget + +Set these defaults when constructing the client. The convenience method's +``polling_interval``, ``max_polling_interval``, and ``timeout`` arguments can +override them for one schedule: .. code-block:: python + client = FlexMeasuresClient( + ..., + job_polling_interval=5.0, + job_polling_timeout=1800.0, + ) + schedule = await client.trigger_and_get_schedule( sensor_id=8, start="2026-09-05T08:00:00+02:00", duration="PT12H", - timeout=1800.0, max_polling_interval=60.0, ) ``get_schedule`` still polls the sensor result endpoint when called directly, and ``trigger_and_get_schedule`` retains that behaviour for servers older than -v0.33.0. This legacy polling is controlled by client settings: +v0.33.0. This legacy polling uses the client's general HTTP request settings. +Those settings also apply to authentication, API discovery, asset and sensor +operations, data transfer, trigger calls, result retrieval, and each individual +job-status lookup: + +- ``request_timeout`` (default 40 s) — timeout for one HTTP attempt +- ``request_retry_interval`` (default 10 s) — initial wait between attempts +- ``request_retry_timeout`` (default 200 s) — total request/retry budget +- ``max_request_attempts`` (default 10) — maximum attempts in that loop -- ``polling_interval`` (default 10 s) — initial wait between attempts -- ``polling_timeout`` (default 200 s) — maximum total wait -- ``max_polling_steps`` (default 10) — maximum number of attempts +They do not control the cadence or total lifetime of a background-job wait; +the ``job_polling_*`` settings above do that. Override them when constructing the client: @@ -250,9 +266,9 @@ Override them when constructing the client: client = FlexMeasuresClient( ..., - polling_interval=5.0, - polling_timeout=300.0, - max_polling_steps=12, + request_retry_interval=5.0, + request_retry_timeout=300.0, + max_request_attempts=12, ) With the jobs API, a job that ends as ``FAILED``, ``STOPPED``, or ``CANCELED`` diff --git a/examples/HEMS/utils/asset_utils.py b/examples/HEMS/utils/asset_utils.py index 728876e9..2c1c7067 100644 --- a/examples/HEMS/utils/asset_utils.py +++ b/examples/HEMS/utils/asset_utils.py @@ -1,4 +1,3 @@ -import asyncio import os from pathlib import Path from typing import Any @@ -43,44 +42,7 @@ async def wait_for_ingestion_jobs( print(f"Waiting for {len(pending_ingestion_jobs)} ingestion job(s)...") for job_id in pending_ingestion_jobs: - deadline = asyncio.get_running_loop().time() + client.polling_timeout - polling_step = 0 - - while True: - # FlexMeasures 0.33 returns HTTP 200 even while a job is in - # progress. Newer versions return 202, which client.request polls - # internally. Inspecting the status field supports both versions. - job, _ = await client.request( - uri=f"jobs/{job_id}", - method="GET", - ) - job_status = ( - str(job.get("status", "")).upper() if isinstance(job, dict) else "" - ) - if job_status == "FINISHED": - break - if job_status not in {"QUEUED", "STARTED", "DEFERRED", "SCHEDULED"}: - raise RuntimeError( - f"Ingestion job {job_id} did not finish successfully: {job}" - ) - - polling_step += 1 - if polling_step >= client.max_polling_steps: - raise ConnectionError( - f"Max polling steps reached while waiting for ingestion job " - f"{job_id}. Last status: {job_status}" - ) - - remaining = deadline - asyncio.get_running_loop().time() - if remaining <= 0: - raise ConnectionError( - f"Client polling timeout while waiting for ingestion job " - f"{job_id}. Last status: {job_status}" - ) - sleep_interval = min( - client.polling_interval * (2 ** (polling_step - 1)), remaining - ) - await asyncio.sleep(sleep_interval) + await client.wait_for_job(job_id) pending_ingestion_jobs.clear() diff --git a/src/flexmeasures_client/client.py b/src/flexmeasures_client/client.py index 4bc71060..2ece70de 100644 --- a/src/flexmeasures_client/client.py +++ b/src/flexmeasures_client/client.py @@ -8,7 +8,7 @@ import socket import time import warnings -from dataclasses import dataclass, field +from dataclasses import InitVar, dataclass, field from datetime import datetime, timedelta from logging import Logger from typing import Any, cast @@ -35,13 +35,18 @@ check_for_status, check_response, ) +from flexmeasures_client.utils import apply_deprecated_parameter_aliases LOGGER = logging.getLogger(__name__) -MAX_POLLING_STEPS: int = 10 # seconds -POLLING_TIMEOUT = 200.0 # seconds +MAX_REQUEST_ATTEMPTS: int = 10 +REQUEST_RETRY_TIMEOUT = 200.0 # seconds, total request/retry budget REQUEST_TIMEOUT = 40.0 # seconds -POLLING_INTERVAL = 10.0 # seconds +REQUEST_RETRY_INTERVAL = 10.0 # seconds +# Backward-compatible module-level aliases; use the REQUEST_* names instead. +MAX_POLLING_STEPS = MAX_REQUEST_ATTEMPTS +POLLING_TIMEOUT = REQUEST_RETRY_TIMEOUT +POLLING_INTERVAL = REQUEST_RETRY_INTERVAL API_VERSIONS_LIST = ["v3_0"] JOB_POLLING_INTERVAL = 2.0 # seconds, first wait between job status polls @@ -156,18 +161,57 @@ class FlexMeasuresClient: path: str = f"/api/{api_version}/" access_token: str | None = None - max_polling_steps: int = MAX_POLLING_STEPS - polling_timeout: float = POLLING_TIMEOUT # seconds + # These govern the HTTP request/retry loop used throughout the client: auth, + # version discovery, assets and sensors, data I/O, trigger requests, result + # retrieval, and each individual job-status lookup. They do not control how + # often or how long a background job is polled; the job_polling_* fields do. + max_request_attempts: int = MAX_REQUEST_ATTEMPTS + request_retry_timeout: float = REQUEST_RETRY_TIMEOUT # total loop budget request_timeout: float = REQUEST_TIMEOUT # seconds - polling_interval: float = POLLING_INTERVAL # seconds + request_retry_interval: float = REQUEST_RETRY_INTERVAL # seconds session: ClientSession | None = None server_version: str | None = None logger: Logger = LOGGER + job_polling_interval: float = JOB_POLLING_INTERVAL # seconds + job_polling_max_interval: float = JOB_POLLING_MAX_INTERVAL # seconds + job_polling_timeout: float = JOB_POLLING_TIMEOUT # seconds + max_polling_steps: InitVar[int | None] = None + polling_timeout: InitVar[float | None] = None + polling_interval: InitVar[float | None] = None _sensor_asset_id_cache: dict[int, int] = field( default_factory=dict, init=False, repr=False ) - def __post_init__(self): + def __post_init__( + self, + max_polling_steps: int | None, + polling_timeout: float | None, + polling_interval: float | None, + ): + apply_deprecated_parameter_aliases( + self, + ( + ( + "max_polling_steps", + max_polling_steps, + "max_request_attempts", + MAX_REQUEST_ATTEMPTS, + ), + ( + "polling_timeout", + polling_timeout, + "request_retry_timeout", + REQUEST_RETRY_TIMEOUT, + ), + ( + "polling_interval", + polling_interval, + "request_retry_interval", + REQUEST_RETRY_INTERVAL, + ), + ), + ) + if self.session is None: self.session = ClientSession() @@ -260,14 +304,20 @@ async def request( ) -> tuple[dict | list, int]: """Send a request to FlexMeasures. + The request-level settings apply to all HTTP calls made by the client, + including each individual job-status lookup. ``request_timeout`` limits + one HTTP attempt; ``request_retry_interval``, ``request_retry_timeout``, + and ``max_request_attempts`` govern the surrounding retry loop. The + separate ``job_polling_*`` settings govern repeated job-status lookups. + Retries if: - - the client request timed out (as indicated by the client's self.request_timeout) + - the client request timed out (as indicated by ``request_timeout``) - the server response indicates a 408 (Request Timeout) status - the server response indicates a 503 (Service Unavailable) status with a Retry-After response header. Fails if: - the server response indicated a status code of 400 or higher - - the client polling timed out (as indicated by the client's self.polling_timeout) + - the request/retry loop exceeded ``request_retry_timeout`` ``pass_through_statuses`` is for callers that interpret particular HTTP statuses themselves. Such responses are returned without generic @@ -281,8 +331,8 @@ async def request( # we allow retrying once if we include authentication headers reauth_once = True if include_auth else False try: - async with async_timeout.timeout(self.polling_timeout): - while polling_step < self.max_polling_steps: + async with async_timeout.timeout(self.request_retry_timeout): + while polling_step < self.max_request_attempts: headers = await self.get_headers(include_auth=include_auth) try: async with async_timeout.timeout(self.request_timeout): @@ -308,7 +358,7 @@ async def request( ) and polling_step == previous_polling_step: break except asyncio.TimeoutError: - sleep_interval = self.polling_interval * (2**polling_step) + sleep_interval = self.request_retry_interval * (2**polling_step) message = f"Client request timeout occurred while connecting to the API. Polling step: {polling_step}. Retrying in {sleep_interval} seconds..." # noqa: E501 self.logger.debug(message) polling_step += 1 @@ -329,12 +379,12 @@ async def request( ) from exception else: raise ConnectionError( - "Max polling steps reached while waiting for the API response." + "Maximum request attempts reached while waiting for the API response." ) except asyncio.TimeoutError as exception: raise ConnectionError( - "Client polling timeout while connection to the API." + "Request retry timeout while connecting to the API." ) from exception check_content_type(response) @@ -1026,9 +1076,9 @@ async def trigger_and_get_schedule( prior: datetime | None = None, scheduler: str | None = None, unit: str | None = None, - timeout: float = JOB_POLLING_TIMEOUT, - polling_interval: float = JOB_POLLING_INTERVAL, - max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, + timeout: float | None = None, + polling_interval: float | None = None, + max_polling_interval: float | None = None, ) -> dict | list[dict]: """Trigger a schedule and then fetch it. @@ -1045,8 +1095,11 @@ async def trigger_and_get_schedule( :param unit: desired unit for the schedule values (e.g. "W", "kW", "MW"). Passed through to get_schedule(); see its docstring for details. :param timeout: total seconds to wait through the jobs endpoint. + Defaults to ``self.job_polling_timeout``. :param polling_interval: seconds before the first repeated status lookup. + Defaults to ``self.job_polling_interval``. :param max_polling_interval: maximum delay between status lookups. + Defaults to ``self.job_polling_max_interval``. :returns: For a single device, returns the schedule as a dictionary. For example: { @@ -1780,9 +1833,9 @@ async def trigger_and_get_forecast( max_forecast_horizon: str | timedelta | None = None, forecast_frequency: str | timedelta | None = None, probabilistic: bool | None = None, - timeout: float = JOB_POLLING_TIMEOUT, - polling_interval: float = JOB_POLLING_INTERVAL, - max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, + timeout: float | None = None, + polling_interval: float | None = None, + max_polling_interval: float | None = None, ) -> dict: """Trigger a forecasting job and then fetch the result. @@ -1791,8 +1844,11 @@ async def trigger_and_get_forecast( retain result-endpoint polling. :param timeout: total seconds to wait through the jobs endpoint. + Defaults to ``self.job_polling_timeout``. :param polling_interval: seconds before the first repeated status lookup. + Defaults to ``self.job_polling_interval``. :param max_polling_interval: maximum delay between status lookups. + Defaults to ``self.job_polling_max_interval``. :returns: forecast as dictionary, for example: { @@ -1880,9 +1936,9 @@ async def get_job_status(self, job_id: str) -> dict: async def wait_for_job( self, job_id: str, - timeout: float = JOB_POLLING_TIMEOUT, - polling_interval: float = JOB_POLLING_INTERVAL, - max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, + timeout: float | None = None, + polling_interval: float | None = None, + max_polling_interval: float | None = None, ) -> dict: """Poll a background job until it reaches a terminal state. @@ -1892,14 +1948,27 @@ async def wait_for_job( :param job_id: UUID of the job, as returned by a trigger endpoint. :param timeout: total number of seconds to wait for the job to finish. - :param polling_interval: seconds to wait before the first status poll. + Defaults to ``self.job_polling_timeout``. + :param polling_interval: seconds to wait before a repeated status lookup. + Defaults to ``self.job_polling_interval``. :param max_polling_interval: upper bound on the wait between polls. + Defaults to ``self.job_polling_max_interval``. :returns: the final job status dictionary (see :func:`get_job_status`). :raises JobFailedError: if the job ended as FAILED, STOPPED or CANCELED. :raises JobTimeoutError: if the job did not finish within ``timeout``. """ + timeout = self.job_polling_timeout if timeout is None else timeout + polling_interval = ( + self.job_polling_interval if polling_interval is None else polling_interval + ) + max_polling_interval = ( + self.job_polling_max_interval + if max_polling_interval is None + else max_polling_interval + ) + deadline = time.monotonic() + timeout interval = polling_interval job = await self.get_job_status(job_id) @@ -1993,15 +2062,22 @@ async def trigger_and_await_report( reporter: str, parameters: dict, config: dict | None = None, - timeout: float = JOB_POLLING_TIMEOUT, - polling_interval: float = JOB_POLLING_INTERVAL, - max_polling_interval: float = JOB_POLLING_MAX_INTERVAL, + timeout: float | None = None, + polling_interval: float | None = None, + max_polling_interval: float | None = None, ) -> dict: """Trigger a one-off reporting job and wait for it to finish. Reports write their result to their output sensors rather than returning it, so use :func:`get_sensor_data` to read the values back. + :param timeout: total seconds to wait for the report job. + Defaults to ``self.job_polling_timeout``. + :param polling_interval: seconds before a repeated status lookup. + Defaults to ``self.job_polling_interval``. + :param max_polling_interval: maximum delay between status lookups. + Defaults to ``self.job_polling_max_interval``. + :returns: the final job status dictionary (see :func:`get_job_status`). :raises JobFailedError: if the report job ended as FAILED, STOPPED or CANCELED. diff --git a/src/flexmeasures_client/response_handling.py b/src/flexmeasures_client/response_handling.py index 936cccc7..76fcd638 100644 --- a/src/flexmeasures_client/response_handling.py +++ b/src/flexmeasures_client/response_handling.py @@ -46,7 +46,7 @@ async def check_response( if status in pass_through_statuses: pass elif status == 202 and method.upper() == "GET": - sleep_interval = self.polling_interval * (2**polling_step) + sleep_interval = self.request_retry_interval * (2**polling_step) job_status = payload.get("status") message = "Server accepted the request but the result is not ready yet." if job_status: @@ -67,7 +67,7 @@ async def check_response( or "Scheduling job has an unknown status" in payload.get("message", "") ): # can be removed in a later version GH issue #645 of the FlexMeasures repo - sleep_interval = self.polling_interval * (2**polling_step) + sleep_interval = self.request_retry_interval * (2**polling_step) message = f"Server indicated to try again later. Retrying in {sleep_interval} seconds..." # noqa: E501 self.logger.debug(message) polling_step += 1 @@ -82,7 +82,7 @@ async def check_response( await self.get_access_token() reauth_once = False elif status == 503 and "Retry-After" in headers: - sleep_interval = self.polling_interval * (2**polling_step) + sleep_interval = self.request_retry_interval * (2**polling_step) polling_step += 1 await asyncio.sleep(sleep_interval) elif payload.get("errors"): diff --git a/src/flexmeasures_client/utils.py b/src/flexmeasures_client/utils.py new file mode 100644 index 00000000..049be0dd --- /dev/null +++ b/src/flexmeasures_client/utils.py @@ -0,0 +1,37 @@ +"""Small shared utilities for the FlexMeasures client.""" + +from __future__ import annotations + +import warnings +from collections.abc import Iterable + +DeprecatedParameterAlias = tuple[str, object | None, str, object] + + +def apply_deprecated_parameter_aliases( + instance: object, + aliases: Iterable[DeprecatedParameterAlias], +) -> None: + """Apply deprecated constructor arguments to their replacement attributes. + + Each alias contains the deprecated name and supplied value followed by the + replacement name and its default value. ``None`` means that the deprecated + argument was not supplied. Supplying both names with a non-default value is + rejected because their precedence would otherwise be ambiguous. + """ + for deprecated_name, deprecated_value, replacement_name, default in aliases: + if deprecated_value is None: + continue + + if getattr(instance, replacement_name) != default: + raise TypeError( + f"Pass either {replacement_name} or the deprecated " + f"{deprecated_name}, not both." + ) + + warnings.warn( + f"{deprecated_name} is deprecated; use {replacement_name} instead.", + DeprecationWarning, + stacklevel=4, + ) + setattr(instance, replacement_name, deprecated_value) diff --git a/tests/client/test_forecast.py b/tests/client/test_forecast.py index 92dbd5c2..b951613f 100644 --- a/tests/client/test_forecast.py +++ b/tests/client/test_forecast.py @@ -130,7 +130,7 @@ async def test_get_forecast_polling() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", ) @@ -166,7 +166,7 @@ async def test_get_forecast_failed_job() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", ) @@ -200,14 +200,14 @@ async def test_get_forecast_polling_max_steps() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0, - max_polling_steps=2, + request_retry_interval=0, + max_request_attempts=2, access_token="skip-auth", ) with pytest.raises( ConnectionError, - match="Max polling steps reached while waiting for the API response.", + match="Maximum request attempts reached while waiting for the API response.", ): await flexmeasures_client.get_forecast( sensor_id=sensor_id, forecast_id=forecast_id @@ -224,7 +224,7 @@ async def test_trigger_and_get_forecast() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, ) flexmeasures_client.access_token = "test-token" diff --git a/tests/client/test_init.py b/tests/client/test_init.py index 5e9202d8..f59bed60 100644 --- a/tests/client/test_init.py +++ b/tests/client/test_init.py @@ -89,11 +89,14 @@ async def test__init__( "ssl": asserted_ssl, "api_version": asserted_version, "path": "/api/v3_0/", - "max_polling_steps": 10, - "polling_timeout": 200.0, + "max_request_attempts": 10, + "request_retry_timeout": 200.0, "server_version": None, "request_timeout": 40.0, - "polling_interval": 10.0, + "request_retry_interval": 10.0, + "job_polling_interval": 2.0, + "job_polling_max_interval": 30.0, + "job_polling_timeout": 600.0, } init_dict = flexmeasures_client.__dict__ init_dict.pop("session") @@ -102,6 +105,60 @@ async def test__init__( assert init_dict == assert_dict +@pytest.mark.parametrize( + ("old_name", "new_name", "value"), + ( + ("max_polling_steps", "max_request_attempts", 4), + ("polling_timeout", "request_retry_timeout", 12.5), + ("polling_interval", "request_retry_interval", 0.25), + ), +) +@pytest.mark.asyncio +async def test_deprecated_request_retry_option_is_still_accepted( + old_name: str, + new_name: str, + value: int | float, +) -> None: + """Old constructor names configure their renamed request-loop options.""" + with pytest.warns( + DeprecationWarning, + match=f"use {new_name} instead", + ): + client = FlexMeasuresClient( + email="test@test.test", + password="test", + **{old_name: value}, + ) + + assert getattr(client, new_name) == value + assert old_name not in client.__dict__ + await client.close() + + +@pytest.mark.parametrize( + ("old_name", "new_name", "old_value", "new_value"), + ( + ("max_polling_steps", "max_request_attempts", 4, 5), + ("polling_timeout", "request_retry_timeout", 12.5, 15.0), + ("polling_interval", "request_retry_interval", 0.25, 0.5), + ), +) +@pytest.mark.asyncio +async def test_old_and_new_request_retry_names_are_rejected( + old_name: str, + new_name: str, + old_value: int | float, + new_value: int | float, +) -> None: + """Ambiguous old and new retry configuration is rejected.""" + with pytest.raises(TypeError, match=f"Pass either {new_name}"): + FlexMeasuresClient( + email="test@test.test", + password="test", + **{new_name: new_value, old_name: old_value}, + ) + + @pytest.mark.parametrize( "kwargs, error_type, error_text", [ @@ -299,7 +356,7 @@ async def test_get_versions() -> None: @pytest.mark.asyncio async def test_get_schedule_timeout() -> None: async def callback(url, **kwargs): - # Sleep longer than the polling_timeout + # Sleep longer than the request_retry_timeout await asyncio.sleep(3) return CallbackResult(status=200) @@ -313,9 +370,9 @@ async def callback(url, **kwargs): flexmeasures_client = FlexMeasuresClient( email="test@test.test", password="test", - polling_timeout=0.5, + request_retry_timeout=0.5, request_timeout=0.2, - polling_interval=0.1, + request_retry_interval=0.1, ) with pytest.raises(ConnectionError): @@ -386,7 +443,7 @@ async def test_503_retry_after(): flexmeasures_client = FlexMeasuresClient( email="test@test.test", password="test", - polling_interval=0.01, + request_retry_interval=0.01, request_timeout=5, ) flexmeasures_client.access_token = "test-token" diff --git a/tests/client/test_report.py b/tests/client/test_report.py index bc91d1d7..1040389e 100644 --- a/tests/client/test_report.py +++ b/tests/client/test_report.py @@ -1,3 +1,5 @@ +from unittest.mock import AsyncMock, patch + import pytest from aioresponses import aioresponses from yarl import URL @@ -201,6 +203,50 @@ async def test_wait_for_job_polls_until_finished() -> None: await client.close() +@pytest.mark.asyncio +async def test_wait_for_job_uses_client_polling_defaults() -> None: + """Stored job settings drive waits when the call has no overrides.""" + with aioresponses() as m: + client = FlexMeasuresClient( + email="test@test.test", + password="test", + access_token="test-token", + job_polling_interval=0.25, + job_polling_max_interval=0.5, + job_polling_timeout=10, + ) + m.get(JOB_URL, status=202, payload=job_payload("QUEUED")) + m.get(JOB_URL, status=202, payload=job_payload("STARTED")) + m.get(JOB_URL, status=200, payload=job_payload("FINISHED")) + + sleep = AsyncMock() + with patch("flexmeasures_client.client.asyncio.sleep", sleep): + job = await client.wait_for_job(JOB_ID) + + assert job["status"] == "FINISHED" + assert [call.args[0] for call in sleep.await_args_list] == [0.25, 0.5] + + await client.close() + + +@pytest.mark.asyncio +async def test_wait_for_job_uses_client_timeout_default() -> None: + """The stored job timeout is used when no call override is supplied.""" + with aioresponses() as m: + client = FlexMeasuresClient( + email="test@test.test", + password="test", + access_token="test-token", + job_polling_timeout=0, + ) + m.get(JOB_URL, status=202, payload=job_payload("QUEUED")) + + with pytest.raises(JobTimeoutError, match="within 0 seconds"): + await client.wait_for_job(JOB_ID) + + await client.close() + + @pytest.mark.asyncio async def test_wait_for_job_raises_on_failed_job() -> None: """Test that a failed job surfaces the server message and traceback.""" diff --git a/tests/client/test_schedule.py b/tests/client/test_schedule.py index ff6b76be..557c531d 100644 --- a/tests/client/test_schedule.py +++ b/tests/client/test_schedule.py @@ -107,7 +107,7 @@ async def test_get_schedule_polling() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", ) @@ -146,7 +146,7 @@ async def test_get_schedule_polling_accepted(caplog) -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", ) @@ -186,7 +186,7 @@ async def test_get_schedule_polling_exponential_backoff() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=1.0, + request_retry_interval=1.0, access_token="skip-auth", ) @@ -239,7 +239,7 @@ async def test_trigger_and_get_schedule() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", ) @@ -304,7 +304,7 @@ async def test_get_fallback_schedule(): email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", ) @@ -457,7 +457,7 @@ async def test_trigger_and_get_schedule_asset_id_flex_model_list(): email="test@test.test", password="test", request_timeout=2, - polling_interval=0.1, + request_retry_interval=0.1, ) client.access_token = "test-token" m.post( @@ -550,7 +550,7 @@ async def test_trigger_and_get_schedule_asset_id_no_flex_model(): email="test@test.test", password="test", request_timeout=2, - polling_interval=0.1, + request_retry_interval=0.1, ) client.access_token = "test-token" m.post( @@ -1081,7 +1081,7 @@ async def test_trigger_and_get_schedule_with_unit(): email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", ) # Older server: client-side conversion @@ -1182,7 +1182,7 @@ async def test_get_schedule_failed_job_raises() -> None: email="test@test.test", password="test", request_timeout=2, - polling_interval=0.2, + request_retry_interval=0.2, access_token="skip-auth", )