Supervisor subsystem

cflib2/__init__.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/_rust.pyi

@@ -411,6 +411,10 @@ class Crazyflie:
         r"""
         Get the platform subsystem
         """
+    def supervisor(self) -> Supervisor:
+        r"""
+        Get the supervisor subsystem
+        """
     def __str__(self) -> builtins.str: ...
     def __repr__(self) -> builtins.str: ...
 
@@ -428,30 +432,6 @@ class DisconnectedError(CrazyflieError):
 
     ...
 
-@typing.final
-class EmergencyControl:
-    r"""
-    Emergency control interface
-
-    Provides emergency stop functionality that immediately stops all motors.
-    """
-    async def send_emergency_stop(self) -> None:
-        r"""
-        Send emergency stop command
-
-        Immediately stops all motors and puts the Crazyflie into a locked state.
-        The drone will require a reboot before it can fly again.
-        """
-    async def send_emergency_stop_watchdog(self) -> None:
-        r"""
-        Send emergency stop watchdog
-
-        Activates/resets a watchdog failsafe that will automatically emergency stop
-        the drone if this message isn't sent every 1000ms. Once activated by the first
-        call, you must continue sending this periodically forever or the drone will
-        automatically emergency stop. Use only if you need automatic failsafe behavior.
-        """
-
 @typing.final
 class ExternalPose:
     r"""
@@ -797,10 +777,7 @@ class LedRingColor:
         Intensity percentage (0-100); values above 100 are clamped to 100
         """
     @intensity.setter
-    def intensity(self, value: builtins.int) -> None:
-        r"""
-        Intensity percentage (0-100); values above 100 are clamped to 100
-        """
+    def intensity(self, value: builtins.int) -> None: ...
     def __new__(
         cls,
         r: builtins.int = 0,
@@ -815,7 +792,7 @@ class LedRingColor:
         * `r` - Red component (0-255, default 0)
         * `g` - Green component (0-255, default 0)
         * `b` - Blue component (0-255, default 0)
-        * `intensity` - Intensity percentage (0-100, default 100); clamped to 100 if higher
+        * `intensity` - Intensity percentage (0-100, default 100); values above 100 are clamped to 100
         """
     def set(
         self,
@@ -970,10 +947,6 @@ class Localization:
     r"""
     Localization subsystem wrapper
     """
-    def emergency(self) -> EmergencyControl:
-        r"""
-        Get the emergency control interface
-        """
     def external_pose(self) -> ExternalPose:
         r"""
         Get the external pose interface
@@ -1480,22 +1453,6 @@ class Platform:
 
         As such, this shall only be used for test purpose in a controlled environment.
         """
-    async def send_arming_request(self, do_arm: builtins.bool) -> None:
-        r"""
-        Send system arm/disarm request
-
-        Arms or disarms the Crazyflie's safety systems. When disarmed, the motors
-        will not spin even if thrust commands are sent.
-
-        # Arguments
-        * `do_arm` - true to arm, false to disarm
-        """
-    async def send_crash_recovery_request(self) -> None:
-        r"""
-        Send crash recovery request
-
-        Requests recovery from a crash state detected by the Crazyflie.
-        """
     async def get_app_channel(self) -> typing.Optional[AppChannel]:
         r"""
         Get the bidirectional app channel for custom communication
@@ -1578,6 +1535,143 @@ class ProtocolVersionNotSupportedError(CrazyflieError):
 
     ...
 
+@typing.final
+class Supervisor:
+    r"""
+    Supervisor subsystem
+
+    Monitors the Crazyflie state and exposes arming, crash recovery,
+    and emergency stop controls. Obtain via `crazyflie.supervisor()`.
+    """
+    async def read(self) -> SupervisorState:
+        r"""
+        Read a consistent snapshot of the supervisor state
+
+        All flags on the returned snapshot are decoded from a single bitfield
+        read, so they are from the same moment and mutually consistent. Uses
+        time-based caching to avoid flooding the link.
+
+        The snapshot does not update itself: re-read to get fresh state, for
+        example on every iteration when polling.
+
+        Example:
+            state = await cf.supervisor().read()
+            if state.can_be_armed and not state.is_armed:
+                await cf.supervisor().send_arming_request(True)
+
+            # When polling, read inside the loop:
+            while not (await cf.supervisor().read()).is_armed:
+                await asyncio.sleep(0.5)
+        """
+    async def send_arming_request(self, do_arm: builtins.bool) -> None:
+        r"""
+        Send arm/disarm request
+
+        Arms or disarms the Crazyflie's motors. When disarmed, the motors
+        will not spin even if thrust commands are sent.
+
+        Args:
+            do_arm: True to arm, False to disarm
+        """
+    async def send_crash_recovery_request(self) -> None:
+        r"""
+        Send crash recovery request
+
+        Requests recovery from a crash state detected by the Crazyflie.
+        """
+    async def send_emergency_stop(self) -> None:
+        r"""
+        Send emergency stop
+
+        Immediately stops all motors and puts the Crazyflie into a locked state.
+        The drone will require a reboot before it can fly again.
+        """
+    async def send_emergency_stop_watchdog(self) -> None:
+        r"""
+        Send emergency stop watchdog
+
+        Activates/resets a watchdog failsafe that will automatically emergency
+        stop the drone if this message is not sent every 1000 ms. Once
+        activated, you must keep sending this periodically or the drone will
+        stop. Use only when you need automatic failsafe behaviour on
+        communication loss.
+        """
+
+@typing.final
+class SupervisorState:
+    r"""
+    A snapshot of the supervisor state
+
+    Decoded from a single supervisor bitfield read: all flags on one snapshot
+    are from the same moment and mutually consistent. The snapshot does not
+    update itself - call `Supervisor.read()` again for fresh state.
+    """
+    @property
+    def raw(self) -> builtins.int:
+        r"""
+        Raw bitfield value
+        """
+    @property
+    def can_be_armed(self) -> builtins.bool:
+        r"""
+        The Crazyflie can be armed - will accept an arming command
+        """
+    @property
+    def is_armed(self) -> builtins.bool:
+        r"""
+        The Crazyflie is armed
+        """
+    @property
+    def is_auto_armed(self) -> builtins.bool:
+        r"""
+        The Crazyflie is configured to automatically arm
+        """
+    @property
+    def can_fly(self) -> builtins.bool:
+        r"""
+        The Crazyflie is ready to fly
+        """
+    @property
+    def is_flying(self) -> builtins.bool:
+        r"""
+        The Crazyflie is flying
+        """
+    @property
+    def is_tumbled(self) -> builtins.bool:
+        r"""
+        The Crazyflie is tumbled (upside down)
+        """
+    @property
+    def is_locked(self) -> builtins.bool:
+        r"""
+        The Crazyflie is in the locked state and must be restarted
+        """
+    @property
+    def is_crashed(self) -> builtins.bool:
+        r"""
+        The Crazyflie has crashed
+        """
+    @property
+    def hl_control_active(self) -> builtins.bool:
+        r"""
+        High level commander is actively flying the drone
+        """
+    @property
+    def hl_traj_finished(self) -> builtins.bool:
+        r"""
+        High level commander trajectory has finished
+        """
+    @property
+    def hl_control_disabled(self) -> builtins.bool:
+        r"""
+        High level commander is disabled and not producing setpoints
+        """
+    def active_states(self) -> builtins.list[builtins.str]:
+        r"""
+        Names of all active states in this snapshot
+        """
+    def __repr__(self) -> builtins.str: ...
+
 class SystemError(CrazyflieError):
     r"""
     Async executor error.

cflib2/commander.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/console.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/error.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/high_level_commander.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/localization.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \
@@ -23,7 +21,6 @@
 """Localization subsystem types"""
 
 from cflib2._rust import (
-    EmergencyControl,
     ExternalPose,
     Lighthouse,
     LighthouseAngleData,
@@ -33,7 +30,6 @@
 )
 
 __all__ = [
-    "EmergencyControl",
     "ExternalPose",
     "Lighthouse",
     "LighthouseAngleData",

cflib2/log.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/memory.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/param.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/platform.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

cflib2/supervisor.py

@@ -0,0 +1,25 @@
+# ,---------,       ____  _ __
+# |  ,-^-,  |      / __ )(_) /_______________ _____  ___
+# | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \
+# | / ,--'  |    / /_/ / / /_/ /__/ /  / /_/ / / /_/  __/
+#    +------`   /_____/_/\__/\___/_/   \__,_/ /___/\___/
+#
+# Copyright (C) 2026 Bitcraze AB
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+"""Supervisor subsystem types"""
+
+from cflib2._rust import Supervisor, SupervisorState
+
+__all__ = ["Supervisor", "SupervisorState"]

cflib2/toc_cache.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # ,---------,       ____  _ __
 # |  ,-^-,  |      / __ )(_) /_______________ _____  ___
 # | (  O  ) |     / __  / / __/ ___/ ___/ __ `/_  / / _ \

examples/arming.py

@@ -54,7 +54,7 @@ async def main() -> None:
     cf = await Crazyflie.connect_from_uri(context, args.uri)
     print("Connected!")
 
-    platform = cf.platform()
+    supervisor = cf.supervisor()
 
     try:
         print("\n⚠️  WARNING: This will ARM the Crazyflie!")
@@ -64,7 +64,7 @@ async def main() -> None:
 
         # Arm the Crazyflie
         print("\n1. Arming the Crazyflie...")
-        await platform.send_arming_request(do_arm=True)
+        await supervisor.send_arming_request(do_arm=True)
         print("   ✓ Armed! Motors can now spin.")
 
         # Wait a few seconds
@@ -76,14 +76,14 @@ async def main() -> None:
 
         # Disarm the Crazyflie
         print("\n3. Disarming the Crazyflie...")
-        await platform.send_arming_request(do_arm=False)
+        await supervisor.send_arming_request(do_arm=False)
         print("   ✓ Disarmed! Motors are now disabled.")
 
         print("\n✓ Arming cycle complete!")
 
     except KeyboardInterrupt:
         print("\n\n⚠️  Interrupted! Disarming for safety...")
-        await platform.send_arming_request(do_arm=False)
+        await supervisor.send_arming_request(do_arm=False)
         print("   ✓ Disarmed!")
 
     finally: