#!/usr/bin/env python3 # © Copyright 2021-2022, Scott Gasch """Utilities for dealing with webcam images.""" import logging import platform import subprocess import warnings from dataclasses import dataclass from typing import Optional import cv2 # type: ignore import numpy as np import requests import decorator_utils import exceptions import scott_secrets logger = logging.getLogger(__name__) @dataclass class RawJpgHsv: """Raw image bytes, the jpeg image and the HSV (hue saturation value) image.""" raw: Optional[bytes] = None jpg: Optional[np.ndarray] = None hsv: Optional[np.ndarray] = None @dataclass class SanityCheckImageMetadata: """Is a Blue Iris image bad (big grey borders around it) or infrared?""" is_infrared_image: bool = False is_bad_image: bool = False 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 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): weird_orange_count += 1 elif is_near(pixel[0], 0) and is_near(pixel[1], 0): hs_zero_count += 1 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), ) @decorator_utils.retry_if_none(tries=2, delay_sec=1, backoff=1.1) 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/{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) 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) logger.debug( 'Converted JPG into %s HSV HSV %s with element type %s', hsv.shape, type(hsv), type(hsv[0][0]), ) ret = sanity_check_image(hsv) if not ret.is_bad_image: return raw except Exception as e: logger.exception(e) msg = f"Got a bad image or HTTP error from {url}; returning None." logger.warning(msg) warnings.warn(msg, stacklevel=2) return None 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", "frontdoor": "frontdoor.house", "cabin_driveway": "driveway.cabin", } camera_name = mapping.get(camera_name, camera_name) if "." not in camera_name: hostname = platform.node() suffix = hostname.split(".")[-1] camera_name += f".{suffix}" return camera_name @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:{scott_secrets.CAMERA_PASSWORD}@{hostname}:554/live" logger.debug('Fetching image from RTSP stream %s', stream) try: cmd = [ "/usr/bin/timeout", "-k 9s", "8s", "/usr/local/bin/ffmpeg", "-y", "-i", f"{stream}", "-f", "singlejpeg", "-vframes", "1", "-vf", f"scale={width}:-1", "-", ] 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) 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: """Fetch a webcam image given the camera name.""" logger.debug("Trying to fetch camera image from video server") if camera_name == 'frontdoor': camera_name = 'front_door' 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") 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") jpg = cv2.imdecode(tmp, cv2.IMREAD_COLOR) hsv = cv2.cvtColor(jpg, cv2.COLOR_BGR2HSV) return RawJpgHsv( raw=raw, jpg=jpg, hsv=hsv, ) 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: """Fetch an image given the camera_name.""" 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()