mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-20 21:52:50 +03:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 dependabot[bot]andGitHub
|
c74b366316 | ||
|
Patryk OrwatGitHubPatryk Watkowskigreptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> |
91e91c68ef | ||
|
Josh HawkinsandGitHub
|
5bb2cde8b3 |
@@ -38,6 +38,7 @@ device_globs=(
|
||||
"/dev/nvidia*"
|
||||
"/dev/memx*"
|
||||
"/dev/dxrt*"
|
||||
"/dev/metis*"
|
||||
)
|
||||
|
||||
IFS=',' read -ra extra_globs <<< "${DEVICE_ACL_PATHS:-}"
|
||||
|
||||
@@ -20,6 +20,11 @@ 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;
|
||||
|
||||
@@ -54,6 +54,13 @@ 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;
|
||||
|
||||
@@ -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 $http_host;
|
||||
proxy_set_header X-Forwarded-Host $forwarded_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;
|
||||
|
||||
@@ -1278,3 +1278,34 @@ 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
|
||||
|
||||
@@ -360,6 +360,7 @@ For example, in Nginx:
|
||||
|
||||
```
|
||||
location /frigate {
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Ingress-Path /frigate;
|
||||
proxy_pass http://frigate_backend;
|
||||
}
|
||||
|
||||
@@ -63,6 +63,10 @@ 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.
|
||||
@@ -100,7 +104,7 @@ models:
|
||||
- openvino:GPU
|
||||
```
|
||||
|
||||
Coral EdgeTPU and MemryX accelerators can only be opened by one process, so those devices can not be repeated.
|
||||
Coral EdgeTPU, MemryX, and Axelera Metis accelerators can only be opened by one process, so those devices can not be repeated.
|
||||
|
||||
### Running more than one model
|
||||
|
||||
@@ -940,3 +944,41 @@ 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
|
||||
```
|
||||
|
||||
@@ -596,6 +596,60 @@ 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.
|
||||
|
||||
@@ -65,11 +65,12 @@ 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`, or `axengine`).
|
||||
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`).
|
||||
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.
|
||||
|
||||
@@ -37,6 +37,8 @@ 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.
|
||||
@@ -161,6 +163,7 @@ 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;
|
||||
|
||||
@@ -55,6 +55,10 @@ 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/`.
|
||||
|
||||
+27
-1
@@ -39,6 +39,7 @@ 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__)
|
||||
|
||||
@@ -718,6 +719,17 @@ 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:
|
||||
@@ -741,7 +753,6 @@ 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
|
||||
@@ -1307,6 +1318,21 @@ 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:
|
||||
|
||||
+24
-1
@@ -40,6 +40,7 @@ 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__)
|
||||
|
||||
@@ -472,6 +473,28 @@ 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)
|
||||
@@ -621,7 +644,7 @@ class WebSocketClient(Communicator):
|
||||
5002,
|
||||
server_class=WSGIServer,
|
||||
handler_class=WebSocketWSGIRequestHandler,
|
||||
app=WebSocketWSGIApplication(handler_cls=_WebSocketHandler),
|
||||
app=OriginCheckedWebSocketApplication(handler_cls=_WebSocketHandler),
|
||||
)
|
||||
self.websocket_server.initialize_websockets_manager()
|
||||
self.websocket_thread = threading.Thread(
|
||||
|
||||
@@ -284,6 +284,19 @@ 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")
|
||||
@@ -344,6 +357,7 @@ PROBES = (
|
||||
detect_hailo,
|
||||
detect_memryx,
|
||||
detect_deepx,
|
||||
detect_axelera,
|
||||
detect_intel_npu,
|
||||
detect_intel_gpu,
|
||||
detect_nvidia_gpu,
|
||||
|
||||
@@ -0,0 +1,972 @@
|
||||
"""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').",
|
||||
)
|
||||
@@ -277,7 +277,8 @@ class AsyncDetectorRunner(FrigateProcess):
|
||||
self._frame_manager = SharedMemoryFrameManager()
|
||||
self._publisher = ObjectDetectorPublisher()
|
||||
self._detector = AsyncLocalObjectDetector(
|
||||
detector_config=self.detector_config, stop_event=self.stop_event
|
||||
detector_config=self.detector_config,
|
||||
stop_event=self.stop_event,
|
||||
)
|
||||
|
||||
for name in self.cameras:
|
||||
@@ -366,8 +367,8 @@ class ObjectDetectProcess:
|
||||
|
||||
self.detection_start.value = 0.0
|
||||
|
||||
# Async path for MemryX
|
||||
if self.detector_config.type == "memryx":
|
||||
# Async path for MemryX and Axelera (send/receive contract)
|
||||
if self.detector_config.type in ("memryx", "axelera"):
|
||||
self.detect_process = AsyncDetectorRunner(
|
||||
f"frigate.detector:{self.name}",
|
||||
self.detection_queue,
|
||||
|
||||
@@ -14,12 +14,11 @@ 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 WebSocket
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.comms.ws import OriginCheckedWebSocketApplication, WebSocket
|
||||
from frigate.config import AuthConfig, FrigateConfig
|
||||
from frigate.config.camera.updater import (
|
||||
CameraConfigUpdateEnum,
|
||||
CameraConfigUpdateSubscriber,
|
||||
@@ -136,7 +135,7 @@ class OutputProcess(FrigateProcess):
|
||||
8082,
|
||||
server_class=WSGIServer,
|
||||
handler_class=WebSocketWSGIRequestHandler,
|
||||
app=WebSocketWSGIApplication(handler_cls=WebSocket),
|
||||
app=OriginCheckedWebSocketApplication(handler_cls=WebSocket),
|
||||
)
|
||||
websocket_server.initialize_websockets_manager()
|
||||
websocket_thread = threading.Thread(target=websocket_server.serve_forever)
|
||||
@@ -154,6 +153,7 @@ 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,6 +194,15 @@ 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:
|
||||
@@ -330,6 +339,7 @@ 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()
|
||||
|
||||
@@ -21,6 +21,7 @@ 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,
|
||||
@@ -481,5 +482,7 @@ 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 []
|
||||
|
||||
@@ -101,6 +101,71 @@ 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()
|
||||
|
||||
|
||||
Executable
+558
@@ -0,0 +1,558 @@
|
||||
"""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()
|
||||
@@ -227,6 +227,14 @@ 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"),
|
||||
|
||||
@@ -34,6 +34,41 @@ 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()
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.config import AuthConfig, FrigateConfig
|
||||
from frigate.output.ws_auth import ws_has_camera_access
|
||||
|
||||
|
||||
@@ -55,3 +55,11 @@ 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))
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,31 @@
|
||||
"""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)
|
||||
@@ -842,6 +842,83 @@ 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:
|
||||
|
||||
Generated
+3
-3
@@ -5498,9 +5498,9 @@
|
||||
"license": "ISC"
|
||||
},
|
||||
"node_modules/follow-redirects": {
|
||||
"version": "1.16.0",
|
||||
"resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.16.0.tgz",
|
||||
"integrity": "sha512-y5rN/uOsadFT/JfYwhxRS5R7Qce+g3zG97+JrtFZlC9klX/W5hD7iiLzScI4nZqUS7DNUdhPgw4xI8W2LuXlUw==",
|
||||
"version": "1.16.1",
|
||||
"resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.16.1.tgz",
|
||||
"integrity": "sha512-FNvFGzoMLWmE6Yj9spb/zjd7yiNCHiAW9/Tg9CXrQ8wuu32HtlJOwWO11OJafl5FfY3DxTdQ0vj42zU1kvv5jg==",
|
||||
"funding": [
|
||||
{
|
||||
"type": "individual",
|
||||
|
||||
@@ -28,7 +28,6 @@ export default defineConfig({
|
||||
},
|
||||
"/live": {
|
||||
target: `ws://${proxyHost}`,
|
||||
changeOrigin: true,
|
||||
ws: true,
|
||||
},
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user