feat: Add a device property to determine local connection status (#547)

Author: allenporter

roborock/devices/channel.py

@@ -22,6 +22,11 @@ def is_connected(self) -> bool:
         """Return true if the channel is connected."""
         ...
 
+    @property
+    def is_local_connected(self) -> bool:
+        """Return true if the channel is connected locally."""
+        ...
+
     async def subscribe(self, callback: Callable[[RoborockMessage], None]) -> Callable[[], None]:
         """Subscribe to messages from the device."""
         ...

roborock/devices/device.py

@@ -87,6 +87,16 @@ def is_connected(self) -> bool:
         """Return whether the device is connected."""
         return self._channel.is_connected
 
+    @property
+    def is_local_connected(self) -> bool:
+        """Return whether the device is connected locally.
+
+        This can be used to determine if the device is reachable over a local
+        network connection, as opposed to a cloud connection. This is useful
+        for adjusting behavior like polling frequency.
+        """
+        return self._channel.is_local_connected
+
     async def connect(self) -> None:
         """Connect to the device using the appropriate protocol channel."""
         if self._unsub:

roborock/devices/local_channel.py

@@ -56,6 +56,11 @@ def is_connected(self) -> bool:
         """Check if the channel is currently connected."""
         return self._is_connected
 
+    @property
+    def is_local_connected(self) -> bool:
+        """Check if the channel is currently connected locally."""
+        return self._is_connected
+
     async def connect(self) -> None:
         """Connect to the device."""
         if self._is_connected:

roborock/devices/mqtt_channel.py

@@ -40,6 +40,11 @@ def is_connected(self) -> bool:
         """
         return self._mqtt_session.connected
 
+    @property
+    def is_local_connected(self) -> bool:
+        """Return true if the channel is connected locally."""
+        return False
+
     @property
     def _publish_topic(self) -> str:
         """Topic to send commands to the device."""

tests/devices/test_mqtt_channel.py

@@ -134,6 +134,18 @@ async def test_publish_success(
     assert decoded_message.protocol == RoborockMessageProtocol.RPC_REQUEST
 
 
+@pytest.mark.parametrize(("connected"), [(True), (False)])
+async def test_connection_status(
+    mqtt_session: Mock,
+    mqtt_channel: MqttChannel,
+    connected: bool,
+) -> None:
+    """Test successful RPC command sending and response handling."""
+    mqtt_session.connected = connected
+    assert mqtt_channel.is_connected is connected
+    assert mqtt_channel.is_local_connected is False
+
+
 async def test_message_decode_error(
     mqtt_channel: MqttChannel,
     mqtt_message_handler: Callable[[bytes], None],

tests/devices/test_v1_device.py

@@ -80,6 +80,28 @@ async def test_device_connection(device: RoborockDevice, channel: AsyncMock) ->
     assert unsub.called
 
 
+@pytest.mark.parametrize(
+    ("connected", "local_connected"),
+    [
+        (True, False),
+        (False, False),
+        (True, True),
+        (False, True),
+    ],
+)
+async def test_connection_status(
+    device: RoborockDevice,
+    channel: AsyncMock,
+    connected: bool,
+    local_connected: bool,
+) -> None:
+    """Test successful RPC command sending and response handling."""
+    channel.is_connected = connected
+    channel.is_local_connected = local_connected
+    assert device.is_connected is connected
+    assert device.is_local_connected is local_connected
+
+
 @pytest.fixture(name="setup_rpc_channel")
 def setup_rpc_channel_fixture(rpc_channel: AsyncMock, payload: pathlib.Path) -> AsyncMock:
     """Fixture to set up the RPC channel for the tests."""