Merge branch 'main' into e2e-tests

allenporter · web-flow · commit 4866534ab538 · 2025-12-14T15:49:43.000-08:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,56 @@
 
 <!-- version list -->
 
+## v3.16.1 (2025-12-14)
+
+### Bug Fixes
+
+- Share a HealthManager instance across all mqtt channels
+  ([#672](https://github.com/Python-roborock/python-roborock/pull/672),
+  [`4ad95dd`](https://github.com/Python-roborock/python-roborock/commit/4ad95ddee4d4d4cd64c7908f150c71d81f45e705))
+
+
+## v3.16.0 (2025-12-14)
+
+### Bug Fixes
+
+- Fix bugs in the subscription idle timeout
+  ([#665](https://github.com/Python-roborock/python-roborock/pull/665),
+  [`85b7bee`](https://github.com/Python-roborock/python-roborock/commit/85b7beeb810cfb3d501658cd44f06b2c0052ca33))
+
+- Harden the device connection logic used in startup
+  ([#666](https://github.com/Python-roborock/python-roborock/pull/666),
+  [`19703f4`](https://github.com/Python-roborock/python-roborock/commit/19703f42fe692a38f8f8639b1136a7585eae76fc))
+
+- Harden the initial startup logic
+  ([#666](https://github.com/Python-roborock/python-roborock/pull/666),
+  [`19703f4`](https://github.com/Python-roborock/python-roborock/commit/19703f42fe692a38f8f8639b1136a7585eae76fc))
+
+### Chores
+
+- Apply suggestions from code review
+  ([#675](https://github.com/Python-roborock/python-roborock/pull/675),
+  [`ab2de5b`](https://github.com/Python-roborock/python-roborock/commit/ab2de5bda7b8e1ff1ad46c7f2bf3b39dc9af4ace))
+
+- Clarify comments and docstrings
+  ([#666](https://github.com/Python-roborock/python-roborock/pull/666),
+  [`19703f4`](https://github.com/Python-roborock/python-roborock/commit/19703f42fe692a38f8f8639b1136a7585eae76fc))
+
+- Fix logging ([#666](https://github.com/Python-roborock/python-roborock/pull/666),
+  [`19703f4`](https://github.com/Python-roborock/python-roborock/commit/19703f42fe692a38f8f8639b1136a7585eae76fc))
+
+- Reduce whitespace changes ([#666](https://github.com/Python-roborock/python-roborock/pull/666),
+  [`19703f4`](https://github.com/Python-roborock/python-roborock/commit/19703f42fe692a38f8f8639b1136a7585eae76fc))
+
+- Revert whitespace change ([#666](https://github.com/Python-roborock/python-roborock/pull/666),
+  [`19703f4`](https://github.com/Python-roborock/python-roborock/commit/19703f42fe692a38f8f8639b1136a7585eae76fc))
+
+### Features
+
+- Add basic schedule getting ([#675](https://github.com/Python-roborock/python-roborock/pull/675),
+  [`ab2de5b`](https://github.com/Python-roborock/python-roborock/commit/ab2de5bda7b8e1ff1ad46c7f2bf3b39dc9af4ace))
+
+
 ## v3.15.0 (2025-12-14)
 
 ### Chores
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "python-roborock"
-version = "3.15.0"
+version = "3.16.1"
 description = "A package to control Roborock vacuums."
 authors = [{ name = "humbertogontijo", email = "humbertogontijo@users.noreply.github.com" }, {name="Lash-L"}, {name="allenporter"}]
 requires-python = ">=3.11, <4"
diff --git a/roborock/data/containers.py b/roborock/data/containers.py
@@ -284,6 +284,15 @@ class HomeDataScene(RoborockBase):
     name: str
 
 
+@dataclass
+class HomeDataSchedule(RoborockBase):
+    id: int
+    cron: str
+    repeated: bool
+    enabled: bool
+    param: dict | None = None
+
+
 @dataclass
 class HomeData(RoborockBase):
     id: int
diff --git a/roborock/devices/device.py b/roborock/devices/device.py
@@ -147,34 +147,45 @@ async def start_connect(self) -> None:
         called. The device will automatically attempt to reconnect if the connection
         is lost.
         """
-        start_attempt: asyncio.Event = asyncio.Event()
+        # The future will be set to True if the first attempt succeeds, False if
+        # it fails, or an exception if an unexpected error occurs.
+        # We use this to wait a short time for the first attempt to complete. We
+        # don't actually care about the result, just that we waited long enough.
+        start_attempt: asyncio.Future[bool] = asyncio.Future()
 
         async def connect_loop() -> None:
-            backoff = MIN_BACKOFF_INTERVAL
             try:
+                backoff = MIN_BACKOFF_INTERVAL
                 while True:
                     try:
                         await self.connect()
-                        start_attempt.set()
+                        if not start_attempt.done():
+                            start_attempt.set_result(True)
                         self._has_connected = True
                         self._ready_callbacks(self)
                         return
                     except RoborockException as e:
-                        start_attempt.set()
+                        if not start_attempt.done():
+                            start_attempt.set_result(False)
                         self._logger.info("Failed to connect (retry %s): %s", backoff.total_seconds(), e)
                         await asyncio.sleep(backoff.total_seconds())
                         backoff = min(backoff * BACKOFF_MULTIPLIER, MAX_BACKOFF_INTERVAL)
+                    except Exception as e:  # pylint: disable=broad-except
+                        if not start_attempt.done():
+                            start_attempt.set_exception(e)
+                        self._logger.exception("Uncaught error during connect: %s", e)
+                        return
             except asyncio.CancelledError:
                 self._logger.debug("connect_loop was cancelled for device %s", self.duid)
-                # Clean exit on cancellation
-                return
             finally:
-                start_attempt.set()
+                if not start_attempt.done():
+                    start_attempt.set_result(False)
 
         self._connect_task = asyncio.create_task(connect_loop())
 
         try:
-            await asyncio.wait_for(start_attempt.wait(), timeout=START_ATTEMPT_TIMEOUT.total_seconds())
+            async with asyncio.timeout(START_ATTEMPT_TIMEOUT.total_seconds()):
+                await start_attempt
         except TimeoutError:
             self._logger.debug("Initial connection attempt took longer than expected, will keep trying in background")
 
@@ -189,6 +200,7 @@ async def connect(self) -> None:
             except RoborockException:
                 unsub()
                 raise
+        self._logger.info("Connected to device")
         self._unsub = unsub
 
     async def close(self) -> None:
diff --git a/roborock/devices/mqtt_channel.py b/roborock/devices/mqtt_channel.py
@@ -6,6 +6,7 @@
 from roborock.callbacks import decoder_callback
 from roborock.data import HomeDataDevice, RRiot, UserData
 from roborock.exceptions import RoborockException
+from roborock.mqtt.health_manager import HealthManager
 from roborock.mqtt.session import MqttParams, MqttSession, MqttSessionException
 from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder
 from roborock.roborock_message import RoborockMessage
@@ -42,6 +43,11 @@ def is_connected(self) -> bool:
         """
         return self._mqtt_session.connected
 
+    @property
+    def health_manager(self) -> HealthManager:
+        """Return the health manager for the session."""
+        return self._mqtt_session.health_manager
+
     @property
     def is_local_connected(self) -> bool:
         """Return true if the channel is connected locally."""
diff --git a/roborock/devices/v1_channel.py b/roborock/devices/v1_channel.py
@@ -181,7 +181,6 @@ def __init__(
         self._logger = RoborockLoggerAdapter(device_uid, _LOGGER)
         self._security_data = security_data
         self._mqtt_channel = mqtt_channel
-        self._mqtt_health_manager = HealthManager(self._mqtt_channel.restart)
         self._local_session = local_session
         self._local_channel: LocalChannel | None = None
         self._mqtt_unsub: Callable[[], None] | None = None
@@ -272,7 +271,7 @@ def _create_mqtt_rpc_strategy(self, decoder: Callable[[RoborockMessage], Any] =
                 security_data=self._security_data,
             ),
             decoder=decoder,
-            health_manager=self._mqtt_health_manager,
+            health_manager=self._mqtt_channel.health_manager,
         )
 
     async def subscribe(self, callback: Callable[[RoborockMessage], None]) -> Callable[[], None]:
diff --git a/roborock/mqtt/health_manager.py b/roborock/mqtt/health_manager.py
@@ -6,8 +6,11 @@
 """
 
 import datetime
+import logging
 from collections.abc import Awaitable, Callable
 
+_LOGGER = logging.getLogger(__name__)
+
 # Number of consecutive timeouts before considering the connection unhealthy.
 TIMEOUT_THRESHOLD = 3
 
@@ -45,7 +48,13 @@ async def on_timeout(self) -> None:
         self._consecutive_timeouts += 1
         if self._consecutive_timeouts >= TIMEOUT_THRESHOLD:
             now = datetime.datetime.now(datetime.UTC)
-            if self._last_restart is None or now - self._last_restart >= RESTART_COOLDOWN:
+            since_last = (now - self._last_restart) if self._last_restart else None
+            if since_last is None or since_last >= RESTART_COOLDOWN:
+                _LOGGER.debug(
+                    "Restarting MQTT session after %d consecutive timeouts (duration since last restart %s)",
+                    self._consecutive_timeouts,
+                    since_last,
+                )
                 await self._restart()
                 self._last_restart = now
                 self._consecutive_timeouts = 0
diff --git a/roborock/mqtt/roborock_session.py b/roborock/mqtt/roborock_session.py
@@ -19,6 +19,7 @@
 
 from roborock.callbacks import CallbackMap
 
+from .health_manager import HealthManager
 from .session import MqttParams, MqttSession, MqttSessionException, MqttSessionUnauthorized
 
 _LOGGER = logging.getLogger(__name__)
@@ -69,17 +70,24 @@ def __init__(
         self._stop = False
         self._backoff = MIN_BACKOFF_INTERVAL
         self._client: aiomqtt.Client | None = None
+        self._client_subscribed_topics: set[str] = set()
         self._client_lock = asyncio.Lock()
         self._listeners: CallbackMap[str, bytes] = CallbackMap(_LOGGER)
         self._connection_task: asyncio.Task[None] | None = None
         self._topic_idle_timeout = topic_idle_timeout
         self._idle_timers: dict[str, asyncio.Task[None]] = {}
+        self._health_manager = HealthManager(self.restart)
 
     @property
     def connected(self) -> bool:
         """True if the session is connected to the broker."""
         return self._healthy
 
+    @property
+    def health_manager(self) -> HealthManager:
+        """Return the health manager for the session."""
+        return self._health_manager
+
     async def start(self) -> None:
         """Start the MQTT session.
 
@@ -218,7 +226,7 @@ async def _mqtt_client(self, params: MqttParams) -> aiomqtt.Client:
                 # Re-establish any existing subscriptions
                 async with self._client_lock:
                     self._client = client
-                    for topic in self._listeners.keys():
+                    for topic in self._client_subscribed_topics:
                         _LOGGER.debug("Re-establishing subscription to topic %s", topic)
                         # TODO: If this fails it will break the whole connection. Make
                         # this retry again in the background with backoff.
@@ -249,32 +257,42 @@ async def subscribe(self, topic: str, callback: Callable[[bytes], None]) -> Call
         unsub = self._listeners.add_callback(topic, callback)
 
         async with self._client_lock:
-            if self._client:
-                _LOGGER.debug("Establishing subscription to topic %s", topic)
-                try:
-                    await self._client.subscribe(topic)
-                except MqttError as err:
-                    # Clean up the callback if subscription fails
-                    unsub()
-                    raise MqttSessionException(f"Error subscribing to topic: {err}") from err
-            else:
-                _LOGGER.debug("Client not connected, will establish subscription later")
-
-        def schedule_unsubscribe():
+            if topic not in self._client_subscribed_topics:
+                self._client_subscribed_topics.add(topic)
+                if self._client:
+                    _LOGGER.debug("Establishing subscription to topic %s", topic)
+                    try:
+                        await self._client.subscribe(topic)
+                    except MqttError as err:
+                        # Clean up the callback if subscription fails
+                        unsub()
+                        self._client_subscribed_topics.discard(topic)
+                        raise MqttSessionException(f"Error subscribing to topic: {err}") from err
+                else:
+                    _LOGGER.debug("Client not connected, will establish subscription later")
+
+        def schedule_unsubscribe() -> None:
             async def idle_unsubscribe():
                 try:
                     await asyncio.sleep(self._topic_idle_timeout.total_seconds())
                     # Only unsubscribe if there are no callbacks left for this topic
                     if not self._listeners.get_callbacks(topic):
                         async with self._client_lock:
+                            # Check again if we have listeners, in case a subscribe happened
+                            # while we were waiting for the lock or after we popped the timer.
+                            if self._listeners.get_callbacks(topic):
+                                _LOGGER.debug("Skipping unsubscribe for %s, new listeners added", topic)
+                                return
+
+                            self._idle_timers.pop(topic, None)
+                            self._client_subscribed_topics.discard(topic)
+
                             if self._client:
                                 _LOGGER.debug("Idle timeout expired, unsubscribing from topic %s", topic)
                                 try:
                                     await self._client.unsubscribe(topic)
                                 except MqttError as err:
                                     _LOGGER.warning("Error unsubscribing from topic %s: %s", topic, err)
-                    # Clean up timer from dict
-                    self._idle_timers.pop(topic, None)
                 except asyncio.CancelledError:
                     _LOGGER.debug("Idle unsubscribe for topic %s cancelled", topic)
 
@@ -286,7 +304,10 @@ def delayed_unsub():
             unsub()  # Remove the callback from CallbackMap
             # If no more callbacks for this topic, start idle timer
             if not self._listeners.get_callbacks(topic):
+                _LOGGER.debug("Unsubscribing topic %s, starting idle timer", topic)
                 schedule_unsubscribe()
+            else:
+                _LOGGER.debug("Unsubscribing topic %s, still have active callbacks", topic)
 
         return delayed_unsub
 
@@ -323,6 +344,11 @@ def connected(self) -> bool:
         """True if the session is connected to the broker."""
         return self._session.connected
 
+    @property
+    def health_manager(self) -> HealthManager:
+        """Return the health manager for the session."""
+        return self._session.health_manager
+
     async def _maybe_start(self) -> None:
         """Start the MQTT session if not already started."""
         async with self._lock:
diff --git a/roborock/mqtt/session.py b/roborock/mqtt/session.py
@@ -5,6 +5,7 @@
 from dataclasses import dataclass
 
 from roborock.exceptions import RoborockException
+from roborock.mqtt.health_manager import HealthManager
 
 DEFAULT_TIMEOUT = 30.0
 
@@ -40,6 +41,11 @@ class MqttSession(ABC):
     def connected(self) -> bool:
         """True if the session is connected to the broker."""
 
+    @property
+    @abstractmethod
+    def health_manager(self) -> HealthManager:
+        """Return the health manager for the session."""
+
     @abstractmethod
     async def subscribe(self, device_id: str, callback: Callable[[bytes], None]) -> Callable[[], None]:
         """Invoke the callback when messages are received on the topic.
diff --git a/roborock/web_api.py b/roborock/web_api.py
@@ -14,6 +14,7 @@
 from aiohttp import ContentTypeError, FormData
 from pyrate_limiter import BucketFullException, Duration, Limiter, Rate
 
+from roborock import HomeDataSchedule
 from roborock.data import HomeData, HomeDataRoom, HomeDataScene, ProductResponse, RRiot, UserData
 from roborock.exceptions import (
     RoborockAccountDoesNotExist,
@@ -607,6 +608,28 @@ async def execute_scene(self, user_data: UserData, scene_id: int) -> None:
         if not execute_scene_response.get("success"):
             raise RoborockException(execute_scene_response)
 
+    async def get_schedules(self, user_data: UserData, device_id: str) -> list[HomeDataSchedule]:
+        rriot = user_data.rriot
+        if rriot is None:
+            raise RoborockException("rriot is none")
+        if rriot.r.a is None:
+            raise RoborockException("Missing field 'a' in rriot reference")
+        schedules_request = PreparedRequest(
+            rriot.r.a,
+            self.session,
+            {
+                "Authorization": _get_hawk_authentication(rriot, f"/user/devices/{device_id}/jobs"),
+            },
+        )
+        schedules_response = await schedules_request.request("get", f"/user/devices/{str(device_id)}/jobs")
+        if not schedules_response.get("success"):
+            raise RoborockException(schedules_response)
+        schedules = schedules_response.get("result")
+        if isinstance(schedules, list):
+            return [HomeDataSchedule.from_dict(schedule) for schedule in schedules]
+        else:
+            raise RoborockException(f"schedule_response result was an unexpected type: {schedules}")
+
     async def get_products(self, user_data: UserData) -> ProductResponse:
         """Gets all products and their schemas, good for determining status codes and model numbers."""
         base_url = await self.base_url
diff --git a/tests/conftest.py b/tests/conftest.py
diff --git a/tests/devices/test_device_manager.py b/tests/devices/test_device_manager.py
diff --git a/tests/mqtt/test_roborock_session.py b/tests/mqtt/test_roborock_session.py
diff --git a/tests/test_web_api.py b/tests/test_web_api.py
diff --git a/uv.lock b/uv.lock
