diff options
Diffstat (limited to 'kvmd/plugins/hid/bt')
| -rw-r--r-- | kvmd/plugins/hid/bt/__init__.py | 17 | ||||
| -rw-r--r-- | kvmd/plugins/hid/bt/bluez.py | 6 | ||||
| -rw-r--r-- | kvmd/plugins/hid/bt/server.py | 18 |
3 files changed, 16 insertions, 25 deletions
diff --git a/kvmd/plugins/hid/bt/__init__.py b/kvmd/plugins/hid/bt/__init__.py index ce33fd26..442822be 100644 --- a/kvmd/plugins/hid/bt/__init__.py +++ b/kvmd/plugins/hid/bt/__init__.py @@ -23,11 +23,8 @@ import multiprocessing import time -from typing import Tuple -from typing import Dict from typing import Iterable from typing import AsyncGenerator -from typing import Optional from ....logging import get_logger @@ -81,7 +78,7 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes select_timeout: float, ) -> None: - self.__proc: Optional[multiprocessing.Process] = None + self.__proc: (multiprocessing.Process | None) = None self.__stop_event = multiprocessing.Event() self.__notifier = aiomulti.AioProcessNotifier() @@ -104,7 +101,7 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes ) @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "manufacturer": Option("PiKVM"), "product": Option("HID Device"), @@ -128,9 +125,9 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes self.__proc = multiprocessing.Process(target=self.__server_worker, daemon=True) self.__proc.start() - async def get_state(self) -> Dict: + async def get_state(self) -> dict: state = await self.__server.get_state() - outputs: Dict = {"available": [], "active": ""} + outputs: dict = {"available": [], "active": ""} return { "online": True, "busy": False, @@ -151,8 +148,8 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes }, } - async def poll_state(self) -> AsyncGenerator[Dict, None]: - prev_state: Dict = {} + async def poll_state(self) -> AsyncGenerator[dict, None]: + prev_state: dict = {} while True: state = await self.get_state() if state != prev_state: @@ -175,7 +172,7 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes # ===== - def send_key_events(self, keys: Iterable[Tuple[str, bool]]) -> None: + def send_key_events(self, keys: Iterable[tuple[str, bool]]) -> None: for (key, state) in keys: self.__server.queue_event(make_keyboard_event(key, state)) diff --git a/kvmd/plugins/hid/bt/bluez.py b/kvmd/plugins/hid/bt/bluez.py index 1b28c3d1..a2c0499e 100644 --- a/kvmd/plugins/hid/bt/bluez.py +++ b/kvmd/plugins/hid/bt/bluez.py @@ -22,8 +22,6 @@ import types -from typing import Type -from typing import Optional from typing import Any import dbus @@ -56,7 +54,7 @@ class BluezIface: self.__pairing_required = pairing_required self.__auth_required = auth_required - self.__bus: Optional[dbus.SystemBus] = None + self.__bus: (dbus.SystemBus | None) = None def get_address(self) -> str: return self.__get_prop("Address") @@ -100,7 +98,7 @@ class BluezIface: def __exit__( self, - _exc_type: Type[BaseException], + _exc_type: type[BaseException], _exc: BaseException, _tb: types.TracebackType, ) -> None: diff --git a/kvmd/plugins/hid/bt/server.py b/kvmd/plugins/hid/bt/server.py index 9cfbce53..1d54baa5 100644 --- a/kvmd/plugins/hid/bt/server.py +++ b/kvmd/plugins/hid/bt/server.py @@ -29,11 +29,7 @@ import contextlib import queue from typing import Literal -from typing import List -from typing import Dict -from typing import Set from typing import Generator -from typing import Optional from ....logging import get_logger @@ -72,8 +68,8 @@ _SockAttrT = Literal["ctl_sock", "int_sock"] @dataclasses.dataclass class _BtClient: addr: str - ctl_sock: Optional[socket.socket] = None - int_sock: Optional[socket.socket] = None + ctl_sock: (socket.socket | None) = None + int_sock: (socket.socket | None) = None # ===== @@ -104,8 +100,8 @@ class BtServer: # pylint: disable=too-many-instance-attributes self.__stop_event = stop_event - self.__clients: Dict[str, _BtClient] = {} - self.__to_read: Set[socket.socket] = set() + self.__clients: dict[str, _BtClient] = {} + self.__to_read: set[socket.socket] = set() self.__events_queue: "multiprocessing.Queue[BaseEvent]" = multiprocessing.Queue() @@ -115,8 +111,8 @@ class BtServer: # pylint: disable=too-many-instance-attributes "scroll": False, "num": False, }, notifier) - self.__modifiers: Set[UsbKey] = set() - self.__keys: List[Optional[UsbKey]] = [None] * 6 + self.__modifiers: set[UsbKey] = set() + self.__keys: list[UsbKey | None] = [None] * 6 self.__mouse_buttons = 0 def run(self) -> None: @@ -132,7 +128,7 @@ class BtServer: # pylint: disable=too-many-instance-attributes self.__close_all_clients(no_change_public=True) self.__set_public(False) - async def get_state(self) -> Dict: + async def get_state(self) -> dict: return (await self.__state_flags.get()) def queue_event(self, event: BaseEvent) -> None: |