step.watchdog_manager ===================== .. py:module:: step.watchdog_manager .. autoapi-nested-parse:: Watchdog manager — deadlines that cancel the main-mission task on expiry. Two usage patterns share one manager: - **Mission deadlines** — armed automatically by the robot runner from a ``Mission.time_budget`` class attribute. Never fed; expiry fires when the budget elapses. - **Keepalive watchdogs** — armed/fed/disarmed imperatively by user code via the ``start_watchdog`` / ``feed_watchdog`` / ``stop_watchdog`` steps. Must be fed before their timeout or expiry fires. On expiry, the manager cancels the registered *main-mission task*. This routes through the existing shutdown path in ``_run_missions``: background tasks drain, the shutdown mission runs, then the process exits. Attributes ---------- .. autoapisummary:: step.watchdog_manager.logger Exceptions ---------- .. autoapisummary:: step.watchdog_manager.WatchdogExpiredError Classes ------- .. autoapisummary:: step.watchdog_manager.WatchdogManager Functions --------- .. autoapisummary:: step.watchdog_manager.get_watchdog_manager Module Contents --------------- .. py:data:: logger .. py:exception:: WatchdogExpiredError Bases: :py:obj:`Exception` Raised on the main-mission task when a watchdog expires. .. py:class:: WatchdogManager Track armed watchdogs and cancel the main-mission task on expiry. Attached to the robot instance as ``robot._watchdog_manager``. Wire the main-mission task via ``attach_main_task`` before arming anything so the manager knows what to cancel. .. py:method:: attach_main_task(task: asyncio.Task[None]) -> None Register the task that should be cancelled when a watchdog fires. .. py:method:: detach_main_task() -> None Clear the main-task reference after main missions finish. .. py:property:: expired_name :type: Optional[str] Name of the watchdog that fired, or ``None`` if no expiry occurred. .. py:method:: arm(name: str, timeout: float, source: str = 'user') -> None Arm a watchdog. Replaces any existing entry with the same name. .. py:method:: feed(name: str) -> None Push the named watchdog's deadline out by its full timeout. .. py:method:: disarm(name: str, *, missing_ok: bool = False) -> None Cancel and remove the named watchdog. .. py:method:: active_names(source: Optional[str] = None) -> list[str] Return names of currently armed watchdogs, optionally filtered by source. .. py:method:: cancel_all() -> None :async: Cancel every armed watchdog and await cleanup. .. py:function:: get_watchdog_manager(robot: object) -> WatchdogManager Get or create the ``WatchdogManager`` attached to *robot*.