import socket from ..config import ConfigUnit from ..linux import 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