X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=camera_utils.py;h=9e7efd6dfccbd7df2d1eb8bd13eb8075a6bfe4f1;hb=31c81f6539969a5eba864d3305f9fb7bf716a367;hp=83664fdc1bad65cb5e9ee4f70de4161df4fad832;hpb=55a3172e37855f388b9ba0dfc91641a6c9ad1376;p=python_utils.git diff --git a/camera_utils.py b/camera_utils.py index 83664fd..9e7efd6 100644 --- 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 @@ -20,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] @@ -27,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 @@ -44,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), ) @@ -65,20 +69,26 @@ 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) @@ -88,7 +98,16 @@ def fetch_camera_image_from_video_server( 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", @@ -108,8 +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", @@ -134,7 +154,7 @@ def fetch_camera_image_from_rtsp_stream( return out except Exception as e: logger.exception(e) - msg = "Failed to retrieve image via RTSP {stream}, returning None." + msg = f"Failed to retrieve image via RTSP {stream}, returning None." logger.warning(msg) warnings.warn(msg, stacklevel=2) return None @@ -150,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") @@ -175,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()