feat(ble): add typed listen_gear/listen_ui_desire broadcast listeners

docs/bluetooth_vehicles.md

@@ -496,6 +496,8 @@ similar in spirit to
 - `listen_vehicle_lock_state(callback)` - `VehicleLockState_E`
 - `listen_vehicle_sleep_status(callback)` - `VehicleSleepStatus_E`
 - `listen_user_presence(callback)` - `UserPresence_E`
+- `listen_gear(callback)` - `Gear_E`
+- `listen_ui_desire(callback)` - `UIDesire_E`
 - `listen_front_driver_door(callback)` - `ClosureState_E`
 - `listen_front_passenger_door(callback)` - `ClosureState_E`
 - `listen_rear_driver_door(callback)` - `ClosureState_E`
@@ -521,14 +523,14 @@ unsubscribe()
 The door/trunk/charge-port/tonneau listeners and `listen_tonneau_percent_open`
 only fire on broadcasts that actually carry the corresponding submessage
 (`closureStatuses`/`detailedClosureStatus`) - not every status broadcast
-includes them. `listen_vehicle_lock_state`, `listen_vehicle_sleep_status`, and
-`listen_user_presence` fire on every `VehicleStatus` broadcast, since proto3
-gives no presence tracking for a scalar enum field.
+includes them. `listen_vehicle_lock_state`, `listen_vehicle_sleep_status`,
+`listen_user_presence`, `listen_gear`, and `listen_ui_desire` fire on every
+`VehicleStatus` broadcast, since proto3 gives no presence tracking for a
+scalar enum field.
 
-`VehicleStatus`'s `uiDesire` and `gear` fields (added by a later proto sync)
-and anything not decoded into `VehicleStatus` - other VCSEC broadcast payloads
+Anything not decoded into `VehicleStatus` - other VCSEC broadcast payloads
 (`CommandStatus`, whitelist events, faults) and any future
-infotainment-domain broadcast - have no typed listener surface. Use the untyped
+infotainment-domain broadcast - has no typed listener surface. Use the untyped
 `listen_broadcast`, which delivers every raw unsolicited `RoutableMessage` for
 a given `Domain`, including decoded `VehicleStatus` broadcasts:
 

tesla_fleet_api/tesla/vehicle/broadcast.py

@@ -4,12 +4,10 @@
 subscription used for command replies (``_on_message`` in ``bluetooth.py``).
 Each modeled leaf field of ``VehicleStatus`` gets its own typed
 ``listen_<field>`` method here, mirroring python-teslemetry-stream's
-per-field listener surface; ``uiDesire`` and ``gear`` (added to
-``VehicleStatus`` in a later proto sync) have no typed listener yet. The
-untyped ``listen_broadcast`` receives every raw broadcast for a domain,
-including ``VehicleStatus`` and payloads that have no typed listener surface:
-other VCSEC broadcast payloads (``CommandStatus``, whitelist events, faults),
-the two unmodeled ``VehicleStatus`` fields above, and any future
+per-field listener surface. The untyped ``listen_broadcast`` receives every
+raw broadcast for a domain, including ``VehicleStatus`` and payloads that
+have no typed listener surface: other VCSEC broadcast payloads
+(``CommandStatus``, whitelist events, faults) and any future
 infotainment-domain broadcast.
 """
 
@@ -24,6 +22,8 @@
 )
 from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import (
     ClosureState_E,
+    Gear_E,
+    UIDesire_E,
     UserPresence_E,
     VehicleLockState_E,
     VehicleSleepStatus_E,
@@ -118,6 +118,18 @@ def listen_user_presence(
             self._status_listeners, lambda status: callback(status.userPresence)
         )
 
+    def listen_gear(self, callback: Callable[[Gear_E], None]) -> Unsubscribe:
+        """Listen for the vehicle's gear selection."""
+        return self._register(
+            self._status_listeners, lambda status: callback(status.gear)
+        )
+
+    def listen_ui_desire(self, callback: Callable[[UIDesire_E], None]) -> Unsubscribe:
+        """Listen for the vehicle's UI desire state."""
+        return self._register(
+            self._status_listeners, lambda status: callback(status.uiDesire)
+        )
+
     def listen_tonneau_percent_open(
         self, callback: Callable[[int], None]
     ) -> Unsubscribe:

tests/test_ble_broadcast_listeners.py

@@ -29,7 +29,9 @@
     CommandStatus,
     DetailedClosureStatus,
     FromVCSECMessage,
+    Gear_E,
     OperationStatus_E,
+    UIDesire_E,
     UserPresence_E,
     VehicleLockState_E,
     VehicleSleepStatus_E,
@@ -175,6 +177,60 @@ async def test_multiple_field_listeners_fire_from_one_broadcast(self) -> None:
         self.assertEqual(lock_seen, [VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED])
         self.assertEqual(presence_seen, [UserPresence_E.VEHICLE_USER_PRESENCE_PRESENT])
 
+    async def test_gear_listener_fires_with_typed_value(self) -> None:
+        vehicle = _make_vehicle()
+        seen: list[Any] = []
+        vehicle.listen_gear(seen.append)
+
+        vehicle._on_message(_status_broadcast(VehicleStatus(gear=Gear_E.GEAR_DRIVE)))
+
+        self.assertEqual(seen, [Gear_E.GEAR_DRIVE])
+
+    async def test_gear_listener_fires_with_default_when_absent(self) -> None:
+        # gear is a scalar proto3 field, so it has no HasField presence -
+        # a broadcast that never sets it still fires with the 0 default.
+        vehicle = _make_vehicle()
+        seen: list[Any] = []
+        vehicle.listen_gear(seen.append)
+
+        vehicle._on_message(
+            _status_broadcast(
+                VehicleStatus(
+                    vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED
+                )
+            )
+        )
+
+        self.assertEqual(seen, [Gear_E.GEAR_UNKNOWN])
+
+    async def test_ui_desire_listener_fires_with_typed_value(self) -> None:
+        vehicle = _make_vehicle()
+        seen: list[Any] = []
+        vehicle.listen_ui_desire(seen.append)
+
+        vehicle._on_message(
+            _status_broadcast(VehicleStatus(uiDesire=UIDesire_E.UI_DESIRE_HAS_DATA))
+        )
+
+        self.assertEqual(seen, [UIDesire_E.UI_DESIRE_HAS_DATA])
+
+    async def test_ui_desire_listener_fires_with_default_when_absent(self) -> None:
+        # uiDesire is a scalar proto3 field, so it has no HasField presence -
+        # a broadcast that never sets it still fires with the 0 default.
+        vehicle = _make_vehicle()
+        seen: list[Any] = []
+        vehicle.listen_ui_desire(seen.append)
+
+        vehicle._on_message(
+            _status_broadcast(
+                VehicleStatus(
+                    vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED
+                )
+            )
+        )
+
+        self.assertEqual(seen, [UIDesire_E.UI_DESIRE_NONE])
+
 
 class ClosureListenerTests(IsolatedAsyncioTestCase):
     async def test_all_eight_closure_listeners_read_their_own_field(self) -> None: