Fix: CEM work session

docker-compose.override.yml

@@ -0,0 +1,75 @@
+# ------------------------------------------------------------------
+# 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
+# ------------------------------------------------------------------
+
+services:
+  dev-db:
+    ports:
+      - "5433:5432"
+  queue-db:
+    ports:
+      - "6380:6379"
+  server:
+    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
+    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
+        flexmeasures db upgrade
+        if ! flexmeasures show accounts | grep -q "Docker Toy Account"; then
+            flexmeasures add toy-account --name 'Docker Toy Account'
+        fi
+        gunicorn --bind 0.0.0.0:5000 --worker-tmp-dir /dev/shm --workers 2 --threads 4 wsgi:application
+  worker:
+    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
+    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
+        flexmeasures jobs run-worker --name flexmeasures-worker --queue forecasting\|scheduling
+  cem:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    depends_on:
+      - server
+    restart: always
+    ports:
+      - "8080:8080"   # aiohttp default; adjust if you change it
+    environment:
+      # Optional, but useful if you later make this configurable
+      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"
+    volumes:
+      # If flexmeasures_client lives in your repo and you want live edits
+      - ../flexmeasures-client:/app/flexmeasures-client:rw
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+    - |
+      # pip install --break-system-packages --no-cache-dir "git+https://github.com/FlexMeasures/flexmeasures-client.git@main#egg=flexmeasures-client[s2]"
+      pip install --break-system-packages -e /app/flexmeasures-client[s2]
+      pip install --break-system-packages aiohttp
+      pip install --break-system-packages pytz
+      pip install --break-system-packages s2-python==0.8.2
+      python3 /app/flexmeasures-client/src/flexmeasures_client/s2/script/websockets_server.py

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 <email> --username <username>
+
 Disclaimer
 ==========
 

src/flexmeasures_client/client.py

@@ -61,6 +61,8 @@ def _parse_asset_json_fields(asset: dict) -> None:
     _parse_json_field(asset, "attributes")
     _parse_json_field(asset, "flex_context")
     _parse_json_field(asset, "flex_model")
+    _parse_json_field(asset, "sensors_to_show")
+    _parse_json_field(asset, "kpi_sensors_to_show")
 
 
 def _parse_sensor_json_fields(sensor: dict) -> None:
@@ -619,6 +621,7 @@ async def get_schedule(
         sensor_id: int,
         schedule_id: str,
         duration: str | timedelta | None = None,
+        unit: str | None = None,
     ) -> dict:
         """Get schedule with given ID.
 
@@ -630,12 +633,14 @@ async def get_schedule(
                       'unit': 'MW'
                   }
         """
+        params = {}
         if duration is not None:
-            params = {
-                "duration": pd.Timedelta(duration).isoformat(),  # for example: PT1H
-            }
-        else:
-            params = {}
+            params["duration"] = pd.Timedelta(duration).isoformat()  # for example: PT1H
+        if unit is not None:
+            message = "get_schedule(): The 'unit' parameter requires FlexMeasures server version 0.31.0 or above. "
+            f"This parameter will be ignored by the server, which is at version {self.server_version}."
+            await self.ensure_minimum_server_version("0.31.0.dev90", message)
+            params["unit"] = unit
         schedule, status = await self.request(
             uri=f"sensors/{sensor_id}/schedules/{schedule_id}",
             method="GET",
@@ -770,7 +775,7 @@ async def get_assets(
             ) < Version("0.31.0"):
                 self.logger.warning(
                     "get_assets(): The 'root', 'depth' and 'fields' parameters require FlexMeasures server version 0.31.0 or above. "
-                    "These parameters will be ignored."
+                    f"These parameters will be ignored for server version {self.server_version}."
                 )
             if root and isinstance(root, int):
                 uri += f"&root={root}"
@@ -855,6 +860,7 @@ async def trigger_and_get_schedule(
         asset_id: int | None = None,
         prior: datetime | None = None,
         scheduler: str | None = None,
+        unit: str | None = None,
     ) -> dict | list[dict]:
         """Trigger a schedule and then fetch it.
 
@@ -890,7 +896,10 @@ async def trigger_and_get_schedule(
         if sensor_id is not None:
             # Get the schedule for a single device
             return await self.get_schedule(
-                sensor_id=sensor_id, schedule_id=schedule_id, duration=duration
+                sensor_id=sensor_id,
+                schedule_id=schedule_id,
+                duration=duration,
+                unit=unit,
             )
         elif flex_model is None:
             # If there is no flex-model referencing power sensors, no power schedules are retrieved
@@ -1140,7 +1149,7 @@ async def update_asset(self, asset_id: int, updates: dict) -> dict:
                 ) < Version("0.31.0"):
                     self.logger.warning(
                         "update_asset(): The 'aggregate-power' flex-context field requires FlexMeasures server version 0.31.0 or above. "
-                        "The 'aggregate-power' field will be ignored by the server."
+                        f"The 'aggregate-power' field will be ignored by the server, which is at version {self.server_version}."
                     )
             updates["flex_context"] = json.dumps(updates["flex_context"])
         if "flex_model" in updates:
@@ -1253,6 +1262,7 @@ async def trigger_schedule(
 
         if prior is not None:
             message["prior"] = pd.Timestamp(prior).isoformat()
+            message["force_new_job_creation"] = True
         if scheduler is not None:
             if asset_id is None:
                 raise ValueError(

src/flexmeasures_client/s2/cem.py

@@ -3,7 +3,7 @@
 import json
 import logging
 import math
-from asyncio import Queue
+import asyncio
 from collections import defaultdict
 from datetime import datetime, timedelta
 from logging import Logger
@@ -58,7 +58,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
@@ -84,7 +84,7 @@ def __init__(
         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
         self._control_types_handlers = dict()
@@ -158,8 +158,8 @@ def register_control_type(self, control_type_handler: ControlTypeHandler):
         # 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
+        # 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
@@ -184,7 +184,19 @@ 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}")
 
         # try to handle the message with the control_type handle
         if (
@@ -215,7 +227,7 @@ 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):
         """
@@ -224,15 +236,24 @@ def update_control_type(self, control_type: ControlType):
         """
         self._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
@@ -272,8 +293,7 @@ async def activate_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
@@ -418,8 +438,11 @@ 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))
+        await fut  # wait until actually sent
 
 
 def get_commodity_unit(commodity_quantity) -> str:

src/flexmeasures_client/s2/control_types/FRBC/__init__.py

@@ -196,4 +196,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)