About Multi-camera viewer optimized for RTSP streams
1import json
2import os
3
4from camera_utils import normalize_reolinkproxy_camera
5from detection import DEFAULT_DETECTION_CONFIG
6from notifications import DEFAULT_EMAIL_CONFIG
7
8
9CONFIG_PATH = "camera_config.json"
10DEFAULT_RECORDING_PATH = os.path.expanduser("~/Videos/Reolink")
11
12
13def snapshot_path_for(recording_path: str) -> str:
14 return os.path.join(recording_path, "snapshots")
15
16
17def config_payload(window) -> dict:
18 detection_config = dict(window.detection_config)
19 detection_config.pop("enabled", None)
20 return {
21 "cameras": window.cameras,
22 "recording_path": window.recording_path,
23 "detection": detection_config,
24 "email": window.email_config,
25 "cameras_per_row": window.cameras_per_row,
26 "next_camera_id": window.next_camera_id,
27 "language": window.language,
28 "order_custom": window._order_custom,
29 "preview_camera_ids": list(window.selected_camera_ids),
30 "selected_camera_id": window.selected_camera_id,
31 }
32
33
34def save_config_data(config: dict, path: str = CONFIG_PATH):
35 with open(path, "w") as f:
36 json.dump(config, f, indent=2)
37
38
39def _coerce_camera_id(camera: dict):
40 try:
41 return int(camera.get("id"))
42 except Exception:
43 return None
44
45
46def _repair_camera_ids(cameras: list[dict]) -> tuple[list[dict], bool]:
47 used_ids = set()
48 max_id = 0
49 fixed = False
50 for camera in cameras:
51 camera_id = _coerce_camera_id(camera)
52 if camera_id is not None:
53 max_id = max(max_id, camera_id)
54
55 repaired = []
56 for camera in cameras:
57 camera_id = _coerce_camera_id(camera)
58 if camera_id is None or camera_id in used_ids:
59 max_id += 1
60 camera["id"] = max_id
61 used_ids.add(max_id)
62 fixed = True
63 else:
64 used_ids.add(camera_id)
65 repaired.append(camera)
66 return repaired, fixed
67
68
69def _dedupe_by_url(cameras: list[dict]) -> tuple[list[dict], bool]:
70 seen_urls = set()
71 deduped = []
72 fixed = False
73 for camera in cameras:
74 url = (camera.get("url") or "").strip()
75 if url:
76 if url in seen_urls:
77 fixed = True
78 continue
79 seen_urls.add(url)
80 if "uid" not in camera:
81 camera["uid"] = ""
82 fixed = True
83 if "detection_enabled" not in camera:
84 camera["detection_enabled"] = False
85 fixed = True
86 deduped.append(camera)
87 return deduped, fixed
88
89
90def repair_cameras(cameras: list[dict], order_custom: bool) -> tuple[list[dict], bool]:
91 repaired, ids_fixed = _repair_camera_ids(cameras)
92 repaired, urls_fixed = _dedupe_by_url(repaired)
93 proxy_fixed = False
94 for camera in repaired:
95 proxy_fixed = normalize_reolinkproxy_camera(camera) or proxy_fixed
96 if not order_custom:
97 try:
98 repaired.sort(key=lambda camera: int(camera.get("id", 0)))
99 except Exception:
100 pass
101 return repaired, ids_fixed or urls_fixed or proxy_fixed
102
103
104def load_config_data(path: str = CONFIG_PATH) -> tuple[dict | None, bool]:
105 try:
106 with open(path, "r") as f:
107 raw_config = json.load(f)
108 except FileNotFoundError:
109 return None, False
110
111 order_custom = bool(raw_config.get("order_custom", False))
112 cameras, fixed = repair_cameras(raw_config.get("cameras", []), order_custom)
113 next_camera_id = max([camera.get("id", 0) for camera in cameras] + [0]) + 1
114 if raw_config.get("next_camera_id") != next_camera_id:
115 fixed = True
116
117 valid_camera_ids = {
118 int(camera.get("id"))
119 for camera in cameras
120 if camera.get("id") is not None
121 }
122 preview_ids = []
123 for camera_id in raw_config.get("preview_camera_ids", []):
124 try:
125 camera_id = int(camera_id)
126 except Exception:
127 continue
128 if camera_id in valid_camera_ids and camera_id not in preview_ids:
129 preview_ids.append(camera_id)
130
131 try:
132 selected_camera_id = (
133 int(raw_config.get("selected_camera_id"))
134 if raw_config.get("selected_camera_id") is not None
135 else None
136 )
137 except Exception:
138 selected_camera_id = None
139 if selected_camera_id not in valid_camera_ids:
140 selected_camera_id = None
141
142 config = {
143 **raw_config,
144 "cameras": cameras,
145 "recording_path": raw_config.get("recording_path", DEFAULT_RECORDING_PATH),
146 "snapshot_path": snapshot_path_for(raw_config.get("recording_path", DEFAULT_RECORDING_PATH)),
147 "detection": {**DEFAULT_DETECTION_CONFIG, **raw_config.get("detection", {})},
148 "email": {**DEFAULT_EMAIL_CONFIG, **raw_config.get("email", {})},
149 "cameras_per_row": raw_config.get("cameras_per_row", 3),
150 "next_camera_id": next_camera_id,
151 "language": raw_config.get("language", "de"),
152 "order_custom": order_custom,
153 "preview_camera_ids": preview_ids,
154 "selected_camera_id": selected_camera_id,
155 }
156 return config, fixed