# Copyright 2021-2023 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
from __future__ import annotations

import asyncio
import enum
import functools
import logging
import struct
from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
from dataclasses import dataclass, field
from typing import ClassVar, SupportsBytes, TypeVar

from bumble import avc, avctp, core, hci, l2cap, sdp, utils
from bumble.colors import color
from bumble.device import Connection, Device
from bumble.sdp import (
    SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
    SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
    SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
    SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
    SDP_PUBLIC_BROWSE_ROOT,
    SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
    SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
    SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
    DataElement,
    ServiceAttribute,
)

# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)


# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
AVRCP_PID = 0x110E
AVRCP_BLUETOOTH_SIG_COMPANY_ID = 0x001958


_UINT64_BE_METADATA = hci.metadata(
    {
        'parser': lambda data, offset: (
            offset + 8,
            int.from_bytes(data[offset : offset + 8], byteorder='big'),
        ),
        'serializer': lambda x: x.to_bytes(8, byteorder='big'),
    }
)


class PduId(utils.OpenIntEnum):
    GET_CAPABILITIES = 0x10
    LIST_PLAYER_APPLICATION_SETTING_ATTRIBUTES = 0x11
    LIST_PLAYER_APPLICATION_SETTING_VALUES = 0x12
    GET_CURRENT_PLAYER_APPLICATION_SETTING_VALUE = 0x13
    SET_PLAYER_APPLICATION_SETTING_VALUE = 0x14
    GET_PLAYER_APPLICATION_SETTING_ATTRIBUTE_TEXT = 0x15
    GET_PLAYER_APPLICATION_SETTING_VALUE_TEXT = 0x16
    INFORM_DISPLAYABLE_CHARACTER_SET = 0x17
    INFORM_BATTERY_STATUS_OF_CT = 0x18
    GET_ELEMENT_ATTRIBUTES = 0x20
    GET_PLAY_STATUS = 0x30
    REGISTER_NOTIFICATION = 0x31
    REQUEST_CONTINUING_RESPONSE = 0x40
    ABORT_CONTINUING_RESPONSE = 0x41
    SET_ABSOLUTE_VOLUME = 0x50
    SET_ADDRESSED_PLAYER = 0x60
    SET_BROWSED_PLAYER = 0x70
    GET_FOLDER_ITEMS = 0x71
    CHANGE_PATH = 0x72
    GET_ITEM_ATTRIBUTES = 0x73
    PLAY_ITEM = 0x74
    GET_TOTAL_NUMBER_OF_ITEMS = 0x75
    SEARCH = 0x80
    ADD_TO_NOW_PLAYING = 0x90


class CharacterSetId(hci.SpecableEnum):
    UTF_8 = 0x6A


class MediaAttributeId(hci.SpecableEnum):
    TITLE = 0x01
    ARTIST_NAME = 0x02
    ALBUM_NAME = 0x03
    TRACK_NUMBER = 0x04
    TOTAL_NUMBER_OF_TRACKS = 0x05
    GENRE = 0x06
    PLAYING_TIME = 0x07
    DEFAULT_COVER_ART = 0x08


class PlayStatus(hci.SpecableEnum):
    STOPPED = 0x00
    PLAYING = 0x01
    PAUSED = 0x02
    FWD_SEEK = 0x03
    REV_SEEK = 0x04
    ERROR = 0xFF


class EventId(hci.SpecableEnum):
    PLAYBACK_STATUS_CHANGED = 0x01
    TRACK_CHANGED = 0x02
    TRACK_REACHED_END = 0x03
    TRACK_REACHED_START = 0x04
    PLAYBACK_POS_CHANGED = 0x05
    BATT_STATUS_CHANGED = 0x06
    SYSTEM_STATUS_CHANGED = 0x07
    PLAYER_APPLICATION_SETTING_CHANGED = 0x08
    NOW_PLAYING_CONTENT_CHANGED = 0x09
    AVAILABLE_PLAYERS_CHANGED = 0x0A
    ADDRESSED_PLAYER_CHANGED = 0x0B
    UIDS_CHANGED = 0x0C
    VOLUME_CHANGED = 0x0D

    def __bytes__(self) -> bytes:
        return bytes([int(self)])


class StatusCode(hci.SpecableEnum):
    INVALID_COMMAND = 0x00
    INVALID_PARAMETER = 0x01
    PARAMETER_CONTENT_ERROR = 0x02
    INTERNAL_ERROR = 0x03
    OPERATION_COMPLETED = 0x04
    UID_CHANGED = 0x05
    INVALID_DIRECTION = 0x07
    NOT_A_DIRECTORY = 0x08
    DOES_NOT_EXIST = 0x09
    INVALID_SCOPE = 0x0A
    RANGE_OUT_OF_BOUNDS = 0x0B
    FOLDER_ITEM_IS_NOT_PLAYABLE = 0x0C
    MEDIA_IN_USE = 0x0D
    NOW_PLAYING_LIST_FULL = 0x0E
    SEARCH_NOT_SUPPORTED = 0x0F
    SEARCH_IN_PROGRESS = 0x10
    INVALID_PLAYER_ID = 0x11
    PLAYER_NOT_BROWSABLE = 0x12
    PLAYER_NOT_ADDRESSED = 0x13
    NO_VALID_SEARCH_RESULTS = 0x14
    NO_AVAILABLE_PLAYERS = 0x15
    ADDRESSED_PLAYER_CHANGED = 0x16


class Scope(hci.SpecableEnum):
    MEDIA_PLAYER_LIST = 0x00
    MEDIA_PLAYER_VIRTUAL_FILESYSTEM = 0x01
    SEARCH = 0x02
    NOW_PLAYING = 0x03


class ControllerFeatures(enum.IntFlag):
    # fmt: off
    CATEGORY_1                                      = 1 << 0
    CATEGORY_2                                      = 1 << 1
    CATEGORY_3                                      = 1 << 2
    CATEGORY_4                                      = 1 << 3
    SUPPORTS_BROWSING                               = 1 << 6
    SUPPORTS_COVER_ART_GET_IMAGE_PROPERTIES_FEATURE = 1 << 7
    SUPPORTS_COVER_ART_GET_IMAGE_FEATURE            = 1 << 8
    SUPPORTS_COVER_ART_GET_LINKED_THUMBNAIL_FEATURE = 1 << 9


class TargetFeatures(enum.IntFlag):
    # fmt: off
    CATEGORY_1                                  = 1 << 0
    CATEGORY_2                                  = 1 << 1
    CATEGORY_3                                  = 1 << 2
    CATEGORY_4                                  = 1 << 3
    PLAYER_APPLICATION_SETTINGS                 = 1 << 4
    GROUP_NAVIGATION                            = 1 << 5
    SUPPORTS_BROWSING                           = 1 << 6
    SUPPORTS_MULTIPLE_MEDIA_PLAYER_APPLICATIONS = 1 << 7
    SUPPORTS_COVER_ART                          = 1 << 8


# -----------------------------------------------------------------------------
@dataclass
class ControllerServiceSdpRecord:
    service_record_handle: int
    avctp_version: tuple[int, int] = (1, 4)
    avrcp_version: tuple[int, int] = (1, 6)
    supported_features: int | ControllerFeatures = ControllerFeatures(1)

    def to_service_attributes(self) -> list[ServiceAttribute]:
        avctp_version_int = self.avctp_version[0] << 8 | self.avctp_version[1]
        avrcp_version_int = self.avrcp_version[0] << 8 | self.avrcp_version[1]

        attributes = [
            ServiceAttribute(
                SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
                DataElement.unsigned_integer_32(self.service_record_handle),
            ),
            ServiceAttribute(
                SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
                DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
            ),
            ServiceAttribute(
                SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
                DataElement.sequence(
                    [
                        DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
                        DataElement.uuid(core.BT_AV_REMOTE_CONTROL_CONTROLLER_SERVICE),
                    ]
                ),
            ),
            ServiceAttribute(
                SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                DataElement.sequence(
                    [
                        DataElement.sequence(
                            [
                                DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
                                DataElement.unsigned_integer_16(avctp.AVCTP_PSM),
                            ]
                        ),
                        DataElement.sequence(
                            [
                                DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
                                DataElement.unsigned_integer_16(avctp_version_int),
                            ]
                        ),
                    ]
                ),
            ),
            ServiceAttribute(
                SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                DataElement.sequence(
                    [
                        DataElement.sequence(
                            [
                                DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
                                DataElement.unsigned_integer_16(avrcp_version_int),
                            ]
                        ),
                    ]
                ),
            ),
            ServiceAttribute(
                SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
                DataElement.unsigned_integer_16(self.supported_features),
            ),
        ]
        if self.supported_features & ControllerFeatures.SUPPORTS_BROWSING:
            attributes.append(
                ServiceAttribute(
                    SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                    DataElement.sequence(
                        [
                            DataElement.sequence(
                                [
                                    DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
                                    DataElement.unsigned_integer_16(
                                        avctp.AVCTP_BROWSING_PSM
                                    ),
                                ]
                            ),
                            DataElement.sequence(
                                [
                                    DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
                                    DataElement.unsigned_integer_16(avctp_version_int),
                                ]
                            ),
                        ]
                    ),
                ),
            )
        return attributes

    @classmethod
    async def find(cls, connection: Connection) -> list[ControllerServiceSdpRecord]:
        async with sdp.Client(connection) as sdp_client:
            search_result = await sdp_client.search_attributes(
                uuids=[core.BT_AV_REMOTE_CONTROL_CONTROLLER_SERVICE],
                attribute_ids=[
                    SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
                    SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                    SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                    SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
                ],
            )

            records: list[ControllerServiceSdpRecord] = []
            for attribute_lists in search_result:
                record = cls(0)
                for attribute in attribute_lists:
                    if attribute.id == SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID:
                        record.service_record_handle = attribute.value.value
                    elif attribute.id == SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID:
                        # [[L2CAP, PSM], [AVCTP, version]]
                        record.avctp_version = (
                            attribute.value.value[1].value[1].value >> 8,
                            attribute.value.value[1].value[1].value & 0xFF,
                        )
                    elif (
                        attribute.id
                        == SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
                    ):
                        # [[AV_REMOTE_CONTROL, version]]
                        record.avrcp_version = (
                            attribute.value.value[0].value[1].value >> 8,
                            attribute.value.value[0].value[1].value & 0xFF,
                        )
                    elif attribute.id == SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID:
                        record.supported_features = ControllerFeatures(
                            attribute.value.value
                        )
                records.append(record)
            return records


# -----------------------------------------------------------------------------
@dataclass
class TargetServiceSdpRecord:
    service_record_handle: int
    avctp_version: tuple[int, int] = (1, 4)
    avrcp_version: tuple[int, int] = (1, 6)
    supported_features: int | TargetFeatures = TargetFeatures(0x23)

    def to_service_attributes(self) -> list[ServiceAttribute]:
        # TODO: support a way to compute the supported features from a feature list
        avctp_version_int = self.avctp_version[0] << 8 | self.avctp_version[1]
        avrcp_version_int = self.avrcp_version[0] << 8 | self.avrcp_version[1]

        attributes = [
            ServiceAttribute(
                SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
                DataElement.unsigned_integer_32(self.service_record_handle),
            ),
            ServiceAttribute(
                SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
                DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
            ),
            ServiceAttribute(
                SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
                DataElement.sequence(
                    [
                        DataElement.uuid(core.BT_AV_REMOTE_CONTROL_TARGET_SERVICE),
                    ]
                ),
            ),
            ServiceAttribute(
                SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                DataElement.sequence(
                    [
                        DataElement.sequence(
                            [
                                DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
                                DataElement.unsigned_integer_16(avctp.AVCTP_PSM),
                            ]
                        ),
                        DataElement.sequence(
                            [
                                DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
                                DataElement.unsigned_integer_16(avctp_version_int),
                            ]
                        ),
                    ]
                ),
            ),
            ServiceAttribute(
                SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                DataElement.sequence(
                    [
                        DataElement.sequence(
                            [
                                DataElement.uuid(core.BT_AV_REMOTE_CONTROL_SERVICE),
                                DataElement.unsigned_integer_16(avrcp_version_int),
                            ]
                        ),
                    ]
                ),
            ),
            ServiceAttribute(
                SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
                DataElement.unsigned_integer_16(self.supported_features),
            ),
        ]
        if self.supported_features & TargetFeatures.SUPPORTS_BROWSING:
            attributes.append(
                ServiceAttribute(
                    SDP_ADDITIONAL_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                    DataElement.sequence(
                        [
                            DataElement.sequence(
                                [
                                    DataElement.uuid(core.BT_L2CAP_PROTOCOL_ID),
                                    DataElement.unsigned_integer_16(
                                        avctp.AVCTP_BROWSING_PSM
                                    ),
                                ]
                            ),
                            DataElement.sequence(
                                [
                                    DataElement.uuid(core.BT_AVCTP_PROTOCOL_ID),
                                    DataElement.unsigned_integer_16(avctp_version_int),
                                ]
                            ),
                        ]
                    ),
                ),
            )
        return attributes

    @classmethod
    async def find(cls, connection: Connection) -> list[TargetServiceSdpRecord]:
        async with sdp.Client(connection) as sdp_client:
            search_result = await sdp_client.search_attributes(
                uuids=[core.BT_AV_REMOTE_CONTROL_TARGET_SERVICE],
                attribute_ids=[
                    SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
                    SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                    SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
                    SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID,
                ],
            )

            records: list[TargetServiceSdpRecord] = []
            for attribute_lists in search_result:
                record = cls(0)
                for attribute in attribute_lists:
                    if attribute.id == SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID:
                        record.service_record_handle = attribute.value.value
                    elif attribute.id == SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID:
                        # [[L2CAP, PSM], [AVCTP, version]]
                        record.avctp_version = (
                            attribute.value.value[1].value[1].value >> 8,
                            attribute.value.value[1].value[1].value & 0xFF,
                        )
                    elif (
                        attribute.id
                        == SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
                    ):
                        # [[AV_REMOTE_CONTROL, version]]
                        record.avrcp_version = (
                            attribute.value.value[0].value[1].value >> 8,
                            attribute.value.value[0].value[1].value & 0xFF,
                        )
                    elif attribute.id == SDP_SUPPORTED_FEATURES_ATTRIBUTE_ID:
                        record.supported_features = TargetFeatures(
                            attribute.value.value
                        )
                records.append(record)
            return records


# -----------------------------------------------------------------------------
def _parse_string(data: bytes, offset: int, length_size: int) -> tuple[int, str]:
    length = int.from_bytes(
        data[offset : offset + length_size], byteorder='big', signed=False
    )
    offset += length_size
    encoded = data[offset : offset + length]
    try:
        decoded = encoded.decode("utf-8")
    except UnicodeDecodeError:
        # This can decode anything.
        decoded = encoded.decode("latin1")
    return offset + length, decoded


# -----------------------------------------------------------------------------
def _serialize_string(value: str, length_size: int) -> bytes:
    encoded = value.encode("utf-8")
    return len(encoded).to_bytes(length_size, byteorder='big', signed=False) + encoded


# -----------------------------------------------------------------------------
def _string_spec(length_size: int):
    return {
        'parser': functools.partial(_parse_string, length_size=length_size),
        'serializer': functools.partial(_serialize_string, length_size=length_size),
    }


# -----------------------------------------------------------------------------
@dataclass
class MediaAttribute(hci.HCI_Dataclass_Object):
    attribute_id: MediaAttributeId = field(
        metadata=MediaAttributeId.type_metadata(4, byteorder='big')
    )
    character_set_id: CharacterSetId = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    attribute_value: str = field(metadata=hci.metadata(_string_spec(2)))


# -----------------------------------------------------------------------------
@dataclass
class SongAndPlayStatus:
    song_length: int
    song_position: int
    play_status: PlayStatus


# -----------------------------------------------------------------------------
class ApplicationSetting:
    class AttributeId(hci.SpecableEnum):
        EQUALIZER_ON_OFF = 0x01
        REPEAT_MODE = 0x02
        SHUFFLE_ON_OFF = 0x03
        SCAN_ON_OFF = 0x04

    class EqualizerOnOffStatus(hci.SpecableEnum):
        OFF = 0x01
        ON = 0x02

    class RepeatModeStatus(hci.SpecableEnum):
        OFF = 0x01
        SINGLE_TRACK_REPEAT = 0x02
        ALL_TRACK_REPEAT = 0x03
        GROUP_REPEAT = 0x04

    class ShuffleOnOffStatus(hci.SpecableEnum):
        OFF = 0x01
        ALL_TRACKS_SHUFFLE = 0x02
        GROUP_SHUFFLE = 0x03

    class ScanOnOffStatus(hci.SpecableEnum):
        OFF = 0x01
        ALL_TRACKS_SCAN = 0x02
        GROUP_SCAN = 0x03

    class GenericValue(hci.SpecableEnum):
        pass


# -----------------------------------------------------------------------------
@dataclass
class AttributeValueEntry(hci.HCI_Dataclass_Object):
    attribute_id: MediaAttributeId = field(
        metadata=MediaAttributeId.type_metadata(4, byteorder='big')
    )
    character_set_id: CharacterSetId = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    attribute_value: str = field(metadata=hci.metadata(_string_spec(2)))


# -----------------------------------------------------------------------------
class BrowseableItem:
    """6.10.2 Browseable items."""

    class Type(hci.SpecableEnum):
        MEDIA_PLAYER = 0x01
        FOLDER = 0x02
        MEDIA_ELEMENT = 0x03

    item_type: ClassVar[Type]
    _payload: bytes | None = None

    subclasses: ClassVar[dict[Type, type[BrowseableItem]]] = {}
    fields: ClassVar[hci.Fields] = ()

    @classmethod
    def parse_from_bytes(cls, data: bytes, offset: int) -> tuple[int, BrowseableItem]:
        item_type, length = struct.unpack_from('>BH', data, offset)
        subclass = cls.subclasses[BrowseableItem.Type(item_type)]
        instance = subclass(
            **hci.HCI_Object.dict_from_bytes(data, offset + 3, subclass.fields)
        )
        instance._payload = data[3:]
        return offset + length + 3, instance

    def __bytes__(self) -> bytes:
        if self._payload is None:
            self._payload = hci.HCI_Object.dict_to_bytes(self.__dict__, self.fields)
        return struct.pack('>BH', self.item_type, len(self._payload)) + self._payload

    _Item = TypeVar('_Item', bound='BrowseableItem')

    @classmethod
    def item(cls, subclass: type[_Item]) -> type[_Item]:
        cls.subclasses[subclass.item_type] = subclass
        subclass.fields = hci.HCI_Object.fields_from_dataclass(subclass)
        return subclass


# -----------------------------------------------------------------------------
@BrowseableItem.item
@dataclass
class MediaPlayerItem(BrowseableItem):
    item_type = BrowseableItem.Type.MEDIA_PLAYER

    class MajorPlayerType(hci.SpecableFlag):
        AUDIO = 0x01
        VIDEO = 0x02
        BROADCASTING_AUDIO = 0x04
        BROADCASTING_VIDEO = 0x08

    class PlayerSubType(hci.SpecableFlag):
        AUDIO_BOOK = 0x01
        PODCAST = 0x02

    class Features(hci.SpecableFlag):
        SELECT = 1 << 0
        UP = 1 << 1
        DOWN = 1 << 2
        LEFT = 1 << 3
        RIGHT = 1 << 4
        RIGHT_UP = 1 << 5
        RIGHT_DOWN = 1 << 6
        LEFT_UP = 1 << 7
        LEFT_DOWN = 1 << 8
        ROOT_MENU = 1 << 9
        SETUP_MENU = 1 << 10
        CONTENTS_MENU = 1 << 11
        FAVORITE_MENU = 1 << 12
        EXIT = 1 << 13
        NUM_0 = 1 << 14
        NUM_1 = 1 << 15
        NUM_2 = 1 << 16
        NUM_3 = 1 << 17
        NUM_4 = 1 << 18
        NUM_5 = 1 << 19
        NUM_6 = 1 << 20
        NUM_7 = 1 << 21
        NUM_8 = 1 << 22
        NUM_9 = 1 << 23
        DOT = 1 << 24
        ENTER = 1 << 25
        CLEAR = 1 << 26
        CHANNEL_UP = 1 << 27
        CHANNEL_DOWN = 1 << 28
        PREVIOUS_CHANNEL = 1 << 29
        SOUND_SELECT = 1 << 30
        INPUT_SELECT = 1 << 31
        DISPLAY_INFORMATION = 1 << 32
        HELP = 1 << 33
        PAGE_UP = 1 << 34
        PAGE_DOWN = 1 << 35
        POWER = 1 << 36
        VOLUME_UP = 1 << 37
        VOLUME_DOWN = 1 << 38
        MUTE = 1 << 39
        PLAY = 1 << 40
        STOP = 1 << 41
        PAUSE = 1 << 42
        RECORD = 1 << 43
        REWIND = 1 << 44
        FAST_FORWARD = 1 << 45
        EJECT = 1 << 46
        FORWARD = 1 << 47
        BACKWARD = 1 << 48
        ANGLE = 1 << 49
        SUBPICTURE = 1 << 50
        F1 = 1 << 51
        F2 = 1 << 52
        F3 = 1 << 53
        F4 = 1 << 54
        F5 = 1 << 55
        VENDOR_UNIQUE = 1 << 56
        BASIC_GROUP_NAVIGATION = 1 << 57
        ADVANCED_CONTROL_PLAYER = 1 << 58
        BROWSING = 1 << 59
        SEARCHING = 1 << 60
        ADD_TO_NOW_PLAYING = 1 << 61
        UI_DS_UNIQUE_IN_PLAYER_BROWSE_TREE = 1 << 62
        ONLY_BROWSABLE_WHEN_ADDRESSED = 1 << 63
        ONLY_SEARCHABLE_WHEN_ADDRESSED = 1 << 64
        NOW_PLAYING = 1 << 65
        UID_PERSISTENCY = 1 << 66
        NUMBER_OF_ITEMS = 1 << 67
        COVER_ART = 1 << 68

    player_id: int = field(metadata=hci.metadata('>2'))
    major_player_type: MajorPlayerType = field(
        metadata=MajorPlayerType.type_metadata(1)
    )
    player_sub_type: PlayerSubType = field(
        metadata=PlayerSubType.type_metadata(4, byteorder='little')
    )
    play_status: PlayStatus = field(metadata=PlayStatus.type_metadata(1))
    feature_bitmask: Features = field(
        metadata=Features.type_metadata(16, byteorder='little')
    )
    character_set_id: CharacterSetId = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    displayable_name: str = field(metadata=hci.metadata(_string_spec(2)))


# -----------------------------------------------------------------------------
@BrowseableItem.item
@dataclass
class FolderItem(BrowseableItem):
    item_type = BrowseableItem.Type.FOLDER

    class FolderType(hci.SpecableEnum):
        MIXED = 0x00
        TITLES = 0x01
        ALBUMS = 0x02
        ARTISTS = 0x03
        GENRES = 0x04
        PLAYLISTS = 0x05
        YEARS = 0x06

    class Playable(hci.SpecableEnum):
        NOT_PLAYABLE = 0x00
        PLAYABLE = 0x01

    folder_uid: int = field(metadata=_UINT64_BE_METADATA)
    folder_type: FolderType = field(metadata=FolderType.type_metadata(1))
    is_playable: Playable = field(metadata=Playable.type_metadata(1))
    character_set_id: CharacterSetId = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    displayable_name: str = field(metadata=hci.metadata(_string_spec(2)))


# -----------------------------------------------------------------------------
@BrowseableItem.item
@dataclass
class MediaElementItem(BrowseableItem):
    item_type = BrowseableItem.Type.MEDIA_ELEMENT

    class MediaType(hci.SpecableEnum):
        AUDIO = 0x00
        VIDEO = 0x01

    media_element_uid: int = field(metadata=_UINT64_BE_METADATA)
    media_type: MediaType = field(metadata=MediaType.type_metadata(1))
    character_set_id: CharacterSetId = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    displayable_name: str = field(metadata=hci.metadata(_string_spec(2)))
    attribute_value_entry_list: Sequence[AttributeValueEntry] = field(
        metadata=hci.metadata(
            AttributeValueEntry.parse_from_bytes, list_begin=True, list_end=True
        )
    )


# -----------------------------------------------------------------------------
class PduAssembler:
    """
    PDU Assembler to support fragmented PDUs are defined in:
    Audio/Video Remote Control / Profile Specification
    6.3.1 AVRCP specific AV//C commands
    """

    pdu_id: PduId | None
    payload: bytes

    def __init__(self, callback: Callable[[PduId, bytes], None]) -> None:
        self.callback = callback
        self.reset()

    def reset(self) -> None:
        self.pdu_id = None
        self.parameter = b''

    def on_pdu(self, pdu: bytes) -> None:
        pdu_id = PduId(pdu[0])
        packet_type = Protocol.PacketType(pdu[1] & 3)
        parameter_length = struct.unpack_from('>H', pdu, 2)[0]
        parameter = pdu[4 : 4 + parameter_length]
        if len(parameter) != parameter_length:
            logger.warning("parameter length exceeds pdu size")
            self.reset()
            return

        if packet_type in (Protocol.PacketType.SINGLE, Protocol.PacketType.START):
            if self.pdu_id is not None:
                # We are already in a PDU
                logger.warning("received START or SINGLE fragment while in pdu")
                self.reset()

        if packet_type in (Protocol.PacketType.CONTINUE, Protocol.PacketType.END):
            if pdu_id != self.pdu_id:
                logger.warning("PID does not match")
                self.reset()
                return
        else:
            self.pdu_id = pdu_id

        self.parameter += parameter

        if packet_type in (Protocol.PacketType.SINGLE, Protocol.PacketType.END):
            self.on_pdu_complete()

    def on_pdu_complete(self) -> None:
        assert self.pdu_id is not None
        try:
            self.callback(self.pdu_id, self.parameter)
        except Exception:
            logger.exception(color('!!! exception in callback', 'red'))

        self.reset()


# -----------------------------------------------------------------------------
class Command:
    pdu_id: ClassVar[PduId]
    _payload: bytes | None = None

    _Command = TypeVar('_Command', bound='Command')
    subclasses: ClassVar[dict[int, type[Command]]] = {}
    fields: ClassVar[hci.Fields] = ()

    @classmethod
    def command(cls, subclass: type[_Command]) -> type[_Command]:
        cls.subclasses[subclass.pdu_id] = subclass
        subclass.fields = hci.HCI_Object.fields_from_dataclass(subclass)
        return subclass

    @classmethod
    def from_bytes(cls, pdu_id: int, pdu: bytes) -> Command:
        if not (subclass := cls.subclasses.get(pdu_id)):
            raise core.InvalidPacketError(f"Unimplemented PDU {pdu_id}")
        instance = subclass(**hci.HCI_Object.dict_from_bytes(pdu, 0, subclass.fields))
        instance._payload = pdu[0:]
        return instance

    def __bytes__(self) -> bytes:
        if self._payload is None:
            self._payload = hci.HCI_Object.dict_to_bytes(self.__dict__, self.fields)
        return self._payload


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetCapabilitiesCommand(Command):
    pdu_id = PduId.GET_CAPABILITIES

    class CapabilityId(hci.SpecableEnum):
        COMPANY_ID = 0x02
        EVENTS_SUPPORTED = 0x03

    capability_id: CapabilityId = field(metadata=CapabilityId.type_metadata(1))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class ListPlayerApplicationSettingAttributesCommand(Command):
    pdu_id = PduId.LIST_PLAYER_APPLICATION_SETTING_ATTRIBUTES


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class ListPlayerApplicationSettingValuesCommand(Command):
    pdu_id = PduId.LIST_PLAYER_APPLICATION_SETTING_VALUES

    attribute: ApplicationSetting.AttributeId = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(1)
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetCurrentPlayerApplicationSettingValueCommand(Command):
    pdu_id = PduId.GET_CURRENT_PLAYER_APPLICATION_SETTING_VALUE

    attribute: Sequence[ApplicationSetting.AttributeId] = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(
            1, list_begin=True, list_end=True
        )
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class SetPlayerApplicationSettingValueCommand(Command):
    pdu_id = PduId.SET_PLAYER_APPLICATION_SETTING_VALUE

    attribute: Sequence[ApplicationSetting.AttributeId] = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(1, list_begin=True)
    )
    value: Sequence[int] = field(metadata=hci.metadata(1, list_end=True))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetPlayerApplicationSettingAttributeTextCommand(Command):
    pdu_id = PduId.GET_PLAYER_APPLICATION_SETTING_ATTRIBUTE_TEXT

    attribute: Sequence[ApplicationSetting.AttributeId] = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(
            1, list_begin=True, list_end=True
        )
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetPlayerApplicationSettingValueTextCommand(Command):
    pdu_id = PduId.GET_PLAYER_APPLICATION_SETTING_VALUE_TEXT

    attribute: ApplicationSetting.AttributeId = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(1)
    )
    value: Sequence[int] = field(
        metadata=hci.metadata(1, list_begin=True, list_end=True)
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class InformDisplayableCharacterSetCommand(Command):
    pdu_id = PduId.INFORM_DISPLAYABLE_CHARACTER_SET

    character_set_id: Sequence[CharacterSetId] = field(
        metadata=CharacterSetId.type_metadata(
            2, list_begin=True, list_end=True, byteorder='big'
        )
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class InformBatteryStatusOfCtCommand(Command):
    pdu_id = PduId.INFORM_BATTERY_STATUS_OF_CT

    class BatteryStatus(hci.SpecableEnum):
        NORMAL = 0x00
        WARNING = 0x01
        CRITICAL = 0x02
        EXTERNAL = 0x03
        FULL_CHARGE = 0x04

    battery_status: BatteryStatus = field(metadata=BatteryStatus.type_metadata(1))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetPlayStatusCommand(Command):
    pdu_id = PduId.GET_PLAY_STATUS


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetElementAttributesCommand(Command):
    pdu_id = PduId.GET_ELEMENT_ATTRIBUTES

    identifier: int = field(metadata=_UINT64_BE_METADATA)
    attribute_ids: Sequence[MediaAttributeId] = field(
        metadata=MediaAttributeId.type_metadata(
            4, list_begin=True, list_end=True, byteorder='big'
        )
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class SetAbsoluteVolumeCommand(Command):
    pdu_id = PduId.SET_ABSOLUTE_VOLUME
    MAXIMUM_VOLUME = 0x7F

    volume: int = field(metadata=hci.metadata(1))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class RegisterNotificationCommand(Command):
    pdu_id = PduId.REGISTER_NOTIFICATION

    event_id: EventId = field(metadata=EventId.type_metadata(1))
    playback_interval: int = field(metadata=hci.metadata('>4'))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class SetAddressedPlayerCommand(Command):
    pdu_id = PduId.SET_ADDRESSED_PLAYER

    player_id: int = field(metadata=hci.metadata('>2'))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class SetBrowsedPlayerCommand(Command):
    pdu_id = PduId.SET_BROWSED_PLAYER

    player_id: int = field(metadata=hci.metadata('>2'))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetFolderItemsCommand(Command):
    pdu_id = PduId.GET_FOLDER_ITEMS

    scope: Scope = field(metadata=Scope.type_metadata(1))
    start_item: int = field(metadata=hci.metadata('>4'))
    end_item: int = field(metadata=hci.metadata('>4'))
    # When attributes is empty, all attributes will be requested.
    attributes: Sequence[MediaAttributeId] = field(
        metadata=MediaAttributeId.type_metadata(
            4, list_begin=True, list_end=True, byteorder='big'
        )
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class ChangePathCommand(Command):
    pdu_id = PduId.CHANGE_PATH

    class Direction(hci.SpecableEnum):
        UP = 0
        DOWN = 1

    uid_counter: int = field(metadata=hci.metadata('>2'))
    direction: Direction = field(metadata=Direction.type_metadata(1))
    folder_uid: int = field(metadata=_UINT64_BE_METADATA)


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetItemAttributesCommand(Command):
    pdu_id = PduId.GET_ITEM_ATTRIBUTES

    scope: Scope = field(metadata=Scope.type_metadata(1))
    uid: int = field(metadata=_UINT64_BE_METADATA)
    uid_counter: int = field(metadata=hci.metadata('>2'))
    start_item: int = field(metadata=hci.metadata('>4'))
    end_item: int = field(metadata=hci.metadata('>4'))
    # When attributes is empty, all attributes will be requested.
    attributes: Sequence[MediaAttributeId] = field(
        metadata=MediaAttributeId.type_metadata(1, list_begin=True, list_end=True)
    )


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class GetTotalNumberOfItemsCommand(Command):
    pdu_id = PduId.GET_TOTAL_NUMBER_OF_ITEMS

    scope: Scope = field(metadata=Scope.type_metadata(1))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class SearchCommand(Command):
    pdu_id = PduId.SEARCH

    character_set_id: CharacterSetId = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    search_string: str = field(metadata=hci.metadata(_string_spec(2)))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class PlayItemCommand(Command):
    pdu_id = PduId.PLAY_ITEM

    scope: Scope = field(metadata=Scope.type_metadata(1))
    uid: int = field(metadata=_UINT64_BE_METADATA)
    uid_counter: int = field(metadata=hci.metadata('>2'))


# -----------------------------------------------------------------------------
@Command.command
@dataclass
class AddToNowPlayingCommand(Command):
    pdu_id = PduId.ADD_TO_NOW_PLAYING

    scope: Scope = field(metadata=Scope.type_metadata(1))
    uid: int = field(metadata=_UINT64_BE_METADATA)
    uid_counter: int = field(metadata=hci.metadata('>2'))


# -----------------------------------------------------------------------------
class Response:
    pdu_id: PduId
    _payload: bytes | None = None

    fields: ClassVar[hci.Fields] = ()
    subclasses: ClassVar[dict[PduId, type[Response]]] = {}

    _Response = TypeVar('_Response', bound='Response')

    @classmethod
    def response(cls, subclass: type[_Response]) -> type[_Response]:
        subclass.fields = hci.HCI_Object.fields_from_dataclass(subclass)
        if pdu_id := getattr(subclass, 'pdu_id', None):
            cls.subclasses[pdu_id] = subclass
        return subclass

    def __bytes__(self) -> bytes:
        if self._payload is None:
            self._payload = hci.HCI_Object.dict_to_bytes(self.__dict__, self.fields)
        return self._payload

    @classmethod
    def from_bytes(cls, pdu: bytes, pdu_id: PduId) -> Response:
        if not (subclass := cls.subclasses.get(pdu_id)):
            raise core.InvalidArgumentError(f"Unimplemented packet {pdu_id.name}")
        return subclass.from_parameters(pdu)

    @classmethod
    def from_parameters(cls, parameters: bytes) -> Response:
        instance = cls(**hci.HCI_Object.dict_from_bytes(parameters, 0, cls.fields))
        instance._payload = parameters
        return instance


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class RejectedResponse(Response):
    pdu_id: PduId
    status_code: StatusCode = field(metadata=StatusCode.type_metadata(1))

    @classmethod
    def from_bytes(cls, pdu: bytes, pdu_id: PduId) -> Response:
        return cls(pdu_id=pdu_id, status_code=StatusCode(pdu[0]))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class NotImplementedResponse(Response):
    pdu_id: PduId
    parameters: bytes = field(metadata=hci.metadata('*'))

    @classmethod
    def from_bytes(cls, pdu: bytes, pdu_id: PduId) -> Response:
        return cls(pdu_id=pdu_id, parameters=pdu)


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetCapabilitiesResponse(Response):
    pdu_id = PduId.GET_CAPABILITIES
    capability_id: GetCapabilitiesCommand.CapabilityId
    capabilities: Sequence[SupportsBytes | bytes]

    @classmethod
    def from_parameters(cls, parameters: bytes) -> Response:
        if len(parameters) < 2:
            # Possibly a reject response.
            return cls(GetCapabilitiesCommand.CapabilityId(0), [])

        # Assume that the payloads all follow the same pattern:
        #  <CapabilityID><CapabilityCount><Capability*>
        capability_id = GetCapabilitiesCommand.CapabilityId(parameters[0])
        capability_count = parameters[1]

        capabilities: list[SupportsBytes | bytes]
        if capability_id == GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED:
            capabilities = [EventId(parameters[2 + x]) for x in range(capability_count)]
        else:
            capability_size = (len(parameters) - 2) // capability_count
            capabilities = [
                parameters[x : x + capability_size]
                for x in range(2, len(parameters), capability_size)
            ]

        return cls(capability_id, capabilities)

    def __post_init__(self) -> None:
        self._payload = bytes([self.capability_id, len(self.capabilities)]) + b''.join(
            bytes(capability) for capability in self.capabilities
        )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class ListPlayerApplicationSettingAttributesResponse(Response):
    pdu_id = PduId.LIST_PLAYER_APPLICATION_SETTING_ATTRIBUTES

    attribute: Sequence[ApplicationSetting.AttributeId] = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(
            1, list_begin=True, list_end=True
        )
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class ListPlayerApplicationSettingValuesResponse(Response):
    pdu_id = PduId.LIST_PLAYER_APPLICATION_SETTING_VALUES

    value: Sequence[int] = field(
        metadata=hci.metadata(1, list_begin=True, list_end=True)
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetCurrentPlayerApplicationSettingValueResponse(Response):
    pdu_id = PduId.GET_CURRENT_PLAYER_APPLICATION_SETTING_VALUE

    attribute: Sequence[ApplicationSetting.AttributeId] = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(1, list_begin=True)
    )
    value: Sequence[int] = field(metadata=hci.metadata(1, list_end=True))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class SetPlayerApplicationSettingValueResponse(Response):
    pdu_id = PduId.SET_PLAYER_APPLICATION_SETTING_VALUE


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetPlayerApplicationSettingAttributeTextResponse(Response):
    pdu_id = PduId.GET_PLAYER_APPLICATION_SETTING_ATTRIBUTE_TEXT

    attribute: Sequence[ApplicationSetting.AttributeId] = field(
        metadata=ApplicationSetting.AttributeId.type_metadata(1, list_begin=True)
    )
    character_set_id: Sequence[CharacterSetId] = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    attribute_string: Sequence[str] = field(
        metadata=hci.metadata(_string_spec(1), list_end=True)
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetPlayerApplicationSettingValueTextResponse(Response):
    pdu_id = PduId.GET_PLAYER_APPLICATION_SETTING_VALUE_TEXT

    value: Sequence[int] = field(metadata=hci.metadata(1, list_begin=True))
    character_set_id: Sequence[CharacterSetId] = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    attribute_string: Sequence[str] = field(
        metadata=hci.metadata(_string_spec(1), list_end=True)
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class InformDisplayableCharacterSetResponse(Response):
    pdu_id = PduId.INFORM_DISPLAYABLE_CHARACTER_SET


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class InformBatteryStatusOfCtResponse(Response):
    pdu_id = PduId.INFORM_BATTERY_STATUS_OF_CT


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetPlayStatusResponse(Response):
    pdu_id = PduId.GET_PLAY_STATUS

    # TG doesn't support Song Length or Position.
    UNAVAILABLE = 0xFFFFFFFF

    song_length: int = field(metadata=hci.metadata(">4"))
    song_position: int = field(metadata=hci.metadata(">4"))
    play_status: PlayStatus = field(metadata=PlayStatus.type_metadata(1))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetElementAttributesResponse(Response):
    pdu_id = PduId.GET_ELEMENT_ATTRIBUTES
    attributes: Sequence[MediaAttribute] = field(
        metadata=hci.metadata(
            MediaAttribute.parse_from_bytes, list_begin=True, list_end=True
        )
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class SetAbsoluteVolumeResponse(Response):
    pdu_id = PduId.SET_ABSOLUTE_VOLUME
    volume: int = field(metadata=hci.metadata(1))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class RegisterNotificationResponse(Response):
    pdu_id = PduId.REGISTER_NOTIFICATION
    event: Event = field(
        metadata=hci.metadata(
            lambda data, offset: (len(data), Event.from_bytes(data[offset:]))
        )
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class SetAddressedPlayerResponse(Response):
    pdu_id = PduId.SET_ADDRESSED_PLAYER

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class SetBrowsedPlayerResponse(Response):
    pdu_id = PduId.SET_BROWSED_PLAYER

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))
    uid_counter: int = field(metadata=hci.metadata('>2'))
    numbers_of_items: int = field(metadata=hci.metadata('>4'))
    character_set_id: CharacterSetId = field(
        metadata=CharacterSetId.type_metadata(2, byteorder='big')
    )
    folder_names: Sequence[str] = field(
        metadata=hci.metadata(_string_spec(2), list_begin=True, list_end=True)
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetFolderItemsResponse(Response):
    pdu_id = PduId.GET_FOLDER_ITEMS

    status: StatusCode
    uid_counter: int
    items: Sequence[BrowseableItem]

    @classmethod
    def from_parameters(cls, parameters: bytes) -> Response:
        status, uid_counter, count = struct.unpack_from('>BHH', parameters)
        items: list[BrowseableItem] = []
        offset = 5
        for _ in range(count):
            offset, item = BrowseableItem.parse_from_bytes(parameters, offset)
            items.append(item)
        instance = cls(status=StatusCode(status), uid_counter=uid_counter, items=items)
        instance._payload = parameters
        return instance

    def __post_init__(self) -> None:
        if self._payload is None:
            self._payload = struct.pack(
                '>BHH', self.status, self.uid_counter, len(self.items)
            ) + b''.join(map(bytes, self.items))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class ChangePathResponse(Response):
    pdu_id = PduId.CHANGE_PATH

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))
    number_of_items: int = field(metadata=hci.metadata('>4'))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetItemAttributesResponse(Response):
    pdu_id = PduId.GET_ITEM_ATTRIBUTES

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))
    attribute_value_entry_list: Sequence[AttributeValueEntry] = field(
        metadata=hci.metadata(
            AttributeValueEntry.parse_from_bytes, list_begin=True, list_end=True
        )
    )


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class GetTotalNumberOfItemsResponse(Response):
    pdu_id = PduId.GET_TOTAL_NUMBER_OF_ITEMS

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))
    uid_counter: int = field(metadata=hci.metadata('>2'))
    number_of_items: int = field(metadata=hci.metadata('>4'))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class SearchResponse(Response):
    pdu_id = PduId.SEARCH

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))
    uid_counter: int = field(metadata=hci.metadata('>2'))
    number_of_items: int = field(metadata=hci.metadata('>4'))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class PlayItemResponse(Response):
    pdu_id = PduId.PLAY_ITEM

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))


# -----------------------------------------------------------------------------
@Response.response
@dataclass
class AddToNowPlayingResponse(Response):
    pdu_id = PduId.ADD_TO_NOW_PLAYING

    status: StatusCode = field(metadata=StatusCode.type_metadata(1))


# -----------------------------------------------------------------------------
class Event:
    event_id: EventId
    _pdu: bytes | None = None

    _Event = TypeVar('_Event', bound='Event')
    subclasses: ClassVar[dict[int, type[Event]]] = {}
    fields: ClassVar[hci.Fields] = ()

    @classmethod
    def event(cls, subclass: type[_Event]) -> type[_Event]:
        cls.subclasses[subclass.event_id] = subclass
        subclass.fields = hci.HCI_Object.fields_from_dataclass(subclass)
        return subclass

    @classmethod
    def from_bytes(cls, pdu: bytes) -> Event:
        if not (subclass := cls.subclasses.get(pdu[0])):
            raise core.InvalidPacketError(f"Unimplemented Event {pdu[0]}")
        instance = subclass(**hci.HCI_Object.dict_from_bytes(pdu, 1, subclass.fields))
        instance._pdu = pdu
        return instance

    def __bytes__(self) -> bytes:
        if self._pdu is None:
            self._pdu = bytes([self.event_id]) + hci.HCI_Object.dict_to_bytes(
                self.__dict__, self.fields
            )
        return self._pdu


# -----------------------------------------------------------------------------
@dataclass
class GenericEvent(Event):
    event_id: EventId = field(metadata=EventId.type_metadata(1))
    data: bytes = field(metadata=hci.metadata('*'))


GenericEvent.fields = hci.HCI_Object.fields_from_dataclass(GenericEvent)


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class PlaybackStatusChangedEvent(Event):
    event_id = EventId.PLAYBACK_STATUS_CHANGED
    play_status: PlayStatus = field(metadata=PlayStatus.type_metadata(1))


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class PlaybackPositionChangedEvent(Event):
    event_id = EventId.PLAYBACK_POS_CHANGED
    playback_position: int = field(metadata=hci.metadata('>4'))


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class TrackChangedEvent(Event):
    event_id = EventId.TRACK_CHANGED
    identifier: bytes = field(metadata=hci.metadata('*'))


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class PlayerApplicationSettingChangedEvent(Event):
    event_id = EventId.PLAYER_APPLICATION_SETTING_CHANGED

    @dataclass
    class Setting(hci.HCI_Dataclass_Object):
        attribute_id: ApplicationSetting.AttributeId = field(
            metadata=ApplicationSetting.AttributeId.type_metadata(1)
        )
        value_id: (
            ApplicationSetting.EqualizerOnOffStatus
            | ApplicationSetting.RepeatModeStatus
            | ApplicationSetting.ShuffleOnOffStatus
            | ApplicationSetting.ScanOnOffStatus
            | ApplicationSetting.GenericValue
        ) = field(metadata=hci.metadata(1))

        def __post_init__(self) -> None:
            super().__post_init__()
            if self.attribute_id == ApplicationSetting.AttributeId.EQUALIZER_ON_OFF:
                self.value_id = ApplicationSetting.EqualizerOnOffStatus(self.value_id)
            elif self.attribute_id == ApplicationSetting.AttributeId.REPEAT_MODE:
                self.value_id = ApplicationSetting.RepeatModeStatus(self.value_id)
            elif self.attribute_id == ApplicationSetting.AttributeId.SHUFFLE_ON_OFF:
                self.value_id = ApplicationSetting.ShuffleOnOffStatus(self.value_id)
            elif self.attribute_id == ApplicationSetting.AttributeId.SCAN_ON_OFF:
                self.value_id = ApplicationSetting.ScanOnOffStatus(self.value_id)
            else:
                self.value_id = ApplicationSetting.GenericValue(self.value_id)

    player_application_settings: Sequence[Setting] = field(
        metadata=hci.metadata(Setting.parse_from_bytes, list_begin=True, list_end=True)
    )


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class NowPlayingContentChangedEvent(Event):
    event_id = EventId.NOW_PLAYING_CONTENT_CHANGED


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class AvailablePlayersChangedEvent(Event):
    event_id = EventId.AVAILABLE_PLAYERS_CHANGED


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class AddressedPlayerChangedEvent(Event):
    event_id = EventId.ADDRESSED_PLAYER_CHANGED

    @dataclass
    class Player(hci.HCI_Dataclass_Object):
        player_id: int = field(metadata=hci.metadata('>2'))
        uid_counter: int = field(metadata=hci.metadata('>2'))

    player: Player = field(metadata=hci.metadata(Player.parse_from_bytes))


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class UidsChangedEvent(Event):
    event_id = EventId.UIDS_CHANGED
    uid_counter: int = field(metadata=hci.metadata('>2'))


# -----------------------------------------------------------------------------
@Event.event
@dataclass
class VolumeChangedEvent(Event):
    event_id = EventId.VOLUME_CHANGED
    volume: int = field(metadata=hci.metadata(1))


# -----------------------------------------------------------------------------
class Delegate:
    """
    Base class for AVRCP delegates.

    All the methods are async, even if they don't always need to be, so that
    delegates that do need to wait for an async result may do so.
    """

    class Error(Exception):
        """The delegate method failed, with a specified status code."""

        def __init__(self, status_code: StatusCode) -> None:
            self.status_code = status_code

    class AvcError(Exception):
        """The delegate AVC method failed, with a specified status code."""

        def __init__(self, status_code: avc.ResponseFrame.ResponseCode) -> None:
            self.status_code = status_code

    supported_events: list[EventId]
    supported_company_ids: list[int]
    volume: int
    playback_status: PlayStatus

    def __init__(
        self,
        supported_events: Iterable[EventId] = (),
        supported_company_ids: Iterable[int] = (AVRCP_BLUETOOTH_SIG_COMPANY_ID,),
    ) -> None:
        self.supported_company_ids = list(supported_company_ids)
        self.supported_events = list(supported_events)
        self.volume = 0
        self.playback_status = PlayStatus.STOPPED

    async def get_supported_events(self) -> list[EventId]:
        return self.supported_events

    async def get_supported_company_ids(self) -> list[int]:
        return self.supported_company_ids

    async def set_absolute_volume(self, volume: int) -> None:
        """
        Set the absolute volume.

        Returns: the effective volume that was set.
        """
        logger.debug(f"@@@ set_absolute_volume: volume={volume}")
        self.volume = volume

    async def get_absolute_volume(self) -> int:
        return self.volume

    async def on_key_event(
        self,
        key: avc.PassThroughFrame.OperationId,
        pressed: bool,
        data: bytes,
    ) -> None:
        logger.debug(
            "@@@ on_key_event: key=%s, pressed=%s, data=%s", key, pressed, data.hex()
        )

    async def get_playback_status(self) -> PlayStatus:
        return self.playback_status

    # TODO add other delegate methods


# -----------------------------------------------------------------------------
class Protocol(utils.EventEmitter):
    """AVRCP Controller and Target protocol."""

    EVENT_CONNECTION = "connection"
    EVENT_START = "start"
    EVENT_STOP = "stop"

    class PacketType(enum.IntEnum):
        SINGLE = 0b00
        START = 0b01
        CONTINUE = 0b10
        END = 0b11

    class InvalidPidError(Exception):
        """A response frame with ipid==1 was received."""

    class NotPendingError(Exception):
        """There is no pending command for a transaction label."""

    class MismatchedResponseError(Exception):
        """The response type does not corresponding to the request type."""

        def __init__(self, response: Response) -> None:
            self.response = response

    class UnexpectedResponseTypeError(Exception):
        """The response type is not the expected one."""

        def __init__(self, response: Protocol.ResponseContext) -> None:
            self.response = response

    class UnexpectedResponseCodeError(Exception):
        """The response code was not the expected one."""

        def __init__(
            self, response_code: avc.ResponseFrame.ResponseCode, response: Response
        ) -> None:
            self.response_code = response_code
            self.response = response

    class PendingCommand:
        response: asyncio.Future

        def __init__(self, transaction_label: int) -> None:
            self.transaction_label = transaction_label
            self.reset()

        def reset(self):
            self.response = asyncio.get_running_loop().create_future()

    @dataclass
    class ReceiveCommandState:
        transaction_label: int
        command_type: avc.CommandFrame.CommandType

    @dataclass
    class ReceiveResponseState:
        transaction_label: int
        response_code: avc.ResponseFrame.ResponseCode

    @dataclass
    class ResponseContext:
        transaction_label: int
        response: Response

    @dataclass
    class FinalResponse(ResponseContext):
        response_code: avc.ResponseFrame.ResponseCode

    @dataclass
    class InterimResponse(ResponseContext):
        final: Awaitable[Protocol.FinalResponse]

    @dataclass
    class NotificationListener:
        transaction_label: int
        register_notification_command: RegisterNotificationCommand

    delegate: Delegate
    send_transaction_label: int
    command_pdu_assembler: PduAssembler
    receive_command_state: ReceiveCommandState | None
    response_pdu_assembler: PduAssembler
    receive_response_state: ReceiveResponseState | None
    avctp_protocol: avctp.Protocol | None
    free_commands: asyncio.Queue
    pending_commands: dict[int, PendingCommand]  # Pending commands, by label
    notification_listeners: dict[EventId, NotificationListener]

    @staticmethod
    def _check_vendor_dependent_frame(
        frame: avc.VendorDependentCommandFrame | avc.VendorDependentResponseFrame,
    ) -> bool:
        if frame.company_id != AVRCP_BLUETOOTH_SIG_COMPANY_ID:
            logger.debug("unsupported company id, ignoring")
            return False

        if frame.subunit_type != avc.Frame.SubunitType.PANEL or frame.subunit_id != 0:
            logger.debug("unsupported subunit")
            return False

        return True

    def __init__(self, delegate: Delegate | None = None) -> None:
        super().__init__()
        self.delegate = delegate if delegate else Delegate()
        self.command_pdu_assembler = PduAssembler(self._on_command_pdu)
        self.receive_command_state = None
        self.response_pdu_assembler = PduAssembler(self._on_response_pdu)
        self.receive_response_state = None
        self.avctp_protocol = None
        self.notification_listeners = {}

        # Create an initial pool of free commands
        self.pending_commands = {}
        self.free_commands = asyncio.Queue()
        for transaction_label in range(16):
            self.free_commands.put_nowait(self.PendingCommand(transaction_label))

    def listen(self, device: Device) -> None:
        """
        Listen for incoming connections.

        A 'connection' event will be emitted when a connection is made, and a 'start'
        event will be emitted when the protocol is ready to be used on that connection.
        """
        device.create_l2cap_server(
            l2cap.ClassicChannelSpec(avctp.AVCTP_PSM), self._on_avctp_connection
        )

    async def connect(self, connection: Connection) -> None:
        """
        Connect to a peer.
        """
        avctp_channel = await connection.create_l2cap_channel(
            l2cap.ClassicChannelSpec(psm=avctp.AVCTP_PSM)
        )
        self._on_avctp_channel_open(avctp_channel)

    async def _obtain_pending_command(self) -> PendingCommand:
        pending_command = await self.free_commands.get()
        self.pending_commands[pending_command.transaction_label] = pending_command
        return pending_command

    def recycle_pending_command(self, pending_command: PendingCommand) -> None:
        pending_command.reset()
        del self.pending_commands[pending_command.transaction_label]
        self.free_commands.put_nowait(pending_command)
        logger.debug(f"recycled pending command, {self.free_commands.qsize()} free")

    _R = TypeVar('_R')

    @staticmethod
    def _check_response(
        response_context: ResponseContext, expected_type: type[_R]
    ) -> _R:
        if isinstance(response_context, Protocol.FinalResponse):
            if (
                response_context.response_code
                != avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE
            ):
                raise Protocol.UnexpectedResponseCodeError(
                    response_context.response_code, response_context.response
                )

            if not (isinstance(response_context.response, expected_type)):
                raise Protocol.MismatchedResponseError(response_context.response)

            return response_context.response

        raise Protocol.UnexpectedResponseTypeError(response_context)

    def _delegate_command(
        self, transaction_label: int, command: Command, method: Awaitable
    ) -> None:
        async def call() -> None:
            try:
                await method
            except Delegate.Error as error:
                self.send_rejected_avrcp_response(
                    transaction_label,
                    command.pdu_id,
                    error.status_code,
                )
            except Exception:
                logger.exception("delegate method raised exception")
                self.send_rejected_avrcp_response(
                    transaction_label,
                    command.pdu_id,
                    StatusCode.INTERNAL_ERROR,
                )

        utils.AsyncRunner.spawn(call())

    async def get_supported_events(self) -> list[EventId]:
        """Get the list of events supported by the connected peer."""
        response_context = await self.send_avrcp_command(
            avc.CommandFrame.CommandType.STATUS,
            GetCapabilitiesCommand(
                GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED
            ),
        )
        response = self._check_response(response_context, GetCapabilitiesResponse)
        return list(
            capability
            for capability in response.capabilities
            if isinstance(capability, EventId)
        )

    async def get_supported_company_ids(self) -> list[int]:
        """Get the list of events supported by the connected peer."""
        response_context = await self.send_avrcp_command(
            avc.CommandFrame.CommandType.STATUS,
            GetCapabilitiesCommand(GetCapabilitiesCommand.CapabilityId.COMPANY_ID),
        )
        response = self._check_response(response_context, GetCapabilitiesResponse)
        return list(
            int.from_bytes(capability, 'big')
            for capability in response.capabilities
            if isinstance(capability, bytes)
        )

    async def get_play_status(self) -> SongAndPlayStatus:
        """Get the play status of the connected peer."""
        response_context = await self.send_avrcp_command(
            avc.CommandFrame.CommandType.STATUS, GetPlayStatusCommand()
        )
        response = self._check_response(response_context, GetPlayStatusResponse)
        return SongAndPlayStatus(
            response.song_length, response.song_position, response.play_status
        )

    async def get_element_attributes(
        self, element_identifier: int, attribute_ids: Sequence[MediaAttributeId]
    ) -> list[MediaAttribute]:
        """Get element attributes from the connected peer."""
        response_context = await self.send_avrcp_command(
            avc.CommandFrame.CommandType.STATUS,
            GetElementAttributesCommand(element_identifier, attribute_ids),
        )
        response = self._check_response(response_context, GetElementAttributesResponse)
        return list(response.attributes)

    async def monitor_events(
        self, event_id: EventId, playback_interval: int = 0
    ) -> AsyncIterator[Event]:
        """
        Monitor events emitted from a peer.

        This generator yields Event objects.
        """

        def check_response(response) -> Event:
            if not isinstance(response, RegisterNotificationResponse):
                raise self.MismatchedResponseError(response)

            return response.event

        while True:
            response = await self.send_avrcp_command(
                avc.CommandFrame.CommandType.NOTIFY,
                RegisterNotificationCommand(event_id, playback_interval),
            )

            if isinstance(response, self.InterimResponse):
                logger.debug(f"interim: {response}")
                yield check_response(response.response)

                logger.debug("waiting for final response")
                response = await response.final

            if not isinstance(response, self.FinalResponse):
                raise self.UnexpectedResponseTypeError(response)

            logger.debug(f"final: {response}")
            if response.response_code != avc.ResponseFrame.ResponseCode.CHANGED:
                raise self.UnexpectedResponseCodeError(
                    response.response_code, response.response
                )

            yield check_response(response.response)

    async def monitor_playback_status(
        self,
    ) -> AsyncIterator[PlayStatus]:
        """Monitor Playback Status changes from the connected peer."""
        async for event in self.monitor_events(EventId.PLAYBACK_STATUS_CHANGED, 0):
            if not isinstance(event, PlaybackStatusChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield event.play_status

    async def monitor_track_changed(
        self,
    ) -> AsyncIterator[bytes]:
        """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

    async def monitor_playback_position(
        self, playback_interval: int
    ) -> AsyncIterator[int]:
        """Monitor Playback Position changes from the connected peer."""
        async for event in self.monitor_events(
            EventId.PLAYBACK_POS_CHANGED, playback_interval
        ):
            if not isinstance(event, PlaybackPositionChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield event.playback_position

    async def monitor_player_application_settings(
        self,
    ) -> AsyncIterator[list[PlayerApplicationSettingChangedEvent.Setting]]:
        """Monitor Player Application Setting changes from the connected peer."""
        async for event in self.monitor_events(
            EventId.PLAYER_APPLICATION_SETTING_CHANGED, 0
        ):
            if not isinstance(event, PlayerApplicationSettingChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield list(event.player_application_settings)

    async def monitor_now_playing_content(self) -> AsyncIterator[None]:
        """Monitor Now Playing changes from the connected peer."""
        async for event in self.monitor_events(EventId.NOW_PLAYING_CONTENT_CHANGED, 0):
            if not isinstance(event, NowPlayingContentChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield None

    async def monitor_available_players(self) -> AsyncIterator[None]:
        """Monitor Available Players changes from the connected peer."""
        async for event in self.monitor_events(EventId.AVAILABLE_PLAYERS_CHANGED, 0):
            if not isinstance(event, AvailablePlayersChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield None

    async def monitor_addressed_player(
        self,
    ) -> AsyncIterator[AddressedPlayerChangedEvent.Player]:
        """Monitor Addressed Player changes from the connected peer."""
        async for event in self.monitor_events(EventId.ADDRESSED_PLAYER_CHANGED, 0):
            if not isinstance(event, AddressedPlayerChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield event.player

    async def monitor_uids(
        self,
    ) -> AsyncIterator[int]:
        """Monitor UID changes from the connected peer."""
        async for event in self.monitor_events(EventId.UIDS_CHANGED, 0):
            if not isinstance(event, UidsChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield event.uid_counter

    async def monitor_volume(
        self,
    ) -> AsyncIterator[int]:
        """Monitor Volume changes from the connected peer."""
        async for event in self.monitor_events(EventId.VOLUME_CHANGED, 0):
            if not isinstance(event, VolumeChangedEvent):
                logger.warning("unexpected event class")
                continue
            yield event.volume

    def notify_event(self, event: Event):
        """Notify an event to the connected peer."""
        if (listener := self.notification_listeners.get(event.event_id)) is None:
            logger.debug(f"no listener for {event.event_id.name}")
            return

        # Emit the notification.
        notification = RegisterNotificationResponse(event)
        self.send_avrcp_response(
            listener.transaction_label,
            avc.ResponseFrame.ResponseCode.CHANGED,
            notification,
        )

        # Remove the listener (they will need to re-register).
        del self.notification_listeners[event.event_id]

    def notify_playback_status_changed(self, status: PlayStatus) -> None:
        """Notify the connected peer of a Playback Status change."""
        self.notify_event(PlaybackStatusChangedEvent(status))

    def notify_track_changed(self, identifier: bytes) -> 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))

    def notify_playback_position_changed(self, position: int) -> None:
        """Notify the connected peer of a Position change."""
        self.notify_event(PlaybackPositionChangedEvent(position))

    def notify_player_application_settings_changed(
        self, settings: Sequence[PlayerApplicationSettingChangedEvent.Setting]
    ) -> None:
        """Notify the connected peer of an Player Application Setting change."""
        self.notify_event(
            PlayerApplicationSettingChangedEvent(settings),
        )

    def notify_now_playing_content_changed(self) -> None:
        """Notify the connected peer of a Now Playing change."""
        self.notify_event(NowPlayingContentChangedEvent())

    def notify_available_players_changed(self) -> None:
        """Notify the connected peer of an Available Players change."""
        self.notify_event(AvailablePlayersChangedEvent())

    def notify_addressed_player_changed(
        self, player: AddressedPlayerChangedEvent.Player
    ) -> None:
        """Notify the connected peer of an Addressed Player change."""
        self.notify_event(AddressedPlayerChangedEvent(player))

    def notify_uids_changed(self, uid_counter: int) -> None:
        """Notify the connected peer of a UID change."""
        self.notify_event(UidsChangedEvent(uid_counter))

    def notify_volume_changed(self, volume: int) -> None:
        """Notify the connected peer of a Volume change."""
        self.notify_event(VolumeChangedEvent(volume))

    def _register_notification_listener(
        self, transaction_label: int, command: RegisterNotificationCommand
    ) -> None:
        listener = self.NotificationListener(transaction_label, command)
        self.notification_listeners[command.event_id] = listener

    def _on_avctp_connection(self, l2cap_channel: l2cap.ClassicChannel) -> None:
        logger.debug("AVCTP connection established")
        l2cap_channel.on(
            l2cap_channel.EVENT_OPEN, lambda: self._on_avctp_channel_open(l2cap_channel)
        )

        self.emit(self.EVENT_CONNECTION)

    def _on_avctp_channel_open(self, l2cap_channel: l2cap.ClassicChannel) -> None:
        logger.debug("AVCTP channel open")
        if self.avctp_protocol is not None:
            # TODO: find a better strategy instead of just closing
            logger.warning("AVCTP protocol already active, closing connection")
            utils.AsyncRunner.spawn(l2cap_channel.disconnect())
            return

        self.avctp_protocol = avctp.Protocol(l2cap_channel)
        self.avctp_protocol.register_command_handler(AVRCP_PID, self._on_avctp_command)
        self.avctp_protocol.register_response_handler(
            AVRCP_PID, self._on_avctp_response
        )
        l2cap_channel.on(l2cap_channel.EVENT_CLOSE, self._on_avctp_channel_close)

        self.emit(self.EVENT_START)

    def _on_avctp_channel_close(self) -> None:
        logger.debug("AVCTP channel closed")
        self.avctp_protocol = None

        self.emit(self.EVENT_STOP)

    def _on_avctp_command(self, transaction_label: int, payload: bytes) -> None:
        command = avc.CommandFrame.from_bytes(payload)
        if not isinstance(command, avc.CommandFrame):
            raise core.InvalidPacketError(
                f"{command} is not a valid AV/C Command Frame"
            )
        logger.debug(
            f"<<< AVCTP Command, transaction_label={transaction_label}: {command}"
        )

        # Only addressing the unit, or the PANEL subunit with subunit ID 0 is supported
        # in this profile.
        if not (
            command.subunit_type == avc.Frame.SubunitType.UNIT
            and command.subunit_id == 7
        ) and not (
            command.subunit_type == avc.Frame.SubunitType.PANEL
            and command.subunit_id == 0
        ):
            logger.debug("subunit not supported")
            self.send_not_implemented_response(transaction_label, command)
            return

        if isinstance(command, avc.VendorDependentCommandFrame):
            if not self._check_vendor_dependent_frame(command):
                return

            if self.receive_command_state is None:
                self.receive_command_state = self.ReceiveCommandState(
                    transaction_label=transaction_label, command_type=command.ctype
                )
            elif (
                self.receive_command_state.transaction_label != transaction_label
                or self.receive_command_state.command_type != command.ctype
            ):
                # We're in the middle of some other PDU
                logger.warning("received interleaved PDU, resetting state")
                self.command_pdu_assembler.reset()
                self.receive_command_state = None
                return
            else:
                self.receive_command_state.command_type = command.ctype
                self.receive_command_state.transaction_label = transaction_label

            self.command_pdu_assembler.on_pdu(command.vendor_dependent_data)
            return

        if isinstance(command, avc.PassThroughCommandFrame):

            async def dispatch_key_event() -> None:
                try:
                    await self.delegate.on_key_event(
                        command.operation_id,
                        command.state_flag == avc.PassThroughFrame.StateFlag.PRESSED,
                        command.operation_data,
                    )
                    response_code = avc.ResponseFrame.ResponseCode.ACCEPTED
                except Delegate.AvcError as error:
                    logger.exception("delegate method raised exception")
                    response_code = error.status_code
                except Exception:
                    logger.exception("delegate method raised exception")
                    response_code = avc.ResponseFrame.ResponseCode.REJECTED
                self.send_passthrough_response(
                    transaction_label=transaction_label,
                    command=command,
                    response_code=response_code,
                )

            utils.AsyncRunner.spawn(dispatch_key_event())
            return

        # TODO handle other types
        self.send_not_implemented_response(transaction_label, command)

    def _on_avctp_response(self, transaction_label: int, payload: bytes | None) -> None:
        response = avc.ResponseFrame.from_bytes(payload) if payload else None
        if not isinstance(response, avc.ResponseFrame):
            raise core.InvalidPacketError(
                f"{response} is not a valid AV/C Response Frame"
            )
        logger.debug(
            f"<<< AVCTP Response, transaction_label={transaction_label}: {response}"
        )

        # Check that we have a pending command that matches this response.
        if not (pending_command := self.pending_commands.get(transaction_label)):
            logger.warning("no pending command with this transaction label")
            return

        # A None response means an invalid PID was used in the request.
        if response is None:
            pending_command.response.set_exception(self.InvalidPidError())

        if isinstance(response, avc.VendorDependentResponseFrame):
            if not self._check_vendor_dependent_frame(response):
                return

            if self.receive_response_state is None:
                self.receive_response_state = self.ReceiveResponseState(
                    transaction_label=transaction_label, response_code=response.response
                )
            elif (
                self.receive_response_state.transaction_label != transaction_label
                or self.receive_response_state.response_code != response.response
            ):
                # We're in the middle of some other PDU
                logger.warning("received interleaved PDU, resetting state")
                self.response_pdu_assembler.reset()
                self.receive_response_state = None
                return
            else:
                self.receive_response_state.response_code = response.response
                self.receive_response_state.transaction_label = transaction_label

            self.response_pdu_assembler.on_pdu(response.vendor_dependent_data)
            return

        if isinstance(response, avc.PassThroughResponseFrame):
            pending_command.response.set_result(response)

        # TODO handle other types

        self.recycle_pending_command(pending_command)

    def _on_command_pdu(self, pdu_id: PduId, pdu: bytes) -> None:
        logger.debug(f"<<< AVRCP command PDU [pdu_id={pdu_id.name}]: {pdu.hex()}")

        assert self.receive_command_state is not None
        transaction_label = self.receive_command_state.transaction_label

        # Dispatch the command.
        # NOTE: with a small number of supported commands, a manual dispatch like this
        # is Ok, but if/when more commands are supported, a lookup dispatch mechanism
        # would be more appropriate.
        # TODO: switch on ctype
        if self.receive_command_state.command_type in (
            avc.CommandFrame.CommandType.CONTROL,
            avc.CommandFrame.CommandType.STATUS,
            avc.CommandFrame.CommandType.NOTIFY,
        ):
            # TODO: catch exceptions from delegates
            command = Command.from_bytes(pdu_id, pdu)
            if isinstance(command, GetCapabilitiesCommand):
                self._on_get_capabilities_command(transaction_label, command)
            elif isinstance(command, SetAbsoluteVolumeCommand):
                self._on_set_absolute_volume_command(transaction_label, command)
            elif isinstance(command, RegisterNotificationCommand):
                self._on_register_notification_command(transaction_label, command)
            elif isinstance(command, GetPlayStatusCommand):
                self._on_get_play_status_command(transaction_label, command)
            else:
                # Not supported.
                # TODO: check that this is the right way to respond in this case.
                logger.debug("unsupported PDU ID")
                self.send_rejected_avrcp_response(
                    transaction_label, pdu_id, StatusCode.INVALID_PARAMETER
                )
        else:
            logger.debug("unsupported command type")
            self.send_rejected_avrcp_response(
                transaction_label, pdu_id, StatusCode.INVALID_COMMAND
            )

        self.receive_command_state = None

    def _on_response_pdu(self, pdu_id: PduId, pdu: bytes) -> None:
        logger.debug(f"<<< AVRCP response PDU [pdu_id={pdu_id.name}]: {pdu.hex()}")

        assert self.receive_response_state is not None

        transaction_label = self.receive_response_state.transaction_label
        response_code = self.receive_response_state.response_code
        self.receive_response_state = None

        # Check that we have a pending command that matches this response.
        if not (pending_command := self.pending_commands.get(transaction_label)):
            logger.warning("no pending command with this transaction label")
            return

        # Convert the PDU bytes into a response object.
        # NOTE: with a small number of supported responses, a manual switch like this
        # is Ok, but if/when more responses are supported, a lookup mechanism would be
        # more appropriate.
        response: Response | None = None
        if response_code == avc.ResponseFrame.ResponseCode.REJECTED:
            response = RejectedResponse(pdu_id=pdu_id, status_code=StatusCode(pdu[0]))
        elif response_code == avc.ResponseFrame.ResponseCode.NOT_IMPLEMENTED:
            response = NotImplementedResponse(pdu_id=pdu_id, parameters=pdu)
        elif response_code in (
            avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE,
            avc.ResponseFrame.ResponseCode.INTERIM,
            avc.ResponseFrame.ResponseCode.CHANGED,
            avc.ResponseFrame.ResponseCode.ACCEPTED,
        ):
            response = Response.from_bytes(pdu=pdu, pdu_id=PduId(pdu_id))
        else:
            logger.debug("unexpected response code")
            pending_command.response.set_exception(
                core.ProtocolError(
                    error_code=None,
                    error_namespace="avrcp",
                    details="unexpected response code",
                )
            )

        if response is None:
            self.recycle_pending_command(pending_command)
            return

        logger.debug(f"<<< AVRCP response: {response}")

        # Make the response available to the waiter.
        if response_code == avc.ResponseFrame.ResponseCode.INTERIM:
            pending_interim_response = pending_command.response
            pending_command.reset()
            pending_interim_response.set_result(
                self.InterimResponse(
                    pending_command.transaction_label,
                    response,
                    pending_command.response,
                )
            )
        else:
            pending_command.response.set_result(
                self.FinalResponse(
                    pending_command.transaction_label,
                    response,
                    response_code,
                )
            )
            self.recycle_pending_command(pending_command)

    def send_command(self, transaction_label: int, command: avc.CommandFrame) -> None:
        logger.debug(f">>> AVRCP command: {command}")

        if self.avctp_protocol is None:
            logger.warning("trying to send command while avctp_protocol is None")
            return

        self.avctp_protocol.send_command(transaction_label, AVRCP_PID, bytes(command))

    async def send_passthrough_command(
        self, command: avc.PassThroughCommandFrame
    ) -> avc.PassThroughResponseFrame:
        # Wait for a free command slot.
        pending_command = await self._obtain_pending_command()

        # Send the command.
        self.send_command(pending_command.transaction_label, command)

        # Wait for the response.
        return await pending_command.response

    async def send_key_event(
        self, key: avc.PassThroughCommandFrame.OperationId, pressed: bool
    ) -> avc.PassThroughResponseFrame:
        """Send a key event to the connected peer."""
        return await self.send_passthrough_command(
            avc.PassThroughCommandFrame(
                avc.CommandFrame.CommandType.CONTROL,
                avc.Frame.SubunitType.PANEL,
                0,
                (
                    avc.PassThroughFrame.StateFlag.PRESSED
                    if pressed
                    else avc.PassThroughFrame.StateFlag.RELEASED
                ),
                key,
                b'',
            )
        )

    async def send_avrcp_command(
        self, command_type: avc.CommandFrame.CommandType, command: Command
    ) -> ResponseContext:
        # Wait for a free command slot.
        pending_command = await self._obtain_pending_command()

        # TODO: fragmentation
        # Send the command.
        logger.debug(f">>> AVRCP command PDU: {command}")
        payload = bytes(command)
        pdu = struct.pack(">BBH", command.pdu_id, 0, len(payload)) + payload
        command_frame = avc.VendorDependentCommandFrame(
            command_type,
            avc.Frame.SubunitType.PANEL,
            0,
            AVRCP_BLUETOOTH_SIG_COMPANY_ID,
            pdu,
        )
        self.send_command(pending_command.transaction_label, command_frame)

        # Wait for the response.
        return await pending_command.response

    def send_response(
        self, transaction_label: int, response: avc.ResponseFrame
    ) -> None:
        assert self.avctp_protocol is not None
        logger.debug(f">>> AVRCP response: {response}")
        self.avctp_protocol.send_response(transaction_label, AVRCP_PID, bytes(response))

    def send_passthrough_response(
        self,
        transaction_label: int,
        command: avc.PassThroughCommandFrame,
        response_code: avc.ResponseFrame.ResponseCode,
    ):
        response = avc.PassThroughResponseFrame(
            response_code,
            avc.Frame.SubunitType.PANEL,
            0,
            command.state_flag,
            command.operation_id,
            command.operation_data,
        )
        self.send_response(transaction_label, response)

    def send_avrcp_response(
        self,
        transaction_label: int,
        response_code: avc.ResponseFrame.ResponseCode,
        response: Response,
    ) -> None:
        # TODO: fragmentation
        logger.debug(f">>> AVRCP response PDU: {response}")
        payload = bytes(response)
        pdu = struct.pack(">BBH", response.pdu_id, 0, len(payload)) + payload
        response_frame = avc.VendorDependentResponseFrame(
            response_code,
            avc.Frame.SubunitType.PANEL,
            0,
            AVRCP_BLUETOOTH_SIG_COMPANY_ID,
            pdu,
        )
        self.send_response(transaction_label, response_frame)

    def send_not_implemented_response(
        self, transaction_label: int, command: avc.CommandFrame
    ) -> None:
        response = avc.ResponseFrame(
            avc.ResponseFrame.ResponseCode.NOT_IMPLEMENTED,
            command.subunit_type,
            command.subunit_id,
            command.opcode,
            command.operands,
        )
        self.send_response(transaction_label, response)

    def send_rejected_avrcp_response(
        self, transaction_label: int, pdu_id: PduId, status_code: StatusCode
    ) -> None:
        self.send_avrcp_response(
            transaction_label,
            avc.ResponseFrame.ResponseCode.REJECTED,
            RejectedResponse(pdu_id, status_code),
        )

    def send_not_implemented_avrcp_response(
        self, transaction_label: int, pdu_id: PduId
    ) -> None:
        self.send_avrcp_response(
            transaction_label,
            avc.ResponseFrame.ResponseCode.NOT_IMPLEMENTED,
            NotImplementedResponse(pdu_id, b''),
        )

    def _on_get_capabilities_command(
        self, transaction_label: int, command: GetCapabilitiesCommand
    ) -> None:
        logger.debug(f"<<< AVRCP command PDU: {command}")

        async def get_supported_events() -> None:
            capabilities: Sequence[bytes | SupportsBytes]
            if (
                command.capability_id
                == GetCapabilitiesCommand.CapabilityId.EVENTS_SUPPORTED
            ):
                capabilities = await self.delegate.get_supported_events()
            elif (
                command.capability_id == GetCapabilitiesCommand.CapabilityId.COMPANY_ID
            ):
                company_ids = await self.delegate.get_supported_company_ids()
                capabilities = [
                    company_id.to_bytes(3, 'big') for company_id in company_ids
                ]
            else:
                raise core.InvalidArgumentError(
                    f"Unsupported capability: {command.capability_id}"
                )
            self.send_avrcp_response(
                transaction_label,
                avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE,
                GetCapabilitiesResponse(command.capability_id, capabilities),
            )

        self._delegate_command(transaction_label, command, get_supported_events())

    def _on_set_absolute_volume_command(
        self, transaction_label: int, command: SetAbsoluteVolumeCommand
    ) -> None:
        logger.debug(f"<<< AVRCP command PDU: {command}")

        async def set_absolute_volume() -> None:
            await self.delegate.set_absolute_volume(command.volume)
            effective_volume = await self.delegate.get_absolute_volume()
            self.send_avrcp_response(
                transaction_label,
                avc.ResponseFrame.ResponseCode.ACCEPTED,
                SetAbsoluteVolumeResponse(effective_volume),
            )

        self._delegate_command(transaction_label, command, set_absolute_volume())

    def _on_get_play_status_command(
        self, transaction_label: int, command: GetPlayStatusCommand
    ) -> None:
        logger.debug("<<< AVRCP command PDU: %s", command)

        async def get_playback_status() -> None:
            play_status: PlayStatus = await self.delegate.get_playback_status()
            self.send_avrcp_response(
                transaction_label,
                avc.ResponseFrame.ResponseCode.IMPLEMENTED_OR_STABLE,
                GetPlayStatusResponse(
                    # TODO: Delegate this.
                    song_length=GetPlayStatusResponse.UNAVAILABLE,
                    song_position=GetPlayStatusResponse.UNAVAILABLE,
                    play_status=play_status,
                ),
            )

        self._delegate_command(transaction_label, command, get_playback_status())

    def _on_register_notification_command(
        self, transaction_label: int, command: RegisterNotificationCommand
    ) -> None:
        logger.debug(f"<<< AVRCP command PDU: {command}")

        async def register_notification() -> None:
            # Check if the event is supported.
            supported_events = await self.delegate.get_supported_events()
            if command.event_id not in supported_events:
                logger.debug("event not supported")
                self.send_not_implemented_avrcp_response(
                    transaction_label, PduId.REGISTER_NOTIFICATION
                )
                return

            response: Response
            if command.event_id == EventId.VOLUME_CHANGED:
                volume = await self.delegate.get_absolute_volume()
                response = RegisterNotificationResponse(VolumeChangedEvent(volume))
            elif command.event_id == EventId.PLAYBACK_STATUS_CHANGED:
                playback_status = await self.delegate.get_playback_status()
                response = RegisterNotificationResponse(
                    PlaybackStatusChangedEvent(play_status=playback_status)
                )
            elif command.event_id == EventId.NOW_PLAYING_CONTENT_CHANGED:
                playback_status = await self.delegate.get_playback_status()
                response = RegisterNotificationResponse(NowPlayingContentChangedEvent())
            else:
                logger.warning("Event supported but not handled %s", command.event_id)
                return

            self.send_avrcp_response(
                transaction_label,
                avc.ResponseFrame.ResponseCode.INTERIM,
                response,
            )
            self._register_notification_listener(transaction_label, command)

        self._delegate_command(transaction_label, command, register_notification())