check every overlapping review item before dropping a recording

Recording retention around review items applied only the first review item, in start order, whose pre and post capture window covered a segment. When an alert and a detection were close together, a quiet segment covered by both was dropped by the detection's motion mode without checking the alert's all mode. The loop also stopped at a review item that started too late to reach the segment, even when a later alert with a longer pre_capture did reach it. Both the recording maintainer and cleanup now check every review item whose window covers the segment and keep it if any of their retain modes would.

docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run

@@ -38,7 +38,6 @@ device_globs=(
     "/dev/nvidia*"
     "/dev/memx*"
     "/dev/dxrt*"
-    "/dev/metis*"
 )
 
 IFS=',' read -ra extra_globs <<< "${DEVICE_ACL_PATHS:-}"

docker/main/rootfs/usr/local/nginx/conf/auth_location.conf

@@ -20,11 +20,6 @@ location /auth {
     proxy_set_header X-Go2rtc-Stream-Proxy $go2rtc_stream_proxy;
     proxy_set_header X-Server-Port $server_port;
     proxy_set_header X-Forwarded-Proto $scheme;
-    # inputs for the cross-origin check on websocket and go2rtc stream requests
-    proxy_set_header Origin $http_origin;
-    proxy_set_header X-Original-Host $http_host;
-    proxy_set_header X-Forwarded-Host $forwarded_host;
-    proxy_set_header X-Original-Upgrade $http_upgrade;
     proxy_set_header Content-Length "";
     # Pass along auth related info
     proxy_set_header Authorization $http_authorization;

docker/main/rootfs/usr/local/nginx/conf/nginx.conf

@@ -54,13 +54,6 @@ http {
         default 1;
     }
 
-    # keep an outer proxy's X-Forwarded-Host so the websocket servers can
-    # match a browser's Origin when that proxy rewrites Host
-    map $http_x_forwarded_host $forwarded_host {
-        "" $http_host;
-        default $http_x_forwarded_host;
-    }
-
     upstream frigate_api {
         server 127.0.0.1:5001;
         keepalive 1024;

docker/main/rootfs/usr/local/nginx/conf/proxy.conf

@@ -4,7 +4,7 @@ proxy_set_header Upgrade $http_upgrade;
 proxy_set_header Connection "Upgrade";
 proxy_set_header X-Original-URL $request_uri;
 proxy_set_header X-Forwarded-Proto $scheme;
-proxy_set_header X-Forwarded-Host $forwarded_host;
+proxy_set_header X-Forwarded-Host $http_host;
 proxy_set_header X-Forwarded-URI $request_uri;
 proxy_set_header X-Forwarded-Ssl on;
 proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

docs/data/object_detectors_models.yaml

@@ -1278,34 +1278,3 @@ axengine:
             input_dtype: int
             input_pixel_format: bgr
             labelmap_path: /labelmap/coco-80.txt
-axelera:
-  title: Axelera Metis
-  models:
-    - key: yolox
-      label: YOLOX
-      recommended: true
-      download: |-
-        Axelera models must be compiled for the Metis NPU with the Voyager SDK before Frigate can run them (compilation is not supported inside the container). Compile a model with the SDK, zip the compiled output, and point the model's `path` at the zip. The default recommendation is the Apache-2.0 licensed YOLOX (Megvii YOLOX, e.g. the `yolox-s-coco-onnx` model-zoo network).
-      ui: |-
-        Navigate to **Settings > System > Detection models** and select **Axelera Metis** from the **Hardware** dropdown. Then, on the same model, open the **Custom Model** tab and configure:
-
-        | Field                                    | Value                                                        |
-        | ---------------------------------------- | ------------------------------------------------------------ |
-        | **Custom object detector model path**    | `/config/model_cache/axelera/yolox-s-coco-onnx.zip`          |
-        | **Label map for custom object detector** | `/labelmap/coco-80.txt`                                      |
-        | **Object detection model input width**   | `640` (must equal the compiled input size)                   |
-        | **Object detection model input height**  | `640`                                                      |
-        | **Model Input Pixel Color Format**       | `bgr` (Frigate delivers BGR; the plugin hands RGB planes to the compiled input) |
-        | **Model Input Tensor Shape**             | `nhwc` (Frigate's default value)                             |
-        | **Model Input D Type**                   | `int` (Frigate's default value)                              |
-        | **Object Detection Model Type**          | `yolox`                                                      |
-      yaml: |-
-        models:
-          - devices:
-              - axelera:PCIe
-            path: /config/model_cache/axelera/yolox-s-coco-onnx.zip
-            model_type: yolox
-            width: 640
-            height: 640
-            input_pixel_format: bgr
-            labelmap_path: /labelmap/coco-80.txt

docs/docs/configuration/advanced/system.md

@@ -360,7 +360,6 @@ For example, in Nginx:
 
 ```
 location /frigate {
-    proxy_set_header Host $host;
     proxy_set_header X-Ingress-Path /frigate;
     proxy_pass http://frigate_backend;
 }

docs/docs/configuration/object_detectors.md

@@ -63,10 +63,6 @@ Frigate supports multiple different detectors that work on different types of ha
 
 - [AXEngine](#axera): axmodels can run on AXERA AI acceleration.
 
-**Axelera** <CommunityBadge />
-
-- [Axelera Metis](#axelera-metis): The Metis AI Accelerator is available in M.2 and PCIe formats, running compiled YOLO models through the vendor runtime.
-
 **For Testing**
 
 - [CPU Detector (not recommended for actual use](#cpu-detector-not-recommended): Use a CPU to run tflite model, this is not recommended and in most cases OpenVINO can be used in CPU mode with better results.
@@ -104,7 +100,7 @@ models:
       - openvino:GPU
 ```
 
-Coral EdgeTPU, MemryX, and Axelera Metis accelerators can only be opened by one process, so those devices can not be repeated.
+Coral EdgeTPU and MemryX accelerators can only be opened by one process, so those devices can not be repeated.
 
 ### Running more than one model
 
@@ -944,41 +940,3 @@ The AXEngine python package is not part of the Frigate image. It is downloaded a
 When configuring the AXEngine detector, you have to specify the model name.
 
 <ModelConfigDropdown detectorTitle="AXEngine" models={objectDetectorsModels.axengine.models} />
-
----
-
-## Axelera Metis
-
-<CommunityBadge />
-
-Hardware accelerated object detection is supported on the Axelera Metis AI Accelerator, available in M.2 and PCIe card formats. Inference runs through Axelera's low-level runtime (`axelera.runtime`), which executes compiled YOLO-family models with a flat per-frame latency.
-
-See the [installation docs](../frigate/installation.md#axelera-metis) for information on configuring the Metis hardware.
-
-:::info
-
-The Axelera runtime is not part of the Frigate image. It is downloaded and installed into `/config/.local` the first time an Axelera detector is configured, verified against pinned checksums, and updated automatically when a Frigate release pins a new version. If the container has no internet access, see [Detector runtimes](/frigate/network_requirements#detector-runtimes) for how to provide the files yourself.
-
-:::
-
-### Configuration {#configuration-axelera}
-
-<ModelConfigDropdown detectorTitle="Axelera Metis" models={objectDetectorsModels.axelera.models} />
-
-#### Using a Custom Model
-
-Models must be compiled for the Metis NPU with the Axelera Voyager SDK before Frigate can run them. The compiled output is a directory containing `1/model.json`, `1/manifest.json`, and the baked `1/postprocess_graph.onnx`.
-
-Compile the model on a machine with the Voyager SDK installed (compilation is not supported inside the Frigate container), for example with the SDK's `deploy.py` against a model-zoo yaml. Then zip the compiled output and point the model's `path` at the zip (bind-mounted, or at an http(s) URL); the archive extracts fresh on every start.
-
-```yaml
-models:
-  - devices:
-      - axelera:PCIe
-    path: /config/model_cache/axelera/yolox-s-coco-onnx.zip
-    width: 640 # must equal the compiled input size
-    height: 640
-    input_pixel_format: bgr # Frigate delivers BGR; the plugin feeds RGB planes
-    model_type: yolox
-    labelmap_path: /labelmap/coco-80.txt
-```

docs/docs/frigate/installation.md

@@ -596,60 +596,6 @@ If you are using `docker run`, add this option to your command `--device /dev/ax
 
 Finally, configure [hardware object detection](/configuration/object_detectors#axera) to complete the setup.
 
-### Axelera Metis
-
-The Axelera Metis AI Accelerator is available in the M.2 and PCIe card form factors and is supported on x86 (Intel/AMD) PCs and Rock 5B class arm64 boards.
-
-The Axelera runtime used inside the container is not part of the Frigate image; Frigate downloads and installs it at first start once an Axelera detector is configured. Containers without internet access can provide the files themselves, see [Detector runtimes](/frigate/network_requirements#detector-runtimes). The host side kernel driver still has to be installed as described below.
-
-#### Installation
-
-Install the Metis PCIe kernel driver and firmware on the **host** following [Axelera's driver installation guide](https://developer.axelera.ai/axelera-documentation/Getting-Started/GettingStartedWithMetis/InstallingAxeleraMetisDriver). The driver creates the `/dev/metis*` device nodes. Reboot after installing it.
-
-#### Setup
-
-To set up Frigate, follow the default installation instructions, for example: `ghcr.io/blakeblackshear/frigate:stable`
-
-The runtime enumerates the card by the colon-separated name the host udev rules create as a symlink (for example `/dev/metis-0:1:0`). Docker cannot pass that name as a `--device` argument, and a bind mount resolves the symlink to the raw node, which the runtime does not enumerate. A CDI device spec defines the container path and host path as separate fields, and a `startContainer` hook recreates the symlink inside the container.
-
-Create `/etc/cdi/axelera.yaml` on the host, matching the node `ls /dev/metis*` prints:
-
-```yaml
-cdiVersion: "0.6.0"
-kind: axelera.ai/metis
-devices:
-  - name: "0"
-    containerEdits:
-      deviceNodes:
-        - path: /dev/metis-0-1-0
-          hostPath: /dev/metis-0-1-0
-        - path: /dev/dma_heap/system
-      hooks:
-        - hookName: startContainer
-          path: /bin/ln
-          args: ["/bin/ln", "-sf", "metis-0-1-0", "/dev/metis-0:1:0"]
-          timeout: 5
-```
-
-Then grant the device in your `docker-compose.yml` file:
-
-```yaml
-devices:
-  - axelera.ai/metis=0
-```
-
-If you are using `docker run`, add `--device axelera.ai/metis=0` to your command. CDI is enabled by default in Docker 28 and later; older versions need `"features": {"cdi": true}` in `daemon.json`. The hook runs as the container user, so this setup needs the container to start as root and does not work under Docker's `user:` mode.
-
-:::warning
-
-DMA-BUF must be available on the host kernel. On Debian/Ubuntu this is provided by the default kernel; on other systems it may need enabling (`CONFIG_DMABUF_HEAPS_SYSTEM`). If Frigate's log reports that no DMA heap can be allocated, check that `/dev/dma_heap/system` exists on the host and is passed into the container.
-
-:::
-
-#### Configuration
-
-Finally, configure [hardware object detection](/configuration/object_detectors#axelera-metis) to complete the setup.
-
 ## Docker
 
 Running through Docker with Docker Compose is the recommended install method.

docs/docs/frigate/network_requirements.md

@@ -65,12 +65,11 @@ The SDKs for a few hardware detectors are not shipped in the Frigate image. They
 | [Hailo 8 / 8L / 8R](/configuration/object_detectors#hailo) | 4.21.0  | `hailort-debian12-amd64.tar.gz` and `hailort-4.21.0-cp311-cp311-linux_x86_64.whl` on x86, `hailort-debian12-arm64.tar.gz` and `hailort-4.21.0-cp311-cp311-linux_aarch64.whl` on arm64 | [GitHub release](https://github.com/frigate-nvr/hailort/releases/tag/v4.21.0)            |
 | [MemryX MX3](/configuration/object_detectors#memryx-mx3)   | 2.1.0   | `mx_accl_frigate-2.1.0.zip` (the release source archive, renamed)                                                                                                                     | [GitHub archive](https://github.com/memryx/mx_accl_frigate/archive/refs/tags/v2.1.0.zip) |
 | [AXERA AXEngine](/configuration/object_detectors#axera)    | 0.1.3   | `axengine-0.1.3-py3-none-any.whl`                                                                                                                                                     | [GitHub release](https://github.com/AXERA-TECH/pyaxengine/releases/tag/0.1.3-frigate)    |
-| [Axelera Metis](/configuration/object_detectors#axelera-metis) | 1.8.0   | `axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl` on x86, `axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl` on arm64, plus `axelera_firmware-1.8.0-py3-none-any.whl` on both | [Axelera Artifactory PyPI](https://software.axelera.ai/artifactory/#web/ui/redirect/pypi/axelera-pypi/simple) |
 
 If the container cannot reach GitHub, provide the files yourself:
 
 1. Download the files for your architecture on a machine with internet access.
-2. Place them, with exactly the file names listed above, in `/config/model_cache/runtimes/<detector>/`, where `<detector>` is the detector named in your config's `devices` (`hailo`, `memryx`, `axengine`, or `axelera`).
+2. Place them, with exactly the file names listed above, in `/config/model_cache/runtimes/<detector>/`, where `<detector>` is the detector named in your config's `devices` (`hailo`, `memryx`, or `axengine`).
 3. Start Frigate. Files whose checksum matches are installed without any download; a file with the wrong checksum is discarded and downloaded again, so a failed startup log names the file to replace.
 
 The `GITHUB_ENDPOINT` mirror variable below applies to these downloads as well.

docs/docs/guides/reverse_proxy.md

@@ -37,8 +37,6 @@ Your reverse proxy must be configured to forward the `Upgrade` and `Connection`
 
 Note that some proxies disable WebSocket support by default. For example, Nginx Proxy Manager has a "Websockets Support" toggle that must be enabled.
 
-Frigate rejects WebSocket connections opened from a page on another site. Your proxy must either preserve the original `Host` header or send it in `X-Forwarded-Host`, otherwise Frigate cannot tell the connection came from its own UI.
-
 ## Proxies
 
 There are many solutions available to implement reverse proxies and the community is invited to help out documenting others through a contribution to this page.
@@ -163,7 +161,6 @@ The settings below enabled connection upgrade, sets up logging (optional) and pr
   error_log /data/logs/proxy-host-40_error.log warn;
 
   location / {
-    proxy_set_header Host $host;
     proxy_set_header Upgrade $http_upgrade;
     proxy_set_header Connection $http_connection;
     proxy_http_version 1.1;

docs/docs/integrations/home-assistant.md

@@ -55,10 +55,6 @@ To install the optional companion Lovelace card, please see the [separate
 installation instructions](https://github.com/dermotduffy/frigate-hass-card) for
 that card.
 
-Frigate rejects browser connections to its live streams that come from another
-origin, such as a Home Assistant dashboard. Configure cards to stream through the
-integration's proxy rather than pointing them directly at Frigate's port.
-
 ## Configuration
 
 When configuring the integration, you will be asked for the `URL` of your Frigate instance which can be pointed at the internal unauthenticated port (`5000`) or the authenticated port (`8971`) for your instance. This may look like `http://<host>:5000/`.

frigate/api/auth.py

@@ -39,7 +39,6 @@ from frigate.config import AuthConfig, ProxyConfig
 from frigate.const import CONFIG_DIR, JWT_SECRET_ENV_VAR, PASSWORD_HASH_ALGORITHM
 from frigate.models import User
 from frigate.notices import raise_notice
-from frigate.util.origin import origin_allowed
 
 logger = logging.getLogger(__name__)
 
@@ -719,17 +718,6 @@ def auth(request: Request):
 
     success_response = Response("", status_code=202)
 
-    original_url = request.headers.get("x-original-url")
-
-    if _is_cross_origin_stream_request(original_url, request):
-        logger.warning(
-            "Rejected cross-origin request: origin=%s, host=%s, x-forwarded-host=%s",
-            request.headers.get("origin"),
-            request.headers.get("x-original-host"),
-            request.headers.get("x-forwarded-host"),
-        )
-        return Response("", status_code=403)
-
     # dont require auth if the request is on the internal port
     # this header is set by Frigate's nginx proxy, so it cant be spoofed.
     # the port is the boot-time snapshot rather than the live config value:
@@ -753,6 +741,7 @@ def auth(request: Request):
         logger.debug("X-Proxy-Secret header does not match configured secret value")
         return fail_response
 
+    original_url = request.headers.get("x-original-url")
     frigate_config = request.app.frigate_config
 
     # if auth is disabled, just apply the proxy header map and return success
@@ -1318,21 +1307,6 @@ def _is_go2rtc_stream_proxy_request(path: str, request: Request) -> bool:
     return any(normalized.startswith(p) for p in GO2RTC_STREAM_PROXY_PATHS)
 
 
-def _is_cross_origin_stream_request(original_url: str | None, request: Request) -> bool:
-    # the internal port grants admin, so a browser on another site must be
-    # stopped before that early return
-    upgrade = request.headers.get("x-original-upgrade", "").lower()
-    path = split_request_uri(original_url or "")[0]
-    if upgrade != "websocket" and not _is_go2rtc_stream_proxy_request(path, request):
-        return False
-
-    return not origin_allowed(
-        request.headers.get("origin"),
-        request.headers.get("x-original-host"),
-        request.headers.get("x-forwarded-host"),
-    )
-
-
 def deny_response_for_go2rtc_stream(
     original_url: str | None, role: str | None, request: Request
 ) -> int | None:

frigate/comms/ws.py

@@ -40,7 +40,6 @@ from frigate.const import (
 )
 from frigate.models import User
 from frigate.output.ws_auth import ws_has_camera_access
-from frigate.util.origin import origin_allowed
 
 logger = logging.getLogger(__name__)
 
@@ -473,28 +472,6 @@ def _materialize_for_ws(
     return None
 
 
-class OriginCheckedWebSocketApplication(WebSocketWSGIApplication):  # type: ignore[misc]
-    """ws4py application that rejects cross-origin browser handshakes."""
-
-    def __call__(self, environ: dict[str, Any], start_response: Callable) -> Any:
-        if not origin_allowed(
-            environ.get("HTTP_ORIGIN"),
-            environ.get("HTTP_HOST"),
-            environ.get("HTTP_X_FORWARDED_HOST"),
-        ):
-            logger.warning(
-                "Rejected cross-origin websocket handshake: origin=%s, host=%s, "
-                "x-forwarded-host=%s",
-                environ.get("HTTP_ORIGIN"),
-                environ.get("HTTP_HOST"),
-                environ.get("HTTP_X_FORWARDED_HOST"),
-            )
-            start_response("403 Forbidden", [("Content-Length", "0")])
-            return []
-
-        return super().__call__(environ, start_response)
-
-
 class WebSocket(WebSocket_):  # type: ignore[misc]
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
@@ -644,7 +621,7 @@ class WebSocketClient(Communicator):
             5002,
             server_class=WSGIServer,
             handler_class=WebSocketWSGIRequestHandler,
-            app=OriginCheckedWebSocketApplication(handler_cls=_WebSocketHandler),
+            app=WebSocketWSGIApplication(handler_cls=_WebSocketHandler),
         )
         self.websocket_server.initialize_websockets_manager()
         self.websocket_thread = threading.Thread(

frigate/detectors/hardware.py

@@ -284,19 +284,6 @@ def detect_deepx() -> DetectionHardware | None:
     return _hardware("deepx", "deepx", "DEEPX NPU", units)
 
 
-def detect_axelera() -> DetectionHardware | None:
-    """Find Axelera Metis accelerators by their device nodes."""
-    nodes = sorted(glob(f"{DEV_ROOT}/metis*"))
-
-    if not nodes:
-        return None
-
-    # the runtime addresses the card through the driver, not per node, so one
-    # unit covers the accelerator
-    units = [HardwareUnit(device="axelera:PCIe", label=os.path.basename(nodes[0]))]
-    return _hardware("axelera", "axelera", "Axelera Metis", units)
-
-
 def detect_rockchip() -> DetectionHardware | None:
     """Find a Rockchip NPU by reading the SoC from the device tree."""
     compatible = _read(f"{PROC_ROOT}/device-tree/compatible")
@@ -357,7 +344,6 @@ PROBES = (
     detect_hailo,
     detect_memryx,
     detect_deepx,
-    detect_axelera,
     detect_intel_npu,
     detect_intel_gpu,
     detect_nvidia_gpu,

frigate/detectors/plugins/axelera.py

@@ -1,972 +0,0 @@
-"""Frigate object detector plugin for the Axelera Metis NPU (`axelera`).
-
-Axelera AI makes the Metis PCIe/M.2 NPU accelerator. Inference runs through the
-vendor's LOW-LEVEL runtime API (``axelera.runtime``, the same direct
-"ModelInstance.run" interface used for hailo8l/memryx), NOT through the
-GStreamer-based ``create_inference_stream`` app pipeline.
-
-Why the low-level API instead of the stream: the video pipeline is depth-4 and
-clocked by the input arrival interval, so a frame's result round-trips in
-~4x the input gap (~150 ms at a ~35 fps aggregate) even though the accelerator
-computes a frame in ~21 ms back-to-back. That latency is what Frigate's async
-runner reports and what a camera waits on. The direct runtime API is a
-synchronous ``ModelInstance.run(inputs, outputs)`` call with no video pipeline,
-so per-frame latency is the flat model time (~21 ms for YOLO-NAS M 704x576,
-measured), independent of scene activity.
-
-Model configuration:
-  - ``path`` = a .zip of the COMPILED model output (containing ``1/model.json``
-    + ``1/manifest.json`` + ``1/postprocess_graph.onnx``), local or http(s)
-    URL (the memryx custom-model pattern); the selected archive extracts
-    fresh on every start.
-  - ``labelmap_path`` / ``labelmap``: Frigate's labelmap names the class
-    index the decode returns, so renames and the coco-80 bus/truck->car
-    mapping need no plugin-side code.
-  - ``width`` / ``height``: must equal the compiled model input size; the
-    plugin raises at init otherwise (no letterboxing).
-
-The SDK's own baked postprocess graph (``1/postprocess_graph.onnx``) converts
-the raw int8 AIPU tensors into boxes + class scores, so the plugin only
-dequantizes the heads and runs a thin class-max + NMS.
-"""
-
-import json
-import logging
-import os
-import queue
-import shutil
-import threading
-import zipfile
-from typing import ClassVar, Literal
-
-import cv2
-import numpy as np
-from pydantic import ConfigDict, Field
-
-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.file import FileLock
-from frigate.util.model import xyxy_to_xywh_for_nms
-from frigate.util.runtime_deps import (
-    Artifact,
-    ArtifactKind,
-    RuntimeManifest,
-    user_site,
-)
-
-logger = logging.getLogger(__name__)
-
-DETECTOR_KEY = "axelera"
-
-# The Axelera runtime SDK is installed at first start rather than shipped in
-# the image. The wheels come from Axelera's public Artifactory PyPI index
-# (readable without credentials) and are pinned by sha256. The runtime wheel
-# is auditwheel-repaired (all native deps bundled under axelera_runtime.libs
-# and resolved through $ORIGIN RPATH) and the firmware wheel ships the AIPU
-# device firmware under axelera/omega so the pip install is fully self
-# contained. axelera.runtime imports axelera.runtime2 only inside a try for
-# its separate "new runtime API"; the low-level API this plugin uses lives in
-# axelera.runtime.objects and works without it, so runtime2 is not installed.
-AXELERA_RUNTIME_VERSION = "1.8.0"
-AXELERA_FIRMWARE_VERSION = "1.8.0"
-AXELERA_PYPI = "https://software.axelera.ai/artifactory/axelera-pypi"
-
-AXELERA_MANIFEST = RuntimeManifest(
-    name=DETECTOR_KEY,
-    version=AXELERA_RUNTIME_VERSION,
-    artifacts=(
-        Artifact(
-            url=(
-                f"{AXELERA_PYPI}/axelera-runtime/{AXELERA_RUNTIME_VERSION}/"
-                "axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_x86_64"
-                ".manylinux_2_28_x86_64.whl"
-            ),
-            sha256="c97b41f427a61ca0fe22dcab38bbf885d12e0a04a0c4f55e87f1f4036a39c869",
-            kind=ArtifactKind.wheel,
-            machines=("x86_64",),
-        ),
-        Artifact(
-            url=(
-                f"{AXELERA_PYPI}/axelera-runtime/{AXELERA_RUNTIME_VERSION}/"
-                "axelera_runtime-1.8.0-cp311-cp311-manylinux_2_27_aarch64"
-                ".manylinux_2_28_aarch64.whl"
-            ),
-            sha256="5ec8a048fa94d261080c11b840044f5f7b62fce1f0a43ba8733fc880f40d86ef",
-            kind=ArtifactKind.wheel,
-            machines=("aarch64",),
-        ),
-        Artifact(
-            url=(
-                f"{AXELERA_PYPI}/axelera-firmware/{AXELERA_FIRMWARE_VERSION}/"
-                "axelera_firmware-1.8.0-py3-none-any.whl"
-            ),
-            sha256="b926c16383e696c6e2b200b9207a6fa8536e9b1f8a3d261987d3f829c0b37d36",
-            kind=ArtifactKind.wheel,
-        ),
-    ),
-    import_check="axelera.runtime",
-)
-
-AX_DEFAULT_NETWORK = "yolox-s-coco-onnx"
-MAX_ROWS = 20
-# int8 zero-point the compiled inputs are offset by (px - 128); also the
-# filler for the padded border. Everything else about a specific compiled
-# model (padded input size, content offset, per-output padded/real channels
-# and dequant scale/zero-point) is read at init from 1/manifest.json, so the
-# decoder works for any compiled YOLO-family model that ships a baked
-# postprocess_graph.onnx (verified: YOLO-NAS M and YOLOX-S).
-INPUT_Q_ZERO = -128
-# confidence / NMS defaults mirror the SDK DecodeYolo operator defaults
-CONF_THRESHOLD = 0.25
-NMS_IOU_THRESHOLD = 0.45
-QUEUE_MAXSIZE = 32
-# ping-pong buffer sets for the producer/consumer pipeline: the producer
-# writes one set while the decoder reads a previous one, so the NPU stays
-# saturated. 3 sets bound the in-flight frames (pool size must equal the
-# decode queue bound).
-BUFFER_SETS = 3
-
-
-# ----------------- Utility Functions ----------------- #
-
-
-def is_url(value: str) -> bool:
-    """Return True for http(s) URLs."""
-    return value.startswith(("http://", "https://"))
-
-
-def _extract_model_zip(zip_path: str, dest_dir: str) -> None:
-    """Unpack a compiled-model zip into ``dest_dir`` (caller wipes it first)."""
-    with zipfile.ZipFile(zip_path) as archive:
-        for member in archive.infolist():
-            if member.is_dir():
-                continue
-            target = os.path.realpath(os.path.join(dest_dir, member.filename))
-            if not target.startswith(os.path.realpath(dest_dir) + os.sep):
-                raise ValueError(
-                    f"model archive member escapes target: {member.filename}"
-                )
-            os.makedirs(os.path.dirname(target), exist_ok=True)
-            with archive.open(member) as source, open(target, "wb") as out:
-                out.write(source.read())
-
-
-def _resolve_model_json(path) -> str:
-    """Resolve ``path`` (a .zip of the compiled output) to its ``model.json``.
-
-    Like the memryx custom-model flow, ``path`` is always a zip of the
-    compiled output: an http(s) URL downloads into the model cache, a local
-    zip is verified, and the selected archive extracts fresh on every start
-    (under a lock, as several processes may start at once), so a replaced
-    archive is always the model that runs. Anything else raises.
-    """
-    if not path:
-        raise ValueError(
-            "axelera: the model path is required (a .zip of the compiled "
-            "model, local or http(s) URL)."
-        )
-    if is_url(path):
-        name = os.path.splitext(os.path.basename(path.split("?")[0]))[0]
-        name = name or AX_DEFAULT_NETWORK
-        cache = os.path.join(MODEL_CACHE_DIR, "axelera")
-        zip_path = os.path.join(cache, f"{name}.zip")
-        if not os.path.isfile(zip_path):
-            os.makedirs(cache, exist_ok=True)
-            try:
-                # function level like runtime_deps: keeps the detector API
-                # import (and this module) free of the downloader's IPC stack
-                from frigate.util.downloader import ModelDownloader
-
-                ModelDownloader.download_from_url(path, zip_path)
-            except OSError as exc:
-                raise RuntimeError(
-                    f"Failed to download model from {path}: {exc}"
-                ) from exc
-    else:
-        if not path.endswith(".zip"):
-            raise ValueError(
-                f"Invalid model path: {path}. Only .zip files are supported. "
-                "Please provide a .zip model archive."
-            )
-        if not os.path.isfile(path):
-            raise FileNotFoundError(f"Custom model zip not found: {path}")
-        zip_path = path
-        name = os.path.splitext(os.path.basename(path))[0] or AX_DEFAULT_NETWORK
-
-    if not zipfile.is_zipfile(zip_path):
-        raise ValueError(f"model archive {zip_path} is not a zip file")
-    cache = os.path.join(MODEL_CACHE_DIR, "axelera")
-    os.makedirs(cache, exist_ok=True)
-    extract_dir = os.path.join(cache, f"{name}-extracted")
-    with FileLock(os.path.join(cache, f".{name}.lock")):
-        if os.path.isdir(extract_dir):
-            shutil.rmtree(extract_dir)
-        _extract_model_zip(zip_path, extract_dir)
-    return _find_model_json(extract_dir)
-
-
-def _find_model_json(where: str) -> str:
-    """Locate the runtime's model.json under an extracted model zip.
-
-    ``where`` is the extract directory, so model.json sits either at its root
-    (``1/model.json`` or ``model.json``) or under the single top-level folder
-    the archive was wrapped in. Only an unambiguous single match is accepted:
-    several candidate subdirectories raise instead of returning whichever
-    sorts first, because silently picking one would run a model the user
-    never selected.
-    """
-    if not os.path.isdir(where):
-        raise FileNotFoundError(
-            f"model.json not found under {where}. The model zip must contain "
-            "1/model.json at its root or under one top-level folder."
-        )
-    for candidate in (
-        os.path.join(where, "1", "model.json"),
-        os.path.join(where, "model.json"),
-    ):
-        if os.path.isfile(candidate):
-            return candidate
-    matches: list[str] = []
-    for entry in sorted(os.listdir(where)):
-        sub = os.path.join(where, entry)
-        if not os.path.isdir(sub):
-            continue
-        for candidate in (
-            os.path.join(sub, "1", "model.json"),
-            os.path.join(sub, "model.json"),
-        ):
-            if os.path.isfile(candidate):
-                matches.append(candidate)
-                break
-    if len(matches) == 1:
-        return matches[0]
-    if len(matches) > 1:
-        listed = ", ".join(os.path.relpath(m, where) for m in matches)
-        raise ValueError(
-            f"multiple compiled models found under {where}: {listed}. The "
-            "model zip must contain exactly one model.json, at its root or "
-            "under one top-level folder."
-        )
-    raise FileNotFoundError(
-        f"model.json not found under {where}. The model zip must contain "
-        "1/model.json at its root or under one top-level folder."
-    )
-
-
-def _nms_per_class(boxes, scores, class_ids, iou_threshold: float) -> np.ndarray:
-    """Per-class greedy NMS via cv2.dnn.NMSBoxes (C++, GIL-free).
-
-    Boxes are xyxy; xyxy_to_xywh_for_nms converts them to the top-left+size
-    format NMSBoxes expects. Running NMS per class keeps a confident car from
-    suppressing an overlapping pedestrian, which class-agnostic NMS does.
-
-    Returns kept indices sorted by descending score.
-    """
-    if len(boxes) == 0:
-        return np.array([], dtype=np.int64)
-
-    order = np.argsort(-scores)
-    kept: list[int] = []
-    for cls in np.unique(class_ids):
-        idx = order[class_ids[order] == cls]
-        if len(idx) == 0:
-            continue
-        indices = cv2.dnn.NMSBoxes(
-            xyxy_to_xywh_for_nms(boxes[idx]).tolist(),
-            [float(s) for s in scores[idx]],
-            # inputs are already filtered above CONF_THRESHOLD by the caller
-            score_threshold=0.0,
-            nms_threshold=float(iou_threshold),
-        )
-        if isinstance(indices, np.ndarray):
-            kept.extend(idx[indices.reshape(-1)].tolist())
-        elif len(indices) > 0:
-            kept.extend(idx[np.asarray(indices, dtype=np.int64).reshape(-1)].tolist())
-
-    if not kept:
-        return np.array([], dtype=np.int64)
-    kept_arr = np.asarray(kept, dtype=np.int64)
-    return kept_arr[np.argsort(-scores[kept_arr])]
-
-
-def _read_model_manifest(model_dir: str) -> dict:
-    """Read the compiled model's manifest (geometry + dequantize params).
-
-    The compiler writes two JSONs next to the kernel: ``1/model.json`` (padded
-    kernel I/O only) and ``1/manifest.json`` (original/padded shapes, the exact
-    content offset inside the padded input, per-output padded channel counts
-    and dequantize (scale, zero-point) for every head).
-    """
-    candidate = os.path.join(model_dir, "manifest.json")
-    if not os.path.isfile(candidate):
-        raise FileNotFoundError(
-            "compiled manifest not found next to the model kernel. Point "
-            "path at a zip of a directory compiled by the Voyager SDK that "
-            "contains 1/model.json + 1/manifest.json + 1/postprocess_graph.onnx."
-        )
-    try:
-        with open(candidate) as f:
-            manifest = json.load(f)
-    except Exception as exc:
-        raise ValueError(
-            f"could not parse compiled manifest {candidate}: {exc}"
-        ) from exc
-    required = (
-        "input_shapes",
-        "input_shapes_original",
-        "output_shapes",
-        "output_shapes_original",
-        "dequantize_params",
-    )
-    missing = [k for k in required if k not in manifest]
-    if missing:
-        raise ValueError(
-            f"compiled manifest {candidate} missing required keys: {missing}"
-        )
-    return manifest
-
-
-# ----------------- AxeleraRuntimeInference (engine) ----------------- #
-
-
-class _AxeleraRuntimeInference:
-    """Runs a compiled model via ``axelera.runtime`` on a two-thread pipeline.
-
-    Frigate's AsyncDetectorRunner calls ``submit(connection_id, frame)`` from
-    the detect thread and ``collect()`` from the result thread. Two engine
-    threads process frames strictly in order: a producer resizes +
-    quantizes + runs the model (``ModelInstance.run``, synchronous and flat:
-    ~21 ms measured), and a consumer feeds the int8 heads through the SDK's
-    baked ``postprocess_graph.onnx`` while the NPU is already busy on the next
-    frame (measured throughput ~25 ms/frame, ~40 fps, vs ~29 ms serial). All
-    ``run`` calls stay on the single producer thread, so in-flight frames never
-    execute the model in parallel (a known card-wedge condition). Because both
-    threads consume their queues FIFO, results are paired to requests by
-    construction - there is no pipeline reorder to guard against.
-    """
-
-    def __init__(
-        self,
-        model_json,
-        model_type: ModelTypeEnum,
-        width: int,
-        height: int,
-        pixel_format: str = "bgr",
-        runtime_factory=None,
-        onnx_factory=None,
-    ) -> None:
-        self._model_type = model_type
-        self._width, self._height = int(width), int(height)
-        # the compiled model consumes RGB planes; Frigate delivers BGR when
-        # input_pixel_format is bgr (flip below), RGB when rgb (no flip)
-        self._flip_channels = pixel_format == "bgr"
-        self._frame_q: queue.Queue = queue.Queue(maxsize=QUEUE_MAXSIZE)
-        self._result_q: queue.Queue = queue.Queue(maxsize=QUEUE_MAXSIZE)
-        self._stop = threading.Event()
-        self._producer_thread: threading.Thread | None = None
-        self._consumer_thread: threading.Thread | None = None
-
-        if runtime_factory is None:
-            try:
-                from axelera.runtime import Context
-            except (ImportError, ModuleNotFoundError) as exc:
-                raise ImportError(
-                    "Axelera runtime (axelera.runtime) is not available in this "
-                    "environment. Frigate installs it at startup when an Axelera "
-                    "detector is configured; check the startup log for errors."
-                ) from exc
-            runtime_factory = Context
-        if onnx_factory is None:
-            import onnxruntime as _ort
-
-            onnx_factory = _ort
-
-        model_dir = os.path.dirname(model_json)
-        self._ctx = runtime_factory()
-        model = self._ctx.load_model(model_json)
-        # enumerate and connect to the DeviceInfo explicitly rather than
-        # letting the runtime pick: it surfaces the lspci/device-node
-        # mismatch as an empty list instead of a late connect failure
-        devices = self._ctx.list_devices()
-        if not devices:
-            raise RuntimeError(
-                "axelera: no Metis device enumerated; the runtime needs the "
-                "raw /dev/metis-<bus>-<dev>-<fn> node in the container and "
-                "the colon-separated name (e.g. /dev/metis-0:1:0) present "
-                "as a symlink to it. See the install docs for the CDI "
-                "device spec that provides both, and check that the metis "
-                "driver is loaded on the host"
-            )
-        # hold the device connection on self: the runtime releases the device
-        # when the connection object is garbage collected, so it must outlive
-        # init for as long as the model instance uses it
-        self._conn = self._ctx.device_connect(devices[0], num_sub_devices=1)
-        self._instance = self._conn.load_model_instance(
-            model, num_sub_devices=1, aipu_cores=1
-        )
-
-        manifest = _read_model_manifest(model_dir)
-
-        # input: padded NHWC (1, H, W, 4) exactly as the runtime expects it,
-        # plus the original NCHW (1, C, H, W) workspace the resize targets
-        in_padded = [int(x) for x in manifest["input_shapes"][0]]
-        in_orig = [int(x) for x in manifest["input_shapes_original"][0]]
-        if len(in_padded) != 4 or in_padded[3] != 4 or in_orig[1] != 3:
-            raise ValueError(
-                f"unsupported compiled input: padded {in_padded}, original {in_orig}"
-            )
-        self._padded_h, self._padded_w = in_padded[1], in_padded[2]
-        self._model_h, self._model_w = in_orig[2], in_orig[3]
-
-        # where the compiler placed the unpadded content inside the padded
-        # buffer: n_padded_ch_inputs = [[N(lo,hi), H(lo,hi), W(lo,hi),
-        # C(lo,hi)]] (one entry per input)
-        npad_data = manifest["n_padded_ch_inputs"]
-        if isinstance(npad_data, list) and npad_data and isinstance(npad_data[0], list):
-            npad_data = npad_data[0]
-        npad = [int(x) for x in npad_data]
-        if len(npad) == 8:
-            self._content_top, self._content_left = npad[2], npad[4]
-        else:
-            self._content_top = (self._padded_h - self._model_h) // 2
-            self._content_left = (self._padded_w - self._model_w) // 2
-
-        out_padded = [tuple(int(x) for x in o) for o in manifest["output_shapes"]]
-        out_real = [
-            tuple(int(x) for x in o) for o in manifest["output_shapes_original"]
-        ]
-        dequant = manifest["dequantize_params"]
-        if not (len(out_padded) == len(out_real) == len(dequant)):
-            raise ValueError(
-                "compiled manifest output counts mismatch: padded "
-                f"{len(out_padded)}, original {len(out_real)}, "
-                f"dequant {len(dequant)}"
-            )
-        # per head: real channel count (crop target) + dequant (scale, zp)
-        self._head_real = [o[1] for o in out_real]
-        self._head_dequant = [(np.float32(d[0]), np.float32(d[1])) for d in dequant]
-        self._out_shapes = out_padded
-
-        pp_path = os.path.join(model_dir, "postprocess_graph.onnx")
-        if not os.path.isfile(pp_path):
-            raise FileNotFoundError(
-                f"postprocess_graph.onnx not found next to {model_json}"
-            )
-        # one frame is in flight through the decoder, so a single intra-op
-        # thread is optimal; extra ORT threads only add context switches
-        sess_opts = onnx_factory.SessionOptions()
-        sess_opts.intra_op_num_threads = 1
-        self._pp = onnx_factory.InferenceSession(
-            pp_path,
-            sess_options=sess_opts,
-            providers=["CPUExecutionProvider"],
-        )
-        pp_inputs = self._pp.get_inputs()
-        self._pp_input_names = [i.name for i in pp_inputs]
-        if len(pp_inputs) != len(self._head_real):
-            raise ValueError(
-                "postprocess graph input count mismatch: "
-                f"{len(pp_inputs)} graph inputs vs "
-                f"{len(self._head_real)} model outputs"
-            )
-        # Map each postprocess input to the model output whose dequantized NCHW
-        # crop has exactly that (1, C, H, W) shape: the compile order of the
-        # raw heads is not guaranteed to match the postprocess graph's input
-        # order (they differ for YOLOX vs YOLO-NAS), shapes are unique per head.
-        self._feed_index: list[int] = []
-        for pin in pp_inputs:
-            want = tuple(int(x) for x in pin.shape)
-            matches = [
-                i
-                for i in range(len(self._head_real))
-                if (
-                    1,
-                    self._head_real[i],
-                    out_padded[i][1],
-                    out_padded[i][2],
-                )
-                == want
-            ]
-            if len(matches) != 1:
-                raise ValueError(
-                    f"cannot map postprocess input {pin.name} {want} to a "
-                    "unique model output"
-                )
-            self._feed_index.append(matches[0])
-
-        # pooled model-input + raw-int8-output sets, ping-ponged between
-        # the producer (runs the model) and the decoder (postprocesses the
-        # previous frame's outputs while the NPU is busy on the next one).
-        self._pool: queue.Queue = queue.Queue(maxsize=BUFFER_SETS)
-        for _ in range(BUFFER_SETS):
-            in_buf = np.full(
-                (1, self._padded_h, self._padded_w, 4),
-                INPUT_Q_ZERO,
-                np.int8,
-            )
-            out_bufs = [np.zeros(shape, np.int8) for shape in out_padded]
-            self._pool.put((in_buf, out_bufs))
-        self._decode_q: queue.Queue = queue.Queue(maxsize=BUFFER_SETS)
-        # fp32 postprocess feeds, sized to each graph input and reused: the
-        # dequantizing copy writes straight into them (copyto fuses the
-        # NHWC->NCHW transpose with the int8->fp32 cast, and the dequant math
-        # then runs in place) so the hot path never allocates
-        self._pp_scratch = {
-            pin.name: np.zeros(
-                tuple(int(x) for x in pin.shape),
-                np.float32,
-            )
-            for pin in pp_inputs
-        }
-        if (self._width, self._height) != (self._model_w, self._model_h):
-            raise ValueError(
-                f"axelera: model width/height must match the compiled input "
-                f"({self._model_w}x{self._model_h}); configured "
-                f"{self._width}x{self._height}"
-            )
-        # per-frame scratch (producer-thread-owned): the resized frame, the
-        # channel-oriented copy, and the quantized int8
-        self._resize_scratch = np.zeros((self._model_h, self._model_w, 3), np.uint8)
-        self._orient_scratch = np.zeros((self._model_h, self._model_w, 3), np.uint8)
-        self._quant_scratch = np.zeros((self._model_h, self._model_w, 3), np.int8)
-
-    # -- public contract used by AxeleraDetector --------------------------
-
-    def submit(self, connection_id: str, frame: np.ndarray) -> None:
-        """Queue one frame for inference, owned by `connection_id`.
-
-        On overload (input queue full) the frame's INFERENCE is dropped but a
-        zero-row result is still emitted for `connection_id`: the async runner
-        contract is strictly 1 submission : 1 result (SHM cleanup + timing
-        pairing), so raising here would leave the runner waiting for a result
-        that never arrives.
-        """
-        try:
-            self._frame_q.put_nowait((connection_id, frame))
-        except queue.Full:
-            logger.warning(
-                "axelera: inference queue full; dropping frame for %s",
-                connection_id,
-            )
-            self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32))
-            return
-        self._ensure_threads()
-
-    def _ensure_threads(self) -> None:
-        if self._consumer_thread is None or not self._consumer_thread.is_alive():
-            self._consumer_thread = threading.Thread(
-                target=self._consumer, name="axelera-decode", daemon=True
-            )
-            self._consumer_thread.start()
-        if self._producer_thread is None or not self._producer_thread.is_alive():
-            self._producer_thread = threading.Thread(
-                target=self._producer, name="axelera-runtime", daemon=True
-            )
-            self._producer_thread.start()
-
-    def collect(self):
-        """Return (connection_id, rows) for the next completed inference.
-
-        Returns (None, None) on stop so the async runner can exit cleanly.
-        """
-        while not self._stop.is_set():
-            try:
-                return self._result_q.get(timeout=0.25)
-            except queue.Empty:
-                continue
-        return None, None
-
-    def stop(self) -> None:
-        """Stop the workers, release the Metis device. Never raises."""
-        self._stop.set()
-        for t in (self._producer_thread, self._consumer_thread):
-            if t is not None and t.is_alive():
-                t.join(timeout=2)
-        try:
-            # Context.release() releases the context and every object it
-            # created (connection, model, instance); there is no per-object
-            # free() in this runtime API
-            self._ctx.release()
-        except Exception:  # noqa: BLE001, S110
-            pass
-        logger.info("axelera: runtime inference stopped")
-
-    # -- internals ---------------------------------------------------------
-
-    def _push_result(self, connection_id: str, rows: np.ndarray) -> None:
-        """Deliver one completion, applying backpressure instead of dropping.
-
-        The async runner contract is strictly 1 submission : 1 delivered
-        result (SHM cleanup + timing pairing), so an accepted submission must
-        never lose its completion: if the result worker is behind, the
-        producing thread blocks until the queue drains. The only consumer is
-        the runner's result worker (which drains until stop), so the block
-        self-throttles rather than deadlocking; it is stop-aware so shutdown
-        stays responsive while the runner is unwinding.
-        """
-        warned = False
-        while not self._stop.is_set():
-            try:
-                self._result_q.put((connection_id, rows), timeout=0.25)
-                return
-            except queue.Full:
-                if not warned:
-                    logger.warning(
-                        "axelera: result queue full, applying backpressure for %s",
-                        connection_id,
-                    )
-                    warned = True
-        logger.warning(
-            "axelera: stopping, dropping undelivered result for %s", connection_id
-        )
-
-    def _producer(self) -> None:
-        """Resize + quantize + run(); hand raw outputs to the decoder.
-
-        All ``ModelInstance.run`` calls happen on THIS single thread, so the
-        three in-flight buffer sets never execute the model in parallel (a
-        known card-wedge condition) while still overlapping with the decoder's
-        CPU postprocessing.
-        """
-        while not self._stop.is_set():
-            try:
-                connection_id, frame = self._frame_q.get(timeout=0.25)
-            except queue.Empty:
-                continue
-            try:
-                in_buf, out_bufs = self._pool.get(timeout=1.0)
-            except queue.Empty:
-                logger.error("axelera: no free buffers; dropping %s", connection_id)
-                self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32))
-                continue
-            try:
-                self._preprocess(frame, in_buf)
-                self._instance.run([in_buf], out_bufs)
-                self._decode_q.put_nowait((connection_id, in_buf, out_bufs))
-            except Exception:
-                logger.exception("axelera: inference failed for %s", connection_id)
-                self._pool.put((in_buf, out_bufs))
-                self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32))
-
-    def _consumer(self) -> None:
-        """Decode the raw int8 heads (dequant + onnx + NMS) to (20,6) rows."""
-        while not self._stop.is_set():
-            try:
-                connection_id, in_buf, out_bufs = self._decode_q.get(timeout=0.25)
-            except queue.Empty:
-                continue
-            try:
-                rows = self._decode(out_bufs)
-            except Exception:
-                logger.exception("axelera: decode failed for %s", connection_id)
-                rows = np.zeros((MAX_ROWS, 6), np.float32)
-            self._pool.put((in_buf, out_bufs))
-            self._push_result(connection_id, rows)
-
-    def _preprocess(self, frame: np.ndarray, buf: np.ndarray) -> None:
-        """Resize the source (H, W, 3) into the model input and quantize.
-
-        All three passes write into preallocated scratch: resize straight into
-        the scratch buffer (cv2 ``dst=``), an in-place channel swap (no strided
-        ``[..., ::-1]`` copy at write time), and an out= subtract for the
-        int8 quantize.
-        """
-        cv2.resize(
-            frame,
-            (self._model_w, self._model_h),
-            interpolation=cv2.INTER_LINEAR,
-            dst=self._resize_scratch,
-        )
-        if self._flip_channels:
-            # BGR -> RGB: Frigate delivers BGR under the documented
-            # input_pixel_format: bgr, while the compiled input expects RGB
-            # semantics; with input_pixel_format: rgb the frame already
-            # carries RGB and is only swapped during quantisation
-            cv2.cvtColor(
-                self._resize_scratch,
-                cv2.COLOR_BGR2RGB,
-                dst=self._orient_scratch,
-            )
-            src_u8 = self._orient_scratch
-        else:
-            src_u8 = self._resize_scratch
-        # uint8 - 128 wraps mod 256, then cast to int8 == px - 128 (exactly);
-        # dtype= on subtract keeps it one pass with no intermediate array
-        np.subtract(
-            src_u8,
-            np.uint8(128),
-            out=self._quant_scratch,
-            dtype=np.int8,
-            casting="unsafe",
-        )
-        # the frame fills the whole content region; anything the compiler
-        # padded around it stays at the zero-point filler
-        buf[0, :, :, :].fill(INPUT_Q_ZERO)
-        dst = buf[
-            0,
-            self._content_top : self._content_top + self._model_h,
-            self._content_left : self._content_left + self._model_w,
-        ]
-        dst[..., 0:3] = self._quant_scratch
-        dst[..., 3] = 127
-
-    def _decode(self, out_bufs) -> np.ndarray:
-        """Dequantize heads + postprocess graph + conf threshold + NMS -> rows."""
-        # dequantize + crop to the real channel count + transpose NHWC -> NCHW
-        # without temporaries: copyto casts int8 -> fp32 (lossless) while it
-        # does the strided transpose copy into the reused scratch feed, then
-        # the zp/scale dequant math runs in place on the contiguous NCHW result
-        feeds = {}
-        for feed_i, name in enumerate(self._pp_input_names):
-            idx = self._feed_index[feed_i]
-            obuf = out_bufs[idx]
-            scale, zp = self._head_dequant[idx]
-            real = self._head_real[idx]
-            dst = self._pp_scratch[name]
-            np.copyto(dst, obuf[..., :real].transpose(0, 3, 1, 2), casting="unsafe")
-            np.subtract(dst, zp, out=dst)
-            np.multiply(dst, scale, out=dst)
-            feeds[name] = dst
-
-        pp_out = self._pp.run(None, feeds)
-        if len(pp_out) == 2:
-            # two outputs: class scores (sigmoid baked into the graph) + xyxy
-            class_scores, boxes = pp_out
-            scores = class_scores[0]  # (8400, 80)
-            boxes = boxes[0]
-            # one argmax pass over the class axis; conf gathers the argmaxed
-            # cells instead of a second full max(axis=1) reduction
-            cls_idx = scores.argmax(axis=1).astype(np.int32)
-            conf = scores[np.arange(cls_idx.shape[0]), cls_idx]
-        else:
-            # one fused output: [cx, cy, w, h, (objectness), C class scores].
-            # The SDK DecodeYolo emits CENTER-format boxes (confirmed by the
-            # vendor's axruntime_yolo11_onnxruntime.py example), absolute in
-            # the model input space.
-            raw = pp_out[0][0]
-            cx, cy, w, h = raw[:, 0], raw[:, 1], raw[:, 2], raw[:, 3]
-            boxes = np.stack([cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2], axis=1)
-            cls = raw[:, 4:]
-            if self._model_type == ModelTypeEnum.yolox:
-                cls = raw[:, 5:] * raw[:, 4:5]  # YOLOX: score = obj * cls
-            if cls.shape[1] == 0:
-                return np.zeros((MAX_ROWS, 6), np.float32)
-            cls_idx = cls.argmax(axis=1).astype(np.int32)
-            conf = cls[np.arange(cls_idx.shape[0]), cls_idx]
-        keep = conf > CONF_THRESHOLD
-        boxes, conf, cls_idx = boxes[keep], conf[keep], cls_idx[keep]
-        if len(conf) == 0:
-            return np.zeros((MAX_ROWS, 6), np.float32)
-        # boxes are already in the frame space: the input is not letterboxed
-        # (width/height must equal the compiled input size)
-        x0i = np.clip(boxes[:, 0], 0, self._width)
-        x1i = np.clip(boxes[:, 2], 0, self._width)
-        y0i = np.clip(boxes[:, 1], 0, self._height)
-        y1i = np.clip(boxes[:, 3], 0, self._height)
-        # NMS on the model-space boxes (scale-invariant IoU), per class
-        keep_nms = _nms_per_class(boxes, conf, cls_idx, NMS_IOU_THRESHOLD)
-        rows = []
-        for i in keep_nms:
-            score = float(conf[i])
-            rows.append(
-                (
-                    int(cls_idx[i]),
-                    score,
-                    float(y0i[i]) / self._height,
-                    float(x0i[i]) / self._width,
-                    float(y1i[i]) / self._height,
-                    float(x1i[i]) / self._width,
-                )
-            )
-        rows.sort(key=lambda r: r[1], reverse=True)
-        out = np.zeros((MAX_ROWS, 6), np.float32)
-        n = min(len(rows), MAX_ROWS)
-        for i in range(n):
-            out[i] = rows[i]
-        return out
-
-
-# ----------------- AxeleraDetector Class ----------------- #
-
-
-class AxeleraDetector(DetectionApi):
-    type_key = DETECTOR_KEY
-    supported_models = [
-        ModelTypeEnum.yolox,
-        ModelTypeEnum.yolonas,
-        ModelTypeEnum.yologeneric,
-    ]
-    runtime_manifest = AXELERA_MANIFEST
-
-    def __init__(self, detector_config: "AxeleraDetectorConfig"):
-        model = detector_config.model
-        if model is None:
-            raise ValueError(
-                "axelera: the model block is required (a .zip of the compiled "
-                "model output, local or http(s) URL)."
-            )
-        # the runtime wheels install into the user site at first start
-        self.activate_dependencies()
-        _ensure_runtime_enumeration_tools()
-        _point_runtime_at_installed_firmware()
-        super().__init__(detector_config)
-
-        self.width = model.width
-        self.height = model.height
-
-        self.working_model = _resolve_model_json(model.path)
-
-        self._inference = _AxeleraRuntimeInference(
-            model_json=self.working_model,
-            model_type=model.model_type,
-            width=self.width,
-            height=self.height,
-            pixel_format=str(
-                getattr(model.input_pixel_format, "value", model.input_pixel_format)
-            ),
-        )
-        logger.info(
-            "axelera: runtime model=%s (%dx%d input, frame %dx%d) ready",
-            self.working_model,
-            self._inference._model_w,
-            self._inference._model_h,
-            self.width,
-            self.height,
-        )
-
-    # --- async API (AsyncDetectorRunner, same contract as the memryx plugin) ---
-
-    def send_input(self, connection_id, tensor_input: np.ndarray) -> None:
-        """Preprocess (if needed) and send a frame to the Metis runtime."""
-        if tensor_input is None:
-            raise ValueError("[send_input] No image data provided for inference")
-
-        frame = np.asarray(tensor_input)
-        if frame.ndim == 4:
-            # Frigate feeds a batched NHWC frame (1, H, W, C); drop the batch dim
-            if frame.shape[0] != 1:
-                raise ValueError(f"Unexpected batch dimension in frame: {frame.shape}")
-            frame = frame[0]
-        # BGR uint8 for the model (resize/quantization happen in the
-        # worker). This must be a REAL copy: the tensor backs the camera's
-        # shared memory, which the runner unmaps once this frame's result is
-        # collected, potentially before the producer thread reads a queued
-        # frame (np.ascontiguousarray would be a no-op on a contiguous view).
-        frame = np.array(frame, dtype=np.uint8, copy=True)
-        self._inference.submit(connection_id, frame)
-
-    def receive_output(self):
-        """Return (connection_id, rows) for the next completed inference."""
-        return self._inference.collect()
-
-    def shutdown(self) -> None:
-        """Gracefully shut down the runtime and release the Metis device."""
-        self._inference.stop()
-
-    def detect_raw(self, tensor_input: np.ndarray):
-        """Synchronous path unused: axelera uses the async detector contract."""
-        return 0  # type: ignore[override]
-
-    def __del__(self):
-        # destructor must not raise
-        try:
-            if getattr(self, "_inference", None) is not None:
-                self.shutdown()
-        except Exception:  # noqa: S110, BLE001
-            pass
-
-
-def _ensure_runtime_enumeration_tools() -> None:
-    """Provide minimal lspci/lsmod stand-ins when the image lacks them.
-
-    The runtime's list_devices() shells out to ``lspci`` (PCI count, matched
-    against the driver count) and ``lsmod`` (AIPU driver presence) through
-    popen and greps their output. Both only read /sys and /proc, which are
-    readable in the container, so when the real tools are absent (slim image)
-    the plugin supplies shell stand-ins in a directory it owns and prepends
-    that to PATH. The real tools always win when installed.
-    """
-    if shutil.which("lspci") is not None and shutil.which("lsmod") is not None:
-        return
-    bin_dir = os.path.join(MODEL_CACHE_DIR, DETECTOR_KEY, "bin")
-    try:
-        os.makedirs(bin_dir, exist_ok=True)
-        scripts = {
-            # lsmod(8)-shaped output built from /proc/modules
-            "lsmod": (
-                "#!/bin/sh\n"
-                'echo "Module                  Size  Used by"\n'
-                "while read -r name size used _; do\n"
-                '    printf \'%-19s %8s  %s\\n\' "$name" "$size" "$used"\n'
-                "done < /proc/modules\n"
-            ),
-            # lspci(8)-shaped Device lines from sysfs; the runtime matches the
-            # vendor:device hex pair (1f9d for Metis)
-            "lspci": (
-                "#!/bin/sh\n"
-                "for dev in /sys/bus/pci/devices/*; do\n"
-                '    [ -r "$dev/vendor" ] || continue\n'
-                '    read -r vendor < "$dev/vendor"\n'
-                '    read -r device < "$dev/device"\n'
-                "    printf '%s Device %s:%s\\n' \"${dev##*/}\" "
-                '"${vendor#0x}" "${device#0x}"\n'
-                "done\n"
-            ),
-        }
-        for name, body in scripts.items():
-            if shutil.which(name) is not None:
-                continue
-            path = os.path.join(bin_dir, name)
-            with open(path, "w") as f:
-                f.write(body)
-            os.chmod(path, 0o755)
-        os.environ["PATH"] = bin_dir + os.pathsep + os.environ.get("PATH", "")
-        logger.debug("axelera: prepended runtime tools dir %s to PATH", bin_dir)
-    except OSError as exc:
-        logger.warning(
-            "axelera: could not install runtime enumeration helpers in %s: %s",
-            bin_dir,
-            exc,
-        )
-
-
-def _point_runtime_at_installed_firmware() -> None:
-    """Export the firmware paths the runtime's native library looks for.
-
-    The pip-installed layout keeps the AIPU device firmware under the user
-    site's ``axelera/omega`` directory (from the axelera-firmware wheel). The
-    native device library resolves firmware through ``AXELERA_DEVICE_DIR`` (and
-    the explicit stage0/elf variables), and has no default inside the Frigate
-    image, so point it at the installed copy before the runtime is imported.
-    """
-    if os.environ.get("AXELERA_DEVICE_DIR"):
-        return
-    omega = os.path.join(str(user_site()), "axelera", "omega")
-    if not os.path.isdir(omega):
-        return
-    os.environ["AXELERA_DEVICE_DIR"] = omega
-    os.environ.setdefault(
-        "AIPU_FIRMWARE_OMEGA", os.path.join(omega, "bin", "start_axelera_runtime.elf")
-    )
-    os.environ.setdefault(
-        "AIPU_RUNTIME_STAGE0_OMEGA",
-        os.path.join(omega, "bin", "start_axelera_runtime_stage0.bin"),
-    )
-
-
-# ----------------- AxeleraDetectorConfig Class ----------------- #
-
-
-class AxeleraDetectorConfig(BaseDetectorConfig):
-    model_config = ConfigDict(title="Axelera Metis")
-
-    # the Metis card can only be opened by one process
-    shareable: ClassVar[bool] = False
-
-    type: Literal[DETECTOR_KEY]
-    device: str = Field(
-        default="PCIe",
-        title="Device Path",
-        description="The Metis device to run inference on (e.g. 'PCIe').",
-    )

frigate/object_detection/base.py

@@ -277,8 +277,7 @@ class AsyncDetectorRunner(FrigateProcess):
         self._frame_manager = SharedMemoryFrameManager()
         self._publisher = ObjectDetectorPublisher()
         self._detector = AsyncLocalObjectDetector(
-            detector_config=self.detector_config,
+            detector_config=self.detector_config, stop_event=self.stop_event
-            stop_event=self.stop_event,
         )
 
         for name in self.cameras:
@@ -367,8 +366,8 @@ class ObjectDetectProcess:
 
         self.detection_start.value = 0.0
 
-        # Async path for MemryX and Axelera (send/receive contract)
+        # Async path for MemryX
-        if self.detector_config.type in ("memryx", "axelera"):
+        if self.detector_config.type == "memryx":
             self.detect_process = AsyncDetectorRunner(
                 f"frigate.detector:{self.name}",
                 self.detection_queue,

frigate/output/output.py

@@ -14,11 +14,12 @@ from ws4py.server.wsgirefserver import (
     WebSocketWSGIRequestHandler,
     WSGIServer,
 )
+from ws4py.server.wsgiutils import WebSocketWSGIApplication
 
 from frigate.comms.config_updater import ConfigSubscriber
 from frigate.comms.detections_updater import DetectionSubscriber, DetectionTypeEnum
-from frigate.comms.ws import OriginCheckedWebSocketApplication, WebSocket
+from frigate.comms.ws import WebSocket
-from frigate.config import AuthConfig, FrigateConfig
+from frigate.config import FrigateConfig
 from frigate.config.camera.updater import (
     CameraConfigUpdateEnum,
     CameraConfigUpdateSubscriber,
@@ -135,7 +136,7 @@ class OutputProcess(FrigateProcess):
             8082,
             server_class=WSGIServer,
             handler_class=WebSocketWSGIRequestHandler,
-            app=OriginCheckedWebSocketApplication(handler_cls=WebSocket),
+            app=WebSocketWSGIApplication(handler_cls=WebSocket),
         )
         websocket_server.initialize_websockets_manager()
         websocket_thread = threading.Thread(target=websocket_server.serve_forever)
@@ -153,7 +154,6 @@ class OutputProcess(FrigateProcess):
             ],
         )
         birdseye_config_subscriber = ConfigSubscriber("config/birdseye", exact=True)
-        auth_config_subscriber = ConfigSubscriber("config/auth", exact=True)
 
         jsmpeg_cameras: dict[str, JsmpegCamera] = {}
         birdseye: Birdseye | None = None
@@ -194,15 +194,6 @@ class OutputProcess(FrigateProcess):
                 self.config.birdseye = birdseye_config
                 logger.debug("Applied dynamic birdseye config update")
 
-            # user endpoints publish None on this topic and role edits an AuthConfig
-            while True:
-                auth_topic, auth_config = auth_config_subscriber.check_for_update()
-                if auth_topic is None:
-                    break
-                if isinstance(auth_config, AuthConfig):
-                    self.config.auth = auth_config
-                    logger.debug("Applied dynamic auth config update")
-
             # drain review updates every iteration, not just when birdseye is
             # being consumed, so a dropped end never strands a camera
             if birdseye is not None:
@@ -339,7 +330,6 @@ class OutputProcess(FrigateProcess):
 
         config_subscriber.stop()
         birdseye_config_subscriber.stop()
-        auth_config_subscriber.stop()
         websocket_server.manager.close_all()
         websocket_server.manager.stop()
         websocket_server.manager.join()

frigate/record/cleanup.py

@@ -198,52 +198,56 @@ class RecordingCleanup(threading.Thread):
         deleted_recordings = set()
         kept_recordings: list[tuple[float, float]] = []
         for recording in recordings:
-            keep = False
+            modes: set[RetainModeEnum] = set()
-            mode = None
+            # Now look for every reason to keep this recording segment
-            # Now look for a reason to keep this recording segment
             for idx in range(review_start, len(reviews)):
                 review = reviews[idx]
                 severity = review.severity
                 pre_capture = config.record.get_review_pre_capture(severity)
                 post_capture = config.record.get_review_post_capture(severity)
 
-                # if the review starts in the future, stop checking reviews
+                # stop once no later review's pre capture can reach this segment
-                # and let this recording segment expire
+                if (
-                if review.start_time - pre_capture > recording.end_time:
+                    review.start_time - config.record.event_pre_capture
-                    keep = False
+                    > recording.end_time
+                ):
                     break
 
-                # if the review is in progress or ends after the recording starts, keep it
+                if review.start_time - pre_capture > recording.end_time:
-                # and stop looking at reviews
+                    continue
+
+                # collect every review that is in progress or ends after the
+                # recording starts, since their windows can overlap
                 if (
                     review.end_time is None
                     or review.end_time + post_capture >= recording.start_time
                 ):
-                    keep = True
+                    modes.add(
-                    mode = (
                         alerts_retain_mode
                         if review.severity == "alert"
                         else detections_retain_mode
                     )
-                    break
 
                 # if the review ends before this recording segment starts, skip
-                # this review and check the next review for an overlap.
+                # it on future segments too. reviews and recordings are sorted,
-                # since the review and recordings are sorted, we can skip review
+                # but an earlier review with a longer post capture may still
-                # that end before the previous recording segment started on future segments
+                # overlap, so only skip while nothing has overlapped yet
-                if review.end_time + post_capture < recording.start_time:
+                elif not modes:
                     review_start = idx
 
-            # Delete recordings outside of the retention window or based on the retention mode
+            # Delete recordings unless an overlapping review's retention mode keeps them
-            if (
+            if not any(
-                not keep
+                mode == RetainModeEnum.all
                 or (
                     mode == RetainModeEnum.motion
-                    and recording.motion == 0
+                    and (
-                    and recording.objects == 0
+                        recording.motion != 0
-                    and recording.dBFS == 0
+                        or recording.objects != 0
+                        or recording.dBFS != 0
                     )
-                or (mode == RetainModeEnum.active_objects and recording.objects == 0)
+                )
+                or (mode == RetainModeEnum.active_objects and recording.objects != 0)
+                for mode in modes
             ):
                 recording_path = Path(recording.path)
                 recording_path.unlink(missing_ok=True)

frigate/record/maintainer.py

@@ -759,39 +759,40 @@ class RecordingMaintainer(threading.Thread):
 
         # we fell through the continuous / motion check, so we need to check the review items
         # if the cached segment overlaps with the review items:
-        overlaps = False
+        modes: set[RetainModeEnum] = set()
         for review in reviews:
             severity = SeverityEnum[review.severity]
 
-            # if the review item starts in the future, stop checking review items
+            # stop once no later review item's pre capture can reach this segment
-            # and remove this segment
+            if (
+                review.start_time - record_config.event_pre_capture
+            ) > end_time.timestamp():
+                break
+
             if (
                 review.start_time - record_config.get_review_pre_capture(severity)
             ) > end_time.timestamp():
-                overlaps = False
+                continue
-                break
 
-            # if the review item is in progress or ends after the recording starts, keep it
+            # collect every review item that is in progress or ends after the
-            # and stop looking at review items
+            # recording starts, since their windows can overlap
             if (
                 review.end_time is None
                 or (review.end_time + record_config.get_review_post_capture(severity))
                 >= start_time.timestamp()
             ):
-                overlaps = True
+                modes.add(
-                break
-
-        if overlaps:
-            record_mode = (
                     alerts_retain_mode
                     if review.severity == "alert"
                     else detections_retain_mode
                 )
 
+        if modes:
             if segment_stats is None:
                 segment_stats = self.segment_stats(camera, start_time, end_time)
 
-            if not segment_stats.should_discard_segment(record_mode):
+            # keep the segment if any overlapping review item's mode keeps it
+            if any(not segment_stats.should_discard_segment(m) for m in modes):
                 # move from cache to recordings immediately
                 return await self.move_segment(
                     camera,

frigate/stats/hardware.py

@@ -21,7 +21,6 @@ from frigate.detectors.hardware import DEV_ROOT, hardware_prober
 from frigate.util.services import (
     get_amd_gpu_stats,
     get_axcl_npu_stats,
-    get_axelera_board_temp,
     get_cpu_stats,
     get_hailo_temps,
     get_intel_gpu_stats,
@@ -482,7 +481,5 @@ def get_hardware_temperatures(detector_type: str) -> list[float | None]:
     elif detector_type == "hailo":
         hailo_temps = get_hailo_temps()
         return [hailo_temps[name] for name in sorted(hailo_temps.keys())]
-    elif detector_type == "axelera":
-        return [get_axelera_board_temp()]
 
     return []

frigate/test/http_api/test_http_auth_internal_port.py

@@ -101,71 +101,6 @@ class TestAuthInternalPort(BaseTestHttp):
             self.assertEqual(resp.headers["remote-user"], "anonymous")
             self.assertEqual(resp.headers["remote-role"], "admin")
 
-    def _internal_auth(self, original_url: str, **headers: str):
-        app = self._create_app()
-
-        with AuthTestClient(app) as client:
-            return client.get(
-                "/auth",
-                headers={
-                    "x-server-port": "5000",
-                    "x-original-url": original_url,
-                    "x-original-host": "frigate.lan:5000",
-                    **{k.replace("_", "-"): v for k, v in headers.items()},
-                },
-            )
-
-    def test_cross_origin_websocket_on_internal_port_denied(self):
-        resp = self._internal_auth(
-            "/ws", origin="https://evil.example", x_original_upgrade="websocket"
-        )
-        self.assertEqual(resp.status_code, 403)
-
-    def test_cross_origin_go2rtc_stream_denied(self):
-        resp = self._internal_auth(
-            "/live/mse/api/ws?src=front_door",
-            origin="https://evil.example",
-            x_original_upgrade="websocket",
-            x_go2rtc_stream_proxy="1",
-        )
-        self.assertEqual(resp.status_code, 403)
-
-    def test_cross_origin_webrtc_post_denied(self):
-        # the offer POST is not an upgrade, so the stream marker alone gates it
-        resp = self._internal_auth(
-            "/api/go2rtc/webrtc?src=front_door",
-            origin="https://evil.example",
-            x_go2rtc_stream_proxy="1",
-        )
-        self.assertEqual(resp.status_code, 403)
-
-    def test_same_origin_websocket_allowed(self):
-        resp = self._internal_auth(
-            "/live/mse/api/ws?src=front_door",
-            origin="http://frigate.lan:5000",
-            x_original_upgrade="websocket",
-            x_go2rtc_stream_proxy="1",
-        )
-        self.assertEqual(resp.status_code, 202)
-
-    def test_websocket_without_origin_allowed(self):
-        resp = self._internal_auth("/ws", x_original_upgrade="websocket")
-        self.assertEqual(resp.status_code, 202)
-
-    def test_forwarded_host_origin_allowed(self):
-        # Home Assistant integration proxy: Host is Frigate, Origin is HA
-        resp = self._internal_auth(
-            "/api/go2rtc/webrtc?src=front_door",
-            origin="https://ha.example.com:8123",
-            x_forwarded_host="ha.example.com:8123",
-            x_go2rtc_stream_proxy="1",
-        )
-        self.assertEqual(resp.status_code, 202)
-
-    def test_cross_origin_plain_http_not_checked(self):
-        resp = self._internal_auth("/api/events", origin="https://evil.example")
-        self.assertEqual(resp.status_code, 202)
-
     def test_external_port_requires_auth(self):
         app = self._create_app()
 

frigate/test/test_axelera_detector.py

@@ -1,558 +0,0 @@
-"""Offline unit tests for the Axelera Metis detector plugin.
-
-The runtime and onnxruntime objects are constructor-injected, so these cover
-the decode paths (fused cxcywh and two-output scores+boxes), the raw class
-index contract, the zip-only model resolution (download, fresh extraction,
-rejection of anything else), and the zip extraction escape guard, without
-the vendor SDK or a Metis card.
-"""
-
-import json
-import os
-import shutil
-import sys
-import tempfile
-import types
-import unittest
-import zipfile
-from typing import ClassVar
-from unittest.mock import patch
-
-import numpy as np
-
-from frigate.detectors.detector_config import ModelTypeEnum
-from frigate.detectors.plugins import axelera as ax
-
-MANIFEST = {
-    "input_shapes": [[1, 642, 656, 4]],
-    "input_shapes_original": [[1, 3, 640, 640]],
-    "n_padded_ch_inputs": [[0, 1, 1, 641, 0, 656, 0, 4]],
-    "output_shapes": [[1, 80, 80, 128]],
-    "output_shapes_original": [[1, 85, 80, 80]],
-    "dequantize_params": [[0.01, -128]],
-}
-
-
-def _make_compiled_model(root: str) -> str:
-    """Write a fake compiled-model tree, return the model_json path."""
-    kernel = os.path.join(root, "fakenet", "1")
-    os.makedirs(kernel)
-    with open(os.path.join(kernel, "manifest.json"), "w") as f:
-        json.dump(MANIFEST, f)
-    with open(os.path.join(kernel, "model.json"), "w") as f:
-        json.dump({"inputs": []}, f)
-    with open(os.path.join(kernel, "postprocess_graph.onnx"), "wb") as f:
-        f.write(b"fake")
-    return os.path.join(kernel, "model.json")
-
-
-def _make_model_zip(zip_path: str, kernel_payload: bytes = b"fake") -> None:
-    """Write a compiled-model zip: fakenet/1/{model.json,manifest.json,onnx}."""
-    with zipfile.ZipFile(zip_path, "w") as archive:
-        archive.writestr("fakenet/1/model.json", json.dumps({"inputs": []}))
-        archive.writestr("fakenet/1/manifest.json", json.dumps(MANIFEST))
-        archive.writestr("fakenet/1/postprocess_graph.onnx", kernel_payload)
-
-
-class _FakeInput:
-    def __init__(self, name, shape):
-        self.name = name
-        self.shape = shape
-
-
-class _FakeInstance:
-    def run(self, ins, outs):
-        pass
-
-
-class _FakeConn:
-    def load_model_instance(self, model, num_sub_devices=1, aipu_cores=1):
-        return _FakeInstance()
-
-
-class _FakeContext:
-    def load_model(self, path):
-        return object()
-
-    def list_devices(self):
-        return ["metis-0:1:0"]
-
-    def device_connect(self, *_args, **_kwargs):
-        return _FakeConn()
-
-    def release(self):
-        pass
-
-
-class _FakeSessionOptions:
-    def __init__(self):
-        self.intra_op_num_threads = 0
-
-
-class _FakeOnnx:
-    """Postprocess-graph fake with a programmable output.
-
-    ``outputs`` is what ``run`` returns; the feed dict keys/shapes are
-    recorded so tests can assert the plugin dequantized and laid out the
-    heads correctly before feeding the graph. Mirrors the onnxruntime
-    module surface the plugin uses: SessionOptions() and InferenceSession().
-    """
-
-    session = None
-    outputs: list = []
-
-    def __init__(self, path, sess_options=None, providers=None):
-        self._inputs = [_FakeInput("head0", [1, 85, 80, 80])]
-        _FakeOnnx.session = self
-
-    def get_inputs(self):
-        return self._inputs
-
-    def run(self, _out, feeds):
-        self.feeds = feeds
-        return _FakeOnnx.outputs
-
-    @staticmethod
-    def SessionOptions():
-        return _FakeSessionOptions()
-
-    @staticmethod
-    def InferenceSession(path, sess_options=None, providers=None):
-        return _FakeOnnx(path, sess_options, providers)
-
-
-class _FakeModelDownloader:
-    """The ModelDownloader stand-in bound into the stubbed downloader module."""
-
-    calls: ClassVar[list] = []
-
-    @staticmethod
-    def download_from_url(download_url, save_path, silent=False):
-        _FakeModelDownloader.calls.append((download_url, save_path))
-        _make_model_zip(save_path)
-
-
-def _make_engine(root: str, outputs, model_type=ModelTypeEnum.yolox):
-    model_json = _make_compiled_model(root)
-    _FakeOnnx.outputs = outputs
-    return ax._AxeleraRuntimeInference(
-        model_json=model_json,
-        model_type=model_type,
-        width=640,
-        height=640,
-        runtime_factory=_FakeContext,
-        onnx_factory=_FakeOnnx,
-    )
-
-
-def _fused_row(cx, cy, w, h, obj, cls0):
-    """Build a (1, 8400, 85) fused output with one detection in row 0."""
-    out = np.zeros((1, 8400, 85), np.float32)
-    out[0, 0, 0:5] = [cx, cy, w, h, obj]
-    out[0, 0, 5] = cls0
-    return [out]
-
-
-class TestDecodeFused(unittest.TestCase):
-    """One fused graph output: [cx, cy, w, h, obj, 80 class scores]."""
-
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory()
-        # a box centered in the 640 model space: model xyxy [288, 288, 352, 352]
-        self.engine = _make_engine(
-            self.tmp.name, _fused_row(320, 320, 64, 64, 1.0, 0.9)
-        )
-
-    def tearDown(self):
-        self.tmp.cleanup()
-
-    def test_dequantizes_heads_into_graph_feed(self):
-        out_bufs = [np.full((1, 80, 80, 128), -128, np.int8)]
-        self.engine._decode(out_bufs)
-        ((name, feed),) = _FakeOnnx.session.feeds.items()
-        self.assertEqual(name, "head0")
-        self.assertEqual(feed.shape, (1, 85, 80, 80))
-        # (q - zero_point) * scale with q=-128, zp=-128, scale=0.01
-        self.assertAlmostEqual(float(feed[0, 0, 0, 0]), 0.0, places=6)
-
-    def test_row_shape_and_dtype(self):
-        out_bufs = [np.zeros((1, 80, 80, 128), np.int8)]
-        rows = self.engine._decode(out_bufs)
-        self.assertEqual(rows.shape, (ax.MAX_ROWS, 6))
-        self.assertEqual(rows.dtype, np.float32)
-
-    def test_boxes_map_straight_to_the_frame_space(self):
-        # no letterbox: the frame is the model input (640x640), so model
-        # xyxy maps to normalized frame coords 1:1
-        out_bufs = [np.zeros((1, 80, 80, 128), np.int8)]
-        rows = self.engine._decode(out_bufs)
-        class_id, score, ymin, xmin, ymax, xmax = rows[0]
-        self.assertEqual(int(class_id), 0)  # raw class index, not a name
-        self.assertAlmostEqual(float(score), 0.9, places=5)  # obj * cls
-        self.assertAlmostEqual(float(xmin), 288 / 640, places=4)
-        self.assertAlmostEqual(float(xmax), 352 / 640, places=4)
-        self.assertAlmostEqual(float(ymin), 288 / 640, places=4)
-        self.assertAlmostEqual(float(ymax), 352 / 640, places=4)
-        self.assertLess(float(ymin), float(ymax))
-        self.assertLess(float(xmin), float(xmax))
-
-    def test_class_index_passes_through_unmapped(self):
-        # detections carry the raw class index; Frigate's labelmap names it
-        out = np.zeros((1, 8400, 85), np.float32)
-        out[0, 0, 0:5] = [320, 320, 64, 64, 1.0]
-        out[0, 0, 5 + 5] = 0.9  # class 5
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            engine = _make_engine(tmp.name, [out])
-            rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)])
-            self.assertEqual(int(rows[0][0]), 5)
-        finally:
-            tmp.cleanup()
-
-    def test_yolox_score_multiplies_objectness(self):
-        out = _fused_row(320, 320, 64, 64, 0.5, 0.9)  # obj 0.5 * cls 0.9
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            engine = _make_engine(tmp.name, out, ModelTypeEnum.yolox)
-            rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)])
-            self.assertAlmostEqual(float(rows[0][1]), 0.45, places=5)
-        finally:
-            tmp.cleanup()
-
-    def test_yolonas_score_skips_objectness(self):
-        out = np.zeros((1, 8400, 84), np.float32)
-        out[0, 0, :4] = [320, 320, 64, 64]
-        out[0, 0, 4] = 0.9  # class 0, with no objectness column
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            engine = _make_engine(tmp.name, [out], ModelTypeEnum.yolonas)
-            rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)])
-            self.assertEqual(int(rows[0][0]), 0)
-            self.assertAlmostEqual(float(rows[0][1]), 0.9, places=5)
-        finally:
-            tmp.cleanup()
-
-    def test_below_conf_threshold_yields_zero_rows(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            engine = _make_engine(tmp.name, _fused_row(320, 320, 64, 64, 1.0, 0.1))
-            rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)])
-            self.assertEqual(float(np.abs(rows).sum()), 0.0)
-        finally:
-            tmp.cleanup()
-
-
-class TestDecodeTwoOutputs(unittest.TestCase):
-    """Two graph outputs: (1, 8400, 80) class scores + (1, 8400, 4) xyxy."""
-
-    def test_scores_and_boxes_branch(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            scores = np.zeros((1, 8400, 80), np.float32)
-            scores[0, 7, 0] = 0.8  # person at row 7
-            boxes = np.zeros((1, 8400, 4), np.float32)
-            boxes[0, 7] = [288, 288, 352, 352]  # xyxy, absolute model space
-            engine = _make_engine(tmp.name, [scores, boxes])
-            rows = engine._decode([np.zeros((1, 80, 80, 128), np.int8)])
-            self.assertEqual(int(rows[0][0]), 0)
-            self.assertAlmostEqual(float(rows[0][1]), 0.8, places=5)
-            self.assertAlmostEqual(float(rows[0][2]), 288 / 640, places=4)
-            self.assertAlmostEqual(float(rows[0][4]), 352 / 640, places=4)
-        finally:
-            tmp.cleanup()
-
-
-class TestModelDimensions(unittest.TestCase):
-    """width/height must equal the compiled input; no letterboxing."""
-
-    def test_mismatched_dimensions_raise(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            model_json = _make_compiled_model(tmp.name)
-            with self.assertRaises(ValueError):
-                ax._AxeleraRuntimeInference(
-                    model_json=model_json,
-                    model_type=ModelTypeEnum.yolox,
-                    width=704,
-                    height=576,
-                    runtime_factory=_FakeContext,
-                    onnx_factory=_FakeOnnx,
-                )
-        finally:
-            tmp.cleanup()
-
-    def test_matching_dimensions_initialize(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            model_json = _make_compiled_model(tmp.name)
-            engine = ax._AxeleraRuntimeInference(
-                model_json=model_json,
-                model_type=ModelTypeEnum.yolox,
-                width=640,
-                height=640,
-                runtime_factory=_FakeContext,
-                onnx_factory=_FakeOnnx,
-            )
-            self.assertEqual((engine._width, engine._height), (640, 640))
-        finally:
-            tmp.cleanup()
-
-
-class TestZipGuard(unittest.TestCase):
-    def test_member_escaping_target_raises(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            zip_path = os.path.join(tmp.name, "evil.zip")
-            with zipfile.ZipFile(zip_path, "w") as zf:
-                zf.writestr("../escaped.txt", "boom")
-            with self.assertRaises(ValueError):
-                ax._extract_model_zip(zip_path, os.path.join(tmp.name, "out"))
-            self.assertFalse(os.path.exists(os.path.join(tmp.name, "escaped.txt")))
-        finally:
-            tmp.cleanup()
-
-    def test_normal_members_extract(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            zip_path = os.path.join(tmp.name, "ok.zip")
-            with zipfile.ZipFile(zip_path, "w") as zf:
-                zf.writestr("net/1/model.json", "{}")
-                zf.writestr("net/1/manifest.json", "{}")
-            dest = os.path.join(tmp.name, "out")
-            ax._extract_model_zip(zip_path, dest)
-            self.assertTrue(
-                os.path.isfile(os.path.join(dest, "net", "1", "model.json"))
-            )
-        finally:
-            tmp.cleanup()
-
-
-class TestFindModelJson(unittest.TestCase):
-    def test_direct_and_wrapper_dir(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            root = tmp.name
-            # direct: root/1/model.json
-            os.makedirs(os.path.join(root, "direct", "1"))
-            direct = os.path.join(root, "direct", "1", "model.json")
-            open(direct, "w").write("{}")
-            self.assertEqual(ax._find_model_json(os.path.join(root, "direct")), direct)
-            # single wrapper dir: root/wrap/net/1/model.json
-            wrapped = os.path.join(root, "wrap", "net", "1")
-            os.makedirs(wrapped)
-            hit = os.path.join(wrapped, "model.json")
-            open(hit, "w").write("{}")
-            self.assertEqual(ax._find_model_json(os.path.join(root, "wrap")), hit)
-        finally:
-            tmp.cleanup()
-
-    def test_ambiguous_subdirs_raise(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            for net in ("aaa", "bbb"):
-                sub = os.path.join(tmp.name, net, "1")
-                os.makedirs(sub)
-                open(os.path.join(sub, "model.json"), "w").write("{}")
-            with self.assertRaises(ValueError):
-                ax._find_model_json(tmp.name)
-        finally:
-            tmp.cleanup()
-
-    def test_missing_raises_filenotfound(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            with self.assertRaises(FileNotFoundError):
-                ax._find_model_json(tmp.name)
-        finally:
-            tmp.cleanup()
-
-
-class TestRuntimeToolShims(unittest.TestCase):
-    def test_ensure_runtime_enumeration_tools_only_installs_missing(self):
-        tmp = tempfile.TemporaryDirectory()
-        saved_path = os.environ["PATH"]
-        # resolve sh before PATH is emptied: the shim check below must still
-        # find a shell once PATH points only at the empty temp dir
-        sh_bin = shutil.which("sh") or "/bin/sh"
-        try:
-            os.environ["PATH"] = tmp.name  # nothing on PATH
-            with tempfile.TemporaryDirectory() as cache:
-                orig_cache = ax.MODEL_CACHE_DIR
-                ax.MODEL_CACHE_DIR = cache
-                try:
-                    ax._ensure_runtime_enumeration_tools()
-                finally:
-                    ax.MODEL_CACHE_DIR = orig_cache
-                bindir = os.path.join(cache, "axelera", "bin")
-                for tool in ("lspci", "lsmod"):
-                    self.assertTrue(os.access(os.path.join(bindir, tool), os.X_OK))
-                self.assertTrue(os.environ["PATH"].startswith(bindir))
-                # the shims are valid sh and produce their /proc, /sys reads
-                import subprocess
-
-                out = subprocess.run(
-                    [sh_bin, os.path.join(bindir, "lsmod")],
-                    capture_output=True,
-                    text=True,
-                )
-                self.assertEqual(out.returncode, 0)
-                self.assertIn("Module", out.stdout.splitlines()[0])
-        finally:
-            os.environ["PATH"] = saved_path
-            tmp.cleanup()
-
-
-class TestModelResolve(unittest.TestCase):
-    """`path` is a .zip of the compiled output, local or http(s) URL."""
-
-    def _resolve(self, path):
-        _FakeModelDownloader.calls = []
-        # the plugin imports the downloader lazily at the call site (its module
-        # chain pulls the IPC stack), so the test binds a stub module instead
-        stub = types.ModuleType("frigate.util.downloader")
-        setattr(stub, "ModelDownloader", _FakeModelDownloader)
-        tmp = tempfile.TemporaryDirectory()
-        saved_cache = ax.MODEL_CACHE_DIR
-        ax.MODEL_CACHE_DIR = os.path.join(tmp.name, "cache")
-        try:
-            with patch.dict(sys.modules, {"frigate.util.downloader": stub}):
-                got = ax._resolve_model_json(path)
-        finally:
-            ax.MODEL_CACHE_DIR = saved_cache
-            tmp.cleanup()
-        return _FakeModelDownloader.calls, got
-
-    def test_https_url_downloads_and_resolves(self):
-        calls, got = self._resolve("https://example.com/models/fakenet.zip")
-
-        self.assertEqual(calls[0][0], "https://example.com/models/fakenet.zip")
-        self.assertTrue(calls[0][1].endswith(os.path.join("axelera", "fakenet.zip")))
-        self.assertTrue(got.endswith(os.path.join("fakenet", "1", "model.json")))
-
-    def test_url_without_filename_falls_back_to_default_name(self):
-        calls, _ = self._resolve("https://example.com/")
-
-        self.assertTrue(calls[0][1].endswith("yolox-s-coco-onnx.zip"))
-
-    def test_scheme_less_host_is_rejected(self):
-        # no scheme means a local path; a host-looking string is not a file
-        with self.assertRaises(FileNotFoundError):
-            self._resolve("www.example.com/models/fakenet.zip")
-
-    def test_local_zip_resolves(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            zip_path = os.path.join(tmp.name, "fakenet.zip")
-            _make_model_zip(zip_path)
-            _FakeModelDownloader.calls = []
-            stub = types.ModuleType("frigate.util.downloader")
-            setattr(stub, "ModelDownloader", _FakeModelDownloader)
-            saved_cache = ax.MODEL_CACHE_DIR
-            ax.MODEL_CACHE_DIR = os.path.join(tmp.name, "cache")
-            try:
-                with patch.dict(sys.modules, {"frigate.util.downloader": stub}):
-                    got = ax._resolve_model_json(zip_path)
-            finally:
-                ax.MODEL_CACHE_DIR = saved_cache
-            self.assertEqual(_FakeModelDownloader.calls, [])
-            self.assertTrue(got.endswith(os.path.join("fakenet", "1", "model.json")))
-        finally:
-            tmp.cleanup()
-
-    def test_non_zip_path_is_rejected(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            model_json = _make_compiled_model(tmp.name)
-            with self.assertRaises(ValueError):
-                self._resolve(model_json)
-            with self.assertRaises(ValueError):
-                self._resolve(os.path.join(tmp.name, "fakenet"))
-        finally:
-            tmp.cleanup()
-
-    def test_missing_local_zip_raises_filenotfound(self):
-        with self.assertRaises(FileNotFoundError):
-            self._resolve("/nope/fakenet.zip")
-
-
-class TestFreshExtract(unittest.TestCase):
-    """The selected archive extracts fresh on every start."""
-
-    def _resolve(self, path, tmp_name):
-        stub = types.ModuleType("frigate.util.downloader")
-        setattr(stub, "ModelDownloader", _FakeModelDownloader)
-        saved_cache = ax.MODEL_CACHE_DIR
-        ax.MODEL_CACHE_DIR = os.path.join(tmp_name, "cache")
-        try:
-            with patch.dict(sys.modules, {"frigate.util.downloader": stub}):
-                return ax._resolve_model_json(path)
-        finally:
-            ax.MODEL_CACHE_DIR = saved_cache
-
-    def test_replaced_archive_becomes_the_model_that_runs(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            _FakeModelDownloader.calls = []
-            self._resolve("https://example.com/models/fakenet.zip", tmp.name)
-            zip_path = _FakeModelDownloader.calls[0][1]
-            # the archive at the cache path is replaced (re-fetched upstream,
-            # a pre-seeded image, a mirror refresh)
-            _make_model_zip(zip_path, kernel_payload=b"replaced")
-            self._resolve("https://example.com/models/fakenet.zip", tmp.name)
-
-            extract_dir = os.path.join(
-                tmp.name, "cache", "axelera", "fakenet-extracted"
-            )
-            with open(
-                os.path.join(extract_dir, "fakenet", "1", "postprocess_graph.onnx"),
-                "rb",
-            ) as f:
-                self.assertEqual(f.read(), b"replaced")
-        finally:
-            tmp.cleanup()
-
-    def test_same_archive_does_not_download_twice(self):
-        tmp = tempfile.TemporaryDirectory()
-        try:
-            _FakeModelDownloader.calls = []
-            self._resolve("https://example.com/models/fakenet.zip", tmp.name)
-            self.assertEqual(len(_FakeModelDownloader.calls), 1)
-            self._resolve("https://example.com/models/fakenet.zip", tmp.name)
-
-            # the zip is cached (no second download), the extract is fresh
-            self.assertEqual(len(_FakeModelDownloader.calls), 1)
-        finally:
-            tmp.cleanup()
-
-
-class TestBackpressure(unittest.TestCase):
-    """Overload keeps the 1 submission : 1 result runner contract."""
-
-    def _engine(self):
-        tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(tmp.cleanup)
-        return _make_engine(tmp.name, [])
-
-    def test_full_input_queue_still_delivers_zero_rows(self):
-        engine = self._engine()
-        frame = np.zeros((1, 640, 640, 3), np.uint8)
-        for i in range(ax.QUEUE_MAXSIZE):
-            engine._frame_q.put_nowait((f"cam{i}", frame))
-
-        overflow = "overflow-cam"
-        engine.submit(overflow, frame)
-
-        cid, rows = engine._result_q.get_nowait()
-        self.assertEqual(cid, overflow)
-        self.assertEqual(rows.shape, (ax.MAX_ROWS, 6))
-        self.assertTrue(np.all(rows == 0))
-
-    def test_stop_ends_collect_and_releases_context(self):
-        engine = self._engine()
-        engine.stop()
-
-        self.assertEqual(engine.collect(), (None, None))
-
-
-if __name__ == "__main__":
-    unittest.main()

frigate/test/test_detector_hardware.py

@@ -227,14 +227,6 @@ class TestAccelerators(HardwareProbeTestCase):
     def test_no_deepx_is_reported_without_a_node(self):
         self.assertNotIn("deepx", self.probe())
 
-    def test_an_axelera_metis_is_a_single_unit(self):
-        write(os.path.join(self.dev_root, "metis0"))
-
-        axelera = self.probe()["axelera"]
-
-        self.assertEqual(axelera.units[0].device, "axelera:PCIe")
-        self.assertEqual(axelera.count, 1)
-
     def test_a_supported_rockchip_soc_is_reported(self):
         write(
             os.path.join(self.proc_root, "device-tree", "compatible"),

frigate/test/test_detector_stats.py

@@ -34,41 +34,6 @@ class TestDetectorTemperatures(unittest.TestCase):
         self.assertEqual(stats["hailo:PCIe#2"]["temperature"], 50.0)
         self.assertEqual(stats["hailo:PCIe:1"]["temperature"], 60.0)
 
-    def test_axelera_temperature_reads_the_board_controller(self):
-        from frigate.stats.hardware import get_hardware_temperatures
-
-        with patch("frigate.stats.hardware.get_axelera_board_temp", return_value=45.4):
-            self.assertEqual(get_hardware_temperatures("axelera"), [45.4])
-
-        with patch("frigate.stats.hardware.get_axelera_board_temp", return_value=None):
-            self.assertEqual(get_hardware_temperatures("axelera"), [None])
-
-
-class TestAxeleraBoardTempCooldown(unittest.TestCase):
-    """A failed board read backs off instead of delaying every stats update."""
-
-    def test_failed_read_starts_a_cooldown(self):
-        import subprocess as sp
-
-        import frigate.util.services as services
-
-        with (
-            patch.object(services, "_axelera_board_temp_cache", None),
-            patch.object(services, "_axelera_board_temp_error_until", 0.0),
-            patch("frigate.util.runtime_deps.find_tool", return_value="/bin/true"),
-            patch.object(
-                services.sp,
-                "run",
-                side_effect=sp.TimeoutExpired(cmd="axcmd", timeout=5),
-            ) as run_mock,
-        ):
-            self.assertIsNone(services.get_axelera_board_temp())
-            # the cooldown prevents the second update from paying the
-            # blocking read cost again while the board is unresponsive
-            self.assertIsNone(services.get_axelera_board_temp())
-
-        self.assertEqual(run_mock.call_count, 1)
-
 
 if __name__ == "__main__":
     unittest.main()

frigate/test/test_maintainer.py

@@ -127,6 +127,116 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
         self.assertIsNone(result)
         maintainer.drop_segment.assert_called_once_with(cache_path)
 
+    async def test_quiet_segment_kept_by_later_overlapping_alert(self):
+        config = MagicMock(spec=FrigateConfig)
+
+        camera_config = MagicMock()
+        camera_config.record.enabled = True
+        camera_config.record.continuous.days = 0
+        camera_config.record.motion.days = 0
+        camera_config.record.alerts.retain.mode = RetainModeEnum.all
+        camera_config.record.detections.retain.mode = RetainModeEnum.motion
+        camera_config.record.event_pre_capture = 60
+        camera_config.record.get_review_pre_capture.return_value = 60
+        camera_config.record.get_review_post_capture.return_value = 60
+        config.cameras = {"test_cam": camera_config}
+
+        maintainer = RecordingMaintainer(config, MagicMock())
+
+        now = datetime.datetime.now(datetime.UTC)
+        start_time = now - datetime.timedelta(seconds=20)
+        end_time = now - datetime.timedelta(seconds=10)
+        cache_path = "/tmp/cache/test_cam@20260417150000+0000.mp4"
+
+        maintainer.end_time_cache = {
+            cache_path: (end_time, 10.0, False, None, None, None, [])
+        }
+        maintainer.object_recordings_info["test_cam"] = [(now.timestamp(), [], [], [])]
+        maintainer.audio_recordings_info["test_cam"] = []
+
+        maintainer.drop_segment = MagicMock()
+        maintainer.move_segment = AsyncMock(return_value=None)
+        maintainer.recordings_publisher = MagicMock()
+
+        # the detection's post capture and the alert's window both cover the segment
+        detection = MagicMock()
+        detection.severity = "detection"
+        detection.start_time = start_time.timestamp() - 50
+        detection.end_time = start_time.timestamp() - 40
+        alert = MagicMock()
+        alert.severity = "alert"
+        alert.start_time = start_time.timestamp() - 10
+        alert.end_time = start_time.timestamp()
+
+        await maintainer.validate_and_move_segment(
+            "test_cam",
+            reviews=[detection, alert],
+            recording={
+                "start_time": start_time,
+                "cache_path": cache_path,
+                "stream_type": "main",
+            },
+        )
+
+        maintainer.move_segment.assert_awaited_once()
+        maintainer.drop_segment.assert_not_called()
+
+    async def test_quiet_segment_kept_by_later_alert_with_longer_pre_capture(self):
+        config = MagicMock(spec=FrigateConfig)
+
+        camera_config = MagicMock()
+        camera_config.record.enabled = True
+        camera_config.record.continuous.days = 0
+        camera_config.record.motion.days = 0
+        camera_config.record.alerts.retain.mode = RetainModeEnum.all
+        camera_config.record.detections.retain.mode = RetainModeEnum.all
+        camera_config.record.event_pre_capture = 30
+        camera_config.record.get_review_pre_capture.side_effect = lambda s: (
+            30 if s == "alert" else 5
+        )
+        camera_config.record.get_review_post_capture.return_value = 5
+        config.cameras = {"test_cam": camera_config}
+
+        maintainer = RecordingMaintainer(config, MagicMock())
+
+        now = datetime.datetime.now(datetime.UTC)
+        start_time = now - datetime.timedelta(seconds=20)
+        end_time = now - datetime.timedelta(seconds=10)
+        cache_path = "/tmp/cache/test_cam@20260417150000+0000.mp4"
+
+        maintainer.end_time_cache = {
+            cache_path: (end_time, 10.0, False, None, None, None, [])
+        }
+        maintainer.object_recordings_info["test_cam"] = [(now.timestamp(), [], [], [])]
+        maintainer.audio_recordings_info["test_cam"] = []
+
+        maintainer.drop_segment = MagicMock()
+        maintainer.move_segment = AsyncMock(return_value=None)
+        maintainer.recordings_publisher = MagicMock()
+
+        # the detection's pre capture starts after the segment, the alert's reaches it
+        detection = MagicMock()
+        detection.severity = "detection"
+        detection.start_time = end_time.timestamp() + 10
+        detection.end_time = end_time.timestamp() + 12
+        alert = MagicMock()
+        alert.severity = "alert"
+        alert.start_time = end_time.timestamp() + 20
+        alert.end_time = end_time.timestamp() + 25
+
+        await maintainer.validate_and_move_segment(
+            "test_cam",
+            reviews=[detection, alert],
+            recording={
+                "start_time": start_time,
+                "cache_path": cache_path,
+                "stream_type": "main",
+            },
+        )
+
+        maintainer.move_segment.assert_awaited_once()
+        maintainer.drop_segment.assert_not_called()
+
     async def test_defers_review_overlap_segment_until_metadata_catches_up(self):
         # Regression: a segment overlapping an active_objects review must not
         # be dropped while detection metadata lags behind the segment end,
@@ -139,6 +249,7 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
         camera_config.record.continuous.days = 7
         camera_config.record.motion.days = 0
         camera_config.record.alerts.retain.mode = RetainModeEnum.active_objects
+        camera_config.record.event_pre_capture = 5
         camera_config.record.get_review_pre_capture.return_value = 5
         camera_config.record.get_review_post_capture.return_value = 5
         config.cameras = {"test_cam": camera_config}

frigate/test/test_output_ws_auth.py

@@ -3,7 +3,7 @@
 import unittest
 from types import SimpleNamespace
 
-from frigate.config import AuthConfig, FrigateConfig
+from frigate.config import FrigateConfig
 from frigate.output.ws_auth import ws_has_camera_access
 
 
@@ -55,11 +55,3 @@ class TestWsHasCameraAccess(unittest.TestCase):
         self.assertFalse(
             ws_has_camera_access(self._make_ws("limited_user"), "birdseye", self.config)
         )
-
-    def test_replaced_auth_config_revokes_camera(self):
-        ws = self._make_ws("limited_user")
-        self.assertTrue(ws_has_camera_access(ws, "front_door", self.config))
-
-        self.config.auth = AuthConfig(roles={"limited_user": ["back_door"]})
-        self.assertFalse(ws_has_camera_access(ws, "front_door", self.config))
-        self.assertTrue(ws_has_camera_access(ws, "back_door", self.config))

frigate/test/test_record_cleanup_reviews.py

@@ -0,0 +1,142 @@
+"""Tests for review-based retention in recording cleanup."""
+
+import datetime
+import unittest
+from unittest.mock import MagicMock
+
+from peewee import SqliteDatabase
+
+from frigate.config import FrigateConfig
+from frigate.models import Previews, Recordings, ReviewSegment, UserReviewStatus
+from frigate.record.cleanup import RecordingCleanup
+
+
+class TestRecordingCleanupReviewOverlap(unittest.TestCase):
+    def setUp(self):
+        self.db = SqliteDatabase(":memory:")
+        models = [Previews, Recordings, ReviewSegment, UserReviewStatus]
+        self.db.bind(models)
+        self.db.create_tables(models)
+        self.base = datetime.datetime.now().timestamp() - 3600
+
+    def tearDown(self):
+        self.db.close()
+
+    def _build_cleanup(self, alerts: dict, detections: dict) -> RecordingCleanup:
+        config = FrigateConfig(
+            **{
+                "mqtt": {"host": "mqtt"},
+                "cameras": {
+                    "front_door": {
+                        "ffmpeg": {
+                            "inputs": [
+                                {
+                                    "path": "rtsp://10.0.0.1:554/video",
+                                    "roles": ["detect", "record"],
+                                }
+                            ]
+                        },
+                        "record": {
+                            "enabled": True,
+                            "continuous": {"days": 0},
+                            "motion": {"days": 0},
+                            "alerts": alerts,
+                            "detections": detections,
+                        },
+                    }
+                },
+            }
+        )
+        return RecordingCleanup(config, MagicMock())
+
+    def _insert_review(self, id: str, severity: str, start: float, end: float):
+        ReviewSegment.create(
+            id=id,
+            camera="front_door",
+            start_time=self.base + start,
+            end_time=self.base + end,
+            severity=severity,
+            thumb_path=f"/media/frigate/clips/review/{id}.webp",
+            data={},
+        )
+
+    def _insert_recording(self, id: str, start: float):
+        Recordings.create(
+            id=id,
+            camera="front_door",
+            path=f"/media/frigate/recordings/{id}.mp4",
+            start_time=self.base + start,
+            end_time=self.base + start + 10,
+            duration=10,
+            motion=0,
+            objects=0,
+            dBFS=0,
+            segment_size=0,
+            stream_type="main",
+        )
+
+    def test_quiet_segment_kept_by_later_overlapping_alert(self):
+        cleanup = self._build_cleanup(
+            alerts={"pre_capture": 60, "post_capture": 60, "retain": {"mode": "all"}},
+            detections={"post_capture": 60, "retain": {"mode": "motion"}},
+        )
+        self._insert_review("detection", "detection", 90, 100)
+        self._insert_review("alert", "alert", 130, 140)
+        self._insert_recording("quiet", 140)
+
+        cleanup.expire_recordings()
+
+        assert Recordings.get_or_none(Recordings.id == "quiet") is not None
+
+    def test_quiet_segment_dropped_when_no_overlapping_mode_keeps_it(self):
+        cleanup = self._build_cleanup(
+            alerts={
+                "pre_capture": 60,
+                "post_capture": 60,
+                "retain": {"mode": "motion"},
+            },
+            detections={"post_capture": 60, "retain": {"mode": "motion"}},
+        )
+        self._insert_review("detection", "detection", 90, 100)
+        self._insert_review("alert", "alert", 130, 140)
+        self._insert_recording("quiet", 140)
+
+        cleanup.expire_recordings()
+
+        assert Recordings.get_or_none(Recordings.id == "quiet") is None
+
+    def test_longer_pre_capture_survives_an_earlier_shorter_review(self):
+        # the detection starts after the segment even with its pre capture,
+        # but the later alert's longer pre capture still reaches it
+        cleanup = self._build_cleanup(
+            alerts={"pre_capture": 30, "retain": {"mode": "all"}},
+            detections={"pre_capture": 5, "retain": {"mode": "all"}},
+        )
+        self._insert_review("detection", "detection", 10, 12)
+        self._insert_review("alert", "alert", 20, 25)
+        self._insert_recording("early", -6)
+
+        cleanup.expire_recordings()
+
+        assert Recordings.get_or_none(Recordings.id == "early") is not None
+
+    def test_longer_post_capture_survives_a_later_shorter_review(self):
+        # the alert's window ends before the detection's, so skipping past
+        # it must not skip the detection for later segments
+        cleanup = self._build_cleanup(
+            alerts={"post_capture": 5, "retain": {"mode": "motion"}},
+            detections={"post_capture": 60, "retain": {"mode": "all"}},
+        )
+        self._insert_review("detection", "detection", 90, 100)
+        self._insert_review("alert", "alert", 110, 115)
+        self._insert_recording("first", 125)
+        self._insert_recording("second", 135)
+
+        cleanup.expire_recordings()
+
+        assert Recordings.get_or_none(Recordings.id == "first") is not None
+        assert Recordings.get_or_none(Recordings.id == "second") is not None
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_ws_origin.py

@@ -1,126 +0,0 @@
-"""Tests for the same-origin check on WebSocket handshakes."""
-
-import unittest
-from unittest.mock import MagicMock
-
-from frigate.comms.ws import OriginCheckedWebSocketApplication, WebSocket
-from frigate.util.origin import origin_allowed
-
-
-def ws_origin_allowed(environ: dict[str, str]) -> bool:
-    return origin_allowed(
-        environ.get("HTTP_ORIGIN"),
-        environ.get("HTTP_HOST"),
-        environ.get("HTTP_X_FORWARDED_HOST"),
-    )
-
-
-class TestOriginAllowed(unittest.TestCase):
-    def test_missing_origin_allowed(self):
-        self.assertTrue(ws_origin_allowed({"HTTP_HOST": "frigate.lan"}))
-
-    def test_direct_access_same_origin(self):
-        # nginx forwards Host without the port and X-Forwarded-Host with it
-        environ = {
-            "HTTP_ORIGIN": "http://192.168.1.10:5000",
-            "HTTP_HOST": "192.168.1.10",
-            "HTTP_X_FORWARDED_HOST": "192.168.1.10:5000",
-        }
-        self.assertTrue(ws_origin_allowed(environ))
-
-    def test_default_port_origin(self):
-        environ = {
-            "HTTP_ORIGIN": "https://frigate.example.com",
-            "HTTP_HOST": "frigate.example.com",
-        }
-        self.assertTrue(ws_origin_allowed(environ))
-
-    def test_host_comparison_is_case_insensitive(self):
-        environ = {
-            "HTTP_ORIGIN": "https://Frigate.Example.com",
-            "HTTP_HOST": "frigate.example.COM",
-        }
-        self.assertTrue(ws_origin_allowed(environ))
-
-    def test_ipv6_host(self):
-        environ = {"HTTP_ORIGIN": "http://[::1]:8971", "HTTP_HOST": "[::1]"}
-        self.assertTrue(ws_origin_allowed(environ))
-
-    def test_cross_origin_rejected(self):
-        environ = {
-            "HTTP_ORIGIN": "https://evil.example",
-            "HTTP_HOST": "frigate.lan",
-            "HTTP_X_FORWARDED_HOST": "frigate.lan:5000",
-        }
-        self.assertFalse(ws_origin_allowed(environ))
-
-    def test_null_origin_rejected(self):
-        self.assertFalse(
-            ws_origin_allowed({"HTTP_ORIGIN": "null", "HTTP_HOST": "frigate.lan"})
-        )
-
-    def test_proxy_rewriting_host_with_forwarded_host(self):
-        environ = {
-            "HTTP_ORIGIN": "https://frigate.example.com",
-            "HTTP_HOST": "10.0.0.5",
-            "HTTP_X_FORWARDED_HOST": "frigate.example.com",
-        }
-        self.assertTrue(ws_origin_allowed(environ))
-
-    def test_forwarded_host_list_uses_first_entry(self):
-        environ = {
-            "HTTP_ORIGIN": "https://frigate.example.com",
-            "HTTP_HOST": "10.0.0.5",
-            "HTTP_X_FORWARDED_HOST": "frigate.example.com, proxy.internal",
-        }
-        self.assertTrue(ws_origin_allowed(environ))
-
-        environ["HTTP_ORIGIN"] = "https://proxy.internal"
-        self.assertFalse(ws_origin_allowed(environ))
-
-    def test_home_assistant_ingress(self):
-        # HA core copies Host and sets X-Forwarded-Host to its own host
-        environ = {
-            "HTTP_ORIGIN": "https://ha.example.com:8123",
-            "HTTP_HOST": "ha.example.com",
-            "HTTP_X_FORWARDED_HOST": "ha.example.com:8123",
-            "HTTP_X_INGRESS_PATH": "/api/hassio_ingress/abc123",
-        }
-        self.assertTrue(ws_origin_allowed(environ))
-
-    def test_ingress_path_does_not_bypass_origin_check(self):
-        environ = {
-            "HTTP_ORIGIN": "https://evil.example",
-            "HTTP_HOST": "frigate.example.com",
-            "HTTP_X_INGRESS_PATH": "/frigate",
-        }
-        self.assertFalse(ws_origin_allowed(environ))
-
-    def test_home_assistant_integration_proxy(self):
-        # hass-web-proxy-lib drops Host, so Host is the integration's Frigate URL
-        environ = {
-            "HTTP_ORIGIN": "https://ha.example.com:8123",
-            "HTTP_HOST": "frigate.lan",
-            "HTTP_X_FORWARDED_HOST": "ha.example.com:8123",
-        }
-        self.assertTrue(ws_origin_allowed(environ))
-
-
-class TestOriginCheckedWebSocketApplication(unittest.TestCase):
-    def test_cross_origin_handshake_gets_403(self):
-        app = OriginCheckedWebSocketApplication(handler_cls=WebSocket)
-        start_response = MagicMock()
-        environ = {
-            "REQUEST_METHOD": "GET",
-            "HTTP_ORIGIN": "https://evil.example",
-            "HTTP_HOST": "frigate.lan",
-        }
-
-        self.assertEqual(app(environ, start_response), [])
-        start_response.assert_called_once()
-        self.assertEqual(start_response.call_args.args[0], "403 Forbidden")
-        self.assertNotIn("ws4py.websocket", environ)
-
-
-if __name__ == "__main__":
-    unittest.main()

frigate/util/origin.py

@@ -1,31 +0,0 @@
-"""Same-origin check for browser WebSocket handshakes and go2rtc stream requests."""
-
-from urllib.parse import urlsplit
-
-
-def _hostname(value: str) -> str | None:
-    try:
-        return urlsplit(value).hostname
-    except ValueError:
-        return None
-
-
-def origin_allowed(
-    origin: str | None, host: str | None, forwarded_host: str | None
-) -> bool:
-    """Return True unless a browser sent the request from a foreign origin.
-
-    Browsers always send Origin on a WebSocket handshake, so a missing Origin
-    means a non-browser client. A page cannot add headers to a handshake, so
-    X-Forwarded-Host comes from a proxy. Ports are ignored because nginx
-    forwards Host without one.
-    """
-    if origin is None:
-        return True
-
-    origin_host = _hostname(origin)
-    if not origin_host:
-        return False
-
-    hosts = ((forwarded_host or "").split(",")[0].strip(), host or "")
-    return any(origin_host == _hostname(f"//{h}") for h in hosts if h)

frigate/util/services.py

@@ -842,83 +842,6 @@ def get_axcl_npu_stats() -> dict[str, str | float] | None:
     return None
 
 
-_axelera_board_temp_cache: tuple[float, float] | None = None
-_axelera_board_temp_error_until = 0.0
-_AXELERA_BOARD_TEMP_TTL = 5.0
-_AXELERA_BOARD_TEMP_ERROR_COOLDOWN = 60.0
-
-
-def get_axelera_board_temp() -> float | None:
-    """Get the Axelera Metis board temperature through the SDK's axcmd tool.
-
-    The read goes through the board controller and does not open an inference
-    stream, so it is safe to call while a detector is running. ``axcmd`` ships
-    with the runtime wheel; ``find_tool`` prefers the runtime-installed copy
-    under the user base. Results are cached for a few seconds so one stats tick
-    spawns axcmd at most once, and a failed read starts a cooldown so an
-    unresponsive board does not delay every stats update.
-
-    NOTE: this function is blocking (up to ~5s on a hung board).
-    """
-    global _axelera_board_temp_cache
-    global _axelera_board_temp_error_until
-
-    now = time.monotonic()
-    if (
-        _axelera_board_temp_cache is not None
-        and now - _axelera_board_temp_cache[0] < _AXELERA_BOARD_TEMP_TTL
-    ):
-        return _axelera_board_temp_cache[1]
-
-    if now < _axelera_board_temp_error_until:
-        # the last read already cost up to 5s; back off instead of paying
-        # that cost again on every stats update while the board is down
-        return None
-
-    try:
-        from frigate.util.runtime_deps import find_tool
-
-        axcmd_path = find_tool("axcmd")
-    except Exception:  # noqa: BLE001
-        axcmd_path = shutil.which("axcmd")
-
-    if axcmd_path is None or not os.path.exists(axcmd_path):
-        return None
-
-    try:
-        result = sp.run(
-            [axcmd_path, "--board-temp"],
-            capture_output=True,
-            text=True,
-            timeout=5,
-            check=False,
-        )
-    except (OSError, sp.TimeoutExpired):
-        _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN
-        return None
-
-    if result.returncode != 0:
-        _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN
-        return None
-
-    # anchor on the board-controller line ("N: temp_sensorX@YY = <value>") so
-    # unrelated "= <number>" output (firmware/version preambles) is not picked up
-    match = re.search(r"temp_sensor[^=\n]*=\s*([\d.]+)", result.stdout)
-
-    if match is None:
-        _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN
-        return None
-
-    try:
-        temp = round(float(match.group(1)), 1)
-    except ValueError:
-        _axelera_board_temp_error_until = now + _AXELERA_BOARD_TEMP_ERROR_COOLDOWN
-        return None
-
-    _axelera_board_temp_cache = (now, temp)
-    return temp
-
-
 def try_get_info(f, h, default="N/A", sensor=None):
     try:
         if h:

web/vite.config.ts

@@ -28,6 +28,7 @@ export default defineConfig({
       },
       "/live": {
         target: `ws://${proxyHost}`,
+        changeOrigin: true,
         ws: true,
       },
     },