From fc1bf36ace085ff97fc0efc2a1db7068a48d721c Mon Sep 17 00:00:00 2001 From: skarnataki Date: Thu, 5 Oct 2023 10:55:28 +0000 Subject: [PATCH] Review changes comment fix. Classes/Subclass/dataclass. Enum constants. Naming conventions --- bumble/hid.py | 172 ++++++++++++++------------------------- bumble/sdp.py | 4 +- examples/run_hid_host.py | 10 +-- tests/sdp_test.py | 16 ++-- 4 files changed, 76 insertions(+), 126 deletions(-) diff --git a/bumble/hid.py b/bumble/hid.py index 8c18b2aa..60102585 100644 --- a/bumble/hid.py +++ b/bumble/hid.py @@ -40,123 +40,104 @@ logger = logging.getLogger(__name__) # Constants # ----------------------------------------------------------------------------- # fmt: off - - +HID_CONTROL_PSM = 0x0011 +HID_INTERRUPT_PSM = 0x0013 class Message(): - class HIDPsm(enum.IntEnum): - HID_CONTROL_PSM = 0x0011 - HID_INTERRUPT_PSM = 0x0013 # Report types class ReportType(enum.IntEnum): - HID_OTHER_REPORT = 0x00 - HID_INPUT_REPORT = 0x01 - HID_OUTPUT_REPORT = 0x02 - HID_FEATURE_REPORT = 0x03 + OTHER_REPORT = 0x00 + INPUT_REPORT = 0x01 + OUTPUT_REPORT = 0x02 + FEATURE_REPORT = 0x03 # Handshake parameters - class HandshakeState(enum.IntEnum): - HANDSHAKE_SUCCESSFUL = 0x00 - HANDSHAKE_NOT_READY = 0x01 - HANDSHAKE_ERR_INVALID_REPORT_ID = 0x02 - HANDSHAKE_ERR_UNSUPPORTED_REQUEST = 0x03 - HANDSHAKE_ERR_UNKNOWN = 0x0E - HANDSHAKE_ERR_FATAL = 0x0F + class Handshake(enum.IntEnum): + SUCCESSFUL = 0x00 + NOT_READY = 0x01 + ERR_INVALID_REPORT_ID = 0x02 + ERR_UNSUPPORTED_REQUEST = 0x03 + ERR_UNKNOWN = 0x0E + ERR_FATAL = 0x0F - class Type(enum.IntEnum): - HID_HANDSHAKE = 0x00 - HID_CONTROL = 0x01 - HID_GET_REPORT = 0x04 - HID_SET_REPORT = 0x05 - HID_GET_PROTOCOL = 0x06 - HID_SET_PROTOCOL = 0x07 - HID_DATA = 0x0A + #Message Type + class MessageType(enum.IntEnum): + HANDSHAKE = 0x00 + CONTROL = 0x01 + GET_REPORT = 0x04 + SET_REPORT = 0x05 + GET_PROTOCOL = 0x06 + SET_PROTOCOL = 0x07 + DATA = 0x0A # Protocol modes class ProtocolMode(enum.IntEnum): - HID_BOOT_PROTOCOL_MODE = 0x00 - HID_REPORT_PROTOCOL_MODE = 0x01 + BOOT_PROTOCOL = 0x00 + REPORT_PROTOCOL = 0x01 # Control Operations class ControlCommand(enum.IntEnum): - HID_SUSPEND = 0x03 - HID_EXIT_SUSPEND = 0x04 - HID_VIRTUAL_CABLE_UNPLUG = 0x05 + SUSPEND = 0x03 + EXIT_SUSPEND = 0x04 + VIRTUAL_CABLE_UNPLUG = 0x05 - # HIDP message types +# HIDP messages @dataclass class GetReportMessage(Message): report_type : int report_id : int buffer_size : int - ''' - def __init__(self, - report_type: Optional[int] = None, - report_id: Optional[int] = None, - buffer_size: Optional[int] = None, - ): - self.report_type = report_type - self.report_id = report_id - self.buffer_size = buffer_size - ''' + def __bytes__(self) -> bytes: - if(self.report_type == Message.ReportType.HID_OTHER_REPORT): + if(self.report_type == Message.ReportType.OTHER_REPORT): param = self.report_type else: param = 0x08 | self.report_type - header = ((Message.Type.HID_GET_REPORT << 4) | param) + header = ((Message.MessageType.GET_REPORT << 4) | param) packet_bytes = bytearray() packet_bytes.append(header) packet_bytes.append(self.report_id) packet_bytes.extend([(self.buffer_size & 0xff), ((self.buffer_size >> 8) & 0xff)]) return bytes(packet_bytes) - +@dataclass class SetReportMessage(Message): - - def __init__(self, - report_type: int, - data : bytes): - self.report_type = report_type - self.data = data + report_type: int + data : bytes def __bytes__(self) -> bytes: - header = ((Message.Type.HID_SET_REPORT << 4) | self.report_type) + header = ((Message.MessageType.SET_REPORT << 4) | self.report_type) packet_bytes = bytearray() packet_bytes.append(header) packet_bytes.extend(self.data) return bytes(packet_bytes) - +@dataclass class GetProtocolMessage(Message): - def __bytes__(self) -> bytes: - header = (Message.Type.HID_GET_PROTOCOL << 4) + header = (Message.MessageType.GET_PROTOCOL << 4) packet_bytes = bytearray() packet_bytes.append(header) return bytes(packet_bytes) - +@dataclass class SetProtocolMessage(Message): - - def __init__(self, protocol_mode: int): - self.protocol_mode = protocol_mode - + protocol_mode: int def __bytes__(self) -> bytes: - header = (Message.Type.HID_SET_PROTOCOL << 4 | self.protocol_mode) + header = (Message.MessageType.SET_PROTOCOL << 4 | self.protocol_mode) packet_bytes = bytearray() packet_bytes.append(header) packet_bytes.append(self.protocol_mode) return bytes(packet_bytes) - +@dataclass class SendData(Message): - def __init__(self, data : bytes): - self.data = data + + data : bytes def __bytes__(self) -> bytes: - header = ((Message.Type.HID_DATA << 4) | Message.ReportType.HID_OUTPUT_REPORT) + header = ((Message.MessageType.DATA << 4) | Message.ReportType.OUTPUT_REPORT) packet_bytes = bytearray() packet_bytes.append(header) packet_bytes.extend(self.data) @@ -173,14 +154,14 @@ class Host(EventEmitter): self.l2cap_intr_channel = None # Register ourselves with the L2CAP channel manager - device.register_l2cap_server(Message.HIDPsm.HID_CONTROL_PSM, self.on_connection) - device.register_l2cap_server(Message.HIDPsm.HID_INTERRUPT_PSM, self.on_connection) + device.register_l2cap_server(HID_CONTROL_PSM, self.on_connection) + device.register_l2cap_server(HID_INTERRUPT_PSM, self.on_connection) async def connect_control_channel(self) -> None: # Create a new L2CAP connection - control channel try: self.l2cap_ctrl_channel = await self.device.l2cap_channel_manager.connect( - self.connection, Message.HIDPsm.HID_CONTROL_PSM + self.connection, HID_CONTROL_PSM ) except ProtocolError as error: logging.exception(f'L2CAP connection failed: {error}') @@ -194,7 +175,7 @@ class Host(EventEmitter): # Create a new L2CAP connection - interrupt channel try: self.l2cap_intr_channel = await self.device.l2cap_channel_manager.connect( - self.connection, Message.HIDPsm.HID_INTERRUPT_PSM + self.connection, HID_INTERRUPT_PSM ) except ProtocolError as error: logging.exception(f'L2CAP connection failed: {error}') @@ -207,7 +188,6 @@ class Host(EventEmitter): async def disconnect_interrupt_channel(self) -> None: if self.l2cap_intr_channel is None: raise InvalidStateError('invalid state') - await self.l2cap_intr_channel.disconnect() # type: ignore channel = self.l2cap_intr_channel self.l2cap_intr_channel = None await channel.disconnect() # type: ignore @@ -215,7 +195,6 @@ class Host(EventEmitter): async def disconnect_control_channel(self) -> None: if self.l2cap_ctrl_channel is None: raise InvalidStateError('invalid state') - await self.l2cap_ctrl_channel.disconnect() # type: ignore channel = self.l2cap_ctrl_channel self.l2cap_ctrl_channel = None await channel.disconnect() # type: ignore @@ -225,7 +204,7 @@ class Host(EventEmitter): l2cap_channel.on('open', lambda: self.on_l2cap_channel_open(l2cap_channel)) def on_l2cap_channel_open(self, l2cap_channel: l2cap.Channel) -> None: - if l2cap_channel.psm == Message.HIDPsm.HID_CONTROL_PSM: + if l2cap_channel.psm == HID_CONTROL_PSM: self.l2cap_ctrl_channel = l2cap_channel self.l2cap_ctrl_channel.sink = self.on_ctrl_pdu else: @@ -239,27 +218,20 @@ class Host(EventEmitter): message_type = pdu[0] >> 4 param = pdu[0] & 0x0f - for command in Message.ControlCommand.__members__items(): - if param == command: - logger.debug(f'<<< ', command + pdu) - self.handle_handshake(param) - self.emit(command, pdu) - ''' - if message_type == Message.Type.HID_HANDSHAKE : - logger.debug('<<< HID HANDSHAKE') - self.handle_handshake(param) - self.emit('handshake', pdu) - elif message_type == Message.Type.HID_DATA : + if message_type == Message.MessageType.HANDSHAKE : + logger.debug(f'<<< HID HANDSHAKE: {Message.Handshake(param).name}') + self.emit('handshake', Message.Handshake(param)) + elif message_type == Message.MessageType.DATA : logger.debug('<<< HID CONTROL DATA') self.emit('data', pdu) - elif message_type == Message.Type.HID_CONTROL : - if param == Message.ControlCommand.HID_SUSPEND : + elif message_type == Message.MessageType.CONTROL : + if param == Message.ControlCommand.SUSPEND : logger.debug('<<< HID SUSPEND') self.emit('suspend', pdu) - elif param == HID_EXIT_SUSPEND : + elif param == Message.ControlCommand.EXIT_SUSPEND : logger.debug('<<< HID EXIT SUSPEND') self.emit('exit_suspend', pdu) - elif param == HID_VIRTUAL_CABLE_UNPLUG : + elif param == Message.ControlCommand.VIRTUAL_CABLE_UNPLUG : logger.debug('<<< HID VIRTUAL CABLE UNPLUG') self.emit('virtual_cable_unplug') else: @@ -267,7 +239,7 @@ class Host(EventEmitter): else: logger.debug('<<< HID CONTROL DATA') self.emit('data', pdu) - ''' + def on_intr_pdu(self, pdu: bytes) -> None: logger.debug(f'<<< HID INTERRUPT PDU: {pdu.hex()}') @@ -304,46 +276,26 @@ class Host(EventEmitter): self.l2cap_intr_channel.send_pdu(msg) # type: ignore def send_data(self, data): - msg = Message(data= data) + msg = SendData(data) hid_message = msg.__bytes__() logger.debug(f'>>> HID INTERRUPT SEND DATA, PDU: {hid_message.hex()}') self.send_pdu_on_intr(hid_message) # type: ignore def suspend(self): - header = (Message.Type.HID_CONTROL << 4 | Message.ControlCommand.HID_SUSPEND) + header = (Message.MessageType.CONTROL << 4 | Message.ControlCommand.SUSPEND) msg = bytearray([header]) logger.debug(f'>>> HID CONTROL SUSPEND, PDU:{msg.hex()}') self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore def exit_suspend(self): - header = (Message.Type.HID_CONTROL << 4 | Message.ControlCommand.HID_EXIT_SUSPEND) + header = (Message.MessageType.CONTROL << 4 | Message.ControlCommand.EXIT_SUSPEND) msg = bytearray([header]) logger.debug(f'>>> HID CONTROL EXIT SUSPEND, PDU:{msg.hex()}') self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore def virtual_cable_unplug(self): - header = (Message.Type.HID_CONTROL << 4 | Message.ControlCommand.HID_VIRTUAL_CABLE_UNPLUG) + header = (Message.MessageType.CONTROL << 4 | Message.ControlCommand.VIRTUAL_CABLE_UNPLUG) msg = bytearray([header]) logger.debug(f'>>> HID CONTROL VIRTUAL CABLE UNPLUG, PDU: {msg.hex()}') self.l2cap_ctrl_channel.send_pdu(msg) # type: ignore - def handle_handshake(self, param: Message.HandshakeState): - for state in Message.HandshakeState.__members__items(): - if param == state: - logger.debug(f'<<< HID HANDSHAKE: ', state) - ''' - if param == HANDSHAKE_SUCCESSFUL : - logger.debug(f'<<< HID HANDSHAKE: SUCCESSFUL') - elif param == HANDSHAKE_NOT_READY : - logger.warning(f'<<< HID HANDSHAKE: NOT_READY') - elif param == HANDSHAKE_ERR_INVALID_REPORT_ID : - logger.warning(f'<<< HID HANDSHAKE: ERR_INVALID_REPORT_ID') - elif param == HANDSHAKE_ERR_UNSUPPORTED_REQUEST : - logger.warning(f'<<< HID HANDSHAKE: ERR_UNSUPPORTED_REQUEST') - elif param == HANDSHAKE_ERR_UNKNOWN : - logger.warning(f'<<< HID HANDSHAKE: ERR_UNKNOWN') - elif param == HANDSHAKE_ERR_FATAL : - logger.warning(f'<<< HID HANDSHAKE: ERR_FATAL') - else: # 0x5-0xD = Reserved - logger.warning("<<< HID HANDSHAKE: RESERVED VALUE") - ''' \ No newline at end of file diff --git a/bumble/sdp.py b/bumble/sdp.py index 389cd558..bcd12cff 100644 --- a/bumble/sdp.py +++ b/bumble/sdp.py @@ -167,7 +167,7 @@ class DataElement: UUID: lambda x: DataElement( DataElement.UUID, core.UUID.from_bytes(bytes(reversed(x))) ), - TEXT_STRING: lambda x: DataElement(DataElement.TEXT_STRING, x.decode('latin1')), + TEXT_STRING: lambda x: DataElement(DataElement.TEXT_STRING, x), BOOLEAN: lambda x: DataElement(DataElement.BOOLEAN, x[0] == 1), SEQUENCE: lambda x: DataElement( DataElement.SEQUENCE, DataElement.list_from_bytes(x) @@ -376,8 +376,6 @@ class DataElement: raise ValueError('invalid value_size') elif self.type == DataElement.UUID: data = bytes(reversed(bytes(self.value))) - elif self.type == DataElement.TEXT_STRING: - data = self.value.encode('latin1') elif self.type == DataElement.URL: data = self.value.encode('utf8') elif self.type == DataElement.BOOLEAN: diff --git a/examples/run_hid_host.py b/examples/run_hid_host.py index efb9768b..17fbfaac 100644 --- a/examples/run_hid_host.py +++ b/examples/run_hid_host.py @@ -169,7 +169,7 @@ async def get_hid_device_sdp_record(device, connection): elif attribute.id == SDP_HID_DESCRIPTOR_LIST_ATTRIBUTE_ID : print(color(' HID Report Descriptor type: ', 'cyan'), hex(attribute.value.value[0].value[0].value)) - print(color(' HID Report DescriptorList: ', 'cyan'), (attribute.value.value[0].value[1].value).encode('latin-1')) + print(color(' HID Report DescriptorList: ', 'cyan'), attribute.value.value[0].value[1].value) HID_Descriptor_Type = attribute.value.value[0].value[0].value HID_Report_Descriptor_List = attribute.value.value[0].value[1].value @@ -243,13 +243,13 @@ async def main(): report_length = len(pdu[1:]) report_id = pdu[1] - if (report_type != Message.ReportType.HID_OTHER_REPORT): + if (report_type != Message.ReportType.OTHER_REPORT): print(color(f' Report type = {report_type}, Report length = {report_length}, Report id = {report_id}', 'blue', None, 'bold')) if ((report_length <= 1) or (report_id == 0)): return - if report_type == Message.ReportType.HID_INPUT_REPORT: + if report_type == Message.ReportType.INPUT_REPORT: ReportParser.parse_input_report(pdu[1:]) #type: ignore async def handle_virtual_cable_unplug(): @@ -383,10 +383,10 @@ async def main(): choice1 = choice1.decode('utf-8').strip() if choice1 == '0': - hid_host.set_protocol(Message.ProtocolMode.HID_BOOT_PROTOCOL_MODE) + hid_host.set_protocol(Message.ProtocolMode.BOOT_PROTOCOL) elif choice1 == '1': - hid_host.set_protocol(Message.ProtocolMode.HID_REPORT_PROTOCOL_MODE) + hid_host.set_protocol(Message.ProtocolMode.REPORT_PROTOCOL) else: print('Incorrect option selected') diff --git a/tests/sdp_test.py b/tests/sdp_test.py index d6491388..29db8751 100644 --- a/tests/sdp_test.py +++ b/tests/sdp_test.py @@ -100,13 +100,13 @@ def test_data_elements() -> None: e = DataElement(DataElement.UUID, UUID('61A3512C-09BE-4DDC-A6A6-0B03667AAFC6')) basic_check(e) - e = DataElement(DataElement.TEXT_STRING, 'hello') + e = DataElement(DataElement.TEXT_STRING, b'hello') basic_check(e) - e = DataElement(DataElement.TEXT_STRING, 'hello' * 60) + e = DataElement(DataElement.TEXT_STRING, b'hello' * 60) basic_check(e) - e = DataElement(DataElement.TEXT_STRING, 'hello' * 20000) + e = DataElement(DataElement.TEXT_STRING, b'hello' * 20000) basic_check(e) e = DataElement(DataElement.BOOLEAN, True) @@ -122,7 +122,7 @@ def test_data_elements() -> None: DataElement.SEQUENCE, [ DataElement(DataElement.BOOLEAN, True), - DataElement(DataElement.TEXT_STRING, 'hello'), + DataElement(DataElement.TEXT_STRING, b'hello'), ], ) basic_check(e) @@ -134,7 +134,7 @@ def test_data_elements() -> None: DataElement.ALTERNATIVE, [ DataElement(DataElement.BOOLEAN, True), - DataElement(DataElement.TEXT_STRING, 'hello'), + DataElement(DataElement.TEXT_STRING, b'hello'), ], ) basic_check(e) @@ -152,19 +152,19 @@ def test_data_elements() -> None: e = DataElement.uuid(UUID.from_16_bits(1234)) basic_check(e) - e = DataElement.text_string('hello') + e = DataElement.text_string(b'hello') basic_check(e) e = DataElement.boolean(True) basic_check(e) e = DataElement.sequence( - [DataElement.signed_integer(0, 1), DataElement.text_string('hello')] + [DataElement.signed_integer(0, 1), DataElement.text_string(b'hello')] ) basic_check(e) e = DataElement.alternative( - [DataElement.signed_integer(0, 1), DataElement.text_string('hello')] + [DataElement.signed_integer(0, 1), DataElement.text_string(b'hello')] ) basic_check(e)