diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 8f85e08b..5ad23f86 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -9,6 +9,10 @@ repos: - id: check-ast - id: check-json - id: check-merge-conflict + # RST section adornments are rows of "=", which this hook reads as conflict + # markers. It scans the whole tree while a merge is in progress, so without + # this exclusion every merge commit is blocked by docs/license.rst. + exclude: ^docs/license\.rst$ - id: check-xml - id: check-yaml - id: debug-statements diff --git a/CHANGELOG.rst b/CHANGELOG.rst index af76e198..201e4610 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -2,6 +2,22 @@ Changelog ========= +Unreleased +========== + +- ``post_sensor_data()`` now awaits asynchronous server-side ingestion by + default (``await_ingestion=True``), restoring read-your-writes semantics + after FlexMeasures PR #2101 made ``POST .../sensors/data`` (and the file + upload endpoint) return ``202 Accepted`` with a background job id instead + of processing synchronously. The client polls the job-status endpoint + (exponential backoff from 0.25s up to a 2s cap, ``ingestion_polling_timeout`` + seconds total, default 60s) until the job finishes. A failed job raises the + new ``IngestionFailedError``; a job still pending when polling times out + logs an ERROR and returns normally, so callers with their own downstream + safety nets are not blocked indefinitely. Pass ``await_ingestion=False`` to + opt out and return as soon as the POST is acknowledged, matching the old + (PR #2101) behavior. + Version 0.1.1 ============= diff --git a/docker-compose.override.yml b/docker-compose.override.yml new file mode 100644 index 00000000..bd23e2cd --- /dev/null +++ b/docker-compose.override.yml @@ -0,0 +1,253 @@ +# ------------------------------------------------------------------ +# This allow to run the S2 CEM from your local FlexMeasures Client code in a docker compose stack. +# Assuming you have flexmeasures the repo next to your flexmeasures-client repo, +# run this from the flexmeasures folder (which contains the Dockerfile): +# docker compose \ +# -f docker-compose.yml \ +# -f ../flexmeasures-client/docker-compose.override.yml \ +# up +# ------------------------------------------------------------------ + +# Shared definition for all CEM instances. Each instance keeps listening on +# port 8080 inside its own container; only the host port mapping differs, +# so RMs connect to ws://localhost:8080/ws, ws://localhost:8081/ws, etc. +x-cem: &cem + build: + context: . + dockerfile: Dockerfile + image: flexmeasures-client-cem + depends_on: + - server + restart: always + environment: + FLEXMEASURES_BASE_URL: http://server:5000 + FLEXMEASURES_USER: toy-user@flexmeasures.io + FLEXMEASURES_PASSWORD: toy-password + LOGGING_LEVEL: DEBUG + SETUPTOOLS_SCM_PRETEND_VERSION_FOR_FLEXMEASURES_CLIENT: "0.0.0" + SETUPTOOLS_SCM_PRETEND_VERSION: "0.0.0" + volumes: + # If flexmeasures_client lives in your repo and you want live edits + - ../flexmeasures-client:/app/flexmeasures-client:rw + entrypoint: ["/bin/sh", "-c"] + command: + - | + # Install and run with the SAME interpreter (`python3 -m pip`, not bare `pip`). + # In images built from FlexMeasures' uv-based Dockerfile, `python3` resolves to + # the /app/.venv virtualenv (which ships without pip and cannot see system + # site-packages) while bare `pip` is the system one under /usr/local - so + # packages installed with `pip` are invisible at runtime (ModuleNotFoundError: + # aiohttp). Older cached images had a single system Python, masking this. + # The ensurepip guard bootstraps pip into the venv on fresh images and is a + # no-op wherever `python3 -m pip` already works. + python3 -m pip --version >/dev/null 2>&1 || python3 -m ensurepip --upgrade + python3 -m pip install --break-system-packages -e "/app/flexmeasures-client[s2]" + python3 -m pip install --break-system-packages aiohttp pytz s2-python==0.8.2 + exec python3 /app/flexmeasures-client/src/flexmeasures_client/s2/script/websockets_server.py + +# Shared definition for all RQ worker instances. Scaled one-per-concurrently- +# simulated-apartment, same as the cem-* services and the server's gunicorn +# worker count below, since otherwise every apartment's scheduling jobs funnel +# through a single worker process and serialize even though gunicorn/CEM are +# already scaled per apartment. Bump alongside those when scaling further. +x-worker: &worker + # worker-2/worker-3 are new services (not present in the base docker-compose.yml, + # unlike "worker" itself), so this anchor must be self-contained rather than + # relying on a base-file merge for build/depends_on/restart/base environment. + build: + context: . + dockerfile: Dockerfile + depends_on: + - dev-db + - queue-db + - mailhog + restart: on-failure + cap_add: + # SYS_PTRACE: lets py-spy attach to the RQ worker subprocess for live debugging + # of slow/stuck scheduling jobs (py-spy itself isn't preinstalled here; install + # it ad hoc with `docker exec pip install --break-system-packages py-spy`). + - SYS_PTRACE + volumes: + # a place for config and plugin code, and custom requirements.txt + - ./flexmeasures-instance/:/usr/var/flexmeasures-instance/:rw + - ../flexmeasures/flexmeasures:/app/flexmeasures:rw + - ./flexmeasures-instance/:/app/instance/:rw + # The client repo (flexmeasures_client + examples/HEMS). The seita_ems + # community plugin is mounted by the ems repo's docker-compose.community.yml + # overlay instead, so this stack also runs without access to that repo. + - ../flexmeasures-client:/app/flexmeasures-client:rw + entrypoint: ["/bin/sh", "-c"] + environment: + SQLALCHEMY_DATABASE_URI: "postgresql://fm-dev-db-user:fm-dev-db-pass@dev-db:5432/fm-dev-db" + FLEXMEASURES_REDIS_URL: queue-db + FLEXMEASURES_REDIS_PASSWORD: fm-redis-pass + SECRET_KEY: notsecret + SECURITY_TOTP_SECRETS: '{"1": "something-secret"}' + FLEXMEASURES_ENV: development + MAIL_SERVER: mailhog + MAIL_PORT: 1025 + LOGGING_LEVEL: INFO + +services: + dev-db: + ports: + - "5433:5432" + queue-db: + ports: + - "6380:6379" + server: + # SYS_PTRACE: lets py-spy attach to the gunicorn workers for live debugging + # of the CEM-asset-reuse hang (see projects/009 HANDOFF.md in pps_flexed). + cap_add: + - SYS_PTRACE + volumes: + # A place for config and plugin code, and custom requirements.txt + # The 1st mount point is for running the FlexMeasures CLI, the 2nd for gunicorn + # We use :rw so flexmeasures CLI commands can write log files + - ./flexmeasures-instance/:/usr/var/flexmeasures-instance/:rw + - ./flexmeasures-instance/:/app/instance/:rw + - ../flexmeasures/flexmeasures:/app/flexmeasures:rw + # The client repo (flexmeasures_client + examples/HEMS). The seita_ems + # community plugin is mounted by the ems repo's docker-compose.community.yml + # overlay instead, so this stack also runs without access to that repo. + - ../flexmeasures-client:/app/flexmeasures-client:rw + command: + - | + pip install --break-system-packages -e /app + pip install --break-system-packages -r /usr/var/flexmeasures-instance/requirements.txt + pip install timely-beliefs -U --break-system-packages + pip install --break-system-packages -e /app/flexmeasures-client[s2] + pip install --break-system-packages py-spy + flexmeasures db upgrade + if ! flexmeasures show accounts | grep -q "Docker Toy Account"; then + flexmeasures add toy-account --name 'Docker Toy Account' + fi + # More worker PROCESSES, fewer threads per worker: a slow CPU-bound + # request (see projects/009 HANDOFF.md - belief-dedup in + # flexmeasures/data/services/time_series.py can be slow) previously + # monopolized the GIL of its gunicorn worker under --threads 4, + # starving unrelated concurrent requests until they hit client-side + # timeouts. Separate OS processes aren't GIL-bound, so this isolates + # a slow request to its own worker instead of blocking the others. + # Worker count matches the number of CEM instances below (one per + # concurrently-simulated apartment), since each apartment's RM/CEM + # can be mid-request at the same time. Bump both together when + # scaling to more apartments (see the cem-* services' own comment). + gunicorn --bind 0.0.0.0:5000 --worker-tmp-dir /dev/shm --workers 5 --threads 1 --timeout 120 wsgi:application + # One RQ worker per concurrently-simulated apartment (see x-worker comment + # above). Add or remove instances alongside the cem-* services. + worker: + <<: *worker + command: + - | + pip install --break-system-packages -e /app + pip install --break-system-packages -r /usr/var/flexmeasures-instance/requirements.txt + pip install timely-beliefs -U --break-system-packages + pip install --break-system-packages -e /app/flexmeasures-client[s2] + # Drop any lingering ingestion registration before this worker starts. + # FlexMeasures decides synchronous vs asynchronous sensor-data ingestion by asking + # Redis which workers are registered for the ingestion queue (PR #2101), and STALE + # registrations count: if a dead worker's entry survives, posts are enqueued to a + # queue nobody serves and the run hangs with no error. Removing only entries that + # name the ingestion queue is safe to run on every start - no worker here registers + # for it - and it is idempotent, so all three workers may do it concurrently. + python3 - <<'PYGUARD' + import os, redis + r = redis.Redis(host=os.environ.get("FLEXMEASURES_REDIS_URL", "queue-db"), + port=int(os.environ.get("FLEXMEASURES_REDIS_PORT", 6379)), + password=os.environ.get("FLEXMEASURES_REDIS_PASSWORD")) + try: + for name in r.smembers("rq:workers"): + key = name.decode() + queues = (r.hget(key, "queues") or b"").decode() + if "ingestion" in queues or not queues: + r.srem("rq:workers", name); r.delete(key) + r.delete("rq:workers:ingestion") + except Exception as exc: + print(f"ingestion-registration guard skipped: {exc}") + PYGUARD + flexmeasures jobs run-worker --name flexmeasures-worker-1 --queue scheduling\|forecasting + worker-2: + <<: *worker + command: + - | + pip install --break-system-packages -e /app + pip install --break-system-packages -r /usr/var/flexmeasures-instance/requirements.txt + pip install timely-beliefs -U --break-system-packages + pip install --break-system-packages -e /app/flexmeasures-client[s2] + # Drop any lingering ingestion registration before this worker starts. + # FlexMeasures decides synchronous vs asynchronous sensor-data ingestion by asking + # Redis which workers are registered for the ingestion queue (PR #2101), and STALE + # registrations count: if a dead worker's entry survives, posts are enqueued to a + # queue nobody serves and the run hangs with no error. Removing only entries that + # name the ingestion queue is safe to run on every start - no worker here registers + # for it - and it is idempotent, so all three workers may do it concurrently. + python3 - <<'PYGUARD' + import os, redis + r = redis.Redis(host=os.environ.get("FLEXMEASURES_REDIS_URL", "queue-db"), + port=int(os.environ.get("FLEXMEASURES_REDIS_PORT", 6379)), + password=os.environ.get("FLEXMEASURES_REDIS_PASSWORD")) + try: + for name in r.smembers("rq:workers"): + key = name.decode() + queues = (r.hget(key, "queues") or b"").decode() + if "ingestion" in queues or not queues: + r.srem("rq:workers", name); r.delete(key) + r.delete("rq:workers:ingestion") + except Exception as exc: + print(f"ingestion-registration guard skipped: {exc}") + PYGUARD + flexmeasures jobs run-worker --name flexmeasures-worker-2 --queue scheduling\|forecasting + worker-3: + <<: *worker + command: + - | + pip install --break-system-packages -e /app + pip install --break-system-packages -r /usr/var/flexmeasures-instance/requirements.txt + pip install timely-beliefs -U --break-system-packages + pip install --break-system-packages -e /app/flexmeasures-client[s2] + # Drop any lingering ingestion registration before this worker starts. + # FlexMeasures decides synchronous vs asynchronous sensor-data ingestion by asking + # Redis which workers are registered for the ingestion queue (PR #2101), and STALE + # registrations count: if a dead worker's entry survives, posts are enqueued to a + # queue nobody serves and the run hangs with no error. Removing only entries that + # name the ingestion queue is safe to run on every start - no worker here registers + # for it - and it is idempotent, so all three workers may do it concurrently. + python3 - <<'PYGUARD' + import os, redis + r = redis.Redis(host=os.environ.get("FLEXMEASURES_REDIS_URL", "queue-db"), + port=int(os.environ.get("FLEXMEASURES_REDIS_PORT", 6379)), + password=os.environ.get("FLEXMEASURES_REDIS_PASSWORD")) + try: + for name in r.smembers("rq:workers"): + key = name.decode() + queues = (r.hget(key, "queues") or b"").decode() + if "ingestion" in queues or not queues: + r.srem("rq:workers", name); r.delete(key) + r.delete("rq:workers:ingestion") + except Exception as exc: + print(f"ingestion-registration guard skipped: {exc}") + PYGUARD + flexmeasures jobs run-worker --name flexmeasures-worker-3 --queue scheduling\|forecasting + # One CEM per simulated household. Add or remove instances as needed; + # scale by copying an entry and bumping the host port. + cem: + <<: *cem + ports: + - "8080:8080" + cem-2: + <<: *cem + ports: + - "8081:8080" + cem-3: + <<: *cem + ports: + - "8082:8080" + cem-4: + <<: *cem + ports: + - "8083:8080" + cem-5: + <<: *cem + ports: + - "8084:8080" diff --git a/docs/CEM.rst b/docs/CEM.rst index ace53690..3defb180 100644 --- a/docs/CEM.rst +++ b/docs/CEM.rst @@ -54,12 +54,39 @@ Then point your Resource Managers (RMs) to ``http://localhost:8080/ws`` and run: uv run src/flexmeasures_client/s2/script/websockets_server.py +We also included a ``docker-compose.override.yaml`` that can be used to set up the CEM including the FlexMeasures server, creating a fully self-hosted HEMS. +Assuming your ``flexmeasures`` and ``flexmeasures-client`` repo folders are located side by side, run this from your flexmeasures folder: + +.. code-block:: bash + + docker compose \ + -f docker-compose.yml \ + -f ../flexmeasures-client/docker-compose.override.yml \ + up + + +This creates the following containers for the CEM: + +- a WebSocket server (FlexMeasures Client) +- web and worker servers (FlexMeasures) +- a database server (Postgres) +- a queue server (Redis) +- a mail server (MailHog) + To test, run the included example RM: .. code-block:: bash uv run src/flexmeasures_client/s2/script/websockets_client.py +For full access via the UI, create an admin user for the Docker Toy Account (here, we assume it has ID 1): + +.. code-block:: bash + + docker exec -it flexmeasures-server-1 bash + flexmeasures show accounts + flexmeasures add user --roles admin --account 1 --email --username + Disclaimer ========== diff --git a/examples/HEMS/configs/aggregate_reporter_param.json b/examples/HEMS/configs/aggregate_reporter_param.json new file mode 100644 index 00000000..585ccad6 --- /dev/null +++ b/examples/HEMS/configs/aggregate_reporter_param.json @@ -0,0 +1,29 @@ +{ + "input": [ + { + "name": "aggregate-1", + "sensor": 3616, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "aggregate-2", + "sensor": 3645, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + } + ], + "output": [ + { + "sensor": 3613 + } + ], + "start": "2030-01-15T20:00:00+01:00", + "end": "2030-01-16T00:00:00+01:00", + "belief_horizon": "PT0H", + "check_output_resolution": false +} diff --git a/examples/HEMS/configs/self-consumption_reporter_param.json b/examples/HEMS/configs/self-consumption_reporter_param.json new file mode 100644 index 00000000..c6abd11e --- /dev/null +++ b/examples/HEMS/configs/self-consumption_reporter_param.json @@ -0,0 +1,74 @@ +{ + "input": [ + { + "name": "production", + "sensor": 3654, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "pv-power", + "sensor": 3655, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "building-consumption", + "sensor": 3643, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "evse1-consumption", + "sensor": 3658, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "evse2-consumption", + "sensor": 3662, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "battery-power", + "sensor": 3656, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "heating-power", + "sensor": 3666, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + } + ], + "output": [ + { + "name": "self-consumption", + "sensor": 3650 + }, + { + "name": "daily-share-of-self-consumption", + "sensor": 3653 + } + ], + "start": "2030-01-15T00:00:00+01:00", + "end": "2030-01-16T00:00:00+01:00", + "belief_horizon": "PT0H", + "check_output_resolution": false +} diff --git a/examples/HEMS/configs/total-energy-costs_reporter_param.json b/examples/HEMS/configs/total-energy-costs_reporter_param.json new file mode 100644 index 00000000..1cfd7e06 --- /dev/null +++ b/examples/HEMS/configs/total-energy-costs_reporter_param.json @@ -0,0 +1,42 @@ +{ + "input": [ + { + "name": "aggregate-power", + "sensor": 3645, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "consumption-production-price", + "sensor": 3609, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + }, + { + "name": "heating-power", + "sensor": 3666, + "exclude_source_types": [ + "scheduler", + "forecaster" + ] + } + ], + "output": [ + { + "name": "total-energy-costs", + "sensor": 3651 + }, + { + "name": "daily-total-energy-costs", + "sensor": 3652 + } + ], + "start": "2030-01-15T00:00:00+01:00", + "end": "2030-01-16T00:00:00+01:00", + "belief_horizon": "PT0H", + "check_output_resolution": false +} diff --git a/pyproject.toml b/pyproject.toml index 3407ad3d..bd056a6b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -76,6 +76,13 @@ packages = ["src/flexmeasures_client"] [tool.hatch.version] source = "vcs" +[tool.hatch.version.raw-options] +# Containers mounting this repo have no git binary, so hatch-vcs +# (setuptools-scm underneath) cannot derive the version at pip-install +# time and the install fails. Fall back to a static version there; +# proper versions still come from git tags wherever git is available. +fallback_version = "0.9.3" + [tool.poe.tasks.test] help = "Run the full test suite." cmd = "pytest" diff --git a/src/flexmeasures_client/client.py b/src/flexmeasures_client/client.py index 2ece70de..4070dfcf 100644 --- a/src/flexmeasures_client/client.py +++ b/src/flexmeasures_client/client.py @@ -566,6 +566,7 @@ async def post_sensor_data( # Parameters for file upload file_path: str | None = None, belief_time_measured_instantly: bool = False, + await_ingestion: bool = True, ): """ Post sensor data for the given time range. @@ -578,6 +579,28 @@ async def post_sensor_data( The method automatically chooses the appropriate API endpoint based on the provided parameters. + Since FlexMeasures PR #2101, the server may process the ingestion of posted + data asynchronously: it returns ``202 Accepted`` with a background job id + instead of ``200 OK`` when an ingestion worker is available, and processes + the data (typically well) after the request returns. By default + (``await_ingestion=True``), this method polls the job-status endpoint until + the job reaches a terminal state, restoring read-your-writes semantics for + callers: once this call returns, the data is confirmed ingested (unless + polling times out - see below). Pass ``await_ingestion=False`` to opt out + and return as soon as the POST is acknowledged, regardless of whether + ingestion has completed. + + :param await_ingestion: If True (default) and the server responds 202 with + a job id, poll GET .../jobs/ (with exponential backoff, capped + at ``self.job_polling_timeout`` seconds) until the job finishes. + - Job status FINISHED: returns normally. + - Job status FAILED (or STOPPED/CANCELED): raises JobFailedError. + - Polling timeout reached while the job is still pending: logs an ERROR + and returns normally (this is a deliberate choice - the caller's own + safety nets, e.g. a compliance check, are expected to catch data that + never lands). + Has no effect on synchronous (200 OK) responses. + This function raises a ValueError when an unhandled status code is returned. """ # Check parameter combinations @@ -627,6 +650,7 @@ async def post_sensor_data( values=values, unit=unit, prior=prior, + await_ingestion=await_ingestion, ) else: # Type assertion to help the type checker understand that file_path is not None @@ -638,6 +662,40 @@ async def post_sensor_data( file_path=file_path, belief_time_measured_instantly=belief_time_measured_instantly, unit=unit, + await_ingestion=await_ingestion, + ) + + @staticmethod + def _ingestion_job_id(response) -> str | None: + """The background job id from a 202 response to a sensor-data post, if any. + + FlexMeasures standardised asynchronous job responses on the ``job`` field + (API v3.0-32); ``job_id`` is the older spelling, still accepted here so that + this client keeps working against servers predating that change. + """ + if not isinstance(response, dict): + return None + job_id = response.get("job") or response.get("job_id") + return job_id if isinstance(job_id, str) else None + + async def _await_sensor_data_ingestion(self, job_id: str) -> None: + """Wait for an ingestion job to finish, so that posted data is readable. + + Delegates to wait_for_job(), so ingestion follows the same polling and + back-off behaviour as every other background job this client waits on. + + A failed job propagates as JobFailedError: callers such as the S2 CEM treat + that like any other failed post and release their de-duplication key, so the + resource manager's next re-send retries it. A timeout, by contrast, is logged + and swallowed: the data may still land moments later, and refusing to continue + would strand a simulation that has its own watchdogs for incomplete data. + """ + try: + await self.wait_for_job(job_id) + except JobTimeoutError as e: + self.logger.error( + f"Ingestion not confirmed for job {job_id}; proceeding - downstream " + f"reads may miss this data. {e}" ) async def _post_sensor_data_json( @@ -648,6 +706,7 @@ async def _post_sensor_data_json( values: list[float], unit: str, prior: str | datetime | None = None, + await_ingestion: bool = True, ): """ Post sensor data using JSON payload. @@ -671,6 +730,11 @@ async def _post_sensor_data_json( ) check_for_status(status, 200) self.logger.info("Sensor data sent successfully via JSON.") + + job_id = self._ingestion_job_id(response) + if await_ingestion and status == 202 and job_id: + await self._await_sensor_data_ingestion(job_id) + return response, status async def _post_sensor_data_file( @@ -679,6 +743,7 @@ async def _post_sensor_data_file( file_path: str, belief_time_measured_instantly: bool = False, unit: str | None = None, + await_ingestion: bool = True, ): """ Post sensor data using file upload. @@ -774,6 +839,11 @@ async def _post_sensor_data_file( f"File uploaded successfully: {os.path.basename(file_path)} " f"(status {response.status})" ) + + job_id = self._ingestion_job_id(response_data) + if await_ingestion and response.status == 202 and job_id: + await self._await_sensor_data_ingestion(job_id) + return response_data, response.status except Exception as e: @@ -1302,9 +1372,9 @@ async def add_asset( self, name: str, account_id: int, - latitude: float, - longitude: float, generic_asset_type_id: int, + latitude: float | None = None, + longitude: float | None = None, parent_asset_id: int | None = None, sensors_to_show: list | None = None, flex_context: dict | None = None, @@ -1333,10 +1403,12 @@ async def add_asset( asset = dict( name=name, account_id=account_id, - latitude=latitude, - longitude=longitude, generic_asset_type_id=generic_asset_type_id, ) + if latitude is not None: + asset["latitude"] = str(latitude) + if longitude is not None: + asset["longitude"] = str(longitude) if parent_asset_id: asset["parent_asset_id"] = parent_asset_id if sensors_to_show: @@ -1580,7 +1652,12 @@ async def trigger_schedule( asset_id: int | None = None, prior: datetime | None = None, scheduler: str | None = None, + metadata: dict | None = None, ) -> str: + """metadata: opaque orchestration metadata passed alongside the domain + payload (a sibling of flex-model/flex-context, mirroring the S2 wrapper + convention). Requires a server that accepts the trigger "metadata" + field; it is passed through verbatim to the Scheduler.""" if (sensor_id is None) == (asset_id is None): raise ValueError("Pass either a sensor_id or an asset_id.") message = { @@ -1593,6 +1670,8 @@ async def trigger_schedule( message["flex-model"] = flex_model if flex_context is not None: message["flex-context"] = flex_context + if metadata is not None: + message["metadata"] = metadata force_new_job_creation_requested = False if prior is not None: diff --git a/src/flexmeasures_client/constants.py b/src/flexmeasures_client/constants.py index 60e832be..573a4e08 100644 --- a/src/flexmeasures_client/constants.py +++ b/src/flexmeasures_client/constants.py @@ -3,3 +3,9 @@ "Content-Type": CONTENT_TYPE, } API_VERSION = "v3_0" + +# Ceiling for the exponential backoff between polls (see client.py / +# response_handling.py): without it, later polling steps sleep for minutes +# (polling_interval * 2**step), dozing far past the moment a result becomes +# available - and taking just as long to conclude that it never will. +MAX_POLLING_SLEEP = 10.0 # seconds diff --git a/src/flexmeasures_client/s2/cem.py b/src/flexmeasures_client/s2/cem.py index 7e83cd8b..f9474b53 100644 --- a/src/flexmeasures_client/s2/cem.py +++ b/src/flexmeasures_client/s2/cem.py @@ -1,13 +1,14 @@ from __future__ import annotations +import asyncio import json import logging import math -from asyncio import Queue from collections import defaultdict from datetime import datetime, timedelta from logging import Logger from typing import Dict, Optional +from zoneinfo import ZoneInfo import pandas as pd import pydantic @@ -33,8 +34,10 @@ from flexmeasures_client.client import FlexMeasuresClient from flexmeasures_client.s2 import Handler, register +from flexmeasures_client.s2.config_utils import configure_site from flexmeasures_client.s2.control_types import ControlTypeHandler from flexmeasures_client.s2.utils import ( + ControlContext, get_latest_compatible_version, get_reception_status, get_unique_id, @@ -42,6 +45,16 @@ _LOGGER = logging.getLogger(__name__) +# Process-global measurement-post bookkeeping. The CEM object is rebuilt for +# every RM websocket (re)connection - in co-simulation that is EVERY replan +# round - so instance-level sets reset each round: the dedupe never held (a +# profiled 1-day run re-posted each house's realized series ~3x, ~13k calls) +# and a new instance's flush barrier could not await the old instance's still +# in-flight posts. One CEM server process serves one apartment, so process +# scope is the correct lifetime for both. +_POSTED_MEASUREMENT_KEYS: set[tuple] = set() +_PENDING_MEASUREMENT_POSTS: set = set() + class CEM(Handler): __version__ = "0.0.2-beta" @@ -49,7 +62,6 @@ class CEM(Handler): _resource_manager_details: ResourceManagerDetails _control_types_handlers: Dict[ControlType | None, ControlTypeHandler] - _control_type = None _is_closed = True _default_control_type: ControlType | None @@ -58,7 +70,7 @@ class CEM(Handler): ] # maps the CommodityQuantity power measurement sensors to FM sensor IDs _fm_client: FlexMeasuresClient - _sending_queue: Queue[pydantic.BaseModel] + _sending_queue: asyncio.Queue[tuple[pydantic.BaseModel, asyncio.Future]] _timers: dict[str, datetime] _datastore: dict @@ -81,12 +93,33 @@ def __init__( """ Customer Energy Manager (CEM) """ + # Initialize per-instance control context and handler build tasks BEFORE calling super().__init__() + # because parent's __init__ calls discover() which accesses control_type property + self._control = ControlContext() + self._handler_build_tasks: dict[ControlType, asyncio.Task] = {} + super(CEM, self).__init__() self._fm_client = fm_client - self._sending_queue = Queue() + self._sending_queue = asyncio.Queue() self._power_sensors = dict() self.power_sensor_id = power_sensor_id + # The apartment's FlexMeasures asset id (set once mapped), and the id of its + # flex-context "aggregate-power" sensor (attached by the community runner AFTER + # this CEM connects, i.e. only resolvable from step 1 onward). Resolved lazily in + # handle_power_measurement so realized apartment power lands on the aggregate-power + # sensor too, not only on measured-power (defect 4a: live apartment realizations). + self._apartment_asset_id: int | None = None + self._aggregate_power_sensor_id: int | None = None + # In-flight measurement posts (see handle_power_measurement): posts run + # concurrently as tasks; flush_measurement_posts() is the barrier that + # schedule triggering awaits so FlexMeasures reads see all realized data. + # Content keys dedupe the series the RM re-sends with every retrigger + # round's flexinput; a failed post drops its key so a later re-send + # retries it. Both live at PROCESS scope (see module globals): the CEM + # object itself is rebuilt on every RM reconnection. + self._pending_measurement_posts = _PENDING_MEASUREMENT_POSTS + self._posted_measurement_keys = _POSTED_MEASUREMENT_KEYS self._control_types_handlers = dict() self._default_control_type = default_control_type @@ -141,7 +174,7 @@ def is_closed(self): @property def control_type(self): - return self._control_type + return self._control.control_type def register_control_type(self, control_type_handler: ControlTypeHandler): """ @@ -151,15 +184,19 @@ def register_control_type(self, control_type_handler: ControlTypeHandler): # skip registering if there's a handler already registered for # the same control type if control_type_handler._control_type in self._control_types_handlers: - self._logger.warning( + self._logger.debug( "Control Type {control_type} already registered. Updating..." ) # add fm_client to control_type handler control_type_handler._fm_client = self._fm_client - # add sending queue - control_type_handler._sending_queue = self._sending_queue + # back-reference so handlers can await flush_measurement_posts() before + # triggering a schedule (realized-power posts run concurrently) + control_type_handler._cem = self + + # add send_message method so the handler can send messages + control_type_handler.send_message = self.send_message # Add logger control_type_handler._logger = self._logger @@ -169,6 +206,9 @@ def register_control_type(self, control_type_handler: ControlTypeHandler): control_type_handler ) + # Mark handler as ready once registered + self._control.handler_ready[control_type_handler._control_type] = True + async def handle_message(self, message: Dict | pydantic.BaseModel | str): """ This method handles the incoming messages to the CEM and routes them to their custom handler. @@ -184,22 +224,59 @@ async def handle_message(self, message: Dict | pydantic.BaseModel | str): if isinstance(message, str): message = json.loads(message) - self._logger.debug(f"Received: {message}") + # Detect wrapper + if isinstance(message, dict) and "message" in message and "metadata" in message: + metadata = message["metadata"] + message = message["message"] + self._logger.debug("Received wrapped message") + self._logger.debug(f"Received message: {message}") + self._logger.debug(f"Received metadata: {metadata}") + if "dt" in metadata: + for control_type in self._control_types_handlers.values(): + control_type.now = lambda: metadata["dt"] # type: ignore + self.now = lambda: metadata["dt"] # type: ignore + else: + self._logger.debug(f"Received: {message}") + + # Implicit control-type activation: per the S2 protocol, an RM only + # sends control-type-specific messages (e.g. FRBC.*) AFTER it has + # accepted our SelectControlType - so such a message is itself proof + # that the control type is active on the RM side. Without this, a + # dropped/mis-routed ReceptionStatus for SelectControlType left the + # CEM stuck in NO_SELECTION forever: every FRBC.SystemDescription was + # answered with TEMPORARY_ERROR, the RM retried indefinitely, and the + # whole simulation hung at its first barrier (observed in vivo, + # 2026-07-26, house APP_4). + message_type = ( + message.get("message_type", "") if isinstance(message, dict) else "" + ) + if ( + self._control.control_type in (None, ControlType.NO_SELECTION) + and isinstance(message_type, str) + and message_type.startswith("FRBC.") + and ControlType.FILL_RATE_BASED_CONTROL in self._control_types_handlers + ): + self._logger.warning( + "Received %s while no control type is active; the RM only " + "sends FRBC messages after accepting SelectControlType, so " + "activating FILL_RATE_BASED_CONTROL implicitly (the " + "SelectControlType confirmation was presumably lost).", + message_type, + ) + self._control.control_type = ControlType.FILL_RATE_BASED_CONTROL # try to handle the message with the control_type handle + ct = self._control.control_type + handler = self._control_types_handlers.get(ct) + ready = self._control.handler_ready.get(ct, False) + if ( - self._control_type is not None - and ( - self._control_type - not in [ControlType.NO_SELECTION, ControlType.NOT_CONTROLABLE] - ) - and self._control_types_handlers[self._control_type].supports_message( - message - ) + handler is not None + and ready + and ct not in [ControlType.NO_SELECTION, ControlType.NOT_CONTROLABLE] + and handler.supports_message(message) ): - response = await self._control_types_handlers[ - self._control_type - ].handle_message(message) + response = await handler.handle_message(message) else: if self.supports_message(message): response = await super().handle_message( @@ -215,24 +292,33 @@ async def handle_message(self, message: Dict | pydantic.BaseModel | str): ) if response is not None: - await self._sending_queue.put(response) + await self.send_message(response) def update_control_type(self, control_type: ControlType): """ Callback function that is triggered when we receive a confirmation that the message has been received. """ - self._control_type = control_type + self._control.control_type = control_type - async def get_message(self) -> str: + async def get_message(self) -> tuple[str, asyncio.Future]: """Call this function to get the messages to be sent to the RM Returns: str: message in JSON format """ - message = await self._sending_queue.get() - return message.model_dump(mode="json") + item = await self._sending_queue.get() + + if not isinstance(item, tuple) or len(item) != 2: + raise RuntimeError( + "Invalid item in sending queue. All messages must go through send_message() rather than _sending_queue.put()." + ) + + message, fut = item + message = message.model_dump(mode="json") + + return message, fut async def activate_control_type( self, control_type: ControlType @@ -242,24 +328,24 @@ async def activate_control_type( """ # check if it's trying to activate the current control_type - if control_type == self._control_type: - self._logger.warning(f"RM is already in `{control_type}` control type.") + if control_type == self._control.control_type: + self._logger.debug(f"RM is already in `{control_type}` control type.") return None # check if the RM supports the control type if control_type not in self._resource_manager_details.available_control_types: - self._logger.warning(f"RM does not support `{control_type}` control type.") + self._logger.debug(f"RM does not support `{control_type}` control type.") return None # RM initialization succeeded - if self._control_type is not None: + if self._control.control_type is not None: message_id = get_unique_id() # the callback `update_control_type` will be called upon arrival of a # ReceptionStatus message with status = ReceptionStatusValues.OK # register callback in CEM handler - if self._control_type in [ + if self._control.control_type in [ ControlType.NOT_CONTROLABLE, ControlType.NO_SELECTION, ]: @@ -268,12 +354,11 @@ async def activate_control_type( ) else: # register callback in control mode handler self._control_types_handlers[ - self._control_type + self._control.control_type ].register_success_callbacks( message_id, self.update_control_type, control_type=control_type ) - - await self._sending_queue.put( + await self.send_message( SelectControlType(message_id=message_id, control_type=control_type) ) return None @@ -300,11 +385,21 @@ async def handle_handshake(self, message: Handshake): async def handle_resource_manager_details(self, message: ResourceManagerDetails): self._resource_manager_details = message + # schedule map_resource_to_asset to run soon concurrently + task = asyncio.create_task(self.map_resource_to_asset(message)) + self._handler_build_tasks[ControlType.FILL_RATE_BASED_CONTROL] = task + # self.background_tasks.add( + # task + # ) # important to avoid a task disappearing mid-execution. + # task.add_done_callback(self.background_tasks.discard) + + # await self.map_resource_to_asset(message) + if ( - not self._control_type + not self._control.control_type ): # initializing. TODO: check if sending resource_manager_details # resets control type - self._control_type = ControlType.NO_SELECTION + self._control.control_type = ControlType.NO_SELECTION # Activate default control type if defined if self._default_control_type: @@ -312,88 +407,265 @@ async def handle_resource_manager_details(self, message: ResourceManagerDetails) return get_reception_status(message) + async def map_resource_to_asset(self, message): + """Map S2 resource to FM asset. + + - Creates a new asset if the resource ID does not yet exist in FlexMeasures. + - Updates the existing asset if the resource details changed. + - Updates the control type for the resource. + """ + assets = await self._fm_client.get_assets() + asset = None + for ast in assets: + if ast["external_id"] == message.resource_id: + asset = ast + if asset is None: + # Fall back to matching by name: servers without external_id support + # cannot persist the resource ID, and RMs generate a fresh resource ID + # on every connection, so a returning RM would otherwise cause a + # duplicate-name asset creation attempt. + for ast in assets: + if ast["name"] == message.name: + asset = ast + if asset is None: + self._logger.debug( + f"HANGDEBUG map_resource_to_asset: no existing asset found for " + f"{message.name!r}, creating a new one" + ) + account = await self._fm_client.get_account() + asset = await self._fm_client.add_asset( + name=message.name, + account_id=account["id"], + generic_asset_type_id=1, + # parent_asset_id=self._asset_id, + attributes=json.loads(message.to_json()), + ) + self._logger.debug( + f"HANGDEBUG map_resource_to_asset: created asset id={asset['id']}" + ) + else: + self._logger.debug( + f"HANGDEBUG map_resource_to_asset: reusing existing asset " + f"id={asset['id']} name={asset['name']!r} for {message.name!r}" + ) + if asset["name"] != message.name: + await self._fm_client.update_asset( + asset_id=asset["id"], updates={"name": message.name} + ) + if asset["attributes"] != message.to_json(): + await self._fm_client.update_asset( + asset_id=asset["id"], + updates={"attributes": json.loads(message.to_json())}, + ) + + # Reconfigure site + ( + price_sensor, + production_price_sensor, + power_sensor, + soc_sensor, + rm_discharge_sensor, + soc_minima_sensor, + soc_maxima_sensor, + usage_forecast_sensor, + leakage_behaviour_sensor, + charging_efficiency_sensor, + measured_power_sensor, + ) = await configure_site(message.name, self._fm_client) + + # Wire up the apartment's dedicated MEASUREMENT sensor (distinct from the + # "power" SCHEDULE sensor above) so incoming S2 PowerMeasurements land on + # their own sensor instead of a hardcoded/wrong one. + if self.power_sensor_id is None: + self.power_sensor_id = {} + self.power_sensor_id["ELECTRIC.POWER.L1"] = measured_power_sensor["id"] + + # Remember the apartment asset so handle_power_measurement can lazily resolve its + # flex-context "aggregate-power" sensor (attached by the community runner only + # after this CEM has connected). Resetting the cached sensor id lets a + # reconnect/reconfigure pick up a freshly-attached aggregate-power sensor. + self._apartment_asset_id = asset["id"] + self._aggregate_power_sensor_id = None + + from flexmeasures_client.s2.control_types.FRBC.frbc_simple import FRBCSimple + + frbc = FRBCSimple( + power_sensor_id=power_sensor["id"], + price_sensor_id=price_sensor["id"], + production_price_sensor_id=production_price_sensor["id"], + soc_sensor_id=soc_sensor["id"], + rm_discharge_sensor_id=rm_discharge_sensor["id"], + soc_minima_sensor_id=soc_minima_sensor["id"], + soc_maxima_sensor_id=soc_maxima_sensor["id"], + usage_forecast_sensor_id=usage_forecast_sensor["id"], + leakage_behaviour_sensor_id=leakage_behaviour_sensor["id"], + charging_efficiency_sensor_id=charging_efficiency_sensor["id"], + ) + self.register_control_type(frbc) + @register(PowerMeasurement) async def handle_power_measurement(self, message: PowerMeasurement): for power_measurement in message.values: commodity_quantity = power_measurement.commodity_quantity.value - if ( - self.power_sensor_id is None - and commodity_quantity == "ELECTRIC.POWER.L1" - ): - sensor_id = 357 - elif self.power_sensor_id: - s_id = self.power_sensor_id.get(commodity_quantity) - if s_id is None: + if self.power_sensor_id: + sensor_id = self.power_sensor_id.get(commodity_quantity) + if sensor_id is None: # TODO: create a new sensor or return ReceptionStatus self._logger.debug( f"No power sensor set up for {commodity_quantity}. Ignoring measurement {power_measurement.value} at {message.measurement_timestamp}." ) continue - sensor_id = s_id else: - self._logger.warning( + self._logger.debug( f"No power sensor IDs set up. Ignoring measurement {power_measurement.value} at {message.measurement_timestamp}." ) continue - # Store the value in the buffer - self._power_buffer[commodity_quantity].append( - (message.measurement_timestamp, power_measurement.value) + # Bin to the SIMULATED measurement timestamp (not wall-clock time): + # this co-sim runs on 2022 simulated time, so datetime.now() would bin + # measurements onto a meaningless (real-world) event_start. + measurement_ts = message.measurement_timestamp + if measurement_ts.tzinfo is None: + # This co-sim treats naive sim times as Europe/Amsterdam (kept + # consistent with the community orchestrator, RM and controller, + # which all localize the same naive sim times as Europe/Amsterdam). + measurement_ts = measurement_ts.replace( + tzinfo=ZoneInfo("Europe/Amsterdam") + ) + # Bin and stamp at the measured-power sensor's own 15-minute event + # resolution (the RM sends one value per 15-minute interval, stamped at + # the interval START). Using the 5-minute _minimum_measurement_period + # here made prior = bin_start + 5min, i.e. a belief 10 minutes BEFORE + # the event's end - an ex-ante horizon on what is a measurement. With + # the event resolution, prior = the interval's end: zero belief horizon. + period = pd.Timedelta(minutes=15) + m = period // pd.Timedelta(minutes=1) + bin_start = measurement_ts.replace( + second=0, microsecond=0, minute=(measurement_ts.minute // m) * m ) - - # Compute bin - now = datetime.now(self._timezone) - m = self._minimum_measurement_period // pd.Timedelta(minutes=1) - bin_end = now.replace( - second=0, microsecond=0, minute=(now.minute // m) * m - ) # e.g. 10:15:00 - bin_start = bin_end - self._minimum_measurement_period - - # If timer not due, just collect values - if not self._is_timer_due(f"power_measurement_{commodity_quantity}"): - self._logger.debug( - f"Collecting 5-minute average for {commodity_quantity} ({bin_start.isoformat()} – {bin_end.isoformat()})" + # Belief time = the SIMULATED instant the measurement became known (the + # moment its interval elapses). Without this, FlexMeasures stamps the + # belief time at wall-clock now (2026), so the UI's horizon view shows + # realized data "recorded in 2026" instead of at simulation time. + prior = (bin_start + period).isoformat() + + # Resolve the apartment's flex-context "aggregate-power" sensor lazily: it is + # attached by the community runner AFTER this CEM connects, so it only exists + # from step 1 onward. When present, mirror the realized value onto it too, so + # advancing the sim produces live apartment realizations on the aggregate-power + # sensor (defect 4a) - not just on the dedicated measured-power sensor. That + # sensor also carries StorageScheduler SCHEDULE data natively; the realized + # posts stay distinguishable by their own (CEM/user) source and simulated + # belief time. Only mirror ELECTRIC.POWER.L1 (the aggregated apartment power). + aggregate_sensor_id = None + if commodity_quantity == "ELECTRIC.POWER.L1": + aggregate_sensor_id = await self._resolve_aggregate_power_sensor_id() + + # Post DIRECTLY, without the wall-clock _is_timer_due throttle or the + # 5-minute buffered-averaging that the original real-time streaming path + # used. In this co-simulation the RM sends exactly one already-aggregated + # apartment-power value per simulated step, and simulated time advances at + # its own (non-real-time) pace; gating on datetime.now() would suppress + # almost every post, and re-averaging a single value is a no-op. Each + # value is simply written at its own simulated event_start. + target_sensor_ids = [sensor_id] + if aggregate_sensor_id is not None and aggregate_sensor_id != sensor_id: + target_sensor_ids.append(aggregate_sensor_id) + for target_sensor_id in target_sensor_ids: + measurement_key = ( + target_sensor_id, + bin_start.isoformat(), + float(power_measurement.value), ) - continue - - # Compute average of all buffered values in last 5 minutes - buffer = self._power_buffer[commodity_quantity] - period_values = [v for (t, v) in buffer if bin_start <= t < bin_end] - - if not period_values: - self._logger.debug( - f"No samples found for {commodity_quantity} in {bin_start}–{bin_end}, skipping." + if measurement_key in self._posted_measurement_keys: + continue + self._posted_measurement_keys.add(measurement_key) + # Post CONCURRENTLY: the RM forwards realized power as one + # PowerMeasurement per 15-min interval, serially, and awaiting + # each post here serialized ~200 HTTP round-trips per house per + # simulated day on the co-sim critical path. Each post keeps its + # own event_start/prior exactly as before; the posts merely + # overlap in time. flush_measurement_posts() is the barrier for + # readers (awaited before any schedule trigger). + task = asyncio.create_task( + self._post_measurement_safely( + measurement_key, + target_sensor_id, + start=bin_start.isoformat(), + duration=period.isoformat(), # TODO: not specified in S2 Protocol + values=[power_measurement.value], + # S2 PowerMeasurement values are in Watts (unlike this codebase's + # S2 power *ranges*, which carry kW-magnitude values that + # get_commodity_unit labels "kW"). Post as W and let FlexMeasures + # convert to the sensor's kW unit, so a ~5000 W realized load is + # stored as 5 kW, not 5000. + unit="W", + prior=prior, + ) ) - continue + self._pending_measurement_posts.add(task) + task.add_done_callback(self._pending_measurement_posts.discard) - avg_value = sum(period_values) / len(period_values) - self._logger.debug( - f"Posting 5-minute average for {commodity_quantity}: " - f"{avg_value} ({bin_start.isoformat()} – {bin_end.isoformat()})" - ) + return get_reception_status(message) - # Send measurement - try: - await self._fm_client.post_sensor_data( - sensor_id, - start=bin_start.isoformat(), - duration=self._minimum_measurement_period.isoformat(), # TODO: not specified in S2 Protocol - values=[avg_value], - unit=get_commodity_unit(commodity_quantity), - ) - except Exception as e: # noqa: B902 - intentional safety net - self._logger.warning( - f"POSTing power measurement failed with error: {e}" - ) + async def _post_measurement_safely( + self, measurement_key, target_sensor_id, **kwargs + ) -> None: + """Post one measurement, swallowing errors exactly like the historical + inline try/except did (a lost measurement must not break the S2 session). + On failure the content key is released so the RM's next re-send of the + same series retries the post instead of being deduplicated away.""" + try: + await self._fm_client.post_sensor_data(target_sensor_id, **kwargs) + except Exception as e: # noqa: B902 - intentional safety net + self._posted_measurement_keys.discard(measurement_key) + self._logger.debug(f"POSTing power measurement failed with error: {e}") + + async def flush_measurement_posts(self) -> None: + """Barrier: wait until all in-flight realized-power posts have landed. + + Called before triggering a FlexMeasures schedule so the scheduler (and the + community compliance check it feeds) sees the complete realized series - + the same guarantee the old serial awaits gave. Tasks never raise (see + _post_measurement_safely).""" + while self._pending_measurement_posts: + await asyncio.gather( + *list(self._pending_measurement_posts), return_exceptions=True + ) - # Keep only samples newer than this bin (for next period) - self._power_buffer[commodity_quantity] = [ - (t, v) for (t, v) in buffer if t >= bin_end - ] + async def _resolve_aggregate_power_sensor_id(self) -> int | None: + """Lazily resolve the apartment's flex-context "aggregate-power" sensor id. - return get_reception_status(message) + The community runner attaches this sensor (and the flex-context key) only after + the CEM has connected, so it is absent during step 0 and appears from step 1 + onward. We re-read the apartment asset's flex_context until the key is present, + then cache the id. Returns None (and posts nothing extra) while it is absent. + """ + if self._aggregate_power_sensor_id is not None: + return self._aggregate_power_sensor_id + if self._apartment_asset_id is None: + return None + try: + asset = await self._fm_client.get_asset( + self._apartment_asset_id, parse_json_fields=True + ) + flex_context = asset.get("flex_context") or {} + entry = flex_context.get("aggregate-power") + if isinstance(entry, dict): + sensor_id = entry.get("sensor") + if sensor_id is not None: + self._aggregate_power_sensor_id = int(sensor_id) + return self._aggregate_power_sensor_id + except ( + Exception + ) as e: # noqa: B902 - best-effort, never break measurement posting + self._logger.debug( + f"Could not resolve aggregate-power sensor for asset " + f"{self._apartment_asset_id}: {e}" + ) + return None @register(RevokeObject) def handle_revoke_object(self, message: RevokeObject): @@ -418,8 +690,10 @@ def handle_revoke_object(self, message: RevokeObject): return get_reception_status(message, ReceptionStatusValues.OK) async def send_message(self, message): + loop = asyncio.get_running_loop() + fut = loop.create_future() self._logger.debug(f"Sent: {message}") - await self._sending_queue.put(message) + await self._sending_queue.put((message, fut)) def get_commodity_unit(commodity_quantity) -> str: diff --git a/src/flexmeasures_client/s2/config_utils.py b/src/flexmeasures_client/s2/config_utils.py new file mode 100644 index 00000000..85eeedee --- /dev/null +++ b/src/flexmeasures_client/s2/config_utils.py @@ -0,0 +1,300 @@ +import asyncio +import logging +import os +from datetime import datetime +from zoneinfo import ZoneInfo + +from flexmeasures_client.client import FlexMeasuresClient + +log_level = os.getenv("LOGGING_LEVEL", "WARNING").upper() +logging.basicConfig( + level=log_level, + format="[CEM][%(asctime)s] %(levelname)s: %(name)s | %(message)s", +) +LOGGER = logging.getLogger(__name__) + +# site_name -> asset id, per CEM server process (see configure_site) +_SITE_ASSET_IDS: dict[str, int] = {} + + +async def configure_site( + site_name: str, fm_client: FlexMeasuresClient +) -> tuple[dict, dict, dict, dict, dict, dict, dict, dict, dict, dict, dict]: + account = await fm_client.get_account() + + # Find the site asset with a lean id/name listing, then fetch only the + # match. This runs on EVERY RM (re)connection - in co-simulation that is + # every replan round - and the previous full-catalog get_assets scan cost + # ~5-7 s per call against a large database (about half the API time of a + # profiled co-simulation day). Falls back to the full scan on servers + # too old for the `fields` parameter (< 0.31). + site_asset: dict | None = None + # Asset ids are stable for the lifetime of this CEM server process (one + # process per apartment; the RM reconnects every replan round, and even + # the lean listing costs seconds against a large database), so remember + # the resolved id and fetch it directly on reconnections. A stale id + # (asset deleted between runs never happens within a process lifetime, + # but be safe) falls through to a fresh lookup. + cached_id = _SITE_ASSET_IDS.get(site_name) + if cached_id is not None: + try: + candidate = await fm_client.get_asset( + asset_id=cached_id, parse_json_fields=True + ) + if candidate.get("name") == site_name: + site_asset = candidate + except Exception: + _SITE_ASSET_IDS.pop(site_name, None) + if site_asset is None: + try: + asset_listing = await fm_client.get_assets( + fields=["id", "name"], parse_json_fields=False + ) + for asset in asset_listing: + if asset.get("name") == site_name: + site_asset = await fm_client.get_asset( + asset_id=asset["id"], parse_json_fields=True + ) + break + except ValueError: + assets = await fm_client.get_assets(parse_json_fields=True) + for asset in assets: + if asset["name"] == site_name: + site_asset = asset + break + + site_asset_specs = dict( + latitude=0, + longitude=0, + generic_asset_type_id=6, # Building asset type + flex_model={ + "power-capacity": f"{3 * 25 * 230} VA", + }, + ) + + if not site_asset: + LOGGER.debug(f"HANGDEBUG configure_site: creating new asset for {site_name!r}") + site_asset = await fm_client.add_asset( + name=site_name, account_id=account["id"], **site_asset_specs + ) + else: + LOGGER.debug( + f"HANGDEBUG configure_site: reusing existing asset id={site_asset['id']} " + f"for {site_name!r}, existing sensors: " + f"{[s['name'] for s in site_asset.get('sensors', [])]}" + ) + _SITE_ASSET_IDS[site_name] = site_asset["id"] + + # Update site asset with the latest specs + LOGGER.debug(f"HANGDEBUG configure_site: updating asset {site_asset['id']} specs") + await fm_client.update_asset(site_asset["id"], site_asset_specs) + LOGGER.debug(f"HANGDEBUG configure_site: asset {site_asset['id']} specs updated") + + sensors = site_asset.get("sensors", []) + price_sensor = None + production_price_sensor = None + power_sensor = None + measured_power_sensor = None + soc_sensor = None + rm_discharge_sensor = None + soc_minima_sensor = None + soc_maxima_sensor = None + usage_forecast_sensor = None + leakage_behaviour_sensor = None + charging_efficiency_sensor = None + for sensor in sensors: + if sensor["name"] == "price": + price_sensor = sensor + if sensor["name"] == "production price": + production_price_sensor = sensor + elif sensor["name"] == "power": + power_sensor = sensor + elif sensor["name"] == "measured-power": + measured_power_sensor = sensor + elif sensor["name"] == "state of charge": + soc_sensor = sensor + elif sensor["name"] == "RM discharge": + rm_discharge_sensor = sensor + elif sensor["name"] == "soc-minima": + soc_minima_sensor = sensor + elif sensor["name"] == "soc-maxima": + soc_maxima_sensor = sensor + elif sensor["name"] == "usage-forecast": + usage_forecast_sensor = sensor + elif sensor["name"] == "leakage-behaviour": + leakage_behaviour_sensor = sensor + elif sensor["name"] == "charging-efficiency": + charging_efficiency_sensor = sensor + + if price_sensor is None: + price_sensor = await fm_client.add_sensor( + name="price", + event_resolution="PT15M", + unit="EUR/kWh", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + if production_price_sensor is None: + production_price_sensor = await fm_client.add_sensor( + name="production price", + event_resolution="PT15M", + unit="EUR/kWh", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + + # Continue immediately without awaiting + LOGGER.debug("Posting 3 days of prices in a background task..") + start_of_today = ( + datetime.now(ZoneInfo("Europe/Amsterdam")) + .replace(hour=0, minute=0, second=0, microsecond=0) + .isoformat() + ) + asyncio.create_task( + fm_client.post_sensor_data( + sensor_id=price_sensor["id"], + start=start_of_today, + prior="2026-01-01T00:00+01", # 2026-01-01T00:00+01 + duration="P3D", # P1M + values=[0.3], + unit="EUR/kWh", + ) + ) + asyncio.create_task( + fm_client.post_sensor_data( + sensor_id=production_price_sensor["id"], + start=start_of_today, + prior="2026-01-01T00:00+01", # 2026-01-01T00:00+01 + duration="P3D", # P1M + values=[0.2], + unit="EUR/kWh", + ) + ) + if power_sensor is None: + power_sensor = await fm_client.add_sensor( + name="power", + event_resolution="PT15M", + unit="kW", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + # is_strictly_non_positive marks this FRBC device (a heat pump) as + # physically unable to produce: FlexMeasures then keeps production + # hard-bounded at zero even when relax-constraints turns directional + # capacities into soft, breach-priced constraints. Without it, a + # tight site capacity made the scheduler "produce" from the heat + # pump (cheap device breach vs expensive site breach). + attributes={ + "consumption_is_positive": True, + "is_strictly_non_positive": True, + }, + ) + if measured_power_sensor is None: + # Dedicated sensor for the REALIZED (measured) aggregated apartment power, + # kept separate from the "power" sensor which holds the StorageScheduler's + # device-power SCHEDULE. Never conflate schedule and measurement on one sensor. + measured_power_sensor = await fm_client.add_sensor( + name="measured-power", + event_resolution="PT15M", + unit="kW", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + attributes={"consumption_is_positive": True}, + ) + if soc_sensor is None: + soc_sensor = await fm_client.add_sensor( + name="state of charge", + event_resolution="PT0M", + unit="kWh", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + if rm_discharge_sensor is None: + rm_discharge_sensor = await fm_client.add_sensor( + name="RM discharge", + event_resolution="PT15M", + unit="dimensionless", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + if soc_minima_sensor is None: + soc_minima_sensor = await fm_client.add_sensor( + name="soc-minima", + event_resolution="PT15M", + unit="kWh", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + if soc_maxima_sensor is None: + soc_maxima_sensor = await fm_client.add_sensor( + name="soc-maxima", + event_resolution="PT15M", + unit="kWh", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + if usage_forecast_sensor is None: + usage_forecast_sensor = await fm_client.add_sensor( + name="usage-forecast", + event_resolution="PT15M", + unit="kW", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + if leakage_behaviour_sensor is None: + leakage_behaviour_sensor = await fm_client.add_sensor( + name="leakage-behaviour", + event_resolution="PT15M", + unit="%", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + if charging_efficiency_sensor is None: + charging_efficiency_sensor = await fm_client.add_sensor( + name="charging-efficiency", + event_resolution="PT15M", + unit="%", + generic_asset_id=site_asset["id"], + timezone="Europe/Amsterdam", + ) + sensors_to_show = [ + { + "title": "State of charge", + # Fit the y-axis to the data: SoC values (large thermal-storage + # fill levels) sit far from zero, so the default zero-including + # axis flattens all variation into a sliver at the top. + "y-axis": "data", + "sensors": [ + soc_minima_sensor["id"], + soc_maxima_sensor["id"], + soc_sensor["id"], + ], + }, + { + "title": "Prices", + "sensors": [price_sensor["id"], production_price_sensor["id"]], + }, + { + "title": "Power", + "sensors": [power_sensor["id"], measured_power_sensor["id"]], + }, + ] + await fm_client.update_asset( + asset_id=site_asset["id"], + updates=dict(sensors_to_show=sensors_to_show), + ) + LOGGER.debug( + f"HANGDEBUG configure_site: done, returning sensors for asset {site_asset['id']}" + ) + return ( + price_sensor, + production_price_sensor, + power_sensor, + soc_sensor, + rm_discharge_sensor, + soc_minima_sensor, + soc_maxima_sensor, + usage_forecast_sensor, + leakage_behaviour_sensor, + charging_efficiency_sensor, + measured_power_sensor, + ) diff --git a/src/flexmeasures_client/s2/control_types/FRBC/__init__.py b/src/flexmeasures_client/s2/control_types/FRBC/__init__.py index 2b4ccf15..a2e57486 100644 --- a/src/flexmeasures_client/s2/control_types/FRBC/__init__.py +++ b/src/flexmeasures_client/s2/control_types/FRBC/__init__.py @@ -3,7 +3,14 @@ import pydantic try: - from s2python.common import ControlType, ReceptionStatusValues + from s2python.common import ( + ControlType, + InstructionStatusUpdate, + ReceptionStatus, + ReceptionStatusValues, + RevokableObjects, + RevokeObject, + ) from s2python.frbc import ( FRBCActuatorStatus, FRBCFillLevelTargetProfile, @@ -62,10 +69,150 @@ def __init__(self, max_size: int = 100) -> None: self._usage_forecast_history = SizeLimitOrderedDict(max_size=max_size) self.background_tasks = set() + # --- Outstanding-work registry (2026-07, instruction-storm fix) --- + # Supersession counters: one per request kind (e.g. "storage_status"). + # Every incoming request bumps its counter; a spawned handler task + # records the value and no-ops if a newer request has arrived by the + # time it runs (or, for long chains, re-checks before queueing + # instructions). In co-simulation, retrigger rounds share one + # simulated timestamp while steering signals change server-side, so + # time-based throttles (live-system style, see the TUNES handler's + # timers) cannot distinguish stale work from fresh work - only + # request ordering can. Without this, backlogged handler tasks each + # completed a full FlexMeasures round trip and dumped a full + # instruction batch, hours-stale batches included (observed in vivo: + # 96 -> 2,225 instructions per burst, ack starvation on the shared + # sending queue, RM retry storms, and 800 s control timeouts). + self._supersession_counters: dict[str, int] = {} + # Instructions of the CURRENT batch, by message_id (cleared when a + # new batch replaces them), plus any status updates the RM reports. + # Modeled after the flexmeasures-s2 plugin's ConnectionState registry. + self._sent_instructions: dict[str, FRBCInstruction] = {} + self._instruction_statuses: dict[str, str] = {} + # Whether replacing a batch also sends RevokeObject for the previous + # batch's instructions. Off by default: not every RM implements + # RevokeObject handling (the FLEXED co-simulation RM does not; it + # keys instructions to its current request and drops late ones). + # Live deployments with a compliant RM should enable this. + self.send_revocations: bool = False + + def _bump_generation(self, key: str) -> int: + """Register a new request of the given kind; newer requests supersede + older ones (see _is_current_generation).""" + generation = self._supersession_counters.get(key, 0) + 1 + self._supersession_counters[key] = generation + return generation + + def _is_current_generation(self, key: str, generation: int) -> bool: + return self._supersession_counters.get(key, 0) == generation + + @staticmethod + def filter_instruction_transitions( + instructions: list[FRBCInstruction], + ) -> list[FRBCInstruction]: + """Keep only instructions that CHANGE the actuator's state. + + FRBC instructions are switch commands: an instruction that repeats + the previous instruction's (actuator, operation mode, factor) is a + no-op for the RM, which holds its state until told otherwise (the + flexmeasures-s2 plugin applies the same filter server-side). One + instruction per 15-min slot of a 24 h schedule is thus typically + 90%+ redundant traffic on the serial S2 link. + """ + filtered: list[FRBCInstruction] = [] + last_state: dict = {} + for instruction in instructions: + state = ( + str(instruction.operation_mode), + float(instruction.operation_mode_factor), + ) + actuator = str(instruction.actuator_id) + if last_state.get(actuator) == state: + continue + last_state[actuator] = state + filtered.append(instruction) + return filtered + + async def send_instruction_batch( + self, + instructions: list[FRBCInstruction], + supersession_key: str | None = None, + generation: int | None = None, + ) -> int: + """Replace the outstanding instruction batch with a new one. + + Applies the transition filter, optionally revokes the previous + batch (send_revocations), checks supersession one last time right + before queueing (a newer request may have arrived while the + schedule was being computed - queueing a stale batch would let the + RM file it under its CURRENT request), sends, and records the new + batch in the registry. Returns the number of instructions sent. + """ + filtered = self.filter_instruction_transitions(instructions) + if ( + supersession_key is not None + and generation is not None + and not self._is_current_generation(supersession_key, generation) + ): + self._logger.info( + f"Dropping stale instruction batch ({len(filtered)} instructions " + f"after transition-filtering {len(instructions)}): a newer " + f"'{supersession_key}' request has since arrived." + ) + return 0 + if self.send_revocations: + for message_id in list(self._sent_instructions): + status = self._instruction_statuses.get(message_id, "NEW") + if status in ("NEW", "ACCEPTED"): + await self.send_message( + RevokeObject( + message_id=get_unique_id(), + object_type=RevokableObjects.FRBC_Instruction, + object_id=self._sent_instructions[message_id].id, + ) + ) + self._sent_instructions = {} + self._instruction_statuses = {} + for instruction in filtered: + await self.send_message(instruction) + self._sent_instructions[str(instruction.message_id)] = instruction + self._logger.debug( + f"Sent instruction batch: {len(filtered)} instructions " + f"({len(instructions)} before transition-filtering)." + ) + return len(filtered) + + @register(InstructionStatusUpdate) + def handle_instruction_status_update( + self, message: InstructionStatusUpdate + ) -> pydantic.BaseModel: + """Track the RM's reported status of outstanding instructions, so a + batch replacement only revokes instructions that are still pending.""" + self._instruction_statuses[str(message.instruction_id)] = str( + getattr(message.status_type, "value", message.status_type) + ) + return get_reception_status(message, status=ReceptionStatusValues.OK) + @register(FRBCSystemDescription) def handle_system_description( self, message: FRBCSystemDescription ) -> pydantic.BaseModel: + # Content-hash dedupe (lifted from the TUNES handler): RMs commonly + # re-send their (static) system description with every request + # payload, under a fresh message_id each time. Re-processing an + # unchanged description would spawn a redundant schedule trigger + # per request on top of the storage status's trigger - half of the + # instruction-storm amplification observed in co-simulation. + message_dict = message.to_dict() + message_dict.pop("message_id") + system_description_hash = hash(str(message_dict)) + if getattr(self, "_last_system_description_hash", 0) == system_description_hash: + self._logger.debug( + "Ignoring re-sent system description (content unchanged)." + ) + return get_reception_status(message, status=ReceptionStatusValues.OK) + self._last_system_description_hash = system_description_hash + system_description_id = str(message.message_id) # store system_description message for later @@ -77,8 +224,116 @@ def handle_system_description( task ) # important to avoid a task disappearing mid-execution. task.add_done_callback(self.background_tasks.discard) + + # schedule send_conversion_efficiencies to run soon concurrently + task = asyncio.create_task(self.send_conversion_efficiencies(message)) + self.background_tasks.add(task) + task.add_done_callback(self.background_tasks.discard) + return get_reception_status(message, status=ReceptionStatusValues.OK) + def _get_operation_mode_efficiency_sensor_map( + self, system_description: FRBCSystemDescription + ) -> dict[str, int]: + """ + Get a mapping of operation mode IDs to efficiency sensor IDs. + + Subclasses can override this method to provide operation mode to efficiency + sensor mappings. Return an empty dict if there are no efficiency sensors. + + Args: + system_description: The system description containing operation mode details. + + Returns: + A dictionary mapping operation mode IDs to efficiency sensor IDs. + Empty dict ({}) if there are no efficiency sensors (default). + """ + return {} + + async def send_conversion_efficiencies( + self, system_description: FRBCSystemDescription + ): + """ + Send conversion efficiencies to FlexMeasures for operation modes. + + This method sends efficiency values for each operation mode that has + an associated efficiency sensor. Subclasses should override + _get_operation_mode_efficiency_sensor_map() to define which operation modes + have efficiency sensors. + + Args: + system_description: The system description containing actuator details. + """ + efficiency_map = self._get_operation_mode_efficiency_sensor_map( + system_description + ) + if not efficiency_map: + # No efficiency sensors defined + return + + try: + from datetime import timedelta + + start = system_description.valid_from + actuator = system_description.actuators[0] + + start_time = start.replace( + minute=(start.minute // 15) * 15, second=0, microsecond=0 + ) + + # Use a default conversion efficiency duration if not set by subclass + duration = getattr( + self, "_conversion_efficiency_duration", timedelta(hours=99) + ) + if isinstance(duration, str): + # If duration is a string like "PT99H", use default timedelta + duration = timedelta(hours=99) + + for operation_mode in actuator.operation_modes: + sensor_id = efficiency_map.get(operation_mode.id) + if sensor_id is None: + # Skip operation modes without an efficiency sensor + continue + + # Calculate efficiency from the last element (characteristic endpoint) + try: + fill_level_scale = getattr(self, "_fill_level_scale", 1.0) + efficiency = ( + 1 + * operation_mode.elements[-1].fill_rate.end_of_range + * fill_level_scale + / (operation_mode.elements[-1].power_ranges[0].end_of_range) + ) + self._logger.debug( + f"operation_mode.elements[-1].fill_rate.end_of_range: {operation_mode.elements[-1].fill_rate.end_of_range}" + ) + self._logger.debug( + f"operation_mode.elements[-1].power_ranges[0].end_of_range: {operation_mode.elements[-1].power_ranges[0].end_of_range}" + ) + self._logger.debug(f"fill_level_scale: {fill_level_scale}") + self._logger.debug(f"efficiency: {efficiency}") + except (IndexError, AttributeError, ZeroDivisionError) as e: + self._logger.debug( + f"Could not calculate efficiency for operation mode {operation_mode.id}: {e}" + ) + continue + + try: + await self._fm_client.post_sensor_data( + sensor_id=sensor_id, + start=start_time, + prior=self.now(), + values=[efficiency], + unit="dimensionless", + duration=duration, + ) + except Exception as e: + self._logger.debug( + f"Error posting efficiency data for sensor {sensor_id}: {e}" + ) + except Exception as e: + self._logger.debug(f"Error sending conversion efficiencies: {e}") + @register(FRBCUsageForecast) def handle_usage_forecast(self, message: FRBCUsageForecast) -> pydantic.BaseModel: message_id = str(message.message_id) @@ -98,7 +353,26 @@ def handle_storage_status(self, message: FRBCStorageStatus) -> pydantic.BaseMode self._storage_status_history[message_id] = message - task = asyncio.create_task(self.send_storage_status(message)) + # Latest-request-wins: each storage status is a (re)planning request; + # if several arrive while earlier ones still await the event loop + # (backlog under load, or an RM retry re-sending under a fresh + # message_id), only the newest may do the heavy work - the older + # requests are superseded and their tasks no-op. Subclasses whose + # send_storage_status runs a long chain should re-check with + # _is_current_generation before queueing instructions (see + # send_instruction_batch). + generation = self._bump_generation("storage_status") + + async def run_if_current(): + if not self._is_current_generation("storage_status", generation): + self._logger.info( + "Skipping superseded storage-status request " + f"(generation {generation})." + ) + return + await self.send_storage_status(message) + + task = asyncio.create_task(run_if_current()) self.background_tasks.add( task ) # important to avoid a task disappearing mid-execution. @@ -148,6 +422,11 @@ def handle_fill_level_target_profile( task.add_done_callback(self.background_tasks.discard) return get_reception_status(message, status=ReceptionStatusValues.OK) + @register(ReceptionStatus) + def handle_reception_status(self, message: ReceptionStatus): + self._logger.debug(message) + self._logger.debug(message.subject_message_id) + @register(FRBCTimerStatus) def handle_frbc_timer_status(self, message: FRBCTimerStatus) -> pydantic.BaseModel: return get_reception_status(message, status=ReceptionStatusValues.OK) @@ -196,4 +475,4 @@ async def trigger_schedule(self, system_description_id: str): ) # put instruction into the sending queue - await self._sending_queue.put(instruction) + await self.send_message(instruction) diff --git a/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py b/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py index eafdaa1b..06d0eb15 100644 --- a/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py +++ b/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py @@ -3,14 +3,22 @@ Used it at your own risk :) """ +import asyncio from datetime import datetime, timedelta from zoneinfo import ZoneInfo +import pandas as pd + +from flexmeasures_client.client import _server_version_at_least + try: from s2python.frbc import ( FRBCActuatorStatus, + FRBCFillLevelTargetProfile, + FRBCLeakageBehaviour, FRBCStorageStatus, FRBCSystemDescription, + FRBCUsageForecast, ) except ImportError: raise ImportError( @@ -21,18 +29,31 @@ from flexmeasures_client.s2.control_types.FRBC import FRBC from flexmeasures_client.s2.control_types.FRBC.utils import ( + clip_fill_level_target_profile, fm_schedule_to_instructions, get_soc_min_max, ) +from flexmeasures_client.s2.control_types.translations import ( + leakage_behaviour_to_storage_efficiency, + translate_fill_level_target_profile, + translate_usage_forecast_to_fm, +) class FRBCSimple(FRBC): _power_sensor_id: int _price_sensor_id: int + _production_price_sensor_id: int _soc_sensor_id: int _rm_discharge_sensor_id: int + _soc_minima_sensor_id: int + _soc_maxima_sensor_id: int + _usage_forecast_sensor_id: int + _leakage_behaviour_sensor_id: int + _charging_efficiency_sensor_id: int _schedule_duration: timedelta - _valid_from_shift: timedelta + _fill_level_scale: float + _resolution = "15min" def __init__( self, @@ -40,34 +61,91 @@ def __init__( soc_sensor_id: int, rm_discharge_sensor_id: int, price_sensor_id: int, + production_price_sensor_id: int, + soc_minima_sensor_id: int, + soc_maxima_sensor_id: int, + usage_forecast_sensor_id: int, + leakage_behaviour_sensor_id: int, + charging_efficiency_sensor_id: int, timezone: str = "UTC", - schedule_duration: timedelta = timedelta(hours=12), + schedule_duration: timedelta = timedelta(hours=24), max_size: int = 100, - valid_from_shift: timedelta = timedelta(days=1), + fill_level_scale: float = 1, + power_unit: str = "W", + energy_unit: str = "J", ) -> None: super().__init__(max_size) self._power_sensor_id = power_sensor_id self._price_sensor_id = price_sensor_id + self._production_price_sensor_id = production_price_sensor_id self._schedule_duration = schedule_duration self._soc_sensor_id = soc_sensor_id self._rm_discharge_sensor_id = rm_discharge_sensor_id + self._soc_minima_sensor_id = soc_minima_sensor_id + self._soc_maxima_sensor_id = soc_maxima_sensor_id + self._usage_forecast_sensor_id = usage_forecast_sensor_id + self._leakage_behaviour_sensor_id = leakage_behaviour_sensor_id + self.charging_efficiency_sensor_id = charging_efficiency_sensor_id self._timezone = ZoneInfo(timezone) - - # delay the start of the schedule from the time `valid_from` - # of the FRBC.SystemDescription. - self._valid_from_shift = valid_from_shift + self._fill_level_scale = fill_level_scale + self.power_unit = power_unit + self.energy_unit = energy_unit def now(self): return datetime.now(self._timezone) + def _get_operation_mode_efficiency_sensor_map(self, system_description) -> dict: + """ + Map operation mode IDs to the charging efficiency sensor. + + For FRBCSimple, all operation modes report their efficiency to the + single charging_efficiency_sensor_id. + """ + efficiency_map = {} + if self.charging_efficiency_sensor_id is None: + return efficiency_map + + # Map each operation mode to the charging efficiency sensor + actuator = system_description.actuators[0] + for operation_mode in actuator.operation_modes: + efficiency_map[operation_mode.id] = self.charging_efficiency_sensor_id + + return efficiency_map + async def send_storage_status(self, status: FRBCStorageStatus): - await self._fm_client.post_measurements( - self._soc_sensor_id, - start=self.now(), - values=[status.present_fill_level], - unit="MWh", - duration=timedelta(minutes=1), - ) + # HANGDEBUG: this runs as a fire-and-forget asyncio.create_task (see + # handle_storage_status), so any exception here is normally silently + # dropped (only surfacing as "Task exception was never retrieved" once + # the task is garbage collected). Log explicitly so nothing is missed. + now = self.now() + # Snapshot this request's generation: the schedule round trip below + # takes tens of seconds, during which a newer storage status (= a + # newer planning request) may arrive. The batch send re-checks this + # so a stale batch is dropped instead of being filed by the RM under + # its CURRENT request (see FRBC.send_instruction_batch). + generation = self._supersession_counters.get("storage_status", 0) + try: + self._logger.debug( + f"HANGDEBUG send_storage_status: posting SoC to sensor {self._soc_sensor_id}" + ) + await self._fm_client.post_sensor_data( + self._soc_sensor_id, + start=now, + prior=now, + values=[status.present_fill_level * self._fill_level_scale], + unit=self.energy_unit, + duration=timedelta(minutes=1), + ) + self._logger.debug( + "HANGDEBUG send_storage_status: SoC posted, calling trigger_schedule" + ) + await self.trigger_schedule(now, generation=generation) + self._logger.debug( + "HANGDEBUG send_storage_status: trigger_schedule returned" + ) + except Exception: + self._logger.exception("HANGDEBUG send_storage_status: raised") + raise async def send_actuator_status(self, status: FRBCActuatorStatus): factor = status.operation_mode_factor @@ -77,62 +155,455 @@ async def send_actuator_status(self, status: FRBCActuatorStatus): power = ( fill_rate.start_of_range + (fill_rate.end_of_range - fill_rate.start_of_range) * factor - ) + ) * self._fill_level_scale - dt = status.transition_timestamp # self.now() + start = status.transition_timestamp or self.now() - await self._fm_client.post_measurements( - self._rm_discharge_sensor_id, - start=dt, - values=[-power], - unit="MWh", + await self._fm_client.post_sensor_data( + self._power_sensor_id, + start=start, + prior=self.now(), + values=[power], + unit=self.power_unit, duration=timedelta(minutes=15), ) - async def trigger_schedule(self, system_description_id: str): - """Translates S2 System Description into FM API calls""" + async def trigger_schedule( + self, + start: datetime, + system_description_id: str | None = None, + generation: int | None = None, + ): + """ + Ask FlexMeasures for a new schedule and create FRBC.Instructions to send back to the ResourceManager + """ - system_description: FRBCSystemDescription = self._system_description_history[ - system_description_id - ] + # Barrier: realized-power posts run concurrently (see + # CEM.handle_power_measurement); the scheduler and the community + # compliance check it feeds must see the complete realized series + # before triggering, exactly as the old serial posting guaranteed. + cem = getattr(self, "_cem", None) + if cem is not None: + await cem.flush_measurement_posts() + + if system_description_id: + system_description: FRBCSystemDescription = ( + self._system_description_history[system_description_id] + ) + else: + # Use last SystemDescription + system_description: FRBCSystemDescription = list( + self._system_description_history.values() + )[-1] + system_descriptions = self._system_description_history.values() + self._logger.debug( + list( + [ + system_description.valid_from + for system_description in system_descriptions + ] + ) + ) + self._logger.debug(f"Using system description: {system_description}") if len(self._storage_status_history) > 0: - soc_at_start = list(self._storage_status_history.values())[ - -1 - ].present_fill_level + soc_at_start = ( + list(self._storage_status_history.values())[-1].present_fill_level + * self._fill_level_scale + ) else: print("Can't trigger schedule without knowing the status of the storage...") return - soc_min, soc_max = get_soc_min_max(system_description) + # Assume a single actuator + actuator = system_description.actuators[0] + + # Derive the overall power range. + # NB an S2 PowerRange runs from start_of_range (lower bound) to + # end_of_range (upper bound). The device can only produce if some + # range extends below zero; a consumption-only device (e.g. a heat + # pump) must get a zero production-capacity, or the scheduler will + # happily "discharge" its thermal storage as if it were an electric + # battery. + overall_min = None + overall_max = None + for operation_mode in actuator.operation_modes: + for element in operation_mode.elements: + for power_range in element.power_ranges: + # todo: distinguish power range per commodity + lo = power_range.start_of_range + hi = power_range.end_of_range + overall_min = lo if overall_min is None else min(overall_min, lo) + overall_max = hi if overall_max is None else max(overall_max, hi) + charging_capacity = max(overall_max, 0) + discharging_capacity = max(-min(overall_min, 0), 0) + + # Translate the S2 operation modes into FM power bands (the flex-model's + # "operation-modes" field): one band per operation-mode element power + # range, so FlexMeasures only schedules power values the device can + # actually run at (e.g. {0 W} U {883.7 W} for an on/off heater, instead + # of a fractional power that the RM would have to round to a full-on + # block, overshooting site capacity limits). Only sent when the bands + # actually restrict the power range (more than one distinct band). + # Which shape the power bands take depends on the server: FlexMeasures from + # v1.0.0 wants consumption and production stated separately and rejects the + # single signed field, while older servers want only that signed field. Asking + # the server rather than pinning it per branch keeps one client working against + # both. + split_power_ranges = _server_version_at_least( + await self._fm_client._resolve_server_version(), "1.0.0" + ) + operation_mode_bands: list[dict] = [] + for operation_mode in actuator.operation_modes: + for element in operation_mode.elements: + for power_range in element.power_ranges: + band_min, band_max = sorted( + (power_range.start_of_range, power_range.end_of_range) + ) + # S2 fixes one sign convention for power (positive means + # consumption), while FlexMeasures asks for the two + # directions separately, each non-negative. Split the S2 + # range by sign: its non-negative part is the consumption + # range, and its negative part becomes the production range + # with the sign flipped. A band spanning zero maps onto + # both, and each of those then starts at zero. + band: dict[str, list[str]] = {} + if not split_power_ranges: + band["power-range"] = [ + f"{band_min} {self.power_unit}", + f"{band_max} {self.power_unit}", + ] + elif band_max > 0: + band["consumption-range"] = [ + f"{max(band_min, 0)} {self.power_unit}", + f"{band_max} {self.power_unit}", + ] + if band_min < 0: + band["production-range"] = [ + f"{max(-band_max, 0)} {self.power_unit}", + f"{-band_min} {self.power_unit}", + ] + if not band: + # A band of exactly {0} on a v1 server: no power either way. + band["consumption-range"] = [ + f"0 {self.power_unit}", + f"0 {self.power_unit}", + ] + if band not in operation_mode_bands: + operation_mode_bands.append(band) + + # The RM's declared fill-level range is a comfort band, steered SOFTLY + # via the soc-minima/-maxima profiles (breach-priced server-side under + # relax-soc-constraints; see send_fill_level_target_profile, which + # clips those profiles into the declared range). The scalar + # soc-min/soc-max, by contrast, are HARD bounds server-side, and a + # hard bound at the declared range made physically-reachable states + # infeasible: free-float above "full", or a time-capped incumbent + # left below "empty" for the next day (a deterministically infeasible + # job). Declare wide safety rails instead; they only exist to keep + # the LP bounded (e.g. negative-price hours combined with a + # soc-value-at-end incentive). + _, range_top = get_soc_min_max(system_description, self._fill_level_scale) + soc_min = 0.0 + soc_max = range_top * 1.5 + + # Support for J energy unit (FM server scheduling trigger endpoint only accepts kWh and MWh) + if self.energy_unit == "J": + f = 3.6 * 10**6 + energy_unit = "kWh" + soc_at_start /= f + soc_min /= f + soc_max /= f + else: + energy_unit = self.energy_unit # call schedule - schedule = await self._fm_client.trigger_and_get_schedule( - start=system_description.valid_from - + self._valid_from_shift, # TODO: localize datetime - sensor_id=self._power_sensor_id, - flex_context={ - "production-price": {"sensor": self._price_sensor_id}, - "consumption-price": {"sensor": self._price_sensor_id}, - "site-power-capacity": f"{3 * 25 * 230} VA", - }, - flex_model={ - "soc-unit": "MWh", - "soc-at-start": soc_at_start, # TODO: use forecast of the SOC instead - "soc-min": soc_min, - "soc-max": soc_max, - }, - duration=self._schedule_duration, # next 12 hours - prior=self.now(), - # TODO: add SOC MAX AND SOC MIN FROM fill_level_range, - # this needs changes on the client + if isinstance(start, str): + start = pd.Timestamp(start) + flex_context = { + "consumption-price": {"sensor": self._price_sensor_id}, + "production-price": {"sensor": self._production_price_sensor_id}, + "site-power-capacity": f"{3 * 25 * 230} VA", + "relax-soc-constraints": True, + # Also relax site-level capacity specifically (not the broader + # relax-constraints/relax-capacity-constraints, which additionally soften + # *device*-level capacity constraints): this fills in a default + # site-consumption/production-breach-price so a site-level capacity + # constraint (e.g. from community steering) becomes a soft, penalized + # violation instead of causing infeasibility that silently falls back to a + # scheduler which ignores the constraint entirely. The broader flag was + # tried and reverted: with this apartment model's large SOC magnitudes + # (hundreds of kWh of thermal storage against a sub-1kW power capacity), + # the extra device-capacity breach-price terms it adds pushed HiGHS from a + # ~10s solve into one that ran 10+ minutes without converging (confirmed + # via py-spy: time was spent inside the solver itself, not a Python hang). + "relax-site-capacity-constraints": True, + } + # Last-resort rail: a start state outside the hard bounds would make + # the whole problem infeasible, so widen them to include it. + soc_min = min(soc_min, soc_at_start) + soc_max = max(soc_max, soc_at_start) + flex_model = { + "soc-unit": energy_unit, + "soc-at-start": soc_at_start, + "soc-min": soc_min, + "soc-max": soc_max, + "soc-minima": {"sensor": self._soc_minima_sensor_id}, + "soc-maxima": {"sensor": self._soc_maxima_sensor_id}, + "state-of-charge": {"sensor": self._soc_sensor_id}, + "soc-usage": [{"sensor": self._usage_forecast_sensor_id}], + "storage-efficiency": {"sensor": self._leakage_behaviour_sensor_id}, + "charging-efficiency": {"sensor": self.charging_efficiency_sensor_id}, + "consumption-capacity": f"{charging_capacity} {self.power_unit}", + "production-capacity": f"{discharging_capacity} {self.power_unit}", + } + if len(operation_mode_bands) > 1: + flex_model["operation-modes"] = operation_mode_bands + self._logger.debug(f"flex_context: {flex_context}") + self._logger.debug(f"flex_model: {flex_model}") + self._logger.debug( + f"HANGDEBUG trigger_schedule: about to call trigger_and_get_schedule " + f"(power_sensor_id={self._power_sensor_id})" + ) + # A scheduling job can fail transiently server-side (e.g. a pandas + # "cannot reindex on an axis with duplicate labels" error observed with + # overlapping recurring schedule windows) and succeed cleanly when re-run + # moments later. Without a retry here, that single failure permanently stalls + # this timestep: the RM never gets an instruction and polls forever, since + # this whole call runs as a fire-and-forget task whose exception is otherwise + # just logged and dropped. Retry a few times before giving up for real. + # "No recent state-of-charge value found" is the same class of transient: + # at startup, this trigger can race the controller's initial battery-SoC + # post (a separate client posting to the flex-model's SoC sensor), and + # FlexMeasures then rejects the trigger with a 422 that resolves itself + # once that post lands moments later. + max_attempts = 5 + transient_errors = ( + "Scheduling job failed", + "No recent state-of-charge value found", ) + # Deterministic server-side failures re-fail identically on every + # retry; re-triggering them only multiplies failed jobs across the + # worker pool (observed with a scheduler bug that violated the + # timed_belief primary key on every attempt). Never retry these, + # even though their message carries a transient-looking marker. + deterministic_errors = ( + "UniqueViolation", + "duplicate key value violates unique constraint", + ) + for attempt in range(1, max_attempts + 1): + try: + schedule = await self._fm_client.trigger_and_get_schedule( + start=start.replace( + minute=(start.minute // 15) * 15, second=0, microsecond=0 + ), + prior=start, + sensor_id=self._power_sensor_id, + flex_context=flex_context, + flex_model=flex_model, + duration=self._schedule_duration, # next 12 hours + # TODO: add SOC MAX AND SOC MIN FROM fill_level_range, + # this needs changes on the client + unit=self.power_unit, + ) + break + except ValueError as exc: + if ( + any(marker in str(exc) for marker in deterministic_errors) + or not any(marker in str(exc) for marker in transient_errors) + or attempt == max_attempts + ): + self._logger.exception( + "HANGDEBUG trigger_schedule: trigger_and_get_schedule raised" + ) + raise + self._logger.warning( + f"HANGDEBUG trigger_schedule: transient trigger failure " + f"(attempt {attempt}/{max_attempts}), retrying: {exc}" + ) + await asyncio.sleep(2.0) + except Exception: + self._logger.exception( + "HANGDEBUG trigger_schedule: trigger_and_get_schedule raised" + ) + raise + self._logger.debug(f"HANGDEBUG trigger_schedule: got schedule back: {schedule}") + + # The server silently substitutes its fallback scheduler's result when + # the real scheduling problem is infeasible (the GET follows the + # fallback job unless FLEXMEASURES_FALLBACK_REDIRECT is set). That + # schedule is a coarse charging policy, not an optimum - and the + # fallback saves NO state-of-charge stream, so any device relying on + # scheduler-saved SoC (e.g. a follow-the-schedule battery) loses its + # SoC trail and later windows fail on soc-at-start resolution + # (observed as a mid-run house starvation in co-simulation). Surface + # it as an ERROR so it is never mistaken for a healthy schedule. + # NB the fallback scheduler is slated for removal in FlexMeasures v1. + scheduler_name = str( + (schedule.get("scheduler_info") or {}).get("scheduler", "") + if isinstance(schedule, dict) + else "" + ) + if "fallback" in scheduler_name.lower(): + self._logger.error( + f"FlexMeasures used its fallback scheduler ({scheduler_name}) " + f"for the window starting {start.isoformat()}: the real " + "scheduling problem was infeasible. The returned schedule is a " + "coarse charging policy and no state-of-charge stream was " + "saved server-side. Investigate the infeasibility." + ) + + if generation is not None and not self._is_current_generation( + "storage_status", generation + ): + self._logger.info( + "HANGDEBUG trigger_schedule: superseded while awaiting the " + "schedule; not building instructions." + ) + return # translate FlexMeasures schedule into instructions. SOC -> Power -> PowerFactor - instructions = fm_schedule_to_instructions( - schedule, system_description, initial_fill_level=soc_at_start + try: + instructions = fm_schedule_to_instructions( + schedule, + system_description, + initial_fill_level=soc_at_start / self._fill_level_scale, + ) + except Exception: + self._logger.exception( + "HANGDEBUG trigger_schedule: fm_schedule_to_instructions raised" + ) + raise + self._logger.debug( + f"HANGDEBUG trigger_schedule: built {len(instructions)} instructions" ) - # put instructions to sending queue - for instruction in instructions: - await self._sending_queue.put(instruction) + # Replace the outstanding batch (transition-filtered; supersession + # re-checked right before queueing - see FRBC.send_instruction_batch). + sent = await self.send_instruction_batch( + instructions, + supersession_key="storage_status" if generation is not None else None, + generation=generation, + ) + self._logger.debug( + f"HANGDEBUG trigger_schedule: {sent} instructions queued for sending" + ) + + async def send_fill_level_target_profile( + self, fill_level_target_profile: FRBCFillLevelTargetProfile + ): + """ + Send FRBC.FillLevelTargetProfile to FlexMeasures. + + Args: + fill_level_target_profile (FRBCFillLevelTargetProfile): The fill level target profile to be translated and sent. + """ + # if not self._is_timer_due("fill_level_target_profile"): + # return + + soc_minima, soc_maxima = translate_fill_level_target_profile( + fill_level_target_profile=fill_level_target_profile, + resolution=self._resolution, + fill_level_scale=self._fill_level_scale, + ) + + # Clip the target profile into the RM's declared fill-level range: + # these profiles steer comfort as breach-priced SOFT constraints + # server-side, while the declared range is no longer sent as a hard + # soc-min/soc-max (see trigger_schedule), so an out-of-range target + # (e.g. a night setback dropping below the range bottom) would + # otherwise price the scheduler into states the RM declared out of + # range. + system_descriptions = list(self._system_description_history.values()) + if system_descriptions: + range_bottom, range_top = get_soc_min_max( + system_descriptions[-1], self._fill_level_scale + ) + soc_minima, soc_maxima, n_crossed = clip_fill_level_target_profile( + soc_minima, soc_maxima, range_bottom, range_top + ) + if n_crossed: + self._logger.warning( + "Fill-level target profile lies outside the declared " + f"fill-level range [{range_bottom}, {range_top}] on " + f"{n_crossed} of {len(soc_minima)} time steps; collapsed " + "minima and maxima to their midpoint there." + ) + + duration = str(pd.Timedelta(self._resolution) * len(soc_maxima)) + + # POST SOC Minima measurements to FlexMeasures + await self._fm_client.post_sensor_data( + sensor_id=self._soc_minima_sensor_id, + start=fill_level_target_profile.start_time, + prior=self.now(), + values=soc_minima.tolist(), + unit=self.energy_unit, + duration=duration, + ) + + # POST SOC Maxima measurements to FlexMeasures + await self._fm_client.post_sensor_data( + sensor_id=self._soc_maxima_sensor_id, + start=fill_level_target_profile.start_time, + prior=self.now(), + values=soc_maxima.tolist(), + unit=self.energy_unit, + duration=duration, + ) + + async def send_usage_forecast(self, usage_forecast: FRBCUsageForecast): + """ + Send FRBC.UsageForecast to FlexMeasures. + + Args: + usage_forecast (FRBCUsageForecast): The usage forecast to be translated and sent. + """ + # if not self._is_timer_due("usage_forecast"): + # return + + start_time = usage_forecast.start_time + + # flooring to previous 15min tick + start_time = start_time.replace( + minute=(start_time.minute // 15) * 15, second=0, microsecond=0 + ) + + usage_forecast = translate_usage_forecast_to_fm( + usage_forecast, + self._resolution, + strategy="mean", + fill_level_scale=self._fill_level_scale, + ) + + await self._fm_client.post_sensor_data( + sensor_id=self._usage_forecast_sensor_id, + start=start_time, + prior=self.now(), + values=usage_forecast.tolist(), + unit=self.power_unit, # e.g. [0, 100] MW/(15 min) # todo: or: f"{self.energy_unit}/s" to scale usage forecast e.g. [0, 100] %/s -> [0, 100] %/(15 min) + duration=str(pd.Timedelta(self._resolution) * len(usage_forecast)), + ) + + async def send_leakage_behaviour(self, leakage: FRBCLeakageBehaviour): + # if not self._is_timer_due("leakage_behaviour"): + # return + + start = leakage.valid_from or self.now() + start = start.replace(minute=(start.minute // 15) * 15, second=0, microsecond=0) + + storage_efficiency = leakage_behaviour_to_storage_efficiency( + message=leakage, + resolution=timedelta(minutes=15), + fill_level_scale=self._fill_level_scale, + ) + self._logger.debug(storage_efficiency) + + await self._fm_client.post_sensor_data( + self._leakage_behaviour_sensor_id, + start=start, + prior=self.now(), + values=[storage_efficiency], + unit="%", + duration=timedelta(hours=48), + ) diff --git a/src/flexmeasures_client/s2/control_types/FRBC/frbc_tunes.py b/src/flexmeasures_client/s2/control_types/FRBC/frbc_tunes.py index e2496182..9f8224a8 100644 --- a/src/flexmeasures_client/s2/control_types/FRBC/frbc_tunes.py +++ b/src/flexmeasures_client/s2/control_types/FRBC/frbc_tunes.py @@ -123,7 +123,6 @@ def __init__( schedule_duration: timedelta = timedelta(hours=12), max_size: int = 100, fill_level_scale: float = 0.1, - valid_from_shift: timedelta = timedelta(days=1), timers: dict[str, datetime] | None = None, datastore: dict | None = None, **kwargs, @@ -154,10 +153,6 @@ def __init__( self._production_price_sensor_id = production_price_sensor self._timezone = ZoneInfo(timezone) - - # delay the start of the schedule from the time `valid_from` of the FRBC.SystemDescription - self._valid_from_shift = valid_from_shift - self._fill_level_scale = fill_level_scale self._active_recurring_schedule = False @@ -215,7 +210,7 @@ async def send_storage_status(self, status: FRBCStorageStatus): subject_message_id=status.message_id, status=ReceptionStatusValues.PERMANENT_ERROR, ) - await self._sending_queue.put(response) + await self.send_message(response) await self.trigger_schedule() async def send_leakage_behaviour(self, leakage: FRBCLeakageBehaviour): @@ -235,7 +230,7 @@ async def send_leakage_behaviour(self, leakage: FRBCLeakageBehaviour): leakage_behaviour_to_storage_efficiency( message=leakage, resolution=timedelta(minutes=15), - fill_level_scale=self._fill_level_scalefill_level_scale, + fill_level_scale=self._fill_level_scale, ) ], unit=PERCENTAGE, @@ -246,7 +241,7 @@ async def send_leakage_behaviour(self, leakage: FRBCLeakageBehaviour): subject_message_id=leakage.message_id, status=ReceptionStatusValues.PERMANENT_ERROR, ) - await self._sending_queue.put(response) + await self.send_message(response) async def send_actuator_status(self, status: FRBCActuatorStatus): if not self._is_timer_due("actuator_status"): @@ -514,12 +509,12 @@ async def trigger_schedule(self): object_id=message_id, ) self._logger.debug(f"Sending revoke instruction for {message_id}") - await self._sending_queue.put(revoke_instruction) + await self.send_message(revoke_instruction) self._datastore["instructions"] = {} # Put the instruction in the sending queue for instruction in instructions: - await self._sending_queue.put(instruction) + await self.send_message(instruction) # Store instructions for instruction in instructions: diff --git a/src/flexmeasures_client/s2/control_types/FRBC/utils.py b/src/flexmeasures_client/s2/control_types/FRBC/utils.py index 04ce1056..e917cc6b 100644 --- a/src/flexmeasures_client/s2/control_types/FRBC/utils.py +++ b/src/flexmeasures_client/s2/control_types/FRBC/utils.py @@ -1,4 +1,6 @@ +import json import logging +import uuid from datetime import timedelta from math import isclose from typing import List @@ -9,6 +11,7 @@ try: from s2python.common import NumberRange from s2python.frbc import ( + FRBCActuatorDescription, FRBCInstruction, FRBCLeakageBehaviour, FRBCOperationMode, @@ -155,12 +158,18 @@ def fm_schedule_to_instructions( f"{len(system_description.actuators)} were provided" ) - operation_modes: list[FRBCOperationMode] = actuator.operation_modes + actuators: dict[uuid.UUID, FRBCActuatorDescription] = { + a.id: a for a in system_description.actuators + } + operation_modes: dict[uuid.UUID, FRBCOperationMode] = { + om.id: om for om in actuator.operation_modes + } fill_level = initial_fill_level deltaT = timedelta(minutes=15) / timedelta(hours=1) + previous_instruction = None for timestamp, row in schedule.iterrows(): power = row["schedule"] usage = row.get("usage_forecast", 0) @@ -169,7 +178,7 @@ def fm_schedule_to_instructions( # Convert from power to fill rate results = [ (om, *power_to_fill_rate_with_metrics(om, power, fill_level)) - for om in operation_modes + for om in operation_modes.values() ] # Step 1: minimize fill-level penalty (primary) @@ -221,10 +230,34 @@ def fm_schedule_to_instructions( execution_time=timestamp, abnormal_condition=False, ) + if previous_instruction and all( + getattr(previous_instruction, attr) == getattr(instruction, attr) + for attr in ( + "actuator_id", + "operation_mode", + "operation_mode_factor", + "abnormal_condition", + ) + ): + logger.info("Instruction removed, no changes to previous instruction") + continue logger.info( f"Instruction created: at {timestamp} set {actuator.diagnostic_label if isinstance(actuator.diagnostic_label, str) else actuator} to {best_operation_mode.diagnostic_label if isinstance(best_operation_mode.diagnostic_label, str) else best_operation_mode} with factor {operation_mode_factor}" ) + previous_instruction = instruction instructions.append(instruction) + logger.debug( + "Instructions JSON: %s", + json.dumps( + [ + serialize_instruction( + instr, actuators=actuators, operation_modes=operation_modes + ) + for instr in instructions + ], + indent=2, + ), + ) # Update fill level fill_level = compute_next_fill_level( @@ -260,6 +293,41 @@ def get_soc_min_max( return soc_min, soc_max +def clip_fill_level_target_profile( + soc_minima: pd.Series, + soc_maxima: pd.Series, + range_bottom: float, + range_top: float, +) -> tuple[pd.Series, pd.Series, int]: + """Clip a translated fill-level target profile into the declared fill-level range. + + The target profile steers comfort as breach-priced SOFT constraints + server-side, while the declared range itself is no longer declared as a + hard soc-min/soc-max (those became wide safety rails). A target outside + the declared range (e.g. a night setback dropping below the range bottom) + would therefore price the scheduler into states the RM declared out of + range, so: + + - minima are clipped up to the range bottom, + - maxima are clipped down to the range top, + - where the clipped bounds cross (i.e. the target band lies entirely + outside the declared range), both collapse to the midpoint of the + clipped pair, keeping minima <= maxima pointwise. + + Returns the clipped (minima, maxima) and the number of time steps on + which the bounds crossed (so the caller can log a warning). + """ + soc_minima = soc_minima.clip(lower=range_bottom) + soc_maxima = soc_maxima.clip(upper=range_top) + crossed = soc_maxima < soc_minima + n_crossed = int(crossed.sum()) + if n_crossed: + midpoint = (soc_minima + soc_maxima) / 2 + soc_minima = soc_minima.mask(crossed, midpoint) + soc_maxima = soc_maxima.mask(crossed, midpoint) + return soc_minima, soc_maxima, n_crossed + + def power_to_fill_rate_with_metrics( operation_mode: FRBCOperationMode, power: float, @@ -381,3 +449,29 @@ def explain_choice( lines.append(f"{label} (element={element_label}): rejected due to {reason}") return "; ".join(lines) + + +def serialize_instruction( + instr: FRBCInstruction, + actuators: dict[uuid.UUID, FRBCActuatorDescription], + operation_modes: dict[uuid.UUID, FRBCOperationMode], +): + """Create dict of instructions suitable for logging.""" + actuator = ( + getattr(actuators[instr.actuator_id], "diagnostic_label", None) + or instr.actuator_id + ) + operation_mode = ( + getattr(operation_modes[instr.operation_mode], "diagnostic_label", None) + or instr.operation_mode + ) + return { + "message_type": instr.message_type, + "message_id": str(instr.message_id), + "instruction_id": str(instr.id), + "actuator": str(actuator), + "operation_mode": str(operation_mode), + "operation_mode_factor": instr.operation_mode_factor, + "execution_time": instr.execution_time.isoformat(), + "abnormal_condition": instr.abnormal_condition, + } diff --git a/src/flexmeasures_client/s2/control_types/__init__.py b/src/flexmeasures_client/s2/control_types/__init__.py index b3e19f5b..c82f4e32 100644 --- a/src/flexmeasures_client/s2/control_types/__init__.py +++ b/src/flexmeasures_client/s2/control_types/__init__.py @@ -1,8 +1,7 @@ from __future__ import annotations -from asyncio import Queue from logging import Logger -from typing import cast +from typing import Any, Callable, cast from pydantic import BaseModel from s2python.common import ( @@ -22,8 +21,12 @@ class ControlTypeHandler(Handler): _instruction_history: SizeLimitOrderedDict[str, BaseModel] _instruction_status_history: SizeLimitOrderedDict[str, InstructionStatus] _fm_client: FlexMeasuresClient - _sending_queue: Queue + send_message: Callable _logger: Logger + #: Back-reference to the CEM that registered this handler, so a handler can + #: await the CEM's flush_measurement_posts() barrier before triggering a + #: schedule. Set by CEM.register_control_type(). + _cem: Any = None def __init__(self, max_size: int = 100) -> None: super().__init__(max_size) @@ -31,6 +34,17 @@ def __init__(self, max_size: int = 100) -> None: self._instruction_history = SizeLimitOrderedDict(max_size=max_size) self._instruction_status_history = SizeLimitOrderedDict(max_size=max_size) + async def close(self): + """Release any resources / stop recurring tasks for this handler. + + Default no-op so CEM.close() (which calls close() on every registered + handler when a websocket tears down) works for any control-type handler. + Subclasses that own recurring tasks override this. Previously FRBCSimple + had no close(), so a websocket teardown raised AttributeError inside + CEM.close(), killing the CEM's request handler and hanging the RM. + """ + self._logger.debug(f"Closing {self.__class__.__name__} handler") + @register(InstructionStatusUpdate) def handle_instruction_status_update(self, message: InstructionStatusUpdate): instruction_id: str = cast(str, message.instruction_id) diff --git a/src/flexmeasures_client/s2/script/websockets_server.py b/src/flexmeasures_client/s2/script/websockets_server.py index 6a11c948..addda7f4 100644 --- a/src/flexmeasures_client/s2/script/websockets_server.py +++ b/src/flexmeasures_client/s2/script/websockets_server.py @@ -10,7 +10,6 @@ from flexmeasures_client.client import FlexMeasuresClient from flexmeasures_client.s2.cem import CEM -from flexmeasures_client.s2.control_types.FRBC.frbc_simple import FRBCSimple log_level = os.getenv("LOGGING_LEVEL", "WARNING").upper() logging.basicConfig( @@ -29,31 +28,34 @@ async def rm_details_watchdog(ws, cem: CEM): """ # wait to get resource manager details - while cem._control_type is None: + while cem._control.control_type is None: await asyncio.sleep(1) await cem.activate_control_type(control_type=ControlType.FILL_RATE_BASED_CONTROL) # check/wait that the control type is set properly - while cem._control_type != ControlType.FILL_RATE_BASED_CONTROL: + while cem._control.control_type != ControlType.FILL_RATE_BASED_CONTROL: cem._logger.debug("waiting for the activation of the control type...") await asyncio.sleep(1) - cem._logger.debug(f"CONTROL TYPE: {cem._control_type}") + cem._logger.debug(f"CONTROL TYPE: {cem._control.control_type}") - # after this, schedule will be triggered on reception of a new system description + # after this, schedule will be triggered on reception of a new storage status async def websocket_producer(ws, cem: CEM): cem._logger.debug("start websocket message producer") cem._logger.debug(f"IS CLOSED? {cem.is_closed()}") while not cem.is_closed(): - message = await cem.get_message() - cem._logger.debug("sending message") + message, fut = await cem.get_message() try: + cem._logger.debug("sending message") await ws.send_json(message) - except aiohttp.ClientConnectionResetError: - break + fut.set_result(True) + # except aiohttp.ClientConnectionResetError: + # break + except Exception as exc: + fut.set_exception(exc) cem._logger.debug("cem closed") @@ -75,6 +77,7 @@ async def websocket_consumer(ws, cem: CEM): elif msg.type == aiohttp.WSMsgType.ERROR: cem._logger.debug("close...") await cem.close() + await ws.close() cem._logger.error(f"ws connection closed with exception {ws.exception()}") # TODO: save cem state? @@ -85,153 +88,46 @@ async def websocket_handler(request): ws = web.WebSocketResponse() await ws.prepare(request) - site_name = "My CEM" - base_url = os.getenv("FLEXMEASURES_BASE_URL", "http://localhost:5000") + base_url = os.getenv( + "FLEXMEASURES_BASE_URL", "http://localhost:5000" + ) # or "server:5000" parsed = urlparse(base_url) fm_client = FlexMeasuresClient( password=os.getenv("FLEXMEASURES_PASSWORD", "toy-password"), email=os.getenv("FLEXMEASURES_USER", "toy-user@flexmeasures.io"), host=parsed.netloc, ssl=parsed.scheme == "https", - ) - - price_sensor, power_sensor, soc_sensor, rm_discharge_sensor = await configure_site( - site_name, fm_client + polling_interval=0.5, ) cem = CEM( - sensor_id=power_sensor["id"], + power_sensor_id=None, # assign CEM a top-level asset directly fm_client=fm_client, logger=LOGGER, ) - frbc = FRBCSimple( - power_sensor_id=power_sensor["id"], - price_sensor_id=price_sensor["id"], - soc_sensor_id=soc_sensor["id"], - rm_discharge_sensor_id=rm_discharge_sensor["id"], - ) - cem.register_control_type(frbc) - - # create "parallel" tasks for the message producer and consumer - await asyncio.gather( - websocket_consumer(ws, cem), - websocket_producer(ws, cem), - rm_details_watchdog(ws, cem), - ) + # Create "parallel" tasks for the message producer and consumer. The + # consumer ends when the websocket closes (RMs reconnect at every replan + # boundary); the producer and watchdog must then be torn down along with + # the CEM itself, or the old session lives on as a zombie - its periodic + # loops keep posting SoC and triggering FlexMeasures schedules for an + # asset the RM no longer implements, racing the RM's new session (seen as + # a parallel steering-less schedule stream in the community co-simulation). + consumer = asyncio.create_task(websocket_consumer(ws, cem)) + side_tasks = [ + asyncio.create_task(websocket_producer(ws, cem)), + asyncio.create_task(rm_details_watchdog(ws, cem)), + ] + try: + await consumer + finally: + await cem.close() + for task in side_tasks: + task.cancel() + await asyncio.gather(*side_tasks, return_exceptions=True) return ws -async def configure_site( - site_name: str, fm_client: FlexMeasuresClient -) -> tuple[dict, dict, dict, dict]: - account = await fm_client.get_account() - assets = await fm_client.get_assets(parse_json_fields=True) - - site_asset = None - for asset in assets: - if asset["name"] == site_name: - site_asset = asset - break - - site_asset_specs = dict( - latitude=0, - longitude=0, - generic_asset_type_id=6, # Building asset type - flex_model={ - "power-capacity": f"{3 * 25 * 230} VA", - }, - ) - - if not site_asset: - site_asset = await fm_client.add_asset( - name=site_name, account_id=account["id"], **site_asset_specs - ) - # Update site asset with the latest specs - await fm_client.update_asset(site_asset["id"], site_asset_specs) - - sensors = site_asset.get("sensors", []) - price_sensor = None - power_sensor = None - soc_sensor = None - rm_discharge_sensor = None - for sensor in sensors: - if sensor["name"] == "price": - price_sensor = sensor - elif sensor["name"] == "power": - power_sensor = sensor - elif sensor["name"] == "state of charge": - soc_sensor = sensor - elif sensor["name"] == "RM discharge": - rm_discharge_sensor = sensor - - if price_sensor is None: - price_sensor = await fm_client.add_sensor( - name="price", - event_resolution="PT15M", - unit="EUR/kWh", - generic_asset_id=site_asset["id"], - timezone="Europe/Amsterdam", - ) - await fm_client.post_sensor_data( - sensor_id=price_sensor["id"], - start="2026-01-15T00:00+01", # 2026-01-01T00:00+01 - duration="P3D", # P1M - values=[ - 0.10, - 0.11, - 0.12, - 0.15, - 0.18, - 0.17, - 0.11, - 0.09, - 0.10, - 0.09, - 0.09, - 0.10, - 0.08, - 0.05, - 0.04, - 0.04, - 0.06, - 0.08, - 0.12, - 0.13, - 0.14, - 0.13, - 0.10, - 0.07, - ], - unit="EUR/kWh", - ) - if power_sensor is None: - power_sensor = await fm_client.add_sensor( - name="power", - event_resolution="PT15M", - unit="kW", - generic_asset_id=site_asset["id"], - timezone="Europe/Amsterdam", - ) - if soc_sensor is None: - soc_sensor = await fm_client.add_sensor( - name="state of charge", - event_resolution="PT0M", - unit="kWh", - generic_asset_id=site_asset["id"], - timezone="Europe/Amsterdam", - ) - if rm_discharge_sensor is None: - rm_discharge_sensor = await fm_client.add_sensor( - name="RM discharge", - event_resolution="PT15M", - unit="dimensionless", - generic_asset_id=site_asset["id"], - timezone="Europe/Amsterdam", - ) - return price_sensor, power_sensor, soc_sensor, rm_discharge_sensor - - app = web.Application() app.add_routes([web.get("/ws", websocket_handler)]) -web.run_app(app) +web.run_app(app, port=int(os.getenv("CEM_PORT", "8080"))) diff --git a/src/flexmeasures_client/s2/utils.py b/src/flexmeasures_client/s2/utils.py index 50eabaab..b7a2f502 100644 --- a/src/flexmeasures_client/s2/utils.py +++ b/src/flexmeasures_client/s2/utils.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections import OrderedDict +from dataclasses import dataclass, field from typing import Mapping, TypeVar from uuid import uuid4 @@ -9,7 +10,7 @@ from packaging.version import Version try: - from s2python.common import ReceptionStatus, ReceptionStatusValues + from s2python.common import ControlType, ReceptionStatus, ReceptionStatusValues except ImportError: raise ImportError( "The 's2-python' package is required for this functionality. " @@ -39,6 +40,12 @@ def __setitem__(self, __key: KT, __value: VT) -> None: return super().__setitem__(__key, __value) +@dataclass +class ControlContext: + control_type: ControlType | None = None + handler_ready: dict[ControlType, bool] = field(default_factory=dict) + + def get_unique_id() -> str: """Generate a random v4 UUID string. diff --git a/tests/client/test_sensor.py b/tests/client/test_sensor.py index 1fd39fc9..8c6e0499 100644 --- a/tests/client/test_sensor.py +++ b/tests/client/test_sensor.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import os import re from unittest.mock import AsyncMock, patch @@ -10,7 +11,10 @@ from aioresponses import aioresponses from flexmeasures_client.client import ContentTypeError, FlexMeasuresClient -from flexmeasures_client.exceptions import InsufficientServerVersionError +from flexmeasures_client.exceptions import ( + InsufficientServerVersionError, + JobFailedError, +) @pytest.mark.asyncio @@ -484,6 +488,13 @@ async def test_post_sensor_data_json_accepted_returns_ingestion_job() -> None: "status": "ACCEPTED", }, ) + # post_sensor_data awaits ingestion by default, so the job it reports + # must resolve before the post returns. + m.get( + "http://localhost:5000/api/v3_0/jobs/ingestion-job-id", + status=200, + payload={"status": "FINISHED", "message": "ok", "result": None}, + ) response, status = await client.post_sensor_data( sensor_id=5, @@ -739,11 +750,16 @@ async def test_post_sensor_data_with_file_accepted(): "http://localhost:5000/api/v3_0/sensors/1/data/upload", status=202, payload={ - "job_id": "test-job-id", + "job": "test-job-id", "message": "Sensor data has been accepted for processing.", "status": "ACCEPTED", }, ) + m.get( + "http://localhost:5000/api/v3_0/jobs/test-job-id", + status=200, + payload={"status": "FINISHED", "message": "ok", "result": None}, + ) response_data, status = await client.post_sensor_data( sensor_id=1, file_path=csv_path, @@ -909,3 +925,186 @@ async def test_get_sensor_data_content_type_error(): resolution="PT15M", ) await client.close() + + +# --- await_ingestion (async server-side ingestion, restoring read-your-writes) --- + + +@pytest.mark.asyncio +async def test_post_sensor_data_await_ingestion_finished(): + """202 + job_id, job later FINISHED: post_sensor_data polls and returns + normally, without raising.""" + with aioresponses() as m: + client = FlexMeasuresClient(email="test@test.test", password="test") + client.access_token = "test-token" + sensor_id = 5 + m.post( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/data", + status=202, + payload={ + "status": "ACCEPTED", + "message": "Sensor data has been accepted for processing.", + "job_monitor_url": "http://localhost:5000/api/v3_0/jobs/job-1", + "job": "job-1", + }, + ) + m.get( + "http://localhost:5000/api/v3_0/jobs/job-1", + status=200, + payload={ + "status": "FINISHED", + "message": "Sensor data ingestion job has finished.", + "result": None, + }, + ) + + await client.post_sensor_data( + sensor_id=sensor_id, + start="2023-01-01T00:00+00:00", + duration="PT1H", + values=[1.0], + unit="MW", + ) + await client.close() + + +@pytest.mark.asyncio +async def test_post_sensor_data_await_ingestion_failed(): + """202 + job_id, job later FAILED: post_sensor_data raises + JobFailedError.""" + with aioresponses() as m: + client = FlexMeasuresClient(email="test@test.test", password="test") + client.access_token = "test-token" + sensor_id = 5 + m.post( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/data", + status=202, + payload={ + "status": "ACCEPTED", + "message": "Sensor data has been accepted for processing.", + "job_monitor_url": "http://localhost:5000/api/v3_0/jobs/job-2", + "job": "job-2", + }, + ) + m.get( + "http://localhost:5000/api/v3_0/jobs/job-2", + status=200, + payload={ + "status": "FAILED", + "message": "Sensor data ingestion job failed with ValueError: boom", + "result": None, + }, + ) + + with pytest.raises(JobFailedError, match="job-2"): + await client.post_sensor_data( + sensor_id=sensor_id, + start="2023-01-01T00:00+00:00", + duration="PT1H", + values=[1.0], + unit="MW", + ) + await client.close() + + +@pytest.mark.asyncio +async def test_post_sensor_data_await_ingestion_timeout(caplog): + """202 + job_id, job stays QUEUED past the polling timeout: post_sensor_data + logs an ERROR and returns normally (does not raise).""" + with aioresponses() as m: + client = FlexMeasuresClient(email="test@test.test", password="test") + client.access_token = "test-token" + # Very small timeout, so the test does not wait out the default. + client.job_polling_timeout = 0.05 + client.job_polling_interval = 0.01 + sensor_id = 5 + m.post( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/data", + status=202, + payload={ + "status": "ACCEPTED", + "message": "Sensor data has been accepted for processing.", + "job_monitor_url": "http://localhost:5000/api/v3_0/jobs/job-3", + "job": "job-3", + }, + ) + m.get( + "http://localhost:5000/api/v3_0/jobs/job-3", + status=200, + payload={ + "status": "QUEUED", + "message": "Sensor data ingestion job waiting to be processed.", + "result": None, + }, + repeat=True, + ) + + with caplog.at_level(logging.ERROR): + await client.post_sensor_data( + sensor_id=sensor_id, + start="2023-01-01T00:00+00:00", + duration="PT1H", + values=[1.0], + unit="MW", + ) + assert any( + "Ingestion not confirmed" in record.message for record in caplog.records + ) + await client.close() + + +@pytest.mark.asyncio +async def test_post_sensor_data_200_no_polling(): + """A synchronous 200 OK response is not followed by any job-status + polling.""" + with aioresponses() as m: + client = FlexMeasuresClient(email="test@test.test", password="test") + client.access_token = "test-token" + sensor_id = 5 + m.post( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/data", + status=200, + payload={"status": "PROCESSED", "message": "ok"}, + ) + # No jobs/* endpoint mocked: if post_sensor_data attempted to poll, the + # unmocked GET request would raise inside aioresponses. + + await client.post_sensor_data( + sensor_id=sensor_id, + start="2023-01-01T00:00+00:00", + duration="PT1H", + values=[1.0], + unit="MW", + ) + await client.close() + + +@pytest.mark.asyncio +async def test_post_sensor_data_await_ingestion_false_no_polling(): + """await_ingestion=False opts out of polling, even for a 202 response.""" + with aioresponses() as m: + client = FlexMeasuresClient(email="test@test.test", password="test") + client.access_token = "test-token" + sensor_id = 5 + m.post( + f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/data", + status=202, + payload={ + "status": "ACCEPTED", + "message": "Sensor data has been accepted for processing.", + "job_monitor_url": "http://localhost:5000/api/v3_0/jobs/job-4", + "job": "job-4", + }, + ) + # No jobs/* endpoint mocked: if post_sensor_data attempted to poll despite + # await_ingestion=False, the unmocked GET request would raise. + + await client.post_sensor_data( + sensor_id=sensor_id, + start="2023-01-01T00:00+00:00", + duration="PT1H", + values=[1.0], + unit="MW", + await_ingestion=False, + ) + await client.close() diff --git a/tests/s2/test_cem.py b/tests/s2/test_cem.py index 37f3aaff..c8748278 100644 --- a/tests/s2/test_cem.py +++ b/tests/s2/test_cem.py @@ -1,5 +1,8 @@ from __future__ import annotations +import asyncio +from unittest.mock import AsyncMock, MagicMock + import pytest from s2python.common import ControlType, ReceptionStatus, ReceptionStatusValues @@ -26,8 +29,8 @@ async def test_handshake(rm_handshake): ) # check that two messages are put to the outgoing queue (ReceptionStatus and HandshakeResponse) # CEM response - response = await cem.get_message() # ReceptionStatus for Handshake - response = await cem.get_message() # HandshakeResponse + response, _ = await cem.get_message() # ReceptionStatus for Handshake + response, _ = await cem.get_message() # HandshakeResponse assert ( response["message_type"] == "HandshakeResponse" @@ -54,8 +57,8 @@ async def test_resource_manager_details(resource_manager_details, rm_handshake): cem._sending_queue.qsize() == 2 ) # check that message is put to the outgoing queue - response = await cem.get_message() # ReceptionStatus for Handshake - response = await cem.get_message() # HandshakeResponse + response, _ = await cem.get_message() # ReceptionStatus for Handshake + response, _ = await cem.get_message() # HandshakeResponse ########################## # ResourceManagerDetails # @@ -63,7 +66,7 @@ async def test_resource_manager_details(resource_manager_details, rm_handshake): # RM sends ResourceManagerDetails await cem.handle_message(resource_manager_details) - response = await cem.get_message() + response, _ = await cem.get_message() # CEM response is ReceptionStatus with an OK status assert response["message_type"] == "ReceptionStatus" @@ -77,6 +80,66 @@ async def test_resource_manager_details(resource_manager_details, rm_handshake): "independently of the original type" ) + # Cleanup: cancel any pending background tasks + for task_id, task in cem._handler_build_tasks.items(): + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +async def test_frbc_message_implicitly_activates_frbc( + frbc_system_description, resource_manager_details, rm_handshake +): + """A lost SelectControlType confirmation must not wedge the CEM. + + Per the S2 protocol, an RM only sends control-type-specific messages + (FRBC.*) AFTER it has accepted the CEM's SelectControlType - so an + incoming FRBC message is itself proof of activation. Without implicit + activation, a dropped/mis-routed ReceptionStatus left the CEM in + NO_SELECTION forever, answering every FRBC.SystemDescription with + TEMPORARY_ERROR while the RM retried indefinitely (in-vivo simulation + hang, 2026-07-26).""" + cem = CEM(fm_client=None) + frbc = FRBCTest() + cem.register_control_type(frbc) + + await cem.handle_message(rm_handshake) + await cem.get_message() # ReceptionStatus for Handshake + await cem.get_message() # HandshakeResponse + await cem.handle_message(resource_manager_details) + await cem.get_message() # ReceptionStatus for ResourceManagerDetails + + # The CEM requests FRBC, but the RM's OK confirmation never arrives. + await cem.activate_control_type(ControlType.FILL_RATE_BASED_CONTROL) + await cem.get_message() # the SelectControlType going out to the RM + assert cem.control_type == ControlType.NO_SELECTION + + # The RM - which did activate FRBC on its side - sends its system + # description anyway. + await cem.handle_message(frbc_system_description) + response, _ = await cem.get_message() + + assert ( + cem.control_type == ControlType.FILL_RATE_BASED_CONTROL + ), "an incoming FRBC message implies the RM accepted FRBC" + assert response["message_type"] == "ReceptionStatus" + assert response["status"] == "OK", ( + "the system description must be accepted, not bounced with" " TEMPORARY_ERROR" + ) + + # Cleanup: cancel any pending background tasks + for task in list(cem._handler_build_tasks.values()) + list(frbc.background_tasks): + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + @pytest.mark.asyncio async def test_activate_control_type( @@ -92,14 +155,14 @@ async def test_activate_control_type( ############# await cem.handle_message(rm_handshake) - response = await cem.get_message() # ReceptionStatus for Handshake - response = await cem.get_message() # HandshakeResponse + response, _ = await cem.get_message() # ReceptionStatus for Handshake + response, _ = await cem.get_message() # HandshakeResponse ########################## # ResourceManagerDetails # ########################## await cem.handle_message(resource_manager_details) - response = await cem.get_message() + response, _ = await cem.get_message() ######################### # Activate control type # @@ -107,7 +170,7 @@ async def test_activate_control_type( # CEM sends a request to change te control type await cem.activate_control_type(ControlType.FILL_RATE_BASED_CONTROL) - message = await cem.get_message() + message, _ = await cem.get_message() assert cem.control_type == ControlType.NO_SELECTION, ( "the control type should still be NO_SELECTION (rather than FRBC)," @@ -124,6 +187,15 @@ async def test_activate_control_type( cem.control_type == ControlType.FILL_RATE_BASED_CONTROL ), "after a positive ResponseStatus, the status changes from NO_SELECTION to FRBC" + # Cleanup: cancel any pending background tasks + for task_id, task in cem._handler_build_tasks.items(): + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + @pytest.mark.asyncio async def test_messages_route_to_control_type_handler( @@ -139,21 +211,21 @@ async def test_messages_route_to_control_type_handler( ############# await cem.handle_message(rm_handshake) - response = await cem.get_message() # ReceptionStatus for Handshake - response = await cem.get_message() # HandshakeResponse + response, _ = await cem.get_message() # ReceptionStatus for Handshake + response, _ = await cem.get_message() # HandshakeResponse ########################## # ResourceManagerDetails # ########################## await cem.handle_message(resource_manager_details) - response = await cem.get_message() + response, _ = await cem.get_message() ######################### # Activate control type # ######################### await cem.activate_control_type(ControlType.FILL_RATE_BASED_CONTROL) - message = await cem.get_message() + message, _ = await cem.get_message() response = ReceptionStatus( subject_message_id=message.get("message_id"), status=ReceptionStatusValues.OK @@ -166,7 +238,7 @@ async def test_messages_route_to_control_type_handler( ######## await cem.handle_message(frbc_system_description) - response = await cem.get_message() + response, _ = await cem.get_message() # checking that FRBC handler is being called assert ( @@ -182,9 +254,9 @@ async def test_messages_route_to_control_type_handler( # change of control type is not performed in case that the RM answers # with a negative response await cem.activate_control_type(ControlType.NO_SELECTION) - response = await cem.get_message() + response, _ = await cem.get_message() assert ( - cem._control_type == ControlType.FILL_RATE_BASED_CONTROL + cem._control.control_type == ControlType.FILL_RATE_BASED_CONTROL ), "control type should not change, confirmation still pending" await cem.handle_message( @@ -195,7 +267,7 @@ async def test_messages_route_to_control_type_handler( ) assert ( - cem._control_type == ControlType.FILL_RATE_BASED_CONTROL + cem._control.control_type == ControlType.FILL_RATE_BASED_CONTROL ), "control type should not change, confirmation state is not 'OK'" assert ( response.get("message_id") @@ -204,6 +276,15 @@ async def test_messages_route_to_control_type_handler( ].success_callbacks ), "success callback should be deleted" + # Cleanup: cancel any pending background tasks + for task_id, task in cem._handler_build_tasks.items(): + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + @pytest.mark.asyncio async def test_automatic_change_control_type(resource_manager_details, rm_handshake): @@ -222,8 +303,8 @@ async def test_automatic_change_control_type(resource_manager_details, rm_handsh cem._sending_queue.qsize() == 2 ) # check that message is put to the outgoing queue - response = await cem.get_message() - response = await cem.get_message() # HandshakeResponse + response, _ = await cem.get_message() + response, _ = await cem.get_message() # HandshakeResponse ########################## # ResourceManagerDetails # @@ -231,14 +312,83 @@ async def test_automatic_change_control_type(resource_manager_details, rm_handsh # RM sends ResourceManagerDetails await cem.handle_message(resource_manager_details) - response = await cem.get_message() + response, _ = await cem.get_message() # CEM sends control type on receiving the ResourceManagerDetails assert response["message_type"] == "SelectControlType" assert response["control_type"] == "FILL_RATE_BASED_CONTROL" - response = await cem.get_message() + response, _ = await cem.get_message() # CEM response is ReceptionStatus with an OK status assert response["message_type"] == "ReceptionStatus" assert response["status"] == "OK" + + # Cleanup: cancel any pending background tasks + for task_id, task in cem._handler_build_tasks.items(): + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio +async def test_handle_message_during_handler_registration_race(): + cem = CEM( + fm_client=MagicMock(), + logger=MagicMock(), + ) + + # --- Fake FRBC handler that we will "register late" + frbc_handler = AsyncMock() + frbc_handler._control_type = ControlType.FILL_RATE_BASED_CONTROL + frbc_handler.supports_message.return_value = True + frbc_handler.handle_message.return_value = {"ok": True} + + # Simulate slow map_resource_to_asset + registration_started = asyncio.Event() + registration_continue = asyncio.Event() + + async def slow_map_resource_to_asset(message): + registration_started.set() + await registration_continue.wait() + + cem.register_control_type(frbc_handler) + + cem.map_resource_to_asset = slow_map_resource_to_asset + + # Set control type BEFORE handler exists + cem.update_control_type(ControlType.FILL_RATE_BASED_CONTROL) + + # Start async registration + task = asyncio.create_task( + cem.map_resource_to_asset(MagicMock(resource_id="x", name="test")) + ) + + # Wait until registration has started but not finished + await registration_started.wait() + + # --- THIS is the race moment + msg = { + "message_type": "TestMessage", + "message_id": "550e8400-e29b-41d4-a716-446655440000", + } + + # Should NOT crash even though handler isn't registered yet + await cem.handle_message(msg) + + # A response should be queued (ReceptionStatus with TEMPORARY_ERROR since handler not ready) + response, _ = await cem.get_message() + assert response.get("message_type") == "ReceptionStatus" + assert response.get("status") == "TEMPORARY_ERROR" + + # Now finish registration + registration_continue.set() + await task + + # Now handler should have been called the second time + await cem.handle_message(msg) + + assert frbc_handler.handle_message.called diff --git a/tests/s2/test_frbc_registry.py b/tests/s2/test_frbc_registry.py new file mode 100644 index 00000000..d1f82c61 --- /dev/null +++ b/tests/s2/test_frbc_registry.py @@ -0,0 +1,204 @@ +"""Tests for the FRBC outstanding-work registry (instruction-storm fix): + +- transition filtering of instruction batches, +- latest-request-wins supersession of storage-status handler work, +- content-hash dedupe of re-sent system descriptions, +- the sent-instruction registry, status updates and opt-in revocation. +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import datetime, timezone + +import pytest +from s2python.common import ( + InstructionStatus, + InstructionStatusUpdate, + ReceptionStatusValues, +) +from s2python.frbc import FRBCInstruction, FRBCStorageStatus, FRBCSystemDescription + +from flexmeasures_client.s2.control_types.FRBC import FRBC, FRBCTest +from flexmeasures_client.s2.utils import get_unique_id + + +def make_instruction( + mode: str, factor: float, slot: int, actuator: str +) -> FRBCInstruction: + return FRBCInstruction( + message_id=get_unique_id(), + id=get_unique_id(), + actuator_id=actuator, + operation_mode=mode, + operation_mode_factor=factor, + execution_time=datetime( + 2022, 12, 1, slot // 4, (slot % 4) * 15, tzinfo=timezone.utc + ), + abnormal_condition=False, + ) + + +class TestTransitionFilter: + def test_repeats_are_dropped_and_changes_kept(self): + on, off = get_unique_id(), get_unique_id() + actuator = get_unique_id() + modes = [on, on, on, off, off, on, on, on] + instructions = [ + make_instruction(m, 1.0, i, actuator) for i, m in enumerate(modes) + ] + filtered = FRBC.filter_instruction_transitions(instructions) + assert [str(i.operation_mode) for i in filtered] == [str(on), str(off), str(on)] + # The first instruction always survives (it establishes the state). + assert filtered[0] is instructions[0] + + def test_factor_change_is_a_transition(self): + mode = get_unique_id() + actuator = get_unique_id() + instructions = [ + make_instruction(mode, f, i, actuator) + for i, f in enumerate([0.5, 0.5, 1.0, 1.0]) + ] + filtered = FRBC.filter_instruction_transitions(instructions) + assert [i.operation_mode_factor for i in filtered] == [0.5, 1.0] + + def test_actuators_are_filtered_independently(self): + mode = get_unique_id() + a1, a2 = get_unique_id(), get_unique_id() + instructions = [ + make_instruction(mode, 1.0, 0, a1), + make_instruction(mode, 1.0, 0, a2), # other actuator: kept + make_instruction(mode, 1.0, 1, a1), # repeat for a1: dropped + ] + filtered = FRBC.filter_instruction_transitions(instructions) + assert len(filtered) == 2 + assert {str(i.actuator_id) for i in filtered} == {str(a1), str(a2)} + + +class _RecordingFRBC(FRBCTest): + """FRBCTest with a recording send_storage_status chain that mimics the + Simple handler's long round trip (SoC post + schedule wait).""" + + def __init__(self, delay_s: float = 0.05, **kwargs): + super().__init__(**kwargs) + self._logger = logging.getLogger("test") + self.completed_generations: list[int] = [] + self.delay_s = delay_s + self.sent = [] + + async def send_message(self, message): # capture instead of queueing + self.sent.append(message) + + async def send_storage_status(self, status: FRBCStorageStatus): + generation = self._supersession_counters.get("storage_status", 0) + await asyncio.sleep(self.delay_s) # the FM round trip + batch = [ + make_instruction(get_unique_id(), 1.0, i, get_unique_id()) for i in range(3) + ] + sent = await self.send_instruction_batch( + batch, supersession_key="storage_status", generation=generation + ) + if sent: + self.completed_generations.append(generation) + + +def make_status(fill_level: float = 0.5) -> FRBCStorageStatus: + return FRBCStorageStatus(message_id=get_unique_id(), present_fill_level=fill_level) + + +@pytest.mark.asyncio +async def test_only_the_latest_storage_status_produces_a_batch(): + """Three rapid statuses (a backlog, or an RM retry storm re-sending under + fresh message ids): only the newest request's batch may reach the RM.""" + frbc = _RecordingFRBC() + + for _ in range(3): + await frbc.handle_message(make_status()) + await asyncio.gather(*frbc.background_tasks) + + assert frbc.completed_generations == [ + 3 + ], "only generation 3 (the newest request) may complete a batch" + # Registry holds exactly the surviving batch. + assert len(frbc._sent_instructions) == 3 + + +@pytest.mark.asyncio +async def test_resent_system_description_is_acked_but_not_reprocessed( + frbc_system_description, +): + frbc = _RecordingFRBC() + + response_1 = await frbc.handle_message(frbc_system_description) + await asyncio.gather(*frbc.background_tasks) + history_size = len(frbc._system_description_history) + + resent = FRBCSystemDescription( + message_id=get_unique_id(), # fresh id, identical content + valid_from=frbc_system_description.valid_from, + actuators=frbc_system_description.actuators, + storage=frbc_system_description.storage, + ) + response_2 = await frbc.handle_message(resent) + await asyncio.gather(*frbc.background_tasks) + + assert str(response_1.status) == str(ReceptionStatusValues.OK) + assert str(response_2.status) == str( + ReceptionStatusValues.OK + ), "the RM's resend must still be acknowledged" + assert ( + len(frbc._system_description_history) == history_size + ), "unchanged content must not be stored or reprocessed" + + +@pytest.mark.asyncio +async def test_batch_replacement_revokes_only_pending_instructions(): + frbc = _RecordingFRBC() + frbc.send_revocations = True + + mode, actuator = get_unique_id(), get_unique_id() + first = [make_instruction(mode, float(i % 2), i, actuator) for i in range(4)] + await frbc.send_instruction_batch(first) + first_sent = list(frbc._sent_instructions.values()) + assert len(first_sent) == 4 # alternating factors: all are transitions + + # The RM reports one instruction finished; it must not be revoked. + finished = first_sent[0] + await frbc.handle_message( + InstructionStatusUpdate( + message_id=get_unique_id(), + instruction_id=finished.message_id, + status_type=InstructionStatus.SUCCEEDED, + timestamp=datetime(2022, 12, 1, 12, 0, tzinfo=timezone.utc), + ) + ) + + frbc.sent.clear() + second = [make_instruction(get_unique_id(), 1.0, 0, actuator)] + await frbc.send_instruction_batch(second) + + revokes = [m for m in frbc.sent if m.__class__.__name__ == "RevokeObject"] + assert len(revokes) == 3, "only NEW/ACCEPTED instructions are revoked" + revoked_ids = {str(r.object_id) for r in revokes} + assert str(finished.id) not in revoked_ids + # Registry now holds only the new batch. + assert len(frbc._sent_instructions) == 1 + + +@pytest.mark.asyncio +async def test_stale_batch_is_dropped_even_after_the_schedule_returned(): + """The supersession re-check at queueing time: a batch computed for an + older request is dropped when a newer request arrived meanwhile.""" + frbc = _RecordingFRBC() + + generation = frbc._bump_generation("storage_status") + frbc._bump_generation("storage_status") # a newer request arrives + + batch = [make_instruction(get_unique_id(), 1.0, 0, get_unique_id())] + sent = await frbc.send_instruction_batch( + batch, supersession_key="storage_status", generation=generation + ) + assert sent == 0 + assert frbc.sent == [] + assert frbc._sent_instructions == {} diff --git a/tests/s2/test_frbc_soft_range.py b/tests/s2/test_frbc_soft_range.py new file mode 100644 index 00000000..05408bd6 --- /dev/null +++ b/tests/s2/test_frbc_soft_range.py @@ -0,0 +1,195 @@ +"""The declared FRBC fill-level range is a SOFT comfort band. + +The CEM steers it via the soc-minima/-maxima profiles (breach-priced +server-side), clipped into the declared range, while the hard scalar +soc-min/soc-max in the flex model are wide safety rails only. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock + +import pandas as pd +import pytest +from s2python.common import Duration, NumberRange +from s2python.frbc import ( + FRBCFillLevelTargetProfile, + FRBCFillLevelTargetProfileElement, + FRBCStorageStatus, +) + +from flexmeasures_client.s2.control_types.FRBC.frbc_simple import FRBCSimple +from flexmeasures_client.s2.control_types.FRBC.utils import ( + clip_fill_level_target_profile, +) +from flexmeasures_client.s2.utils import get_unique_id + +RANGE_BOTTOM = 278.0 +RANGE_TOP = 355.0 + + +def series(values): + index = pd.date_range( + "2024-01-01T00:00:00+00:00", periods=len(values), freq="15min" + ) + return pd.Series(values, index=index, dtype=float) + + +def test_clip_is_a_noop_inside_the_range(): + minima, maxima, n_crossed = clip_fill_level_target_profile( + series([300, 310]), series([320, 330]), RANGE_BOTTOM, RANGE_TOP + ) + assert minima.tolist() == [300, 310] + assert maxima.tolist() == [320, 330] + assert n_crossed == 0 + + +def test_night_setback_minima_clip_up_to_the_range_bottom(): + # Night setback drops the target floor below the declared range bottom; + # the posted minima must clip UP to the bottom (maxima untouched). + minima, maxima, n_crossed = clip_fill_level_target_profile( + series([250, 300]), series([320, 330]), RANGE_BOTTOM, RANGE_TOP + ) + assert minima.tolist() == [RANGE_BOTTOM, 300] + assert maxima.tolist() == [320, 330] + assert n_crossed == 0 + + +def test_maxima_clip_down_to_the_range_top(): + minima, maxima, n_crossed = clip_fill_level_target_profile( + series([300]), series([400]), RANGE_BOTTOM, RANGE_TOP + ) + assert minima.tolist() == [300] + assert maxima.tolist() == [RANGE_TOP] + assert n_crossed == 0 + + +def test_target_band_entirely_outside_the_range_collapses_to_midpoint(): + # Target band entirely below the range bottom: clipped bounds cross + # (minima -> bottom, maxima stays below it), so both collapse to the + # midpoint of the clipped pair, keeping minima <= maxima pointwise. + minima, maxima, n_crossed = clip_fill_level_target_profile( + series([250, 300]), series([260, 330]), RANGE_BOTTOM, RANGE_TOP + ) + expected_midpoint = (RANGE_BOTTOM + 260) / 2 + assert minima.tolist() == [expected_midpoint, 300] + assert maxima.tolist() == [expected_midpoint, 330] + assert (minima <= maxima).all() + assert n_crossed == 1 + + +def make_frbc(fill_level_scale: float = 1.0) -> FRBCSimple: + frbc = FRBCSimple( + power_sensor_id=1, + soc_sensor_id=2, + rm_discharge_sensor_id=3, + price_sensor_id=4, + production_price_sensor_id=5, + soc_minima_sensor_id=6, + soc_maxima_sensor_id=7, + usage_forecast_sensor_id=8, + leakage_behaviour_sensor_id=9, + charging_efficiency_sensor_id=10, + fill_level_scale=fill_level_scale, + energy_unit="kWh", + ) + # normally attached by CEM.register_control_type + frbc._logger = logging.getLogger("test_frbc_soft_range") + return frbc + + +@pytest.mark.asyncio +async def test_flex_model_declares_wide_safety_rails(frbc_system_description): + """soc-min/soc-max are wide rails (0 and 1.5x the declared range top), + not the declared fill-level range.""" + frbc = make_frbc() + sd = frbc_system_description + frbc._system_description_history[str(sd.message_id)] = sd + status = FRBCStorageStatus(message_id=get_unique_id(), present_fill_level=0.5) + frbc._storage_status_history[str(status.message_id)] = status + + frbc._fm_client = AsyncMock() + frbc._fm_client.trigger_and_get_schedule = AsyncMock( + side_effect=RuntimeError("stop after capturing the flex model") + ) + with pytest.raises(RuntimeError, match="stop after capturing"): + await frbc.trigger_schedule(datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc)) + + kwargs = frbc._fm_client.trigger_and_get_schedule.call_args.kwargs + flex_model = kwargs["flex_model"] + range_top = sd.storage.fill_level_range.end_of_range + assert flex_model["soc-min"] == 0.0 + assert flex_model["soc-max"] == pytest.approx(range_top * 1.5) + # comfort steering stays on the (clipped) profile sensors + assert flex_model["soc-minima"] == {"sensor": 6} + assert flex_model["soc-maxima"] == {"sensor": 7} + + +@pytest.mark.asyncio +async def test_safety_rails_still_widen_to_include_the_start_state( + frbc_system_description, +): + """A start state above the upper rail widens the rail (last resort + against infeasibility).""" + frbc = make_frbc() + sd = frbc_system_description + frbc._system_description_history[str(sd.message_id)] = sd + range_top = sd.storage.fill_level_range.end_of_range + status = FRBCStorageStatus( + message_id=get_unique_id(), present_fill_level=range_top * 2 + ) + frbc._storage_status_history[str(status.message_id)] = status + + frbc._fm_client = AsyncMock() + frbc._fm_client.trigger_and_get_schedule = AsyncMock( + side_effect=RuntimeError("stop after capturing the flex model") + ) + with pytest.raises(RuntimeError, match="stop after capturing"): + await frbc.trigger_schedule(datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc)) + + flex_model = frbc._fm_client.trigger_and_get_schedule.call_args.kwargs["flex_model"] + assert flex_model["soc-max"] == pytest.approx(range_top * 2) + + +@pytest.mark.asyncio +async def test_fill_level_target_profile_is_posted_clipped(frbc_system_description): + """The posted soc-minima/-maxima series are clipped into the declared + fill-level range (night setback clips up to the range bottom).""" + frbc = make_frbc(fill_level_scale=RANGE_TOP) # declared range: 0..1 -> 0..355 + sd = frbc_system_description + frbc._system_description_history[str(sd.message_id)] = sd + range_bottom = sd.storage.fill_level_range.start_of_range * RANGE_TOP # 0.0 + range_top = sd.storage.fill_level_range.end_of_range * RANGE_TOP # 355.0 + + profile = FRBCFillLevelTargetProfile( + message_id=get_unique_id(), + start_time=datetime(2024, 1, 1, 0, 0, tzinfo=timezone.utc), + elements=[ + # night setback: target floor below the declared range bottom + FRBCFillLevelTargetProfileElement( + duration=Duration.from_timedelta(timedelta(minutes=15)), + fill_level_range=NumberRange(start_of_range=-0.5, end_of_range=0.8), + ), + # daytime: target ceiling above the declared range top + FRBCFillLevelTargetProfileElement( + duration=Duration.from_timedelta(timedelta(minutes=15)), + fill_level_range=NumberRange(start_of_range=0.8, end_of_range=1.2), + ), + ], + ) + + frbc._fm_client = AsyncMock() + await frbc.send_fill_level_target_profile(profile) + + posts = { + call.kwargs["sensor_id"]: call.kwargs["values"] + for call in frbc._fm_client.post_sensor_data.call_args_list + } + minima, maxima = posts[6], posts[7] + assert minima[0] == pytest.approx(range_bottom) # clipped UP to the bottom + assert maxima[0] == pytest.approx(0.8 * RANGE_TOP) # untouched + assert minima[1] == pytest.approx(0.8 * RANGE_TOP) # untouched + assert maxima[1] == pytest.approx(range_top) # clipped DOWN to the top + assert all(lo <= hi for lo, hi in zip(minima, maxima)) diff --git a/tests/s2/test_frbc_tunes.py b/tests/s2/test_frbc_tunes.py index 645692c9..0efb5832 100644 --- a/tests/s2/test_frbc_tunes.py +++ b/tests/s2/test_frbc_tunes.py @@ -78,21 +78,21 @@ async def setup_cem(resource_manager_details, rm_handshake): ############# await cem.handle_message(rm_handshake) - response = await cem.get_message() # ReceptionStatus for Handshake - response = await cem.get_message() # HandshakeResponse + response, _ = await cem.get_message() # ReceptionStatus for Handshake + response, _ = await cem.get_message() # HandshakeResponse ########################## # ResourceManagerDetails # ########################## await cem.handle_message(resource_manager_details) - response = await cem.get_message() + response, _ = await cem.get_message() ######################### # Activate control type # ######################### await cem.activate_control_type(ControlType.FILL_RATE_BASED_CONTROL) - message = await cem.get_message() + message, _ = await cem.get_message() response = ReceptionStatus( subject_message_id=message.get("message_id"), status=ReceptionStatusValues.OK @@ -112,7 +112,7 @@ async def cem_in_frbc_control_type(setup_cem, frbc_system_description): ######## await cem.handle_message(frbc_system_description) - await cem.get_message() + _, _ = await cem.get_message() return cem, fm_client, frbc_system_description diff --git a/tests/s2/test_timezone.py b/tests/s2/test_timezone.py index 59ab27b6..11039fcf 100644 --- a/tests/s2/test_timezone.py +++ b/tests/s2/test_timezone.py @@ -16,6 +16,12 @@ def test_frbc_simple_now_uses_zoneinfo_timezone(): soc_sensor_id=2, rm_discharge_sensor_id=3, price_sensor_id=4, + production_price_sensor_id=5, + soc_minima_sensor_id=6, + soc_maxima_sensor_id=7, + usage_forecast_sensor_id=8, + leakage_behaviour_sensor_id=9, + charging_efficiency_sensor_id=10, timezone="Europe/Amsterdam", )