Replace | typing usage with Optional and Union

This commit is contained in:
Josh Wu
2023-09-12 15:45:05 +08:00
parent 1e00c8f456
commit f39f5f531c
8 changed files with 19 additions and 11 deletions

View File

@@ -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 = '',

View File

@@ -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

View File

@@ -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")} '

View File

@@ -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)

View File

@@ -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')

View File

@@ -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)

View File

@@ -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

View File

@@ -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