administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-08 15:52:48 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Dynamically install and load detector dependencies (#24156)

Browse Source 
* Dynamically install and load detector dependencies

* Cleanup

* Cleanup
This commit is contained in:
Nicolas Mowen
2026-09-12 07:30:04 -06:00
parent 53b04f44ef
commit d183f03fee
18 changed files with 1386 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/app.py
+23
View File
@@ -50,6 +50,7 @@ from frigate.debug_replay import (
cleanup_replay_cameras,
)
from frigate.detectors.detector_config import SceneEnum
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
@@ -88,6 +89,7 @@ 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
@@ -361,6 +363,26 @@ class FrigateApp:
)
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[SceneEnum, list[str]] = {
model.scene: [] for model in self.config.models
@@ -590,6 +612,7 @@ class FrigateApp:
# Ensure global state.
self.ensure_dirs()
self.ensure_detector_dependencies()
# Set soft file limits.
set_file_limit()
frigate/detectors/detection_api.py
+23
View File
@@ -1,9 +1,11 @@
import logging
from abc import ABC, abstractmethod
from typing import ClassVar
import numpy as np
from frigate.detectors.detector_config import BaseDetectorConfig, ModelTypeEnum
from frigate.util.runtime_deps import RuntimeManifest, activate, ensure_installed
logger = logging.getLogger(__name__)
@@ -12,6 +14,10 @@ class DetectionApi(ABC):
type_key: str
supported_models: list[ModelTypeEnum]
# pinned SDK artifacts that are installed at runtime instead of being
# shipped in the image; None when the runtime is already available
runtime_manifest: ClassVar[RuntimeManifest | None] = None
@abstractmethod
def __init__(self, detector_config: BaseDetectorConfig):
self.detector_config = detector_config
@@ -23,6 +29,23 @@ class DetectionApi(ABC):
def detect_raw(self, tensor_input):
pass
@classmethod
def ensure_dependencies(cls) -> None:
"""Download and install this detector's runtime if it is not present.
Runs once in the main process before detector processes start, so a
single install serves every process and the user site is on sys.path
before it is inherited.
"""
if cls.runtime_manifest is not None:
ensure_installed(cls.runtime_manifest)
@classmethod
def activate_dependencies(cls) -> None:
"""Make the installed runtime importable in the current process."""
if cls.runtime_manifest is not None:
activate(cls.runtime_manifest)
def calculate_grids_strides(self, expanded=True) -> None:
grids = []
expanded_strides = []
frigate/detectors/plugins/axengine.py
+24 -1
View File
@@ -10,11 +10,28 @@ from frigate.const import MODEL_CACHE_DIR
from frigate.detectors.detection_api import DetectionApi
from frigate.detectors.detector_config import BaseDetectorConfig, ModelTypeEnum
from frigate.util.model import post_process_yolo
from frigate.util.runtime_deps import Artifact, ArtifactKind, RuntimeManifest
logger = logging.getLogger(__name__)
DETECTOR_KEY = "axengine"
# The AXEngine python package is installed at first start rather than shipped
# in the image; its native libraries are bind mounted from the host.
AXENGINE_VERSION = "0.1.3"
AXENGINE_MANIFEST = RuntimeManifest(
name=DETECTOR_KEY,
version=AXENGINE_VERSION,
artifacts=(
Artifact(
url=f"https://github.com/AXERA-TECH/pyaxengine/releases/download/{AXENGINE_VERSION}-frigate/axengine-{AXENGINE_VERSION}-py3-none-any.whl",
sha256="e995b8a887b067dc3456512aae2fa9c84f70e708c28b11caf184efdc254c64ae",
kind=ArtifactKind.wheel,
),
),
import_check="axengine",
)
supported_models = {
ModelTypeEnum.yologeneric: "frigate-yolov9-.*$",
}
@@ -34,12 +51,18 @@ class AxengineDetectorConfig(BaseDetectorConfig):
class Axengine(DetectionApi):
type_key = DETECTOR_KEY
runtime_manifest = AXENGINE_MANIFEST
def __init__(self, config: AxengineDetectorConfig):
self.activate_dependencies()
try:
import axengine as axe
except ModuleNotFoundError:
raise ImportError("AXEngine is not installed.") from None
raise ImportError(
"AXEngine is not installed. Frigate installs it at startup when an "
"axengine detector is configured; check the startup log for errors."
) from None
logger.info("__init__ axengine")
super().__init__(config)
frigate/detectors/plugins/hailo8l.py
+62 -2
View File
@@ -16,6 +16,14 @@ from frigate.detectors.detector_config import (
BaseDetectorConfig,
)
from frigate.object_detection.util import RequestStore, ResponseStore
from frigate.util.runtime_deps import (
ArchiveDest,
ArchiveMapping,
Artifact,
ArtifactKind,
RuntimeManifest,
find_tool,
)
logger = logging.getLogger(__name__)
@@ -52,11 +60,58 @@ H8L_DEFAULT_MODEL = "yolov6n.hef"
H8_DEFAULT_URL = "https://hailo-model-zoo.s3.eu-west-2.amazonaws.com/ModelZoo/Compiled/v2.14.0/hailo8/yolov6n.hef"
H8L_DEFAULT_URL = "https://hailo-model-zoo.s3.eu-west-2.amazonaws.com/ModelZoo/Compiled/v2.14.0/hailo8l/yolov6n.hef"
# HailoRT is installed at first start rather than shipped in the image. The
# tarball carries libhailort and hailortcli, the wheel the python bindings.
HAILORT_VERSION = "4.21.0"
HAILORT_RELEASE = (
f"https://github.com/frigate-nvr/hailort/releases/download/v{HAILORT_VERSION}"
)
HAILORT_MAPPINGS = (
ArchiveMapping("rootfs/usr/local/lib/", ArchiveDest.lib),
ArchiveMapping("rootfs/usr/local/bin/", ArchiveDest.bin),
)
HAILORT_MANIFEST = RuntimeManifest(
name=DETECTOR_KEY,
version=HAILORT_VERSION,
artifacts=(
Artifact(
url=f"{HAILORT_RELEASE}/hailort-debian12-amd64.tar.gz",
sha256="0a57ac5f7cc8c2c3668133189d9285b55f498e8cb219797e203f6f5015fec4b3",
kind=ArtifactKind.archive,
mappings=HAILORT_MAPPINGS,
machines=("x86_64",),
),
Artifact(
url=f"{HAILORT_RELEASE}/hailort-debian12-arm64.tar.gz",
sha256="dd840548eb5d0d147c99aee2cb013d39d64be09c5bc63061171fcfacf4547b3f",
kind=ArtifactKind.archive,
mappings=HAILORT_MAPPINGS,
machines=("aarch64",),
),
Artifact(
url=f"{HAILORT_RELEASE}/hailort-{HAILORT_VERSION}-cp311-cp311-linux_x86_64.whl",
sha256="8112a973ab48095399b29d883f31987828df5861b8553f614c89f098a67b3fb6",
kind=ArtifactKind.wheel,
machines=("x86_64",),
),
Artifact(
url=f"{HAILORT_RELEASE}/hailort-{HAILORT_VERSION}-cp311-cp311-linux_aarch64.whl",
sha256="658432a43573280d472f6402d7934669effe7f163ba3dffa31c50bbeeaa7c01d",
kind=ArtifactKind.wheel,
machines=("aarch64",),
),
),
preload=(f"libhailort.so.{HAILORT_VERSION}",),
import_check="hailo_platform",
)
def detect_hailo_arch():
try:
result = subprocess.run(
["hailortcli", "fw-control", "identify"], capture_output=True, text=True
[find_tool("hailortcli"), "fw-control", "identify"],
capture_output=True,
text=True,
)
if result.returncode != 0:
logger.error(f"Inference error: {result.stderr}")
@@ -96,7 +151,10 @@ class HailoAsyncInference:
VDevice,
)
except ModuleNotFoundError:
pass
raise ImportError(
"HailoRT is not installed. Frigate installs it at startup when a "
"Hailo detector is configured; check the startup log for errors."
) from None
self.input_store = input_store
self.output_store = output_store
@@ -202,9 +260,11 @@ class HailoAsyncInference:
# ----------------- HailoDetector Class ----------------- #
class HailoDetector(DetectionApi):
type_key = DETECTOR_KEY
runtime_manifest = HAILORT_MANIFEST
def __init__(self, detector_config: "HailoDetectorConfig"):
global ARCH
self.activate_dependencies()
ARCH = detect_hailo_arch()
self.cache_dir = MODEL_CACHE_DIR
self.device_type = detector_config.device
frigate/detectors/plugins/memryx.py
+53 -3
View File
@@ -18,11 +18,58 @@ from frigate.detectors.detector_config import (
)
from frigate.util.file import FileLock
from frigate.util.model import xyxy_to_xywh_for_nms
from frigate.util.runtime_deps import (
ArchiveDest,
ArchiveMapping,
Artifact,
ArtifactKind,
RuntimeManifest,
)
logger = logging.getLogger(__name__)
DETECTOR_KEY = "memryx"
# The MemryX SDK is installed at first start rather than shipped in the image.
# The release archive holds the python package with its compiled extensions
# and, per architecture, the shared libraries they link against. libmemx has
# no SONAME, so the libraries are resolved through LD_LIBRARY_PATH.
MEMRYX_VERSION = "2.1.0"
MEMRYX_ARCHIVE_ROOT = f"mx_accl_frigate-{MEMRYX_VERSION}/memryx/"
MEMRYX_LIBS = ("libmemx.so*", "libmx_accl.so*")
MEMRYX_MANIFEST = RuntimeManifest(
name=DETECTOR_KEY,
version=MEMRYX_VERSION,
artifacts=(
Artifact(
url=f"https://github.com/memryx/mx_accl_frigate/archive/refs/tags/v{MEMRYX_VERSION}.zip",
sha256="54d1971dee54688541561b7bd9136342be51707b68c11a2a5e575fa19e69bc51",
kind=ArtifactKind.archive,
filename=f"mx_accl_frigate-{MEMRYX_VERSION}.zip",
mappings=(
ArchiveMapping(
MEMRYX_ARCHIVE_ROOT, ArchiveDest.site_packages, subdir="memryx"
),
ArchiveMapping(
f"{MEMRYX_ARCHIVE_ROOT}x86/",
ArchiveDest.lib,
include=MEMRYX_LIBS,
machines=("x86_64",),
),
ArchiveMapping(
f"{MEMRYX_ARCHIVE_ROOT}arm/",
ArchiveDest.lib,
include=MEMRYX_LIBS,
machines=("aarch64",),
),
),
),
),
preload=("libmemx.so", "libmx_accl.so.2"),
import_check="memryx",
needs_ld_library_path=True,
)
# Configuration class for model settings
class ModelConfig(BaseModel):
@@ -56,17 +103,20 @@ class MemryXDetector(DetectionApi):
ModelTypeEnum.yologeneric, # Treated as yolov9 in MemryX implementation
ModelTypeEnum.yolox,
]
runtime_manifest = MEMRYX_MANIFEST
def __init__(self, detector_config):
"""Initialize MemryX detector with the provided configuration."""
self.activate_dependencies()
try:
# Import MemryX SDK
from memryx import AsyncAccl
except ModuleNotFoundError:
raise ImportError(
"MemryX SDK is not installed. Install it and set up MIX environment."
"MemryX SDK is not installed. Frigate installs it at startup when "
"a MemryX detector is configured; check the startup log for errors."
) from None
return
# Initialize stop_event as None, will be set later by set_stop_event()
self.stop_event = None
@@ -78,7 +128,7 @@ class MemryXDetector(DetectionApi):
detector_config.model.model_type = model_cfg.model_type
else:
logger.info(
"model_type not set in config — defaulting to yolonas for MemryX."
"model_type not set in config, defaulting to yolonas for MemryX"
)
detector_config.model.model_type = ModelTypeEnum.yolonas
frigate/test/test_runtime_deps.py
+563
View File
@@ -0,0 +1,563 @@
"""Tests for runtime installation of detector SDKs into the user site."""
import ctypes
import hashlib
import io
import json
import os
import stat
import subprocess
import sys
import tarfile
import tempfile
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch
from frigate.util import runtime_deps
from frigate.util.runtime_deps import (
ArchiveDest,
ArchiveMapping,
Artifact,
ArtifactKind,
RuntimeDependencyError,
RuntimeManifest,
activate,
ensure_installed,
find_tool,
)
LIB_PREFIX = "rootfs/usr/local/lib/"
BIN_PREFIX = "rootfs/usr/local/bin/"
MAPPINGS = (
ArchiveMapping(LIB_PREFIX, ArchiveDest.lib),
ArchiveMapping(BIN_PREFIX, ArchiveDest.bin),
)
def sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def make_tarball(members: dict[str, bytes | str], exec_names=()) -> bytes:
"""Build a tar.gz; a str value is a symlink target, bytes a file."""
buffer = io.BytesIO()
with tarfile.open(fileobj=buffer, mode="w:gz") as archive:
for name, content in members.items():
info = tarfile.TarInfo(name)
if isinstance(content, str):
info.type = tarfile.SYMTYPE
info.linkname = content
archive.addfile(info)
else:
info.size = len(content)
info.mode = 0o755 if name in exec_names else 0o644
archive.addfile(info, io.BytesIO(content))
return buffer.getvalue()
def make_zip(members: dict[str, bytes | str]) -> bytes:
"""Build a zip the way GitHub archives do, symlinks included."""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as archive:
for name, content in members.items():
info = zipfile.ZipInfo(name)
if isinstance(content, str):
info.external_attr = (stat.S_IFLNK | 0o777) << 16
archive.writestr(info, content)
else:
info.external_attr = (stat.S_IFREG | 0o644) << 16
archive.writestr(info, content)
return buffer.getvalue()
class RuntimeDepsTestCase(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.TemporaryDirectory()
root = Path(self.tmp.name)
self.base = root / "base"
self.site = self.base / "lib" / "python" / "site-packages"
self.cache = root / "cache"
self.downloads: list[str] = []
self.download_content: bytes | None = None
self.sys_path = list(sys.path)
runtime_deps._loaded_libs.clear()
def download(url: str, save_path: str, silent: bool = False) -> Path:
self.downloads.append(url)
if self.download_content is None:
raise OSError("network is off")
Path(save_path).write_bytes(self.download_content)
return Path(save_path)
self.patches = [
patch.object(runtime_deps, "user_base", lambda: self.base),
patch.object(runtime_deps, "user_site", lambda: self.site),
patch.object(runtime_deps, "cache_dir", lambda name: self.cache / name),
patch.object(runtime_deps.site, "ENABLE_USER_SITE", True),
patch(
"frigate.util.downloader.ModelDownloader.download_from_url", download
),
patch("os.geteuid", return_value=1000),
patch.dict("os.environ", {}, clear=True),
]
for p in self.patches:
p.start()
def tearDown(self) -> None:
for p in reversed(self.patches):
p.stop()
sys.path[:] = self.sys_path
self.tmp.cleanup()
def archive_manifest(
self, data: bytes, version: str = "1.0", name: str = "test"
) -> RuntimeManifest:
return RuntimeManifest(
name=name,
version=version,
artifacts=(
Artifact(
url=f"https://github.com/org/repo/releases/download/v{version}/runtime.tar.gz",
sha256=sha256(data),
kind=ArtifactKind.archive,
mappings=MAPPINGS,
),
),
)
def preseed(self, manifest: RuntimeManifest, data: bytes) -> Path:
path = self.cache / manifest.name / manifest.artifacts[0].name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
return path
class TestFetch(RuntimeDepsTestCase):
def test_preseeded_file_with_matching_sha256_skips_download(self) -> None:
data = make_tarball({f"{LIB_PREFIX}libfoo.so": b"lib"})
manifest = self.archive_manifest(data)
self.preseed(manifest, data)
ensure_installed(manifest)
self.assertEqual(self.downloads, [])
self.assertEqual((self.base / "lib" / "libfoo.so").read_bytes(), b"lib")
def test_preseeded_file_with_wrong_sha256_is_downloaded_again(self) -> None:
data = make_tarball({f"{LIB_PREFIX}libfoo.so": b"lib"})
manifest = self.archive_manifest(data)
self.preseed(manifest, b"corrupt")
self.download_content = data
ensure_installed(manifest)
self.assertEqual(len(self.downloads), 1)
self.assertTrue((self.base / "lib" / "libfoo.so").is_file())
def test_download_with_wrong_sha256_raises_and_removes_file(self) -> None:
data = make_tarball({f"{LIB_PREFIX}libfoo.so": b"lib"})
manifest = self.archive_manifest(data)
self.download_content = b"tampered"
with self.assertRaises(RuntimeDependencyError):
ensure_installed(manifest)
self.assertFalse((self.cache / "test" / "runtime.tar.gz").exists())
self.assertFalse((self.base / "lib" / "libfoo.so").exists())
def test_download_failure_names_the_preseed_directory(self) -> None:
manifest = self.archive_manifest(b"whatever")
with self.assertRaises(RuntimeDependencyError) as ctx:
ensure_installed(manifest)
self.assertIn(str(self.cache / "test"), str(ctx.exception))
def test_github_endpoint_mirror_is_honored(self) -> None:
data = make_tarball({f"{LIB_PREFIX}libfoo.so": b"lib"})
manifest = self.archive_manifest(data)
self.download_content = data
with patch.dict("os.environ", {"GITHUB_ENDPOINT": "https://mirror.test/"}):
ensure_installed(manifest)
self.assertEqual(
self.downloads,
["https://mirror.test/org/repo/releases/download/v1.0/runtime.tar.gz"],
)
def test_non_github_urls_are_left_alone(self) -> None:
with patch.dict("os.environ", {"GITHUB_ENDPOINT": "https://mirror.test"}):
self.assertEqual(
runtime_deps.resolve_url("https://example.com/a.whl"),
"https://example.com/a.whl",
)
class TestInstall(RuntimeDepsTestCase):
def test_archive_extracts_lib_and_bin_with_exec_bits_and_symlinks(self) -> None:
data = make_tarball(
{
f"{LIB_PREFIX}libfoo.so.1": b"lib",
f"{LIB_PREFIX}libfoo.so": "libfoo.so.1",
f"{BIN_PREFIX}tool": b"#!/bin/sh\n",
"rootfs/etc/ignored": b"no",
},
exec_names=(f"{BIN_PREFIX}tool",),
)
manifest = self.archive_manifest(data)
self.preseed(manifest, data)
ensure_installed(manifest)
lib = self.base / "lib"
self.assertEqual((lib / "libfoo.so.1").read_bytes(), b"lib")
self.assertEqual(os.readlink(lib / "libfoo.so"), "libfoo.so.1")
self.assertTrue(os.access(self.base / "bin" / "tool", os.X_OK))
self.assertFalse((self.base / "etc").exists())
stamp = json.loads(
(self.base / "share" / "frigate" / "runtimes" / "test.json").read_text()
)
self.assertEqual(stamp["version"], "1.0")
self.assertEqual(len(stamp["files"]), 3)
def test_zip_archive_with_include_filter_and_site_packages_dest(self) -> None:
data = make_zip(
{
"pkg-1.0/pkg/__init__.py": b"",
"pkg-1.0/pkg/x86/libx.so.2": b"lib",
"pkg-1.0/pkg/x86/libx.so": "libx.so.2",
"pkg-1.0/pkg/x86/other.txt": b"skip",
}
)
manifest = RuntimeManifest(
name="test",
version="1.0",
artifacts=(
Artifact(
url="https://github.com/org/pkg/archive/refs/tags/v1.0.zip",
sha256=sha256(data),
kind=ArtifactKind.archive,
filename="pkg-1.0.zip",
mappings=(
ArchiveMapping(
"pkg-1.0/pkg/", ArchiveDest.site_packages, subdir="pkg"
),
ArchiveMapping(
"pkg-1.0/pkg/x86/", ArchiveDest.lib, include=("libx.so*",)
),
),
),
),
)
self.preseed(manifest, data)
ensure_installed(manifest)
self.assertTrue((self.site / "pkg" / "__init__.py").is_file())
self.assertEqual(os.readlink(self.base / "lib" / "libx.so"), "libx.so.2")
self.assertTrue((self.base / "lib" / "libx.so.2").is_file())
self.assertFalse((self.base / "lib" / "other.txt").exists())
def test_archive_member_escaping_destination_is_rejected(self) -> None:
data = make_tarball({f"{LIB_PREFIX}../../../../escaped": b"evil"})
manifest = self.archive_manifest(data)
self.preseed(manifest, data)
with self.assertRaises(RuntimeDependencyError):
ensure_installed(manifest)
self.assertFalse((Path(self.tmp.name) / "escaped").exists())
def test_symlink_pointing_outside_its_directory_is_rejected(self) -> None:
data = make_tarball({f"{LIB_PREFIX}libfoo.so": "../../etc/passwd"})
manifest = self.archive_manifest(data)
self.preseed(manifest, data)
with self.assertRaises(RuntimeDependencyError):
ensure_installed(manifest)
def test_wheel_is_installed_with_pip_into_the_user_site(self) -> None:
data = b"not really a wheel"
manifest = RuntimeManifest(
name="test",
version="1.0",
artifacts=(
Artifact(
url="https://example.com/thing-1.0-py3-none-any.whl",
sha256=sha256(data),
kind=ArtifactKind.wheel,
),
),
)
self.preseed(manifest, data)
completed = subprocess.CompletedProcess([], 0, stdout="", stderr="")
with patch.object(
runtime_deps.subprocess, "run", return_value=completed
) as run:
ensure_installed(manifest)
args = run.call_args.args[0]
self.assertEqual(args[:4], [sys.executable, "-m", "pip", "install"])
self.assertIn("--user", args)
self.assertIn("--no-index", args)
self.assertIn("--no-deps", args)
self.assertTrue(args[-1].endswith("thing-1.0-py3-none-any.whl"))
# the staged copy is installed, not the one in the cache
self.assertFalse(args[-1].startswith(str(self.cache)))
def test_failed_pip_install_raises(self) -> None:
data = b"wheel"
manifest = RuntimeManifest(
name="test",
version="1.0",
artifacts=(
Artifact(
url="https://example.com/thing-1.0-py3-none-any.whl",
sha256=sha256(data),
kind=ArtifactKind.wheel,
),
),
)
self.preseed(manifest, data)
completed = subprocess.CompletedProcess([], 1, stdout="", stderr="boom")
with (
patch.object(runtime_deps.subprocess, "run", return_value=completed),
self.assertRaises(RuntimeDependencyError),
):
ensure_installed(manifest)
def test_matching_stamp_skips_a_second_install(self) -> None:
data = make_tarball({f"{LIB_PREFIX}libfoo.so": b"lib"})
manifest = self.archive_manifest(data)
self.preseed(manifest, data)
ensure_installed(manifest)
with patch.object(runtime_deps, "_extract_archive") as extract:
ensure_installed(manifest)
extract.assert_not_called()
def test_missing_installed_file_triggers_a_reinstall(self) -> None:
data = make_tarball({f"{LIB_PREFIX}libfoo.so": b"lib"})
manifest = self.archive_manifest(data)
self.preseed(manifest, data)
ensure_installed(manifest)
(self.base / "lib" / "libfoo.so").unlink()
ensure_installed(manifest)
self.assertTrue((self.base / "lib" / "libfoo.so").is_file())
def test_version_bump_reinstalls_and_removes_old_files(self) -> None:
old = make_tarball({f"{LIB_PREFIX}libfoo.so.1": b"old"})
old_manifest = self.archive_manifest(old, version="1.0")
self.preseed(old_manifest, old)
ensure_installed(old_manifest)
new = make_tarball({f"{LIB_PREFIX}libfoo.so.2": b"new"})
new_manifest = self.archive_manifest(new, version="2.0")
self.preseed(new_manifest, new)
ensure_installed(new_manifest)
self.assertFalse((self.base / "lib" / "libfoo.so.1").exists())
self.assertEqual((self.base / "lib" / "libfoo.so.2").read_bytes(), b"new")
def test_only_artifacts_for_the_current_machine_are_used(self) -> None:
x86 = make_tarball({f"{LIB_PREFIX}libx86.so": b"x"})
arm = make_tarball({f"{LIB_PREFIX}libarm.so": b"a"})
manifest = RuntimeManifest(
name="test",
version="1.0",
artifacts=(
Artifact(
url="https://example.com/x86.tar.gz",
sha256=sha256(x86),
kind=ArtifactKind.archive,
mappings=MAPPINGS,
machines=("x86_64",),
),
Artifact(
url="https://example.com/arm.tar.gz",
sha256=sha256(arm),
kind=ArtifactKind.archive,
mappings=MAPPINGS,
machines=("aarch64",),
),
),
)
(self.cache / "test").mkdir(parents=True)
(self.cache / "test" / "x86.tar.gz").write_bytes(x86)
(self.cache / "test" / "arm.tar.gz").write_bytes(arm)
with patch.object(runtime_deps.platform, "machine", return_value="aarch64"):
ensure_installed(manifest)
self.assertTrue((self.base / "lib" / "libarm.so").is_file())
self.assertFalse((self.base / "lib" / "libx86.so").exists())
class TestRootGuard(RuntimeDepsTestCase):
"""A root frigate service must never load code from a uid-1000 tree."""
def test_refuses_a_runtime_user_writable_base_under_granular_root(self) -> None:
manifest = self.archive_manifest(b"data")
with (
patch.object(runtime_deps, "user_base", lambda: Path("/config/.local")),
patch("os.geteuid", return_value=0),
patch.dict("os.environ", {"FRIGATE_ROOT_SERVICES": "frigate"}),
):
with self.assertRaises(RuntimeDependencyError) as ctx:
ensure_installed(manifest)
self.assertIn("FRIGATE_ROOT_SERVICES", str(ctx.exception))
self.assertIsNotNone(runtime_deps._usable_reason())
def test_root_owned_base_is_allowed_under_granular_root(self) -> None:
with (
patch.object(runtime_deps, "user_base", lambda: Path("/root/.local")),
patch("os.geteuid", return_value=0),
patch.dict("os.environ", {"FRIGATE_ROOT_SERVICES": "frigate"}),
):
self.assertIsNone(runtime_deps._usable_reason())
def test_escape_hatch_is_not_guarded(self) -> None:
with (
patch.object(runtime_deps, "user_base", lambda: Path("/config/.local")),
patch("os.geteuid", return_value=0),
patch.dict("os.environ", {"FRIGATE_RUN_AS_ROOT": "true"}),
):
self.assertIsNone(runtime_deps._usable_reason())
def test_disabled_user_site_is_refused(self) -> None:
with patch.object(runtime_deps.site, "ENABLE_USER_SITE", False):
self.assertIsNotNone(runtime_deps._usable_reason())
def test_activate_skips_a_refused_runtime(self) -> None:
self.site.mkdir(parents=True)
with (
patch.object(runtime_deps, "user_base", lambda: Path("/config/.local")),
patch("os.geteuid", return_value=0),
patch.dict("os.environ", {"FRIGATE_ROOT_SERVICES": "frigate"}),
):
activate(RuntimeManifest(name="test", version="1", artifacts=()))
self.assertNotIn(str(self.site), sys.path)
class TestActivate(RuntimeDepsTestCase):
def test_adds_the_user_site_ahead_of_system_site_packages(self) -> None:
self.site.mkdir(parents=True)
activate(RuntimeManifest(name="test", version="1", artifacts=()))
self.assertIn(str(self.site), sys.path)
system = [
i
for i, p in enumerate(sys.path)
if p.endswith(("site-packages", "dist-packages")) and p != str(self.site)
]
if system:
self.assertLess(sys.path.index(str(self.site)), system[0])
def test_preloads_libraries_in_order_with_rtld_global(self) -> None:
lib = self.base / "lib"
lib.mkdir(parents=True)
(lib / "liba.so").write_bytes(b"")
(lib / "libb.so").write_bytes(b"")
manifest = RuntimeManifest(
name="test", version="1", artifacts=(), preload=("liba.so", "libb.so")
)
with patch.object(runtime_deps.ctypes, "CDLL") as cdll:
activate(manifest)
activate(manifest)
self.assertEqual(
[c.args[0] for c in cdll.call_args_list],
[str(lib / "liba.so"), str(lib / "libb.so")],
)
self.assertTrue(
all(c.kwargs["mode"] == ctypes.RTLD_GLOBAL for c in cdll.call_args_list)
)
def test_missing_ld_library_path_is_reported(self) -> None:
manifest = RuntimeManifest(
name="test", version="1", artifacts=(), needs_ld_library_path=True
)
with self.assertLogs(runtime_deps.logger, level="WARNING") as logs:
activate(manifest)
self.assertTrue(any("LD_LIBRARY_PATH" in line for line in logs.output))
with (
patch.dict(
"os.environ", {"LD_LIBRARY_PATH": f"/usr/lib:{self.base / 'lib'}"}
),
self.assertNoLogs(runtime_deps.logger, level="WARNING"),
):
activate(manifest)
def test_find_tool_prefers_the_user_bin(self) -> None:
tool = self.base / "bin" / "sh"
tool.parent.mkdir(parents=True)
tool.write_text("#!/bin/sh\n")
tool.chmod(0o755)
self.assertEqual(find_tool("sh"), str(tool))
self.assertEqual(find_tool("definitely-not-a-tool"), "definitely-not-a-tool")
class TestDeclaredManifests(unittest.TestCase):
"""Every manifest a detector declares must be complete and pinned."""
def test_manifests_are_pinned_for_both_architectures(self) -> None:
from frigate.detectors import api_types
manifests = [
api.runtime_manifest
for api in api_types.values()
if api.runtime_manifest is not None
]
self.assertGreaterEqual(len(manifests), 3)
for manifest in manifests:
with self.subTest(manifest=manifest.name):
self.assertTrue(manifest.version)
self.assertTrue(manifest.import_check)
for artifact in manifest.artifacts:
self.assertRegex(artifact.sha256, r"^[0-9a-f]{64}$")
self.assertTrue(artifact.url.startswith("https://"))
for machine in ("x86_64", "aarch64"):
self.assertTrue(
any(
not a.machines or machine in a.machines
for a in manifest.artifacts
),
f"no artifact for {machine}",
)
if __name__ == "__main__":
unittest.main()
frigate/util/config.py
+2 -2
View File
@@ -69,7 +69,7 @@ def frigate_service_is_granular_root() -> bool:
return any("".join(entry.split()) == "frigate" for entry in entries)
def _is_runtime_user_writable(path: str) -> bool:
def is_runtime_user_writable(path: str) -> bool:
"""Report whether a path resolves inside a runtime-user-writable tree."""
resolved = os.path.realpath(path)
return any(
@@ -104,7 +104,7 @@ def resolve_ffmpeg_path(path: str, binary: str = "ffmpeg") -> str:
elif path in INCLUDED_FFMPEG_VERSIONS:
version = path
else:
if not (frigate_service_is_granular_root() and _is_runtime_user_writable(path)):
if not (frigate_service_is_granular_root() and is_runtime_user_writable(path)):
return f"{path}/bin/{binary}"
_warn_ignored_ffmpeg_path(path)
frigate/util/rknn_converter.py
+35 -21
View File
@@ -1,7 +1,9 @@
"""RKNN model conversion utility for Frigate."""
import importlib
import logging
import os
import site
import subprocess
import sys
import time
@@ -101,30 +103,42 @@ def ensure_torch_dependencies() -> bool:
except ImportError:
logger.info("PyTorch not found, attempting to install...")
try:
subprocess.check_call(
[
sys.executable,
"-m",
"pip",
"install",
"--break-system-packages",
"setuptools<81",
"torch",
"torchvision",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
result = subprocess.run(
[
sys.executable,
"-m",
"pip",
"install",
"--break-system-packages",
"setuptools<81",
"torch",
"torchvision",
],
capture_output=True,
text=True,
)
import torch # type: ignore # noqa: F401
logger.info("PyTorch installed successfully")
return True
except (subprocess.CalledProcessError, ImportError) as e:
logger.error(f"Failed to install PyTorch: {e}")
if result.returncode != 0:
logger.error("Failed to install PyTorch:\n%s", result.stderr[-4000:])
return False
# as an unprivileged user pip falls back to the user site, which is
# only on sys.path at startup if it already existed
user_site = site.getusersitepackages()
if os.path.isdir(user_site) and user_site not in sys.path:
site.addsitedir(user_site)
importlib.invalidate_caches()
try:
import torch # type: ignore # noqa: F401
except ImportError as e:
logger.error(f"Failed to import PyTorch after installing it: {e}")
return False
logger.info("PyTorch installed successfully")
return True
def ensure_rknn_toolkit() -> bool:
"""Ensure RKNN toolkit is available."""
frigate/util/runtime_deps.py
+545
View File
@@ -0,0 +1,545 @@
"""Runtime installation of optional accelerator SDKs into the user site.
Detector runtimes that are only useful with specific hardware are not shipped
in the image. A detector declares a RuntimeManifest of pinned, checksummed
artifacts, and ensure_installed() fetches and installs them into the runtime
user's home (pip's --user location) the first time that detector is
configured. Everything is derived from site.getuserbase() so the location
follows $HOME: /config/.local for the unprivileged service, /root/.local for
a root service, never a mix of the two.
"""
import ctypes
import fnmatch
import hashlib
import importlib
import importlib.util
import json
import logging
import os
import platform
import shutil
import site
import stat
import subprocess
import sys
import tarfile
import tempfile
import zipfile
from dataclasses import dataclass
from enum import StrEnum
from pathlib import Path
from frigate.const import MODEL_CACHE_DIR
from frigate.util.config import (
frigate_service_is_granular_root,
is_runtime_user_writable,
)
from frigate.util.file import FileLock
logger = logging.getLogger(__name__)
GITHUB_URL = "https://github.com"
DOCS_URL = "https://docs.frigate.video/frigate/network_requirements#detector-runtimes"
class RuntimeDependencyError(RuntimeError):
"""A runtime could not be installed or is not allowed to be used."""
class ArtifactKind(StrEnum):
wheel = "wheel"
archive = "archive"
class ArchiveDest(StrEnum):
lib = "lib"
bin = "bin"
site_packages = "site_packages"
@dataclass(frozen=True)
class ArchiveMapping:
"""Copy archive members under `prefix` into a directory of the user base.
`include` holds optional basename globs; when set, only matching members
are extracted. `machines` restricts the mapping to platform.machine()
values, empty meaning all.
"""
prefix: str
dest: ArchiveDest
subdir: str = ""
include: tuple[str, ...] = ()
machines: tuple[str, ...] = ()
@dataclass(frozen=True)
class Artifact:
url: str
sha256: str
kind: ArtifactKind
filename: str | None = None
mappings: tuple[ArchiveMapping, ...] = ()
machines: tuple[str, ...] = ()
@property
def name(self) -> str:
return self.filename or os.path.basename(self.url.split("?")[0])
@dataclass(frozen=True)
class RuntimeManifest:
"""The pinned artifacts that make up one detector's runtime.
`preload` lists shared libraries under <user base>/lib to load with
RTLD_GLOBAL before the SDK import, in dependency order. Libraries without
a SONAME cannot be satisfied that way and need <user base>/lib on
LD_LIBRARY_PATH at exec time; `needs_ld_library_path` makes activate()
warn when it is missing.
"""
name: str
version: str
artifacts: tuple[Artifact, ...]
preload: tuple[str, ...] = ()
import_check: str = ""
needs_ld_library_path: bool = False
def digest(self) -> str:
parts = [self.name, self.version, *sorted(a.sha256 for a in self.artifacts)]
return hashlib.sha256("\n".join(parts).encode()).hexdigest()
_loaded_libs: dict[str, ctypes.CDLL] = {}
def user_base() -> Path:
return Path(site.getuserbase())
def user_site() -> Path:
return Path(site.getusersitepackages())
def cache_dir(name: str) -> Path:
"""The directory downloads land in, and where offline users pre-seed them."""
return Path(MODEL_CACHE_DIR) / "runtimes" / name
def resolve_url(url: str) -> str:
"""Apply the GITHUB_ENDPOINT mirror to GitHub release URLs."""
if url.startswith(f"{GITHUB_URL}/"):
endpoint = os.environ.get("GITHUB_ENDPOINT", GITHUB_URL).rstrip("/")
return f"{endpoint}{url[len(GITHUB_URL) :]}"
return url
def find_tool(name: str) -> str:
"""Resolve a CLI tool, preferring the runtime-installed copy."""
candidate = user_base() / "bin" / name
if candidate.is_file() and os.access(candidate, os.X_OK):
return str(candidate)
return shutil.which(name) or name
def sha256_of(path: Path) -> str:
digest = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _usable_reason() -> str | None:
"""Why the user site must not be used, or None when it may be."""
if not site.ENABLE_USER_SITE:
return "the user site-packages directory is disabled for this interpreter"
base = user_base()
if frigate_service_is_granular_root() and is_runtime_user_writable(str(base)):
return (
f"{base} is writable by the unprivileged user while "
"FRIGATE_ROOT_SERVICES runs frigate as root"
)
return None
def _artifacts_for_machine(manifest: RuntimeManifest) -> list[Artifact]:
machine = platform.machine()
return [a for a in manifest.artifacts if not a.machines or machine in a.machines]
def _fetch(artifact: Artifact, cache: Path) -> Path:
"""Return a verified copy of the artifact in the cache, downloading if needed."""
path = cache / artifact.name
if path.is_file():
if sha256_of(path) == artifact.sha256:
logger.info("Using pre-seeded runtime file %s", path)
return path
logger.warning("Checksum mismatch for %s, downloading again", path)
path.unlink()
# imported here so the detector API does not pull in the IPC stack
from frigate.util.downloader import ModelDownloader
url = resolve_url(artifact.url)
logger.info("Downloading runtime file %s", url)
try:
ModelDownloader.download_from_url(url, str(path), silent=True)
except Exception as err:
raise RuntimeDependencyError(
f"Unable to download {url}: {err}. Without internet access, "
f"download it elsewhere and place it in {cache} (see {DOCS_URL})"
) from err
if sha256_of(path) != artifact.sha256:
path.unlink()
raise RuntimeDependencyError(
f"Checksum mismatch for {url}; the file was removed from {cache}"
)
return path
def _stage(path: Path, artifact: Artifact, staging: Path) -> Path:
"""Copy into the private staging dir and verify there.
The cache directory is writable by the runtime user, so the copy that is
installed is the one that was verified, not whatever is in the cache at
install time.
"""
staged = staging / path.name
shutil.copyfile(path, staged)
if sha256_of(staged) != artifact.sha256:
raise RuntimeDependencyError(f"{path} changed while being installed")
return staged
def _install_wheel(path: Path) -> None:
result = subprocess.run(
[
sys.executable,
"-m",
"pip",
"install",
"--user",
"--no-index",
"--no-deps",
"--force-reinstall",
"--no-warn-script-location",
str(path),
],
capture_output=True,
text=True,
env={**os.environ, "PIP_BREAK_SYSTEM_PACKAGES": "1"},
)
if result.returncode != 0:
logger.error("pip install of %s failed:\n%s", path.name, result.stderr[-4000:])
raise RuntimeDependencyError(f"pip install of {path.name} failed")
logger.debug("pip install of %s:\n%s", path.name, result.stdout)
def _dest_root(mapping: ArchiveMapping) -> Path:
if mapping.dest is ArchiveDest.site_packages:
root = user_site()
else:
root = user_base() / mapping.dest.value
return root / mapping.subdir if mapping.subdir else root
def _mappings_for(
member_name: str, mappings: list[ArchiveMapping]
) -> list[tuple[ArchiveMapping, str]]:
"""Every mapping a member falls under; prefixes may overlap."""
matches = []
for mapping in mappings:
if not member_name.startswith(mapping.prefix):
continue
relative = member_name[len(mapping.prefix) :]
if not relative or relative.endswith("/"):
continue
if mapping.include and not any(
fnmatch.fnmatch(os.path.basename(relative), p) for p in mapping.include
):
continue
matches.append((mapping, relative))
return matches
def _safe_dest(root: Path, relative: str) -> Path:
dest = root / relative
resolved_root = os.path.realpath(root)
if not os.path.realpath(dest).startswith(f"{resolved_root}{os.sep}"):
raise RuntimeDependencyError(f"Archive member {relative} escapes {root}")
return dest
def _write_member(dest: Path, data_source, mode: int) -> None:
dest.parent.mkdir(parents=True, exist_ok=True)
tmp = dest.with_name(f"{dest.name}.tmp")
with open(tmp, "wb") as f:
shutil.copyfileobj(data_source, f)
os.chmod(tmp, (mode & 0o777) or 0o644)
os.replace(tmp, dest)
def _write_symlink(dest: Path, target: str) -> None:
# only links to a sibling file are recreated; anything else is dropped
if "/" in target or target in ("", ".", ".."):
raise RuntimeDependencyError(f"Archive symlink {dest.name} -> {target}")
dest.parent.mkdir(parents=True, exist_ok=True)
dest.unlink(missing_ok=True)
os.symlink(target, dest)
def _extract_archive(path: Path, artifact: Artifact) -> list[str]:
"""Extract mapped members into the user base and return their paths."""
machine = platform.machine()
mappings = [m for m in artifact.mappings if not m.machines or machine in m.machines]
written: list[str] = []
if path.name.endswith(".zip"):
with zipfile.ZipFile(path) as archive:
for info in archive.infolist():
if info.is_dir():
continue
mode = info.external_attr >> 16
for mapping, relative in _mappings_for(info.filename, mappings):
dest = _safe_dest(_dest_root(mapping), relative)
if stat.S_ISLNK(mode):
_write_symlink(dest, archive.read(info).decode())
else:
with archive.open(info) as source:
_write_member(dest, source, mode)
written.append(str(dest))
else:
with tarfile.open(path, "r:*") as archive:
for member in archive:
if not (member.isfile() or member.issym()):
continue
for mapping, relative in _mappings_for(member.name, mappings):
dest = _safe_dest(_dest_root(mapping), relative)
if member.issym():
_write_symlink(dest, member.linkname)
else:
source = archive.extractfile(member)
assert source is not None
with source:
_write_member(dest, source, member.mode)
written.append(str(dest))
if not written:
raise RuntimeDependencyError(f"{path.name} contained no expected files")
return written
def _stamp_path(name: str) -> Path:
return user_base() / "share" / "frigate" / "runtimes" / f"{name}.json"
def _read_stamp(name: str) -> dict | None:
try:
with open(_stamp_path(name)) as f:
stamp = json.load(f)
except (OSError, ValueError):
return None
return stamp if isinstance(stamp, dict) else None
def _write_stamp(manifest: RuntimeManifest, files: list[str]) -> None:
path = _stamp_path(manifest.name)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(
{
"version": manifest.version,
"digest": manifest.digest(),
"files": files,
},
f,
indent=2,
)
def _remove_stamped_files(stamp: dict | None) -> None:
for file in (stamp or {}).get("files", []):
Path(file).unlink(missing_ok=True)
def _is_current(manifest: RuntimeManifest, stamp: dict | None) -> bool:
if (
stamp is None
or stamp.get("version") != manifest.version
or stamp.get("digest") != manifest.digest()
):
return False
if not all(os.path.lexists(f) for f in stamp.get("files", [])):
return False
if manifest.import_check:
_add_user_site()
return importlib.util.find_spec(manifest.import_check) is not None
return True
def _add_user_site() -> None:
"""Put the user site on sys.path the way interpreter startup would.
site.addsitedir() appends, but at startup the user site precedes the
system site-packages, so the entry is moved ahead of them to keep that
precedence.
"""
path = str(user_site())
if path in sys.path or not os.path.isdir(path):
return
site.addsitedir(path)
importlib.invalidate_caches()
system_sites = [
i
for i, p in enumerate(sys.path)
if p.endswith(("site-packages", "dist-packages")) and p != path
]
if system_sites:
sys.path.remove(path)
sys.path.insert(system_sites[0], path)
def activate(manifest: RuntimeManifest) -> None:
"""Make an installed runtime importable in the current process.
Idempotent, and safe to call when nothing is installed; the SDK import
then fails with its own error.
"""
reason = _usable_reason()
if reason is not None:
logger.warning("Ignoring the %s runtime because %s", manifest.name, reason)
return
_add_user_site()
lib_dir = user_base() / "lib"
if manifest.needs_ld_library_path and str(lib_dir) not in os.environ.get(
"LD_LIBRARY_PATH", ""
).split(":"):
logger.warning(
"%s is not on LD_LIBRARY_PATH; the %s runtime may fail to load",
lib_dir,
manifest.name,
)
for soname in manifest.preload:
path = str(lib_dir / soname)
if path in _loaded_libs:
continue
if not os.path.exists(path):
logger.debug("Runtime library %s is not installed", path)
continue
try:
_loaded_libs[path] = ctypes.CDLL(path, mode=ctypes.RTLD_GLOBAL)
except OSError as err:
logger.warning("Unable to preload %s: %s", path, err)
def ensure_installed(manifest: RuntimeManifest) -> None:
"""Install the manifest into the user site unless it already is.
Meant to run once in the main process before detector processes start,
so sys.path is inherited by them. Raises RuntimeDependencyError when the
runtime cannot be installed or must not be used.
"""
reason = _usable_reason()
if reason is not None:
raise RuntimeDependencyError(
f"Refusing to install the {manifest.name} runtime because {reason}"
)
cache = cache_dir(manifest.name)
cache.mkdir(parents=True, exist_ok=True)
with FileLock(cache / ".install.lock", timeout=600):
stamp = _read_stamp(manifest.name)
if _is_current(manifest, stamp):
logger.debug(
"%s runtime %s is already installed", manifest.name, manifest.version
)
activate(manifest)
return
logger.info("Installing the %s runtime %s", manifest.name, manifest.version)
_remove_stamped_files(stamp)
staging = Path(tempfile.mkdtemp(prefix=f"frigate-{manifest.name}-"))
files: list[str] = []
try:
for artifact in _artifacts_for_machine(manifest):
staged = _stage(_fetch(artifact, cache), artifact, staging)
if artifact.kind is ArtifactKind.wheel:
_install_wheel(staged)
else:
files.extend(_extract_archive(staged, artifact))
except (OSError, tarfile.TarError, zipfile.BadZipFile) as err:
raise RuntimeDependencyError(
f"Unable to install the {manifest.name} runtime: {err}"
) from err
finally:
shutil.rmtree(staging, ignore_errors=True)
_write_stamp(manifest, files)
activate(manifest)
logger.info(
"Installed the %s runtime %s into %s",
manifest.name,
manifest.version,
user_base(),
)