chore: address PR feedback

Author: allenporter

roborock/devices/rpc/v1_channel.py

@@ -308,9 +308,8 @@ async def subscribe(self, callback: Callable[[RoborockMessage], None]) -> Callab
             loop = asyncio.get_running_loop()
             self._reconnect_task = loop.create_task(self._background_reconnect())
 
-        # Always attempt to subscribe to MQTT to receive protocol updates (data points)
-        # even if we have a local connection. Protocol updates only come via cloud/MQTT.
-        # Local connection is used for RPC commands, but push notifications come via MQTT.
+        # We maintain an active MQTT subscription even when connected locally to receive
+        # unsolicited status updates (DPS push messages) directly from the cloud.
         try:
             self._mqtt_unsub = await self._mqtt_channel.subscribe(self._on_mqtt_message)
         except RoborockException as err:
@@ -342,7 +341,7 @@ def add_dps_listener(self, listener: Callable[[dict[RoborockDataProtocol, Any]],
         This will attach a listener to the existing subscription, invoking
         the listener whenever new DPS values arrive from the subscription.
         This will only work if a subscription has already been setup, which is
-        handled by the device setup.
+        handled by the device start.
         """
         return self._dps_listeners.add_callback(listener)
 
@@ -447,10 +446,16 @@ def _on_mqtt_message(self, message: RoborockMessage) -> None:
         if self._callback:
             self._callback(message)
         try:
-            if datapoints := decode_data_protocol_message(message):
-                self._dps_listeners(datapoints)
+            datapoints = decode_data_protocol_message(message)
         except RoborockException as e:
             self._logger.debug("Error decoding data protocol message: %s", e)
+            return
+
+        if datapoints:
+            try:
+                self._dps_listeners(datapoints)
+            except Exception:
+                self._logger.exception("Error in DPS listener callback")
 
     def _on_local_message(self, message: RoborockMessage) -> None:
         """Handle incoming local messages."""

tests/protocols/test_v1_protocol.py

@@ -320,9 +320,7 @@ def test_decode_data_protocol_message() -> None:
         payload=payload,
     )
     result = decode_data_protocol_message(message)
-    assert result is not None
-    assert result[RoborockDataProtocol.STATE] == 8
-    assert result[RoborockDataProtocol.BATTERY] == 95
+    assert result == {RoborockDataProtocol.STATE: 8, RoborockDataProtocol.BATTERY: 95}
 
 
 def test_decode_data_protocol_message_all_status_fields() -> None:
@@ -338,13 +336,14 @@ def test_decode_data_protocol_message_all_status_fields() -> None:
         payload=payload,
     )
     result = decode_data_protocol_message(message)
-    assert result is not None
-    assert result[RoborockDataProtocol.ERROR_CODE] == 0
-    assert result[RoborockDataProtocol.STATE] == 5
-    assert result[RoborockDataProtocol.BATTERY] == 100
-    assert result[RoborockDataProtocol.FAN_POWER] == 102
-    assert result[RoborockDataProtocol.WATER_BOX_MODE] == 204
-    assert result[RoborockDataProtocol.CHARGE_STATUS] == 1
+    assert result == {
+        RoborockDataProtocol.ERROR_CODE: 0,
+        RoborockDataProtocol.STATE: 5,
+        RoborockDataProtocol.BATTERY: 100,
+        RoborockDataProtocol.FAN_POWER: 102,
+        RoborockDataProtocol.WATER_BOX_MODE: 204,
+        RoborockDataProtocol.CHARGE_STATUS: 1,
+    }
 
 
 def test_decode_data_protocol_message_unknown_codes() -> None:
@@ -355,9 +354,7 @@ def test_decode_data_protocol_message_unknown_codes() -> None:
         payload=payload,
     )
     result = decode_data_protocol_message(message)
-    assert result is not None
-    assert len(result) == 1
-    assert result[RoborockDataProtocol.STATE] == 8
+    assert result == {RoborockDataProtocol.STATE: 8}
 
 
 def test_decode_data_protocol_message_empty_payload() -> None: