X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=camera_utils.py;h=c2bd04da0adccdb439711e3df6fdacf700d92e84;hb=300300e13b91ed6a382ccbb1d7bdb3e3c3c46788;hp=c789ed61e19d304391c9d10393bcb22094b0b7d7;hpb=5c212d7639f62fcb936f9d7a0bbe704a9f7b213d;p=python_utils.git diff --git a/camera_utils.py b/camera_utils.py index c789ed6..c2bd04d 100644 --- a/camera_utils.py +++ b/camera_utils.py @@ -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: