X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=camera_utils.py;h=99ccdb32a5e949e7edd1b6ce89e866b6b6c63119;hb=f2600f30801c849fc1d139386e3ddc3c9eb43e30;hp=66ef6ac8e4d77f76f1775569f808f1ab57839176;hpb=52290f9c9e0eeaba3d5a067043f5ba98c9b386e5;p=python_utils.git diff --git a/camera_utils.py b/camera_utils.py index 66ef6ac..99ccdb3 100644 --- a/camera_utils.py +++ b/camera_utils.py @@ -5,8 +5,9 @@ import logging import platform import subprocess -from typing import NamedTuple, Optional import warnings +from dataclasses import dataclass +from typing import Optional import cv2 # type: ignore import numpy as np @@ -18,21 +19,26 @@ import exceptions 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_bad_image: bool = False + is_infrared_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 @@ -43,17 +49,14 @@ def sanity_check_image(hsv: np.ndarray) -> SanityCheckImageMetadata: 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) - ): + 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)): + 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}") + 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) + hs_zero_count > (num_pixels * 0.75), + weird_orange_count > (num_pixels * 0.75), ) @@ -64,27 +67,38 @@ 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.226:8080/Umtxxf1uKMBniFblqeQ9KRbb6DDzN4/jpeg/GKlT2FfiSQ/{camera_name}/s.jpg" - logger.debug(f'Fetching image from {url}') + 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(f'Read {len(response.content)} byte image from HTTP server') + logger.debug('Read %d byte image from HTTP server', len(response.content)) tmp = np.frombuffer(raw, dtype="uint8") logger.debug( - f'Translated raw content into {tmp.shape} {type(tmp)} with element type {type(tmp[0])}.' + '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( - f'Decoded into {jpg.shape} jpeg {type(jpg)} with element type {type(jpg[0][0])}' + '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( - f'Converted JPG into HSV {hsv.shape} HSV {type(hsv)} with element type {type(hsv[0][0])}' + '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: + ret = sanity_check_image(hsv) + if not ret.is_bad_image: return raw except Exception as e: logger.exception(e) @@ -119,13 +133,11 @@ 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]: +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" - logger.debug(f'Fetching image from RTSP stream {stream}') + logger.debug('Fetching image from RTSP stream %s', stream) try: cmd = [ "/usr/bin/timeout", @@ -143,34 +155,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) - 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 @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. - - """ +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") @@ -187,9 +189,7 @@ def _fetch_camera_image( 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 exceptions.TimeoutError: @@ -199,4 +199,5 @@ def fetch_camera_image( if __name__ == '__main__': import doctest + doctest.testmod()