mypy again

kvmd/aiofs.py

@@ -29,7 +29,12 @@ from . import aiotools
 
 
 # =====
+async def read(path: str) -> str:
+    async with aiofiles.open(path) as afile:  # type: ignore
+        return (await afile.read())
+
+
 async def afile_write_now(afile: aiofiles.base.AiofilesContextManager, data: bytes) -> None:
-    await afile.write(data)
+    await afile.write(data)  # type: ignore
-    await afile.flush()
+    await afile.flush()  # type: ignore
-    await aiotools.run_async(os.fsync, afile.fileno())
+    await aiotools.run_async(os.fsync, afile.fileno())  # type: ignore

kvmd/apps/kvmd/info/hw.py

@@ -29,11 +29,10 @@ from typing import AsyncGenerator
 from typing import TypeVar
 from typing import Optional
 
-import aiofiles
-
 from ....logging import get_logger
 
 from .... import env
+from .... import aiofs
 from .... import aioproc
 
 from .base import BaseInfoSubmanager
@@ -89,8 +88,7 @@ class HwInfoSubmanager(BaseInfoSubmanager):
     async def __get_dt_model(self) -> Optional[str]:
         model_path = f"{env.PROCFS_PREFIX}/proc/device-tree/model"
         try:
-            async with aiofiles.open(model_path) as model_file:
+            return (await aiofs.read(model_path)).strip(" \t\r\n\0")
-                return (await model_file.read()).strip(" \t\r\n\0")
         except Exception as err:
             get_logger(0).error("Can't read DT model from %s: %s", model_path, err)
             return None
@@ -98,8 +96,7 @@ class HwInfoSubmanager(BaseInfoSubmanager):
     async def __get_cpu_temp(self) -> Optional[float]:
         temp_path = f"{env.SYSFS_PREFIX}/sys/class/thermal/thermal_zone0/temp"
         try:
-            async with aiofiles.open(temp_path) as temp_file:
+            return int((await aiofs.read(temp_path)).strip()) / 1000
-                return int((await temp_file.read()).strip()) / 1000
         except Exception as err:
             get_logger(0).error("Can't read CPU temp from %s: %s", temp_path, err)
             return None

kvmd/apps/vnc/vncauth.py

@@ -25,10 +25,10 @@ import dataclasses
 from typing import Tuple
 from typing import Dict
 
-import aiofiles
-
 from ...logging import get_logger
 
+from ... import aiofs
+
 
 # =====
 class VncAuthError(Exception):
@@ -64,8 +64,7 @@ class VncAuthManager:
         return ({}, (not self.__enabled))
 
     async def __inner_read_credentials(self) -> Dict[str, VncAuthKvmdCredentials]:
-        async with aiofiles.open(self.__path) as vc_file:
+        lines = (await aiofs.read(self.__path)).split("\n")
-            lines = (await vc_file.read()).split("\n")
 
         credentials: Dict[str, VncAuthKvmdCredentials] = {}
         for (lineno, line) in enumerate(lines):

kvmd/plugins/msd/otg/__init__.py

@@ -306,7 +306,7 @@ class Plugin(BaseMsd):  # pylint: disable=too-many-instance-attributes
                         await self.__remount_storage(rw=True)
                         self.__set_image_complete(name, False)
                         self.__new_file_written = 0
-                        self.__new_file = await aiofiles.open(path, mode="w+b", buffering=0)
+                        self.__new_file = await aiofiles.open(path, mode="w+b", buffering=0)  # type: ignore
 
                     await self.__notifier.notify()
                     yield
@@ -364,7 +364,7 @@ class Plugin(BaseMsd):  # pylint: disable=too-many-instance-attributes
         try:
             if self.__new_file:
                 get_logger().info("Closing new image file ...")
-                await self.__new_file.close()
+                await self.__new_file.close()  # type: ignore
         except Exception:
             get_logger().exception("Can't close device file")
         finally:

kvmd/plugins/msd/relay.py

@@ -355,7 +355,7 @@ class Plugin(BaseMsd):  # pylint: disable=too-many-instance-attributes
                     if self.__connected:
                         raise MsdConnectedError()
 
-                    self.__device_file = await aiofiles.open(self.__device_info.path, mode="w+b", buffering=0)
+                    self.__device_file = await aiofiles.open(self.__device_info.path, mode="w+b", buffering=0)  # type: ignore
                     self.__written = 0
 
                     await self.__write_image_info(name, complete=False)
@@ -391,9 +391,9 @@ class Plugin(BaseMsd):  # pylint: disable=too-many-instance-attributes
         assert self.__device_file
         assert self.__device_info
         if self.__device_info.size - self.__written > _IMAGE_INFO_SIZE:
-            await self.__device_file.seek(self.__device_info.size - _IMAGE_INFO_SIZE)
+            await self.__device_file.seek(self.__device_info.size - _IMAGE_INFO_SIZE)  # type: ignore
             await aiofs.afile_write_now(self.__device_file, _make_image_info_bytes(name, self.__written, complete))
-            await self.__device_file.seek(0)
+            await self.__device_file.seek(0)  # type: ignore
         else:
             get_logger().error("Can't write image info because device is full")
 
@@ -401,7 +401,7 @@ class Plugin(BaseMsd):  # pylint: disable=too-many-instance-attributes
         try:
             if self.__device_file:
                 get_logger().info("Closing device file ...")
-                await self.__device_file.close()
+                await self.__device_file.close()  # type: ignore
         except Exception:
             get_logger().exception("Can't close device file")
         finally: