refactoring

kvmd/apps/kvmd/api/streamer.py

@@ -20,17 +20,12 @@
 # ========================================================================== #
 
 
-import io
-import functools
-
 from typing import List
 from typing import Dict
 
 from aiohttp.web import Request
 from aiohttp.web import Response
 
-from PIL import Image as PilImage
-
 from ....validators import check_string_in_list
 from ....validators.basic import valid_bool
 from ....validators.basic import valid_number
@@ -38,13 +33,10 @@ from ....validators.basic import valid_int_f0
 from ....validators.basic import valid_string_list
 from ....validators.kvm import valid_stream_quality
 
-from .... import aiotools
-
 from ..http import UnavailableError
 from ..http import exposed_http
 from ..http import make_json_response
 
-from ..streamer import StreamerSnapshot
 from ..streamer import Streamer
 
 from ..tesseract import TesseractOcr
@@ -89,8 +81,7 @@ class StreamerApi:
                     content_type="text/plain",
                 )
             elif valid_bool(request.query.get("preview", "false")):
-                data = await self.__make_preview(
-                    snapshot=snapshot,
+                data = await snapshot.make_preview(
                     max_width=valid_int_f0(request.query.get("preview_max_width", "0")),
                     max_height=valid_int_f0(request.query.get("preview_max_height", "0")),
                     quality=valid_stream_quality(request.query.get("preview_quality", "80")),
@@ -131,30 +122,3 @@ class StreamerApi:
     @exposed_http("GET", "/streamer/ocr")
     async def __ocr_handler(self, _: Request) -> Response:
         return make_json_response(await self.get_ocr())
-
-    # =====
-
-    async def __make_preview(self, snapshot: StreamerSnapshot, max_width: int, max_height: int, quality: int) -> bytes:
-        if max_width == 0 and max_height == 0:
-            max_width = snapshot.width // 5
-            max_height = snapshot.height // 5
-        else:
-            max_width = min((max_width or snapshot.width), snapshot.width)
-            max_height = min((max_height or snapshot.height), snapshot.height)
-
-        if max_width == snapshot.width and max_height == snapshot.height:
-            return snapshot.data
-        else:
-            return (await aiotools.run_async(self.__inner_make_preview, snapshot, max_width, max_height, quality))
-
-    @functools.lru_cache(maxsize=1)
-    def __inner_make_preview(self, snapshot: StreamerSnapshot, max_width: int, max_height: int, quality: int) -> bytes:
-        assert 0 < max_width <= snapshot.width
-        assert 0 < max_height <= snapshot.height
-        assert not (max_width == snapshot.width and max_height == snapshot.height)
-        with io.BytesIO(snapshot.data) as snapshot_bio:
-            with io.BytesIO() as preview_bio:
-                with PilImage.open(snapshot_bio) as image:
-                    image.thumbnail((max_width, max_height), PilImage.ANTIALIAS)
-                    image.save(preview_bio, format="jpeg", quality=quality)
-                    return preview_bio.getvalue()

kvmd/apps/kvmd/streamer.py

@@ -20,10 +20,12 @@
 # ========================================================================== #
 
 
+import io
 import signal
 import asyncio
 import asyncio.subprocess
 import dataclasses
+import functools
 
 from typing import Tuple
 from typing import List
@@ -34,6 +36,8 @@ from typing import Any
 
 import aiohttp
 
+from PIL import Image as PilImage
+
 from ...logging import get_logger
 
 from ... import tools
@@ -52,6 +56,31 @@ class StreamerSnapshot:
     headers: Tuple[Tuple[str, str], ...]
     data: bytes
 
+    async def make_preview(self, max_width: int, max_height: int, quality: int) -> bytes:
+        assert max_width >= 0
+        assert max_height >= 0
+        assert quality > 0
+
+        if max_width == 0 and max_height == 0:
+            max_width = self.width // 5
+            max_height = self.height // 5
+        else:
+            max_width = min((max_width or self.width), self.width)
+            max_height = min((max_height or self.height), self.height)
+
+        if (max_width, max_height) == (self.width, self.height):
+            return self.data
+        return (await aiotools.run_async(self.__inner_make_preview, max_width, max_height, quality))
+
+    @functools.lru_cache(maxsize=1)
+    def __inner_make_preview(self, max_width: int, max_height: int, quality: int) -> bytes:
+        with io.BytesIO(self.data) as snapshot_bio:
+            with io.BytesIO() as preview_bio:
+                with PilImage.open(snapshot_bio) as image:
+                    image.thumbnail((max_width, max_height), PilImage.ANTIALIAS)
+                    image.save(preview_bio, format="jpeg", quality=quality)
+                    return preview_bio.getvalue()
+
 
 class _StreamerParams:
     __DESIRED_FPS = "desired_fps"