X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=camera_utils.py;h=03ac621a52eb4d855ec3ac5458f59712e7eea657;hb=36fea7f15ed17150691b5b3ead75450e575229ef;hp=204a3374a78e3521daba62447644b39ea67b7bc5;hpb=bb3ebebb1a2458fa52b042b35d8e5fbad408ff80;p=python_utils.git diff --git a/camera_utils.py b/camera_utils.py index 204a337..03ac621 100644 --- a/camera_utils.py +++ b/camera_utils.py @@ -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 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), ) @@ -70,7 +74,9 @@ def fetch_camera_image_from_video_server( 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( + 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])}.' @@ -160,9 +166,7 @@ def fetch_camera_image_from_rtsp_stream( def _fetch_camera_image( camera_name: str, *, width: int = 256, quality: int = 70 ) -> RawJpgHsv: - """Fetch a webcam image given the camera name. - - """ + """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 @@ -181,7 +185,9 @@ def _fetch_camera_image( jpg=jpg, hsv=hsv, ) - msg = "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) @@ -199,4 +205,5 @@ def fetch_camera_image( if __name__ == '__main__': import doctest + doctest.testmod()