summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
committerMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
commitee3e224e396494cd0d69bb6167087a071a20349c (patch)
tree5becd28570e58a03c6e1e231d0db24c264a73f88
parent4b75221e9470b4a009955d7677f16adf8e23e302 (diff)
new typing style
-rwxr-xr-xgenmap.py18
-rw-r--r--kvmd/aiogp.py12
-rw-r--r--kvmd/aiohelpers.py4
-rw-r--r--kvmd/aiomulti.py13
-rw-r--r--kvmd/aioproc.py19
-rw-r--r--kvmd/aiotools.py21
-rw-r--r--kvmd/apps/__init__.py28
-rw-r--r--kvmd/apps/cleanup/__init__.py5
-rw-r--r--kvmd/apps/edidconf/__init__.py4
-rw-r--r--kvmd/apps/htpasswd/__init__.py4
-rw-r--r--kvmd/apps/ipmi/__init__.py5
-rw-r--r--kvmd/apps/ipmi/auth.py7
-rw-r--r--kvmd/apps/ipmi/server.py21
-rw-r--r--kvmd/apps/janus/__init__.py5
-rw-r--r--kvmd/apps/janus/runner.py18
-rw-r--r--kvmd/apps/janus/stun.py20
-rw-r--r--kvmd/apps/kvmd/__init__.py5
-rw-r--r--kvmd/apps/kvmd/api/export.py5
-rw-r--r--kvmd/apps/kvmd/api/hid.py30
-rw-r--r--kvmd/apps/kvmd/api/info.py4
-rw-r--r--kvmd/apps/kvmd/api/msd.py13
-rw-r--r--kvmd/apps/kvmd/api/streamer.py9
-rw-r--r--kvmd/apps/kvmd/auth.py20
-rw-r--r--kvmd/apps/kvmd/info/__init__.py4
-rw-r--r--kvmd/apps/kvmd/info/auth.py4
-rw-r--r--kvmd/apps/kvmd/info/base.py6
-rw-r--r--kvmd/apps/kvmd/info/extras.py13
-rw-r--r--kvmd/apps/kvmd/info/fan.py12
-rw-r--r--kvmd/apps/kvmd/info/hw.py21
-rw-r--r--kvmd/apps/kvmd/info/meta.py5
-rw-r--r--kvmd/apps/kvmd/info/system.py11
-rw-r--r--kvmd/apps/kvmd/logreader.py5
-rw-r--r--kvmd/apps/kvmd/streamer.py54
-rw-r--r--kvmd/apps/kvmd/sysunit.py14
-rw-r--r--kvmd/apps/kvmd/tesseract.py19
-rw-r--r--kvmd/apps/kvmd/ugpio.py35
-rw-r--r--kvmd/apps/otg/__init__.py8
-rw-r--r--kvmd/apps/otg/hid/keyboard.py4
-rw-r--r--kvmd/apps/otg/hid/mouse.py8
-rw-r--r--kvmd/apps/otgconf/__init__.py7
-rw-r--r--kvmd/apps/otgmsd/__init__.py5
-rw-r--r--kvmd/apps/otgnet/__init__.py25
-rw-r--r--kvmd/apps/otgnet/netctl.py42
-rw-r--r--kvmd/apps/pst/__init__.py5
-rw-r--r--kvmd/apps/pst/server.py7
-rw-r--r--kvmd/apps/pstrun/__init__.py17
-rw-r--r--kvmd/apps/vnc/__init__.py9
-rw-r--r--kvmd/apps/vnc/rfb/__init__.py11
-rw-r--r--kvmd/apps/vnc/rfb/crypto.py4
-rw-r--r--kvmd/apps/vnc/rfb/encodings.py14
-rw-r--r--kvmd/apps/vnc/rfb/stream.py7
-rw-r--r--kvmd/apps/vnc/server.py35
-rw-r--r--kvmd/apps/vnc/vncauth.py9
-rw-r--r--kvmd/apps/watchdog/__init__.py5
-rw-r--r--kvmd/clients/kvmd.py27
-rw-r--r--kvmd/clients/streamer.py11
-rw-r--r--kvmd/helpers/remount/__init__.py4
-rw-r--r--kvmd/htclient.py6
-rw-r--r--kvmd/htserver.py37
-rw-r--r--kvmd/inotify.py23
-rw-r--r--kvmd/keyboard/keysym.py15
-rw-r--r--kvmd/keyboard/mappings.py.mako4
-rw-r--r--kvmd/keyboard/printer.py6
-rw-r--r--kvmd/logging.py4
-rw-r--r--kvmd/plugins/__init__.py6
-rw-r--r--kvmd/plugins/atx/__init__.py8
-rw-r--r--kvmd/plugins/atx/disabled.py5
-rw-r--r--kvmd/plugins/atx/gpio.py16
-rw-r--r--kvmd/plugins/auth/__init__.py4
-rw-r--r--kvmd/plugins/auth/htpasswd.py4
-rw-r--r--kvmd/plugins/auth/http.py9
-rw-r--r--kvmd/plugins/auth/pam.py9
-rw-r--r--kvmd/plugins/auth/radius.py4
-rw-r--r--kvmd/plugins/hid/__init__.py14
-rw-r--r--kvmd/plugins/hid/_mcu/__init__.py24
-rw-r--r--kvmd/plugins/hid/_mcu/gpio.py9
-rw-r--r--kvmd/plugins/hid/bt/__init__.py17
-rw-r--r--kvmd/plugins/hid/bt/bluez.py6
-rw-r--r--kvmd/plugins/hid/bt/server.py18
-rw-r--r--kvmd/plugins/hid/otg/__init__.py25
-rw-r--r--kvmd/plugins/hid/otg/device.py8
-rw-r--r--kvmd/plugins/hid/otg/events.py13
-rw-r--r--kvmd/plugins/hid/otg/keyboard.py10
-rw-r--r--kvmd/plugins/hid/otg/mouse.py5
-rw-r--r--kvmd/plugins/hid/serial.py7
-rw-r--r--kvmd/plugins/hid/spi.py13
-rw-r--r--kvmd/plugins/msd/__init__.py29
-rw-r--r--kvmd/plugins/msd/disabled.py14
-rw-r--r--kvmd/plugins/msd/otg/__init__.py45
-rw-r--r--kvmd/plugins/msd/otg/drive.py4
-rw-r--r--kvmd/plugins/msd/otg/fs.py4
-rw-r--r--kvmd/plugins/msd/relay/__init__.py26
-rw-r--r--kvmd/plugins/msd/relay/drive.py5
-rw-r--r--kvmd/plugins/msd/relay/gpio.py8
-rw-r--r--kvmd/plugins/ugpio/__init__.py9
-rw-r--r--kvmd/plugins/ugpio/cmd.py9
-rw-r--r--kvmd/plugins/ugpio/cmdret.py9
-rw-r--r--kvmd/plugins/ugpio/ezcoo.py15
-rw-r--r--kvmd/plugins/ugpio/gpio.py16
-rw-r--r--kvmd/plugins/ugpio/hidrelay.py13
-rw-r--r--kvmd/plugins/ugpio/hue.py16
-rw-r--r--kvmd/plugins/ugpio/ipmi.py11
-rw-r--r--kvmd/plugins/ugpio/locator.py15
-rw-r--r--kvmd/plugins/ugpio/pway.py15
-rw-r--r--kvmd/plugins/ugpio/pwm.py13
-rw-r--r--kvmd/plugins/ugpio/servo.py4
-rw-r--r--kvmd/plugins/ugpio/tesmart.py8
-rw-r--r--kvmd/plugins/ugpio/wol.py6
-rw-r--r--kvmd/plugins/ugpio/xh_hk4401.py15
-rw-r--r--kvmd/tools.py15
-rw-r--r--kvmd/validators/__init__.py9
-rw-r--r--kvmd/validators/auth.py3
-rw-r--r--kvmd/validators/basic.py15
-rw-r--r--kvmd/validators/kvm.py3
-rw-r--r--kvmd/validators/net.py11
-rw-r--r--kvmd/validators/os.py5
-rw-r--r--kvmd/validators/ugpio.py9
-rw-r--r--kvmd/yamlconf/__init__.py20
-rw-r--r--kvmd/yamlconf/loader.py8
-rw-r--r--testenv/linters/mypy.ini2
-rw-r--r--testenv/tests/apps/htpasswd/test_main.py3
-rw-r--r--testenv/tests/apps/kvmd/test_auth.py7
-rw-r--r--testenv/tests/plugins/auth/test_http.py3
-rw-r--r--testenv/tests/plugins/auth/test_pam.py8
-rw-r--r--testenv/tests/test_aiotools.py4
-rw-r--r--testenv/tests/validators/test_auth.py3
-rw-r--r--testenv/tests/validators/test_basic.py5
-rw-r--r--testenv/tests/validators/test_net.py3
-rw-r--r--testenv/tests/validators/test_os.py3
129 files changed, 593 insertions, 941 deletions
diff --git a/genmap.py b/genmap.py
index da512d63..5e827725 100755
--- a/genmap.py
+++ b/genmap.py
@@ -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