1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
#!/usr/bin/env python3
"""Utilities for dealing with webcam images."""
import logging
import platform
import subprocess
from typing import NamedTuple, Optional
import cv2 # type: ignore
import numpy as np
import requests
import decorator_utils
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):
"""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."""
rows, cols, _ = hsv.shape
num_pixels = rows * cols
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:
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)
)
@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.56:81/image/{camera_name}?w={width}&q={quality}"
try:
response = requests.get(url, stream=False, timeout=10.0)
if response.ok:
raw = response.content
tmp = np.frombuffer(raw, dtype="uint8")
jpg = cv2.imdecode(tmp, cv2.IMREAD_COLOR)
hsv = cv2.cvtColor(jpg, cv2.COLOR_BGR2HSV)
(is_bad_image, _) = analyze_blue_iris_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}")
return None
def blue_iris_camera_name_to_hostname(camera_name: str) -> str:
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 = blue_iris_camera_name_to_hostname(camera_name)
try:
cmd = [
"/usr/bin/timeout",
"-k 9s",
"8s",
"/usr/local/bin/ffmpeg",
"-y",
"-i",
f"rtsp://camera:IaLaIok@{hostname}:554/live",
"-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)
logger.warning("Failed to retrieve image from RTSP stream")
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")
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,
)
logger.warning(
"Failed to retieve image from both video server and direct RTSP stream"
)
return RawJpgHsv(None, None, None)
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:
return RawJpgHsv(None, None, None)
|
