X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=camera_utils.py;h=c789ed61e19d304391c9d10393bcb22094b0b7d7;hb=e8fbbb7306430478dec55d2c963eed116d8330cc;hp=e85bd6e4dbbedc060aad4944d16519b91bd1c746;hpb=497fb9e21f45ec08e1486abaee6dfa7b20b8a691;p=python_utils.git diff --git a/camera_utils.py b/camera_utils.py index e85bd6e..c789ed6 100644 --- 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 @@ -12,39 +13,47 @@ import numpy as np import requests import decorator_utils +import exceptions 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] -class BlueIrisImageMetadata(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 analyze_blue_iris_image(hsv: np.ndarray) -> BlueIrisImageMetadata: - """See if a Blue Iris image is bad and infrared.""" +def sanity_check_image(hsv: np.ndarray) -> SanityCheckImageMetadata: + """See if a Blue Iris or Shinobi image is bad and infrared.""" + + def is_near(a, b) -> bool: + return abs(a - b) < 3 + rows, cols, _ = hsv.shape num_pixels = rows * cols + weird_orange_count = 0 hs_zero_count = 0 - gray_count = 0 for r in range(rows): for c in range(cols): pixel = hsv[(r, c)] - if pixel[0] == 0 and pixel[1] == 0: + 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 - if abs(pixel[2] - 64) <= 10: - gray_count += 1 - logger.debug(f"gray#={gray_count}, hs0#={hs_zero_count}") - return BlueIrisImageMetadata( - gray_count > (num_pixels * 0.33), hs_zero_count > (num_pixels * 0.75) + 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), ) @@ -55,25 +64,51 @@ 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.56:81/image/{camera_name}?w={width}&q={quality}" + 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('Read %d byte image from HTTP server', len(response.content)) 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]), + ) 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]) + ) hsv = cv2.cvtColor(jpg, cv2.COLOR_BGR2HSV) - (is_bad_image, _) = analyze_blue_iris_image(hsv) + logger.debug( + '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: - 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", @@ -89,11 +124,11 @@ def blue_iris_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 = 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('Fetching image from RTSP stream %s', stream) try: cmd = [ "/usr/bin/timeout", @@ -102,7 +137,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", @@ -111,30 +146,24 @@ 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: 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 @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") @@ -145,16 +174,21 @@ 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) -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 decorator_utils.TimeoutError: + except exceptions.TimeoutError: + logger.warning('Fetching camera image operation timed out.') return RawJpgHsv(None, None, None) + + +if __name__ == '__main__': + import doctest + + doctest.testmod()