frigate/frigate/camera/maintainer.py

"""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=config.onvif.autotracking.enabled
            )
            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()