From f39f5f531c2d00287f881af4638152ee27d0b4ad Mon Sep 17 00:00:00 2001 From: Josh Wu Date: Tue, 12 Sep 2023 15:45:05 +0800 Subject: [PATCH] Replace | typing usage with Optional and Union --- bumble/core.py | 2 +- bumble/hci.py | 2 +- bumble/l2cap.py | 6 +++--- bumble/transport/android_emulator.py | 6 ++++-- bumble/transport/android_netsim.py | 2 +- bumble/transport/hci_socket.py | 4 +++- bumble/transport/pty.py | 4 +++- bumble/transport/vhci.py | 4 +++- 8 files changed, 19 insertions(+), 11 deletions(-) diff --git a/bumble/core.py b/bumble/core.py index b00c40e..ba8f876 100644 --- a/bumble/core.py +++ b/bumble/core.py @@ -80,7 +80,7 @@ class BaseError(Exception): def __init__( self, - error_code: int | None, + error_code: Optional[int], error_namespace: str = '', error_name: str = '', details: str = '', diff --git a/bumble/hci.py b/bumble/hci.py index b7014dd..41deed2 100644 --- a/bumble/hci.py +++ b/bumble/hci.py @@ -4397,7 +4397,7 @@ class HCI_Event(HCI_Packet): if len(parameters) != length: raise ValueError('invalid packet length') - cls: Type[HCI_Event | HCI_LE_Meta_Event] | None + cls: Any if event_code == HCI_LE_META_EVENT: # We do this dispatch here and not in the subclass in order to avoid call # loops diff --git a/bumble/l2cap.py b/bumble/l2cap.py index 270d909..9798e87 100644 --- a/bumble/l2cap.py +++ b/bumble/l2cap.py @@ -757,7 +757,7 @@ class Channel(EventEmitter): ) self.state = new_state - def send_pdu(self, pdu: SupportsBytes | bytes) -> None: + def send_pdu(self, pdu: Union[SupportsBytes, bytes]) -> None: self.manager.send_pdu(self.connection, self.destination_cid, pdu) def send_control_frame(self, frame: L2CAP_Control_Frame) -> None: @@ -1098,7 +1098,7 @@ class LeConnectionOrientedChannel(EventEmitter): elif new_state == self.DISCONNECTED: self.emit('close') - def send_pdu(self, pdu: SupportsBytes | bytes) -> None: + def send_pdu(self, pdu: Union[SupportsBytes, bytes]) -> None: self.manager.send_pdu(self.connection, self.destination_cid, pdu) def send_control_frame(self, frame: L2CAP_Control_Frame) -> None: @@ -1569,7 +1569,7 @@ class ChannelManager: if connection_handle in self.identifiers: del self.identifiers[connection_handle] - def send_pdu(self, connection, cid: int, pdu: SupportsBytes | bytes) -> None: + def send_pdu(self, connection, cid: int, pdu: Union[SupportsBytes, bytes]) -> None: pdu_str = pdu.hex() if isinstance(pdu, bytes) else str(pdu) logger.debug( f'{color(">>> Sending L2CAP PDU", "blue")} ' diff --git a/bumble/transport/android_emulator.py b/bumble/transport/android_emulator.py index 5ef0047..8621f56 100644 --- a/bumble/transport/android_emulator.py +++ b/bumble/transport/android_emulator.py @@ -18,6 +18,8 @@ import logging import grpc.aio +from typing import Optional, Union + from .common import PumpedTransport, PumpedPacketSource, PumpedPacketSink, Transport # pylint: disable=no-name-in-module @@ -33,7 +35,7 @@ logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- -async def open_android_emulator_transport(spec: str | None) -> Transport: +async def open_android_emulator_transport(spec: Optional[str]) -> Transport: ''' Open a transport connection to an Android emulator via its gRPC interface. The parameter string has this syntax: @@ -82,7 +84,7 @@ async def open_android_emulator_transport(spec: str | None) -> Transport: logger.debug(f'connecting to gRPC server at {server_address}') channel = grpc.aio.insecure_channel(server_address) - service: EmulatedBluetoothServiceStub | VhciForwardingServiceStub + service: Union[EmulatedBluetoothServiceStub, VhciForwardingServiceStub] if mode == 'host': # Connect as a host service = EmulatedBluetoothServiceStub(channel) diff --git a/bumble/transport/android_netsim.py b/bumble/transport/android_netsim.py index 76a7385..55f2502 100644 --- a/bumble/transport/android_netsim.py +++ b/bumble/transport/android_netsim.py @@ -122,7 +122,7 @@ def publish_grpc_port(grpc_port) -> bool: # ----------------------------------------------------------------------------- async def open_android_netsim_controller_transport( - server_host: str | None, server_port: int + server_host: Optional[str], server_port: int ) -> Transport: if not server_port: raise ValueError('invalid port') diff --git a/bumble/transport/hci_socket.py b/bumble/transport/hci_socket.py index 9891c5b..f071e52 100644 --- a/bumble/transport/hci_socket.py +++ b/bumble/transport/hci_socket.py @@ -23,6 +23,8 @@ import socket import ctypes import collections +from typing import Optional + from .common import Transport, ParserSource @@ -33,7 +35,7 @@ logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- -async def open_hci_socket_transport(spec: str | None) -> Transport: +async def open_hci_socket_transport(spec: Optional[str]) -> Transport: ''' Open an HCI Socket (only available on some platforms). The parameter string is either empty (to use the first/default Bluetooth adapter) diff --git a/bumble/transport/pty.py b/bumble/transport/pty.py index 7765b09..2f46e75 100644 --- a/bumble/transport/pty.py +++ b/bumble/transport/pty.py @@ -23,6 +23,8 @@ import atexit import os import logging +from typing import Optional + from .common import Transport, StreamPacketSource, StreamPacketSink # ----------------------------------------------------------------------------- @@ -32,7 +34,7 @@ logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- -async def open_pty_transport(spec: str | None) -> Transport: +async def open_pty_transport(spec: Optional[str]) -> Transport: ''' Open a PTY transport. The parameter string may be empty, or a path name where a symbolic link diff --git a/bumble/transport/vhci.py b/bumble/transport/vhci.py index 5795840..2b19085 100644 --- a/bumble/transport/vhci.py +++ b/bumble/transport/vhci.py @@ -17,6 +17,8 @@ # ----------------------------------------------------------------------------- import logging +from typing import Optional + from .common import Transport from .file import open_file_transport @@ -27,7 +29,7 @@ logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- -async def open_vhci_transport(spec: str | None) -> Transport: +async def open_vhci_transport(spec: Optional[str]) -> Transport: ''' Open a VHCI transport (only available on some platforms). The parameter string is either empty (to use the default VHCI device