diff options
author | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
---|---|---|
committer | Maxim Devaev <[email protected]> | 2022-09-04 18:08:40 +0300 |
commit | ee3e224e396494cd0d69bb6167087a071a20349c (patch) | |
tree | 5becd28570e58a03c6e1e231d0db24c264a73f88 /kvmd/plugins | |
parent | 4b75221e9470b4a009955d7677f16adf8e23e302 (diff) |
new typing style
Diffstat (limited to 'kvmd/plugins')
45 files changed, 214 insertions, 329 deletions
diff --git a/kvmd/plugins/__init__.py b/kvmd/plugins/__init__.py index 9ee8eede..f3afd693 100644 --- a/kvmd/plugins/__init__.py +++ b/kvmd/plugins/__init__.py @@ -23,8 +23,6 @@ import importlib import functools -from typing import Dict -from typing import Type from typing import Any @@ -44,12 +42,12 @@ class BasePlugin: return name[name.rindex(".") + 1:] @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return {} # pragma: nocover @functools.lru_cache() -def get_plugin_class(sub: str, name: str) -> Type[BasePlugin]: +def get_plugin_class(sub: str, name: str) -> type[BasePlugin]: assert sub assert name if name.startswith("_"): diff --git a/kvmd/plugins/atx/__init__.py b/kvmd/plugins/atx/__init__.py index f7e85057..4ac183c5 100644 --- a/kvmd/plugins/atx/__init__.py +++ b/kvmd/plugins/atx/__init__.py @@ -20,9 +20,7 @@ # ========================================================================== # -from typing import Dict from typing import AsyncGenerator -from typing import Type from ...errors import OperationError from ...errors import IsBusyError @@ -47,10 +45,10 @@ class AtxIsBusyError(IsBusyError, AtxError): # ===== class BaseAtx(BasePlugin): - async def get_state(self) -> Dict: + async def get_state(self) -> dict: raise NotImplementedError - async def poll_state(self) -> AsyncGenerator[Dict, None]: + async def poll_state(self) -> AsyncGenerator[dict, None]: yield {} raise NotImplementedError @@ -84,5 +82,5 @@ class BaseAtx(BasePlugin): # ===== -def get_atx_class(name: str) -> Type[BaseAtx]: +def get_atx_class(name: str) -> type[BaseAtx]: return get_plugin_class("atx", name) # type: ignore diff --git a/kvmd/plugins/atx/disabled.py b/kvmd/plugins/atx/disabled.py index 4f09cba4..9bfb28b2 100644 --- a/kvmd/plugins/atx/disabled.py +++ b/kvmd/plugins/atx/disabled.py @@ -20,7 +20,6 @@ # ========================================================================== # -from typing import Dict from typing import AsyncGenerator from ... import aiotools @@ -37,7 +36,7 @@ class AtxDisabledError(AtxOperationError): # ===== class Plugin(BaseAtx): - async def get_state(self) -> Dict: + async def get_state(self) -> dict: return { "enabled": False, "busy": False, @@ -47,7 +46,7 @@ class Plugin(BaseAtx): }, } - async def poll_state(self) -> AsyncGenerator[Dict, None]: + async def poll_state(self) -> AsyncGenerator[dict, None]: while True: yield (await self.get_state()) await aiotools.wait_infinite() diff --git a/kvmd/plugins/atx/gpio.py b/kvmd/plugins/atx/gpio.py index 8d41e53d..e7089481 100644 --- a/kvmd/plugins/atx/gpio.py +++ b/kvmd/plugins/atx/gpio.py @@ -20,9 +20,7 @@ # ========================================================================== # -from typing import Dict from typing import AsyncGenerator -from typing import Optional import gpiod @@ -76,9 +74,9 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes self.__notifier = aiotools.AioNotifier() self.__region = aiotools.AioExclusiveRegion(AtxIsBusyError, self.__notifier) - self.__chip: Optional[gpiod.Chip] = None - self.__power_switch_line: Optional[gpiod.Line] = None - self.__reset_switch_line: Optional[gpiod.Line] = None + self.__chip: (gpiod.Chip | None) = None + self.__power_switch_line: (gpiod.Line | None) = None + self.__reset_switch_line: (gpiod.Line | None) = None self.__reader = aiogp.AioReader( path=self.__device_path, @@ -91,7 +89,7 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes ) @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="device_path"), @@ -122,7 +120,7 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes self.__reset_switch_line = self.__chip.get_line(self.__reset_switch_pin) self.__reset_switch_line.request("kvmd::atx::reset_switch", gpiod.LINE_REQ_DIR_OUT, default_vals=[0]) - async def get_state(self) -> Dict: + async def get_state(self) -> dict: return { "enabled": True, "busy": self.__region.is_busy(), @@ -132,8 +130,8 @@ class Plugin(BaseAtx): # 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: diff --git a/kvmd/plugins/auth/__init__.py b/kvmd/plugins/auth/__init__.py index 59a59a13..c532829c 100644 --- a/kvmd/plugins/auth/__init__.py +++ b/kvmd/plugins/auth/__init__.py @@ -20,8 +20,6 @@ # ========================================================================== # -from typing import Type - from .. import BasePlugin from .. import get_plugin_class @@ -36,5 +34,5 @@ class BaseAuthService(BasePlugin): # ===== -def get_auth_service_class(name: str) -> Type[BaseAuthService]: +def get_auth_service_class(name: str) -> type[BaseAuthService]: return get_plugin_class("auth", name) # type: ignore diff --git a/kvmd/plugins/auth/htpasswd.py b/kvmd/plugins/auth/htpasswd.py index 51364a18..1e3b1010 100644 --- a/kvmd/plugins/auth/htpasswd.py +++ b/kvmd/plugins/auth/htpasswd.py @@ -20,8 +20,6 @@ # ========================================================================== # -from typing import Dict - import passlib.apache from ...yamlconf import Option @@ -37,7 +35,7 @@ class Plugin(BaseAuthService): self.__path = path @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "file": Option("/etc/kvmd/htpasswd", type=valid_abs_file, unpack_as="path"), } diff --git a/kvmd/plugins/auth/http.py b/kvmd/plugins/auth/http.py index bfa3aa9f..aa80c2eb 100644 --- a/kvmd/plugins/auth/http.py +++ b/kvmd/plugins/auth/http.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import Dict -from typing import Optional - import aiohttp import aiohttp.web @@ -57,10 +54,10 @@ class Plugin(BaseAuthService): self.__passwd = passwd self.__timeout = timeout - self.__http_session: Optional[aiohttp.ClientSession] = None + self.__http_session: (aiohttp.ClientSession | None) = None @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "url": Option("http://localhost/auth"), "verify": Option(True, type=valid_bool), @@ -102,7 +99,7 @@ class Plugin(BaseAuthService): def __ensure_http_session(self) -> aiohttp.ClientSession: if not self.__http_session: - kwargs: Dict = {} + kwargs: dict = {} if self.__user: kwargs["auth"] = aiohttp.BasicAuth(login=self.__user, password=self.__passwd) if not self.__verify: diff --git a/kvmd/plugins/auth/pam.py b/kvmd/plugins/auth/pam.py index aef38102..6a83efa0 100644 --- a/kvmd/plugins/auth/pam.py +++ b/kvmd/plugins/auth/pam.py @@ -23,9 +23,6 @@ import asyncio import pwd -from typing import List -from typing import Dict - import pam from ...yamlconf import Option @@ -45,8 +42,8 @@ class Plugin(BaseAuthService): def __init__( # pylint: disable=super-init-not-called self, service: str, - allow_users: List[str], - deny_users: List[str], + allow_users: list[str], + deny_users: list[str], allow_uids_at: int, ) -> None: @@ -58,7 +55,7 @@ class Plugin(BaseAuthService): self.__lock = asyncio.Lock() @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "service": Option("login"), "allow_users": Option([], type=valid_users_list), diff --git a/kvmd/plugins/auth/radius.py b/kvmd/plugins/auth/radius.py index b3bd94cc..d7c6390e 100644 --- a/kvmd/plugins/auth/radius.py +++ b/kvmd/plugins/auth/radius.py @@ -22,8 +22,6 @@ import io -from typing import Dict - import pyrad.client import pyrad.packet import pyrad.dictionary @@ -413,7 +411,7 @@ class Plugin(BaseAuthService): self.__timeout = timeout @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "host": Option("localhost", type=valid_ip_or_host), "port": Option(1812, type=valid_port), diff --git a/kvmd/plugins/hid/__init__.py b/kvmd/plugins/hid/__init__.py index a0b582c7..2a6c075e 100644 --- a/kvmd/plugins/hid/__init__.py +++ b/kvmd/plugins/hid/__init__.py @@ -20,12 +20,8 @@ # ========================================================================== # -from typing import Tuple -from typing import Dict from typing import Iterable from typing import AsyncGenerator -from typing import Type -from typing import Optional from .. import BasePlugin from .. import get_plugin_class @@ -36,10 +32,10 @@ class BaseHid(BasePlugin): def sysprep(self) -> None: raise NotImplementedError - async def get_state(self) -> Dict: + async def get_state(self) -> dict: raise NotImplementedError - async def poll_state(self) -> AsyncGenerator[Dict, None]: + async def poll_state(self) -> AsyncGenerator[dict, None]: yield {} raise NotImplementedError @@ -51,7 +47,7 @@ class BaseHid(BasePlugin): # ===== - def send_key_events(self, keys: Iterable[Tuple[str, bool]]) -> None: + def send_key_events(self, keys: Iterable[tuple[str, bool]]) -> None: raise NotImplementedError def send_mouse_button_event(self, button: str, state: bool) -> None: @@ -68,7 +64,7 @@ class BaseHid(BasePlugin): def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None: raise NotImplementedError - def set_params(self, keyboard_output: Optional[str]=None, mouse_output: Optional[str]=None) -> None: + def set_params(self, keyboard_output: (str | None)=None, mouse_output: (str | None)=None) -> None: _ = keyboard_output _ = mouse_output @@ -80,5 +76,5 @@ class BaseHid(BasePlugin): # ===== -def get_hid_class(name: str) -> Type[BaseHid]: +def get_hid_class(name: str) -> type[BaseHid]: return get_plugin_class("hid", name) # type: ignore diff --git a/kvmd/plugins/hid/_mcu/__init__.py b/kvmd/plugins/hid/_mcu/__init__.py index 0de017df..ff92136c 100644 --- a/kvmd/plugins/hid/_mcu/__init__.py +++ b/kvmd/plugins/hid/_mcu/__init__.py @@ -25,13 +25,9 @@ import contextlib import queue import time -from typing import Tuple -from typing import List -from typing import Dict from typing import Iterable from typing import Generator from typing import AsyncGenerator -from typing import Optional from ....logging import get_logger @@ -144,7 +140,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many- self.__stop_event = multiprocessing.Event() @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "gpio_device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="gpio_device_path"), "reset_pin": Option(4, type=valid_gpio_pin_optional), @@ -162,7 +158,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many- get_logger(0).info("Starting HID daemon ...") self.start() - async def get_state(self) -> Dict: + async def get_state(self) -> dict: state = await self.__state_flags.get() online = bool(state["online"]) pong = (state["status"] >> 16) & 0xFF @@ -174,8 +170,8 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many- if online and active_mouse in ["usb_rel", "ps2"]: absolute = False - keyboard_outputs: Dict = {"available": [], "active": ""} - mouse_outputs: Dict = {"available": [], "active": ""} + keyboard_outputs: dict = {"available": [], "active": ""} + mouse_outputs: dict = {"available": [], "active": ""} if outputs1 & 0b10000000: # Dynamic if outputs2 & 0b00000001: # USB @@ -222,8 +218,8 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many- }, } - 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: @@ -244,7 +240,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many- # ===== - 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.__queue_event(KeyEvent(key, state)) @@ -260,8 +256,8 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many- def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None: self.__queue_event(MouseWheelEvent(delta_x, delta_y)) - def set_params(self, keyboard_output: Optional[str]=None, mouse_output: Optional[str]=None) -> None: - events: List[BaseEvent] = [] + def set_params(self, keyboard_output: (str | None)=None, mouse_output: (str | None)=None) -> None: + events: list[BaseEvent] = [] if keyboard_output is not None: events.append(SetKeyboardOutputEvent(keyboard_output)) if mouse_output is not None: @@ -346,7 +342,7 @@ class BaseMcuHid(BaseHid, multiprocessing.Process): # pylint: disable=too-many- def __process_request(self, conn: BasePhyConnection, request: bytes) -> bool: # pylint: disable=too-many-branches logger = get_logger() - error_messages: List[str] = [] + error_messages: list[str] = [] live_log_errors = False common_retries = self.__common_retries diff --git a/kvmd/plugins/hid/_mcu/gpio.py b/kvmd/plugins/hid/_mcu/gpio.py index e3c7f7b7..f4985061 100644 --- a/kvmd/plugins/hid/_mcu/gpio.py +++ b/kvmd/plugins/hid/_mcu/gpio.py @@ -23,9 +23,6 @@ import types import time -from typing import Type -from typing import Optional - import gpiod from ....logging import get_logger @@ -46,8 +43,8 @@ class Gpio: self.__reset_inverted = reset_inverted self.__reset_delay = reset_delay - self.__chip: Optional[gpiod.Chip] = None - self.__reset_line: Optional[gpiod.Line] = None + self.__chip: (gpiod.Chip | None) = None + self.__reset_line: (gpiod.Line | None) = None def __enter__(self) -> None: if self.__reset_pin >= 0: @@ -59,7 +56,7 @@ class Gpio: def __exit__( self, - _exc_type: Type[BaseException], + _exc_type: type[BaseException], _exc: BaseException, _tb: types.TracebackType, ) -> None: 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: diff --git a/kvmd/plugins/hid/otg/__init__.py b/kvmd/plugins/hid/otg/__init__.py index 41fcff29..3c0ad221 100644 --- a/kvmd/plugins/hid/otg/__init__.py +++ b/kvmd/plugins/hid/otg/__init__.py @@ -20,11 +20,8 @@ # ========================================================================== # -from typing import Tuple -from typing import Dict from typing import Iterable from typing import AsyncGenerator -from typing import Optional from typing import Any from ....logging import get_logger @@ -49,9 +46,9 @@ from .mouse import MouseProcess class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes def __init__( # pylint: disable=super-init-not-called self, - keyboard: Dict[str, Any], - mouse: Dict[str, Any], - mouse_alt: Dict[str, Any], + keyboard: dict[str, Any], + mouse: dict[str, Any], + mouse_alt: dict[str, Any], noop: bool, udc: str, # XXX: Not from options, see /kvmd/apps/kvmd/__init__.py for details ) -> None: @@ -66,8 +63,8 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes self.__keyboard_proc = KeyboardProcess(**common, **keyboard) self.__mouse_current = self.__mouse_proc = MouseProcess(**common, **mouse) - self.__mouse_alt_proc: Optional[MouseProcess] = None - self.__mouses: Dict[str, MouseProcess] = {} + self.__mouse_alt_proc: (MouseProcess | None) = None + self.__mouses: dict[str, MouseProcess] = {} if mouse_alt["device_path"]: self.__mouse_alt_proc = MouseProcess( absolute=(not mouse["absolute"]), @@ -84,7 +81,7 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes self.__mouses["usb_win98"] = self.__mouses["usb"] @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "keyboard": { "device": Option("/dev/kvmd-hid-keyboard", type=valid_abs_path, unpack_as="device_path"), @@ -121,7 +118,7 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes if self.__mouse_alt_proc: self.__mouse_alt_proc.start(udc) - async def get_state(self) -> Dict: + async def get_state(self) -> dict: keyboard_state = await self.__keyboard_proc.get_state() mouse_state = await self.__mouse_current.get_state() return { @@ -146,8 +143,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: @@ -173,7 +170,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: self.__keyboard_proc.send_key_events(keys) def send_mouse_button_event(self, button: str, state: bool) -> None: @@ -188,7 +185,7 @@ class Plugin(BaseHid): # pylint: disable=too-many-instance-attributes def send_mouse_wheel_event(self, delta_x: int, delta_y: int) -> None: self.__mouse_current.send_wheel_event(delta_x, delta_y) - def set_params(self, keyboard_output: Optional[str]=None, mouse_output: Optional[str]=None) -> None: + def set_params(self, keyboard_output: (str | None)=None, mouse_output: (str | None)=None) -> None: _ = keyboard_output if mouse_output in self.__mouses and mouse_output != self.__get_current_mouse_mode(): self.__mouse_current.send_clear_event() diff --git a/kvmd/plugins/hid/otg/device.py b/kvmd/plugins/hid/otg/device.py index 2dc11d6c..630ad186 100644 --- a/kvmd/plugins/hid/otg/device.py +++ b/kvmd/plugins/hid/otg/device.py @@ -28,9 +28,7 @@ import errno import logging import time -from typing import Dict from typing import Generator -from typing import Optional from ....logging import get_logger @@ -48,7 +46,7 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in self, name: str, read_size: int, - initial_state: Dict, + initial_state: dict, notifier: aiomulti.AioProcessNotifier, device_path: str, @@ -76,7 +74,7 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in self.__stop_event = multiprocessing.Event() self.__no_device_reported = False - self.__logger: Optional[logging.Logger] = None + self.__logger: (logging.Logger | None) = None def start(self, udc: str) -> None: # type: ignore # pylint: disable=arguments-differ self.__udc_state_path = usb.get_udc_path(udc, usb.U_STATE) @@ -125,7 +123,7 @@ class BaseDeviceProcess(multiprocessing.Process): # pylint: disable=too-many-in self.__close_device() - async def get_state(self) -> Dict: + async def get_state(self) -> dict: return (await self.__state_flags.get()) # ===== diff --git a/kvmd/plugins/hid/otg/events.py b/kvmd/plugins/hid/otg/events.py index e1f0f35c..49243dfe 100644 --- a/kvmd/plugins/hid/otg/events.py +++ b/kvmd/plugins/hid/otg/events.py @@ -23,11 +23,6 @@ import struct import dataclasses -from typing import List -from typing import Set -from typing import Optional -from typing import Union - from ....keyboard.mappings import UsbKey from ....keyboard.mappings import KEYMAP @@ -66,7 +61,7 @@ class ModifierEvent(BaseEvent): assert self.modifier.is_modifier -def make_keyboard_event(key: str, state: bool) -> Union[KeyEvent, ModifierEvent]: +def make_keyboard_event(key: str, state: bool) -> (KeyEvent | ModifierEvent): usb_key = KEYMAP[key].usb if usb_key.is_modifier: return ModifierEvent(usb_key, state) @@ -87,8 +82,8 @@ def get_led_num(flags: int) -> bool: def make_keyboard_report( - pressed_modifiers: Set[UsbKey], - pressed_keys: List[Optional[UsbKey]], + pressed_modifiers: set[UsbKey], + pressed_keys: list[UsbKey | None], ) -> bytes: modifiers = 0 @@ -168,7 +163,7 @@ def make_mouse_report( buttons: int, move_x: int, move_y: int, - wheel_x: Optional[int], + wheel_x: (int | None), wheel_y: int, ) -> bytes: diff --git a/kvmd/plugins/hid/otg/keyboard.py b/kvmd/plugins/hid/otg/keyboard.py index 2403dc2c..6bf3a184 100644 --- a/kvmd/plugins/hid/otg/keyboard.py +++ b/kvmd/plugins/hid/otg/keyboard.py @@ -20,12 +20,8 @@ # ========================================================================== # -from typing import Tuple -from typing import List -from typing import Set from typing import Iterable from typing import Generator -from typing import Optional from typing import Any from ....logging import get_logger @@ -56,8 +52,8 @@ class KeyboardProcess(BaseDeviceProcess): **kwargs, ) - self.__pressed_modifiers: Set[UsbKey] = set() - self.__pressed_keys: List[Optional[UsbKey]] = [None] * 6 + self.__pressed_modifiers: set[UsbKey] = set() + self.__pressed_keys: list[UsbKey | None] = [None] * 6 def cleanup(self) -> None: self._stop() @@ -72,7 +68,7 @@ class KeyboardProcess(BaseDeviceProcess): self._clear_queue() self._queue_event(ResetEvent()) - 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._queue_event(make_keyboard_event(key, state)) diff --git a/kvmd/plugins/hid/otg/mouse.py b/kvmd/plugins/hid/otg/mouse.py index 0c1a0946..91c2967e 100644 --- a/kvmd/plugins/hid/otg/mouse.py +++ b/kvmd/plugins/hid/otg/mouse.py @@ -21,7 +21,6 @@ from typing import Generator -from typing import Optional from typing import Any from ....logging import get_logger @@ -145,8 +144,8 @@ class MouseProcess(BaseDeviceProcess): def __make_report( self, - relative_event: Optional[MouseRelativeEvent]=None, - wheel_event: Optional[MouseWheelEvent]=None, + relative_event: (MouseRelativeEvent | None)=None, + wheel_event: (MouseWheelEvent | None)=None, ) -> bytes: if self.__absolute: diff --git a/kvmd/plugins/hid/serial.py b/kvmd/plugins/hid/serial.py index 044a8d9e..59bc79a7 100644 --- a/kvmd/plugins/hid/serial.py +++ b/kvmd/plugins/hid/serial.py @@ -23,7 +23,6 @@ import os import contextlib -from typing import Dict from typing import Generator from typing import Any @@ -85,21 +84,21 @@ class _SerialPhy(BasePhy): # ===== class Plugin(BaseMcuHid): def __init__(self, **kwargs: Any) -> None: - phy_kwargs: Dict = { + phy_kwargs: dict = { (option.unpack_as or key): kwargs.pop(option.unpack_as or key) for (key, option) in self.__get_phy_options().items() } super().__init__(phy=_SerialPhy(**phy_kwargs), **kwargs) @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { **cls.__get_phy_options(), **BaseMcuHid.get_plugin_options(), } @classmethod - def __get_phy_options(cls) -> Dict: + def __get_phy_options(cls) -> dict: return { "device": Option("/dev/kvmd-hid", type=valid_abs_path, unpack_as="device_path"), "speed": Option(115200, type=valid_tty_speed), diff --git a/kvmd/plugins/hid/spi.py b/kvmd/plugins/hid/spi.py index b3c0db00..89689a18 100644 --- a/kvmd/plugins/hid/spi.py +++ b/kvmd/plugins/hid/spi.py @@ -24,11 +24,8 @@ import os import contextlib import time -from typing import List -from typing import Dict from typing import Generator from typing import Callable -from typing import Optional from typing import Any import spidev @@ -75,7 +72,7 @@ class _SpiPhyConnection(BasePhyConnection): self.__xfer(request) - response: List[int] = [] + response: list[int] = [] deadline_ts = time.monotonic() + self.__read_timeout found = False while time.monotonic() < deadline_ts: @@ -143,7 +140,7 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes ) @contextlib.contextmanager - def __sw_cs_connected(self) -> Generator[Optional[gpiod.Line], None, None]: + def __sw_cs_connected(self) -> Generator[(gpiod.Line | None), None, None]: if self.__sw_cs_pin > 0: with contextlib.closing(gpiod.Chip(self.__gpio_device_path)) as chip: line = chip.get_line(self.__sw_cs_pin) @@ -156,19 +153,19 @@ class _SpiPhy(BasePhy): # pylint: disable=too-many-instance-attributes # ===== class Plugin(BaseMcuHid): def __init__(self, **kwargs: Any) -> None: - phy_kwargs: Dict = {key: kwargs.pop(key) for key in self.__get_phy_options()} + phy_kwargs: dict = {key: kwargs.pop(key) for key in self.__get_phy_options()} phy_kwargs["gpio_device_path"] = kwargs["gpio_device_path"] super().__init__(phy=_SpiPhy(**phy_kwargs), **kwargs) @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { **cls.__get_phy_options(), **BaseMcuHid.get_plugin_options(), } @classmethod - def __get_phy_options(cls) -> Dict: + def __get_phy_options(cls) -> dict: return { "bus": Option(-1, type=valid_int_f0), "chip": Option(-1, type=valid_int_f0), diff --git a/kvmd/plugins/msd/__init__.py b/kvmd/plugins/msd/__init__.py index 1e4a521e..21afd86b 100644 --- a/kvmd/plugins/msd/__init__.py +++ b/kvmd/plugins/msd/__init__.py @@ -24,10 +24,7 @@ import os import contextlib import time -from typing import Dict -from typing import Type from typing import AsyncGenerator -from typing import Optional import aiofiles import aiofiles.base @@ -105,7 +102,7 @@ class MsdRwNotSupported(MsdOperationError): # ===== class BaseMsdReader: - def get_state(self) -> Dict: + def get_state(self) -> dict: raise NotImplementedError() def get_total_size(self) -> int: @@ -121,7 +118,7 @@ class BaseMsdReader: class BaseMsdWriter: - def get_state(self) -> Dict: + def get_state(self) -> dict: raise NotImplementedError() def get_chunk_size(self) -> int: @@ -132,10 +129,10 @@ class BaseMsdWriter: class BaseMsd(BasePlugin): - async def get_state(self) -> Dict: + async def get_state(self) -> dict: raise NotImplementedError() - async def poll_state(self) -> AsyncGenerator[Dict, None]: + async def poll_state(self) -> AsyncGenerator[dict, None]: if self is not None: # XXX: Vulture and pylint hack raise NotImplementedError() yield @@ -150,9 +147,9 @@ class BaseMsd(BasePlugin): async def set_params( self, - name: Optional[str]=None, - cdrom: Optional[bool]=None, - rw: Optional[bool]=None, + name: (str | None)=None, + cdrom: (bool | None)=None, + rw: (bool | None)=None, ) -> None: raise NotImplementedError() @@ -168,7 +165,7 @@ class BaseMsd(BasePlugin): yield BaseMsdReader() @contextlib.asynccontextmanager - async def write_image(self, name: str, size: int, remove_incomplete: Optional[bool]) -> AsyncGenerator[BaseMsdWriter, None]: + async def write_image(self, name: str, size: int, remove_incomplete: (bool | None)) -> AsyncGenerator[BaseMsdWriter, None]: _ = name _ = size _ = remove_incomplete @@ -187,12 +184,12 @@ class MsdFileReader(BaseMsdReader): # pylint: disable=too-many-instance-attribu self.__path = path self.__chunk_size = chunk_size - self.__file: Optional[aiofiles.base.AiofilesContextManager] = None + self.__file: (aiofiles.base.AiofilesContextManager | None) = None self.__file_size = 0 self.__readed = 0 self.__tick = 0.0 - def get_state(self) -> Dict: + def get_state(self) -> dict: return { "name": self.__name, "size": self.__file_size, @@ -248,12 +245,12 @@ class MsdFileWriter(BaseMsdWriter): # pylint: disable=too-many-instance-attribu self.__sync_size = sync_size self.__chunk_size = chunk_size - self.__file: Optional[aiofiles.base.AiofilesContextManager] = None + self.__file: (aiofiles.base.AiofilesContextManager | None) = None self.__written = 0 self.__unsynced = 0 self.__tick = 0.0 - def get_state(self) -> Dict: + def get_state(self) -> dict: return { "name": self.__name, "size": self.__file_size, @@ -315,5 +312,5 @@ class MsdFileWriter(BaseMsdWriter): # pylint: disable=too-many-instance-attribu # ===== -def get_msd_class(name: str) -> Type[BaseMsd]: +def get_msd_class(name: str) -> type[BaseMsd]: return get_plugin_class("msd", name) # type: ignore diff --git a/kvmd/plugins/msd/disabled.py b/kvmd/plugins/msd/disabled.py index a1c2b752..1c96f077 100644 --- a/kvmd/plugins/msd/disabled.py +++ b/kvmd/plugins/msd/disabled.py @@ -22,9 +22,7 @@ import contextlib -from typing import Dict from typing import AsyncGenerator -from typing import Optional from ... import aiotools @@ -42,7 +40,7 @@ class MsdDisabledError(MsdOperationError): # ===== class Plugin(BaseMsd): - async def get_state(self) -> Dict: + async def get_state(self) -> dict: return { "enabled": False, "online": False, @@ -56,7 +54,7 @@ class Plugin(BaseMsd): }, } - async def poll_state(self) -> AsyncGenerator[Dict, None]: + async def poll_state(self) -> AsyncGenerator[dict, None]: while True: yield (await self.get_state()) await aiotools.wait_infinite() @@ -68,9 +66,9 @@ class Plugin(BaseMsd): async def set_params( self, - name: Optional[str]=None, - cdrom: Optional[bool]=None, - rw: Optional[bool]=None, + name: (str | None)=None, + cdrom: (bool | None)=None, + rw: (bool | None)=None, ) -> None: raise MsdDisabledError() @@ -85,7 +83,7 @@ class Plugin(BaseMsd): yield BaseMsdReader() @contextlib.asynccontextmanager - async def write_image(self, name: str, size: int, remove_incomplete: Optional[bool]) -> AsyncGenerator[BaseMsdWriter, None]: + async def write_image(self, name: str, size: int, remove_incomplete: (bool | None)) -> AsyncGenerator[BaseMsdWriter, None]: if self is not None: # XXX: Vulture and pylint hack raise MsdDisabledError() yield BaseMsdWriter() diff --git a/kvmd/plugins/msd/otg/__init__.py b/kvmd/plugins/msd/otg/__init__.py index 5a8b86a6..50f737a6 100644 --- a/kvmd/plugins/msd/otg/__init__.py +++ b/kvmd/plugins/msd/otg/__init__.py @@ -27,10 +27,7 @@ import dataclasses import functools import time -from typing import List -from typing import Dict from typing import AsyncGenerator -from typing import Optional from ....logging import get_logger @@ -77,7 +74,7 @@ class _DriveImage: @dataclasses.dataclass(frozen=True) class _DriveState: - image: Optional[_DriveImage] + image: (_DriveImage | None) cdrom: bool rw: bool @@ -86,13 +83,13 @@ class _DriveState: class _StorageState: size: int free: int - images: Dict[str, _DriveImage] + images: dict[str, _DriveImage] # ===== @dataclasses.dataclass class _VirtualDriveState: - image: Optional[_DriveImage] + image: (_DriveImage | None) connected: bool cdrom: bool rw: bool @@ -111,8 +108,8 @@ class _State: def __init__(self, notifier: aiotools.AioNotifier) -> None: self.__notifier = notifier - self.storage: Optional[_StorageState] = None - self.vd: Optional[_VirtualDriveState] = None + self.storage: (_StorageState | None) = None + self.vd: (_VirtualDriveState | None) = None self._lock = asyncio.Lock() self._region = aiotools.AioExclusiveRegion(MsdIsBusyError) @@ -143,9 +140,9 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes storage_path: str, - remount_cmd: List[str], + remount_cmd: list[str], - initial: Dict, + initial: dict, gadget: str, # XXX: Not from options, see /kvmd/apps/kvmd/__init__.py for details ) -> None: @@ -165,8 +162,8 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes self.__drive = Drive(gadget, instance=0, lun=0) - self.__reader: Optional[MsdFileReader] = None - self.__writer: Optional[MsdFileWriter] = None + self.__reader: (MsdFileReader | None) = None + self.__writer: (MsdFileWriter | None) = None self.__notifier = aiotools.AioNotifier() self.__state = _State(self.__notifier) @@ -176,7 +173,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes aiotools.run_sync(self.__reload_state(notify=False)) @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "read_chunk_size": Option(65536, type=functools.partial(valid_number, min=1024)), "write_chunk_size": Option(65536, type=functools.partial(valid_number, min=1024)), @@ -195,9 +192,9 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes }, } - async def get_state(self) -> Dict: + async def get_state(self) -> dict: async with self.__state._lock: # pylint: disable=protected-access - storage: Optional[Dict] = None + storage: (dict | None) = None if self.__state.storage: storage = dataclasses.asdict(self.__state.storage) for name in list(storage["images"]): @@ -215,7 +212,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes else: storage["uploading"] = None - vd: Optional[Dict] = None + vd: (dict | None) = None if self.__state.vd: vd = dataclasses.asdict(self.__state.vd) if vd["image"]: @@ -234,8 +231,8 @@ class Plugin(BaseMsd): # 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: @@ -267,9 +264,9 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes @aiotools.atomic_fg async def set_params( self, - name: Optional[str]=None, - cdrom: Optional[bool]=None, - rw: Optional[bool]=None, + name: (str | None)=None, + cdrom: (bool | None)=None, + rw: (bool | None)=None, ) -> None: async with self.__state.busy(): @@ -359,7 +356,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes self.__notifier.notify() @contextlib.asynccontextmanager - async def write_image(self, name: str, size: int, remove_incomplete: Optional[bool]) -> AsyncGenerator[MsdFileWriter, None]: + async def write_image(self, name: str, size: int, remove_incomplete: (bool | None)) -> AsyncGenerator[MsdFileWriter, None]: try: async with self.__state._region: # pylint: disable=protected-access path: str = "" @@ -540,7 +537,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes # ===== def __get_storage_state(self) -> _StorageState: - images: Dict[str, _DriveImage] = {} + images: dict[str, _DriveImage] = {} for name in os.listdir(self.__images_path): path = os.path.join(self.__images_path, name) if os.path.exists(path): @@ -562,7 +559,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes ) def __get_drive_state(self) -> _DriveState: - image: Optional[_DriveImage] = None + image: (_DriveImage | None) = None path = self.__drive.get_image_path() if path: name = os.path.basename(path) diff --git a/kvmd/plugins/msd/otg/drive.py b/kvmd/plugins/msd/otg/drive.py index 11af7f81..407aef9e 100644 --- a/kvmd/plugins/msd/otg/drive.py +++ b/kvmd/plugins/msd/otg/drive.py @@ -23,8 +23,6 @@ import os import errno -from typing import List - from .... import usb from .. import MsdOperationError @@ -47,7 +45,7 @@ class Drive: def is_enabled(self) -> bool: return os.path.exists(self.__profile_func_path) - def get_watchable_paths(self) -> List[str]: + def get_watchable_paths(self) -> list[str]: return [self.__lun_path, self.__profile_path] # ===== diff --git a/kvmd/plugins/msd/otg/fs.py b/kvmd/plugins/msd/otg/fs.py index 17306fe1..d5018190 100644 --- a/kvmd/plugins/msd/otg/fs.py +++ b/kvmd/plugins/msd/otg/fs.py @@ -23,8 +23,6 @@ import os import dataclasses -from typing import Optional - from ....logging import get_logger @@ -44,7 +42,7 @@ def get_file_size(path: str) -> int: return -1 -def get_fs_space(path: str, fatal: bool) -> Optional[FsSpace]: +def get_fs_space(path: str, fatal: bool) -> (FsSpace | None): try: st = os.statvfs(path) except Exception as err: diff --git a/kvmd/plugins/msd/relay/__init__.py b/kvmd/plugins/msd/relay/__init__.py index d80cd410..a93d5e34 100644 --- a/kvmd/plugins/msd/relay/__init__.py +++ b/kvmd/plugins/msd/relay/__init__.py @@ -25,9 +25,7 @@ import contextlib import dataclasses import functools -from typing import Dict from typing import AsyncGenerator -from typing import Optional from ....logging import get_logger @@ -86,16 +84,16 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes self.__gpio = Gpio(gpio_device_path, target_pin, reset_pin, reset_inverted, reset_delay) - self.__device_info: Optional[DeviceInfo] = None + self.__device_info: (DeviceInfo | None) = None self.__connected = False - self.__device_writer: Optional[MsdFileWriter] = None + self.__device_writer: (MsdFileWriter | None) = None self.__notifier = aiotools.AioNotifier() self.__region = aiotools.AioExclusiveRegion(MsdIsBusyError, self.__notifier) @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "upload_chunk_size": Option(65536, type=functools.partial(valid_number, min=1024)), "sync_chunk_size": Option(4194304, type=functools.partial(valid_number, min=1024)), @@ -121,9 +119,9 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes log = (logger.error if isinstance(err, MsdError) else logger.exception) log("MSD is offline: %s", err) - async def get_state(self) -> Dict: - storage: Optional[Dict] = None - drive: Optional[Dict] = None + async def get_state(self) -> dict: + storage: (dict | None) = None + drive: (dict | None) = None if self.__device_info: storage = { "size": self.__device_info.size, @@ -147,8 +145,8 @@ class Plugin(BaseMsd): # 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: @@ -183,9 +181,9 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes @aiotools.atomic_fg async def set_params( self, - name: Optional[str]=None, - cdrom: Optional[bool]=None, - rw: Optional[bool]=None, + name: (str | None)=None, + cdrom: (bool | None)=None, + rw: (bool | None)=None, ) -> None: async with self.__working(): @@ -226,7 +224,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes yield BaseMsdReader() @contextlib.asynccontextmanager - async def write_image(self, name: str, size: int, remove_incomplete: Optional[bool]) -> AsyncGenerator[MsdFileWriter, None]: + async def write_image(self, name: str, size: int, remove_incomplete: (bool | None)) -> AsyncGenerator[MsdFileWriter, None]: async with self.__working(): if remove_incomplete is not None: raise MsdMultiNotSupported() diff --git a/kvmd/plugins/msd/relay/drive.py b/kvmd/plugins/msd/relay/drive.py index e1301f99..51a85392 100644 --- a/kvmd/plugins/msd/relay/drive.py +++ b/kvmd/plugins/msd/relay/drive.py @@ -27,7 +27,6 @@ import struct import dataclasses from typing import IO -from typing import Optional from .... import aiotools from .... import aiofs @@ -57,7 +56,7 @@ class ImageInfo: complete: bool @classmethod - def from_bytes(cls, data: bytes) -> Optional["ImageInfo"]: + def from_bytes(cls, data: bytes) -> ("ImageInfo" | None): try: parsed = list(struct.unpack(_IMAGE_INFO_FORMAT, data)) except struct.error: @@ -97,7 +96,7 @@ class DeviceInfo: path: str size: int free: int - image: Optional[ImageInfo] + image: (ImageInfo | None) @classmethod async def read(cls, device_path: str) -> "DeviceInfo": diff --git a/kvmd/plugins/msd/relay/gpio.py b/kvmd/plugins/msd/relay/gpio.py index dc3013a9..cec31f1a 100644 --- a/kvmd/plugins/msd/relay/gpio.py +++ b/kvmd/plugins/msd/relay/gpio.py @@ -20,8 +20,6 @@ # ========================================================================== # -from typing import Optional - import gpiod from .... import aiogp @@ -44,9 +42,9 @@ class Gpio: # pylint: disable=too-many-instance-attributes self.__reset_inverted = reset_inverted self.__reset_delay = reset_delay - self.__chip: Optional[gpiod.Chip] = None - self.__target_line: Optional[gpiod.Line] = None - self.__reset_line: Optional[gpiod.Line] = None + self.__chip: (gpiod.Chip | None) = None + self.__target_line: (gpiod.Line | None) = None + self.__reset_line: (gpiod.Line | None) = None def open(self) -> None: assert self.__chip is None diff --git a/kvmd/plugins/ugpio/__init__.py b/kvmd/plugins/ugpio/__init__.py index 54ee2aea..f5bee635 100644 --- a/kvmd/plugins/ugpio/__init__.py +++ b/kvmd/plugins/ugpio/__init__.py @@ -20,10 +20,7 @@ # ========================================================================== # -from typing import Set -from typing import Type from typing import Callable -from typing import Optional from typing import Any from ...errors import OperationError @@ -71,7 +68,7 @@ class BaseUserGpioDriver(BasePlugin): return self._instance_name @classmethod - def get_modes(cls) -> Set[str]: + def get_modes(cls) -> set[str]: return set(UserGpioModes.ALL) @classmethod @@ -84,7 +81,7 @@ class BaseUserGpioDriver(BasePlugin): _ = pin _ = debounce - def register_output(self, pin: str, initial: Optional[bool]) -> None: + def register_output(self, pin: str, initial: (bool | None)) -> None: _ = pin _ = initial @@ -105,5 +102,5 @@ class BaseUserGpioDriver(BasePlugin): # ===== -def get_ugpio_driver_class(name: str) -> Type[BaseUserGpioDriver]: +def get_ugpio_driver_class(name: str) -> type[BaseUserGpioDriver]: return get_plugin_class("ugpio", name) # type: ignore diff --git a/kvmd/plugins/ugpio/cmd.py b/kvmd/plugins/ugpio/cmd.py index 8e1922a4..afd9c7b2 100644 --- a/kvmd/plugins/ugpio/cmd.py +++ b/kvmd/plugins/ugpio/cmd.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Dict -from typing import Set from typing import Callable from typing import Any @@ -48,7 +45,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute instance_name: str, notifier: aiotools.AioNotifier, - cmd: List[str], + cmd: list[str], ) -> None: super().__init__(instance_name, notifier) @@ -56,13 +53,13 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__cmd = cmd @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "cmd": Option([], type=valid_command), } @classmethod - def get_modes(cls) -> Set[str]: + def get_modes(cls) -> set[str]: return set([UserGpioModes.OUTPUT]) @classmethod diff --git a/kvmd/plugins/ugpio/cmdret.py b/kvmd/plugins/ugpio/cmdret.py index db55ee05..05b514d6 100644 --- a/kvmd/plugins/ugpio/cmdret.py +++ b/kvmd/plugins/ugpio/cmdret.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Dict -from typing import Set from typing import Callable from typing import Any @@ -48,7 +45,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute instance_name: str, notifier: aiotools.AioNotifier, - cmd: List[str], + cmd: list[str], ) -> None: super().__init__(instance_name, notifier) @@ -56,13 +53,13 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__cmd = cmd @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "cmd": Option([], type=valid_command), } @classmethod - def get_modes(cls) -> Set[str]: + def get_modes(cls) -> set[str]: return set([UserGpioModes.INPUT]) @classmethod diff --git a/kvmd/plugins/ugpio/ezcoo.py b/kvmd/plugins/ugpio/ezcoo.py index ac7ee0f3..4af8e087 100644 --- a/kvmd/plugins/ugpio/ezcoo.py +++ b/kvmd/plugins/ugpio/ezcoo.py @@ -26,10 +26,7 @@ import functools import errno import time -from typing import Tuple -from typing import Dict from typing import Callable -from typing import Optional from typing import Any import serial @@ -72,14 +69,14 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__protocol = protocol self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue() - self.__channel_queue: "multiprocessing.Queue[Optional[int]]" = multiprocessing.Queue() - self.__channel: Optional[int] = -1 + self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue() + self.__channel: (int | None) = -1 - self.__proc: Optional[multiprocessing.Process] = None + self.__proc: (multiprocessing.Process | None) = None self.__stop_event = multiprocessing.Event() @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "device": Option("", type=valid_abs_path, unpack_as="device_path"), "speed": Option(115200, type=valid_tty_speed), @@ -164,8 +161,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute def __get_serial(self) -> serial.Serial: return serial.Serial(self.__device_path, self.__speed, timeout=self.__read_timeout) - def __recv_channel(self, tty: serial.Serial, data: bytes) -> Tuple[Optional[int], bytes]: - channel: Optional[int] = None + def __recv_channel(self, tty: serial.Serial, data: bytes) -> tuple[(int | None), bytes]: + channel: (int | None) = None if tty.in_waiting: data += tty.read_all() found = re.findall(b"V[0-9a-fA-F]{2}S", data) diff --git a/kvmd/plugins/ugpio/gpio.py b/kvmd/plugins/ugpio/gpio.py index 692ee884..4e4533ed 100644 --- a/kvmd/plugins/ugpio/gpio.py +++ b/kvmd/plugins/ugpio/gpio.py @@ -20,9 +20,7 @@ # ========================================================================== # -from typing import Dict from typing import Callable -from typing import Optional from typing import Any import gpiod @@ -52,16 +50,16 @@ class Plugin(BaseUserGpioDriver): self.__device_path = device_path - self.__input_pins: Dict[int, aiogp.AioReaderPinParams] = {} - self.__output_pins: Dict[int, Optional[bool]] = {} + self.__input_pins: dict[int, aiogp.AioReaderPinParams] = {} + self.__output_pins: dict[int, (bool | None)] = {} - self.__reader: Optional[aiogp.AioReader] = None + self.__reader: (aiogp.AioReader | None) = None - self.__chip: Optional[gpiod.Chip] = None - self.__output_lines: Dict[int, gpiod.Line] = {} + self.__chip: (gpiod.Chip | None) = None + self.__output_lines: dict[int, gpiod.Line] = {} @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="device_path"), } @@ -73,7 +71,7 @@ class Plugin(BaseUserGpioDriver): def register_input(self, pin: str, debounce: float) -> None: self.__input_pins[int(pin)] = aiogp.AioReaderPinParams(False, debounce) - def register_output(self, pin: str, initial: Optional[bool]) -> None: + def register_output(self, pin: str, initial: (bool | None)) -> None: self.__output_pins[int(pin)] = initial def prepare(self) -> None: diff --git a/kvmd/plugins/ugpio/hidrelay.py b/kvmd/plugins/ugpio/hidrelay.py index d1786908..0606c85f 100644 --- a/kvmd/plugins/ugpio/hidrelay.py +++ b/kvmd/plugins/ugpio/hidrelay.py @@ -24,10 +24,7 @@ import asyncio import contextlib import functools -from typing import Dict -from typing import Set from typing import Callable -from typing import Optional from typing import Any import hid @@ -67,27 +64,27 @@ class Plugin(BaseUserGpioDriver): self.__device_path = device_path self.__state_poll = state_poll - self.__device: Optional[hid.device] = None + self.__device: (hid.device | None) = None self.__stop = False - self.__initials: Dict[int, Optional[bool]] = {} + self.__initials: dict[int, (bool | None)] = {} @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "device": Option("", type=valid_abs_path, unpack_as="device_path"), "state_poll": Option(5.0, type=valid_float_f01), } @classmethod - def get_modes(cls) -> Set[str]: + def get_modes(cls) -> set[str]: return set([UserGpioModes.OUTPUT]) @classmethod def get_pin_validator(cls) -> Callable[[Any], Any]: return functools.partial(valid_number, min=0, max=7, name="HID relay channel") - def register_output(self, pin: str, initial: Optional[bool]) -> None: + def register_output(self, pin: str, initial: (bool | None)) -> None: self.__initials[int(pin)] = initial def prepare(self) -> None: diff --git a/kvmd/plugins/ugpio/hue.py b/kvmd/plugins/ugpio/hue.py index b34cb719..7270c4d8 100644 --- a/kvmd/plugins/ugpio/hue.py +++ b/kvmd/plugins/ugpio/hue.py @@ -22,9 +22,7 @@ import asyncio -from typing import Dict from typing import Callable -from typing import Optional from typing import Any import aiohttp @@ -70,15 +68,15 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__state_poll = state_poll self.__timeout = timeout - self.__initial: Dict[str, Optional[bool]] = {} + self.__initial: dict[str, (bool | None)] = {} - self.__state: Dict[str, Optional[bool]] = {} + self.__state: dict[str, (bool | None)] = {} self.__update_notifier = aiotools.AioNotifier() - self.__http_session: Optional[aiohttp.ClientSession] = None + self.__http_session: (aiohttp.ClientSession | None) = None @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "url": Option("", type=valid_stripped_string_not_empty), "verify": Option(True, type=valid_bool), @@ -95,7 +93,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute _ = debounce self.__state[pin] = None - def register_output(self, pin: str, initial: Optional[bool]) -> None: + def register_output(self, pin: str, initial: (bool | None)) -> None: self.__initial[pin] = initial self.__state[pin] = None @@ -109,7 +107,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute aiotools.run_sync(inner_prepare()) async def run(self) -> None: - prev_state: Optional[Dict] = None + prev_state: (dict | None) = None while True: session = self.__ensure_http_session() try: @@ -152,7 +150,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute def __ensure_http_session(self) -> aiohttp.ClientSession: if not self.__http_session: - kwargs: Dict = { + kwargs: dict = { "headers": { "User-Agent": htclient.make_user_agent("KVMD"), }, diff --git a/kvmd/plugins/ugpio/ipmi.py b/kvmd/plugins/ugpio/ipmi.py index e822065d..5376b9da 100644 --- a/kvmd/plugins/ugpio/ipmi.py +++ b/kvmd/plugins/ugpio/ipmi.py @@ -23,10 +23,7 @@ import asyncio import functools -from typing import List -from typing import Dict from typing import Callable -from typing import Optional from typing import Any from ...logging import get_logger @@ -71,7 +68,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute passwd: str, passwd_env: str, - cmd: List[str], + cmd: list[str], state_poll: float, ) -> None: @@ -92,7 +89,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__power = False @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "host": Option("", type=valid_ip_or_host), "port": Option(623, type=valid_port), @@ -121,7 +118,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute if pin not in ["0", "status"]: raise RuntimeError(f"Unsupported mode 'input' for pin={pin} on {self}") - def register_output(self, pin: str, initial: Optional[bool]) -> None: + def register_output(self, pin: str, initial: (bool | None)) -> None: _ = initial if pin not in [*_OUTPUTS, *_OUTPUTS.values()]: raise RuntimeError(f"Unsupported mode 'output' for pin={pin} on {self}") @@ -181,7 +178,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__online = False @functools.lru_cache() - def __make_ipmitool_kwargs(self, action: str) -> Dict: + def __make_ipmitool_kwargs(self, action: str) -> dict: return { "cmd": [ part.format( diff --git a/kvmd/plugins/ugpio/locator.py b/kvmd/plugins/ugpio/locator.py index a0ec5aef..8eeddd6b 100644 --- a/kvmd/plugins/ugpio/locator.py +++ b/kvmd/plugins/ugpio/locator.py @@ -22,10 +22,7 @@ import asyncio -from typing import Dict -from typing import Set from typing import Callable -from typing import Optional from typing import Any import gpiod @@ -55,26 +52,26 @@ class Plugin(BaseUserGpioDriver): self.__device_path = device_path - self.__tasks: Dict[int, Optional[asyncio.Task]] = {} + self.__tasks: dict[int, (asyncio.Task | None)] = {} - self.__chip: Optional[gpiod.Chip] = None - self.__lines: Dict[int, gpiod.Line] = {} + self.__chip: (gpiod.Chip | None) = None + self.__lines: dict[int, gpiod.Line] = {} @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "device": Option("/dev/gpiochip0", type=valid_abs_path, unpack_as="device_path"), } @classmethod - def get_modes(cls) -> Set[str]: + def get_modes(cls) -> set[str]: return set([UserGpioModes.OUTPUT]) @classmethod def get_pin_validator(cls) -> Callable[[Any], Any]: return valid_gpio_pin - def register_output(self, pin: str, initial: Optional[bool]) -> None: + def register_output(self, pin: str, initial: (bool | None)) -> None: _ = initial self.__tasks[int(pin)] = None diff --git a/kvmd/plugins/ugpio/pway.py b/kvmd/plugins/ugpio/pway.py index 969838f3..1337806c 100644 --- a/kvmd/plugins/ugpio/pway.py +++ b/kvmd/plugins/ugpio/pway.py @@ -28,10 +28,7 @@ import functools import errno import time -from typing import Tuple -from typing import Dict from typing import Callable -from typing import Optional from typing import Any import serial @@ -74,14 +71,14 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__protocol = protocol self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue() - self.__channel_queue: "multiprocessing.Queue[Optional[int]]" = multiprocessing.Queue() - self.__channel: Optional[int] = -1 + self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue() + self.__channel: (int | None) = -1 - self.__proc: Optional[multiprocessing.Process] = None + self.__proc: (multiprocessing.Process | None) = None self.__stop_event = multiprocessing.Event() @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "device": Option("", type=valid_abs_path, unpack_as="device_path"), "speed": Option(19200, type=valid_tty_speed), @@ -167,8 +164,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute def __get_serial(self) -> serial.Serial: return serial.Serial(self.__device_path, self.__speed, timeout=self.__read_timeout) - def __recv_channel(self, tty: serial.Serial, data: bytes) -> Tuple[Optional[int], bytes]: - channel: Optional[int] = None + def __recv_channel(self, tty: serial.Serial, data: bytes) -> tuple[(int | None), bytes]: + channel: (int | None) = None if tty.in_waiting: data += tty.read_all() # When you switch ports you see something like "VGA_SWITCH_CONTROL=[0-15]" for ports 1-16 diff --git a/kvmd/plugins/ugpio/pwm.py b/kvmd/plugins/ugpio/pwm.py index 47883b57..2b3d5125 100644 --- a/kvmd/plugins/ugpio/pwm.py +++ b/kvmd/plugins/ugpio/pwm.py @@ -21,10 +21,7 @@ # ========================================================================== # -from typing import Dict -from typing import Set from typing import Callable -from typing import Optional from typing import Any from periphery import PWM @@ -64,11 +61,11 @@ class Plugin(BaseUserGpioDriver): self.__duty_cycle_push = duty_cycle_push self.__duty_cycle_release = duty_cycle_release - self.__channels: Dict[int, Optional[bool]] = {} - self.__pwms: Dict[int, PWM] = {} + self.__channels: dict[int, (bool | None)] = {} + self.__pwms: dict[int, PWM] = {} @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "chip": Option(0, type=valid_int_f0), "period": Option(20000000, type=valid_int_f0), @@ -77,14 +74,14 @@ class Plugin(BaseUserGpioDriver): } @classmethod - def get_modes(cls) -> Set[str]: + def get_modes(cls) -> set[str]: return set([UserGpioModes.OUTPUT]) @classmethod def get_pin_validator(cls) -> Callable[[Any], Any]: return valid_gpio_pin - def register_output(self, pin: str, initial: Optional[bool]) -> None: + def register_output(self, pin: str, initial: (bool | None)) -> None: self.__channels[int(pin)] = initial def prepare(self) -> None: diff --git a/kvmd/plugins/ugpio/servo.py b/kvmd/plugins/ugpio/servo.py index daf2104a..1595b2bc 100644 --- a/kvmd/plugins/ugpio/servo.py +++ b/kvmd/plugins/ugpio/servo.py @@ -21,8 +21,6 @@ # ========================================================================== # -from typing import Dict - from ... import aiotools from ...yamlconf import Option @@ -69,7 +67,7 @@ class Plugin(PwmPlugin): ) @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: valid_angle = (lambda arg: valid_number(arg, min=-360.0, max=360.0, type=float)) return { "chip": Option(0, type=valid_int_f0), diff --git a/kvmd/plugins/ugpio/tesmart.py b/kvmd/plugins/ugpio/tesmart.py index e3f5a3bd..c7e410f6 100644 --- a/kvmd/plugins/ugpio/tesmart.py +++ b/kvmd/plugins/ugpio/tesmart.py @@ -23,9 +23,7 @@ import asyncio import functools -from typing import Dict from typing import Callable -from typing import Optional from typing import Any import serial_asyncio @@ -79,13 +77,13 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__switch_delay = switch_delay self.__state_poll = state_poll - self.__reader: Optional[asyncio.StreamReader] = None - self.__writer: Optional[asyncio.StreamWriter] = None + self.__reader: (asyncio.StreamReader | None) = None + self.__writer: (asyncio.StreamWriter | None) = None self.__active: int = -1 self.__update_notifier = aiotools.AioNotifier() @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "host": Option("", type=valid_ip_or_host, if_empty=""), "port": Option(5000, type=valid_port), diff --git a/kvmd/plugins/ugpio/wol.py b/kvmd/plugins/ugpio/wol.py index 42383d84..67875ad9 100644 --- a/kvmd/plugins/ugpio/wol.py +++ b/kvmd/plugins/ugpio/wol.py @@ -23,9 +23,7 @@ import socket import functools -from typing import Dict from typing import Callable -from typing import Optional from typing import Any from ...logging import get_logger @@ -61,7 +59,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__mac = mac @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "ip": Option("255.255.255.255", type=functools.partial(valid_ip, v6=False)), "port": Option(9, type=valid_port), @@ -81,7 +79,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute if not state: return - sock: Optional[socket.socket] = None + sock: (socket.socket | None) = None try: # TODO: IPv6 support: http://lists.cluenet.de/pipermail/ipv6-ops/2014-September/010139.html sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) diff --git a/kvmd/plugins/ugpio/xh_hk4401.py b/kvmd/plugins/ugpio/xh_hk4401.py index d2ed34a5..b25aea03 100644 --- a/kvmd/plugins/ugpio/xh_hk4401.py +++ b/kvmd/plugins/ugpio/xh_hk4401.py @@ -27,10 +27,7 @@ import functools import errno import time -from typing import Tuple -from typing import Dict from typing import Callable -from typing import Optional from typing import Any import serial @@ -71,14 +68,14 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute self.__read_timeout = read_timeout self.__ctl_queue: "multiprocessing.Queue[int]" = multiprocessing.Queue() - self.__channel_queue: "multiprocessing.Queue[Optional[int]]" = multiprocessing.Queue() - self.__channel: Optional[int] = -1 + self.__channel_queue: "multiprocessing.Queue[int | None]" = multiprocessing.Queue() + self.__channel: (int | None) = -1 - self.__proc: Optional[multiprocessing.Process] = None + self.__proc: (multiprocessing.Process | None) = None self.__stop_event = multiprocessing.Event() @classmethod - def get_plugin_options(cls) -> Dict: + def get_plugin_options(cls) -> dict: return { "device": Option("", type=valid_abs_path, unpack_as="device_path"), "speed": Option(19200, type=valid_tty_speed), @@ -165,8 +162,8 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute def __get_serial(self) -> serial.Serial: return serial.Serial(self.__device_path, self.__speed, timeout=self.__read_timeout) - def __recv_channel(self, tty: serial.Serial, data: bytes) -> Tuple[Optional[int], bytes]: - channel: Optional[int] = None + def __recv_channel(self, tty: serial.Serial, data: bytes) -> tuple[(int | None), bytes]: + channel: (int | None) = None if tty.in_waiting: data += tty.read_all() found = re.findall(b"AG0[1-4]gA", data) |