Merge pull request #285 from google/gbg/fix-mypy-errors

mypy: ignore false positive errors

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

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

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:
@@ -1571,7 +1571,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")} '

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)

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

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)
@@ -45,9 +47,9 @@ async def open_hci_socket_transport(spec: str | None) -> Transport:
     # Create a raw HCI socket
     try:
         hci_socket = socket.socket(
-            socket.AF_BLUETOOTH,
+            socket.AF_BLUETOOTH,  # type: ignore[attr-defined]
-            socket.SOCK_RAW | socket.SOCK_NONBLOCK,
+            socket.SOCK_RAW | socket.SOCK_NONBLOCK,  # type: ignore[attr-defined]
-            socket.BTPROTO_HCI,  # type: ignore
+            socket.BTPROTO_HCI,  # type: ignore[attr-defined]
         )
     except AttributeError as error:
         # Not supported on this platform
@@ -78,7 +80,7 @@ async def open_hci_socket_transport(spec: str | None) -> Transport:
     bind_address = struct.pack(
         # pylint: disable=no-member
         '<HHH',
-        socket.AF_BLUETOOTH,
+        socket.AF_BLUETOOTH,  # type: ignore[attr-defined]
         adapter_index,
         HCI_CHANNEL_USER,
     )

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

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

setup.cfg

@@ -35,7 +35,7 @@ install_requires =
     appdirs >= 1.4; platform_system!='Emscripten'
     bt-test-interfaces >= 0.0.2; platform_system!='Emscripten'
     click == 8.1.3; platform_system!='Emscripten'
-    cryptography == 39; platform_system!='Emscripten'
+    cryptography == 39
     grpcio == 1.57.0; platform_system!='Emscripten'
     humanize >= 4.6.0; platform_system!='Emscripten'
     libusb1 >= 2.0.1; platform_system!='Emscripten'
@@ -84,7 +84,7 @@ development =
     black == 22.10
     grpcio-tools >= 1.57.0
     invoke >= 1.7.3
-    mypy == 1.2.0
+    mypy == 1.5.0
     nox >= 2022
     pylint == 2.15.8
     types-appdirs >= 1.4.3

web/bumble.js

@@ -74,7 +74,6 @@ export async function loadBumble(pyodide, bumblePackage) {
     await pyodide.loadPackage("micropip");
     await pyodide.runPythonAsync(`
         import micropip
-        await micropip.install("cryptography")
         await micropip.install("${bumblePackage}")
         package_list = micropip.list()
         print(package_list)