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