Report removed instances as a Removal

AGENTS.md

@@ -50,6 +50,9 @@ work on a branch.
 
 - **Public API**: everything users need is exported from
   `innergrid/__init__.py` and listed in `__all__`.
+- **Topics**: the prefix is the slug of the customer site. The platform's
+  topics are retained; an empty message means the instance is gone, which
+  `parse_message` returns as a `Removal`.
 - **Connection**: `listen()` connects once and raises
   `InnergridConnectionError` (or `InnergridAuthError`) when the
   connection fails or is lost; reconnecting is up to the caller.

README.md

@@ -35,6 +35,10 @@ async def main() -> None:
 asyncio.run(main())
 ```
 
+The callback of `listen()` gets an `Update` with the data of a topic, or a
+`Removal` when the instance of a topic is gone (the platform then publishes an
+empty message on its topic).
+
 `listen()` and `wait_for_data()` raise `InnergridAuthError` when the
 broker rejects the login and `InnergridConnectionError` when the connection
 fails or is lost. `wait_for_data()` raises `InnergridTimeoutError` when no
@@ -55,10 +59,12 @@ data arrives in time.
 | `<prefix>/events/chargepoints/<id>/<event>` | `ChargePointEvent` |
 | `milesight/uplink/<devEUI>` | `MilesightSensor` |
 
-Every update has a key (`<kind>/<id>`, for example `meter/10005`); the
-latest update of a key replaces the earlier ones. The topics are not
-retained, so the data arrives as it is published, every 10 seconds to 3
-minutes.
+`<prefix>` is the slug of the customer site. Every update and removal has a
+key (`<kind>/<id>`, for example `meter/10005`); the latest update of a key
+replaces the earlier ones. The platform's topics are retained, so the current
+state of every topic arrives right after subscribing; after that, a topic is
+published when it changes and every refresh interval (60 seconds by default).
+The Milesight sensors are not retained and publish every few minutes.
 
 `parse_data(kind, id, raw)` parses a payload that was stored earlier, for
 example `update.raw`, into an `Update` again.

innergrid/__init__.py

@@ -9,6 +9,7 @@
 )
 from .innergrid import (
     Innergrid,
+    Removal,
     Update,
     parse_data,
     parse_message,
@@ -51,6 +52,7 @@
     "MilesightSensor",
     "Model",
     "Quantity",
+    "Removal",
     "Site",
     "SiteMeter",
     "Space",

innergrid/innergrid.py

@@ -90,6 +90,23 @@ def key(self) -> str:
         return f"{self.kind}/{self.id}"
 
 
+@dataclass(frozen=True)
+class Removal:
+    """The instance of a topic is gone, for example a removed charge point.
+
+    The platform publishes an empty retained message on its topic, which
+    also clears the topic on the broker.
+    """
+
+    kind: Kind
+    id: str
+
+    @property
+    def key(self) -> str:
+        """Return the key of the topic."""
+        return f"{self.kind}/{self.id}"
+
+
 def valid_topic_prefix(prefix: str) -> bool:
     """Return whether a topic prefix is valid: not empty, without wildcards."""
     prefix = prefix.strip().strip("/")
@@ -109,23 +126,27 @@ def topic_kind(prefix: str, topic: str) -> tuple[Kind, str] | None:
     return None
 
 
-def parse_message(prefix: str, topic: str, payload: Any) -> Update | None:
+def parse_message(prefix: str, topic: str, payload: Any) -> Update | Removal | None:
     """Parse a message, or return None if it is not used or not valid.
 
     Args:
     ----
-        prefix: The topic prefix, for example mysite.
+        prefix: The topic prefix: the slug of the customer site, for example
+            mysite.
         topic: The topic the message was published on.
-        payload: The payload of the message, a JSON object.
+        payload: The payload of the message, a JSON object. An empty payload
+            means that the instance of the topic is gone.
 
     Returns:
     -------
-        The parsed data, or None.
+        The parsed data, a removal, or None.
 
     """
     if (kind_id := topic_kind(prefix, topic)) is None:
         return None
     kind, idx = kind_id
+    if payload in (b"", "", None):
+        return Removal(kind=kind, id=idx)
     try:
         data = json.loads(payload)
     except (TypeError, ValueError):
@@ -217,7 +238,7 @@ def _connection_error(self, err: aiomqtt.MqttError) -> InnergridConnectionError:
 
     async def listen(
         self,
-        callback: Callable[[Update], None],
+        callback: Callable[[Update | Removal], None],
         on_connect: Callable[[], None] | None = None,
     ) -> None:
         """Connect and call the callback with the data of every message.
@@ -226,7 +247,8 @@ async def listen(
 
         Args:
         ----
-            callback: Called with the data of every message.
+            callback: Called with the data of every message, or a removal when
+                the instance of a topic is gone.
             on_connect: Called after connecting and subscribing.
 
         Raises:
@@ -281,7 +303,8 @@ async def wait_for_data(self, wait_seconds: float = 30.0) -> Update:
                         update = parse_message(
                             self.topic_prefix, message.topic.value, message.payload
                         )
-                        if update is not None:
+                        # A removal is no data; the next topic may have some.
+                        if isinstance(update, Update):
                             return update
         except aiomqtt.MqttError as err:
             raise self._connection_error(err) from err

tests/test_innergrid.py

@@ -19,6 +19,7 @@
     InnergridTimeoutError,
     Kind,
     MilesightSensor,
+    Removal,
     SystemStatus,
     Update,
     parse_data,
@@ -117,7 +118,7 @@ def test_parse_message() -> None:
         "mysite/electricitymeters/10005",
         load_fixture("meter_single_phase.json").encode(),
     )
-    assert update is not None
+    assert isinstance(update, Update)
     assert update.key == "meter/10005"
     assert isinstance(update.data, ElectricityMeter)
     assert update.raw["id"] == 10005
@@ -131,7 +132,20 @@ def test_parse_data() -> None:
     assert update.data.temperature == 21.3  # type: ignore[union-attr]
 
 
-@pytest.mark.parametrize("payload", [b"not json", b"[1, 2]", None])
+@pytest.mark.parametrize("payload", [b"", "", None])
+def test_parse_removal(payload: Any) -> None:
+    """Test that an empty message means that the instance is gone."""
+    removal = parse_message("mysite", "mysite/chargepoints/19463", payload)
+    assert removal == Removal(kind=Kind.CHARGE_POINT, id="19463")
+    assert removal.key == "chargepoint/19463"
+
+
+def test_parse_removal_of_unused_topic() -> None:
+    """Test that an empty message on a topic that is not used is ignored."""
+    assert parse_message("mysite", "mysite/unknown", b"") is None
+
+
+@pytest.mark.parametrize("payload", [b"not json", b"[1, 2]"])
 def test_parse_invalid_message(payload: Any) -> None:
     """Test that a payload that is not a JSON object is ignored."""
     assert parse_message("mysite", "mysite/system", payload) is None
@@ -178,9 +192,10 @@ async def test_listen() -> None:
             ("mysite/system", load_fixture("system.json")),
             ("mysite/unknown", "{}"),
             ("milesight/uplink/a81758fffe054f0d", load_fixture("milesight.json")),
+            ("mysite/spaces/1/live", b""),
         ]
     )
-    updates: list[Update] = []
+    updates: list[Update | Removal] = []
     connected: list[bool] = []
     with (
         patch("aiomqtt.Client", fake),
@@ -191,10 +206,11 @@ async def test_listen() -> None:
         )
     assert fake.subscriptions == ["mysite/#", "milesight/uplink/#"]
     assert connected == [True]
-    assert [type(update.data) for update in updates] == [
+    assert [type(u.data) for u in updates[:2] if isinstance(u, Update)] == [
         SystemStatus,
         MilesightSensor,
     ]
+    assert updates[2] == Removal(kind=Kind.SPACE, id="1")
 
 
 async def test_listen_without_on_connect() -> None:
@@ -239,9 +255,13 @@ async def test_listen_other_code() -> None:
 
 
 async def test_wait_for_data() -> None:
-    """Test waiting for the first message with data."""
+    """Test waiting for the first message with data, skipping removals."""
     fake = FakeClient(
-        [("mysite/unknown", "{}"), ("mysite/system", load_fixture("system.json"))]
+        [
+            ("mysite/unknown", "{}"),
+            ("mysite/chargepoints/1", b""),
+            ("mysite/system", load_fixture("system.json")),
+        ]
     )
     with patch("aiomqtt.Client", fake):
         update = await Innergrid("broker", topic_prefix="mysite").wait_for_data()