Bugfixes in math_utils.
[python_utils.git] / camera_utils.py
index c789ed61e19d304391c9d10393bcb22094b0b7d7..c2bd04da0adccdb439711e3df6fdacf700d92e84 100644 (file)
@@ -1,12 +1,15 @@
 #!/usr/bin/env python3
 
+# © Copyright 2021-2022, Scott Gasch
+
 """Utilities for dealing with webcam images."""
 
 import logging
 import platform
 import subprocess
 import warnings
-from typing import NamedTuple, Optional
+from dataclasses import dataclass
+from typing import Optional
 
 import cv2  # type: ignore
 import numpy as np
@@ -14,23 +17,26 @@ import requests
 
 import decorator_utils
 import exceptions
+import scott_secrets
 
 logger = logging.getLogger(__name__)
 
 
-class RawJpgHsv(NamedTuple):
+@dataclass
+class RawJpgHsv:
     """Raw image bytes, the jpeg image and the HSV (hue saturation value) image."""
 
-    raw: Optional[bytes]
-    jpg: Optional[np.ndarray]
-    hsv: Optional[np.ndarray]
+    raw: Optional[bytes] = None
+    jpg: Optional[np.ndarray] = None
+    hsv: Optional[np.ndarray] = None
 
 
-class SanityCheckImageMetadata(NamedTuple):
+@dataclass
+class SanityCheckImageMetadata:
     """Is a Blue Iris image bad (big grey borders around it) or infrared?"""
 
-    is_bad_image: bool
-    is_infrared_image: bool
+    is_infrared_image: bool = False
+    is_bad_image: bool = False
 
 
 def sanity_check_image(hsv: np.ndarray) -> SanityCheckImageMetadata:
@@ -62,11 +68,10 @@ def fetch_camera_image_from_video_server(
     camera_name: str, *, width: int = 256, quality: int = 70
 ) -> Optional[bytes]:
     """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"
-    )
+    url = f"http://10.0.0.226:8080/{scott_secrets.SHINOBI_KEY1}/jpeg/{scott_secrets.SHINOBI_KEY2}/{camera_name}/s.jpg"
     logger.debug('Fetching image from %s', url)
     try:
         response = requests.get(url, stream=False, timeout=10.0)
@@ -76,20 +81,26 @@ def fetch_camera_image_from_video_server(
             tmp = np.frombuffer(raw, dtype="uint8")
             logger.debug(
                 'Translated raw content into %s %s with element type %s',
-                tmp.shape, type(tmp), type(tmp[0]),
+                tmp.shape,
+                type(tmp),
+                type(tmp[0]),
             )
             jpg = cv2.imdecode(tmp, cv2.IMREAD_COLOR)
             logger.debug(
                 'Decoded into %s jpeg %s with element type %s',
-                jpg.shape, type(jpg), type(jpg[0][0])
+                jpg.shape,
+                type(jpg),
+                type(jpg[0][0]),
             )
             hsv = cv2.cvtColor(jpg, cv2.COLOR_BGR2HSV)
             logger.debug(
                 'Converted JPG into %s HSV HSV %s with element type %s',
-                hsv.shape, type(hsv), type(hsv[0][0])
+                hsv.shape,
+                type(hsv),
+                type(hsv[0][0]),
             )
-            (_, is_bad_image) = sanity_check_image(hsv)
-            if not is_bad_image:
+            ret = sanity_check_image(hsv)
+            if not ret.is_bad_image:
                 return raw
     except Exception as e:
         logger.exception(e)
@@ -107,8 +118,8 @@ def camera_name_to_hostname(camera_name: str) -> str:
 
     >>> camera_name_to_hostname('cabin_driveway')
     'driveway.cabin'
-
     """
+
     mapping = {
         "driveway": "driveway.house",
         "backyard": "backyard.house",
@@ -126,8 +137,9 @@ 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]:
     """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"
+    stream = f"rtsp://camera:{scott_secrets.CAMERA_PASSWORD}@{hostname}:554/live"
     logger.debug('Fetching image from RTSP stream %s', stream)
     try:
         cmd = [
@@ -160,6 +172,7 @@ def fetch_camera_image_from_rtsp_stream(camera_name: str, *, width: int = 256) -
 @decorator_utils.timeout(seconds=30, use_signals=False)
 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)
     if raw is None:
@@ -181,6 +194,8 @@ def _fetch_camera_image(camera_name: str, *, width: int = 256, quality: int = 70
 
 
 def fetch_camera_image(camera_name: str, *, width: int = 256, quality: int = 70) -> RawJpgHsv:
+    """Fetch an image given the camera_name."""
+
     try:
         return _fetch_camera_image(camera_name, width=width, quality=quality)
     except exceptions.TimeoutError: