Use OpenCV instead of Pillow for Base64 Encoding (#410)

* Using `cv2.imencode()`. (#408)

* Merged unnecessary encoding. (#408)

* Added param `quality`. (#408)

* Supported uncompressed image for comm stream. (#408)

leads_gui/prototype.py

@@ -1,12 +1,10 @@
 from abc import ABCMeta as _ABCMeta, abstractmethod as _abstractmethod
-from io import BytesIO as _BytesIO
 from json import dumps as _dumps
 from time import time as _time
 from tkinter import Misc as _Misc, Event as _Event, PhotoImage as _PhotoImage
 from typing import Callable as _Callable, Self as _Self, TypeVar as _TypeVar, Generic as _Generic, Any as _Any, \
     Literal as _Literal, override as _override
 
-from PIL.Image import Image as _Image
 from customtkinter import CTk as _CTk, CTkCanvas as _CTkCanvas, get_appearance_mode as _get_appearance_mode, \
     ThemeManager as _ThemeManager, Variable as _Variable, ScalingTracker as _ScalingTracker, \
     set_appearance_mode as _set_appearance_mode, CTkToplevel as _CTkToplevel
@@ -231,11 +229,9 @@ def comm_notify(self, d: _DataContainer | dict[str, _Any]) -> None:
         if self.comm:
             self.comm.broadcast(d.encode() if isinstance(d, _DataContainer) else _dumps(d).encode())
 
-    def comm_stream_notify(self, tag: _Literal["frvc", "lfvc", "rtvc", "revc"], frame: _Image,
-                           quality: int = 90) -> None:
+    def comm_stream_notify(self, tag: _Literal["frvc", "lfvc", "rtvc", "revc"], frame: bytes) -> None:
         if self.comm_stream:
-            frame.save(buffer := _BytesIO(), "JPEG", quality=quality)
-            self.comm_stream.broadcast(tag.encode() + b":" + buffer.getvalue())
+            self.comm_stream.broadcast(tag.encode() + b":" + frame)
 
 
 _runtime_data_singleton_flag: bool = False

leads_vec/cli.py

@@ -1,3 +1,4 @@
+from base64 import b64encode
 from datetime import datetime as _datetime
 from threading import Thread as _Thread
 from time import time as _time, sleep as _sleep
@@ -18,7 +19,7 @@
     Typography, Speedometer, ProxyCanvas, SpeedTrendMeter, GForceMeter, Stopwatch, Hazard, initialize, Battery, Brake, \
     ESC, Satellite, Motor, Speed, Photo, Light, ImageVariable
 from leads_vec.__version__ import __version__
-from leads_video import get_camera
+from leads_video import get_camera, Base64Camera
 
 
 class CustomRuntimeData(RuntimeData):
@@ -72,8 +73,8 @@ def _() -> None:
             if rd.comm_stream.num_connections() < 1:
                 _sleep(.01)
             for tag in FRONT_VIEW_CAMERA, LEFT_VIEW_CAMERA, RIGHT_VIEW_CAMERA, REAR_VIEW_CAMERA:
-                if (cam := get_camera(tag)) and (frame := cam.read_pil()):
-                    rd.comm_stream_notify(tag, frame)
+                if (cam := get_camera(tag)) and (frame := cam.read_numpy()) is not None:
+                    rd.comm_stream_notify(tag, b64encode(Base64Camera.encode(frame)))
 
     _Thread(name="comm streamer", target=_, daemon=True).start()
 

leads_vec/devices_visual.py

@@ -14,19 +14,19 @@
 config: Config = require_config()
 CAMERA_RESOLUTION: tuple[int, int] | None = config.get("camera_resolution")
 CAMERA_TAGS: list[str] = []
-CAMERA_ARGS: list[tuple[int, tuple[int, int] | None]] = []
+CAMERA_ARGS: list[tuple[int, tuple[int, int] | None, int]] = []
 if (port := config.get("front_view_camera_port")) is not None:
     CAMERA_TAGS.append(FRONT_VIEW_CAMERA)
-    CAMERA_ARGS.append((port, CAMERA_RESOLUTION))
+    CAMERA_ARGS.append((port, CAMERA_RESOLUTION, 25))
 if (port := config.get("left_view_camera_port")) is not None:
     CAMERA_TAGS.append(LEFT_VIEW_CAMERA)
-    CAMERA_ARGS.append((port, CAMERA_RESOLUTION))
+    CAMERA_ARGS.append((port, CAMERA_RESOLUTION, 25))
 if (port := config.get("right_view_camera_port")) is not None:
     CAMERA_TAGS.append(RIGHT_VIEW_CAMERA)
-    CAMERA_ARGS.append((port, CAMERA_RESOLUTION))
+    CAMERA_ARGS.append((port, CAMERA_RESOLUTION, 25))
 if (port := config.get("rear_view_camera_port")) is not None:
     CAMERA_TAGS.append(REAR_VIEW_CAMERA)
-    CAMERA_ARGS.append((port, CAMERA_RESOLUTION))
+    CAMERA_ARGS.append((port, CAMERA_RESOLUTION, 25))
 
 
 @device(CAMERA_TAGS, MAIN_CONTROLLER, CAMERA_ARGS)

leads_video/camera.py

@@ -4,8 +4,9 @@
 from time import time as _time, sleep as _sleep
 from typing import override as _override
 
-from PIL.Image import fromarray as _fromarray, Image as _Image
-from cv2 import VideoCapture as _VideoCapture, cvtColor as _cvtColor, COLOR_BGR2RGB as _COLOR_BGR2RGB
+from PIL.Image import fromarray as _fromarray, Image as _Image, open as _open
+from cv2 import VideoCapture as _VideoCapture, cvtColor as _cvtColor, COLOR_BGR2RGB as _COLOR_BGR2RGB, \
+    imencode as _imencode, COLOR_RGB2BGR as _COLOR_RGB2BGR, IMWRITE_JPEG_QUALITY as _IMWRITE_JPEG_QUALITY
 from numpy import ndarray as _ndarray, pad as _pad, array as _array
 
 from leads import Device as _Device, ShadowDevice as _ShadowDevice
@@ -83,21 +84,25 @@ def read(self) -> _ndarray | None:
 
 
 class Base64Camera(LowLatencyCamera):
-    def __init__(self, port: int, resolution: tuple[int, int] | None = None) -> None:
+    def __init__(self, port: int, resolution: tuple[int, int] | None = None, quality: int = 90) -> None:
         super().__init__(port, resolution)
+        self._quality: int = quality
         self._shadow_thread2: _Thread | None = None
-        self._pil: _Image | None = None
+        self._bytes: bytes = b""
         self._base64: str = ""
 
     @_override
     def loop(self) -> None:
         super().loop()
 
+    @staticmethod
+    def encode(frame: _ndarray, quality: int = 100) -> bytes:
+        return _imencode(".jpg", _cvtColor(frame, _COLOR_RGB2BGR), (_IMWRITE_JPEG_QUALITY, quality))[1].tobytes()
+
     def loop2(self) -> None:
-        if (local_frame := self._frame) is not None:
-            self._pil = _fromarray(local_frame)
-            self._pil.save(buffer := _BytesIO(), "JPEG", quality=25)
-            self._base64 = _b64encode(buffer.getvalue()).decode()
+        if (frame := self._frame) is not None:
+            self._bytes = Base64Camera.encode(frame, self._quality)
+            self._base64 = _b64encode(self._bytes).decode()
 
     def run2(self) -> None:
         while True:
@@ -120,7 +125,7 @@ def read_numpy(self) -> _ndarray | None:
 
     @_override
     def read_pil(self) -> _Image | None:
-        return self._pil
+        return _open(_BytesIO(self._bytes)) if self._bytes else None
 
 
 class LightweightBase64Camera(Base64Camera):