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
|
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}
}
}
},
}
}
},
'areas': {
'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) -> 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 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 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),)
|
