AVRCP: Delegate UID and Addressed Player

bumble/avrcp.py

@@ -1517,7 +1517,7 @@ class PlaybackPositionChangedEvent(Event):
 @dataclass
 class TrackChangedEvent(Event):
     event_id = EventId.TRACK_CHANGED
-    identifier: bytes = field(metadata=hci.metadata('*'))
+    uid: int = field(metadata=_UINT64_BE_METADATA)
 
 
 # -----------------------------------------------------------------------------
@@ -1651,6 +1651,8 @@ class Delegate:
             else {}
         )
         self.player_app_settings = {}
+        self.uid_counter = 0
+        self.addressed_player_id = 0
 
     async def get_supported_events(self) -> list[EventId]:
         return self.supported_events
@@ -1706,6 +1708,12 @@ class Delegate:
             uid_counter,
         )
 
+    async def get_uid_counter(self) -> int:
+        return self.uid_counter
+
+    async def get_addressed_player_id(self) -> int:
+        return self.addressed_player_id
+
     # TODO add other delegate methods
 
 
@@ -2049,13 +2057,13 @@ class Protocol(utils.EventEmitter):
 
     async def monitor_track_changed(
         self,
-    ) -> AsyncIterator[bytes]:
+    ) -> AsyncIterator[int]:
         """Monitor Track changes from the connected peer."""
         async for event in self.monitor_events(EventId.TRACK_CHANGED, 0):
             if not isinstance(event, TrackChangedEvent):
                 logger.warning("unexpected event class")
                 continue
-            yield event.identifier
+            yield event.uid
 
     async def monitor_playback_position(
         self, playback_interval: int
@@ -2148,11 +2156,9 @@ class Protocol(utils.EventEmitter):
         """Notify the connected peer of a Playback Status change."""
         self.notify_event(PlaybackStatusChangedEvent(status))
 
-    def notify_track_changed(self, identifier: bytes) -> None:
+    def notify_track_changed(self, uid: int) -> None:
         """Notify the connected peer of a Track change."""
-        if len(identifier) != 8:
-            raise core.InvalidArgumentError("identifier must be 8 bytes")
-        self.notify_event(TrackChangedEvent(identifier))
+        self.notify_event(TrackChangedEvent(uid))
 
     def notify_playback_position_changed(self, position: int) -> None:
         """Notify the connected peer of a Position change."""
@@ -2830,6 +2836,15 @@ class Protocol(utils.EventEmitter):
                     )
                 case EventId.AVAILABLE_PLAYERS_CHANGED:
                     event = AvailablePlayersChangedEvent()
+                case EventId.ADDRESSED_PLAYER_CHANGED:
+                    event = AddressedPlayerChangedEvent(
+                        AddressedPlayerChangedEvent.Player(
+                            player_id=await self.delegate.get_addressed_player_id(),
+                            uid_counter=await self.delegate.get_uid_counter(),
+                        )
+                    )
+                case EventId.UIDS_CHANGED:
+                    event = UidsChangedEvent(await self.delegate.get_uid_counter())
                 case _:
                     logger.warning(
                         "Event supported but not handled %s", command.event_id

tests/avrcp_test.py

@@ -598,7 +598,7 @@ async def test_list_player_application_settings():
             avrcp.ApplicationSetting.ShuffleOnOffStatus.GROUP_SHUFFLE,
         ],
     }
-    delegate = two_devices.protocols[1].delegate = avrcp.Delegate(
+    two_devices.protocols[1].delegate = avrcp.Delegate(
         supported_player_app_settings=expected_settings
     )
     actual_settings = await two_devices.protocols[
@@ -715,6 +715,50 @@ async def test_monitor_now_playing_content():
         await anext(now_playing_iter)
 
 
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_monitor_uid_changed():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    delegate = two_devices.protocols[1].delegate = avrcp.Delegate(
+        [avrcp.EventId.UIDS_CHANGED]
+    )
+    delegate.uid_counter = 0
+    uid_iter = two_devices.protocols[0].monitor_uids()
+
+    # Interim
+    assert (await anext(uid_iter)) == 0
+    # Changed
+    # Changed
+    two_devices.protocols[1].notify_uids_changed(1)
+    assert (await anext(uid_iter)) == 1
+
+
+# -----------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_monitor_addressed_player():
+    two_devices = await TwoDevices.create_with_avdtp()
+
+    delegate = two_devices.protocols[1].delegate = avrcp.Delegate(
+        [avrcp.EventId.ADDRESSED_PLAYER_CHANGED]
+    )
+    delegate.uid_counter = 0
+    delegate.addressed_player_id = 0
+    addressed_player_iter = two_devices.protocols[0].monitor_addressed_player()
+
+    # Interim
+    assert (
+        await anext(addressed_player_iter)
+    ) == avrcp.AddressedPlayerChangedEvent.Player(player_id=0, uid_counter=0)
+    # Changed
+    two_devices.protocols[1].notify_addressed_player_changed(
+        avrcp.AddressedPlayerChangedEvent.Player(player_id=1, uid_counter=1)
+    )
+    assert (
+        await anext(addressed_player_iter)
+    ) == avrcp.AddressedPlayerChangedEvent.Player(player_id=1, uid_counter=1)
+
+
 # -----------------------------------------------------------------------------
 @pytest.mark.asyncio
 async def test_monitor_player_app_settings():