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 | |
parent | 4b75221e9470b4a009955d7677f16adf8e23e302 (diff) |
new typing style
129 files changed, 593 insertions, 941 deletions
@@ -26,10 +26,6 @@ import csv import textwrap import dataclasses -from typing import Set -from typing import List -from typing import Optional - import Xlib.keysymdef.latin1 import Xlib.keysymdef.miscellany import Xlib.keysymdef.xf86 @@ -77,11 +73,11 @@ class _KeyMapping: usb_key: _UsbKey ps2_key: _Ps2Key at1_code: int - x11_keys: Set[_X11Key] + x11_keys: set[_X11Key] def _resolve_keysym(name: str) -> int: - code: Optional[int] = None + code: (int | None) = None for module in [ Xlib.keysymdef.latin1, Xlib.keysymdef.miscellany, @@ -95,8 +91,8 @@ def _resolve_keysym(name: str) -> int: return code -def _parse_x11_names(names: str) -> Set[_X11Key]: - keys: Set[_X11Key] = set() +def _parse_x11_names(names: str) -> set[_X11Key]: + keys: set[_X11Key] = set() for name in filter(None, names.split(",")): shift = name.startswith("^") name = (name[1:] if shift else name) @@ -119,8 +115,8 @@ def _parse_ps2_key(key: str) -> _Ps2Key: ) -def _read_keymap_csv(path: str) -> List[_KeyMapping]: - keymap: List[_KeyMapping] = [] +def _read_keymap_csv(path: str) -> list[_KeyMapping]: + keymap: list[_KeyMapping] = [] with open(path) as keymap_file: for row in csv.DictReader(keymap_file): if len(row) >= 6: @@ -135,7 +131,7 @@ def _read_keymap_csv(path: str) -> List[_KeyMapping]: return keymap -def _render_keymap(keymap: List[_KeyMapping], template_path: str, out_path: str) -> None: +def _render_keymap(keymap: list[_KeyMapping], template_path: str, out_path: str) -> None: with open(template_path) as template_file: with open(out_path, "w") as out_file: template = textwrap.dedent(template_file.read()) diff --git a/kvmd/aiogp.py b/kvmd/aiogp.py index a55c9f71..8fe01a9a 100644 --- a/kvmd/aiogp.py +++ b/kvmd/aiogp.py @@ -24,10 +24,6 @@ import asyncio import threading import dataclasses -from typing import Tuple -from typing import Dict -from typing import Optional - import gpiod from . import aiotools @@ -55,7 +51,7 @@ class AioReader: # pylint: disable=too-many-instance-attributes self, path: str, consumer: str, - pins: Dict[int, AioReaderPinParams], + pins: dict[int, AioReaderPinParams], notifier: aiotools.AioNotifier, ) -> None: @@ -64,12 +60,12 @@ class AioReader: # pylint: disable=too-many-instance-attributes self.__pins = pins self.__notifier = notifier - self.__values: Optional[Dict[int, _DebouncedValue]] = None + self.__values: (dict[int, _DebouncedValue] | None) = None self.__thread = threading.Thread(target=self.__run, daemon=True) self.__stop_event = threading.Event() - self.__loop: Optional[asyncio.AbstractEventLoop] = None + self.__loop: (asyncio.AbstractEventLoop | None) = None def get(self, pin: int) -> bool: value = (self.__values[pin].get() if self.__values is not None else False) @@ -123,7 +119,7 @@ class AioReader: # pylint: disable=too-many-instance-attributes for (pin, value) in zip(pins, lines.get_values()): self.__values[pin].set(bool(value)) - def __parse_event(self, event: gpiod.LineEvent) -> Tuple[int, int]: + def __parse_event(self, event: gpiod.LineEvent) -> tuple[int, int]: pin = event.source.offset() if event.type == gpiod.LineEvent.RISING_EDGE: return (pin, 1) diff --git a/kvmd/aiohelpers.py b/kvmd/aiohelpers.py index ae943d23..032afb87 100644 --- a/kvmd/aiohelpers.py +++ b/kvmd/aiohelpers.py @@ -22,8 +22,6 @@ import subprocess -from typing import List - from .logging import get_logger from . import tools @@ -31,7 +29,7 @@ from . import aioproc # ===== -async def remount(name: str, base_cmd: List[str], rw: bool) -> bool: +async def remount(name: str, base_cmd: list[str], rw: bool) -> bool: logger = get_logger(1) mode = ("rw" if rw else "ro") cmd = [ diff --git a/kvmd/aiomulti.py b/kvmd/aiomulti.py index 72ec5cda..2222b9be 100644 --- a/kvmd/aiomulti.py +++ b/kvmd/aiomulti.py @@ -23,12 +23,9 @@ import multiprocessing import queue -from typing import Tuple -from typing import Dict from typing import Type from typing import TypeVar from typing import Generic -from typing import Optional from . import aiotools @@ -40,7 +37,7 @@ _QueueItemT = TypeVar("_QueueItemT") async def queue_get_last( # pylint: disable=invalid-name q: "multiprocessing.Queue[_QueueItemT]", timeout: float, -) -> Tuple[bool, Optional[_QueueItemT]]: +) -> tuple[bool, (_QueueItemT | None)]: return (await aiotools.run_async(queue_get_last_sync, q, timeout)) @@ -48,7 +45,7 @@ async def queue_get_last( # pylint: disable=invalid-name def queue_get_last_sync( # pylint: disable=invalid-name q: "multiprocessing.Queue[_QueueItemT]", timeout: float, -) -> Tuple[bool, Optional[_QueueItemT]]: +) -> tuple[bool, (_QueueItemT | None)]: try: item = q.get(timeout=timeout) @@ -79,7 +76,7 @@ _SharedFlagT = TypeVar("_SharedFlagT", int, bool) class AioSharedFlags(Generic[_SharedFlagT]): def __init__( self, - initial: Dict[str, _SharedFlagT], + initial: dict[str, _SharedFlagT], notifier: AioProcessNotifier, type: Type[_SharedFlagT]=bool, # pylint: disable=redefined-builtin ) -> None: @@ -105,10 +102,10 @@ class AioSharedFlags(Generic[_SharedFlagT]): if changed: self.__notifier.notify() - async def get(self) -> Dict[str, _SharedFlagT]: + async def get(self) -> dict[str, _SharedFlagT]: return (await aiotools.run_async(self.__inner_get)) - def __inner_get(self) -> Dict[str, _SharedFlagT]: + def __inner_get(self) -> dict[str, _SharedFlagT]: with self.__lock: return { key: self.__type(shared.value) diff --git a/kvmd/aioproc.py b/kvmd/aioproc.py index 5df2aaf2..a06628c5 100644 --- a/kvmd/aioproc.py +++ b/kvmd/aioproc.py @@ -26,11 +26,6 @@ import asyncio import asyncio.subprocess import logging -from typing import Tuple -from typing import List -from typing import Dict -from typing import Optional - import setproctitle from .logging import get_logger @@ -38,9 +33,9 @@ from .logging import get_logger # ===== async def run_process( - cmd: List[str], + cmd: list[str], err_to_null: bool=False, - env: Optional[Dict[str, str]]=None, + env: (dict[str, str] | None)=None, ) -> asyncio.subprocess.Process: # pylint: disable=no-member return (await asyncio.create_subprocess_exec( @@ -53,10 +48,10 @@ async def run_process( async def read_process( - cmd: List[str], + cmd: list[str], err_to_null: bool=False, - env: Optional[Dict[str, str]]=None, -) -> Tuple[asyncio.subprocess.Process, str]: # pylint: disable=no-member + env: (dict[str, str] | None)=None, +) -> tuple[asyncio.subprocess.Process, str]: # pylint: disable=no-member proc = await run_process(cmd, err_to_null, env) (stdout, _) = await proc.communicate() @@ -64,9 +59,9 @@ async def read_process( async def log_process( - cmd: List[str], + cmd: list[str], logger: logging.Logger, - env: Optional[Dict[str, str]]=None, + env: (dict[str, str] | None)=None, prefix: str="", ) -> asyncio.subprocess.Process: # pylint: disable=no-member diff --git a/kvmd/aiotools.py b/kvmd/aiotools.py index 4f05501c..a5a28fde 100644 --- a/kvmd/aiotools.py +++ b/kvmd/aiotools.py @@ -29,22 +29,17 @@ import types import typing -from typing import Tuple -from typing import List -from typing import Set from typing import Callable from typing import Awaitable from typing import Coroutine -from typing import Type from typing import TypeVar -from typing import Optional from typing import Any from .logging import get_logger # ===== -def run(coro: Coroutine, final: Optional[Coroutine]=None) -> None: +def run(coro: Coroutine, final: (Coroutine | None)=None) -> None: # https://github.com/aio-libs/aiohttp/blob/a1d4dac1d/aiohttp/web.py#L515 def sigint_handler() -> None: @@ -177,7 +172,7 @@ def create_deadly_task(name: str, coro: Coroutine) -> asyncio.Task: async def stop_all_deadly_tasks() -> None: - tasks: List[asyncio.Task] = [] + tasks: list[asyncio.Task] = [] for task in asyncio.all_tasks(): if getattr(task, _ATTR_DEADLY_TASK, False): tasks.append(task) @@ -200,7 +195,7 @@ async def wait_infinite() -> None: await asyncio.sleep(3600) -async def wait_first(*aws: Awaitable) -> Tuple[Set[asyncio.Task], Set[asyncio.Task]]: +async def wait_first(*aws: Awaitable) -> tuple[set[asyncio.Task], set[asyncio.Task]]: return (await asyncio.wait(list(aws), return_when=asyncio.FIRST_COMPLETED)) @@ -235,7 +230,7 @@ class AioNotifier: def notify(self) -> None: self.__queue.put_nowait(None) - async def wait(self, timeout: Optional[float]=None) -> None: + async def wait(self, timeout: (float | None)=None) -> None: if timeout is None: await self.__queue.get() else: @@ -276,8 +271,8 @@ class AioStage: class AioExclusiveRegion: def __init__( self, - exc_type: Type[Exception], - notifier: Optional[AioNotifier]=None, + exc_type: type[Exception], + notifier: (AioNotifier | None)=None, ) -> None: self.__exc_type = exc_type @@ -285,7 +280,7 @@ class AioExclusiveRegion: self.__busy = False - def get_exc_type(self) -> Type[Exception]: + def get_exc_type(self) -> type[Exception]: return self.__exc_type def is_busy(self) -> bool: @@ -313,7 +308,7 @@ class AioExclusiveRegion: async def __aexit__( self, - _exc_type: Type[BaseException], + _exc_type: type[BaseException], _exc: BaseException, _tb: types.TracebackType, ) -> None: diff --git a/kvmd/apps/__init__.py b/kvmd/apps/__init__.py index b03b7893..ffd8daae 100644 --- a/kvmd/apps/__init__.py +++ b/kvmd/apps/__init__.py @@ -27,12 +27,6 @@ import argparse import logging import logging.config -from typing import Tuple -from typing import List -from typing import Dict -from typing import Type -from typing import Optional - import pygments import pygments.lexers.data import pygments.formatters @@ -109,14 +103,14 @@ from ..validators.hw import valid_otg_ethernet # ===== def init( - prog: Optional[str]=None, - description: Optional[str]=None, + prog: (str | None)=None, + description: (str | None)=None, add_help: bool=True, check_run: bool=False, cli_logging: bool=False, - argv: Optional[List[str]]=None, + argv: (list[str] | None)=None, **load: bool, -) -> Tuple[argparse.ArgumentParser, List[str], Section]: +) -> tuple[argparse.ArgumentParser, list[str], Section]: argv = (argv or sys.argv) assert len(argv) > 0 @@ -170,10 +164,10 @@ def init( # ===== -def _init_config(config_path: str, override_options: List[str], **load_flags: bool) -> Section: +def _init_config(config_path: str, override_options: list[str], **load_flags: bool) -> Section: config_path = os.path.expanduser(config_path) try: - raw_config: Dict = load_yaml_file(config_path) + raw_config: dict = load_yaml_file(config_path) except Exception as err: raise SystemExit(f"ConfigError: Can't read config file {config_path!r}:\n{tools.efmt(err)}") if not isinstance(raw_config, dict): @@ -194,7 +188,7 @@ def _init_config(config_path: str, override_options: List[str], **load_flags: bo raise SystemExit(f"ConfigError: {err}") -def _patch_raw(raw_config: Dict) -> None: # pylint: disable=too-many-branches +def _patch_raw(raw_config: dict) -> None: # pylint: disable=too-many-branches if isinstance(raw_config.get("otg"), dict): for (old, new) in [ ("msd", "msd"), @@ -250,9 +244,9 @@ def _patch_raw(raw_config: Dict) -> None: # pylint: disable=too-many-branches def _patch_dynamic( # pylint: disable=too-many-locals - raw_config: Dict, + raw_config: dict, config: Section, - scheme: Dict, + scheme: dict, load_auth: bool=False, load_hid: bool=False, load_atx: bool=False, @@ -279,7 +273,7 @@ def _patch_dynamic( # pylint: disable=too-many-locals if load_gpio: driver: str - drivers: Dict[str, Type[BaseUserGpioDriver]] = {} # Name to drivers + drivers: dict[str, type[BaseUserGpioDriver]] = {} # Name to drivers for (driver, params) in { # type: ignore "__gpio__": {}, **tools.rget(raw_config, "kvmd", "gpio", "drivers"), @@ -343,7 +337,7 @@ def _dump_config(config: Section) -> None: print(dump) -def _get_config_scheme() -> Dict: +def _get_config_scheme() -> dict: return { "logging": Option({}), diff --git a/kvmd/apps/cleanup/__init__.py b/kvmd/apps/cleanup/__init__.py index 3e33d09f..5e83bd9a 100644 --- a/kvmd/apps/cleanup/__init__.py +++ b/kvmd/apps/cleanup/__init__.py @@ -24,9 +24,6 @@ import os import signal import time -from typing import List -from typing import Optional - import psutil from ...logging import get_logger @@ -74,7 +71,7 @@ def _remove_sockets(config: Section) -> None: # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd-cleanup", description="Kill KVMD and clear resources", diff --git a/kvmd/apps/edidconf/__init__.py b/kvmd/apps/edidconf/__init__.py index 53edf5fd..e6dcc610 100644 --- a/kvmd/apps/edidconf/__init__.py +++ b/kvmd/apps/edidconf/__init__.py @@ -28,11 +28,9 @@ import contextlib import argparse import time -from typing import List from typing import IO from typing import Generator from typing import Callable -from typing import Optional from ...validators.basic import valid_bool from ...validators.basic import valid_int_f0 @@ -180,7 +178,7 @@ def _make_format_hex(size: int) -> Callable[[int], str]: # ===== -def main(argv: Optional[List[str]]=None) -> None: # pylint: disable=too-many-branches +def main(argv: (list[str] | None)=None) -> None: # pylint: disable=too-many-branches # (parent_parser, argv, _) = init( # add_help=False, # argv=argv, diff --git a/kvmd/apps/htpasswd/__init__.py b/kvmd/apps/htpasswd/__init__.py index c46564dd..dd1213f7 100644 --- a/kvmd/apps/htpasswd/__init__.py +++ b/kvmd/apps/htpasswd/__init__.py @@ -28,9 +28,7 @@ import contextlib import textwrap import argparse -from typing import List from typing import Generator -from typing import Optional import passlib.apache @@ -125,7 +123,7 @@ def _cmd_delete(config: Section, options: argparse.Namespace) -> None: # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: (parent_parser, argv, config) = init( add_help=False, cli_logging=True, diff --git a/kvmd/apps/ipmi/__init__.py b/kvmd/apps/ipmi/__init__.py index fc450228..13aa9730 100644 --- a/kvmd/apps/ipmi/__init__.py +++ b/kvmd/apps/ipmi/__init__.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Optional - from ...clients.kvmd import KvmdClient from ... import htclient @@ -34,7 +31,7 @@ from .server import IpmiServer # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd-ipmi", description="IPMI to KVMD proxy", diff --git a/kvmd/apps/ipmi/auth.py b/kvmd/apps/ipmi/auth.py index 0ad3d208..2e495af8 100644 --- a/kvmd/apps/ipmi/auth.py +++ b/kvmd/apps/ipmi/auth.py @@ -22,9 +22,6 @@ import dataclasses -from typing import List -from typing import Dict - # ===== class IpmiPasswdError(Exception): @@ -55,8 +52,8 @@ class IpmiAuthManager: def get_credentials(self, ipmi_user: str) -> IpmiUserCredentials: return self.__credentials[ipmi_user] - def __parse_passwd_file(self, lines: List[str]) -> Dict[str, IpmiUserCredentials]: - credentials: Dict[str, IpmiUserCredentials] = {} + def __parse_passwd_file(self, lines: list[str]) -> dict[str, IpmiUserCredentials]: + credentials: dict[str, IpmiUserCredentials] = {} for (lineno, line) in enumerate(lines): if len(line.strip()) == 0 or line.lstrip().startswith("#"): continue diff --git a/kvmd/apps/ipmi/server.py b/kvmd/apps/ipmi/server.py index 597e6dce..87265607 100644 --- a/kvmd/apps/ipmi/server.py +++ b/kvmd/apps/ipmi/server.py @@ -28,9 +28,6 @@ import multiprocessing import functools import queue -from typing import Dict -from typing import Optional - import aiohttp import serial @@ -83,8 +80,8 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute self.__sol_proxy_port = (sol_proxy_port or port) self.__sol_lock = threading.Lock() - self.__sol_console: Optional[IpmiConsole] = None - self.__sol_thread: Optional[threading.Thread] = None + self.__sol_console: (IpmiConsole | None) = None + self.__sol_thread: (threading.Thread | None) = None self.__sol_stop = False def run(self) -> None: @@ -100,7 +97,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute # ===== - def handle_raw_request(self, request: Dict, session: IpmiServerSession) -> None: + def handle_raw_request(self, request: dict, session: IpmiServerSession) -> None: handler = { (6, 1): (lambda _, session: self.send_device_id(session)), # Get device ID (6, 7): self.__get_power_state_handler, # Power state @@ -125,13 +122,13 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute # ===== - def __get_power_state_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __get_power_state_handler(self, _: dict, session: IpmiServerSession) -> None: # https://github.com/arcress0/ipmiutil/blob/e2f6e95127d22e555f959f136d9bb9543c763896/util/ireset.c#L654 result = self.__make_request(session, "atx.get_state() [power]", "atx.get_state") data = [(0 if result["leds"]["power"] else 5)] session.send_ipmi_response(data=data) - def __get_selftest_status_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __get_selftest_status_handler(self, _: dict, session: IpmiServerSession) -> None: # https://github.com/arcress0/ipmiutil/blob/e2f6e95127d22e555f959f136d9bb9543c763896/util/ihealth.c#L858 data = [0x0055] try: @@ -140,12 +137,12 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute data = [0] session.send_ipmi_response(data=data) - def __get_chassis_status_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __get_chassis_status_handler(self, _: dict, session: IpmiServerSession) -> None: result = self.__make_request(session, "atx.get_state() [chassis]", "atx.get_state") data = [int(result["leds"]["power"]), 0, 0] session.send_ipmi_response(data=data) - def __chassis_control_handler(self, request: Dict, session: IpmiServerSession) -> None: + def __chassis_control_handler(self, request: dict, session: IpmiServerSession) -> None: action = { 0: "off_hard", 1: "on", @@ -179,7 +176,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute # ===== - def __activate_sol_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __activate_sol_handler(self, _: dict, session: IpmiServerSession) -> None: with self.__sol_lock: if not self.__sol_device_path: session.send_ipmi_response(code=0x81) # SOL disabled @@ -198,7 +195,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute ]) self.__start_sol_worker(session) - def __deactivate_sol_handler(self, _: Dict, session: IpmiServerSession) -> None: + def __deactivate_sol_handler(self, _: dict, session: IpmiServerSession) -> None: with self.__sol_lock: if not self.__sol_device_path: session.send_ipmi_response(code=0x81) diff --git a/kvmd/apps/janus/__init__.py b/kvmd/apps/janus/__init__.py index fc67aeb6..d8945e57 100644 --- a/kvmd/apps/janus/__init__.py +++ b/kvmd/apps/janus/__init__.py @@ -20,16 +20,13 @@ # ========================================================================== # -from typing import List -from typing import Optional - from .. import init from .runner import JanusRunner # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd-Janus", description="Janus WebRTC Gateway Runner", diff --git a/kvmd/apps/janus/runner.py b/kvmd/apps/janus/runner.py index 799fd94c..c2d01dad 100644 --- a/kvmd/apps/janus/runner.py +++ b/kvmd/apps/janus/runner.py @@ -3,10 +3,6 @@ import asyncio.subprocess import socket import dataclasses -from typing import Tuple -from typing import List -from typing import Optional - import netifaces from ... import tools @@ -42,9 +38,9 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes check_retries: int, check_retries_delay: float, - cmd: List[str], - cmd_remove: List[str], - cmd_append: List[str], + cmd: list[str], + cmd_remove: list[str], + cmd_append: list[str], ) -> None: self.__stun = Stun(stun_host, stun_port, stun_timeout, stun_retries, stun_retries_delay) @@ -55,8 +51,8 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes self.__cmd = tools.build_cmd(cmd, cmd_remove, cmd_append) - self.__janus_task: Optional[asyncio.Task] = None - self.__janus_proc: Optional[asyncio.subprocess.Process] = None # pylint: disable=no-member + self.__janus_task: (asyncio.Task | None) = None + self.__janus_proc: (asyncio.subprocess.Process | None) = None # pylint: disable=no-member def run(self) -> None: logger = get_logger(0) @@ -68,7 +64,7 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes async def __run(self) -> None: logger = get_logger(0) - prev_netcfg: Optional[_Netcfg] = None + prev_netcfg: (_Netcfg | None) = None while True: retry = 0 netcfg = _Netcfg() @@ -117,7 +113,7 @@ class JanusRunner: # pylint: disable=too-many-instance-attributes get_logger().error("Can't get default IP: %s", tools.efmt(err)) return "" - async def __get_stun_info(self, src_ip: str) -> Tuple[Stun, Tuple[str, str]]: + async def __get_stun_info(self, src_ip: str) -> tuple[Stun, tuple[str, str]]: try: return (self.__stun, (await self.__stun.get_info(src_ip, 0))) except Exception as err: diff --git a/kvmd/apps/janus/stun.py b/kvmd/apps/janus/stun.py index dc41b961..1233d7ea 100644 --- a/kvmd/apps/janus/stun.py +++ b/kvmd/apps/janus/stun.py @@ -5,10 +5,6 @@ import struct import secrets import dataclasses -from typing import Tuple -from typing import Dict -from typing import Optional - from ... import tools from ... import aiotools @@ -25,9 +21,9 @@ class StunAddress: @dataclasses.dataclass(frozen=True) class StunResponse: ok: bool - ext: Optional[StunAddress] = dataclasses.field(default=None) - src: Optional[StunAddress] = dataclasses.field(default=None) - changed: Optional[StunAddress] = dataclasses.field(default=None) + ext: (StunAddress | None) = dataclasses.field(default=None) + src: (StunAddress | None) = dataclasses.field(default=None) + changed: (StunAddress | None) = dataclasses.field(default=None) class StunNatType: @@ -60,9 +56,9 @@ class Stun: self.__retries = retries self.__retries_delay = retries_delay - self.__sock: Optional[socket.socket] = None + self.__sock: (socket.socket | None) = None - async def get_info(self, src_ip: str, src_port: int) -> Tuple[str, str]: + async def get_info(self, src_ip: str, src_port: int) -> tuple[str, str]: (family, _, _, _, addr) = socket.getaddrinfo(src_ip, src_port, type=socket.SOCK_DGRAM)[0] try: with socket.socket(family, socket.SOCK_DGRAM) as self.__sock: @@ -74,7 +70,7 @@ class Stun: finally: self.__sock = None - async def __get_nat_type(self, src_ip: str) -> Tuple[str, StunResponse]: # pylint: disable=too-many-return-statements + async def __get_nat_type(self, src_ip: str) -> tuple[str, StunResponse]: # pylint: disable=too-many-return-statements first = await self.__make_request("First probe") if not first.ok: return (StunNatType.BLOCKED, first) @@ -128,7 +124,7 @@ class Stun: ctx, self.__retries, error) return StunResponse(ok=False) - parsed: Dict[str, StunAddress] = {} + parsed: dict[str, StunAddress] = {} offset = 0 remaining = len(response) while remaining > 0: @@ -146,7 +142,7 @@ class Stun: remaining -= (4 + attr_len) return StunResponse(ok=True, **parsed) - async def __inner_make_request(self, trans_id: bytes, request: bytes, host: str, port: int) -> Tuple[bytes, str]: + async def __inner_make_request(self, trans_id: bytes, request: bytes, host: str, port: int) -> tuple[bytes, str]: assert self.__sock is not None request = struct.pack(">HH", 0x0001, len(request)) + trans_id + request # Bind Request diff --git a/kvmd/apps/kvmd/__init__.py b/kvmd/apps/kvmd/__init__.py index f7536577..5249bdd7 100644 --- a/kvmd/apps/kvmd/__init__.py +++ b/kvmd/apps/kvmd/__init__.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Optional - from ...logging import get_logger from ...plugins.hid import get_hid_class @@ -42,7 +39,7 @@ from .server import KvmdServer # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd", description="The main PiKVM daemon", diff --git a/kvmd/apps/kvmd/api/export.py b/kvmd/apps/kvmd/api/export.py index ca367f30..384fe903 100644 --- a/kvmd/apps/kvmd/api/export.py +++ b/kvmd/apps/kvmd/api/export.py @@ -23,7 +23,6 @@ import asyncio from typing import Any -from typing import List from aiohttp.web import Request from aiohttp.web import Response @@ -56,7 +55,7 @@ class ExportApi: self.__info_manager.get_submanager("fan").get_state(), self.__user_gpio.get_state(), ]) - rows: List[str] = [] + rows: list[str] = [] self.__append_prometheus_rows(rows, atx_state["enabled"], "pikvm_atx_enabled") self.__append_prometheus_rows(rows, atx_state["leds"]["power"], "pikvm_atx_power") @@ -71,7 +70,7 @@ class ExportApi: return Response(text="\n".join(rows)) - def __append_prometheus_rows(self, rows: List[str], value: Any, path: str) -> None: + def __append_prometheus_rows(self, rows: list[str], value: Any, path: str) -> None: if isinstance(value, bool): value = int(value) if isinstance(value, (int, float)): diff --git a/kvmd/apps/kvmd/api/hid.py b/kvmd/apps/kvmd/api/hid.py index 93658e88..6d993d4a 100644 --- a/kvmd/apps/kvmd/api/hid.py +++ b/kvmd/apps/kvmd/api/hid.py @@ -24,10 +24,6 @@ import os import stat import functools -from typing import Tuple -from typing import List -from typing import Dict -from typing import Set from typing import Callable from aiohttp.web import Request @@ -64,10 +60,10 @@ class HidApi: hid: BaseHid, keymap_path: str, - ignore_keys: List[str], + ignore_keys: list[str], - mouse_x_range: Tuple[int, int], - mouse_y_range: Tuple[int, int], + mouse_x_range: tuple[int, int], + mouse_y_range: tuple[int, int], ) -> None: self.__hid = hid @@ -112,8 +108,8 @@ class HidApi: # ===== - async def get_keymaps(self) -> Dict: # Ugly hack to generate hid_keymaps_state (see server.py) - keymaps: Set[str] = set() + async def get_keymaps(self) -> dict: # Ugly hack to generate hid_keymaps_state (see server.py) + keymaps: set[str] = set() for keymap_name in os.listdir(self.__keymaps_dir_path): path = os.path.join(self.__keymaps_dir_path, keymap_name) if os.access(path, os.R_OK) and stat.S_ISREG(os.stat(path).st_mode): @@ -139,7 +135,7 @@ class HidApi: self.__hid.send_key_events(text_to_web_keys(text, symmap)) return make_json_response() - def __ensure_symmap(self, keymap_name: str) -> Dict[int, Dict[int, str]]: + def __ensure_symmap(self, keymap_name: str) -> dict[int, dict[int, str]]: keymap_name = valid_printable_filename(keymap_name, "keymap") path = os.path.join(self.__keymaps_dir_path, keymap_name) try: @@ -151,14 +147,14 @@ class HidApi: return self.__inner_ensure_symmap(path, st.st_mtime) @functools.lru_cache(maxsize=10) - def __inner_ensure_symmap(self, path: str, mtime: int) -> Dict[int, Dict[int, str]]: + def __inner_ensure_symmap(self, path: str, mtime: int) -> dict[int, dict[int, str]]: _ = mtime # For LRU return build_symmap(path) # ===== @exposed_ws("key") - async def __ws_key_handler(self, _: WsSession, event: Dict) -> None: + async def __ws_key_handler(self, _: WsSession, event: dict) -> None: try: key = valid_hid_key(event["key"]) state = valid_bool(event["state"]) @@ -168,7 +164,7 @@ class HidApi: self.__hid.send_key_events([(key, state)]) @exposed_ws("mouse_button") - async def __ws_mouse_button_handler(self, _: WsSession, event: Dict) -> None: + async def __ws_mouse_button_handler(self, _: WsSession, event: dict) -> None: try: button = valid_hid_mouse_button(event["button"]) state = valid_bool(event["state"]) @@ -177,7 +173,7 @@ class HidApi: self.__hid.send_mouse_button_event(button, state) @exposed_ws("mouse_move") - async def __ws_mouse_move_handler(self, _: WsSession, event: Dict) -> None: + async def __ws_mouse_move_handler(self, _: WsSession, event: dict) -> None: try: to_x = valid_hid_mouse_move(event["to"]["x"]) to_y = valid_hid_mouse_move(event["to"]["y"]) @@ -186,14 +182,14 @@ class HidApi: self.__send_mouse_move_event_remapped(to_x, to_y) @exposed_ws("mouse_relative") - async def __ws_mouse_relative_handler(self, _: WsSession, event: Dict) -> None: + async def __ws_mouse_relative_handler(self, _: WsSession, event: dict) -> None: self.__process_delta_ws_request(event, self.__hid.send_mouse_relative_event) @exposed_ws("mouse_wheel") - async def __ws_mouse_wheel_handler(self, _: WsSession, event: Dict) -> None: + async def __ws_mouse_wheel_handler(self, _: WsSession, event: dict) -> None: self.__process_delta_ws_request(event, self.__hid.send_mouse_wheel_event) - def __process_delta_ws_request(self, event: Dict, handler: Callable[[int, int], None]) -> None: + def __process_delta_ws_request(self, event: dict, handler: Callable[[int, int], None]) -> None: try: raw_delta = event["delta"] deltas = [ diff --git a/kvmd/apps/kvmd/api/info.py b/kvmd/apps/kvmd/api/info.py index bb7caf79..5517ced0 100644 --- a/kvmd/apps/kvmd/api/info.py +++ b/kvmd/apps/kvmd/api/info.py @@ -22,8 +22,6 @@ import asyncio -from typing import List - from aiohttp.web import Request from aiohttp.web import Response @@ -51,7 +49,7 @@ class InfoApi: ]))) return make_json_response(results) - def __valid_info_fields(self, request: Request) -> List[str]: + def __valid_info_fields(self, request: Request) -> list[str]: subs = self.__info_manager.get_subs() return sorted(valid_info_fields( arg=request.query.get("fields", ",".join(subs)), diff --git a/kvmd/apps/kvmd/api/msd.py b/kvmd/apps/kvmd/api/msd.py index 18d3ba2c..16118141 100644 --- a/kvmd/apps/kvmd/api/msd.py +++ b/kvmd/apps/kvmd/api/msd.py @@ -23,10 +23,7 @@ import lzma import time -from typing import Dict from typing import AsyncGenerator -from typing import Optional -from typing import Union import aiohttp import zstandard @@ -153,7 +150,7 @@ class MsdApi: return make_json_response(self.__make_write_info(name, size, written)) @exposed_http("POST", "/msd/write_remote") - async def __write_remote_handler(self, request: Request) -> Union[Response, StreamResponse]: # pylint: disable=too-many-locals + async def __write_remote_handler(self, request: Request) -> (Response | StreamResponse): # pylint: disable=too-many-locals url = valid_url(request.query.get("url")) insecure = valid_bool(request.query.get("insecure", False)) timeout = valid_float_f01(request.query.get("timeout", 10.0)) @@ -161,7 +158,7 @@ class MsdApi: name = "" size = written = 0 - response: Optional[StreamResponse] = None + response: (StreamResponse | None) = None async def stream_write_info() -> None: assert response is not None @@ -206,11 +203,11 @@ class MsdApi: return make_json_exception(err, 400) raise - def __get_remove_incomplete(self, request: Request) -> Optional[bool]: - flag: Optional[str] = request.query.get("remove_incomplete") + def __get_remove_incomplete(self, request: Request) -> (bool | None): + flag: (str | None) = request.query.get("remove_incomplete") return (valid_bool(flag) if flag is not None else None) - def __make_write_info(self, name: str, size: int, written: int) -> Dict: + def __make_write_info(self, name: str, size: int, written: int) -> dict: return {"image": {"name": name, "size": size, "written": written}} # ===== diff --git a/kvmd/apps/kvmd/api/streamer.py b/kvmd/apps/kvmd/api/streamer.py index 401795ce..3030a925 100644 --- a/kvmd/apps/kvmd/api/streamer.py +++ b/kvmd/apps/kvmd/api/streamer.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Dict - from aiohttp.web import Request from aiohttp.web import Response @@ -102,10 +99,10 @@ class StreamerApi: # ===== - async def get_ocr(self) -> Dict: # XXX: Ugly hack + async def get_ocr(self) -> dict: # XXX: Ugly hack enabled = self.__ocr.is_available() - default: List[str] = [] - available: List[str] = [] + default: list[str] = [] + available: list[str] = [] if enabled: default = self.__ocr.get_default_langs() available = self.__ocr.get_available_langs() diff --git a/kvmd/apps/kvmd/auth.py b/kvmd/apps/kvmd/auth.py index 0f437a6b..cfc2fb6a 100644 --- a/kvmd/apps/kvmd/auth.py +++ b/kvmd/apps/kvmd/auth.py @@ -22,10 +22,6 @@ import secrets -from typing import List -from typing import Dict -from typing import Optional - from ...logging import get_logger from ... import aiotools @@ -40,12 +36,12 @@ class AuthManager: self, internal_type: str, - internal_kwargs: Dict, + internal_kwargs: dict, external_type: str, - external_kwargs: Dict, + external_kwargs: dict, - force_internal_users: List[str], + force_internal_users: list[str], enabled: bool, ) -> None: @@ -53,19 +49,19 @@ class AuthManager: if not enabled: get_logger().warning("AUTHORIZATION IS DISABLED") - self.__internal_service: Optional[BaseAuthService] = None + self.__internal_service: (BaseAuthService | None) = None if enabled: self.__internal_service = get_auth_service_class(internal_type)(**internal_kwargs) get_logger().info("Using internal auth service %r", self.__internal_service.get_plugin_name()) - self.__external_service: Optional[BaseAuthService] = None + self.__external_service: (BaseAuthService | None) = None if enabled and external_type: self.__external_service = get_auth_service_class(external_type)(**external_kwargs) get_logger().info("Using external auth service %r", self.__external_service.get_plugin_name()) self.__force_internal_users = force_internal_users - self.__tokens: Dict[str, str] = {} # {token: user} + self.__tokens: dict[str, str] = {} # {token: user} def is_auth_enabled(self) -> bool: return self.__enabled @@ -88,7 +84,7 @@ class AuthManager: get_logger().error("Got access denied for user %r from auth service %r", user, service.get_plugin_name()) return ok - async def login(self, user: str, passwd: str) -> Optional[str]: + async def login(self, user: str, passwd: str) -> (str | None): assert user == user.strip() assert user assert self.__enabled @@ -109,7 +105,7 @@ class AuthManager: if user: get_logger().info("Logged out user %r", user) - def check(self, token: str) -> Optional[str]: + def check(self, token: str) -> (str | None): assert self.__enabled return self.__tokens.get(token) diff --git a/kvmd/apps/kvmd/info/__init__.py b/kvmd/apps/kvmd/info/__init__.py index f8871d72..032dbb2d 100644 --- a/kvmd/apps/kvmd/info/__init__.py +++ b/kvmd/apps/kvmd/info/__init__.py @@ -20,8 +20,6 @@ # ========================================================================== # -from typing import Set - from ....yamlconf import Section from .base import BaseInfoSubmanager @@ -45,7 +43,7 @@ class InfoManager: "fan": FanInfoSubmanager(**config.kvmd.info.fan._unpack()), } - def get_subs(self) -> Set[str]: + def get_subs(self) -> set[str]: return set(self.__subs) def get_submanager(self, name: str) -> BaseInfoSubmanager: diff --git a/kvmd/apps/kvmd/info/auth.py b/kvmd/apps/kvmd/info/auth.py index bf1c6160..6aef3348 100644 --- a/kvmd/apps/kvmd/info/auth.py +++ b/kvmd/apps/kvmd/info/auth.py @@ -20,8 +20,6 @@ # ========================================================================== # -from typing import Dict - from .base import BaseInfoSubmanager @@ -30,5 +28,5 @@ class AuthInfoSubmanager(BaseInfoSubmanager): def __init__(self, enabled: bool) -> None: self.__enabled = enabled - async def get_state(self) -> Dict: + async def get_state(self) -> dict: return {"enabled": self.__enabled} diff --git a/kvmd/apps/kvmd/info/base.py b/kvmd/apps/kvmd/info/base.py index ae447cf5..51dbda18 100644 --- a/kvmd/apps/kvmd/info/base.py +++ b/kvmd/apps/kvmd/info/base.py @@ -20,11 +20,7 @@ # ========================================================================== # -from typing import Dict -from typing import Optional - - # ===== class BaseInfoSubmanager: - async def get_state(self) -> Optional[Dict]: + async def get_state(self) -> (dict | None): raise NotImplementedError diff --git a/kvmd/apps/kvmd/info/extras.py b/kvmd/apps/kvmd/info/extras.py index 9e70d05c..07a5eabb 100644 --- a/kvmd/apps/kvmd/info/extras.py +++ b/kvmd/apps/kvmd/info/extras.py @@ -24,9 +24,6 @@ import os import re import asyncio -from typing import Dict -from typing import Optional - from ....logging import get_logger from ....yamlconf import Section @@ -45,7 +42,7 @@ class ExtrasInfoSubmanager(BaseInfoSubmanager): def __init__(self, global_config: Section) -> None: self.__global_config = global_config - async def get_state(self) -> Optional[Dict]: + async def get_state(self) -> (dict | None): try: sui = sysunit.SystemdUnitInfo() await sui.open() @@ -53,7 +50,7 @@ class ExtrasInfoSubmanager(BaseInfoSubmanager): get_logger(0).error("Can't open systemd bus to get extras state: %s", tools.efmt(err)) sui = None try: - extras: Dict[str, Dict] = {} + extras: dict[str, dict] = {} for extra in (await asyncio.gather(*[ self.__read_extra(sui, name) for name in os.listdir(self.__get_extras_path()) @@ -71,7 +68,7 @@ class ExtrasInfoSubmanager(BaseInfoSubmanager): def __get_extras_path(self, *parts: str) -> str: return os.path.join(self.__global_config.kvmd.info.extras, *parts) - async def __read_extra(self, sui: Optional[sysunit.SystemdUnitInfo], name: str) -> Dict: + async def __read_extra(self, sui: (sysunit.SystemdUnitInfo | None), name: str) -> dict: try: extra = await aiotools.run_async(load_yaml_file, self.__get_extras_path(name, "manifest.yaml")) await self.__rewrite_app_daemon(sui, extra) @@ -81,7 +78,7 @@ class ExtrasInfoSubmanager(BaseInfoSubmanager): get_logger(0).exception("Can't read extra %r", name) return {} - async def __rewrite_app_daemon(self, sui: Optional[sysunit.SystemdUnitInfo], extra: Dict) -> None: + async def __rewrite_app_daemon(self, sui: (sysunit.SystemdUnitInfo | None), extra: dict) -> None: daemon = extra.get("daemon", "") if isinstance(daemon, str) and daemon.strip(): extra["enabled"] = extra["started"] = False @@ -91,7 +88,7 @@ class ExtrasInfoSubmanager(BaseInfoSubmanager): except Exception as err: get_logger(0).error("Can't get info about the service %r: %s", daemon, tools.efmt(err)) - def __rewrite_app_port(self, extra: Dict) -> None: + def __rewrite_app_port(self, extra: dict) -> None: port_path = extra.get("port", "") if isinstance(port_path, str) and port_path.strip(): extra["port"] = 0 diff --git a/kvmd/apps/kvmd/info/fan.py b/kvmd/apps/kvmd/info/fan.py index afa7afce..7cff7217 100644 --- a/kvmd/apps/kvmd/info/fan.py +++ b/kvmd/apps/kvmd/info/fan.py @@ -23,9 +23,7 @@ import copy import asyncio -from typing import Dict from typing import AsyncGenerator -from typing import Optional import aiohttp @@ -55,15 +53,15 @@ class FanInfoSubmanager(BaseInfoSubmanager): self.__timeout = timeout self.__state_poll = state_poll - async def get_state(self) -> Dict: + async def get_state(self) -> dict: monitored = await self.__get_monitored() return { "monitored": monitored, "state": ((await self.__get_fan_state() if monitored else None)), } - async def poll_state(self) -> AsyncGenerator[Dict, None]: - prev_state: Dict = {} + async def poll_state(self) -> AsyncGenerator[dict, None]: + prev_state: dict = {} while True: if self.__unix_path: pure = state = await self.get_state() @@ -93,7 +91,7 @@ class FanInfoSubmanager(BaseInfoSubmanager): get_logger(0).error("Can't get info about the service %r: %s", self.__daemon, tools.efmt(err)) return False - async def __get_fan_state(self) -> Optional[Dict]: + async def __get_fan_state(self) -> (dict | None): try: async with self.__make_http_session() as session: async with session.get("http://localhost/state") as response: @@ -104,7 +102,7 @@ class FanInfoSubmanager(BaseInfoSubmanager): return None def __make_http_session(self) -> aiohttp.ClientSession: - kwargs: Dict = { + kwargs: dict = { "headers": { "User-Agent": htclient.make_user_agent("KVMD"), }, diff --git a/kvmd/apps/kvmd/info/hw.py b/kvmd/apps/kvmd/info/hw.py index a99fb09d..86efd761 100644 --- a/kvmd/apps/kvmd/info/hw.py +++ b/kvmd/apps/kvmd/info/hw.py @@ -23,12 +23,9 @@ import os import asyncio -from typing import List -from typing import Dict from typing import Callable from typing import AsyncGenerator from typing import TypeVar -from typing import Optional from ....logging import get_logger @@ -48,16 +45,16 @@ _RetvalT = TypeVar("_RetvalT") class HwInfoSubmanager(BaseInfoSubmanager): def __init__( self, - vcgencmd_cmd: List[str], + vcgencmd_cmd: list[str], state_poll: float, ) -> None: self.__vcgencmd_cmd = vcgencmd_cmd self.__state_poll = state_poll - self.__dt_cache: Dict[str, str] = {} + self.__dt_cache: dict[str, str] = {} - async def get_state(self) -> Dict: + async def get_state(self) -> dict: (model, serial, cpu_temp, throttling) = await asyncio.gather( self.__read_dt_file("model"), self.__read_dt_file("serial-number"), @@ -78,8 +75,8 @@ class HwInfoSubmanager(BaseInfoSubmanager): }, } - 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: @@ -89,7 +86,7 @@ class HwInfoSubmanager(BaseInfoSubmanager): # ===== - async def __read_dt_file(self, name: str) -> Optional[str]: + async def __read_dt_file(self, name: str) -> (str | None): if name not in self.__dt_cache: path = os.path.join(f"{env.PROCFS_PREFIX}/proc/device-tree", name) try: @@ -99,7 +96,7 @@ class HwInfoSubmanager(BaseInfoSubmanager): return None return self.__dt_cache[name] - async def __get_cpu_temp(self) -> Optional[float]: + async def __get_cpu_temp(self) -> (float | None): temp_path = f"{env.SYSFS_PREFIX}/sys/class/thermal/thermal_zone0/temp" try: return int((await aiofs.read(temp_path)).strip()) / 1000 @@ -107,7 +104,7 @@ class HwInfoSubmanager(BaseInfoSubmanager): get_logger(0).error("Can't read CPU temp from %s: %s", temp_path, err) return None - async def __get_throttling(self) -> Optional[Dict]: + async def __get_throttling(self) -> (dict | None): # https://www.raspberrypi.org/forums/viewtopic.php?f=63&t=147781&start=50#p972790 flags = await self.__parse_vcgencmd( arg="get_throttled", @@ -133,7 +130,7 @@ class HwInfoSubmanager(BaseInfoSubmanager): } return None - async def __parse_vcgencmd(self, arg: str, parser: Callable[[str], _RetvalT]) -> Optional[_RetvalT]: + async def __parse_vcgencmd(self, arg: str, parser: Callable[[str], _RetvalT]) -> (_RetvalT | None): cmd = [*self.__vcgencmd_cmd, arg] try: text = (await aioproc.read_process(cmd, err_to_null=True))[1] diff --git a/kvmd/apps/kvmd/info/meta.py b/kvmd/apps/kvmd/info/meta.py index 86a9be26..6a92d7ea 100644 --- a/kvmd/apps/kvmd/info/meta.py +++ b/kvmd/apps/kvmd/info/meta.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import Dict -from typing import Optional - from ....logging import get_logger from ....yamlconf.loader import load_yaml_file @@ -37,7 +34,7 @@ class MetaInfoSubmanager(BaseInfoSubmanager): def __init__(self, meta_path: str) -> None: self.__meta_path = meta_path - async def get_state(self) -> Optional[Dict]: + async def get_state(self) -> (dict | None): try: return ((await aiotools.run_async(load_yaml_file, self.__meta_path)) or {}) except Exception: diff --git a/kvmd/apps/kvmd/info/system.py b/kvmd/apps/kvmd/info/system.py index 5c46a5c6..9037a0c2 100644 --- a/kvmd/apps/kvmd/info/system.py +++ b/kvmd/apps/kvmd/info/system.py @@ -24,9 +24,6 @@ import os import asyncio import platform -from typing import List -from typing import Dict - from ....logging import get_logger from .... import aioproc @@ -38,10 +35,10 @@ from .base import BaseInfoSubmanager # ===== class SystemInfoSubmanager(BaseInfoSubmanager): - def __init__(self, streamer_cmd: List[str]) -> None: + def __init__(self, streamer_cmd: list[str]) -> None: self.__streamer_cmd = streamer_cmd - async def get_state(self) -> Dict: + async def get_state(self) -> dict: streamer_info = await self.__get_streamer_info() uname_info = platform.uname() # Uname using the internal cache return { @@ -55,9 +52,9 @@ class SystemInfoSubmanager(BaseInfoSubmanager): # ===== - async def __get_streamer_info(self) -> Dict: + async def __get_streamer_info(self) -> dict: version = "" - features: Dict[str, bool] = {} + features: dict[str, bool] = {} try: path = self.__streamer_cmd[0] ((_, version), (_, features_text)) = await asyncio.gather( diff --git a/kvmd/apps/kvmd/logreader.py b/kvmd/apps/kvmd/logreader.py index dfe2125a..cdd46744 100644 --- a/kvmd/apps/kvmd/logreader.py +++ b/kvmd/apps/kvmd/logreader.py @@ -24,7 +24,6 @@ import re import asyncio import time -from typing import Dict from typing import AsyncGenerator import systemd.journal @@ -32,7 +31,7 @@ import systemd.journal # ===== class LogReader: - async def poll_log(self, seek: int, follow: bool) -> AsyncGenerator[Dict, None]: + async def poll_log(self, seek: int, follow: bool) -> AsyncGenerator[dict, None]: reader = systemd.journal.Reader() reader.this_boot() reader.this_machine() @@ -59,7 +58,7 @@ class LogReader: else: await asyncio.sleep(1) - def __entry_to_record(self, entry: Dict) -> Dict[str, Dict]: + def __entry_to_record(self, entry: dict) -> dict[str, dict]: return { "dt": entry["__REALTIME_TIMESTAMP"], "service": entry["_SYSTEMD_UNIT"], diff --git a/kvmd/apps/kvmd/streamer.py b/kvmd/apps/kvmd/streamer.py index fb209033..096737c8 100644 --- a/kvmd/apps/kvmd/streamer.py +++ b/kvmd/apps/kvmd/streamer.py @@ -27,11 +27,7 @@ import asyncio.subprocess import dataclasses import functools -from typing import Tuple -from typing import List -from typing import Dict from typing import AsyncGenerator -from typing import Optional from typing import Any import aiohttp @@ -53,7 +49,7 @@ class StreamerSnapshot: width: int height: int mtime: float - headers: Tuple[Tuple[str, str], ...] + headers: tuple[tuple[str, str], ...] data: bytes async def make_preview(self, max_width: int, max_height: int, quality: int) -> bytes: @@ -98,7 +94,7 @@ class _StreamerParams: quality: int, resolution: str, - available_resolutions: List[str], + available_resolutions: list[str], desired_fps: int, desired_fps_min: int, @@ -117,8 +113,8 @@ class _StreamerParams: self.__has_resolution = bool(resolution) self.__has_h264 = bool(h264_bitrate) - self.__params: Dict = {self.__DESIRED_FPS: min(max(desired_fps, desired_fps_min), desired_fps_max)} - self.__limits: Dict = {self.__DESIRED_FPS: {"min": desired_fps_min, "max": desired_fps_max}} + self.__params: dict = {self.__DESIRED_FPS: min(max(desired_fps, desired_fps_min), desired_fps_max)} + self.__limits: dict = {self.__DESIRED_FPS: {"min": desired_fps_min, "max": desired_fps_max}} if self.__has_quality: self.__params[self.__QUALITY] = quality @@ -133,23 +129,23 @@ class _StreamerParams: self.__params[self.__H264_GOP] = min(max(h264_gop, h264_gop_min), h264_gop_max) self.__limits[self.__H264_GOP] = {"min": h264_gop_min, "max": h264_gop_max} - def get_features(self) -> Dict: + def get_features(self) -> dict: return { self.__QUALITY: self.__has_quality, self.__RESOLUTION: self.__has_resolution, "h264": self.__has_h264, } - def get_limits(self) -> Dict: + def get_limits(self) -> dict: limits = dict(self.__limits) if self.__has_resolution: limits[self.__AVAILABLE_RESOLUTIONS] = list(limits[self.__AVAILABLE_RESOLUTIONS]) return limits - def get_params(self) -> Dict: + def get_params(self) -> dict: return dict(self.__params) - def set_params(self, params: Dict) -> None: + def set_params(self, params: dict) -> None: new_params = dict(self.__params) if self.__QUALITY in params and self.__has_quality: @@ -187,9 +183,9 @@ class Streamer: # pylint: disable=too-many-instance-attributes process_name_prefix: str, - cmd: List[str], - cmd_remove: List[str], - cmd_append: List[str], + cmd: list[str], + cmd_remove: list[str], + cmd_append: list[str], **params_kwargs: Any, ) -> None: @@ -207,15 +203,15 @@ class Streamer: # pylint: disable=too-many-instance-attributes self.__params = _StreamerParams(**params_kwargs) - self.__stop_task: Optional[asyncio.Task] = None + self.__stop_task: (asyncio.Task | None) = None self.__stop_wip = False - self.__streamer_task: Optional[asyncio.Task] = None - self.__streamer_proc: Optional[asyncio.subprocess.Process] = None # pylint: disable=no-member + self.__streamer_task: (asyncio.Task | None) = None + self.__streamer_proc: (asyncio.subprocess.Process | None) = None # pylint: disable=no-member - self.__http_session: Optional[aiohttp.ClientSession] = None + self.__http_session: (aiohttp.ClientSession | None) = None - self.__snapshot: Optional[StreamerSnapshot] = None + self.__snapshot: (StreamerSnapshot | None) = None self.__notifier = aiotools.AioNotifier() @@ -280,16 +276,16 @@ class Streamer: # pylint: disable=too-many-instance-attributes # ===== - def set_params(self, params: Dict) -> None: + def set_params(self, params: dict) -> None: assert not self.__streamer_task return self.__params.set_params(params) - def get_params(self) -> Dict: + def get_params(self) -> dict: return self.__params.get_params() # ===== - async def get_state(self) -> Dict: + async def get_state(self) -> dict: streamer_state = None if self.__streamer_task: session = self.__ensure_http_session() @@ -302,7 +298,7 @@ class Streamer: # pylint: disable=too-many-instance-attributes except Exception: get_logger().exception("Invalid streamer response from /state") - snapshot: Optional[Dict] = None + snapshot: (dict | None) = None if self.__snapshot: snapshot = dataclasses.asdict(self.__snapshot) del snapshot["headers"] @@ -316,7 +312,7 @@ class Streamer: # pylint: disable=too-many-instance-attributes "features": self.__params.get_features(), } - async def poll_state(self) -> AsyncGenerator[Dict, None]: + async def poll_state(self) -> AsyncGenerator[dict, None]: def signal_handler(*_: Any) -> None: get_logger(0).info("Got SIGUSR2, checking the stream state ...") self.__notifier.notify() @@ -324,8 +320,8 @@ class Streamer: # pylint: disable=too-many-instance-attributes get_logger(0).info("Installing SIGUSR2 streamer handler ...") asyncio.get_event_loop().add_signal_handler(signal.SIGUSR2, signal_handler) - waiter_task: Optional[asyncio.Task] = None - prev_state: Dict = {} + waiter_task: (asyncio.Task | None) = None + prev_state: dict = {} while True: state = await self.get_state() if state != prev_state: @@ -339,7 +335,7 @@ class Streamer: # pylint: disable=too-many-instance-attributes # ===== - async def take_snapshot(self, save: bool, load: bool, allow_offline: bool) -> Optional[StreamerSnapshot]: + async def take_snapshot(self, save: bool, load: bool, allow_offline: bool) -> (StreamerSnapshot | None): if load: return self.__snapshot else: @@ -395,7 +391,7 @@ class Streamer: # pylint: disable=too-many-instance-attributes def __ensure_http_session(self) -> aiohttp.ClientSession: if not self.__http_session: - kwargs: Dict = { + kwargs: dict = { "headers": {"User-Agent": htclient.make_user_agent("KVMD")}, "connector": aiohttp.UnixConnector(path=self.__unix_path), "timeout": aiohttp.ClientTimeout(total=self.__timeout), diff --git a/kvmd/apps/kvmd/sysunit.py b/kvmd/apps/kvmd/sysunit.py index dea5e961..1358ee61 100644 --- a/kvmd/apps/kvmd/sysunit.py +++ b/kvmd/apps/kvmd/sysunit.py @@ -22,10 +22,6 @@ import types -from typing import Tuple -from typing import Type -from typing import Optional - import dbus_next import dbus_next.aio import dbus_next.aio.proxy_object @@ -36,11 +32,11 @@ import dbus_next.errors # ===== class SystemdUnitInfo: def __init__(self) -> None: - self.__bus: Optional[dbus_next.aio.MessageBus] = None - self.__intr: Optional[dbus_next.introspection.Node] = None - self.__manager: Optional[dbus_next.aio.proxy_object.ProxyInterface] = None + self.__bus: (dbus_next.aio.MessageBus | None) = None + self.__intr: (dbus_next.introspection.Node | None) = None + self.__manager: (dbus_next.aio.proxy_object.ProxyInterface | None) = None - async def get_status(self, name: str) -> Tuple[bool, bool]: + async def get_status(self, name: str) -> tuple[bool, bool]: assert self.__bus is not None assert self.__intr is not None assert self.__manager is not None @@ -89,7 +85,7 @@ class SystemdUnitInfo: async def __aexit__( self, - _exc_type: Type[BaseException], + _exc_type: type[BaseException], _exc: BaseException, _tb: types.TracebackType, ) -> None: diff --git a/kvmd/apps/kvmd/tesseract.py b/kvmd/apps/kvmd/tesseract.py index 58cba9ae..fff4df2d 100644 --- a/kvmd/apps/kvmd/tesseract.py +++ b/kvmd/apps/kvmd/tesseract.py @@ -36,10 +36,7 @@ from ctypes import c_char_p from ctypes import c_void_p from ctypes import c_char -from typing import List -from typing import Set from typing import Generator -from typing import Optional from PIL import ImageOps from PIL import Image as PilImage @@ -60,7 +57,7 @@ class _TessBaseAPI(Structure): pass -def _load_libtesseract() -> Optional[ctypes.CDLL]: +def _load_libtesseract() -> (ctypes.CDLL | None): try: path = ctypes.util.find_library("tesseract") if not path: @@ -88,7 +85,7 @@ _libtess = _load_libtesseract() @contextlib.contextmanager -def _tess_api(data_dir_path: str, langs: List[str]) -> Generator[_TessBaseAPI, None, None]: +def _tess_api(data_dir_path: str, langs: list[str]) -> Generator[_TessBaseAPI, None, None]: if not _libtess: raise OcrError("Tesseract is not available") api = _libtess.TessBaseAPICreate() @@ -107,19 +104,19 @@ _LANG_SUFFIX = ".traineddata" # ===== class TesseractOcr: - def __init__(self, data_dir_path: str, default_langs: List[str]) -> None: + def __init__(self, data_dir_path: str, default_langs: list[str]) -> None: self.__data_dir_path = data_dir_path self.__default_langs = default_langs def is_available(self) -> bool: return bool(_libtess) - def get_default_langs(self) -> List[str]: + def get_default_langs(self) -> list[str]: return list(self.__default_langs) - def get_available_langs(self) -> List[str]: + def get_available_langs(self) -> list[str]: # Это быстрее чем, инициализация либы и TessBaseAPIGetAvailableLanguagesAsVector() - langs: Set[str] = set() + langs: set[str] = set() for lang_name in os.listdir(self.__data_dir_path): if lang_name.endswith(_LANG_SUFFIX): path = os.path.join(self.__data_dir_path, lang_name) @@ -129,12 +126,12 @@ class TesseractOcr: langs.add(lang) return sorted(langs) - async def recognize(self, data: bytes, langs: List[str], left: int, top: int, right: int, bottom: int) -> str: + async def recognize(self, data: bytes, langs: list[str], left: int, top: int, right: int, bottom: int) -> str: if not langs: langs = self.__default_langs return (await aiotools.run_async(self.__inner_recognize, data, langs, left, top, right, bottom)) - def __inner_recognize(self, data: bytes, langs: List[str], left: int, top: int, right: int, bottom: int) -> str: + def __inner_recognize(self, data: bytes, langs: list[str], left: int, top: int, right: int, bottom: int) -> str: with _tess_api(self.__data_dir_path, langs) as api: assert _libtess with io.BytesIO(data) as bio: diff --git a/kvmd/apps/kvmd/ugpio.py b/kvmd/apps/kvmd/ugpio.py index b29b7ce4..8b7b31f8 100644 --- a/kvmd/apps/kvmd/ugpio.py +++ b/kvmd/apps/kvmd/ugpio.py @@ -22,11 +22,8 @@ import asyncio -from typing import List -from typing import Dict from typing import AsyncGenerator from typing import Callable -from typing import Optional from typing import Any from ...logging import get_logger @@ -83,7 +80,7 @@ class _GpioInput: self.__driver = driver self.__driver.register_input(self.__pin, config.debounce) - def get_scheme(self) -> Dict: + def get_scheme(self) -> dict: return { "hw": { "driver": self.__driver.get_instance_id(), @@ -91,7 +88,7 @@ class _GpioInput: }, } - async def get_state(self) -> Dict: + async def get_state(self) -> dict: (online, state) = (True, False) try: state = (await self.__driver.read(self.__pin) ^ self.__inverted) @@ -139,7 +136,7 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes self.__region = aiotools.AioExclusiveRegion(GpioChannelIsBusyError, notifier) - def get_scheme(self) -> Dict: + def get_scheme(self) -> dict: return { "switch": self.__switch, "pulse": { @@ -153,7 +150,7 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes }, } - async def get_state(self) -> Dict: + async def get_state(self) -> dict: busy = self.__region.is_busy() (online, state) = (True, False) if not busy: @@ -246,8 +243,8 @@ class UserGpio: for (driver, drv_config) in tools.sorted_kvs(config.drivers) } - self.__inputs: Dict[str, _GpioInput] = {} - self.__outputs: Dict[str, _GpioOutput] = {} + self.__inputs: dict[str, _GpioInput] = {} + self.__outputs: dict[str, _GpioOutput] = {} for (channel, ch_config) in tools.sorted_kvs(config.scheme): driver = self.__drivers[ch_config.driver] @@ -256,7 +253,7 @@ class UserGpio: else: # output: self.__outputs[channel] = _GpioOutput(channel, ch_config, driver, self.__notifier) - async def get_model(self) -> Dict: + async def get_model(self) -> dict: return { "scheme": { "inputs": {channel: gin.get_scheme() for (channel, gin) in self.__inputs.items()}, @@ -265,14 +262,14 @@ class UserGpio: "view": self.__make_view(), } - async def get_state(self) -> Dict: + async def get_state(self) -> dict: return { "inputs": {channel: await gin.get_state() for (channel, gin) in self.__inputs.items()}, "outputs": {channel: await gout.get_state() for (channel, gout) in self.__outputs.items()}, } - 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: @@ -313,14 +310,14 @@ class UserGpio: # ===== - def __make_view(self) -> Dict: - table: List[Optional[List[Dict]]] = [] + def __make_view(self) -> dict: + table: list[list[dict] | None] = [] for row in self.__view["table"]: if len(row) == 0: table.append(None) continue - items: List[Dict] = [] + items: list[dict] = [] for item in map(str.strip, row): if item.startswith("#") or len(item) == 0: items.append(self.__make_view_label(item)) @@ -338,14 +335,14 @@ class UserGpio: "table": table, } - def __make_view_label(self, item: str) -> Dict: + def __make_view_label(self, item: str) -> dict: assert item.startswith("#") return { "type": "label", "text": item[1:].strip(), } - def __make_view_input(self, parts: List[str]) -> Dict: + def __make_view_input(self, parts: list[str]) -> dict: assert len(parts) >= 1 color = (parts[1] if len(parts) > 1 else None) if color not in ["green", "yellow", "red"]: @@ -356,7 +353,7 @@ class UserGpio: "color": color, } - def __make_view_output(self, parts: List[str]) -> Dict: + def __make_view_output(self, parts: list[str]) -> dict: assert len(parts) >= 1 confirm = False text = "Click" diff --git a/kvmd/apps/otg/__init__.py b/kvmd/apps/otg/__init__.py index 9b6f5e69..4c811d2b 100644 --- a/kvmd/apps/otg/__init__.py +++ b/kvmd/apps/otg/__init__.py @@ -29,10 +29,6 @@ import argparse from os.path import join # pylint: disable=ungrouped-imports -from typing import List -from typing import Optional -from typing import Union - from ...logging import get_logger from ...yamlconf import Section @@ -78,7 +74,7 @@ def _unlink(path: str, optional: bool=False) -> None: os.unlink(path) -def _write(path: str, value: Union[str, int], optional: bool=False) -> None: +def _write(path: str, value: (str | int), optional: bool=False) -> None: logger = get_logger() if optional and not os.access(path, os.F_OK): logger.info("WRITE --- [SKIPPED] %s", path) @@ -320,7 +316,7 @@ def _cmd_stop(config: Section) -> None: # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: (parent_parser, argv, config) = init( add_help=False, argv=argv, diff --git a/kvmd/apps/otg/hid/keyboard.py b/kvmd/apps/otg/hid/keyboard.py index d10a4b73..6a4d0184 100644 --- a/kvmd/apps/otg/hid/keyboard.py +++ b/kvmd/apps/otg/hid/keyboard.py @@ -20,13 +20,11 @@ # ========================================================================== # -from typing import Optional - from . import Hid # ===== -def make_keyboard_hid(report_id: Optional[int]=None) -> Hid: +def make_keyboard_hid(report_id: (int | None)=None) -> Hid: return Hid( protocol=1, # Keyboard protocol subclass=1, # Boot interface subclass diff --git a/kvmd/apps/otg/hid/mouse.py b/kvmd/apps/otg/hid/mouse.py index 71249569..e0f7350b 100644 --- a/kvmd/apps/otg/hid/mouse.py +++ b/kvmd/apps/otg/hid/mouse.py @@ -20,13 +20,11 @@ # ========================================================================== # -from typing import Optional - from . import Hid # ===== -def make_mouse_hid(absolute: bool, horizontal_wheel: bool, report_id: Optional[int]=None) -> Hid: +def make_mouse_hid(absolute: bool, horizontal_wheel: bool, report_id: (int | None)=None) -> Hid: maker = (_make_absolute_hid if absolute else _make_relative_hid) return maker(horizontal_wheel, report_id) @@ -42,7 +40,7 @@ _HORIZONTAL_WHEEL = [ ] -def _make_absolute_hid(horizontal_wheel: bool, report_id: Optional[int]) -> Hid: +def _make_absolute_hid(horizontal_wheel: bool, report_id: (int | None)) -> Hid: return Hid( protocol=0, # None protocol subclass=0, # No subclass @@ -106,7 +104,7 @@ def _make_absolute_hid(horizontal_wheel: bool, report_id: Optional[int]) -> Hid: ) -def _make_relative_hid(horizontal_wheel: bool, report_id: Optional[int]) -> Hid: +def _make_relative_hid(horizontal_wheel: bool, report_id: (int | None)) -> Hid: return Hid( protocol=2, # Mouse protocol subclass=1, # Boot interface subclass diff --git a/kvmd/apps/otgconf/__init__.py b/kvmd/apps/otgconf/__init__.py index 389e1721..1a1e0f99 100644 --- a/kvmd/apps/otgconf/__init__.py +++ b/kvmd/apps/otgconf/__init__.py @@ -26,10 +26,7 @@ import contextlib import argparse import time -from typing import List -from typing import Dict from typing import Generator -from typing import Optional import yaml @@ -64,7 +61,7 @@ class _GadgetControl: with open(udc_path, "w") as udc_file: udc_file.write(udc) - def __read_metas(self) -> Generator[Dict, None, None]: + def __read_metas(self) -> Generator[dict, None, None]: for meta_name in sorted(os.listdir(self.__meta_path)): with open(os.path.join(self.__meta_path, meta_name)) as meta_file: yield json.loads(meta_file.read()) @@ -111,7 +108,7 @@ class _GadgetControl: # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: (parent_parser, argv, config) = init( add_help=False, cli_logging=True, diff --git a/kvmd/apps/otgmsd/__init__.py b/kvmd/apps/otgmsd/__init__.py index 0d32331b..2540eb0b 100644 --- a/kvmd/apps/otgmsd/__init__.py +++ b/kvmd/apps/otgmsd/__init__.py @@ -24,9 +24,6 @@ import os import errno import argparse -from typing import List -from typing import Optional - from ...validators.basic import valid_bool from ...validators.basic import valid_int_f0 from ...validators.os import valid_abs_file @@ -57,7 +54,7 @@ def _set_param(gadget: str, instance: int, param: str, value: str) -> None: # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: (parent_parser, argv, config) = init( add_help=False, cli_logging=True, diff --git a/kvmd/apps/otgnet/__init__.py b/kvmd/apps/otgnet/__init__.py index 7a5e4cc7..98213346 100644 --- a/kvmd/apps/otgnet/__init__.py +++ b/kvmd/apps/otgnet/__init__.py @@ -26,9 +26,6 @@ import dataclasses import itertools import argparse -from typing import List -from typing import Optional - from ...logging import get_logger from ...yamlconf import Section @@ -66,25 +63,25 @@ class _Netcfg: # pylint: disable=too-many-instance-attributes class _Service: # pylint: disable=too-many-instance-attributes def __init__(self, config: Section) -> None: self.__iface_net: str = config.otgnet.iface.net - self.__ip_cmd: List[str] = config.otgnet.iface.ip_cmd + self.__ip_cmd: list[str] = config.otgnet.iface.ip_cmd self.__allow_icmp: bool = config.otgnet.firewall.allow_icmp - self.__allow_tcp: List[int] = sorted(set(config.otgnet.firewall.allow_tcp)) - self.__allow_udp: List[int] = sorted(set(config.otgnet.firewall.allow_udp)) + self.__allow_tcp: list[int] = sorted(set(config.otgnet.firewall.allow_tcp)) + self.__allow_udp: list[int] = sorted(set(config.otgnet.firewall.allow_udp)) self.__forward_iface: str = config.otgnet.firewall.forward_iface - self.__iptables_cmd: List[str] = config.otgnet.firewall.iptables_cmd + self.__iptables_cmd: list[str] = config.otgnet.firewall.iptables_cmd - def build_cmd(key: str) -> List[str]: + def build_cmd(key: str) -> list[str]: return tools.build_cmd( getattr(config.otgnet.commands, key), getattr(config.otgnet.commands, f"{key}_remove"), getattr(config.otgnet.commands, f"{key}_append"), ) - self.__pre_start_cmd: List[str] = build_cmd("pre_start_cmd") - self.__post_start_cmd: List[str] = build_cmd("post_start_cmd") - self.__pre_stop_cmd: List[str] = build_cmd("pre_stop_cmd") - self.__post_stop_cmd: List[str] = build_cmd("post_stop_cmd") + self.__pre_start_cmd: list[str] = build_cmd("pre_start_cmd") + self.__post_start_cmd: list[str] = build_cmd("post_start_cmd") + self.__pre_stop_cmd: list[str] = build_cmd("pre_stop_cmd") + self.__post_stop_cmd: list[str] = build_cmd("post_stop_cmd") self.__gadget: str = config.otg.gadget self.__driver: str = config.otg.devices.ethernet.driver @@ -101,7 +98,7 @@ class _Service: # pylint: disable=too-many-instance-attributes key: str(value) for (key, value) in dataclasses.asdict(netcfg).items() } - ctls: List[BaseCtl] = [ + ctls: list[BaseCtl] = [ CustomCtl(self.__pre_start_cmd, self.__post_stop_cmd, placeholders), IfaceUpCtl(self.__ip_cmd, netcfg.iface), *([IptablesAllowIcmpCtl(self.__iptables_cmd, netcfg.iface)] if self.__allow_icmp else []), @@ -185,7 +182,7 @@ class _Service: # pylint: disable=too-many-instance-attributes # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: (parent_parser, argv, config) = init( add_help=False, argv=argv, diff --git a/kvmd/apps/otgnet/netctl.py b/kvmd/apps/otgnet/netctl.py index f186ea34..15b5ea8c 100644 --- a/kvmd/apps/otgnet/netctl.py +++ b/kvmd/apps/otgnet/netctl.py @@ -20,50 +20,46 @@ # ========================================================================== # -from typing import List -from typing import Dict - - # ===== class BaseCtl: - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: raise NotImplementedError class IfaceUpCtl(BaseCtl): - def __init__(self, base_cmd: List[str], iface: str) -> None: + def __init__(self, base_cmd: list[str], iface: str) -> None: self.__base_cmd = base_cmd self.__iface = iface - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [*self.__base_cmd, "link", "set", self.__iface, ("up" if direct else "down")] class IfaceAddIpCtl(BaseCtl): - def __init__(self, base_cmd: List[str], iface: str, cidr: str) -> None: + def __init__(self, base_cmd: list[str], iface: str, cidr: str) -> None: self.__base_cmd = base_cmd self.__iface = iface self.__cidr = cidr - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [*self.__base_cmd, "address", ("add" if direct else "del"), self.__cidr, "dev", self.__iface] class IptablesDropAllCtl(BaseCtl): - def __init__(self, base_cmd: List[str], iface: str) -> None: + def __init__(self, base_cmd: list[str], iface: str) -> None: self.__base_cmd = base_cmd self.__iface = iface - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [*self.__base_cmd, ("-A" if direct else "-D"), "INPUT", "-i", self.__iface, "-j", "DROP"] class IptablesAllowIcmpCtl(BaseCtl): - def __init__(self, base_cmd: List[str], iface: str) -> None: + def __init__(self, base_cmd: list[str], iface: str) -> None: self.__base_cmd = base_cmd self.__iface = iface - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [ *self.__base_cmd, ("-A" if direct else "-D"), "INPUT", "-i", self.__iface, "-p", "icmp", "-j", "ACCEPT", @@ -71,13 +67,13 @@ class IptablesAllowIcmpCtl(BaseCtl): class IptablesAllowPortCtl(BaseCtl): - def __init__(self, base_cmd: List[str], iface: str, port: int, tcp: bool) -> None: + def __init__(self, base_cmd: list[str], iface: str, port: int, tcp: bool) -> None: self.__base_cmd = base_cmd self.__iface = iface self.__port = port self.__proto = ("tcp" if tcp else "udp") - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [ *self.__base_cmd, ("-A" if direct else "-D"), "INPUT", "-i", self.__iface, "-p", self.__proto, @@ -86,11 +82,11 @@ class IptablesAllowPortCtl(BaseCtl): class IptablesForwardOut(BaseCtl): - def __init__(self, base_cmd: List[str], iface: str) -> None: + def __init__(self, base_cmd: list[str], iface: str) -> None: self.__base_cmd = base_cmd self.__iface = iface - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [ *self.__base_cmd, "--table", "nat", @@ -100,11 +96,11 @@ class IptablesForwardOut(BaseCtl): class IptablesForwardIn(BaseCtl): - def __init__(self, base_cmd: List[str], iface: str) -> None: + def __init__(self, base_cmd: list[str], iface: str) -> None: self.__base_cmd = base_cmd self.__iface = iface - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [ *self.__base_cmd, ("-A" if direct else "-D"), "FORWARD", @@ -115,16 +111,16 @@ class IptablesForwardIn(BaseCtl): class CustomCtl(BaseCtl): def __init__( self, - direct_cmd: List[str], - reverse_cmd: List[str], - placeholders: Dict[str, str], + direct_cmd: list[str], + reverse_cmd: list[str], + placeholders: dict[str, str], ) -> None: self.__direct_cmd = direct_cmd self.__reverse_cmd = reverse_cmd self.__placeholders = placeholders - def get_command(self, direct: bool) -> List[str]: + def get_command(self, direct: bool) -> list[str]: return [ part.format(**self.__placeholders) for part in (self.__direct_cmd if direct else self.__reverse_cmd) diff --git a/kvmd/apps/pst/__init__.py b/kvmd/apps/pst/__init__.py index 2bb29a08..8f92e39d 100644 --- a/kvmd/apps/pst/__init__.py +++ b/kvmd/apps/pst/__init__.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Optional - from ...logging import get_logger from .. import init @@ -31,7 +28,7 @@ from .server import PstServer # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd-pst", description="The KVMD persistent storage manager", diff --git a/kvmd/apps/pst/server.py b/kvmd/apps/pst/server.py index a492a29b..1df26f61 100644 --- a/kvmd/apps/pst/server.py +++ b/kvmd/apps/pst/server.py @@ -23,9 +23,6 @@ import os import asyncio -from typing import List -from typing import Dict - from aiohttp.web import Request from aiohttp.web import WebSocketResponse @@ -48,7 +45,7 @@ class PstServer(HttpServer): # pylint: disable=too-many-arguments,too-many-inst storage_path: str, ro_retries_delay: float, ro_cleanup_delay: float, - remount_cmd: List[str], + remount_cmd: list[str], ) -> None: super().__init__() @@ -69,7 +66,7 @@ class PstServer(HttpServer): # pylint: disable=too-many-arguments,too-many-inst return (await self._ws_loop(ws)) @exposed_ws("ping") - async def __ws_ping_handler(self, ws: WsSession, _: Dict) -> None: + async def __ws_ping_handler(self, ws: WsSession, _: dict) -> None: await ws.send_event("pong", {}) # ===== SYSTEM STUFF diff --git a/kvmd/apps/pstrun/__init__.py b/kvmd/apps/pstrun/__init__.py index 2b1b895a..c164e533 100644 --- a/kvmd/apps/pstrun/__init__.py +++ b/kvmd/apps/pstrun/__init__.py @@ -27,9 +27,6 @@ import asyncio import asyncio.subprocess import argparse -from typing import List -from typing import Optional - import aiohttp from ...logging import get_logger @@ -53,7 +50,7 @@ def _preexec() -> None: get_logger(0).info("Can't perform tcsetpgrp(0): %s", tools.efmt(err)) -async def _run_process(cmd: List[str], data_path: str) -> asyncio.subprocess.Process: # pylint: disable=no-member +async def _run_process(cmd: list[str], data_path: str) -> asyncio.subprocess.Process: # pylint: disable=no-member # https://stackoverflow.com/questions/58918188/why-is-stdin-not-propagated-to-child-process-of-different-process-group if os.isatty(0): signal.signal(signal.SIGTTOU, signal.SIG_IGN) @@ -67,11 +64,11 @@ async def _run_process(cmd: List[str], data_path: str) -> asyncio.subprocess.Pro )) -async def _run_cmd_ws(cmd: List[str], ws: aiohttp.ClientWebSocketResponse) -> int: # pylint: disable=too-many-branches +async def _run_cmd_ws(cmd: list[str], ws: aiohttp.ClientWebSocketResponse) -> int: # pylint: disable=too-many-branches logger = get_logger(0) - receive_task: Optional[asyncio.Task] = None - proc_task: Optional[asyncio.Task] = None - proc: Optional[asyncio.subprocess.Process] = None # pylint: disable=no-member + receive_task: (asyncio.Task | None) = None + proc_task: (asyncio.Task | None) = None + proc: (asyncio.subprocess.Process | None) = None # pylint: disable=no-member try: # pylint: disable=too-many-nested-blocks while True: @@ -120,7 +117,7 @@ async def _run_cmd_ws(cmd: List[str], ws: aiohttp.ClientWebSocketResponse) -> in return 1 -async def _run_cmd(cmd: List[str], unix_path: str) -> None: +async def _run_cmd(cmd: list[str], unix_path: str) -> None: get_logger(0).info("Opening PST session ...") async with aiohttp.ClientSession( headers={"User-Agent": htclient.make_user_agent("KVMD-PSTRun")}, @@ -132,7 +129,7 @@ async def _run_cmd(cmd: List[str], unix_path: str) -> None: # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: (parent_parser, argv, config) = init( add_help=False, cli_logging=True, diff --git a/kvmd/apps/vnc/__init__.py b/kvmd/apps/vnc/__init__.py index d7e2c68a..3653f416 100644 --- a/kvmd/apps/vnc/__init__.py +++ b/kvmd/apps/vnc/__init__.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Optional - from ...clients.kvmd import KvmdClient from ...clients.streamer import StreamFormats from ...clients.streamer import BaseStreamerClient @@ -38,7 +35,7 @@ from .server import VncServer # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: config = init( prog="kvmd-vnc", description="VNC to KVMD proxy", @@ -48,12 +45,12 @@ def main(argv: Optional[List[str]]=None) -> None: user_agent = htclient.make_user_agent("KVMD-VNC") - def make_memsink_streamer(name: str, fmt: int) -> Optional[MemsinkStreamerClient]: + def make_memsink_streamer(name: str, fmt: int) -> (MemsinkStreamerClient | None): if getattr(config.memsink, name).sink: return MemsinkStreamerClient(name.upper(), fmt, **getattr(config.memsink, name)._unpack()) return None - streamers: List[BaseStreamerClient] = list(filter(None, [ + streamers: list[BaseStreamerClient] = list(filter(None, [ make_memsink_streamer("h264", StreamFormats.H264), make_memsink_streamer("jpeg", StreamFormats.JPEG), HttpStreamerClient(name="JPEG", user_agent=user_agent, **config.streamer._unpack()), diff --git a/kvmd/apps/vnc/rfb/__init__.py b/kvmd/apps/vnc/rfb/__init__.py index 059d4d1d..f715f9b7 100644 --- a/kvmd/apps/vnc/rfb/__init__.py +++ b/kvmd/apps/vnc/rfb/__init__.py @@ -23,9 +23,6 @@ import asyncio import ssl -from typing import Tuple -from typing import List -from typing import Dict from typing import Callable from typing import Coroutine @@ -66,7 +63,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute width: int, height: int, name: str, - vnc_passwds: List[str], + vnc_passwds: list[str], vencrypt: bool, none_auth_only: bool, ) -> None: @@ -103,7 +100,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute finally: await aiotools.shield_fg(self.__cleanup(tasks)) - async def __cleanup(self, tasks: List[asyncio.Task]) -> None: + async def __cleanup(self, tasks: list[asyncio.Task]) -> None: for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) @@ -150,7 +147,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute async def _on_ext_key_event(self, code: int, state: bool) -> None: raise NotImplementedError - async def _on_pointer_event(self, buttons: Dict[str, bool], wheel: Dict[str, int], move: Dict[str, int]) -> None: + async def _on_pointer_event(self, buttons: dict[str, bool], wheel: dict[str, int], move: dict[str, int]) -> None: raise NotImplementedError async def _on_cut_event(self, text: str) -> None: @@ -232,7 +229,7 @@ class RfbClient(RfbClientStream): # pylint: disable=too-many-instance-attribute # ===== async def __handshake_security(self) -> None: - sec_types: Dict[int, Tuple[str, Callable]] = {} + sec_types: dict[int, tuple[str, Callable]] = {} if self.__vencrypt and self.__rfb_version > 3: sec_types[19] = ("VeNCrypt", self.__handshake_security_vencrypt) if self.__none_auth_only: diff --git a/kvmd/apps/vnc/rfb/crypto.py b/kvmd/apps/vnc/rfb/crypto.py index 854dceb3..8953bef5 100644 --- a/kvmd/apps/vnc/rfb/crypto.py +++ b/kvmd/apps/vnc/rfb/crypto.py @@ -22,8 +22,6 @@ import os -from typing import List - import passlib.crypto.des @@ -43,7 +41,7 @@ def rfb_encrypt_challenge(challenge: bytes, passwd: bytes) -> bytes: def _make_key(passwd: bytes) -> bytes: passwd = (passwd + b"\0" * 8)[:8] - key: List[int] = [] + key: list[int] = [] for ch in passwd: btgt = 0 for index in range(8): diff --git a/kvmd/apps/vnc/rfb/encodings.py b/kvmd/apps/vnc/rfb/encodings.py index de51dde5..e153b5e8 100644 --- a/kvmd/apps/vnc/rfb/encodings.py +++ b/kvmd/apps/vnc/rfb/encodings.py @@ -22,10 +22,6 @@ import dataclasses -from typing import List -from typing import Dict -from typing import FrozenSet -from typing import Union from typing import Any @@ -45,13 +41,13 @@ class RfbEncodings: H264 = 50 # Open H.264 Encoding -def _make_meta(variants: Union[int, FrozenSet[int]]) -> Dict: +def _make_meta(variants: (int | frozenset[int])) -> dict: return {"variants": (frozenset([variants]) if isinstance(variants, int) else variants)} @dataclasses.dataclass(frozen=True) class RfbClientEncodings: # pylint: disable=too-many-instance-attributes - encodings: FrozenSet[int] + encodings: frozenset[int] has_resize: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RESIZE)) # noqa: E224 has_rename: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.RENAME)) # noqa: E224 @@ -63,8 +59,8 @@ class RfbClientEncodings: # pylint: disable=too-many-instance-attributes has_h264: bool = dataclasses.field(default=False, metadata=_make_meta(RfbEncodings.H264)) # noqa: E224 - def get_summary(self) -> List[str]: - summary: List[str] = [f"encodings -- {sorted(self.encodings)}"] + def get_summary(self) -> list[str]: + summary: list[str] = [f"encodings -- {sorted(self.encodings)}"] for field in dataclasses.fields(self): if field.name != "encodings": found = ", ".join(map(str, sorted(map(int, self.__get_found(field))))) @@ -80,7 +76,7 @@ class RfbClientEncodings: # pylint: disable=too-many-instance-attributes def __set_value(self, key: str, value: Any) -> None: object.__setattr__(self, key, value) - def __get_found(self, field: dataclasses.Field) -> FrozenSet[int]: + def __get_found(self, field: dataclasses.Field) -> frozenset[int]: return self.encodings.intersection(field.metadata["variants"]) def __get_tight_jpeg_quality(self) -> int: diff --git a/kvmd/apps/vnc/rfb/stream.py b/kvmd/apps/vnc/rfb/stream.py index ee42436e..dfa6cb13 100644 --- a/kvmd/apps/vnc/rfb/stream.py +++ b/kvmd/apps/vnc/rfb/stream.py @@ -24,9 +24,6 @@ import asyncio import ssl import struct -from typing import Tuple -from typing import Union - from .... import aiotools from .errors import RfbConnectionError @@ -57,7 +54,7 @@ class RfbClientStream: except (ConnectionError, asyncio.IncompleteReadError) as err: raise RfbConnectionError(f"Can't read {msg}", err) - async def _read_struct(self, msg: str, fmt: str) -> Tuple[int, ...]: + async def _read_struct(self, msg: str, fmt: str) -> tuple[int, ...]: assert len(fmt) > 1 try: fmt = f">{fmt}" @@ -73,7 +70,7 @@ class RfbClientStream: # ===== - async def _write_struct(self, msg: str, fmt: str, *values: Union[int, bytes], drain: bool=True) -> None: + async def _write_struct(self, msg: str, fmt: str, *values: (int | bytes), drain: bool=True) -> None: try: if not fmt: for value in values: diff --git a/kvmd/apps/vnc/server.py b/kvmd/apps/vnc/server.py index 7fac9aa4..c6ea076a 100644 --- a/kvmd/apps/vnc/server.py +++ b/kvmd/apps/vnc/server.py @@ -26,11 +26,6 @@ import socket import dataclasses import contextlib -from typing import List -from typing import Dict -from typing import Union -from typing import Optional - import aiohttp from ...logging import get_logger @@ -83,12 +78,12 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes desired_fps: int, keymap_name: str, - symmap: Dict[int, Dict[int, str]], + symmap: dict[int, dict[int, str]], kvmd: KvmdClient, - streamers: List[BaseStreamerClient], + streamers: list[BaseStreamerClient], - vnc_credentials: Dict[str, VncAuthKvmdCredentials], + vnc_credentials: dict[str, VncAuthKvmdCredentials], vencrypt: bool, none_auth_only: bool, shared_params: _SharedParams, @@ -122,15 +117,15 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes self.__stage2_encodings_accepted = aiotools.AioStage() self.__stage3_ws_connected = aiotools.AioStage() - self.__kvmd_session: Optional[KvmdClientSession] = None - self.__kvmd_ws: Optional[KvmdClientWs] = None + self.__kvmd_session: (KvmdClientSession | None) = None + self.__kvmd_ws: (KvmdClientWs | None) = None self.__fb_notifier = aiotools.AioNotifier() - self.__fb_queue: "asyncio.Queue[Dict]" = asyncio.Queue() + self.__fb_queue: "asyncio.Queue[dict]" = asyncio.Queue() # Эти состояния шарить не обязательно - бекенд исключает дублирующиеся события. # Все это нужно только чтобы не посылать лишние жсоны в сокет KVMD - self.__mouse_buttons: Dict[str, Optional[bool]] = dict.fromkeys(["left", "right", "middle"], None) + self.__mouse_buttons: dict[str, (bool | None)] = dict.fromkeys(["left", "right", "middle"], None) self.__mouse_move = {"x": -1, "y": -1} self.__lock = asyncio.Lock() @@ -175,7 +170,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes finally: self.__kvmd_ws = None - async def __process_ws_event(self, event_type: str, event: Dict) -> None: + async def __process_ws_event(self, event_type: str, event: dict) -> None: if event_type == "info_meta_state": try: host = event["server"]["host"] @@ -225,7 +220,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes StreamFormats.JPEG: "has_tight", StreamFormats.H264: "has_h264", } - streamer: Optional[BaseStreamerClient] = None + streamer: (BaseStreamerClient | None) = None for streamer in self.__streamers: if getattr(self._encodings, formats[streamer.get_format()]): get_logger(0).info("%s [streamer]: Using preferred %s", self._remote, streamer) @@ -237,12 +232,12 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes get_logger(0).info("%s [streamer]: Using default %s", self._remote, streamer) return streamer - async def __queue_frame(self, frame: Union[Dict, str]) -> None: + async def __queue_frame(self, frame: (dict | str)) -> None: if isinstance(frame, str): frame = await self.__make_text_frame(frame) self.__fb_queue.put_nowait(frame) - async def __make_text_frame(self, text: str) -> Dict: + async def __make_text_frame(self, text: str) -> dict: return { "data": (await make_text_jpeg(self._width, self._height, self._encodings.tight_jpeg_quality, text)), "width": self._width, @@ -252,7 +247,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes async def __fb_sender_task_loop(self) -> None: # pylint: disable=too-many-branches has_h264_key = False - last: Optional[Dict] = None + last: (dict | None) = None while True: await self.__fb_notifier.wait() @@ -351,7 +346,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes if self.__kvmd_ws: await self.__kvmd_ws.send_key_event(web_key, state) - def __switch_modifiers(self, key: Union[int, str], state: bool) -> bool: + def __switch_modifiers(self, key: (int | str), state: bool) -> bool: mod = 0 if key in X11Modifiers.SHIFTS or key in WebModifiers.SHIFTS: mod = SymmapModifiers.SHIFT @@ -367,7 +362,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes self.__modifiers &= ~mod return True - async def _on_pointer_event(self, buttons: Dict[str, bool], wheel: Dict[str, int], move: Dict[str, int]) -> None: + async def _on_pointer_event(self, buttons: dict[str, bool], wheel: dict[str, int], move: dict[str, int]) -> None: if self.__kvmd_ws: for (button, state) in buttons.items(): if self.__mouse_buttons[button] != state: @@ -434,7 +429,7 @@ class VncServer: # pylint: disable=too-many-instance-attributes keymap_path: str, kvmd: KvmdClient, - streamers: List[BaseStreamerClient], + streamers: list[BaseStreamerClient], vnc_auth_manager: VncAuthManager, ) -> None: diff --git a/kvmd/apps/vnc/vncauth.py b/kvmd/apps/vnc/vncauth.py index b0f702d1..c7c04864 100644 --- a/kvmd/apps/vnc/vncauth.py +++ b/kvmd/apps/vnc/vncauth.py @@ -22,9 +22,6 @@ import dataclasses -from typing import Tuple -from typing import Dict - from ...logging import get_logger from ... import aiofs @@ -53,7 +50,7 @@ class VncAuthManager: self.__path = path self.__enabled = enabled - async def read_credentials(self) -> Tuple[Dict[str, VncAuthKvmdCredentials], bool]: + async def read_credentials(self) -> tuple[dict[str, VncAuthKvmdCredentials], bool]: if self.__enabled: try: return (await self.__inner_read_credentials(), True) @@ -63,10 +60,10 @@ class VncAuthManager: get_logger(0).exception("Unhandled exception while reading VNCAuth passwd file") return ({}, (not self.__enabled)) - async def __inner_read_credentials(self) -> Dict[str, VncAuthKvmdCredentials]: + async def __inner_read_credentials(self) -> dict[str, VncAuthKvmdCredentials]: lines = (await aiofs.read(self.__path)).split("\n") - credentials: Dict[str, VncAuthKvmdCredentials] = {} + credentials: dict[str, VncAuthKvmdCredentials] = {} for (lineno, line) in enumerate(lines): if len(line.strip()) == 0 or line.lstrip().startswith("#"): continue diff --git a/kvmd/apps/watchdog/__init__.py b/kvmd/apps/watchdog/__init__.py index e7271ac1..c8a50729 100644 --- a/kvmd/apps/watchdog/__init__.py +++ b/kvmd/apps/watchdog/__init__.py @@ -24,9 +24,6 @@ import argparse import errno import time -from typing import List -from typing import Optional - from ...logging import get_logger from ...yamlconf import Section @@ -104,7 +101,7 @@ def _cmd_cancel(config: Section) -> None: # ===== -def main(argv: Optional[List[str]]=None) -> None: +def main(argv: (list[str] | None)=None) -> None: (parent_parser, argv, config) = init(add_help=False, argv=argv) parser = argparse.ArgumentParser( prog="kvmd-watchdog", diff --git a/kvmd/clients/kvmd.py b/kvmd/clients/kvmd.py index f8b0d811..f98c1064 100644 --- a/kvmd/clients/kvmd.py +++ b/kvmd/clients/kvmd.py @@ -24,13 +24,8 @@ import asyncio import contextlib import types -from typing import Tuple -from typing import Dict -from typing import Set from typing import Callable -from typing import Type from typing import AsyncGenerator -from typing import Optional import aiohttp @@ -65,13 +60,13 @@ class _AuthApiPart(_BaseApiPart): class _StreamerApiPart(_BaseApiPart): - async def get_state(self) -> Dict: + async def get_state(self) -> dict: session = self._ensure_http_session() async with session.get(self._make_url("streamer")) as response: htclient.raise_not_200(response) return (await response.json())["result"] - async def set_params(self, quality: Optional[int]=None, desired_fps: Optional[int]=None) -> None: + async def set_params(self, quality: (int | None)=None, desired_fps: (int | None)=None) -> None: session = self._ensure_http_session() async with session.post( url=self._make_url("streamer/set_params"), @@ -88,7 +83,7 @@ class _StreamerApiPart(_BaseApiPart): class _HidApiPart(_BaseApiPart): - async def get_keymaps(self) -> Tuple[str, Set[str]]: + async def get_keymaps(self) -> tuple[str, set[str]]: session = self._ensure_http_session() async with session.get(self._make_url("hid/keymaps")) as response: htclient.raise_not_200(response) @@ -106,7 +101,7 @@ class _HidApiPart(_BaseApiPart): class _AtxApiPart(_BaseApiPart): - async def get_state(self) -> Dict: + async def get_state(self) -> dict: session = self._ensure_http_session() async with session.get(self._make_url("atx")) as response: htclient.raise_not_200(response) @@ -132,14 +127,14 @@ class KvmdClientWs: def __init__(self, ws: aiohttp.ClientWebSocketResponse) -> None: self.__ws = ws - self.__writer_queue: "asyncio.Queue[Tuple[str, Dict]]" = asyncio.Queue() + self.__writer_queue: "asyncio.Queue[tuple[str, dict]]" = asyncio.Queue() self.__communicated = False - async def communicate(self) -> AsyncGenerator[Tuple[str, Dict], None]: # pylint: disable=too-many-branches + async def communicate(self) -> AsyncGenerator[tuple[str, dict], None]: # pylint: disable=too-many-branches assert not self.__communicated self.__communicated = True - receive_task: Optional[asyncio.Task] = None - writer_task: Optional[asyncio.Task] = None + receive_task: (asyncio.Task | None) = None + writer_task: (asyncio.Task | None) = None try: while True: if receive_task is None: @@ -199,7 +194,7 @@ class KvmdClientSession: self.__make_http_session = make_http_session self.__make_url = make_url - self.__http_session: Optional[aiohttp.ClientSession] = None + self.__http_session: (aiohttp.ClientSession | None) = None args = (self.__ensure_http_session, make_url) @@ -229,7 +224,7 @@ class KvmdClientSession: async def __aexit__( self, - _exc_type: Type[BaseException], + _exc_type: type[BaseException], _exc: BaseException, _tb: types.TracebackType, ) -> None: @@ -256,7 +251,7 @@ class KvmdClient: ) def __make_http_session(self, user: str, passwd: str) -> aiohttp.ClientSession: - kwargs: Dict = { + kwargs: dict = { "headers": { "X-KVMD-User": user, "X-KVMD-Passwd": passwd, diff --git a/kvmd/clients/streamer.py b/kvmd/clients/streamer.py index dd4167f9..f26fed1e 100644 --- a/kvmd/clients/streamer.py +++ b/kvmd/clients/streamer.py @@ -22,7 +22,6 @@ import types -from typing import Dict from typing import AsyncGenerator import aiohttp @@ -61,7 +60,7 @@ class BaseStreamerClient: def get_format(self) -> int: raise NotImplementedError() - async def read_stream(self) -> AsyncGenerator[Dict, None]: + async def read_stream(self) -> AsyncGenerator[dict, None]: if self is not None: # XXX: Vulture and pylint hack raise NotImplementedError() yield @@ -84,7 +83,7 @@ class HttpStreamerClient(BaseStreamerClient): def get_format(self) -> int: return StreamFormats.JPEG - async def read_stream(self) -> AsyncGenerator[Dict, None]: + async def read_stream(self) -> AsyncGenerator[dict, None]: try: async with self.__make_http_session() as session: async with session.get( @@ -118,7 +117,7 @@ class HttpStreamerClient(BaseStreamerClient): raise StreamerTempError("Reached EOF") def __make_http_session(self) -> aiohttp.ClientSession: - kwargs: Dict = { + kwargs: dict = { "headers": {"User-Agent": self.__user_agent}, "connector": aiohttp.UnixConnector(path=self.__unix_path), "timeout": aiohttp.ClientTimeout( @@ -162,7 +161,7 @@ class MemsinkStreamerClient(BaseStreamerClient): self.__name = name self.__fmt = fmt - self.__kwargs: Dict = { + self.__kwargs: dict = { "obj": obj, "lock_timeout": lock_timeout, "wait_timeout": wait_timeout, @@ -172,7 +171,7 @@ class MemsinkStreamerClient(BaseStreamerClient): def get_format(self) -> int: return self.__fmt - async def read_stream(self) -> AsyncGenerator[Dict, None]: + async def read_stream(self) -> AsyncGenerator[dict, None]: if ustreamer is None: raise StreamerPermError("Missing ustreamer library") try: diff --git a/kvmd/helpers/remount/__init__.py b/kvmd/helpers/remount/__init__.py index 9ea2889d..80e683f5 100644 --- a/kvmd/helpers/remount/__init__.py +++ b/kvmd/helpers/remount/__init__.py @@ -27,8 +27,6 @@ import shutil import dataclasses import subprocess -from typing import List - # ==== _MOUNT_PATH = "/bin/mount" @@ -98,7 +96,7 @@ def main() -> None: raise SystemExit(f"Usage: {sys.argv[0]} [ro|rw]") target = "" - dirs: List[str] = [] + dirs: list[str] = [] app = os.path.basename(sys.argv[0]) if app == "kvmd-helper-otgmsd-remount": target = "otgmsd" diff --git a/kvmd/htclient.py b/kvmd/htclient.py index e2ae0de4..d8592175 100644 --- a/kvmd/htclient.py +++ b/kvmd/htclient.py @@ -23,9 +23,7 @@ import os import contextlib -from typing import Dict from typing import AsyncGenerator -from typing import Optional import aiohttp import aiohttp.multipart @@ -68,11 +66,11 @@ async def download( url: str, verify: bool=True, timeout: float=10.0, - read_timeout: Optional[float]=None, + read_timeout: (float | None)=None, app: str="KVMD", ) -> AsyncGenerator[aiohttp.ClientResponse, None]: - kwargs: Dict = { + kwargs: dict = { "headers": {"User-Agent": make_user_agent(app)}, "timeout": aiohttp.ClientTimeout( connect=timeout, diff --git a/kvmd/htserver.py b/kvmd/htserver.py index a83fc17c..f47c1fc9 100644 --- a/kvmd/htserver.py +++ b/kvmd/htserver.py @@ -29,13 +29,8 @@ import inspect import urllib.parse import json -from typing import Tuple -from typing import List -from typing import Dict from typing import Callable from typing import AsyncGenerator -from typing import Optional -from typing import Union from typing import Any from aiohttp import ClientWebSocketResponse @@ -111,7 +106,7 @@ def exposed_http(http_method: str, path: str, auth_required: bool=True) -> Calla return set_attrs -def _get_exposed_http(obj: object) -> List[HttpExposed]: +def _get_exposed_http(obj: object) -> list[HttpExposed]: return [ HttpExposed( method=getattr(handler, _HTTP_METHOD), @@ -143,7 +138,7 @@ def exposed_ws(event_type: str) -> Callable: return set_attrs -def _get_exposed_ws(obj: object) -> List[WsExposed]: +def _get_exposed_ws(obj: object) -> list[WsExposed]: return [ WsExposed( event_type=getattr(handler, _WS_EVENT_TYPE), @@ -156,9 +151,9 @@ def _get_exposed_ws(obj: object) -> List[WsExposed]: # ===== def make_json_response( - result: Optional[Dict]=None, + result: (dict | None)=None, status: int=200, - set_cookies: Optional[Dict[str, str]]=None, + set_cookies: (dict[str, str] | None)=None, wrap_result: bool=True, ) -> Response: @@ -176,7 +171,7 @@ def make_json_response( return response -def make_json_exception(err: Exception, status: Optional[int]=None) -> Response: +def make_json_exception(err: Exception, status: (int | None)=None) -> Response: name = type(err).__name__ msg = str(err) if isinstance(err, HttpError): @@ -208,7 +203,7 @@ async def start_streaming( return response -async def stream_json(response: StreamResponse, result: Dict, ok: bool=True) -> None: +async def stream_json(response: StreamResponse, result: dict, ok: bool=True) -> None: await response.write(json.dumps({ "ok": ok, "result": result, @@ -226,9 +221,9 @@ async def stream_json_exception(response: StreamResponse, err: Exception) -> Non async def send_ws_event( - wsr: Union[ClientWebSocketResponse, WebSocketResponse], + wsr: (ClientWebSocketResponse | WebSocketResponse), event_type: str, - event: Optional[Dict], + event: (dict | None), ) -> None: await wsr.send_str(json.dumps({ @@ -237,7 +232,7 @@ async def send_ws_event( })) -def parse_ws_event(msg: str) -> Tuple[str, Dict]: +def parse_ws_event(msg: str) -> tuple[str, dict]: data = json.loads(msg) if not isinstance(data, dict): raise RuntimeError("Top-level event structure is not a dict") @@ -269,20 +264,20 @@ def set_request_auth_info(request: BaseRequest, info: str) -> None: @dataclasses.dataclass(frozen=True) class WsSession: wsr: WebSocketResponse - kwargs: Dict[str, Any] + kwargs: dict[str, Any] def __str__(self) -> str: return f"WsSession(id={id(self)}, {self.kwargs})" - async def send_event(self, event_type: str, event: Optional[Dict]) -> None: + async def send_event(self, event_type: str, event: (dict | None)) -> None: await send_ws_event(self.wsr, event_type, event) class HttpServer: def __init__(self) -> None: - self.__ws_heartbeat: Optional[float] = None - self.__ws_handlers: Dict[str, Callable] = {} - self.__ws_sessions: List[WsSession] = [] + self.__ws_heartbeat: (float | None) = None + self.__ws_handlers: dict[str, Callable] = {} + self.__ws_sessions: list[WsSession] = [] self.__ws_sessions_lock = asyncio.Lock() def run( @@ -373,7 +368,7 @@ class HttpServer: logger.error("Unknown websocket event: %r", msg.data) return ws.wsr - async def _broadcast_ws_event(self, event_type: str, event: Optional[Dict]) -> None: + async def _broadcast_ws_event(self, event_type: str, event: (dict | None)) -> None: if self.__ws_sessions: await asyncio.gather(*[ ws.send_event(event_type, event) @@ -391,7 +386,7 @@ class HttpServer: await self.__close_ws(ws) return bool(wss) - def _get_wss(self) -> List[WsSession]: + def _get_wss(self) -> list[WsSession]: return list(self.__ws_sessions) async def __close_ws(self, ws: WsSession) -> None: diff --git a/kvmd/inotify.py b/kvmd/inotify.py index 6317e62d..0511f088 100644 --- a/kvmd/inotify.py +++ b/kvmd/inotify.py @@ -31,12 +31,7 @@ import dataclasses import types import errno -from typing import Tuple -from typing import List -from typing import Dict -from typing import Type from typing import Generator -from typing import Optional from .logging import get_logger @@ -53,7 +48,7 @@ _FS_ENCODING = (sys.getfilesystemencoding() or _FS_FALLBACK_ENCODING) # ===== -def _inotify_parsed_buffer(data: bytes) -> Generator[Tuple[int, int, int, bytes], None, None]: +def _inotify_parsed_buffer(data: bytes) -> Generator[tuple[int, int, int, bytes], None, None]: offset = 0 while offset + _EVENT_HEAD_SIZE <= len(data): (wd, mask, cookie, length) = struct.unpack_from("iIII", data, offset) @@ -157,7 +152,7 @@ class InotifyMask: @classmethod def to_string(cls, mask: int) -> str: - flags: List[str] = [] + flags: list[str] = [] for name in dir(cls): if ( name[0].isupper() @@ -188,10 +183,10 @@ class Inotify: def __init__(self) -> None: self.__fd = -1 - self.__wd_by_path: Dict[str, int] = {} - self.__path_by_wd: Dict[int, str] = {} + self.__wd_by_path: dict[str, int] = {} + self.__path_by_wd: dict[int, str] = {} - self.__moved: Dict[int, str] = {} + self.__moved: dict[int, str] = {} self.__events_queue: "asyncio.Queue[InotifyEvent]" = asyncio.Queue() @@ -215,15 +210,15 @@ class Inotify: # def has_events(self) -> bool: # return (not self.__events_queue.empty()) - async def get_event(self, timeout: float) -> Optional[InotifyEvent]: + async def get_event(self, timeout: float) -> (InotifyEvent | None): assert timeout > 0 try: return (await asyncio.wait_for(self.__events_queue.get(), timeout=timeout)) except asyncio.TimeoutError: return None - async def get_series(self, timeout: float) -> List[InotifyEvent]: - series: List[InotifyEvent] = [] + async def get_series(self, timeout: float) -> list[InotifyEvent]: + series: list[InotifyEvent] = [] event = await self.get_event(timeout) if event: series.append(event) @@ -283,7 +278,7 @@ class Inotify: def __exit__( self, - _exc_type: Type[BaseException], + _exc_type: type[BaseException], _exc: BaseException, _tb: types.TracebackType, ) -> None: diff --git a/kvmd/keyboard/keysym.py b/kvmd/keyboard/keysym.py index 623155f3..e9697717 100644 --- a/kvmd/keyboard/keysym.py +++ b/kvmd/keyboard/keysym.py @@ -24,9 +24,6 @@ import pkgutil import functools import importlib.machinery -from typing import List -from typing import Dict - import Xlib.keysymdef from ..logging import get_logger @@ -44,11 +41,11 @@ class SymmapModifiers: CTRL: int = 0x4 -def build_symmap(path: str) -> Dict[int, Dict[int, str]]: +def build_symmap(path: str) -> dict[int, dict[int, str]]: # https://github.com/qemu/qemu/blob/95a9457fd44ad97c518858a4e1586a5498f9773c/ui/keymaps.c logger = get_logger() - symmap: Dict[int, Dict[int, str]] = {} + symmap: dict[int, dict[int, str]] = {} for (src, items) in [ ("<builtin>", list(X11_TO_AT1.items())), (path, list(_read_keyboard_layout(path).items())), @@ -78,8 +75,8 @@ def build_symmap(path: str) -> Dict[int, Dict[int, str]]: # ===== @functools.lru_cache() -def _get_keysyms() -> Dict[str, int]: - keysyms: Dict[str, int] = {} +def _get_keysyms() -> dict[str, int]: + keysyms: dict[str, int] = {} for (finder, module_name, _) in pkgutil.walk_packages(Xlib.keysymdef.__path__): if not isinstance(finder, importlib.machinery.FileFinder): continue @@ -109,14 +106,14 @@ def _resolve_keysym(name: str) -> int: return 0 -def _read_keyboard_layout(path: str) -> Dict[int, List[At1Key]]: # Keysym to evdev (at1) +def _read_keyboard_layout(path: str) -> dict[int, list[At1Key]]: # Keysym to evdev (at1) logger = get_logger(0) logger.info("Reading keyboard layout %s ...", path) with open(path) as layout_file: lines = list(map(str.strip, layout_file.read().split("\n"))) - layout: Dict[int, List[At1Key]] = {} + layout: dict[int, list[At1Key]] = {} for (lineno, line) in enumerate(lines): if len(line) == 0 or line.startswith(("#", "map ", "include ")): continue diff --git a/kvmd/keyboard/mappings.py.mako b/kvmd/keyboard/mappings.py.mako index 91d46297..b8a6596c 100644 --- a/kvmd/keyboard/mappings.py.mako +++ b/kvmd/keyboard/mappings.py.mako @@ -22,8 +22,6 @@ import dataclasses -from typing import Dict - # ===== @dataclasses.dataclass(frozen=True) @@ -43,7 +41,7 @@ class Key: usb: UsbKey <%! import operator %> -KEYMAP: Dict[str, Key] = { +KEYMAP: dict[str, Key] = { % for km in sorted(keymap, key=operator.attrgetter("mcu_code")): "${km.web_name}": Key(mcu=McuKey(code=${km.mcu_code}), usb=UsbKey(code=${km.usb_key.code}, is_modifier=${km.usb_key.is_modifier})), % endfor diff --git a/kvmd/keyboard/printer.py b/kvmd/keyboard/printer.py index 20ff9a56..7261a8ef 100644 --- a/kvmd/keyboard/printer.py +++ b/kvmd/keyboard/printer.py @@ -20,8 +20,6 @@ # ========================================================================== # -from typing import Tuple -from typing import Dict from typing import Generator from .keysym import SymmapModifiers @@ -31,9 +29,9 @@ from .mappings import WebModifiers # ===== def text_to_web_keys( # pylint: disable=too-many-branches text: str, - symmap: Dict[int, Dict[int, str]], + symmap: dict[int, dict[int, str]], shift_key: str=WebModifiers.SHIFT_LEFT, -) -> Generator[Tuple[str, bool], None, None]: +) -> Generator[tuple[str, bool], None, None]: assert shift_key in WebModifiers.SHIFTS diff --git a/kvmd/logging.py b/kvmd/logging.py index e1a75bc3..c4dadbfd 100644 --- a/kvmd/logging.py +++ b/kvmd/logging.py @@ -24,12 +24,10 @@ import sys import types import logging -from typing import Optional - # ===== def get_logger(depth: int=1) -> logging.Logger: - frame: Optional[types.FrameType] = sys._getframe(1) # pylint: disable=protected-access + frame: (types.FrameType | None) = sys._getframe(1) # pylint: disable=protected-access assert frame frames = [] while frame: 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) diff --git a/kvmd/tools.py b/kvmd/tools.py index 77a81d64..a21e085e 100644 --- a/kvmd/tools.py +++ b/kvmd/tools.py @@ -26,9 +26,6 @@ import multiprocessing.queues import queue import shlex -from typing import Tuple -from typing import List -from typing import Dict from typing import Hashable from typing import TypeVar @@ -39,7 +36,7 @@ def remap(value: int, in_min: int, in_max: int, out_min: int, out_max: int) -> i # ===== -def cmdfmt(cmd: List[str]) -> str: +def cmdfmt(cmd: list[str]) -> str: return " ".join(map(shlex.quote, cmd)) @@ -48,7 +45,7 @@ def efmt(err: Exception) -> str: # ===== -def merge(dest: Dict, src: Dict) -> None: +def merge(dest: dict, src: dict) -> None: for key in src: if key in dest: if isinstance(dest[key], dict) and isinstance(src[key], dict): @@ -57,7 +54,7 @@ def merge(dest: Dict, src: Dict) -> None: dest[key] = src[key] -def rget(dct: Dict, *keys: Hashable) -> Dict: +def rget(dct: dict, *keys: Hashable) -> dict: result = functools.reduce((lambda nxt, key: nxt.get(key, {})), keys, dct) if not isinstance(result, dict): raise TypeError(f"Not a dict as result: {result!r} from {dct!r} at {list(keys)}") @@ -68,11 +65,11 @@ _DictKeyT = TypeVar("_DictKeyT") _DictValueT = TypeVar("_DictValueT") -def sorted_kvs(dct: Dict[_DictKeyT, _DictValueT]) -> List[Tuple[_DictKeyT, _DictValueT]]: +def sorted_kvs(dct: dict[_DictKeyT, _DictValueT]) -> list[tuple[_DictKeyT, _DictValueT]]: return sorted(dct.items(), key=operator.itemgetter(0)) -def swapped_kvs(dct: Dict[_DictKeyT, _DictValueT]) -> Dict[_DictValueT, _DictKeyT]: +def swapped_kvs(dct: dict[_DictKeyT, _DictValueT]) -> dict[_DictValueT, _DictKeyT]: return {value: key for (key, value) in dct.items()} @@ -86,7 +83,7 @@ def clear_queue(q: multiprocessing.queues.Queue) -> None: # pylint: disable=inv # ===== -def build_cmd(cmd: List[str], cmd_remove: List[str], cmd_append: List[str]) -> List[str]: +def build_cmd(cmd: list[str], cmd_remove: list[str], cmd_append: list[str]) -> list[str]: assert len(cmd) >= 1, cmd return [ cmd[0], # Executable diff --git a/kvmd/validators/__init__.py b/kvmd/validators/__init__.py index 0e9886cf..db8fb6a7 100644 --- a/kvmd/validators/__init__.py +++ b/kvmd/validators/__init__.py @@ -22,14 +22,11 @@ import re -from typing import List -from typing import Set from typing import Mapping from typing import Sequence from typing import Callable from typing import TypeVar from typing import NoReturn -from typing import Union from typing import Any @@ -63,7 +60,7 @@ def check_not_none_string(arg: Any, name: str, strip: bool=True) -> str: return arg -def check_in_list(arg: Any, name: str, variants: Union[Sequence, Mapping, Set]) -> Any: +def check_in_list(arg: Any, name: str, variants: (Sequence | Mapping | set)) -> Any: if arg not in variants: raise_error(arg, name) return arg @@ -72,7 +69,7 @@ def check_in_list(arg: Any, name: str, variants: Union[Sequence, Mapping, Set]) def check_string_in_list( arg: Any, name: str, - variants: Union[Sequence[str], Mapping[str, Any], Set[str]], + variants: (Sequence[str] | Mapping[str, Any] | set[str]), lower: bool=True, ) -> str: @@ -95,7 +92,7 @@ def check_len(arg: _RetvalSeqT, name: str, limit: int) -> _RetvalSeqT: return arg -def check_any(arg: Any, name: str, validators: List[Callable[[Any], Any]]) -> Any: # pylint: disable=inconsistent-return-statements +def check_any(arg: Any, name: str, validators: list[Callable[[Any], Any]]) -> Any: # pylint: disable=inconsistent-return-statements for validator in validators: try: return validator(arg) diff --git a/kvmd/validators/auth.py b/kvmd/validators/auth.py index bb2fad14..025110e6 100644 --- a/kvmd/validators/auth.py +++ b/kvmd/validators/auth.py @@ -20,7 +20,6 @@ # ========================================================================== # -from typing import List from typing import Any from .basic import valid_string_list @@ -33,7 +32,7 @@ def valid_user(arg: Any) -> str: return check_re_match(arg, "username characters", r"^[a-z_][a-z0-9_-]*$") -def valid_users_list(arg: Any) -> List[str]: +def valid_users_list(arg: Any) -> list[str]: return valid_string_list(arg, subval=valid_user, name="users list") diff --git a/kvmd/validators/basic.py b/kvmd/validators/basic.py index e72ad558..059ec226 100644 --- a/kvmd/validators/basic.py +++ b/kvmd/validators/basic.py @@ -22,11 +22,8 @@ import re -from typing import List from typing import Type from typing import Callable -from typing import Optional -from typing import Union from typing import Any from . import ValidatorError @@ -63,11 +60,11 @@ def valid_bool(arg: Any) -> bool: def valid_number( arg: Any, - min: Union[int, float, None]=None, # pylint: disable=redefined-builtin - max: Union[int, float, None]=None, # pylint: disable=redefined-builtin - type: Union[Type[int], Type[float]]=int, # pylint: disable=redefined-builtin + min: (int | float | None)=None, # pylint: disable=redefined-builtin + max: (int | float | None)=None, # pylint: disable=redefined-builtin + type: (Type[int] | Type[float])=int, # pylint: disable=redefined-builtin name: str="", -) -> Union[int, float]: +) -> (int | float): name = (name or type.__name__) @@ -103,9 +100,9 @@ def valid_float_f01(arg: Any) -> float: def valid_string_list( arg: Any, delim: str=r"[,\t ]+", - subval: Optional[Callable[[Any], Any]]=None, + subval: (Callable[[Any], Any] | None)=None, name: str="", -) -> List[str]: +) -> list[str]: if not name: name = "string list" diff --git a/kvmd/validators/kvm.py b/kvmd/validators/kvm.py index a6a6060a..600aa5de 100644 --- a/kvmd/validators/kvm.py +++ b/kvmd/validators/kvm.py @@ -20,7 +20,6 @@ # ========================================================================== # -from typing import Set from typing import Any from . import raise_error @@ -45,7 +44,7 @@ def valid_msd_image_name(arg: Any) -> str: return valid_printable_filename(arg, name="MSD image name") # pragma: nocover -def valid_info_fields(arg: Any, variants: Set[str]) -> Set[str]: +def valid_info_fields(arg: Any, variants: set[str]) -> set[str]: return set(valid_string_list( arg=str(arg).strip(), subval=(lambda field: check_string_in_list(field, "info field", variants)), diff --git a/kvmd/validators/net.py b/kvmd/validators/net.py index 2ff8f495..f92dc978 100644 --- a/kvmd/validators/net.py +++ b/kvmd/validators/net.py @@ -23,7 +23,6 @@ import ipaddress import ssl -from typing import List from typing import Callable from typing import Any @@ -52,8 +51,8 @@ def valid_ip_or_host(arg: Any) -> str: def valid_ip(arg: Any, v4: bool=True, v6: bool=True) -> str: assert v4 or v6 - validators: List[Callable] = [] - versions: List[str] = [] + validators: list[Callable] = [] + versions: list[str] = [] if v4: validators.append(lambda arg: str(ipaddress.IPv4Address(arg))) versions.append("4") @@ -70,8 +69,8 @@ def valid_ip(arg: Any, v4: bool=True, v6: bool=True) -> str: def valid_net(arg: Any, v4: bool=True, v6: bool=True) -> str: assert v4 or v6 - validators: List[Callable] = [] - versions: List[str] = [] + validators: list[Callable] = [] + versions: list[str] = [] if v4: validators.append(lambda arg: str(ipaddress.IPv4Network(arg))) versions.append("4") @@ -99,7 +98,7 @@ def valid_port(arg: Any) -> int: return int(valid_number(arg, min=0, max=65535, name="network port")) -def valid_ports_list(arg: Any) -> List[int]: +def valid_ports_list(arg: Any) -> list[int]: return list(map(int, valid_string_list(arg, subval=valid_port, name="ports list"))) diff --git a/kvmd/validators/os.py b/kvmd/validators/os.py index 4cf107a7..97aabdf0 100644 --- a/kvmd/validators/os.py +++ b/kvmd/validators/os.py @@ -23,7 +23,6 @@ import os import stat -from typing import List from typing import Any from . import raise_error @@ -94,13 +93,13 @@ def valid_unix_mode(arg: Any) -> int: return int(valid_number(arg, min=0, name="UNIX mode")) -def valid_options(arg: Any, name: str="") -> List[str]: +def valid_options(arg: Any, name: str="") -> list[str]: if not name: name = "options" return valid_string_list(arg, delim=r"[,\t]+", name=name) -def valid_command(arg: Any) -> List[str]: +def valid_command(arg: Any) -> list[str]: cmd = valid_options(arg, name="command") if len(cmd) == 0: raise_error(arg, "command") diff --git a/kvmd/validators/ugpio.py b/kvmd/validators/ugpio.py index 6f32a26f..c05aa75d 100644 --- a/kvmd/validators/ugpio.py +++ b/kvmd/validators/ugpio.py @@ -20,9 +20,6 @@ # ========================================================================== # -from typing import List -from typing import Set -from typing import Optional from typing import Any from . import raise_error @@ -32,7 +29,7 @@ from . import check_len # ===== -def valid_ugpio_driver(arg: Any, variants: Optional[Set[str]]=None) -> str: +def valid_ugpio_driver(arg: Any, variants: (set[str] | None)=None) -> str: name = "GPIO driver" arg = check_len(check_re_match(arg, name, r"^[a-zA-Z_][a-zA-Z0-9_-]*$"), name, 255) if variants is not None: @@ -45,11 +42,11 @@ def valid_ugpio_channel(arg: Any) -> str: return check_len(check_re_match(arg, name, r"^[a-zA-Z_][a-zA-Z0-9_.-]*$"), name, 255) -def valid_ugpio_mode(arg: Any, variants: Set[str]) -> str: +def valid_ugpio_mode(arg: Any, variants: set[str]) -> str: return check_string_in_list(arg, "GPIO driver's pin mode", variants) -def valid_ugpio_view_table(arg: Any) -> List[List[str]]: # pylint: disable=inconsistent-return-statements +def valid_ugpio_view_table(arg: Any) -> list[list[str]]: # pylint: disable=inconsistent-return-statements try: return [list(map(str, row)) for row in list(arg)] except Exception: diff --git a/kvmd/yamlconf/__init__.py b/kvmd/yamlconf/__init__.py index d32e5fd3..faf8fa63 100644 --- a/kvmd/yamlconf/__init__.py +++ b/kvmd/yamlconf/__init__.py @@ -23,12 +23,8 @@ import contextlib import json -from typing import Tuple -from typing import List -from typing import Dict from typing import Generator from typing import Callable -from typing import Optional from typing import Any @@ -38,8 +34,8 @@ class ConfigError(ValueError): # ===== -def build_raw_from_options(options: List[str]) -> Dict[str, Any]: - raw: Dict[str, Any] = {} +def build_raw_from_options(options: list[str]) -> dict[str, Any]: + raw: dict[str, Any] = {} for option in options: (key, value) = (option.split("=", 1) + [None])[:2] # type: ignore if len(key.strip()) == 0: @@ -71,12 +67,12 @@ def _parse_value(value: str) -> Any: class Section(dict): def __init__(self) -> None: dict.__init__(self) - self.__meta: Dict[str, Dict[str, Any]] = {} + self.__meta: dict[str, dict[str, Any]] = {} - def _unpack(self, ignore: Optional[List[str]]=None) -> Dict[str, Any]: + def _unpack(self, ignore: (list[str] | None)=None) -> dict[str, Any]: if ignore is None: ignore = [] - unpacked: Dict[str, Any] = {} + unpacked: dict[str, Any] = {} for (key, value) in self.items(): if key not in ignore: if isinstance(value, Section): @@ -118,7 +114,7 @@ class Option: def __init__( self, default: Any, - type: Optional[Callable[[Any], Any]]=None, # pylint: disable=redefined-builtin + type: (Callable[[Any], Any] | None)=None, # pylint: disable=redefined-builtin if_none: Any=Stub, if_empty: Any=Stub, only_if: str="", @@ -150,13 +146,13 @@ def manual_validated(value: Any, *path: str) -> Generator[None, None, None]: raise ConfigError(f"Invalid value {value!r} for key {'/'.join(path)!r}: {err}") -def make_config(raw: Dict[str, Any], scheme: Dict[str, Any], _keys: Tuple[str, ...]=()) -> Section: +def make_config(raw: dict[str, Any], scheme: dict[str, Any], _keys: tuple[str, ...]=()) -> Section: if not isinstance(raw, dict): raise ConfigError(f"The node {('/'.join(_keys) or '/')!r} must be a dictionary") config = Section() - def make_full_key(key: str) -> Tuple[str, ...]: + def make_full_key(key: str) -> tuple[str, ...]: return _keys + (key,) def make_full_name(key: str) -> str: diff --git a/kvmd/yamlconf/loader.py b/kvmd/yamlconf/loader.py index 682d5979..2dedc017 100644 --- a/kvmd/yamlconf/loader.py +++ b/kvmd/yamlconf/loader.py @@ -22,8 +22,6 @@ import os -from typing import List -from typing import Dict from typing import IO from typing import Any @@ -52,7 +50,7 @@ class _YamlLoader(yaml.SafeLoader): self.__root = os.path.dirname(yaml_file.name) def include(self, node: yaml.nodes.Node) -> Any: - incs: List[str] + incs: list[str] if isinstance(node, yaml.nodes.SequenceNode): incs = [ str(child) @@ -63,8 +61,8 @@ class _YamlLoader(yaml.SafeLoader): incs = [str(self.construct_scalar(node))] # type: ignore return self.__inner_include(list(filter(None, incs))) - def __inner_include(self, incs: List[str]) -> Any: - tree: Dict = {} + def __inner_include(self, incs: list[str]) -> Any: + tree: dict = {} for inc in filter(None, incs): assert inc, inc inc_path = os.path.join(self.__root, inc) diff --git a/testenv/linters/mypy.ini b/testenv/linters/mypy.ini index 00dd6881..33877875 100644 --- a/testenv/linters/mypy.ini +++ b/testenv/linters/mypy.ini @@ -1,5 +1,5 @@ [mypy] -python_version = 3.9 +python_version = 3.10 ignore_missing_imports = true disallow_untyped_defs = true strict_optional = true diff --git a/testenv/tests/apps/htpasswd/test_main.py b/testenv/tests/apps/htpasswd/test_main.py index 670539f2..e98281e1 100644 --- a/testenv/tests/apps/htpasswd/test_main.py +++ b/testenv/tests/apps/htpasswd/test_main.py @@ -26,7 +26,6 @@ import tempfile import builtins import getpass -from typing import List from typing import Generator from typing import Any @@ -54,7 +53,7 @@ def _htpasswd_fixture(request) -> Generator[passlib.apache.HtpasswdFile, None, N os.remove(path) -def _run_htpasswd(cmd: List[str], htpasswd_path: str, internal_type: str="htpasswd") -> None: +def _run_htpasswd(cmd: list[str], htpasswd_path: str, internal_type: str="htpasswd") -> None: cmd = ["kvmd-htpasswd", *cmd, "--set-options"] if internal_type != "htpasswd": # By default cmd.append("kvmd/auth/internal/type=" + internal_type) diff --git a/testenv/tests/apps/kvmd/test_auth.py b/testenv/tests/apps/kvmd/test_auth.py index 742e7a2c..4a9b8766 100644 --- a/testenv/tests/apps/kvmd/test_auth.py +++ b/testenv/tests/apps/kvmd/test_auth.py @@ -23,10 +23,7 @@ import os import contextlib -from typing import List -from typing import Dict from typing import AsyncGenerator -from typing import Optional import passlib.apache @@ -40,7 +37,7 @@ from kvmd.plugins.auth import get_auth_service_class # ===== -def _make_service_kwargs(path: str) -> Dict: +def _make_service_kwargs(path: str) -> dict: cls = get_auth_service_class("htpasswd") scheme = cls.get_plugin_options() return make_config({"file": path}, scheme)._unpack() @@ -50,7 +47,7 @@ def _make_service_kwargs(path: str) -> Dict: async def _get_configured_manager( internal_path: str, external_path: str="", - force_internal_users: Optional[List[str]]=None, + force_internal_users: (list[str] | None)=None, ) -> AsyncGenerator[AuthManager, None]: manager = AuthManager( diff --git a/testenv/tests/plugins/auth/test_http.py b/testenv/tests/plugins/auth/test_http.py index 7a13f9b7..a02df543 100644 --- a/testenv/tests/plugins/auth/test_http.py +++ b/testenv/tests/plugins/auth/test_http.py @@ -20,7 +20,6 @@ # ========================================================================== # -from typing import Dict from typing import AsyncGenerator import aiohttp.web @@ -68,7 +67,7 @@ async def _auth_server_port_fixture(aiohttp_server) -> AsyncGenerator[int, None] {"verify": False}, {"user": "server-admin", "passwd": "server-pass"}, ]) -async def test_ok(auth_server_port: int, kwargs: Dict) -> None: +async def test_ok(auth_server_port: int, kwargs: dict) -> None: url = "http://localhost:%d/%s" % ( auth_server_port, ("auth_plus_basic" if kwargs.get("user") else "auth"), diff --git a/testenv/tests/plugins/auth/test_pam.py b/testenv/tests/plugins/auth/test_pam.py index ea524cb4..d165365b 100644 --- a/testenv/tests/plugins/auth/test_pam.py +++ b/testenv/tests/plugins/auth/test_pam.py @@ -24,9 +24,7 @@ import os import asyncio import pwd -from typing import Dict from typing import AsyncGenerator -from typing import Optional import pytest import pytest_asyncio @@ -41,7 +39,7 @@ _PASSWD = "query" # ===== -async def _run_process(cmd: str, input: Optional[str]=None) -> None: # pylint: disable=redefined-builtin +async def _run_process(cmd: str, input: (str | None)=None) -> None: # pylint: disable=redefined-builtin proc = await asyncio.create_subprocess_exec( *cmd.split(" "), stdin=(asyncio.subprocess.PIPE if input is not None else None), @@ -75,7 +73,7 @@ async def _test_user() -> AsyncGenerator[None, None]: {"allow_users": [_USER]}, {"allow_uids_at": _UID}, ]) -async def test_ok(test_user, kwargs: Dict) -> None: # type: ignore +async def test_ok(test_user, kwargs: dict) -> None: # type: ignore _ = test_user async with get_configured_auth_service("pam", **kwargs) as service: assert not (await service.authorize(_USER, "invalid_password")) @@ -88,7 +86,7 @@ async def test_ok(test_user, kwargs: Dict) -> None: # type: ignore {"deny_users": [_USER]}, {"allow_uids_at": _UID + 1}, ]) -async def test_fail(test_user, kwargs: Dict) -> None: # type: ignore +async def test_fail(test_user, kwargs: dict) -> None: # type: ignore _ = test_user async with get_configured_auth_service("pam", **kwargs) as service: assert not (await service.authorize(_USER, "invalid_password")) diff --git a/testenv/tests/test_aiotools.py b/testenv/tests/test_aiotools.py index 9ccaee0a..1ceae318 100644 --- a/testenv/tests/test_aiotools.py +++ b/testenv/tests/test_aiotools.py @@ -22,8 +22,6 @@ import asyncio -from typing import List - import pytest from kvmd.aiotools import AioExclusiveRegion @@ -123,7 +121,7 @@ async def test_fail__region__access_two() -> None: # ===== @pytest.mark.asyncio async def test_ok__shield_fg() -> None: - ops: List[str] = [] + ops: list[str] = [] async def foo(op: str, delay: float) -> None: # pylint: disable=disallowed-name await asyncio.sleep(delay) diff --git a/testenv/tests/validators/test_auth.py b/testenv/tests/validators/test_auth.py index ad56bbe0..789d8ece 100644 --- a/testenv/tests/validators/test_auth.py +++ b/testenv/tests/validators/test_auth.py @@ -20,7 +20,6 @@ # ========================================================================== # -from typing import List from typing import Any import pytest @@ -72,7 +71,7 @@ def test_fail__valid_user(arg: Any) -> None: (", foo, ", ["foo"]), ([], []), ]) -def test_ok__valid_users_list(arg: Any, retval: List) -> None: +def test_ok__valid_users_list(arg: Any, retval: list) -> None: assert valid_users_list(arg) == retval diff --git a/testenv/tests/validators/test_basic.py b/testenv/tests/validators/test_basic.py index c0e7b9ba..6303ed96 100644 --- a/testenv/tests/validators/test_basic.py +++ b/testenv/tests/validators/test_basic.py @@ -20,7 +20,6 @@ # ========================================================================== # -from typing import List from typing import Any import pytest @@ -151,7 +150,7 @@ def test_fail__valid_float_f01(arg: Any) -> None: (", a, ", ["a"]), ([], []), ]) -def test_ok__valid_string_list(arg: Any, retval: List) -> None: +def test_ok__valid_string_list(arg: Any, retval: list) -> None: assert valid_string_list(arg) == retval @@ -165,7 +164,7 @@ def test_ok__valid_string_list(arg: Any, retval: List) -> None: (", 1, ", [1]), ([], []), ]) -def test_ok__valid_string_list__subval(arg: Any, retval: List) -> None: # pylint: disable=invalid-name +def test_ok__valid_string_list__subval(arg: Any, retval: list) -> None: # pylint: disable=invalid-name assert valid_string_list(arg, subval=int) == retval diff --git a/testenv/tests/validators/test_net.py b/testenv/tests/validators/test_net.py index 288f4799..8067322e 100644 --- a/testenv/tests/validators/test_net.py +++ b/testenv/tests/validators/test_net.py @@ -20,7 +20,6 @@ # ========================================================================== # -from typing import List from typing import Any import pytest @@ -168,7 +167,7 @@ def test_fail__valid_port(arg: Any) -> None: ("80,443,443,", [80, 443, 443]), (65535, [65535]), ]) -def test_ok__valid_ports_list(arg: Any, retval: List[int]) -> None: +def test_ok__valid_ports_list(arg: Any, retval: list[int]) -> None: assert valid_ports_list(arg) == retval diff --git a/testenv/tests/validators/test_os.py b/testenv/tests/validators/test_os.py index 4cdd5aa4..cabb8e51 100644 --- a/testenv/tests/validators/test_os.py +++ b/testenv/tests/validators/test_os.py @@ -22,7 +22,6 @@ import os -from typing import List from typing import Any import pytest @@ -148,7 +147,7 @@ def test_fail__valid_unix_mode(arg: Any) -> None: ("/bin/true, 1, 2, 3,", ["/bin/true", "1", "2", "3"]), ("/bin/true", ["/bin/true"]), ]) -def test_ok__valid_command(arg: Any, retval: List[str]) -> None: +def test_ok__valid_command(arg: Any, retval: list[str]) -> None: assert valid_command(arg) == retval |