summaryrefslogtreecommitdiff
path: root/kvmd/plugins/msd/otg
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/msd/otg
parent4b75221e9470b4a009955d7677f16adf8e23e302 (diff)
new typing style
Diffstat (limited to 'kvmd/plugins/msd/otg')
-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
3 files changed, 23 insertions, 30 deletions
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: