From faa1062191ace0325ffb80f2fcda2602cc7e9b27 Mon Sep 17 00:00:00 2001 From: Robbin Janssen Date: Thu, 5 Nov 2026 10:01:55 +0100 Subject: [PATCH] Report removed instances as a Removal The platform publishes an empty retained message on the topic of an instance that is gone. parse_message returns a Removal for it, and listen() passes it to the callback. Document that the topics are retained and that the prefix is the slug of the customer site. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 3 +++ README.md | 14 ++++++++++---- innergrid/__init__.py | 2 ++ innergrid/innergrid.py | 37 ++++++++++++++++++++++++++++++------- tests/test_innergrid.py | 32 ++++++++++++++++++++++++++------ 5 files changed, 71 insertions(+), 17 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2a6b08d..6f65bf8 100644 --- a/AGENTS.md +++ b/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. diff --git a/README.md b/README.md index 07c7307..328d105 100644 --- a/README.md +++ b/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. | `/events/chargepoints//` | `ChargePointEvent` | | `milesight/uplink/` | `MilesightSensor` | -Every update has a key (`/`, 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. +`` is the slug of the customer site. Every update and removal has a +key (`/`, 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. diff --git a/innergrid/__init__.py b/innergrid/__init__.py index bc8ee1e..709734c 100644 --- a/innergrid/__init__.py +++ b/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", diff --git a/innergrid/innergrid.py b/innergrid/innergrid.py index c025d9a..6612e0c 100644 --- a/innergrid/innergrid.py +++ b/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 diff --git a/tests/test_innergrid.py b/tests/test_innergrid.py index 789304c..9b25175 100644 --- a/tests/test_innergrid.py +++ b/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()