diff --git a/CODEOWNERS b/CODEOWNERS index c37041c2c3..e38d8e3ea7 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -5,3 +5,4 @@ /docker/rockchip/ @MarcA711 /docker/rocm/ @harakas /docker/hailo8l/ @spanner3003 +/docker/deepx/ @sixfab diff --git a/docker/deepx/user_installation.sh b/docker/deepx/user_installation.sh new file mode 100644 index 0000000000..2e9564d5ea --- /dev/null +++ b/docker/deepx/user_installation.sh @@ -0,0 +1,82 @@ +#!/bin/bash + +# Installs the DEEPX NPU kernel driver on the Docker host. +# +# The driver has to be built and loaded on the host rather than in the Frigate +# container: containers share the host kernel, so a kernel module cannot be +# loaded from inside one. Once loaded it exposes the NPU as /dev/dxrt* device +# nodes, which are passed through to the container (see the DEEPX section of +# the installation docs). +# +# A DEEPX install has three separately versioned pieces that have to agree: +# +# DX-RT runtime v3.4.2 the Frigate image, docker/main/install_dxrt.sh +# Kernel driver v2.6.0 this file (provides dxrt_driver and dx_dma) +# NPU firmware v2.7.4 the module itself, flashed from the host +# +# DX-RT v3.4.2 requires driver 2.5.0 and firmware 2.7.0 as a minimum, so the +# versions pinned here sit above that floor. A mismatch is the most common +# cause of inference requests that are accepted but never complete, and it +# fails silently rather than at startup. This script only installs the driver: +# reading or flashing the firmware version requires the DX-RT runtime on the +# host, which conflicts with the copy Frigate runs in its container, so that is +# a separate manual step covered in the installation docs. +# +# DEEPX NPU support in Frigate is developed and maintained by Sixfab +# (https://sixfab.com). + +set -euo pipefail + +driver_version="v2.6.0" +firmware_version="v2.7.4" + +sudo apt-get update +sudo apt-get install -y git build-essential "linux-headers-$(uname -r)" pciutils + +if ! lspci -d 1ff4: | grep -q .; then + echo "No DEEPX device found on the PCIe bus (lspci -d 1ff4:)." + echo "Check that the module is seated correctly before continuing." + exit 1 +fi + +git clone --depth 1 --branch "${driver_version}" \ + https://github.com/DEEPX-AI/dx_rt_npu_linux_driver.git +cd dx_rt_npu_linux_driver/modules + +# --reload unloads any running modules and loads the newly built ones, so the +# NPU is usable without a reboot. +sudo ./build.sh -c install --reload + +sudo depmod -A + +# dx_dma is the PCIe transport, dxrt_driver is the NPU driver on top of it. +for module in dx_dma dxrt_driver; do + if ! sudo modprobe "${module}"; then + echo "Unable to load the ${module} kernel module, common reasons are:" + echo "- Secure Boot is enabled and is rejecting the unsigned module." + echo "- The running kernel does not match the installed linux-headers." + exit 1 + fi +done + +if ! compgen -G "/dev/dxrt*" > /dev/null; then + echo "Modules loaded but no /dev/dxrt* device node appeared." + echo "Run ./sanity_check.sh from the driver repo to diagnose." + exit 1 +fi + +# dxrtd runs inside the Frigate container and may only run once per system, so +# a host copy left over from a runtime install has to be disabled. +if systemctl list-unit-files dxrt.service &> /dev/null; then + echo "Disabling the host dxrt.service so the container can run dxrtd." + sudo systemctl disable --now dxrt.service +fi + +echo "DEEPX driver installation complete." +echo "Driver version: $(modinfo -F version dxrt_driver)" +echo "Device node(s): $(echo /dev/dxrt*)" +echo +echo "This driver expects NPU firmware ${firmware_version}. The firmware version" +echo "can only be read with the DX-RT runtime installed on the host, so check it" +echo "before starting Frigate, then remove the host runtime: Frigate runs its own" +echo "copy of dxrtd and only one may run per system." diff --git a/docker/main/install_dxrt.sh b/docker/main/install_dxrt.sh new file mode 100644 index 0000000000..bee184ebc2 --- /dev/null +++ b/docker/main/install_dxrt.sh @@ -0,0 +1,55 @@ +#!/bin/bash + +set -euxo pipefail + +# DEEPX DX-RT runtime (dxrtd daemon + the dx_engine python bindings) used by +# the `deepx` detector plugin. +# +# A DEEPX install has three separately versioned pieces that have to agree: +# +# DX-RT runtime v3.4.2 this file, installed into the Frigate image +# Kernel driver v2.6.0 the host, docker/deepx/user_installation.sh +# NPU firmware v2.7.4 the module itself, flashed from the host +# +# DX-RT v3.4.2 requires driver 2.5.0 and firmware 2.7.0 as a minimum, so the +# versions above sit just over that floor. The runtime is pinned rather than +# resolved to "latest" so image builds stay reproducible, and so there is a +# known version for the host side to match. A mismatch usually presents as +# inference requests that are accepted but never complete, not as a startup +# error, so bump all three together and keep the table in +# docs/docs/frigate/installation.md in sync. +dxrt_version="v3.4.2" + +# The release directory and the package filename both use the bare version, +# while the git tag carries a "v". +dxrt_release="${dxrt_version#v}" +deb_arch=$(dpkg --print-architecture) +deb_file="/tmp/libdxrt-bin_${dxrt_release}_${deb_arch}.deb" + +# Fetch just the one package rather than the source archive: the repository +# keeps every past release under release/, so the tag archive is several +# hundred MB of .deb files we would immediately throw away. Note that +# release/latest is a symlink and would not survive the archive anyway. +wget -qO "${deb_file}" \ + "https://raw.githubusercontent.com/DEEPX-AI/dx_rt/${dxrt_version}/release/${dxrt_release}/libdxrt-bin_${dxrt_release}_${deb_arch}.deb" + +apt-get -qq update +apt-get -qq install -y "${deb_file}" +rm -rf /var/lib/apt/lists/* + +# The package bundles a dx_engine wheel per cpython ABI tag. Ask dpkg what it +# installed rather than hardcoding a path, so this keeps working if DEEPX moves +# the wheels between releases. +package=$(dpkg-deb -f "${deb_file}" Package) +py_tag="cp$(python3 -c 'import sys; print(f"{sys.version_info[0]}{sys.version_info[1]}")')" +wheel=$(dpkg -L "${package}" | grep -E "/dx_engine-.*\.whl$" | grep -- "${py_tag}" | sort | head -1 || true) + +if [[ -z "${wheel}" ]]; then + echo "No dx_engine wheel for ${py_tag} in ${package}. Wheels available:" + dpkg -L "${package}" | grep -E "/dx_engine-.*\.whl$" || echo " (none)" + exit 1 +fi + +pip3 install "${wheel}" + +rm -f "${deb_file}" diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/consumer-for b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/consumer-for new file mode 100644 index 0000000000..c8ae1cc5b9 --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/consumer-for @@ -0,0 +1 @@ +dxrtd diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/dependencies.d/log-prepare b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/dependencies.d/log-prepare new file mode 100644 index 0000000000..e69de29bb2 diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/pipeline-name b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/pipeline-name new file mode 100644 index 0000000000..55265a254e --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/pipeline-name @@ -0,0 +1 @@ +dxrtd-pipeline diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/run b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/run new file mode 100644 index 0000000000..a02011f742 --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/run @@ -0,0 +1,4 @@ +#!/command/with-contenv bash +# shellcheck shell=bash + +exec logutil-service /dev/shm/logs/dxrtd diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/type b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/type new file mode 100644 index 0000000000..5883cff0cd --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd-log/type @@ -0,0 +1 @@ +longrun diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/dependencies.d/prepare b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/dependencies.d/prepare new file mode 100644 index 0000000000..e69de29bb2 diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/finish b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/finish new file mode 100644 index 0000000000..fefd404cb2 --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/finish @@ -0,0 +1,33 @@ +#!/command/with-contenv bash +# shellcheck shell=bash +# Take down the S6 supervision tree when the service fails + +set -o errexit -o nounset -o pipefail + +# Logs should be sent to stdout so that s6 can collect them + +declare exit_code_container +exit_code_container=$(cat /run/s6-linux-init-container-results/exitcode) +readonly exit_code_container +readonly exit_code_service="${1}" +readonly exit_code_signal="${2}" +readonly service="dxrtd" + +echo "[INFO] Service ${service} exited with code ${exit_code_service} (by signal ${exit_code_signal})" + +# A clean exit means the service took itself down because no DEEPX hardware is +# present, which is the normal case on images without an NPU. Only a genuine +# failure should bring the container down. +if [[ "${exit_code_service}" -eq 256 ]]; then + if [[ "${exit_code_container}" -eq 0 ]]; then + echo $((128 + exit_code_signal)) >/run/s6-linux-init-container-results/exitcode + fi + if [[ "${exit_code_signal}" -eq 15 ]]; then + exec /run/s6/basedir/bin/halt + fi +elif [[ "${exit_code_service}" -ne 0 ]]; then + if [[ "${exit_code_container}" -eq 0 ]]; then + echo "${exit_code_service}" >/run/s6-linux-init-container-results/exitcode + fi + exec /run/s6/basedir/bin/halt +fi diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/producer-for b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/producer-for new file mode 100644 index 0000000000..29cfb55297 --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/producer-for @@ -0,0 +1 @@ +dxrtd-log diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/run b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/run new file mode 100644 index 0000000000..27ff22217c --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/run @@ -0,0 +1,32 @@ +#!/command/with-contenv bash +# shellcheck shell=bash +# Start the DEEPX DX-RT runtime daemon (dxrtd) + +set -o errexit -o nounset -o pipefail + +# Logs should be sent to stdout so that s6 can collect them +exec 2>&1 + +# The DEEPX kernel driver runs on the Docker host, not in this container, and +# exposes the NPU as /dev/dxrt* device nodes that are passed through. Without +# them there is no NPU to serve, so exit cleanly and let the rest of Frigate +# start normally: this service ships in the default image and most users do +# not have DEEPX hardware. +if ! compgen -G "/dev/dxrt*" > /dev/null; then + echo "[INFO] No DEEPX NPU device found, not starting dxrtd" + exec s6-svc -Od . +fi + +# dxrtd is a system-wide singleton. If the host's own dxrt.service is still +# running it will refuse to start, so surface that up front rather than +# leaving the user to decode the daemon's own error. +if pgrep -x dxrtd > /dev/null; then + echo "[ERROR] Another dxrtd instance is already running" + echo "[ERROR] dxrtd may only run once per system. Stop the host service with:" + echo "[ERROR] sudo systemctl disable --now dxrt.service" + exit 1 +fi + +echo "[INFO] Starting dxrtd for DEEPX NPU device(s): $(echo /dev/dxrt*)" + +exec /usr/local/bin/dxrtd diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/timeout-kill b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/timeout-kill new file mode 100644 index 0000000000..3a05c8b3ee --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/timeout-kill @@ -0,0 +1 @@ +30000 diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/type b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/type new file mode 100644 index 0000000000..5883cff0cd --- /dev/null +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/dxrtd/type @@ -0,0 +1 @@ +longrun diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/user/contents.d/dxrtd-pipeline b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/user/contents.d/dxrtd-pipeline new file mode 100644 index 0000000000..e69de29bb2 diff --git a/docs/docs/configuration/object_detectors.md b/docs/docs/configuration/object_detectors.md index 5dd163299b..42e78f6941 100644 --- a/docs/docs/configuration/object_detectors.md +++ b/docs/docs/configuration/object_detectors.md @@ -24,6 +24,7 @@ Frigate supports multiple different detectors that work on different types of ha - [Coral EdgeTPU](#edge-tpu-detector): The Google Coral EdgeTPU is available in USB, Mini PCIe, and m.2 formats allowing for a wide range of compatibility with devices. - [Hailo](#hailo-8): The Hailo8 and Hailo8L AI Acceleration module is available in m.2 format with a HAT for RPi devices, offering a wide range of compatibility with devices. - [MemryX](#memryx-mx3): The MX3 Acceleration module is available in m.2 format, offering broad compatibility across various platforms. +- [DEEPX](#deepx-npu): The DEEPX NPU is available in m.2 format and as a HAT+ for the Raspberry Pi 5, offering broad compatibility across various platforms. **AMD** @@ -636,6 +637,94 @@ For detailed instructions on compiling models, refer to the [MemryX Compiler](ht --- +## DEEPX NPU + +This detector is available for use with the DEEPX NPU, both the DX-M1 M.2 module and the DX-M1M on the Sixfab AI HAT+ for the Raspberry Pi 5. The configuration below applies unchanged to either form factor. DEEPX NPU support in Frigate is developed and maintained by [Sixfab](https://sixfab.com). + +See the [installation docs](../frigate/installation.md#deepx-npu) for information on installing the DEEPX kernel driver and passing the NPU through to the container. + +To configure a DEEPX detector, set the `type` attribute to `deepx` and follow the configuration guide below. + +### Configuration {#configuration-deepx} + + + +Frigate does not bundle a model for this detector. Models must be compiled to DEEPX's `.dxnn` format. + +The quickest way to get one is the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo), which publishes pre-compiled `.dxnn` files for a range of YOLO variants alongside a JSON file describing how each was compiled. Pick a YOLO detection model, download the `.dxnn`, bind-mount it into the container, and point `model.path` at it. Note the variant you picked, as it determines the `model_format` value below. + +Alternatively, compile your own model with the DX-COM compiler. + +`model.width` and `model.height` must match the resolution the model was compiled for. Quantization parameters are baked into the `.dxnn` file at compile time, so no normalization is applied on the host and Frigate's default `input_tensor`, `input_pixel_format`, and `input_dtype` values do not need to be overridden. + +#### Label maps + +Models from the DEEPX ModelZoo are trained on the standard 80-class COCO label set, so `labelmap_path` must be set to `/labelmap/coco-80.txt`. Frigate's default label map uses an extended 91-class scheme, and leaving it in place will cause detections to be reported as the wrong object type. + +#### Model formats + +DX-COM preserves the detection head of the model it compiles, so the YOLO variant determines how the output has to be read. Set `model_format` to the variant your `.dxnn` file was compiled from: + +```yaml +detectors: + deepx: + type: deepx + model_format: yolov8 +``` + +Supported values fall into three decode families: + +| `model_format` | Output layout | PPU support | +| ------------------------------------------ | ------------------------------------------------------------------------ | ----------- | +| `yolov5`, `yolov7` | Anchor-based, with a separate objectness score | Yes | +| `yolov8`, `yolov9`, `yolov11`, `yolov12` | Anchor-free, no objectness, boxes already in pixels | Yes | +| `yolov10`, `yolov26` | NMS runs in the detection head, output is final corner boxes | No | + +Leaving `model_format` unset falls back to inferring the layout from the output shape. That inference cannot distinguish an anchor-based head from an anchor-free one, so it is only reliable for YOLOv8 and newer. Setting it explicitly is recommended. + +Two mistakes this option prevents: + +- Decoding an anchor-free model with the anchor formula squares the box dimensions, producing boxes hundreds of thousands of pixels wide that clip to the entire frame. +- Decoding an anchor-based model as anchor-free reads the objectness column as a class score, so confidences and class IDs are both wrong. + +`yolov10` and `yolov26` run NMS inside the head, so `nms_threshold` does not apply to them. + +#### PPU models + +Models compiled with Post-Processing Unit (PPU) support perform candidate selection on the NPU itself and emit a list of surviving detections rather than raw feature maps. This leaves only the box decode and NMS for the host, which lowers CPU usage. + +Set `ppu: true` on the detector when using such a model: + +```yaml +detectors: + deepx: + type: deepx + model_format: yolov8 + ppu: true + score_threshold: 0.25 + nms_threshold: 0.45 +``` + +`ppu` and `model_format` are independent: `ppu` says whether candidate selection already happened on the NPU, and `model_format` says how to read the boxes that come back. Both must match how the model was compiled. + +DX-COM cannot compile `yolov10` or `yolov26` models with PPU support, since their heads already produce final detections. Combining either with `ppu: true` is rejected at startup. + +#### Multiple NPUs + +By default the detector binds to every NPU it can find. To pin a detector to specific devices, set `device_ids` to the indices you want it to use. This allows multiple detectors to be configured against separate modules: + +```yaml +detectors: + deepx_0: + type: deepx + device_ids: [0] + deepx_1: + type: deepx + device_ids: [1] +``` + +--- + ## NVidia TensorRT Detector Nvidia Jetson devices may be used for object detection using the TensorRT libraries. Due to the size of the additional libraries, this detector is only provided in images with the `-tensorrt-jp6` tag suffix, e.g. `ghcr.io/blakeblackshear/frigate:stable-tensorrt-jp6`. This detector is designed to work with Yolo models for object detection. diff --git a/docs/docs/frigate/hardware.md b/docs/docs/frigate/hardware.md index 1df3bb8368..2de31896e9 100644 --- a/docs/docs/frigate/hardware.md +++ b/docs/docs/frigate/hardware.md @@ -65,6 +65,11 @@ Frigate supports multiple different detectors that work on different types of ha - [Supports many model architectures](../../configuration/object_detectors#memryx-mx3) - Runs best with tiny, small, or medium-size models +- [DEEPX](#deepx-npu): The DEEPX NPU is available in m.2 format and as a HAT+ for the Raspberry Pi 5, allowing for a wide range of compatibility with devices. + - [Supports YOLO model architectures](../../configuration/object_detectors#deepx-npu) + - Runs best with tiny or small size models + - Runs efficiently on low power hardware + **AMD** - [ROCm](#rocm---amd-gpu): ROCm can run on AMD Discrete GPUs to provide efficient object detection @@ -256,6 +261,24 @@ The MX3 is a pipelined architecture, where the maximum frames per second support Inference speeds may vary depending on the host platform. The above data was measured on an **Intel 13700 CPU**. Platforms like Raspberry Pi, Orange Pi, and other ARM-based SBCs have different levels of processing capability, which may limit total FPS. +### DEEPX NPU + +Frigate supports the DEEPX NPU in both of its form factors: the **DX-M1** M.2 module, which works on x86 (Intel/AMD) and ARM-based SBCs such as the Raspberry Pi 5, and the **DX-M1M** on the [Sixfab AI HAT+](https://docs.sixfab.com/docs/ai-hat-plus-raspberry-pi-5-quickstart) for the Raspberry Pi 5. Both use the same driver and runtime, so the configuration is identical for either one. + +DEEPX NPU support in Frigate is developed and maintained by [Sixfab](https://sixfab.com). + +The NPU runs models compiled to the `.dxnn` format with DEEPX's DX-COM compiler. Pre-compiled YOLO models can be downloaded from the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo), and models compiled with Post-Processing Unit (PPU) support move candidate selection onto the NPU, which reduces host CPU usage. + +The DEEPX kernel driver has to be installed on the Docker host before the NPU can be used. See the [installation docs](installation.md#deepx-npu) for the setup steps. + +Detailed information is available [in the detector docs](/configuration/object_detectors#deepx-npu). + +| Name | Input Size | Inference Time | +| ------- | ---------- | -------------- | +| YOLOv8s | 640 | ~ 19 ms | + +Inference times were measured on a Raspberry Pi 5. The host platform affects the pre- and post-processing cost rather than the inference itself. + ### Nvidia Jetson Jetson devices are supported via the TensorRT or ONNX detectors when running Jetpack 6. It will [make use of the Jetson's hardware media engine](/configuration/hardware_acceleration_video#nvidia-jetson) when configured with the [appropriate presets](/configuration/ffmpeg_presets#hwaccel-presets), and will make use of the Jetson's GPU and DLA for object detection when configured with the [TensorRT detector](/configuration/object_detectors#nvidia-tensorrt-detector). diff --git a/docs/docs/frigate/installation.md b/docs/docs/frigate/installation.md index 880292f232..78ad105a6a 100644 --- a/docs/docs/frigate/installation.md +++ b/docs/docs/frigate/installation.md @@ -381,6 +381,89 @@ If you can't use Docker Compose, you can run the container with something simila Finally, configure [hardware object detection](/configuration/object_detectors#memryx-mx3) to complete the setup. +### DEEPX NPU + +The DEEPX NPU is available in two form factors, and Frigate supports both: + +- **DX-M1** in the M.2 2280 form factor (like an NVMe SSD), for x86 (Intel/AMD) PCs, the Raspberry Pi 5, and other ARM SBCs with an exposed PCIe M.2 slot. +- **DX-M1M** on the [Sixfab AI HAT+](https://docs.sixfab.com/docs/ai-hat-plus-raspberry-pi-5-quickstart), a HAT+ board that connects to the Raspberry Pi 5 over PCIe Gen 3 x1. + +Both present the NPU through the same PCIe driver and DX-RT runtime, so the setup below and the detector configuration are identical for either one. Nothing needs to change when moving between them. + +DEEPX NPU support in Frigate is developed and maintained by [Sixfab](https://sixfab.com). + +#### Versions + +A DEEPX install has three separately versioned pieces, and all three have to agree. The DX-RT runtime is built into the Frigate image, so the two host-side pieces must be brought up to match it: + +| Component | Version | Installed on | Installed by | +| ------------- | -------- | ------------- | -------------------------- | +| DX-RT runtime | `v3.4.2` | Frigate image | Built in, no action needed | +| Kernel driver | `v2.6.0` | Docker host | `user_installation.sh` | +| NPU firmware | `v2.7.4` | The module | Flashed from the host | + +:::warning + +A version mismatch does not produce a startup error. It typically shows up as inference requests that are accepted but never return a result, so detections simply stop appearing while Frigate looks healthy. If that happens after a Frigate upgrade, check all three versions before anything else. + +::: + +The firmware version cannot be read without the DX-RT runtime installed on the host, so checking or updating it is a manual step done before Frigate is running: + +1. Install the DX-RT runtime on the host by following DEEPX's own instructions. If you are using Sixfab hardware, their [AI HAT+ quickstart](https://docs.sixfab.com/docs/ai-hat-plus-raspberry-pi-5-quickstart) covers this. +2. Check the reported firmware version and update the module if it does not match the table above. +3. **Remove or disable the host runtime before starting the Frigate container.** + +:::danger + +Step 3 is not optional when running Frigate in Docker. The DX-RT daemon (`dxrtd`) may only run once per system, and Frigate runs its own copy inside the container. Leaving the host runtime active means two daemons competing for the same NPU, and the container's copy will fail to start. + +This only applies to Docker. If you run Frigate directly on the host, it uses the host's runtime and there is no conflict. + +::: + +#### Installation + +The DEEPX kernel driver must be installed on the host rather than in the container, because containers share the host kernel and cannot load kernel modules. Installing it creates the `/dev/dxrt*` device nodes that are passed through to Frigate. + +1. Copy or download [this script](https://github.com/blakeblackshear/frigate/blob/dev/docker/deepx/user_installation.sh). +2. Ensure it has execution permissions with `sudo chmod +x user_installation.sh` +3. Run the script with `./user_installation.sh` +4. **Restart your computer** to complete driver installation. + +Confirm the NPU is visible before continuing: + +```bash +ls /dev/dxrt* +``` + +:::note + +The installation script disables the host's `dxrt.service` if it finds one, for the reason described above. If you install or reinstall the DEEPX runtime on the host afterwards, disable it again before starting the container: + +```bash +sudo systemctl disable --now dxrt.service +``` + +::: + +#### Setup + +To set up Frigate, follow the default installation instructions, for example: `ghcr.io/blakeblackshear/frigate:stable` + +Next, grant Docker access to the NPU by adding the device to your `docker-compose.yml` file: + +```yaml +devices: + - /dev/dxrt0 +``` + +If you can't use Docker Compose, add `--device /dev/dxrt0` to your `docker run` command. + +#### Configuration + +Finally, configure [hardware object detection](/configuration/object_detectors#deepx-npu) to complete the setup. + ### Rockchip platform Make sure that you use a linux distribution that comes with the rockchip BSP kernel 5.10 or 6.1 and necessary drivers (especially rkvdec2 and rknpu). To check, enter the following commands: diff --git a/frigate/detectors/plugins/deepx.py b/frigate/detectors/plugins/deepx.py new file mode 100644 index 0000000000..4a691e7d61 --- /dev/null +++ b/frigate/detectors/plugins/deepx.py @@ -0,0 +1,565 @@ +"""DEEPX NPU detector running compiled .dxnn models via the DX-RT runtime.""" + +import glob +import logging +import os +from enum import Enum +from typing import Literal + +import cv2 +import numpy as np +from pydantic import ConfigDict, Field, model_validator + +from frigate.detectors.detection_api import DetectionApi +from frigate.detectors.detector_config import BaseDetectorConfig, ModelTypeEnum +from frigate.util.model import post_process_yolo + +logger = logging.getLogger(__name__) + +DETECTOR_KEY = "deepx" + +# PPU-compiled models still use yolo-generic in the config; the `ppu` option is +# what selects the decode path, since ModelTypeEnum has no PPU-specific value. +supported_models = [ModelTypeEnum.yologeneric] + +DXNN_DEVICES_ENV = "DXNN_DEVICES" + + +class ModelFormatEnum(str, Enum): + """The YOLO variant a .dxnn file was compiled from. + + DX-COM preserves the detection head of the source model, so the variant + determines how the output has to be read. Three families exist, and every + supported variant belongs to exactly one of them. + """ + + yolov5 = "yolov5" + yolov7 = "yolov7" + yolov8 = "yolov8" + yolov9 = "yolov9" + yolov10 = "yolov10" + yolov11 = "yolov11" + yolov12 = "yolov12" + yolov26 = "yolov26" + + +# Anchor-based heads. Raw output is (N, 5+C) with a separate objectness column, +# and the box columns hold sigmoid activations that must be rescaled against the +# grid cell and anchor that produced them. +ANCHOR_FORMATS = frozenset( + { + ModelFormatEnum.yolov5, + ModelFormatEnum.yolov7, + } +) + +# Anchor-free heads. Raw output is (4+C, N) channel-major with no objectness +# column, and the box columns are already in input pixels. +ANCHOR_FREE_FORMATS = frozenset( + { + ModelFormatEnum.yolov8, + ModelFormatEnum.yolov9, + ModelFormatEnum.yolov11, + ModelFormatEnum.yolov12, + } +) + +# Heads that run NMS on-device and emit final boxes as (N, 6) corner records. +# DX-COM cannot compile these with PPU support. +NMS_IN_HEAD_FORMATS = frozenset( + { + ModelFormatEnum.yolov10, + ModelFormatEnum.yolov26, + } +) + +# Anchor table for YOLOv5/YOLOv7 PPU models, keyed by stride. +PPU_ANCHORS = { + 8: np.array([[10, 13], [16, 30], [33, 23]], dtype=np.float32), + 16: np.array([[30, 61], [62, 45], [59, 119]], dtype=np.float32), + 32: np.array([[116, 90], [156, 198], [373, 326]], dtype=np.float32), +} +PPU_STRIDES = np.array([8, 16, 32], dtype=np.float32) + +# Byte layout of the fixed-width detection record the PPU emits. +PPU_RECORD_SIZE = 32 +PPU_BOX_BYTES = (0, 16) +PPU_GRID_BYTES = (16, 20) +PPU_SCORE_BYTES = (20, 24) +PPU_LABEL_BYTES = (24, 28) + + +def detect_device_count() -> int: + """Count the NPU devices visible to this process. + + Falls back from the DX-RT API to a scan of the passed-through device + nodes, then to a single device. + """ + try: + from dx_engine import get_device_count + + count = get_device_count() + except (ImportError, RuntimeError) as err: + logger.debug("Could not query DX-RT for a device count: %s", err) + else: + if count > 0: + return count + + devices = glob.glob("/dev/dxrt*") + + return len(devices) if devices else 1 + + +def resolve_devices(configured: list[int] | None) -> list[int]: + """Resolve which NPU device indices this detector should bind to.""" + if configured: + return configured + + env_value = os.environ.get(DXNN_DEVICES_ENV) + + if env_value: + try: + devices = [int(d.strip()) for d in env_value.split(",") if d.strip()] + except ValueError: + logger.warning( + "Ignoring unparsable %s=%s, detecting devices instead", + DXNN_DEVICES_ENV, + env_value, + ) + else: + if devices: + return devices + + return list(range(detect_device_count())) + + +def reinterpret(tensor: np.ndarray, byte_range: tuple[int, int], dtype) -> np.ndarray: + """Read a byte column range of a uint8 PPU tensor as `dtype`. + + A column slice is not contiguous, and numpy will not reinterpret + non-contiguous memory as a differently sized dtype, so it is copied first. + """ + lo, hi = byte_range + return np.ascontiguousarray(tensor[:, lo:hi]).view(dtype) + + +def ppu_records(outputs: list[np.ndarray]) -> np.ndarray | None: + """Return the fixed-width record array a PPU model emits, or None.""" + if not outputs or outputs[0].ndim < 2: + return None + + # drop the batch dimension, leaving one uint8 record per candidate + records = outputs[0][0] + + if ( + records.ndim != 2 + or records.shape[1] != PPU_RECORD_SIZE + or records.shape[0] == 0 + ): + logger.debug("Unexpected PPU output shape %s, skipping frame", records.shape) + return None + + return records + + +def fill_detections( + x_min: np.ndarray, + y_min: np.ndarray, + x_max: np.ndarray, + y_max: np.ndarray, + scores: np.ndarray, + labels: np.ndarray, + width: int, + height: int, + order: np.ndarray, +) -> np.ndarray: + """Normalize the surviving boxes into Frigate's (20, 6) detection array.""" + detections = np.zeros((20, 6), np.float32) + + for i, idx in enumerate(order[:20]): + detections[i] = [ + labels[idx], + scores[idx], + np.clip(y_min[idx] / height, 0, 1), + np.clip(x_min[idx] / width, 0, 1), + np.clip(y_max[idx] / height, 0, 1), + np.clip(x_max[idx] / width, 0, 1), + ] + + return detections + + +def run_nms( + x_min: np.ndarray, + y_min: np.ndarray, + box_w: np.ndarray, + box_h: np.ndarray, + scores: np.ndarray, + score_threshold: float, + nms_threshold: float, +) -> np.ndarray: + """Return the indices surviving NMS, in descending score order.""" + boxes_xywh = np.column_stack([x_min, y_min, box_w, box_h]) + indices = cv2.dnn.NMSBoxes( + boxes_xywh.tolist(), scores.tolist(), score_threshold, nms_threshold + ) + + if len(indices) == 0: + return np.empty(0, dtype=np.int32) + + return np.array(indices).reshape(-1) + + +def decode_yolo_ppu_anchor( + outputs: list[np.ndarray], + width: int, + height: int, + score_threshold: float, + nms_threshold: float, +) -> np.ndarray: + """Decode PPU output from an anchor-based head (YOLOv5, YOLOv7). + + Models compiled with DEEPX's Post-Processing Unit perform candidate + selection on the NPU and emit one fixed-width record per surviving + detection, so only the anchor decode and NMS are left for the host. This + is cheaper than decoding raw feature maps because every sub-threshold + anchor has already been discarded. + + Returns detections in Frigate's (20, 6) format of + [class_id, score, y_min, x_min, y_max, x_max] with normalized coordinates. + """ + records = ppu_records(outputs) + + if records is None: + return np.zeros((20, 6), np.float32) + + boxes = reinterpret(records, PPU_BOX_BYTES, np.float32).reshape(-1, 4) + scores = reinterpret(records, PPU_SCORE_BYTES, np.float32).flatten() + labels = reinterpret(records, PPU_LABEL_BYTES, np.uint32).flatten() + + grid = records[:, PPU_GRID_BYTES[0] : PPU_GRID_BYTES[1]] + grid_y = grid[:, 0].astype(np.float32) + grid_x = grid[:, 1].astype(np.float32) + anchor_idx = grid[:, 2] + layer_idx = grid[:, 3] + + stride = PPU_STRIDES[layer_idx] + anchor_w = np.zeros(len(boxes), dtype=np.float32) + anchor_h = np.zeros(len(boxes), dtype=np.float32) + + for layer_stride, anchors in PPU_ANCHORS.items(): + mask = stride == layer_stride + + if np.any(mask): + anchor_w[mask] = anchors[anchor_idx[mask], 0] + anchor_h[mask] = anchors[anchor_idx[mask], 1] + + # YOLOv5-style anchor decode: the model emits sigmoid outputs that are + # rescaled against the grid cell and its anchor to recover pixel geometry. + center_x = (boxes[:, 0] * 2.0 - 0.5 + grid_x) * stride + center_y = (boxes[:, 1] * 2.0 - 0.5 + grid_y) * stride + box_w = (boxes[:, 2] ** 2 * 4.0) * anchor_w + box_h = (boxes[:, 3] ** 2 * 4.0) * anchor_h + + x_min = center_x - box_w * 0.5 + y_min = center_y - box_h * 0.5 + + order = run_nms(x_min, y_min, box_w, box_h, scores, score_threshold, nms_threshold) + + return fill_detections( + x_min, + y_min, + x_min + box_w, + y_min + box_h, + scores, + labels, + width, + height, + order, + ) + + +def decode_yolo_ppu_anchor_free( + outputs: list[np.ndarray], + width: int, + height: int, + score_threshold: float, + nms_threshold: float, +) -> np.ndarray: + """Decode PPU output from an anchor-free head (YOLOv8, v9, v11, v12). + + These heads have no anchors and no objectness, so the PPU writes the box + directly as (center_x, center_y, width, height) in input pixels. The grid + columns an anchor-based record carries are unused here, and applying the + anchor formula to them is what produces boxes hundreds of thousands of + pixels wide. + """ + records = ppu_records(outputs) + + if records is None: + return np.zeros((20, 6), np.float32) + + boxes = reinterpret(records, PPU_BOX_BYTES, np.float32).reshape(-1, 4) + scores = reinterpret(records, PPU_SCORE_BYTES, np.float32).flatten() + labels = reinterpret(records, PPU_LABEL_BYTES, np.uint32).flatten() + + box_w = boxes[:, 2] + box_h = boxes[:, 3] + x_min = boxes[:, 0] - box_w * 0.5 + y_min = boxes[:, 1] - box_h * 0.5 + + order = run_nms(x_min, y_min, box_w, box_h, scores, score_threshold, nms_threshold) + + return fill_detections( + x_min, + y_min, + x_min + box_w, + y_min + box_h, + scores, + labels, + width, + height, + order, + ) + + +def decode_yolo_raw_anchor( + outputs: list[np.ndarray], + width: int, + height: int, + score_threshold: float, + nms_threshold: float, +) -> np.ndarray: + """Decode raw output from an anchor-based head (YOLOv5, YOLOv7). + + The exported head emits (N, 5+C) rows of + [center_x, center_y, width, height, objectness, class scores...] already in + input pixels. Confidence is objectness times the best class score, which is + what separates this layout from the anchor-free one: reading column 4 as a + class score there would treat objectness as a class. + """ + tensor = outputs[0] + + if tensor.ndim < 2 or tensor.shape[-1] == 0: + return np.zeros((20, 6), np.float32) + + # squeeze would collapse the row axis too when a single candidate is + # returned, so fold every leading axis into rows instead + predictions = tensor.reshape(-1, tensor.shape[-1]) + + if predictions.shape[0] == 0: + return np.zeros((20, 6), np.float32) + + objectness = predictions[:, 4] + class_scores = predictions[:, 5:] + labels = np.argmax(class_scores, axis=1) + scores = objectness * class_scores[np.arange(len(labels)), labels] + + box_w = predictions[:, 2] + box_h = predictions[:, 3] + x_min = predictions[:, 0] - box_w * 0.5 + y_min = predictions[:, 1] - box_h * 0.5 + + order = run_nms(x_min, y_min, box_w, box_h, scores, score_threshold, nms_threshold) + + return fill_detections( + x_min, + y_min, + x_min + box_w, + y_min + box_h, + scores, + labels, + width, + height, + order, + ) + + +def decode_yolo_raw_nms_in_head( + outputs: list[np.ndarray], + width: int, + height: int, + score_threshold: float, +) -> np.ndarray: + """Decode raw output from a head that already ran NMS (YOLOv10, YOLOv26). + + These heads emit (N, 6) rows of [x_min, y_min, x_max, y_max, score, class] + in input pixels, sorted by score. Running NMS again would be wasted work, + so only the score filter and normalization are applied. + """ + tensor = outputs[0] + + if tensor.ndim < 2 or tensor.shape[-1] == 0: + return np.zeros((20, 6), np.float32) + + # squeeze would collapse the row axis too when a single candidate is + # returned, so fold every leading axis into rows instead + predictions = tensor.reshape(-1, tensor.shape[-1]) + + if predictions.shape[0] == 0: + return np.zeros((20, 6), np.float32) + + scores = predictions[:, 4] + order = np.flatnonzero(scores >= score_threshold) + + return fill_detections( + predictions[:, 0], + predictions[:, 1], + predictions[:, 2], + predictions[:, 3], + scores, + predictions[:, 5].astype(np.int32), + width, + height, + order, + ) + + +class DeepxDetectorConfig(BaseDetectorConfig): + """DEEPX NPU detector running .dxnn models via the DX-RT runtime.""" + + model_config = ConfigDict(title="DEEPX NPU") + + type: Literal[DETECTOR_KEY] + device_ids: list[int] | None = Field( + default=None, + title="DEEPX device indices", + description="NPU device indices this detector should bind to. Defaults " + "to auto-detection via the DXNN_DEVICES environment variable, then the " + "DX-RT device count, then a scan of /dev/dxrt*.", + ) + ppu: bool = Field( + default=False, + title="PPU-compiled model", + description="Set to true when the model was compiled with DEEPX's " + "Post-Processing Unit support, which moves candidate selection onto " + "the NPU and changes the output format.", + ) + model_format: ModelFormatEnum | None = Field( + default=None, + title="YOLO variant the model was compiled from", + description="Selects how the model's output is decoded. Leave unset to " + "infer the layout from the output shape, which cannot tell an " + "anchor-based head from an anchor-free one and is only reliable for " + "YOLOv8 and newer.", + ) + score_threshold: float = Field( + default=0.25, + title="Score threshold used when decoding detector output", + ) + nms_threshold: float = Field( + default=0.45, + title="NMS IoU threshold used when decoding detector output", + ) + + @model_validator(mode="after") + def validate_ppu_is_supported(self): + """Reject a PPU/format pairing DX-COM cannot produce.""" + if self.ppu and self.model_format in NMS_IN_HEAD_FORMATS: + raise ValueError( + f"model_format '{self.model_format.value}' runs NMS in the " + "detection head and cannot be compiled with PPU support by " + "DX-COM. Set ppu to false." + ) + + return self + + +class Deepx(DetectionApi): + type_key = DETECTOR_KEY + + def __init__(self, config: DeepxDetectorConfig): + try: + from dx_engine import InferenceEngine, InferenceOption + except ImportError as err: + raise ImportError( + "The DX-RT runtime is not available. Check that the DEEPX " + "kernel driver is loaded on the Docker host and that " + "/dev/dxrt* is passed through to this container." + ) from err + + super().__init__(config) + + if config.model.model_type not in supported_models: + raise ValueError( + f'Model type "{config.model.model_type}" is not supported by the ' + "deepx detector" + ) + + model_path = config.model.path + + if not model_path or not os.path.isfile(model_path): + raise FileNotFoundError( + f"DEEPX model '{model_path}' was not found. Compile a model with " + "DX-COM or download one from the DEEPX ModelZoo, then point " + "model.path at the .dxnn file." + ) + + devices = resolve_devices(config.device_ids) + logger.info("Loading DEEPX model %s on device(s) %s", model_path, devices) + + options = InferenceOption() + options.devices = devices + options.bound_option = InferenceOption.BOUND_OPTION.NPU_ALL + + self.session = InferenceEngine(str(model_path), options) + self.ppu = config.ppu + self.model_format = config.model_format + self.score_threshold = config.score_threshold + self.nms_threshold = config.nms_threshold + + logger.info( + "DEEPX decoding %s output as %s", + "PPU" if self.ppu else "raw", + self.model_format.value if self.model_format else "auto-detected", + ) + + def decode(self, outputs: list[np.ndarray]) -> np.ndarray: + """Decode raw or PPU output according to the configured YOLO variant.""" + if self.ppu: + # An unset model_format keeps the original anchor-based behaviour, + # which is the only layout this detector decoded before the option + # existed. + if self.model_format in ANCHOR_FREE_FORMATS: + return decode_yolo_ppu_anchor_free( + outputs, + self.width, + self.height, + self.score_threshold, + self.nms_threshold, + ) + + return decode_yolo_ppu_anchor( + outputs, + self.width, + self.height, + self.score_threshold, + self.nms_threshold, + ) + + if self.model_format in NMS_IN_HEAD_FORMATS: + return decode_yolo_raw_nms_in_head( + outputs, self.width, self.height, self.score_threshold + ) + + # Only single-output anchor heads need the dedicated decoder. A + # multi-part output is a set of raw feature maps, which the shared + # helper already decodes with the same anchor formula. + if self.model_format in ANCHOR_FORMATS and len(outputs) == 1: + return decode_yolo_raw_anchor( + outputs, + self.width, + self.height, + self.score_threshold, + self.nms_threshold, + ) + + return post_process_yolo(outputs, self.width, self.height) + + def detect_raw(self, tensor_input): + outputs = self.session.run([tensor_input]) + + if not isinstance(outputs, list): + outputs = [outputs] + + return self.decode(outputs) diff --git a/frigate/test/test_deepx.py b/frigate/test/test_deepx.py new file mode 100644 index 0000000000..6a661daaef --- /dev/null +++ b/frigate/test/test_deepx.py @@ -0,0 +1,298 @@ +"""Tests for the DEEPX detector's per-format output decoding.""" + +import unittest + +import numpy as np + +from frigate.detectors.plugins.deepx import ( + ANCHOR_FORMATS, + ANCHOR_FREE_FORMATS, + NMS_IN_HEAD_FORMATS, + PPU_ANCHORS, + PPU_RECORD_SIZE, + ModelFormatEnum, + decode_yolo_ppu_anchor, + decode_yolo_ppu_anchor_free, + decode_yolo_raw_anchor, + decode_yolo_raw_nms_in_head, +) + + +def build_ppu_record( + box: tuple[float, float, float, float], + grid_y: int, + grid_x: int, + anchor_idx: int, + layer_idx: int, + score: float, + label: int, +) -> np.ndarray: + """Build a single fixed-width detection record as the PPU emits it.""" + record = np.zeros(PPU_RECORD_SIZE, dtype=np.uint8) + record[0:16] = np.array(box, dtype=np.float32).view(np.uint8) + record[16:20] = [grid_y, grid_x, anchor_idx, layer_idx] + record[20:24] = np.array([score], dtype=np.float32).view(np.uint8) + record[24:28] = np.array([label], dtype=np.uint32).view(np.uint8) + return record + + +class TestDeepxPpuDecode(unittest.TestCase): + def test_decodes_a_single_detection(self): + # stride 8 (layer 0) with the first anchor of that layer + outputs = [ + build_ppu_record( + box=(0.5, 0.5, 0.5, 0.5), + grid_y=10, + grid_x=10, + anchor_idx=0, + layer_idx=0, + score=0.9, + label=2, + ).reshape(1, 1, PPU_RECORD_SIZE) + ] + + detections = decode_yolo_ppu_anchor(outputs, 640, 640, 0.25, 0.45) + + anchor_w, anchor_h = PPU_ANCHORS[8][0] + # center = (0.5 * 2 - 0.5 + 10) * 8, size = (0.5 ** 2 * 4) * anchor + center = (0.5 * 2.0 - 0.5 + 10) * 8 + box_w = 0.5**2 * 4.0 * anchor_w + box_h = 0.5**2 * 4.0 * anchor_h + + self.assertEqual(detections[0][0], 2) + self.assertAlmostEqual(detections[0][1], 0.9, places=5) + self.assertAlmostEqual(detections[0][2], (center - box_h / 2) / 640, places=5) + self.assertAlmostEqual(detections[0][3], (center - box_w / 2) / 640, places=5) + self.assertAlmostEqual(detections[0][4], (center + box_h / 2) / 640, places=5) + self.assertAlmostEqual(detections[0][5], (center + box_w / 2) / 640, places=5) + + # remaining slots stay empty + self.assertTrue(np.all(detections[1:] == 0)) + + def test_uses_the_anchor_for_the_records_layer(self): + # the same box on stride 32 (layer 2) must decode to a larger box + def decode(layer_idx: int) -> np.ndarray: + outputs = [ + build_ppu_record( + box=(0.5, 0.5, 0.5, 0.5), + grid_y=1, + grid_x=1, + anchor_idx=0, + layer_idx=layer_idx, + score=0.9, + label=0, + ).reshape(1, 1, PPU_RECORD_SIZE) + ] + return decode_yolo_ppu_anchor(outputs, 640, 640, 0.25, 0.45) + + small = decode(0) + large = decode(2) + + small_width = small[0][5] - small[0][3] + large_width = large[0][5] - large[0][3] + + self.assertGreater(large_width, small_width) + + def test_drops_detections_below_the_score_threshold(self): + outputs = [ + build_ppu_record( + box=(0.5, 0.5, 0.5, 0.5), + grid_y=10, + grid_x=10, + anchor_idx=0, + layer_idx=0, + score=0.1, + label=2, + ).reshape(1, 1, PPU_RECORD_SIZE) + ] + + detections = decode_yolo_ppu_anchor(outputs, 640, 640, 0.25, 0.45) + + self.assertTrue(np.all(detections == 0)) + + def test_caps_output_at_twenty_detections(self): + records = np.stack( + [ + build_ppu_record( + box=(0.5, 0.5, 0.5, 0.5), + grid_y=i, + grid_x=i, + anchor_idx=0, + layer_idx=0, + score=0.9, + label=1, + ) + for i in range(40) + ] + ) + + detections = decode_yolo_ppu_anchor( + [records.reshape(1, 40, PPU_RECORD_SIZE)], 640, 640, 0.25, 0.45 + ) + + self.assertEqual(detections.shape, (20, 6)) + + def test_returns_empty_detections_for_no_output(self): + self.assertTrue(np.all(decode_yolo_ppu_anchor([], 640, 640, 0.25, 0.45) == 0)) + + def test_returns_empty_detections_for_an_unexpected_record_width(self): + # a non-PPU model would produce a different record width + outputs = [np.zeros((1, 4, 16), dtype=np.uint8)] + + self.assertTrue( + np.all(decode_yolo_ppu_anchor(outputs, 640, 640, 0.25, 0.45) == 0) + ) + + +class TestDeepxModelFormats(unittest.TestCase): + def test_every_format_belongs_to_exactly_one_family(self): + families = [ANCHOR_FORMATS, ANCHOR_FREE_FORMATS, NMS_IN_HEAD_FORMATS] + + for fmt in ModelFormatEnum: + matches = [family for family in families if fmt in family] + self.assertEqual( + len(matches), 1, f"{fmt.value} must be in exactly one family" + ) + + def test_families_do_not_overlap(self): + self.assertFalse(ANCHOR_FORMATS & ANCHOR_FREE_FORMATS) + self.assertFalse(ANCHOR_FORMATS & NMS_IN_HEAD_FORMATS) + self.assertFalse(ANCHOR_FREE_FORMATS & NMS_IN_HEAD_FORMATS) + + +class TestDeepxPpuAnchorFreeDecode(unittest.TestCase): + def test_reads_the_box_bytes_as_pixel_geometry(self): + """An anchor-free PPU record holds cx, cy, w, h directly in pixels.""" + record = build_ppu_record( + box=(320.0, 160.0, 64.0, 32.0), + # grid columns carry no meaning for an anchor-free head, and must + # not influence the result + grid_y=7, + grid_x=9, + anchor_idx=2, + layer_idx=2, + score=0.9, + label=0, + ) + outputs = [record.reshape(1, 1, PPU_RECORD_SIZE)] + + detections = decode_yolo_ppu_anchor_free(outputs, 640, 640, 0.25, 0.45) + + # x: 320 +/- 32 -> 288..352, y: 160 +/- 16 -> 144..176 + self.assertAlmostEqual(detections[0][2], 144 / 640, places=5) + self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) + self.assertAlmostEqual(detections[0][4], 176 / 640, places=5) + self.assertAlmostEqual(detections[0][5], 352 / 640, places=5) + + def test_anchor_formula_would_blow_the_box_up(self): + """Guards the bug this decoder exists to prevent.""" + record = build_ppu_record( + box=(320.0, 160.0, 64.0, 32.0), + grid_y=7, + grid_x=9, + anchor_idx=2, + layer_idx=2, + score=0.9, + label=0, + ) + outputs = [record.reshape(1, 1, PPU_RECORD_SIZE)] + + anchor_free = decode_yolo_ppu_anchor_free(outputs, 640, 640, 0.25, 0.45) + anchor = decode_yolo_ppu_anchor(outputs, 640, 640, 0.25, 0.45) + + # the anchor decode squares the width, saturating the clip to the frame + self.assertEqual(anchor[0][5], 1.0) + self.assertLess(anchor_free[0][5], 1.0) + + +class TestDeepxRawAnchorDecode(unittest.TestCase): + def build_raw_anchor_output(self, rows): + """Build an (1, N, 5+C) tensor with 80 classes.""" + out = np.zeros((1, len(rows), 85), dtype=np.float32) + + for i, (cx, cy, w, h, obj, label, cls_score) in enumerate(rows): + out[0, i, 0:4] = [cx, cy, w, h] + out[0, i, 4] = obj + out[0, i, 5 + label] = cls_score + + return [out] + + def test_confidence_is_objectness_times_class_score(self): + outputs = self.build_raw_anchor_output( + [(320.0, 320.0, 40.0, 80.0, 0.8, 3, 0.5)] + ) + + detections = decode_yolo_raw_anchor(outputs, 640, 640, 0.25, 0.45) + + self.assertEqual(detections[0][0], 3) + self.assertAlmostEqual(detections[0][1], 0.4, places=5) + + def test_drops_rows_whose_combined_score_is_below_threshold(self): + # 0.4 * 0.5 = 0.2, under a 0.25 threshold even though both parts are + # individually above it + outputs = self.build_raw_anchor_output( + [(320.0, 320.0, 40.0, 80.0, 0.4, 3, 0.5)] + ) + + detections = decode_yolo_raw_anchor(outputs, 640, 640, 0.25, 0.45) + + self.assertTrue(np.all(detections == 0)) + + def test_converts_center_form_to_normalized_corners(self): + outputs = self.build_raw_anchor_output( + [(320.0, 160.0, 64.0, 32.0, 1.0, 0, 1.0)] + ) + + detections = decode_yolo_raw_anchor(outputs, 640, 640, 0.25, 0.45) + + self.assertAlmostEqual(detections[0][2], 144 / 640, places=5) + self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) + self.assertAlmostEqual(detections[0][4], 176 / 640, places=5) + self.assertAlmostEqual(detections[0][5], 352 / 640, places=5) + + +class TestDeepxRawNmsInHeadDecode(unittest.TestCase): + def test_reads_corner_records_without_running_nms(self): + """Two heavily overlapping boxes both survive: the head already ran NMS.""" + out = np.array( + [ + [ + [100.0, 100.0, 200.0, 200.0, 0.9, 2.0], + [102.0, 102.0, 202.0, 202.0, 0.8, 2.0], + ] + ], + dtype=np.float32, + ) + + detections = decode_yolo_raw_nms_in_head([out], 640, 640, 0.25) + + self.assertEqual(detections[0][0], 2) + self.assertAlmostEqual(detections[0][1], 0.9, places=5) + self.assertAlmostEqual(detections[0][3], 100 / 640, places=5) + self.assertEqual(detections[1][0], 2) + self.assertAlmostEqual(detections[1][1], 0.8, places=5) + + def test_applies_the_score_threshold(self): + out = np.array( + [ + [ + [100.0, 100.0, 200.0, 200.0, 0.9, 2.0], + [300.0, 300.0, 400.0, 400.0, 0.1, 5.0], + ] + ], + dtype=np.float32, + ) + + detections = decode_yolo_raw_nms_in_head([out], 640, 640, 0.25) + + self.assertAlmostEqual(detections[0][1], 0.9, places=5) + self.assertTrue(np.all(detections[1] == 0)) + + def test_returns_empty_detections_for_an_empty_output(self): + out = np.zeros((1, 0, 6), dtype=np.float32) + + self.assertTrue(np.all(decode_yolo_raw_nms_in_head([out], 640, 640, 0.25) == 0)) + + +if __name__ == "__main__": + unittest.main()