Upgrade to Python 3.9 (bullseye) (#60)

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ jobs:
  - name: Setup Python
  uses: actions/setup-python@v2
  with:
- python-version: 3.7
+ python-version: 3.9
  - name: Install libgirepository
  run: sudo apt install -y libgirepository1.0-dev
  - name: Install dependencies

adapter.py

@@ -2,7 +2,7 @@
 
 import asyncio
 from datetime import datetime, timedelta
-from typing import Awaitable, Callable, Dict, List, Optional
+from typing import Awaitable, Callable, Optional, TypedDict, cast
 
 from dasbus.connection import InterfaceProxy, SystemMessageBus
 import dasbus.typing as dt
@@ -12,19 +12,19 @@
 from hid_devices import HIDDeviceRegistry
 
 
-"""class _Device(TypedDict):
+class _Device(TypedDict):
  path: str
  address: str
  alias: str
  paired: bool
  trusted: bool
  connected: bool
  host: bool
+
+
 class _GetDevices(TypedDict):
- devices: List[Dict[str, str]]
- scanning: bool"""
-_Device = Dict[str, object]
-_GetDevices = Dict[str, object]
+ devices: list[_Device]
+ scanning: bool
 
 
 DBUS_PATH_PROFILE = '/ruundii/btkb_profile'
@@ -119,7 +119,7 @@ async def wait_till_adapter_present_then_init(self) -> None:
  self.initialising_adapter = False
 
 
- def interfaces_added(self, obj_name: str, interfaces: List[str]) -> None:
+ def interfaces_added(self, obj_name: str, interfaces: list[str]) -> None:
  self.on_interface_changed()
  if not self.adapter_exists():
  return
@@ -133,7 +133,7 @@ def interfaces_added(self, obj_name: str, interfaces: List[str]) -> None:
  elif INPUT_DEVICE_INTERFACE in interfaces:
  self.bluetooth_devices.add_device(obj_name, False)
 
- def interfaces_removed(self, obj_name: str, interfaces: List[str]) -> None:
+ def interfaces_removed(self, obj_name: str, interfaces: list[str]) -> None:
  if(obj_name==ADAPTER_OBJECT or obj_name==ROOT_OBJECT):
  self.adapter = None
  self.bluetooth_devices.remove_devices()
@@ -205,18 +205,18 @@ def get_devices(self) -> _GetDevices:
  return {"devices": [], "scanning": False }
  om = self.bus.get_proxy(service_name="org.bluez", object_path="/", interface_name=OBJECT_MANAGER_INTERFACE)
  objs = om.GetManagedObjects()
- devices: List[_Device] = []
+ devices: list[_Device] = []
  for path in objs:
  obj = objs[path]
  if DEVICE_INTERFACE in obj:
  dev = obj[DEVICE_INTERFACE]
  devices.append({
  "path" : path,
- "address" : dt.unwrap_variant(dev["Address"]),
- "alias": dt.unwrap_variant(dev["Alias"]),
- "paired": dt.unwrap_variant(dev["Paired"]),
- "trusted": dt.unwrap_variant(dev["Trusted"]),
- "connected": dt.unwrap_variant(dev["Connected"]),
+ "address" : cast(str, dt.unwrap_variant(dev["Address"])),
+ "alias": cast(str, dt.unwrap_variant(dev["Alias"])),
+ "paired": cast(bool, dt.unwrap_variant(dev["Paired"])),
+ "trusted": cast(bool, dt.unwrap_variant(dev["Trusted"])),
+ "connected": cast(bool, dt.unwrap_variant(dev["Connected"])),
  "host" : INPUT_HOST_INTERFACE in obj
  })
  return {"devices": devices, "scanning": self.adapter.Discovering}

agent.py

@@ -2,20 +2,19 @@
 
 import asyncio
 from datetime import datetime, timedelta
-from typing import Callable, Dict, Optional
+from typing import Callable, Optional, TypedDict
 
 import dasbus.typing as dt
 from dasbus.server.interface import dbus_interface
 from dasbus.connection import SystemMessageBus
 
 
-"""class Action(TypedDict, total=False):
+class Action(TypedDict, total=False):
  action: str
  device: dt.ObjPath
  entered: dt.UInt16
- passkey: dt.UInt32
- pincode: str"""
-Action = Dict[str, object]
+ passkey: str
+ pincode: str
 
 
 bus = SystemMessageBus()
@@ -66,11 +65,6 @@ def RequestPasskey(self, device: dt.ObjPath) -> dt.UInt32:
  passkey = int(ask("Enter passkey: "))
  return dt.UInt32(passkey)
 
- def DisplayPasskey(self, device: dt.ObjPath, passkey: dt.UInt32, entered: dt.UInt16) -> None:
- print("DisplayPasskey (%s, %06u entered %u)" %
- (device, passkey, entered))
- self.on_agent_action({'action':'display_passkey', 'passkey':passkey, 'device':device, 'entered':entered})
-
  def DisplayPinCode(self, device: dt.ObjPath, pincode: dt.Str) -> None:
  print("DisplayPinCode (%s, %s)" % (device, pincode))
  self.on_agent_action({'action':'display_pin_code', 'pincode':pincode, 'device':device})

bluetooth_devices.py

@@ -3,7 +3,7 @@
 import asyncio
 import socket
 import os
-from typing import Awaitable, Callable, Dict, List, Optional, TYPE_CHECKING
+from typing import Awaitable, Callable, Optional, TYPE_CHECKING
 
 from dasbus.connection import SystemMessageBus
 
@@ -162,9 +162,9 @@ class BluetoothDeviceRegistry:
  def __init__(self, bus: SystemMessageBus, loop: asyncio.AbstractEventLoop):
  self.bus = bus
  self.loop = loop
- self.all: Dict[str, BluetoothDevice] = {}
- self.connected_hosts: List[BluetoothDevice] = []
- self.connected_devices: List[BluetoothDevice] = []
+ self.all: dict[str, BluetoothDevice] = {}
+ self.connected_hosts: list[BluetoothDevice] = []
+ self.connected_devices: list[BluetoothDevice] = []
  self.on_devices_changed_handler: Optional[Callable[[], Awaitable[None]]] = None
  self.hid_devices: Optional["HIDDeviceRegistry"] = None
  self.current_host_index = 0
@@ -233,7 +233,7 @@ def remove_device(self, device_object_path: str) -> None:
  def switch_host(self) -> None:
  self.current_host_index = (self.current_host_index + 1) % len(self.connected_hosts)
 
- def __get_current_host_as_list(self) -> List[BluetoothDevice]:
+ def __get_current_host_as_list(self) -> list[BluetoothDevice]:
  if len(self.connected_hosts) <= self.current_host_index:
  return []
  return [self.connected_hosts[self.current_host_index]]
@@ -242,7 +242,7 @@ def send_message(self, msg: bytes, send_to_hosts: bool, is_control_channel: bool
  if IGNORE_INPUT_DEVICES and not send_to_hosts and not is_control_channel and self.hid_devices is not None:
  asyncio.run_coroutine_threadsafe(self.hid_devices.send_message_to_devices(msg), loop=self.loop)
  return
- targets: List[BluetoothDevice] = self.__get_current_host_as_list() if send_to_hosts else self.connected_devices
+ targets: list[BluetoothDevice] = self.__get_current_host_as_list() if send_to_hosts else self.connected_devices
  for target in list(targets):
  try:
  socket = target.control_socket if is_control_channel else target.interrupt_socket

compatibility_device.py

@@ -1,5 +1,4 @@
 import asyncio
-from typing import List
 
 from evdev import InputDevice, categorize, ecodes
 from hidtools.uhid import UHIDDevice
@@ -149,8 +148,8 @@ def __init__(self, loop: asyncio.AbstractEventLoop, device_path: str):
  self.hidraw_device.phys = "0"
  self.hidraw_device.rdesc = bytearray.fromhex(
  "05010906a1018501050719e029e715002501750195088102950175088103950575010508190129059102950175039103950675081500256d05071900296d8100c0050C0901A1018502050C150025017501950709B509B609B709CD09E209E909EA810295018101C0")
- self.pressed_keys:List[int]=[]
- self.pressed_consumer_keys:List[int]=[]
+ self.pressed_keys: list[int] = []
+ self.pressed_consumer_keys: list[int] = []
 
  self.hidraw_device.create_kernel_device()
  asyncio.run_coroutine_threadsafe(self.__read_events(),self.loop)

hid_devices.py

@@ -7,7 +7,7 @@
 import json
 import re
 import time
-from typing import Awaitable, Callable, Dict, List, Optional, cast
+from typing import Awaitable, Callable, Literal, Optional, TypedDict, cast
 
 import evdev
 from watchgod import awatch, AllWatcher
@@ -19,16 +19,19 @@
 from mouse_g502_message_filter import G502MessageFilter
 from mouse_message_filter import MouseMessageFilter
 from mouse_mx510_message_filter import MX510MessageFilter
-from typing import Dict, List
 
 
-# TODO PY38: Swap below code and remove all casts.
-"""class _Device(TypedDict):
+class __Device(TypedDict, total=False):
+ capture: bool
+ filter: str
+
+
+class _Device(__Device):
  id: str
  instance: str
  name: str
  hidraw: str
- events: List[str]
+ events: list[str]
  compatibility_mode: bool
 
 
@@ -38,20 +41,22 @@ class _InputDevice(TypedDict):
  phys: str
  compatibility_mode: bool
 
+
 class _HIDDevices(TypedDict):
- devices: List[_Device]
- filters: List[Dict[str, str]]
- input_devices: List[_InputDevice]
-"""
-_Device = Dict[str, object]
-_InputDevice = Dict[str, object]
-_HIDDevices = Dict[str, List[Dict[str, object]]]
+ devices: list[_Device]
+ filters: list[dict[str, str]]
+ input_devices: list[_InputDevice]
+
+
+class _DeviceConfig(TypedDict, total=False):
+ capture: bool
+ filter: str
 
 
 DEVICES_CONFIG_FILE_NAME = 'devices_config.json'
 DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY = 'compatibility_devices'
-CAPTURE_ELEMENT = 'capture'
-FILTER_ELEMENT = 'filter'
+CAPTURE_ELEMENT: Literal['capture'] = 'capture'
+FILTER_ELEMENT: Literal['filter'] = 'filter'
 
 FILTERS = [
  {"id":"Default", "name":"Default"},
@@ -74,8 +79,8 @@ def __init__(self, device: _Device, filter: HIDMessageFilter,
  self.device_id = device["instance"]
  self.device_class = device["id"]
  self.name = device["name"]
- self.hidraw = cast(str, device["hidraw"])
- self.events = cast(List[str], device["events"])
+ self.hidraw = device["hidraw"]
+ self.events = device["events"]
  self.events_devices = []
  for event in self.events:
  event_device = evdev.InputDevice('/dev/input/'+event)
@@ -170,13 +175,13 @@ def __init__(self, loop: asyncio.AbstractEventLoop):
  self.loop = loop
  try:
  with open(DEVICES_CONFIG_FILE_NAME) as devices_config:
- self.devices_config: Dict[str, Dict[str, object]] = json.load(devices_config)
+ self.devices_config: dict[str, _DeviceConfig] = json.load(devices_config)
  except Exception:
  self.devices_config = {}
- self.devices: List[_Device] = []
- self.capturing_devices: Dict[str, HIDDevice] = {}
- self.input_devices: List[_InputDevice] = []
- self.compatibility_mode_devices: Dict[str, CompatibilityModeDevice] = {}
+ self.devices: list[_Device] = []
+ self.capturing_devices: dict[str, HIDDevice] = {}
+ self.input_devices: list[_InputDevice] = []
+ self.compatibility_mode_devices: dict[str, CompatibilityModeDevice] = {}
  asyncio.run_coroutine_threadsafe(self.__watch_device_changes(), loop=self.loop)
  self.on_devices_changed_handler: Optional[Callable[[], Awaitable[None]]] = None
  self.__scan_devices()
@@ -224,7 +229,7 @@ def _filter(d: evdev.InputDevice) -> bool:
  comp_device.finalise()
  del comp_device
 
- devs: List[_Device] = []
+ devs: list[_Device] = []
  devs_dict = {}
  devs_in_compatibility_mode = []
  for device in os.listdir('/sys/bus/hid/devices'):
@@ -242,7 +247,7 @@ def _filter(d: evdev.InputDevice) -> bool:
  input_events = [e for e in os.listdir('/sys/bus/hid/devices/' + device + '/input/'+input) if e.startswith('event')]
  for event in input_events:
  for input_device in self.input_devices:
- if input_device["compatibility_mode"] and cast(str, input_device["path"]).find(event)>=0:
+ if input_device["compatibility_mode"] and input_device["path"].find(event)>=0:
  compatibility_mode = True
  break
  events.extend(input_events)
@@ -266,9 +271,9 @@ def _filter(d: evdev.InputDevice) -> bool:
  del hid_device
 
  for dev_dict in devs:
- if dev_dict["instance"] not in self.capturing_devices and self.__is_configured_capturing_device(cast(str, dev_dict["id"])) and dev_dict["instance"] not in devs_in_compatibility_mode:
+ if dev_dict["instance"] not in self.capturing_devices and self.__is_configured_capturing_device(dev_dict["id"]) and dev_dict["instance"] not in devs_in_compatibility_mode:
  #create capturing device
- self.capturing_devices[cast(str, dev_dict["instance"])] = HIDDevice(dev_dict, self.__get_configured_device_filter(cast(str, dev_dict["id"])), self.loop, self)
+ self.capturing_devices[dev_dict["instance"]] = HIDDevice(dev_dict, self.__get_configured_device_filter(dev_dict["id"]), self.loop, self)
  self.devices = devs
 
 
@@ -291,9 +296,9 @@ def set_compatibility_device(self, device_path: str, compatibility_state: bool)
  if DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY not in self.devices_config:
  self.devices_config[DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY] = [] # type: ignore[assignment]
  if compatibility_state and device_path not in self.devices_config[DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY]:
- cast(List[str], self.devices_config[DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY]).append(device_path)
+ cast(list[str], self.devices_config[DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY]).append(device_path)
  elif not compatibility_state and device_path in self.devices_config[DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY]:
- cast(List[str], self.devices_config[DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY]).remove(device_path)
+ cast(list[str], self.devices_config[DEVICES_CONFIG_COMPATIBILITY_DEVICE_KEY]).remove(device_path)
  self.__save_config()
  self.__scan_devices()
 
@@ -304,23 +309,23 @@ def __save_config(self) -> None:
  def __is_configured_capturing_device(self, device_id: str) -> bool:
  if device_id in self.devices_config:
  if CAPTURE_ELEMENT in self.devices_config[device_id]:
- return cast(bool, self.devices_config[device_id][CAPTURE_ELEMENT])
+ return self.devices_config[device_id][CAPTURE_ELEMENT]
  return False
 
  def __get_configured_device_filter(self, device_id: str) -> HIDMessageFilter:
  if device_id in self.devices_config:
  if FILTER_ELEMENT in self.devices_config[device_id]:
- filter_id = cast(str, self.devices_config[device_id][FILTER_ELEMENT])
+ filter_id = self.devices_config[device_id][FILTER_ELEMENT]
  return FILTER_INSTANCES[filter_id]
  return FILTER_INSTANCES["Default"]
 
  def get_hid_devices_with_config(self) -> _HIDDevices:
  for device in self.devices:
  if device["id"] in self.devices_config:
- if CAPTURE_ELEMENT in self.devices_config[cast(str, device["id"])]:
- device[CAPTURE_ELEMENT] = self.devices_config[cast(str, device["id"])][CAPTURE_ELEMENT]
+ if CAPTURE_ELEMENT in self.devices_config[device["id"]]:
+ device[CAPTURE_ELEMENT] = self.devices_config[device["id"]][CAPTURE_ELEMENT]
  else:
  device[CAPTURE_ELEMENT] = False
- if FILTER_ELEMENT in self.devices_config[cast(str, device["id"])]:
- device[FILTER_ELEMENT] = self.devices_config[cast(str, device["id"])][FILTER_ELEMENT]
- return {"devices": self.devices, "filters": cast(List[Dict[str, object]], FILTERS), "input_devices": self.input_devices}
+ if FILTER_ELEMENT in self.devices_config[device["id"]]:
+ device[FILTER_ELEMENT] = self.devices_config[device["id"]][FILTER_ELEMENT]
+ return {"devices": self.devices, "filters": FILTERS, "input_devices": self.input_devices}

requirements.txt

@@ -6,7 +6,7 @@ bitarray # TODO: remove
 dasbus
 evdev
 hid-tools
-mypy==1.4.1
+mypy==1.5.1
 PyGObject
 pyudev
 watchgod