new typing style

8 files changed, 59 insertions, 76 deletions

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