summaryrefslogtreecommitdiff
path: root/kvmd/plugins/msd
diff options
context:
space:
mode:
Diffstat (limited to 'kvmd/plugins/msd')
-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
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
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-23 16:59:21 +0000