Used isort to sort imports. Also added to the git pre-commit hook.

[python_utils.git] / camera_utils.py

diff --git a/camera_utils.py b/camera_utils.py
index acf760d55ca8b61d69871260ac55e18844f46acb..9e7efd6dfccbd7df2d1eb8bd13eb8075a6bfe4f1 100644 (file)
--- a/camera_utils.py
+++ b/camera_utils.py
@@ -5,6 +5,7 @@
 import logging
 import platform
 import subprocess
+import warnings
 from typing import NamedTuple, Optional
 
 import cv2  # type: ignore
@@ -19,6 +20,7 @@ logger = logging.getLogger(__name__)
 
 class RawJpgHsv(NamedTuple):
     """Raw image bytes, the jpeg image and the HSV (hue saturation value) image."""
+
     raw: Optional[bytes]
     jpg: Optional[np.ndarray]
     hsv: Optional[np.ndarray]
@@ -26,12 +28,14 @@ class RawJpgHsv(NamedTuple):
 
 class SanityCheckImageMetadata(NamedTuple):
     """Is a Blue Iris image bad (big grey borders around it) or infrared?"""
+
     is_bad_image: bool
     is_infrared_image: bool
 
 
 def sanity_check_image(hsv: np.ndarray) -> SanityCheckImageMetadata:
-    """See if a Blue Iris image is bad and infrared."""
+    """See if a Blue Iris or Shinobi image is bad and infrared."""
+
     def is_near(a, b) -> bool:
         return abs(a - b) < 3
 
@@ -43,16 +47,17 @@ def sanity_check_image(hsv: np.ndarray) -> SanityCheckImageMetadata:
         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)
+                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)):
+            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}")
     return SanityCheckImageMetadata(
-        hs_zero_count > (num_pixels * 0.75), weird_orange_count > (num_pixels * 0.75)
+        hs_zero_count > (num_pixels * 0.75),
+        weird_orange_count > (num_pixels * 0.75),
     )
 
 
@@ -64,28 +69,45 @@ def fetch_camera_image_from_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}')
     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')
             tmp = np.frombuffer(raw, dtype="uint8")
-            logger.debug(f'Translated raw content into {tmp.shape} {type(tmp)} with element type {type(tmp[0])}.')
+            logger.debug(
+                f'Translated raw content into {tmp.shape} {type(tmp)} with element type {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])}')
+            logger.debug(
+                f'Decoded into {jpg.shape} jpeg {type(jpg)} with element type {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])}')
+            logger.debug(
+                f'Converted JPG into HSV {hsv.shape} HSV {type(hsv)} with element type {type(hsv[0][0])}'
+            )
             (_, is_bad_image) = sanity_check_image(hsv)
             if not is_bad_image:
-                logger.debug(f"Got a good image from {url}")
                 return raw
     except Exception as e:
         logger.exception(e)
-    logger.warning(f"Got a bad image or HTTP error from {url}")
+    msg = f"Got a bad image or HTTP error from {url}; returning None."
+    logger.warning(msg)
+    warnings.warn(msg, stacklevel=2)
     return None
 
 
-def blue_iris_camera_name_to_hostname(camera_name: str) -> str:
+def camera_name_to_hostname(camera_name: str) -> str:
+    """Map a camera name to a hostname
+
+    >>> camera_name_to_hostname('driveway')
+    'driveway.house'
+
+    >>> camera_name_to_hostname('cabin_driveway')
+    'driveway.cabin'
+
+    """
     mapping = {
         "driveway": "driveway.house",
         "backyard": "backyard.house",
@@ -105,7 +127,9 @@ 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 = blue_iris_camera_name_to_hostname(camera_name)
+    hostname = camera_name_to_hostname(camera_name)
+    stream = f"rtsp://camera:IaLaIok@{hostname}:554/live"
+    logger.debug(f'Fetching image from RTSP stream {stream}')
     try:
         cmd = [
             "/usr/bin/timeout",
@@ -114,7 +138,7 @@ def fetch_camera_image_from_rtsp_stream(
             "/usr/local/bin/ffmpeg",
             "-y",
             "-i",
-            f"rtsp://camera:IaLaIok@{hostname}:554/live",
+            f"{stream}",
             "-f",
             "singlejpeg",
             "-vframes",
@@ -130,7 +154,9 @@ def fetch_camera_image_from_rtsp_stream(
             return out
     except Exception as e:
         logger.exception(e)
-    logger.warning("Failed to retrieve image from RTSP stream")
+    msg = f"Failed to retrieve image via RTSP {stream}, returning None."
+    logger.warning(msg)
+    warnings.warn(msg, stacklevel=2)
     return None
 
 
@@ -144,9 +170,7 @@ def _fetch_camera_image(
         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")
@@ -157,9 +181,9 @@ def _fetch_camera_image(
             jpg=jpg,
             hsv=hsv,
         )
-    logger.warning(
-        "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)
 
 
@@ -169,4 +193,11 @@ def fetch_camera_image(
     try:
         return _fetch_camera_image(camera_name, width=width, quality=quality)
     except exceptions.TimeoutError:
+        logger.warning('Fetching camera image operation timed out.')
         return RawJpgHsv(None, None, None)
+
+
+if __name__ == '__main__':
+    import doctest
+
+    doctest.testmod()