More cleanup, yey!

[python_utils.git] / camera_utils.py

diff --git a/camera_utils.py b/camera_utils.py
index 03ac621a52eb4d855ec3ac5458f59712e7eea657..c789ed61e19d304391c9d10393bcb22094b0b7d7 100644 (file)
--- a/camera_utils.py
+++ b/camera_utils.py
@@ -5,8 +5,8 @@
 import logging
 import platform
 import subprocess
-from typing import NamedTuple, Optional
 import warnings
+from typing import NamedTuple, Optional
 
 import cv2  # type: ignore
 import numpy as np
@@ -46,15 +46,11 @@ def sanity_check_image(hsv: np.ndarray) -> SanityCheckImageMetadata:
     for r in range(rows):
         for c in range(cols):
             pixel = hsv[(r, c)]
-            if (
-                is_near(pixel[0], 16)
-                and is_near(pixel[1], 117)
-                and is_near(pixel[2], 196)
-            ):
+            if is_near(pixel[0], 16) and is_near(pixel[1], 117) and is_near(pixel[2], 196):
                 weird_orange_count += 1
             elif is_near(pixel[0], 0) and is_near(pixel[1], 0):
                 hs_zero_count += 1
-    logger.debug(f"hszero#={hs_zero_count}, weird_orange={weird_orange_count}")
+    logger.debug("hszero#=%d, weird_orange=%d", hs_zero_count, weird_orange_count)
     return SanityCheckImageMetadata(
         hs_zero_count > (num_pixels * 0.75),
         weird_orange_count > (num_pixels * 0.75),
@@ -68,26 +64,29 @@ def fetch_camera_image_from_video_server(
     """Fetch the raw webcam image from the video server."""
     camera_name = camera_name.replace(".house", "")
     camera_name = camera_name.replace(".cabin", "")
-    url = f"http://10.0.0.226:8080/Umtxxf1uKMBniFblqeQ9KRbb6DDzN4/jpeg/GKlT2FfiSQ/{camera_name}/s.jpg"
-    logger.debug(f'Fetching image from {url}')
+    url = (
+        f"http://10.0.0.226:8080/Umtxxf1uKMBniFblqeQ9KRbb6DDzN4/jpeg/GKlT2FfiSQ/{camera_name}/s.jpg"
+    )
+    logger.debug('Fetching image from %s', url)
     try:
         response = requests.get(url, stream=False, timeout=10.0)
         if response.ok:
             raw = response.content
-            logger.debug(
-                f'Read {len(response.content)} byte image from HTTP server'
-            )
+            logger.debug('Read %d byte image from HTTP server', len(response.content))
             tmp = np.frombuffer(raw, dtype="uint8")
             logger.debug(
-                f'Translated raw content into {tmp.shape} {type(tmp)} with element type {type(tmp[0])}.'
+                'Translated raw content into %s %s with element type %s',
+                tmp.shape, type(tmp), type(tmp[0]),
             )
             jpg = cv2.imdecode(tmp, cv2.IMREAD_COLOR)
             logger.debug(
-                f'Decoded into {jpg.shape} jpeg {type(jpg)} with element type {type(jpg[0][0])}'
+                'Decoded into %s jpeg %s with element type %s',
+                jpg.shape, type(jpg), type(jpg[0][0])
             )
             hsv = cv2.cvtColor(jpg, cv2.COLOR_BGR2HSV)
             logger.debug(
-                f'Converted JPG into HSV {hsv.shape} HSV {type(hsv)} with element type {type(hsv[0][0])}'
+                'Converted JPG into %s HSV HSV %s with element type %s',
+                hsv.shape, type(hsv), type(hsv[0][0])
             )
             (_, is_bad_image) = sanity_check_image(hsv)
             if not is_bad_image:
@@ -125,13 +124,11 @@ def camera_name_to_hostname(camera_name: str) -> str:
 
 
 @decorator_utils.retry_if_none(tries=2, delay_sec=1, backoff=1.1)
-def fetch_camera_image_from_rtsp_stream(
-    camera_name: str, *, width: int = 256
-) -> Optional[bytes]:
+def fetch_camera_image_from_rtsp_stream(camera_name: str, *, width: int = 256) -> Optional[bytes]:
     """Fetch the raw webcam image straight from the webcam's RTSP stream."""
     hostname = camera_name_to_hostname(camera_name)
     stream = f"rtsp://camera:IaLaIok@{hostname}:554/live"
-    logger.debug(f'Fetching image from RTSP stream {stream}')
+    logger.debug('Fetching image from RTSP stream %s', stream)
     try:
         cmd = [
             "/usr/bin/timeout",
@@ -149,9 +146,7 @@ def fetch_camera_image_from_rtsp_stream(
             f"scale={width}:-1",
             "-",
         ]
-        with subprocess.Popen(
-            cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL
-        ) as proc:
+        with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) as proc:
             out, _ = proc.communicate(timeout=10)
             return out
     except Exception as e:
@@ -163,18 +158,12 @@ def fetch_camera_image_from_rtsp_stream(
 
 
 @decorator_utils.timeout(seconds=30, use_signals=False)
-def _fetch_camera_image(
-    camera_name: str, *, width: int = 256, quality: int = 70
-) -> RawJpgHsv:
+def _fetch_camera_image(camera_name: str, *, width: int = 256, quality: int = 70) -> RawJpgHsv:
     """Fetch a webcam image given the camera name."""
     logger.debug("Trying to fetch camera image from video server")
-    raw = fetch_camera_image_from_video_server(
-        camera_name, width=width, quality=quality
-    )
+    raw = fetch_camera_image_from_video_server(camera_name, width=width, quality=quality)
     if raw is None:
-        logger.debug(
-            "Reading from video server failed; trying direct RTSP stream"
-        )
+        logger.debug("Reading from video server failed; trying direct RTSP stream")
         raw = fetch_camera_image_from_rtsp_stream(camera_name, width=width)
     if raw is not None and len(raw) > 0:
         tmp = np.frombuffer(raw, dtype="uint8")
@@ -185,17 +174,13 @@ def _fetch_camera_image(
             jpg=jpg,
             hsv=hsv,
         )
-    msg = (
-        "Failed to retieve image from both video server and direct RTSP stream"
-    )
+    msg = "Failed to retieve image from both video server and direct RTSP stream"
     logger.warning(msg)
     warnings.warn(msg, stacklevel=2)
     return RawJpgHsv(None, None, None)
 
 
-def fetch_camera_image(
-    camera_name: str, *, width: int = 256, quality: int = 70
-) -> RawJpgHsv:
+def fetch_camera_image(camera_name: str, *, width: int = 256, quality: int = 70) -> RawJpgHsv:
     try:
         return _fetch_camera_image(camera_name, width=width, quality=quality)
     except exceptions.TimeoutError: