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
|
import socket
from ..config import ConfigUnit, LinuxBoardsConfig
from typing import Optional
from .types import CameraType, VideoContainerType, VideoCodecType
_lbc = LinuxBoardsConfig()
def _validate_roi_line(field, value, error) -> bool:
p = value.split(' ')
if len(p) != 4:
error(field, f'{field}: must contain four coordinates separated by space')
for n in p:
if not n.isnumeric():
error(field, f'{field}: invalid coordinates (not a number)')
return True
class IpcamConfig(ConfigUnit):
NAME = 'ipcam'
@classmethod
def schema(cls) -> Optional[dict]:
return {
'cameras': {
'type': 'dict',
'keysrules': {'type': ['string', 'integer']},
'valuesrules': {
'type': 'dict',
'schema': {
'type': {'type': 'string', 'allowed': [t.value for t in CameraType], 'required': True},
'enabled': {'type': 'boolean'},
'motion': {
'type': 'dict',
'schema': {
'threshold': {'type': ['float', 'integer']},
'roi': {
'type': 'list',
'schema': {'type': 'string', 'check_with': _validate_roi_line}
}
}
},
}
}
},
'zones': {
'type': 'dict',
'keysrules': {'type': 'string'},
'valuesrules': {
'type': 'list',
'schema': {'type': ['string', 'integer']} # same type as for 'cameras' keysrules
}
},
'camera_ip_template': {'type': 'string', 'required': True},
'motion_padding': {'type': 'integer', 'required': True},
'motion_telegram': {'type': 'boolean', 'required': True},
'fix_interval': {'type': 'integer', 'required': True},
'fix_enabled': {'type': 'boolean', 'required': True},
'cleanup_min_gb': {'type': 'integer', 'required': True},
'cleanup_interval': {'type': 'integer', 'required': True},
# TODO FIXME
'fragment_url_templates': cls._url_templates_schema(),
'original_file_url_templates': cls._url_templates_schema(),
'hls_path': {'type': 'string', 'required': True},
'motion_processing_tmpfs_path': {'type': 'string', 'required': True},
'rtsp_creds': {
'required': True,
'type': 'dict',
'schema': {
'login': {'type': 'string', 'required': True},
'password': {'type': 'string', 'required': True},
}
},
'web_creds': {
'required': True,
'type': 'dict',
'schema': {
'login': {'type': 'string', 'required': True},
'password': {'type': 'string', 'required': True},
}
}
}
@staticmethod
def custom_validator(data):
pass
# FIXME rewrite or delete, looks kinda obsolete
# for n, cam in data['cameras'].items():
# linux_box = _lbc[cam['server']]
# if 'ext_hdd' not in linux_box:
# raise ValueError(f'cam-{n}: linux box {cam["server"]} must have ext_hdd defined')
# disk = cam['disk']-1
# if disk < 0 or disk >= len(linux_box['ext_hdd']):
# raise ValueError(f'cam-{n}: invalid disk index for linux box {cam["server"]}')
@classmethod
def _url_templates_schema(cls) -> dict:
return {
'type': 'list',
'empty': False,
'schema': {
'type': 'list',
'empty': False,
'schema': {'type': 'string'}
}
}
# FIXME
def get_all_cam_names(self,
filter_by_server: Optional[str] = None,
filter_by_disk: Optional[int] = None,
only_enabled=True) -> list[int]:
cams = []
if filter_by_server is not None and filter_by_server not in _lbc:
raise ValueError(f'invalid filter_by_server: {filter_by_server} not found in {_lbc.__class__.__name__}')
for cam, params in self['cameras'].items():
if only_enabled and not self.is_camera_enabled(cam):
continue
if filter_by_server is None or params['server'] == filter_by_server:
if filter_by_disk is None or params['disk'] == filter_by_disk:
cams.append(int(cam))
return cams
# def get_all_cam_names_for_this_server(self,
# filter_by_disk: Optional[int] = None):
# return self.get_all_cam_names(filter_by_server=socket.gethostname(),
# filter_by_disk=filter_by_disk)
# def get_cam_server_and_disk(self, cam: int) -> tuple[str, int]:
# return self['cameras'][cam]['server'], self['cameras'][cam]['disk']
def has_camera(self, camera: int) -> bool:
return camera in tuple(self['cameras'].keys())
def has_zone(self, zone: str) -> bool:
return zone in tuple(self['zones'].keys())
def get_camera_container(self, camera: int) -> VideoContainerType:
return self.get_camera_type(camera).get_container()
def get_camera_type(self, camera: int) -> CameraType:
return CameraType(self['cameras'][camera]['type'])
def get_rtsp_creds(self) -> tuple[str, str]:
return self['rtsp_creds']['login'], self['rtsp_creds']['password']
def get_camera_ip(self, camera: int) -> str:
return self['camera_ip_template'] % (str(camera),)
def is_camera_enabled(self, camera: int) -> bool:
try:
return self['cameras'][camera]['enabled']
except KeyError:
return True
|
