"""Create and maintain camera processes / management.""" import logging import multiprocessing as mp import threading from multiprocessing import Queue from multiprocessing.managers import DictProxy, SyncManager from multiprocessing.synchronize import Event as MpEvent from frigate.camera import CameraMetrics, PTZMetrics from frigate.config import FrigateConfig from frigate.config.camera import CameraConfig from frigate.config.camera.updater import ( CameraConfigUpdateEnum, CameraConfigUpdateSubscriber, ) from frigate.const import REPLAY_CAMERA_PREFIX from frigate.models import Regions from frigate.util.builtin import empty_and_close_queue from frigate.util.image import SharedMemoryFrameManager, UntrackedSharedMemory from frigate.util.object import get_camera_regions_grid from frigate.util.services import calculate_shm_requirements from frigate.video import CameraCapture, CameraTracker logger = logging.getLogger(__name__) class CameraMaintainer(threading.Thread): def __init__( self, config: FrigateConfig, detection_queue: Queue, detected_frames_queue: Queue, camera_metrics: DictProxy, ptz_metrics: dict[str, PTZMetrics], stop_event: MpEvent, metrics_manager: SyncManager, ): super().__init__(name="camera_processor") self.config = config self.detection_queue = detection_queue self.detected_frames_queue = detected_frames_queue self.stop_event = stop_event self.camera_metrics = camera_metrics self.ptz_metrics = ptz_metrics self.frame_manager = SharedMemoryFrameManager() self.region_grids: dict[str, list[list[dict[str, int]]]] = {} self.update_subscriber = CameraConfigUpdateSubscriber( self.config, {}, [ CameraConfigUpdateEnum.add, CameraConfigUpdateEnum.remove, CameraConfigUpdateEnum.refresh, ], ) self.shm_count = self.__calculate_shm_frame_count() self.camera_processes: dict[str, mp.Process] = {} self.capture_processes: dict[str, mp.Process] = {} self.camera_stop_events: dict[str, MpEvent] = {} self.metrics_manager = metrics_manager def __ensure_camera_stop_event(self, camera: str) -> MpEvent: camera_stop_event = self.camera_stop_events.get(camera) if camera_stop_event is None: camera_stop_event = mp.Event() self.camera_stop_events[camera] = camera_stop_event else: camera_stop_event.clear() return camera_stop_event def __init_historical_regions(self) -> None: # delete region grids for removed or renamed cameras cameras = list(self.config.cameras.keys()) Regions.delete().where(~(Regions.camera << cameras)).execute() # create or update region grids for each camera for camera in self.config.cameras.values(): assert camera.name is not None self.region_grids[camera.name] = get_camera_regions_grid( camera.name, camera.detect, max(self.config.model.width, self.config.model.height), ) def __calculate_shm_frame_count(self) -> int: shm_stats = calculate_shm_requirements(self.config) if not shm_stats: # /dev/shm not available return 0 logger.debug( f"Calculated total camera size {shm_stats['available']} / " f"{shm_stats['camera_frame_size']} :: {shm_stats['shm_frame_count']} " f"frames for each camera in SHM" ) if shm_stats["shm_frame_count"] < 20: logger.warning( f"The current SHM size of {shm_stats['total']}MB is too small, " f"recommend increasing it to at least {shm_stats['min_shm']}MB." ) return int(shm_stats["shm_frame_count"]) def __start_camera_processor( self, name: str, config: CameraConfig, runtime: bool = False ) -> None: if not config.enabled_in_config: logger.info(f"Camera processor not started for disabled camera {name}") return camera_stop_event = self.__ensure_camera_stop_event(name) if runtime: self.camera_metrics[name] = CameraMetrics(self.metrics_manager) self.ptz_metrics[name] = PTZMetrics(autotracker_enabled=False) self.region_grids[name] = get_camera_regions_grid( name, config.detect, max(self.config.model.width, self.config.model.height), ) try: largest_frame = max( [ det.model.height * det.model.width * 3 if det.model is not None else 320 for det in self.config.detectors.values() ] ) UntrackedSharedMemory(name=f"out-{name}", create=True, size=20 * 6 * 4) UntrackedSharedMemory( name=name, create=True, size=largest_frame, ) except FileExistsError: pass camera_process = CameraTracker( config, self.config.model, self.config.model.merged_labelmap, self.detection_queue, self.detected_frames_queue, self.camera_metrics[name], self.ptz_metrics[name], self.region_grids[name], camera_stop_event, self.config.logger, ) self.camera_processes[name] = camera_process camera_process.start() self.camera_metrics[name].process_pid.value = camera_process.pid logger.info(f"Camera processor started for {name}: {camera_process.pid}") def __start_camera_capture( self, name: str, config: CameraConfig, runtime: bool = False ) -> None: if not config.enabled_in_config: logger.info(f"Capture process not started for disabled camera {name}") return camera_stop_event = self.__ensure_camera_stop_event(name) # pre-create shms count = 10 if runtime else self.shm_count for i in range(count): frame_size = config.frame_shape_yuv[0] * config.frame_shape_yuv[1] self.frame_manager.create(f"{config.name}_frame{i}", frame_size) capture_process = CameraCapture( config, count, self.camera_metrics[name], camera_stop_event, self.config.logger, ) capture_process.daemon = True self.capture_processes[name] = capture_process capture_process.start() self.camera_metrics[name].capture_process_pid.value = capture_process.pid logger.info(f"Capture process started for {name}: {capture_process.pid}") def __stop_camera_capture_process(self, camera: str) -> None: capture_process = self.capture_processes.get(camera) if capture_process is not None: logger.info(f"Waiting for capture process for {camera} to stop") camera_stop_event = self.camera_stop_events.get(camera) if camera_stop_event is not None: camera_stop_event.set() capture_process.join(timeout=10) if capture_process.is_alive(): logger.warning( f"Capture process for {camera} didn't exit, forcing termination" ) capture_process.terminate() capture_process.join() def __unlink_camera_frame_slots(self, camera: str) -> None: """Drop the camera's per-frame YUV SHM segments from this process's frame_manager and unlink them at the OS level. Safe to call after the camera's capture/processor subprocesses have been joined — they no longer hold mappings, so unlink frees the segments immediately. Other long-lived processes that opened these slots will continue using their existing mappings until they call frame_manager.get with a shape that no longer fits (the get path drops and reopens stale refs). """ prefix = f"{camera}_frame" names = [n for n in list(self.frame_manager.shm_store) if n.startswith(prefix)] for name in names: try: self.frame_manager.delete(name) except Exception as exc: logger.debug("Could not unlink SHM %s: %s", name, exc) def __stop_camera_process(self, camera: str) -> None: camera_process = self.camera_processes.get(camera) if camera_process is not None: logger.info(f"Waiting for process for {camera} to stop") camera_stop_event = self.camera_stop_events.get(camera) if camera_stop_event is not None: camera_stop_event.set() camera_process.join(timeout=10) if camera_process.is_alive(): logger.warning(f"Process for {camera} didn't exit, forcing termination") camera_process.terminate() camera_process.join() logger.info(f"Closing frame queue for {camera}") empty_and_close_queue(self.camera_metrics[camera].frame_queue) def run(self) -> None: self.__init_historical_regions() # start camera processes for camera, config in self.config.cameras.items(): self.__start_camera_processor(camera, config) self.__start_camera_capture(camera, config) while not self.stop_event.wait(1): updates = self.update_subscriber.check_for_updates() for update_type, updated_cameras in updates.items(): if update_type == CameraConfigUpdateEnum.add.name: for camera in updated_cameras: if ( camera in self.camera_processes or camera in self.capture_processes ): continue self.__start_camera_processor( camera, self.update_subscriber.camera_configs[camera], runtime=True, ) self.__start_camera_capture( camera, self.update_subscriber.camera_configs[camera], runtime=True, ) elif update_type == CameraConfigUpdateEnum.remove.name: for camera in updated_cameras: self.__stop_camera_capture_process(camera) self.__stop_camera_process(camera) self.__unlink_camera_frame_slots(camera) self.capture_processes.pop(camera, None) self.camera_processes.pop(camera, None) self.camera_stop_events.pop(camera, None) self.region_grids.pop(camera, None) self.camera_metrics.pop(camera, None) self.ptz_metrics.pop(camera, None) elif update_type == CameraConfigUpdateEnum.refresh.name: # Recycle replay cameras so detect width/height/fps # propagate through ffmpeg args, SHM sizing, and the # region grid. Regular cameras detect change still # requires a full restart. for camera in updated_cameras: if not camera.startswith(REPLAY_CAMERA_PREFIX): continue new_config = self.update_subscriber.camera_configs.get(camera) if new_config is None: # remove arrived in the same batch continue if ( camera not in self.camera_processes and camera not in self.capture_processes ): continue # rebuild ffmpeg cmds on the shared config so the # new subprocesses spawn with current args new_config.recreate_ffmpeg_cmds() self.__stop_camera_capture_process(camera) self.__stop_camera_process(camera) self.__unlink_camera_frame_slots(camera) self.capture_processes.pop(camera, None) self.camera_processes.pop(camera, None) self.__start_camera_processor(camera, new_config, runtime=True) self.__start_camera_capture(camera, new_config, runtime=True) # ensure the capture processes are done for camera in self.capture_processes.keys(): self.__stop_camera_capture_process(camera) # ensure the camera processors are done for camera in self.camera_processes.keys(): self.__stop_camera_process(camera) self.update_subscriber.stop() self.frame_manager.cleanup()