mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-24 23:44:43 +03:00
* materialize event counts in metrics endpoint The metrics endpoint stored a lazy peewee query on the shared Prometheus collector, and collect() ran it later. Overlapping scrapes run in separate threadpool threads, so two requests could execute and step the same query's sqlite cursor at once. The rows are now read with list() before they reach the collector. * fix fps overflow restarting ffmpeg after 3 seconds instead of 3 retry intervals This is a regression from 0.18. The watchdog has restarted ffmpeg after 3 consecutive fps overflow checks since #13651, which was 3 retry intervals (30s by default) when the loop ran every retry_interval. #21297 moved the loop to 1s ticks without converting the count, so the window shrank to 3s. camera_fps is a 10s average, so one burst of buffered frames after a stream hiccup could keep it over the limit long enough to restart a healthy camera, and the next reconnect burst could repeat it. The threshold is now 3 * retry_interval ticks. When the count filled while restarts were still rate limited, the watchdog logged "Exiting ffmpeg...", marked the camera offline and reset the count without restarting. The check now waits for can_restart, so the log, status and restart always happen together. * keep recording cleanup running after a failed pass An exception anywhere in a cleanup pass, such as an OSError from a read-only or unmounted storage path, ended the recording cleanup thread with nothing logged. Retention stopped until restart, and recordings, previews and database rows kept growing. Each pass now logs the exception and the loop continues. * expose frigate_camera_events as a gauge The metric reports the current count of events in the database, which drops when retention prunes old events. As a counter, each prune read as a counter reset, so increase() and rate() inflated event counts. The series name changes from frigate_camera_events_total to frigate_camera_events. The test's porch/inside assertion was missing its `in event.text` and always passed. * keep uvicorn idle connections open longer than nginx's upstream keepalive uvicorn closed idle connections after its 5s default while nginx kept them pooled for 60s, so a request nginx sent on a connection uvicorn was closing got a 502. nginx retries that on a fresh connection for most methods, but not for POST or PATCH. * snapshot websocket clients under the manager lock in the output loop The output loop iterated ws4py's live websocket dict without its lock, so a client connecting or disconnecting mid-iteration raised "dictionary changed size during iteration" and killed the output loop. It now copies the clients under manager.lock once per frame and skips terminated ones, matching the jsmpeg and birdseye broadcasters. * only cache preview frames when recording is enabled The segment-end check tested the record config object instead of record.enabled, and the first frame after startup was written unconditionally. With recording disabled, the first segment still cached frames and encoded a preview. * fix untranslated unknown resolution text in camera info dialog * detect all AMD GPUs from vainfo The vainfo check only matched "AMD Radeon Graphics", so drivers reporting names like "AMD Radeon Vega 8 Graphics", "AMD Radeon RX 6600", "AMD Custom GPU 0405", or older codename strings fell through and their GPU stats were collected as Intel. It now matches Mesa's "for AMD" driver string. * return a 500 instead of raising when go2rtc is unreachable The two go2rtc stream GETs had no exception handling or timeout, so a go2rtc outage raised a connection error with a traceback, and a hung go2rtc held the request open indefinitely. Both now catch request errors, log the cause, and use a timeout. The per-stream probe gets a 35s read timeout because go2rtc probes each source in turn, and one unresponsive source takes about 10s. * retry camera snapshots after a failed load AutoUpdatingCameraImage only scheduled the next frame from onLoad, so a single failed latest.webp request froze the image until the component remounted. It now retries on error, dropping the periodic cache key so the retry requests a fresh URL. * reuse pooled upstream connections for nginx auth subrequests The /auth subrequest proxied to a variable address, which nginx can't pool, and its location-level headers dropped the inherited Connection header so nginx sent Connection: close. Every authenticated request opened and closed its own loopback connection to the API. It now proxies to the frigate_api upstream with an empty Connection header so the keepalive pool is reused.
823 lines
30 KiB
Python
823 lines
30 KiB
Python
import datetime
|
|
import logging
|
|
import multiprocessing as mp
|
|
import os
|
|
import secrets
|
|
import shutil
|
|
from collections.abc import Callable
|
|
from multiprocessing import Queue
|
|
from multiprocessing.managers import DictProxy, SyncManager
|
|
from multiprocessing.synchronize import Event as MpEvent
|
|
from pathlib import Path
|
|
|
|
import psutil
|
|
import uvicorn
|
|
from peewee import SqliteDatabase
|
|
from peewee_migrate import Router
|
|
|
|
from frigate.api.auth import hash_password
|
|
from frigate.api.fastapi_app import create_fastapi_app
|
|
from frigate.camera import CameraMetrics, PTZMetrics
|
|
from frigate.camera.maintainer import CameraMaintainer
|
|
from frigate.comms.base_communicator import Communicator
|
|
from frigate.comms.dispatcher import Dispatcher
|
|
from frigate.comms.event_metadata_updater import EventMetadataPublisher
|
|
from frigate.comms.inter_process import InterProcessCommunicator
|
|
from frigate.comms.mqtt import MqttClient
|
|
from frigate.comms.object_detector_signaler import DetectorProxy
|
|
from frigate.comms.webpush import WebPushClient
|
|
from frigate.comms.ws import WebSocketClient
|
|
from frigate.comms.zmq_proxy import ZmqProxy
|
|
from frigate.config.camera.updater import CameraConfigUpdatePublisher
|
|
from frigate.config.config import FrigateConfig
|
|
from frigate.config.holder import ConfigHolder
|
|
from frigate.config.profile_manager import ProfileManager
|
|
from frigate.const import (
|
|
CACHE_DIR,
|
|
CLIPS_DIR,
|
|
CONFIG_DIR,
|
|
EXPORT_DIR,
|
|
FACE_DIR,
|
|
MODEL_CACHE_DIR,
|
|
RECORD_DIR,
|
|
THUMB_DIR,
|
|
TRIGGER_DIR,
|
|
)
|
|
from frigate.data_processing.types import DataProcessorMetrics
|
|
from frigate.db.sqlitevecq import SqliteVecQueueDatabase
|
|
from frigate.debug_replay import (
|
|
DebugReplayManager,
|
|
cleanup_replay_cameras,
|
|
)
|
|
from frigate.detectors.detector_types import api_types
|
|
from frigate.detectors.device import build_detector_config, runner_names
|
|
from frigate.embeddings import EmbeddingProcess, EmbeddingsContext
|
|
from frigate.events.audio import AudioProcessor
|
|
from frigate.events.cleanup import EventCleanup
|
|
from frigate.events.maintainer import EventProcessor
|
|
from frigate.jobs.export import reap_stale_exports
|
|
from frigate.jobs.motion_search import stop_all_motion_search_jobs
|
|
from frigate.log import _stop_logging
|
|
from frigate.models import (
|
|
Event,
|
|
Export,
|
|
Notice,
|
|
NoticeStats,
|
|
Previews,
|
|
Recordings,
|
|
RecordingsToDelete,
|
|
Regions,
|
|
ReviewSegment,
|
|
Timeline,
|
|
Trigger,
|
|
User,
|
|
)
|
|
from frigate.notices import install_registry
|
|
from frigate.notices.registry import NoticeRegistry
|
|
from frigate.object_detection.base import ObjectDetectProcess
|
|
from frigate.object_detection.util import detection_frame_size
|
|
from frigate.output.output import OutputProcess
|
|
from frigate.ptz.autotrack import PtzAutoTracker
|
|
from frigate.ptz.onvif import OnvifController
|
|
from frigate.record.cleanup import RecordingCleanup
|
|
from frigate.record.export import migrate_exports
|
|
from frigate.record.record import RecordProcess
|
|
from frigate.review.review import ReviewProcess
|
|
from frigate.stats.emitter import StatsEmitter
|
|
from frigate.stats.util import stats_init
|
|
from frigate.storage import StorageMaintainer
|
|
from frigate.timeline import TimelineProcessor
|
|
from frigate.track.object_processing import TrackedObjectProcessor
|
|
from frigate.util.builtin import empty_and_close_queue
|
|
from frigate.util.image import UntrackedSharedMemory
|
|
from frigate.util.ownership import chown_to_runtime
|
|
from frigate.util.process import FrigateProcess
|
|
from frigate.util.runtime_deps import RuntimeDependencyError
|
|
from frigate.util.services import set_file_limit
|
|
from frigate.version import VERSION
|
|
from frigate.watchdog import FrigateWatchdog
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FrigateApp:
|
|
def __init__(
|
|
self, config: FrigateConfig, manager: SyncManager, stop_event: MpEvent
|
|
) -> None:
|
|
self.metrics_manager = manager
|
|
self.audio_process: mp.Process | None = None
|
|
self.stop_event = stop_event
|
|
self.detection_queues: dict[str, Queue] = {
|
|
model.scene: mp.Queue() for model in config.models
|
|
}
|
|
self.detectors: dict[str, ObjectDetectProcess] = {}
|
|
self.detection_shms: list[mp.shared_memory.SharedMemory] = []
|
|
self.log_queue: Queue = mp.Queue()
|
|
self.camera_metrics: DictProxy = self.metrics_manager.dict()
|
|
|
|
self.embeddings_metrics = DataProcessorMetrics(
|
|
self.metrics_manager, list(config.classification.custom.keys())
|
|
)
|
|
self.ptz_metrics: dict[str, PTZMetrics] = {}
|
|
self.processes: dict[str, int] = {}
|
|
self.embeddings: EmbeddingsContext | None = None
|
|
self.config_holder = ConfigHolder(config)
|
|
|
|
@property
|
|
def config(self) -> FrigateConfig:
|
|
"""The current config, not the one Frigate booted with.
|
|
|
|
Read through the holder so the deferred watchdog factories below build
|
|
a replacement process from the config as it is now. There is no setter
|
|
on purpose: a plain attribute would let a caller pin this back to a
|
|
single object and reintroduce the staleness.
|
|
"""
|
|
return self.config_holder.config
|
|
|
|
def ensure_dirs(self) -> None:
|
|
dirs = [
|
|
CONFIG_DIR,
|
|
RECORD_DIR,
|
|
THUMB_DIR,
|
|
f"{CLIPS_DIR}/cache",
|
|
CACHE_DIR,
|
|
MODEL_CACHE_DIR,
|
|
EXPORT_DIR,
|
|
]
|
|
|
|
if self.config.face_recognition.enabled:
|
|
dirs.append(FACE_DIR)
|
|
|
|
if self.config.semantic_search.enabled:
|
|
dirs.append(TRIGGER_DIR)
|
|
|
|
for d in dirs:
|
|
if not os.path.exists(d) and not os.path.islink(d):
|
|
logger.info(f"Creating directory: {d}")
|
|
os.makedirs(d, exist_ok=True)
|
|
chown_to_runtime(d)
|
|
else:
|
|
logger.debug(f"Skipping directory: {d}")
|
|
|
|
def init_debug_replay_manager(self) -> None:
|
|
self.replay_manager = DebugReplayManager()
|
|
|
|
def init_camera_metrics(self) -> None:
|
|
# create camera_metrics
|
|
for camera_name in self.config.cameras.keys():
|
|
self.camera_metrics[camera_name] = CameraMetrics(self.metrics_manager)
|
|
self.ptz_metrics[camera_name] = PTZMetrics()
|
|
|
|
def init_queues(self) -> None:
|
|
# Queue for cameras to push tracked objects to
|
|
# leaving room for 2 extra cameras to be added
|
|
self.detected_frames_queue: Queue = mp.Queue(
|
|
maxsize=(
|
|
sum(
|
|
camera.enabled_in_config == True
|
|
for camera in self.config.cameras.values()
|
|
)
|
|
+ 2
|
|
)
|
|
* 2
|
|
)
|
|
|
|
# Queue for timeline events
|
|
self.timeline_queue: Queue = mp.Queue()
|
|
|
|
def init_database(self) -> None:
|
|
def vacuum_db(db: SqliteDatabase) -> None:
|
|
logger.info("Running database vacuum")
|
|
db.execute_sql("VACUUM;")
|
|
|
|
try:
|
|
with open(f"{CONFIG_DIR}/.vacuum", "w") as f:
|
|
f.write(str(datetime.datetime.now().timestamp()))
|
|
except PermissionError:
|
|
logger.error("Unable to write to /config to save DB state")
|
|
|
|
# Migrate DB schema
|
|
migrate_db = SqliteDatabase(self.config.database.path)
|
|
|
|
# Run migrations
|
|
del logging.getLogger("peewee_migrate").handlers[:]
|
|
router = Router(migrate_db)
|
|
|
|
if len(router.diff) > 0:
|
|
logger.info("Making backup of DB before migrations...")
|
|
shutil.copyfile(
|
|
self.config.database.path,
|
|
self.config.database.path.replace("frigate.db", "backup.db"),
|
|
)
|
|
|
|
router.run()
|
|
|
|
# check if vacuum needs to be run
|
|
if os.path.exists(f"{CONFIG_DIR}/.vacuum"):
|
|
with open(f"{CONFIG_DIR}/.vacuum") as f:
|
|
try:
|
|
timestamp = round(float(f.readline()))
|
|
except Exception:
|
|
timestamp = 0
|
|
|
|
if (
|
|
timestamp
|
|
< (
|
|
datetime.datetime.now() - datetime.timedelta(weeks=2)
|
|
).timestamp()
|
|
):
|
|
vacuum_db(migrate_db)
|
|
else:
|
|
vacuum_db(migrate_db)
|
|
|
|
migrate_db.close()
|
|
|
|
# a root frigate service creates these as root; wal and shm recreated
|
|
# later in the run are realigned by the per-boot /config sweep
|
|
for db_file in (
|
|
self.config.database.path,
|
|
f"{self.config.database.path}-wal",
|
|
f"{self.config.database.path}-shm",
|
|
self.config.database.path.replace("frigate.db", "backup.db"),
|
|
):
|
|
if os.path.exists(db_file):
|
|
chown_to_runtime(db_file)
|
|
|
|
def init_go2rtc(self) -> None:
|
|
for proc in psutil.process_iter(["pid", "name"]):
|
|
if proc.info["name"] == "go2rtc":
|
|
logger.info(f"go2rtc process pid: {proc.info['pid']}")
|
|
self.processes["go2rtc"] = proc.info["pid"]
|
|
|
|
def init_recording_manager(self) -> None:
|
|
recording_process = RecordProcess(self.config, self.stop_event)
|
|
self.recording_process = recording_process
|
|
recording_process.start()
|
|
self.processes["recording"] = recording_process.pid or 0
|
|
logger.info(f"Recording process started: {recording_process.pid}")
|
|
|
|
def init_review_segment_manager(self) -> None:
|
|
review_segment_process = ReviewProcess(self.config, self.stop_event)
|
|
self.review_segment_process = review_segment_process
|
|
review_segment_process.start()
|
|
self.processes["review_segment"] = review_segment_process.pid or 0
|
|
logger.info(f"Review process started: {review_segment_process.pid}")
|
|
|
|
def init_embeddings_manager(self) -> None:
|
|
# always start the embeddings process
|
|
embedding_process = EmbeddingProcess(
|
|
self.config, self.embeddings_metrics, self.stop_event
|
|
)
|
|
self.embedding_process = embedding_process
|
|
embedding_process.start()
|
|
self.processes["embeddings"] = embedding_process.pid or 0
|
|
logger.info(f"Embedding process started: {embedding_process.pid}")
|
|
|
|
def bind_database(self) -> None:
|
|
"""Bind db to the main process."""
|
|
# NOTE: all db accessing processes need to be created before the db can be bound to the main process
|
|
self.db = SqliteVecQueueDatabase(
|
|
self.config.database.path,
|
|
pragmas={
|
|
"auto_vacuum": "FULL", # Does not defragment database
|
|
"cache_size": -512 * 1000, # 512MB of cache,
|
|
"synchronous": "NORMAL", # Safe when using WAL https://www.sqlite.org/pragma.html#pragma_synchronous
|
|
},
|
|
timeout=max(
|
|
60,
|
|
10
|
|
* len([c for c in self.config.cameras.values() if c.enabled_in_config]),
|
|
),
|
|
load_vec_extension=True,
|
|
)
|
|
models = [
|
|
Event,
|
|
Export,
|
|
Notice,
|
|
NoticeStats,
|
|
Previews,
|
|
Recordings,
|
|
RecordingsToDelete,
|
|
Regions,
|
|
ReviewSegment,
|
|
Timeline,
|
|
User,
|
|
Trigger,
|
|
]
|
|
self.db.bind(models)
|
|
|
|
def check_db_data_migrations(self) -> None:
|
|
# check if vacuum needs to be run
|
|
if not os.path.exists(f"{CONFIG_DIR}/.exports"):
|
|
try:
|
|
with open(f"{CONFIG_DIR}/.exports", "w") as f:
|
|
f.write(str(datetime.datetime.now().timestamp()))
|
|
except PermissionError:
|
|
logger.error("Unable to write to /config to save export state")
|
|
|
|
migrate_exports(self.config.ffmpeg, list(self.config.cameras.keys()))
|
|
|
|
def install_notice_registry(self) -> None:
|
|
self.notice_registry = NoticeRegistry()
|
|
install_registry(self.notice_registry)
|
|
|
|
def init_embeddings_client(self) -> None:
|
|
# Create a client for other processes to use
|
|
self.embeddings = EmbeddingsContext(self.db)
|
|
|
|
def init_inter_process_communicator(self) -> None:
|
|
self.inter_process_communicator = InterProcessCommunicator()
|
|
self.inter_config_updater = CameraConfigUpdatePublisher()
|
|
self.event_metadata_updater = EventMetadataPublisher()
|
|
self.inter_zmq_proxy = ZmqProxy()
|
|
self.detection_proxy = DetectorProxy()
|
|
|
|
def init_onvif(self) -> None:
|
|
self.onvif_controller = OnvifController(self.config, self.ptz_metrics)
|
|
|
|
def init_dispatcher(self) -> None:
|
|
comms: list[Communicator] = []
|
|
|
|
if self.config.mqtt.enabled:
|
|
comms.append(MqttClient(self.config))
|
|
|
|
notification_cameras = [
|
|
c
|
|
for c in self.config.cameras.values()
|
|
if c.enabled and c.notifications.enabled_in_config
|
|
]
|
|
|
|
if notification_cameras:
|
|
comms.append(WebPushClient(self.config, self.stop_event))
|
|
|
|
comms.append(WebSocketClient(self.config))
|
|
comms.append(self.inter_process_communicator)
|
|
|
|
self.dispatcher = Dispatcher(
|
|
self.config,
|
|
self.inter_config_updater,
|
|
self.onvif_controller,
|
|
self.ptz_metrics,
|
|
comms,
|
|
notice_registry=self.notice_registry,
|
|
)
|
|
self.dispatcher.start_communicators()
|
|
|
|
def init_profile_manager(self) -> None:
|
|
self.profile_manager = ProfileManager(
|
|
self.config, self.inter_config_updater, self.dispatcher
|
|
)
|
|
self.dispatcher.profile_manager = self.profile_manager
|
|
|
|
def ensure_detector_dependencies(self) -> None:
|
|
"""Install runtimes for the configured detector types.
|
|
|
|
Runs before any detector process starts so one install serves them
|
|
all and the user site is on sys.path before the forkserver copies it.
|
|
A failure is logged and startup continues; the detector process then
|
|
fails on its own with a clear import error.
|
|
"""
|
|
detector_types = {
|
|
spec.detector
|
|
for model in self.config.models
|
|
for spec in self.config.devices_for_model(model)
|
|
}
|
|
|
|
for detector_type in sorted(detector_types):
|
|
try:
|
|
api_types[detector_type].ensure_dependencies()
|
|
except RuntimeDependencyError as err:
|
|
logger.error("Unable to prepare the %s runtime: %s", detector_type, err)
|
|
|
|
def start_detectors(self) -> None:
|
|
model_cameras: dict[str, list[str]] = {
|
|
model.scene: [] for model in self.config.models
|
|
}
|
|
|
|
for name in self.config.cameras.keys():
|
|
model = self.config.model_for_camera(name)
|
|
model_cameras[model.scene].append(name)
|
|
|
|
try:
|
|
shm_in = UntrackedSharedMemory(
|
|
name=name,
|
|
create=True,
|
|
size=detection_frame_size(model),
|
|
)
|
|
except FileExistsError:
|
|
shm_in = UntrackedSharedMemory(name=name)
|
|
|
|
try:
|
|
shm_out = UntrackedSharedMemory(
|
|
name=f"out-{name}", create=True, size=20 * 6 * 4
|
|
)
|
|
except FileExistsError:
|
|
shm_out = UntrackedSharedMemory(name=f"out-{name}")
|
|
|
|
self.detection_shms.append(shm_in)
|
|
self.detection_shms.append(shm_out)
|
|
|
|
# a device may be listed more than once to run additional inference
|
|
# processes on it, so names are only unique once de-duplicated
|
|
all_devices = [
|
|
device
|
|
for model in self.config.models
|
|
for device in self.config.devices_for_model(model)
|
|
]
|
|
names = iter(runner_names(all_devices))
|
|
|
|
for model in self.config.models:
|
|
for device in self.config.devices_for_model(model):
|
|
name = next(names)
|
|
self.detectors[name] = ObjectDetectProcess(
|
|
name,
|
|
self.detection_queues[model.scene],
|
|
model_cameras[model.scene],
|
|
self.config,
|
|
build_detector_config(device, model),
|
|
self.stop_event,
|
|
)
|
|
|
|
def start_ptz_autotracker(self) -> None:
|
|
self.ptz_autotracker_thread = PtzAutoTracker(
|
|
self.config,
|
|
self.onvif_controller,
|
|
self.ptz_metrics,
|
|
self.dispatcher,
|
|
self.stop_event,
|
|
)
|
|
self.ptz_autotracker_thread.start()
|
|
|
|
def start_detected_frames_processor(self) -> None:
|
|
self.detected_frames_processor = TrackedObjectProcessor(
|
|
self.config,
|
|
self.dispatcher,
|
|
self.detected_frames_queue,
|
|
self.ptz_autotracker_thread,
|
|
self.stop_event,
|
|
)
|
|
self.detected_frames_processor.start()
|
|
|
|
def start_video_output_processor(self) -> None:
|
|
output_processor = OutputProcess(self.config, self.stop_event)
|
|
self.output_processor = output_processor
|
|
output_processor.start()
|
|
logger.info(f"Output process started: {output_processor.pid}")
|
|
|
|
def start_camera_processor(self) -> None:
|
|
self.camera_maintainer = CameraMaintainer(
|
|
self.config,
|
|
self.detection_queues,
|
|
self.detected_frames_queue,
|
|
self.camera_metrics,
|
|
self.ptz_metrics,
|
|
self.stop_event,
|
|
self.metrics_manager,
|
|
)
|
|
self.camera_maintainer.start()
|
|
|
|
def start_audio_processor(self) -> None:
|
|
self.audio_process = AudioProcessor(
|
|
self.config, self.camera_metrics, self.embeddings_metrics, self.stop_event
|
|
)
|
|
self.audio_process.start()
|
|
self.processes["audio_detector"] = self.audio_process.pid or 0
|
|
|
|
def start_timeline_processor(self) -> None:
|
|
self.timeline_processor = TimelineProcessor(
|
|
self.config, self.timeline_queue, self.stop_event
|
|
)
|
|
self.timeline_processor.start()
|
|
|
|
def start_event_processor(self) -> None:
|
|
self.event_processor = EventProcessor(
|
|
self.config,
|
|
self.timeline_queue,
|
|
self.stop_event,
|
|
)
|
|
self.event_processor.start()
|
|
|
|
def start_event_cleanup(self) -> None:
|
|
self.event_cleanup = EventCleanup(self.config, self.stop_event, self.db)
|
|
self.event_cleanup.start()
|
|
|
|
def start_record_cleanup(self) -> None:
|
|
self.record_cleanup = RecordingCleanup(self.config, self.stop_event)
|
|
self.record_cleanup.start()
|
|
|
|
def start_storage_maintainer(self) -> None:
|
|
self.storage_maintainer = StorageMaintainer(self.config, self.stop_event)
|
|
self.storage_maintainer.start()
|
|
|
|
def start_stats_emitter(self) -> None:
|
|
self.stats_emitter = StatsEmitter(
|
|
self.config,
|
|
stats_init(
|
|
self.config,
|
|
self.camera_metrics,
|
|
self.embeddings_metrics,
|
|
self.detectors,
|
|
self.processes,
|
|
self.storage_maintainer,
|
|
),
|
|
self.stop_event,
|
|
)
|
|
self.stats_emitter.start()
|
|
|
|
def start_watchdog(self) -> None:
|
|
self.frigate_watchdog = FrigateWatchdog(self.detectors, self.stop_event)
|
|
|
|
# (attribute on self, key in self.processes, factory)
|
|
specs: list[tuple[str, str, Callable[[], FrigateProcess]]] = [
|
|
(
|
|
"embedding_process",
|
|
"embeddings",
|
|
lambda: EmbeddingProcess(
|
|
self.config, self.embeddings_metrics, self.stop_event
|
|
),
|
|
),
|
|
(
|
|
"recording_process",
|
|
"recording",
|
|
lambda: RecordProcess(self.config, self.stop_event),
|
|
),
|
|
(
|
|
"review_segment_process",
|
|
"review_segment",
|
|
lambda: ReviewProcess(self.config, self.stop_event),
|
|
),
|
|
(
|
|
"output_processor",
|
|
"output",
|
|
lambda: OutputProcess(self.config, self.stop_event),
|
|
),
|
|
(
|
|
"audio_process",
|
|
"audio_detector",
|
|
lambda: AudioProcessor(
|
|
self.config,
|
|
self.camera_metrics,
|
|
self.embeddings_metrics,
|
|
self.stop_event,
|
|
),
|
|
),
|
|
]
|
|
|
|
for attr, key, factory in specs:
|
|
if getattr(self, attr, None) is None:
|
|
continue
|
|
|
|
def on_restart(
|
|
proc: FrigateProcess, _attr: str = attr, _key: str = key
|
|
) -> None:
|
|
setattr(self, _attr, proc)
|
|
self.processes[_key] = proc.pid or 0
|
|
|
|
self.frigate_watchdog.register(
|
|
key, getattr(self, attr), factory, on_restart
|
|
)
|
|
|
|
self.frigate_watchdog.start()
|
|
|
|
def init_auth(self) -> None:
|
|
if self.config.auth.enabled:
|
|
if (
|
|
User.select().count() == 0
|
|
or User.select()
|
|
.where((User.username == "admin") & (User.password_hash == ""))
|
|
.exists()
|
|
):
|
|
password = secrets.token_hex(16)
|
|
password_hash = hash_password(
|
|
password, iterations=self.config.auth.hash_iterations
|
|
)
|
|
User.insert(
|
|
{
|
|
User.username: "admin",
|
|
User.role: "admin",
|
|
User.password_hash: password_hash,
|
|
User.notification_tokens: [],
|
|
}
|
|
).on_conflict(
|
|
conflict_target=[User.username],
|
|
update={User.password_hash: password_hash},
|
|
).execute()
|
|
|
|
self.config.auth.admin_first_time_login = True
|
|
|
|
logger.info("********************************************************")
|
|
logger.info("********************************************************")
|
|
logger.info("*** Auth is enabled, but no users exist. ***")
|
|
logger.info("*** Created a default user: ***")
|
|
logger.info("*** User: admin ***")
|
|
logger.info(f"*** Password: {password} ***")
|
|
logger.info("********************************************************")
|
|
logger.info("********************************************************")
|
|
elif self.config.auth.reset_admin_password:
|
|
password = secrets.token_hex(16)
|
|
password_hash = hash_password(
|
|
password, iterations=self.config.auth.hash_iterations
|
|
)
|
|
User.replace(
|
|
username="admin",
|
|
role="admin",
|
|
password_hash=password_hash,
|
|
password_changed_at=datetime.datetime.now(),
|
|
notification_tokens=[],
|
|
).execute()
|
|
|
|
logger.info("********************************************************")
|
|
logger.info("********************************************************")
|
|
logger.info("*** Reset admin password set in the config. ***")
|
|
logger.info(f"*** Password: {password} ***")
|
|
logger.info("********************************************************")
|
|
logger.info("********************************************************")
|
|
else:
|
|
# Notification registrations store tokens on the admin row. Its
|
|
# empty password hash marks it for a generated password if auth
|
|
# is enabled later.
|
|
if User.select().count() == 0:
|
|
User.insert(
|
|
{
|
|
User.username: "admin",
|
|
User.role: "admin",
|
|
User.password_hash: "",
|
|
User.notification_tokens: [],
|
|
}
|
|
).execute()
|
|
|
|
def start(self) -> None:
|
|
logger.info(f"Starting Frigate ({VERSION})")
|
|
|
|
# Ensure global state.
|
|
self.ensure_dirs()
|
|
self.ensure_detector_dependencies()
|
|
|
|
# Set soft file limits.
|
|
set_file_limit()
|
|
|
|
# Start frigate services.
|
|
self.init_debug_replay_manager()
|
|
self.init_camera_metrics()
|
|
self.init_queues()
|
|
self.init_database()
|
|
self.init_onvif()
|
|
self.init_recording_manager()
|
|
self.init_review_segment_manager()
|
|
self.init_go2rtc()
|
|
self.init_embeddings_manager()
|
|
self.bind_database()
|
|
self.check_db_data_migrations()
|
|
self.install_notice_registry()
|
|
|
|
# Clean up any stale replay camera artifacts (filesystem + DB)
|
|
cleanup_replay_cameras()
|
|
|
|
# Reap any Export rows still marked in_progress from a previous
|
|
# session (crash, kill, broken migration). Runs synchronously before
|
|
# uvicorn binds so no API request can observe a stale row.
|
|
reap_stale_exports()
|
|
|
|
self.init_inter_process_communicator()
|
|
self.start_detectors()
|
|
self.init_dispatcher()
|
|
self.init_profile_manager()
|
|
|
|
# workers get a copy of the config and can miss the broadcast below, so
|
|
# apply both layers here. must stay after init_profile_manager(), which
|
|
# snapshots the base config that profile deactivation resets to
|
|
self.profile_manager.restore_persisted_profile_to_config()
|
|
self.dispatcher.reapply_runtime_state_to_config()
|
|
|
|
self.init_embeddings_client()
|
|
self.start_video_output_processor()
|
|
self.start_ptz_autotracker()
|
|
self.start_detected_frames_processor()
|
|
self.start_camera_processor()
|
|
self.start_audio_processor()
|
|
self.start_storage_maintainer()
|
|
self.start_stats_emitter()
|
|
self.start_timeline_processor()
|
|
self.start_event_processor()
|
|
self.start_event_cleanup()
|
|
self.start_record_cleanup()
|
|
self.start_watchdog()
|
|
|
|
# publish for the recording/review/embeddings processes, which start
|
|
# before the config can be corrected, and for the retained MQTT states
|
|
self.profile_manager.restore_persisted_profile()
|
|
self.dispatcher.restore_runtime_state()
|
|
|
|
self.init_auth()
|
|
|
|
try:
|
|
uvicorn.run(
|
|
create_fastapi_app(
|
|
self.config,
|
|
self.db,
|
|
self.embeddings,
|
|
self.detected_frames_processor,
|
|
self.storage_maintainer,
|
|
self.onvif_controller,
|
|
self.stats_emitter,
|
|
self.event_metadata_updater,
|
|
self.inter_config_updater,
|
|
self.replay_manager,
|
|
self.dispatcher,
|
|
self.profile_manager,
|
|
config_holder=self.config_holder,
|
|
notice_registry=self.notice_registry,
|
|
),
|
|
host="127.0.0.1",
|
|
port=5001,
|
|
log_level="error",
|
|
# outlive nginx's 60s upstream keepalive so nginx closes idle connections first
|
|
timeout_keep_alive=75,
|
|
)
|
|
finally:
|
|
self.stop()
|
|
|
|
def stop(self) -> None:
|
|
logger.info("Stopping...")
|
|
|
|
# used by the docker healthcheck
|
|
Path("/dev/shm/.frigate-is-stopping").touch()
|
|
|
|
# Cancel any running motion search jobs before setting stop_event
|
|
stop_all_motion_search_jobs()
|
|
|
|
self.stop_event.set()
|
|
|
|
# set an end_time on entries without an end_time before exiting
|
|
Event.update(
|
|
end_time=datetime.datetime.now().timestamp(), has_snapshot=False
|
|
).where(Event.end_time == None).execute()
|
|
ReviewSegment.update(end_time=datetime.datetime.now().timestamp()).where(
|
|
ReviewSegment.end_time == None
|
|
).execute()
|
|
|
|
# stop the audio process
|
|
if self.audio_process:
|
|
self.audio_process.terminate()
|
|
self.audio_process.join()
|
|
|
|
# stop the onvif controller
|
|
if self.onvif_controller:
|
|
self.onvif_controller.close()
|
|
|
|
# ensure the detectors are done
|
|
for detector in self.detectors.values():
|
|
detector.stop()
|
|
|
|
for detection_queue in self.detection_queues.values():
|
|
empty_and_close_queue(detection_queue)
|
|
|
|
logger.info("Detection queues closed")
|
|
|
|
self.detected_frames_processor.join()
|
|
empty_and_close_queue(self.detected_frames_queue)
|
|
logger.info("Detected frames queue closed")
|
|
|
|
self.timeline_processor.join()
|
|
self.event_processor.join()
|
|
empty_and_close_queue(self.timeline_queue)
|
|
logger.info("Timeline queue closed")
|
|
|
|
self.output_processor.terminate()
|
|
self.output_processor.join()
|
|
|
|
self.recording_process.terminate()
|
|
self.recording_process.join()
|
|
|
|
self.review_segment_process.terminate()
|
|
self.review_segment_process.join()
|
|
|
|
self.dispatcher.stop()
|
|
self.ptz_autotracker_thread.join()
|
|
|
|
self.event_cleanup.join()
|
|
self.record_cleanup.join()
|
|
self.stats_emitter.join()
|
|
self.frigate_watchdog.join()
|
|
self.camera_maintainer.join()
|
|
self.db.stop()
|
|
|
|
# Save embeddings stats to disk
|
|
if self.embeddings:
|
|
self.embeddings.stop()
|
|
|
|
# Stop Communicators
|
|
self.inter_process_communicator.stop()
|
|
self.inter_config_updater.stop()
|
|
self.event_metadata_updater.stop()
|
|
self.inter_zmq_proxy.stop()
|
|
self.detection_proxy.stop()
|
|
|
|
while len(self.detection_shms) > 0:
|
|
shm = self.detection_shms.pop()
|
|
shm.close()
|
|
shm.unlink()
|
|
|
|
_stop_logging()
|
|
self.metrics_manager.shutdown()
|