From af0ba191966812cf9ac8515d95b1dd221363d17e Mon Sep 17 00:00:00 2001 From: "A. Ahmet" Date: Mon, 21 Sep 2026 15:50:44 +0300 Subject: [PATCH] Feat/deepx npu detector (#24336) * feat(deepx): add DEEPX NPU detector and runtime integration. * feat(deepx): enforce model_format requirement when ppu is enabled and add integrity checks for driver installation * Update frigate/detectors/plugins/deepx.py Public method lacks docstring Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> * Refactor DEEPX detector tests, support SSD and DAMO-YOLO * feat(deepx): add anchor-free output decoding and corresponding tests * Add tests and updates for DEEPX detector and refactor DEEPX accelerator code structure. * fix: enhance model type validation and update documentation for DEEPX detector * fix: add support for customizable score and NMS thresholds * refactor: infer YOLO layout from the model, drop per-detector options and the dxrtd placeholder * fix: keep only the anchor-free PPU verdict, re-read anchor-based each frame * Update latency data for DEEPX NPU * Expanding PPU support for DEEPX and set yolo-generic as default. * enhance scale count resolution logic * Extend PPU layout handling and YOLOX support to DEEPX detector * fix: assume the largest PPU anchor table when the .dxnn has no layout * Improve PPU decoding and introduce strides handling * Improve PPU scope and fix box format mismatch * Fix unnamed node issue that breaks traversal * fix: update object detection model type description to remove outdated architecture --------- Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --- CODEOWNERS | 1 + docker/deepx/user_installation.sh | 131 ++ .../etc/s6-overlay/s6-rc.d/init-devices/run | 1 + docs/data/object_detectors_models.yaml | 59 + docs/docs/configuration/object_detectors.md | 58 + docs/docs/frigate/hardware.md | 31 + docs/docs/frigate/installation.md | 93 ++ frigate/detectors/hardware.py | 11 + frigate/detectors/plugins/deepx.py | 1166 +++++++++++++++++ frigate/test/test_deepx.py | 896 +++++++++++++ frigate/test/test_detector_hardware.py | 20 + 11 files changed, 2467 insertions(+) create mode 100755 docker/deepx/user_installation.sh create mode 100644 frigate/detectors/plugins/deepx.py create mode 100644 frigate/test/test_deepx.py 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 100755 index 0000000000..ff332172f7 --- /dev/null +++ b/docker/deepx/user_installation.sh @@ -0,0 +1,131 @@ +#!/bin/bash + +# Installs the DEEPX NPU kernel driver and the DX-RT runtime on the Docker host, +# then enables the vendor's dxrt.service. A container cannot load kernel +# modules, so this runs outside the image; the driver creates the /dev/dxrt* +# nodes and the daemon multiplexes the NPU across host and container. +# +# Driver, runtime and firmware versions must agree or inference hangs instead +# of failing at startup. The set this script installs is pinned in +# driver_version, runtime_version and firmware_version below; move them +# together, never one at a time. +# +# DEEPX NPU support in Frigate is maintained by Sixfab (https://sixfab.com). + +set -euo pipefail + +driver_version="v2.6.0" +# the commit the tag resolves to, since DEEPX signs neither tags nor releases +# and this is compiled and installed as root. Update both together +driver_commit="7074748e7104f470b02f517583abba652b3f05fa" +firmware_version="v2.7.4" + +sudo apt-get update +sudo apt-get install -y git build-essential "linux-headers-$(uname -r)" pciutils wget + +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 + +# fetch the pinned commit rather than cloning the tag, so a retag cannot swap +# in different source. The build directory is reused so a second run after a +# failure does not stop on the directory already being there +mkdir -p dx_rt_npu_linux_driver +cd dx_rt_npu_linux_driver +git init -q +git remote get-url origin > /dev/null 2>&1 || + git remote add origin https://github.com/DEEPX-AI/dx_rt_npu_linux_driver.git +git fetch --depth 1 origin "${driver_commit}" +git checkout -q FETCH_HEAD + +fetched_commit=$(git rev-parse HEAD) +if [[ "${fetched_commit}" != "${driver_commit}" ]]; then + echo "Fetched commit ${fetched_commit} does not match pinned driver_commit ${driver_commit}." + echo "Refusing to build unverified driver source." + exit 1 +fi + +cd modules + +sudo ./build.sh -c install --reload + +sudo depmod -A + +# dx_dma is the PCIe transport, dxrt_driver 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 + +runtime_version="v3.4.0" +declare -A runtime_sha256=( + [amd64]="736cfef009ce9e974ab1ab610d867239d19d72a426a53e367ddcbd53297b6e20" + [arm64]="eb6107f5f02f2ad76ae89f414e8b5f346f34fbc6f0888236136853a26be6f6a0" +) + +runtime_release="${runtime_version#v}" +deb_arch=$(dpkg --print-architecture) +deb_file="/tmp/libdxrt-bin_${runtime_release}_${deb_arch}.deb" + +wget -qO "${deb_file}" \ + "https://raw.githubusercontent.com/DEEPX-AI/dx_rt/${runtime_version}/release/${runtime_release}/libdxrt-bin_${runtime_release}_${deb_arch}.deb" + +expected_sha256="${runtime_sha256[${deb_arch}]:-}" +if [[ -z "${expected_sha256}" ]]; then + echo "No pinned SHA-256 for architecture ${deb_arch}; refusing to install." + exit 1 +fi +if [[ "$(sha256sum "${deb_file}" | cut -d' ' -f1)" != "${expected_sha256}" ]]; then + echo "SHA-256 mismatch for ${deb_file}; refusing to install." + exit 1 +fi + +sudo dpkg -i "${deb_file}" +sudo ldconfig +rm -f "${deb_file}" + +sudo cp /usr/share/libdxrt-bin/service/dxrt.service /etc/systemd/system/ + +# With an endpoint set, dxrtd binds that path only, so the socket goes in a +# directory Frigate can mount (kept across restarts so the mount stays valid) +# and a symlink at the default /tmp path keeps host tools that do not set the +# variable working through their own fallback. +sudo mkdir -p /etc/systemd/system/dxrt.service.d +sudo tee /etc/systemd/system/dxrt.service.d/frigate.conf > /dev/null <<'UNIT' +[Service] +RuntimeDirectory=dxrt +RuntimeDirectoryMode=0755 +RuntimeDirectoryPreserve=yes +Environment=DXRT_DYNAMIC_IPC_ENDPOINT=/run/dxrt/dxrt_dynamic_ipc.sock +ExecStartPost=/bin/ln -sfn /run/dxrt/dxrt_dynamic_ipc.sock /tmp/dxrt_dynamic_ipc.sock +UNIT + +sudo systemctl daemon-reload +sudo systemctl enable dxrt.service +sudo systemctl restart dxrt.service + +if ! sudo systemctl is-active --quiet dxrt.service; then + echo "dxrt.service did not start. Check: sudo journalctl -u dxrt.service" + exit 1 +fi + +echo "DEEPX driver and runtime installation complete." +echo "Driver version: $(modinfo -F version dxrt_driver) (expected ${driver_version#v})" +echo "Runtime version: ${runtime_release}" +echo "Device node(s): $(echo /dev/dxrt*)" +echo +echo "This driver expects NPU firmware ${firmware_version}. Check it with:" +echo " dxrt-cli --status" +echo "Update the module if it does not match before starting Frigate." diff --git a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run index 72bbc0434c..7b20bf3914 100755 --- a/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run +++ b/docker/main/rootfs/etc/s6-overlay/s6-rc.d/init-devices/run @@ -37,6 +37,7 @@ device_globs=( "/dev/nvmap" "/dev/nvidia*" "/dev/memx*" + "/dev/dxrt*" ) IFS=',' read -ra extra_globs <<< "${DEVICE_ACL_PATHS:-}" diff --git a/docs/data/object_detectors_models.yaml b/docs/data/object_detectors_models.yaml index b50a5992b7..f2e560d570 100644 --- a/docs/data/object_detectors_models.yaml +++ b/docs/data/object_detectors_models.yaml @@ -967,6 +967,65 @@ memryx: # The .zip file must contain: # ├── ssdlite_mobilenet.dfp (a file ending with .dfp) # └── ssdlite_mobilenet_post.onnx (optional; only if the model includes a cropped post-processing network) +deepx: + title: DEEPX NPU + models: + - key: yolo + label: YOLO + recommended: true + download: No model is bundled with Frigate. Download a pre-compiled YOLO `.dxnn` model from the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo) or compile your own with DX-COM, then bind-mount it into the container and point the model's `path` at it. The recommended model is `yolox-s_640x640_ppu.dxnn`. Its Post-Processing Unit (PPU) compile moves candidate selection onto the NPU, which makes it the fastest ModelZoo model measured through Frigate (about 13 ms on a DX-M1). The output layout is read from the compiled model, so anchor-based, anchor-free, NMS-in-head and PPU models (anchor-based or anchor-free) all need no extra configuration; prefer a `PPU` variant whenever the ModelZoo offers one. PPU models must be compiled with DX-COM 2.4.0 or later, which writes the head layout Frigate reads into the file. + ui: |- + Navigate to **Settings > System > Detection models** and select **DEEPX NPU** 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/deepx/yolox-s_640x640_ppu.dxnn` | + | **Label map for custom object detector** | `/labelmap/coco-80.txt` | + | **Object detection model input width** | `640` | + | **Object detection model input height** | `640` | + | **Model Input Pixel Color Format** | `rgb` (Frigate's default value) | + | **Model Input Tensor Shape** | `nhwc` (Frigate's default value) | + | **Model Input D Type** | `int` (Frigate's default value) | + | **Object Detection Model Type** | `yolo-generic` | + + Quantization is baked into the compiled model, so no normalization is applied on the host and the input defaults do not need to be overridden. + yaml: |- + models: + - devices: + - deepx:PCIe:0 + path: /config/model_cache/deepx/yolox-s_640x640_ppu.dxnn + labelmap_path: /labelmap/coco-80.txt + model_type: yolo-generic + width: 640 + height: 640 + - key: yolox + label: YOLOX + recommended: false + download: No model is bundled with Frigate. Download a pre-compiled YOLOX `.dxnn` model from the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo), then bind-mount it into the container and point the model's `path` at it. The `_ppu` variant is faster and also works with the `yolo-generic` model type; the plain export needs `yolox` so its raw head is decoded. + ui: |- + Navigate to **Settings > System > Detection models** and select **DEEPX NPU** 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/deepx/yolox-s_640x640.dxnn`| + | **Label map for custom object detector** | `/labelmap/coco-80.txt` | + | **Object detection model input width** | `640` | + | **Object detection model input height** | `640` | + | **Model Input Pixel Color Format** | `rgb` (Frigate's default value) | + | **Model Input Tensor Shape** | `nhwc` (Frigate's default value) | + | **Model Input D Type** | `int` (Frigate's default value) | + | **Object Detection Model Type** | `yolox` | + + The width and height must match the resolution the `.dxnn` file was compiled for. + yaml: |- + models: + - devices: + - deepx:PCIe:0 + path: /config/model_cache/deepx/yolox-s_640x640.dxnn + labelmap_path: /labelmap/coco-80.txt + model_type: yolox + width: 640 + height: 640 tensorrt: title: TensorRT models: diff --git a/docs/docs/configuration/object_detectors.md b/docs/docs/configuration/object_detectors.md index cd2ca1fdfb..27f03ca1f4 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): The Hailo-8, Hailo-8L and Hailo-8R AI Acceleration modules are 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** @@ -612,6 +613,63 @@ 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 runtime on the host and passing the NPU through to the container. + +To run a model on a DEEPX NPU, list a `deepx` device on that model. + +:::info + +The DX-RT Python bindings are not part of the Frigate image. They are downloaded and installed into `/config/.local` the first time a DEEPX device 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-deepx} + + + +Frigate does not bundle a model for this detector. Models must be compiled to DEEPX's `.dxnn` format. Two model types are supported: + +- `yolo-generic` for YOLO object detection models, the recommended default. The detector reads the model's output layout from the compiled file, so anchor-based, anchor-free and NMS-in-head models all work with the same configuration, as do models compiled with DEEPX's Post-Processing Unit (PPU) support. +- `yolox` for YOLOX models compiled without PPU support, whose raw head needs Frigate's YOLOX decoder. A YOLOX model compiled with PPU support works under either `yolox` or `yolo-generic`. + +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 object detection models. Download the `.dxnn`, bind-mount it into the container, and point the model's `path` at it. Alternatively, compile your own model with the DX-COM compiler. The recommended starting point is `yolox-s_640x640_ppu.dxnn`, the fastest ModelZoo model measured through Frigate: + +```yaml +models: + - devices: + - deepx:PCIe:0 + path: /config/model_cache/deepx/yolox-s_640x640_ppu.dxnn + labelmap_path: /labelmap/coco-80.txt + model_type: yolo-generic + width: 640 + height: 640 +``` + +For PPU models, use a `.dxnn` compiled with DX-COM 2.4.0 or later. Frigate reads the PPU head layout the compiler writes into the file and refuses to load a PPU model without it. + +`model_type` must be set to `yolo-generic` or `yolox` to match the model; `yolo-generic` is the recommended default unless the model is a raw YOLOX export. Frigate defaults it to `ssd`, which this detector does not support, so the detector refuses to start on a model that leaves it unset. + +`width` and `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. + +A DEEPX device is `PCIe:`, as reported on the detector settings page. The NPU daemon multiplexes across processes, so the same device may be listed more than once to run additional inference processes against it: + +```yaml +models: + - devices: + - deepx:PCIe:0 + - deepx:PCIe:0 +``` + +#### Label maps + +The object detection models in 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 COCO scheme, and leaving it in place will cause detections to be reported as the wrong object type. For `yolo-generic` models the label map is also what the detector uses to tell the output layout, so a label map with the wrong number of classes is reported as an error at startup. + +--- + ## 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 3301eb5d42..2bc95938d4 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 @@ -257,6 +262,32 @@ 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 DEEPX driver and runtime run on the Docker host rather than inside the Frigate container and must be installed before the NPU can be used. See the [installation docs](installation.md#deepx-npu) for the setup steps and [the detector docs](/configuration/object_detectors#deepx-npu) for the configuration. + +Frigate does not bundle a model for this detector. Models use DEEPX's `.dxnn` format, and pre-compiled YOLO models can be downloaded from the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo). Prefer a model with a `_ppu` suffix whenever one is available for the architecture you want: these run part of the post-processing on the NPU itself and are considerably faster, roughly 2.5x for the same architecture and input size. **YOLOX-S with PPU is the recommended starting point.** + +Inference times for a few recommended models, measured through Frigate's own stats on a DX-M1: + +| Model | Input Size | DX-M1 Inference Time | +| ----------------- | ---------- | -------------------- | +| YOLOX-S (PPU) | 640 | ~ 13 ms | +| YOLOv9-t (PPU) | 640 | ~ 18 ms | +| YOLOv4 (PPU) | 512 | ~ 20 ms | +| YOLOX-S | 640 | ~ 34 ms | +| YOLOv9-s | 640 | ~ 39 ms | + +Other ModelZoo YOLO variants are also supported but have not been measured. Inference speeds vary with the host platform, so a slower host such as a Raspberry Pi 5 will report higher times than those above. + +:::note + +A few ModelZoo models can not be used with Frigate: SSD models (they are trained on Pascal VOC, so their labels do not match Frigate's), DAMO-YOLO models, face and pose models, and the PPU builds of YOLOv7. + +::: + ### 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 239cf4290a..7773c33ea6 100644 --- a/docs/docs/frigate/installation.md +++ b/docs/docs/frigate/installation.md @@ -381,6 +381,99 @@ 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 several separately versioned pieces, and they all have to agree. The driver, the runtime, and the daemon live on the Docker host; Frigate itself carries only the Python bindings, which it downloads on first start: + +| Component | Version | Installed on | Installed by | +| -------------- | -------- | ------------ | ------------------------- | +| Kernel driver | `v2.6.0` | Host | `user_installation.sh` | +| DX-RT runtime | `v3.4.0` | Host | `user_installation.sh` | +| NPU firmware | `v2.7.4` | The module | Flashed from the host | +| DX-RT bindings | `v3.4.0` | Frigate | Downloaded at first start | + +:::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 every version in the table before anything else. + +::: + +The installation script installs the DX-RT runtime on the host and enables `dxrt.service`, so the daemon starts at boot and any other program on the host can share the NPU with Frigate. Check the firmware version with `dxrt-cli --status` and update the module if it does not match the table above. + +#### 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. The same script installs the DX-RT runtime and enables `dxrt.service`, the daemon that owns the NPU and hands work to it on behalf of Frigate and anything else on the host. + +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* +``` + +Then confirm the daemon is running and listening in `/run/dxrt`: + +```bash +systemctl is-active dxrt.service +ls /run/dxrt/ +``` + +#### Setup + +To set up Frigate, follow the default installation instructions, for example: `ghcr.io/blakeblackshear/frigate:stable` + +#### Docker configuration + +Frigate needs the NPU device node and the directory holding the daemon's socket: + +```yaml +services: + frigate: + devices: + - /dev/dxrt0:/dev/dxrt0 + volumes: + - /run/dxrt:/run/dxrt +``` + +If you can't use Docker Compose, add `--device /dev/dxrt0:/dev/dxrt0 -v /run/dxrt:/run/dxrt` to your `docker run` command. + +Add one `--device` per NPU, contiguously from `/dev/dxrt0`, since the client stops enumerating at the first gap. + +The installation script configures `dxrt.service` to place its socket in `/run/dxrt` through a systemd drop-in. Mounting the directory rather than the socket file means the container sees the new socket after `dxrt.service` is restarted, rather than holding on to a deleted one. + +`dxrtd` listens on an abstract socket as well, but that one does not cross into a container, so Frigate names the filesystem socket through `DXRT_DYNAMIC_IPC_ENDPOINT` on your behalf. Set that variable on the container yourself only if the daemon listens somewhere else, which means you also set it for `dxrtd` through its own systemd drop-in. The script writes `/etc/systemd/system/dxrt.service.d/frigate.conf` for exactly that, and has `dxrt.service` link the socket to `/tmp/dxrt_dynamic_ipc.sock` when it starts, so the host's own `dxrt-cli` and `dxtop` keep finding it at the default path they fall back to. + +:::note + +The DX-RT client exits when `dxrt.service` stops, so restart the Frigate container after restarting `dxrt.service`. + +::: + +The device node is needed as well as the socket, because the client opens the NPU directly even though the daemon arbitrates access. Without it, inference fails with `Device not found`. + +`/dev/shm` does not need sharing. + +The DX-RT python bindings are not shipped in the Frigate image. Frigate downloads them on first start when a DEEPX detector is configured, and caches them under `/config`. + +#### 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/hardware.py b/frigate/detectors/hardware.py index 37b1f48e8e..1eb1111a76 100644 --- a/frigate/detectors/hardware.py +++ b/frigate/detectors/hardware.py @@ -273,6 +273,16 @@ def detect_memryx() -> DetectionHardware | None: return _hardware("memryx", "memryx", "MemryX MX3", units) +def detect_deepx() -> DetectionHardware | None: + """Find DEEPX NPUs by their device nodes.""" + units = _dev_units("dxrt*", "deepx:PCIe:{index}", "PCIe") + + if not units: + return None + + return _hardware("deepx", "deepx", "DEEPX NPU", 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") @@ -319,6 +329,7 @@ PROBES = ( detect_coral_usb, detect_hailo, detect_memryx, + detect_deepx, detect_intel_npu, detect_intel_gpu, detect_nvidia_gpu, diff --git a/frigate/detectors/plugins/deepx.py b/frigate/detectors/plugins/deepx.py new file mode 100644 index 0000000000..7ae0d04761 --- /dev/null +++ b/frigate/detectors/plugins/deepx.py @@ -0,0 +1,1166 @@ +"""DEEPX NPU detector running compiled .dxnn models via the DX-RT runtime.""" + +import json +import logging +import os +import re +import struct +from dataclasses import dataclass +from enum import Enum +from typing import Literal + +import cv2 +import numpy as np +from pydantic import ConfigDict, Field, field_validator + +from frigate.detectors.detection_api import DetectionApi +from frigate.detectors.detector_config import BaseDetectorConfig, ModelTypeEnum +from frigate.util.model import post_process_yolo, post_process_yolox +from frigate.util.runtime_deps import Artifact, ArtifactKind, RuntimeManifest + +logger = logging.getLogger(__name__) + +DETECTOR_KEY = "deepx" + +# Installed at first start; DEEPX's PyPI wheels match the libdxrt-bin build. +DXRT_VERSION = "3.4.0" + +# Where the host dxrtd accepts clients; a container cannot reach its abstract +# socket, and a mounted socket file pins a stale inode after a daemon restart. +DXRT_IPC_ENDPOINT_ENV = "DXRT_DYNAMIC_IPC_ENDPOINT" +DXRT_IPC_SOCKET = "/run/dxrt/dxrt_dynamic_ipc.sock" + +# Pre-NMS filter; Frigate applies per-object min_score and threshold afterwards. +SCORE_THRESHOLD = 0.4 +NMS_THRESHOLD = 0.4 + +# The device half of a `deepx:...` string, as the hardware probe writes it +DEVICE_INDEX = re.compile(r"(?:PCIe:)?(\d+)") + +# YOLOX's raw head concatenates one cell per position of its three strides +YOLOX_STRIDES = (8, 16, 32) + +# Fixed-width record the PPU emits, DeviceBoundingBox_t in DX-RT's +# datatype.h: x, y, w, h (float32), grid_y, grid_x, box_idx, layer_idx +# (uint8), score (float32), label (uint32), 4 bytes of padding. +PPU_RECORD_SIZE = 32 +PPU_BOX_BYTES = (0, 16) +PPU_GRID_BYTES = (16, 20) +PPU_SCORE_BYTES = (20, 24) +PPU_LABEL_BYTES = (24, 28) + +# Anchor sizes are in neither the PPU record nor the .dxnn, so a head is read +# against the table for its scale count: DEEPX's own reference decoder's, which +# every model it ships shares except YOLOv7. +PPU_ANCHORS_BY_SCALES = { + 2: np.array( + [ + [[10, 14], [23, 27], [37, 58]], + [[81, 82], [135, 169], [344, 319]], + ], + dtype=np.float32, + ), + 3: np.array( + [ + [[10, 13], [16, 30], [33, 23]], + [[30, 61], [62, 45], [59, 119]], + [[116, 90], [156, 198], [373, 326]], + ], + dtype=np.float32, + ), +} + +# .dxnn: "DXNN", a 4-byte LE version, then a JSON index padded to +# DXNN_HEADER_SIZE; every offset in the index counts from there. +DXNN_MAGIC = b"DXNN" +DXNN_HEADER_SIZE = 8192 +# The graph DX-COM keeps for visualisation, an ONNX ModelProto beside the +# encrypted model; only its nodes are read, for how the box tensor was built. +DXNN_GRAPH_SECTION = "vis_npu_models" +# Field numbers from onnx.proto +ONNX_GRAPH_FIELD = 7 +ONNX_NODE_FIELD = 1 +ONNX_NODE_FIELDS = {1: "input", 2: "output", 3: "name", 4: "op_type"} +# A YOLO head splits the DFL output into the two distances a box is built from +PPU_DISTANCE_OPS = ("Split", "Slice") +ONNX_WALK_LIMIT = 16 +PPU_TYPE_ANCHOR_BASED = 0 +PPU_TYPE_ANCHOR_FREE = 1 +# The PPU tensor table, ppu_info_header_t then one ppu_info_t per output +# tensor, as DX-RT's ppu_binary_parser.h defines them. +PPU_TABLE_HEADER = struct.Struct(" int: + """Resolve the NPU index from the device half of a `deepx:...` string: empty, a + bare index, or `PCIe:` as the hardware probe writes it.""" + if not configured: + return 0 + + # the whole string has to match: reading only the tail would take the 1 out + # of "PCIe:0,PCIe:1" and silently bind to an NPU the config never named + index = DEVICE_INDEX.fullmatch(configured.strip()) + + if index is None: + raise ValueError( + f'"{configured}" is not an NPU index; expected a number or ' + '"PCIe:". Run several NPUs by listing each one as its own ' + "devices entry, not by joining them into one string." + ) + + return int(index.group(1)) + + +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.""" + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "NMS input: %d boxes, best score %.4f, %d over threshold %.2f", + len(scores), + float(scores.max()) if len(scores) else 0.0, + int((scores >= score_threshold).sum()), + score_threshold, + ) + + 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) + + +class YoloLayout(str, Enum): + """How a yolo-generic .dxnn output is read; DX-COM keeps the source + model's head, so the head decides and the model is what gets asked.""" + + ppu = "ppu" # fixed-width records from the NPU's post-processing unit + yolox = "yolox" # (N, 5+C) grid-relative offsets and log sizes, model_type yolox + anchor = "anchor" # (N, 5+C) with an objectness column, pixel boxes + anchor_free = "anchor_free" # (N, 4+C) or (4+C, N), no objectness + nms_in_head = "nms_in_head" # (N, 6) final corner boxes + multipart = "multipart" # three NCHW feature maps + + +@dataclass(frozen=True) +class YoloOutput: + """The decoder a model needs and, for single-tensor layouts, its row + width, so the decoder can orient the tensor.""" + + layout: YoloLayout + columns: int | None = None + + +def class_count(labelmap: dict[int, str]) -> int: + """Highest label id plus one, which bounds the classes even when ids are sparse.""" + if not labelmap: + raise ValueError("the label map is empty; set labelmap_path") + + return max(labelmap) + 1 + + +def _significant_dims(shape: tuple[int, ...]) -> list[int]: + return [int(d) for d in shape if d != 1] + + +def infer_yolo_layout( + shapes: list[tuple[int, ...]], + num_classes: int, + ppu: bool, + dynamic_output: bool, +) -> YoloOutput: + """Pick the decoder from the output shapes the runtime reports.""" + if ppu: + return YoloOutput(YoloLayout.ppu) + + if len(shapes) == 0: + raise ValueError("the model reports no output tensor") + + if len(shapes) > 1: + return _infer_multipart(shapes, num_classes) + + if dynamic_output: + return YoloOutput(YoloLayout.nms_in_head, NMS_IN_HEAD_COLUMNS) + + dims = _significant_dims(shapes[0]) + + if len(dims) == 1: + dims = [1, dims[0]] + + if len(dims) != 2: + raise ValueError(f"unexpected output shape {tuple(shapes[0])}") + + candidates = [ + (NMS_IN_HEAD_COLUMNS, YoloLayout.nms_in_head), + (4 + num_classes, YoloLayout.anchor_free), + (5 + num_classes, YoloLayout.anchor), + ] + hits = [(d, layout) for d in dims for width, layout in candidates if d == width] + + if not hits: + raise ValueError( + f"output shape {tuple(shapes[0])} does not match a YOLO head with " + f"{num_classes} classes (expected a {4 + num_classes}, " + f"{5 + num_classes} or {NMS_IN_HEAD_COLUMNS} column output). Check that " + "labelmap_path matches the model, usually /labelmap/coco-80.txt." + ) + + widths = {d for d, _ in hits} + + if len(widths) > 1: + names = ", ".join(f"{d} columns ({layout.value})" for d, layout in hits) + raise ValueError( + f"output shape {tuple(shapes[0])} fits two layouts, {names}, and " + "the label map cannot tell them apart" + ) + + width = widths.pop() + rows = next(d for d in dims if d != width) if dims[0] != dims[1] else dims[0] + # a square output matches the same width on both axes, which is one layout + layouts = {layout for _, layout in hits} + + if len(layouts) == 1: + return YoloOutput(next(iter(layouts)), width) + + # 6 columns is both NMS-in-head and a 1 or 2 class raw head + raw = next(layout for layout in layouts if layout is not YoloLayout.nms_in_head) + if rows > NMS_IN_HEAD_MAX_ROWS: + return YoloOutput(raw, width) + + return YoloOutput(YoloLayout.nms_in_head, width) + + +def _infer_multipart(shapes: list[tuple[int, ...]], num_classes: int) -> YoloOutput: + reason = None + + if num_classes != MULTIPART_CLASSES: + reason = f"the shared decoder reads {MULTIPART_CLASSES}-class heads only" + elif len(shapes) != MULTIPART_OUTPUTS: + reason = f"expected {MULTIPART_OUTPUTS} feature maps, got {len(shapes)}" + elif any(len(shape) != 4 or shape[1] != MULTIPART_CHANNELS for shape in shapes): + reason = f"expected NCHW maps with {MULTIPART_CHANNELS} channels" + + if reason is not None: + raise ValueError( + f"output shapes {[tuple(shape) for shape in shapes]} are not the " + f"per-scale feature maps of an anchor-based YOLO head ({reason}). " + "A two-tensor output is a DAMO-YOLO head, which this detector does " + "not decode; otherwise export the model with its head included." + ) + + return YoloOutput(YoloLayout.multipart) + + +def validate_yolox_outputs( + shapes: list[tuple[int, ...]], num_classes: int, width: int, height: int +) -> int: + """Fail unless the shapes are YOLOX's raw (1, N, 5+C) head for this + input size; returns the row width to orient a channel-major export.""" + columns = 5 + num_classes + cells = sum((height // s) * (width // s) for s in YOLOX_STRIDES) + dims = [_significant_dims(shape) for shape in shapes] + + if ( + len(dims) == 1 + and len(dims[0]) == 2 + and sorted(dims[0]) == sorted((cells, columns)) + ): + return columns + + raise ValueError( + f"output shapes {[tuple(shape) for shape in shapes]} are not YOLOX's " + f"(1, {cells}, {columns}) raw head for a {width}x{height} input. Check " + "that width and height match the compiled model, that labelmap_path " + "matches it, usually /labelmap/coco-80.txt, and that model_type " + "matches it; any other YOLO head needs model_type yolo-generic." + ) + + +def rows_with_columns(tensor: np.ndarray, columns: int) -> np.ndarray: + """(N, columns) rows, transposing a channel-major export first.""" + if tensor.ndim >= 2 and tensor.shape[-1] != columns and tensor.shape[-2] == columns: + tensor = np.swapaxes(tensor, -1, -2) + + return tensor.reshape(-1, columns) + + +def reinterpret(tensor: np.ndarray, byte_range: tuple[int, int], dtype) -> np.ndarray: + """Read a byte column range of a uint8 record array as `dtype`.""" + lo, hi = byte_range + return np.ascontiguousarray(tensor[:, lo:hi]).view(dtype) + + +def ppu_records(outputs: list[np.ndarray]) -> np.ndarray | None: + """The (N, 32) uint8 record array a PPU model emits, or None.""" + if not outputs or outputs[0].ndim < 2 or outputs[0].size == 0: + # no candidate cleared the on-NPU filter this frame + return None + + records = outputs[0][0] + + if records.ndim != 2 or records.shape[1] != PPU_RECORD_SIZE: + logger.debug("Unexpected PPU output shape %s, skipping frame", records.shape) + return None + + return records + + +@dataclass(frozen=True) +class PpuLayout: + """The PPU head's kind (None if unnamed), each scale's grid, finest first, and + whether its boxes are a centre and size, as the compiled model states them.""" + + anchor_based: bool | None + grids: tuple[tuple[int, int], ...] + centre_boxes: bool | None = None + + @property + def scale_count(self) -> int: + return len(self.grids) + + def strides(self, width: int) -> tuple[int, ...]: + """One stride per scale, finest first, from the input width.""" + return tuple(round(width / grid_w) for grid_w, _ in self.grids) + + +def read_ppu_layout(path: str) -> PpuLayout | None: + """Read the PPU head layout the compiler wrote into a .dxnn, or None when there + is none to read.""" + try: + with open(path, "rb") as model: + header = model.read(DXNN_HEADER_SIZE) + if header[:4] != DXNN_MAGIC: + return None + + index = json.JSONDecoder().raw_decode( + header[8:].decode("utf-8", "replace") + )[0] + data = index["data"] + + def section(entry: dict) -> bytes: + model.seek(DXNN_HEADER_SIZE + int(entry["offset"])) + return model.read(int(entry["size"])) + + anchor_based = None + ppu = None + if data.get("compile_config"): + ppu = json.loads(section(data["compile_config"])).get("ppu") + if ppu is None: + return None + + anchor_based = { + PPU_TYPE_ANCHOR_BASED: True, + PPU_TYPE_ANCHOR_FREE: False, + }.get(ppu.get("type")) + + for chip in data.get("compiled_data", {}).values(): + for npu in chip.values(): + # v8 has a PPU table; v7's output tensor shapes hold the same grids + if npu.get("ppu", {}).get("size"): + grids = _ppu_grids_from_table(section(npu["ppu"])) + elif npu.get("rmap_info", {}).get("size"): + outputs = json.loads(section(npu["rmap_info"])).get( + "outputs", [] + ) + grids = _ppu_grids_from_outputs(outputs, ppu) + if grids and anchor_based is None: + anchor_based = _ppu_outputs_are_anchor_based(outputs, ppu) + else: + continue + + if not grids: + continue + + scales = tuple(grids[k] for k in sorted(grids)) + centre_boxes = None + bbox_node = _ppu_bbox_node(ppu) + if anchor_based is False and len(scales) == 1 and bbox_node: + nodes = [] + try: + for entry in (data.get(DXNN_GRAPH_SECTION) or {}).values(): + nodes += _onnx_nodes(section(entry)) + except (OSError, ValueError, IndexError, KeyError, TypeError): + nodes = [] + + centre_boxes = ppu_boxes_are_centres(nodes, bbox_node) + + return PpuLayout(anchor_based, scales, centre_boxes) + + return None + except ( + OSError, + ValueError, + KeyError, + IndexError, + TypeError, + AttributeError, + struct.error, + ): + return None + + +def _ppu_grids_from_table(table: bytes) -> dict[int, tuple[int, int]]: + _, tensor_count, _, _ = PPU_TABLE_HEADER.unpack_from(table) + if len(table) < PPU_TABLE_HEADER.size + tensor_count * PPU_TABLE_ENTRY.size: + raise ValueError("PPU table shorter than its tensor count") + + grids: dict[int, tuple[int, int]] = {} + for i in range(tensor_count): + fields = PPU_TABLE_ENTRY.unpack_from( + table, PPU_TABLE_HEADER.size + i * PPU_TABLE_ENTRY.size + ) + grids.setdefault(fields[3], (fields[8], fields[9])) + + return grids + + +def _ppu_output_scale(name: str, ppu: dict | None) -> int | None: + entry = ((ppu or {}).get("outputs") or {}).get(name) + if isinstance(entry, dict) and "conv_idx" in entry: + return int(entry["conv_idx"]) + + match = re.fullmatch(r"PPU_\w*?Output_(\d+)(?:_anchor_\d+)?", name) + return int(match.group(1)) if match else None + + +def _ppu_grids_from_outputs( + outputs: list[dict], ppu: dict | None +) -> dict[int, tuple[int, int]]: + grids: dict[int, tuple[int, int]] = {} + for output in outputs: + scale = _ppu_output_scale(str(output.get("name", "")), ppu) + shape = output.get("shape") or [] + if scale is None or len(shape) not in (3, 4): + continue + + grid = (int(shape[2]), int(shape[1])) if len(shape) == 4 else (int(shape[1]), 1) + grids.setdefault(scale, grid) + + return grids + + +def _ppu_outputs_are_anchor_based(outputs: list[dict], ppu: dict | None) -> bool | None: + mapping = (ppu or {}).get("outputs") or {} + if any( + isinstance(entry, dict) and "anchor_idx" in entry for entry in mapping.values() + ): + return True + + if any("_anchor_" in str(output.get("name", "")) for output in outputs): + return True + + return None + + +def _ppu_bbox_node(ppu: dict | None) -> str | None: + layers = (ppu or {}).get("layer") + if not isinstance(layers, list) or len(layers) != 1: + return None + + layer = layers[0] + return layer.get("bbox") if isinstance(layer, dict) else None + + +def _proto_varint(blob: bytes, at: int) -> tuple[int, int]: + value = shift = 0 + while True: + byte = blob[at] + at += 1 + value |= (byte & 0x7F) << shift + if not byte & 0x80: + return value, at + + shift += 7 + + +def _proto_fields(blob: bytes): + at = 0 + while at < len(blob): + key, at = _proto_varint(blob, at) + field, wire = key >> 3, key & 7 + if wire == 0: + _, at = _proto_varint(blob, at) + elif wire == 2: + size, at = _proto_varint(blob, at) + yield field, blob[at : at + size] + at += size + elif wire == 5: + at += 4 + elif wire == 1: + at += 8 + else: + raise ValueError(f"unknown protobuf wire type {wire}") + + +def _onnx_nodes(blob: bytes) -> list[dict]: + nodes = [] + for field, graph in _proto_fields(blob): + if field != ONNX_GRAPH_FIELD: + continue + + for graph_field, node in _proto_fields(graph): + if graph_field != ONNX_NODE_FIELD: + continue + + entry = {"input": [], "output": [], "name": "", "op_type": ""} + for node_field, value in _proto_fields(node): + key = ONNX_NODE_FIELDS.get(node_field) + if key is None: + continue + + text = value.decode("utf-8", "replace") + if key in ("input", "output"): + entry[key].append(text) + else: + entry[key] = text + + nodes.append(entry) + + return nodes + + +def _upstream_concat(node: dict | None, producers: dict) -> dict | None: + seen: set[str] = set() + queue = [node] if node else [] + while queue and len(seen) <= ONNX_WALK_LIMIT: + current = queue.pop(0) + if current["op_type"] == "Concat": + return current + + for name in current["input"]: + if name in seen: + continue + + seen.add(name) + producer = producers.get(name) + if producer is not None: + queue.append(producer) + + return None + + +def _box_tensor_sources(producers: dict, tensor: str) -> set[str]: + sources: set[str] = set() + seen: set[str] = set() + pending = [(tensor, 0)] + while pending: + name, depth = pending.pop() + if name in seen: + continue + + seen.add(name) + node = producers.get(name) + if ( + node is None + or depth >= ONNX_WALK_LIMIT + or node["op_type"] in PPU_DISTANCE_OPS + ): + sources.add(name) + continue + + pending += [(read, depth + 1) for read in node["input"]] + + return sources + + +def ppu_boxes_are_centres(nodes: list[dict], bbox_node: str) -> bool | None: + """Whether the PPU's box tensor holds a centre and size rather than two + corners, read from how the compiled graph built it. A YOLO head turns + the two DFL distances into corners as (anchor - left, anchor + right), + one distance per half, or into a centre and size as their mean and + their difference, both halves drawing on both distances. None when the + graph is not there or built the boxes some other way; nothing here + reads a node's name, only the shape of the graph around it.""" + producers = {out: node for node in nodes for out in node["output"]} + by_name = {node["name"]: node for node in nodes} + + concat = _upstream_concat(by_name.get(bbox_node), producers) + if concat is None or len(concat["input"]) != 2: + return None + + first, second = (_box_tensor_sources(producers, t) for t in concat["input"]) + distances = { + tensor + for tensor in first | second + if producers.get(tensor, {}).get("op_type") in PPU_DISTANCE_OPS + } + if len(distances) != 2: + return None + + first &= distances + second &= distances + if first == second == distances: + return True + + if first != second and len(first) == len(second) == 1: + return False + + return None + + +def ppu_grid_regression_geometry( + records: np.ndarray, boxes: np.ndarray, strides: np.ndarray +) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]: + """Pixel centre and size for a YOLOX-style grid-relative box: centre + offset from the grid cell, size a log-scale multiple of the stride. + `strides` is one per scale, finest first.""" + grid = reinterpret(records, PPU_GRID_BYTES, np.uint8) + grid_y = grid[:, 0].astype(np.float32) + grid_x = grid[:, 1].astype(np.float32) + layer_idx = grid[:, 3].astype(np.int32) + + known = layer_idx < len(strides) + layer = np.where(known, layer_idx, 0) + stride = strides[layer] + + centre_x = (boxes[:, 0] + grid_x) * stride + centre_y = (boxes[:, 1] + grid_y) * stride + box_w = np.exp(boxes[:, 2]) * stride + box_h = np.exp(boxes[:, 3]) * stride + + return centre_x, centre_y, box_w, box_h, known + + +def ppu_anchor_geometry( + records: np.ndarray, boxes: np.ndarray, strides: np.ndarray +) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]: + """Pixel centre and size for anchor-based PPU records, with the mask of + records the anchor table covers. `strides` is one per scale, finest + first; a scale count with no table leaves every record unplaced.""" + grid = reinterpret(records, PPU_GRID_BYTES, np.uint8) + grid_y = grid[:, 0].astype(np.float32) + grid_x = grid[:, 1].astype(np.float32) + box_idx = grid[:, 2].astype(np.int32) + layer_idx = grid[:, 3].astype(np.int32) + + anchor_table = PPU_ANCHORS_BY_SCALES.get(len(strides)) + if anchor_table is None: + zeros = np.zeros(len(layer_idx), np.float32) + return zeros, zeros, zeros, zeros, np.zeros(len(layer_idx), dtype=bool) + + levels, per_level = anchor_table.shape[:2] + known = (layer_idx < levels) & (box_idx < per_level) + + # fold the out-of-range rows onto a real entry and let the mask drop them + layer = np.where(known, layer_idx, 0) + anchors = anchor_table[layer, np.where(known, box_idx, 0)] + stride = strides[layer] + + centre_x = (boxes[:, 0] * 2.0 - 0.5 + grid_x) * stride + centre_y = (boxes[:, 1] * 2.0 - 0.5 + grid_y) * stride + box_w = (boxes[:, 2] * 2.0) ** 2 * anchors[:, 0] + box_h = (boxes[:, 3] * 2.0) ** 2 * anchors[:, 1] + + return centre_x, centre_y, box_w, box_h, known + + +def decode_ppu( + outputs: list[np.ndarray], + width: int, + height: int, + score_threshold: float, + nms_threshold: float, + *, + strides: tuple[int, ...] | np.ndarray, + anchor_based: bool, + centre_boxes: bool = False, + records: np.ndarray | None = None, +) -> np.ndarray: + """Decode PPU records, leaving only NMS. `strides` is one per scale, + finest first.""" + if records is None: + 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() + strides = np.asarray(strides, np.float32) + + if anchor_based: + centre_x, centre_y, box_w, box_h, known = ppu_anchor_geometry( + records, boxes, strides + ) + # a record the table cannot place is dropped by the score filter + scores = np.where(known, scores, 0.0) + x_min, y_min = centre_x - box_w * 0.5, centre_y - box_h * 0.5 + elif len(strides) > 1: + # a multi-scale anchor-free head is YOLOX-style, grid-relative + centre_x, centre_y, box_w, box_h, known = ppu_grid_regression_geometry( + records, boxes, strides + ) + scores = np.where(known, scores, 0.0) + x_min, y_min = centre_x - box_w * 0.5, centre_y - box_h * 0.5 + elif centre_boxes: + box_w, box_h = boxes[:, 2], boxes[:, 3] + x_min, y_min = boxes[:, 0] - box_w * 0.5, boxes[:, 1] - box_h * 0.5 + else: + # the head wrote two corners, so its edges are the answer already + x_min, y_min = boxes[:, 0], boxes[:, 1] + box_w, box_h = boxes[:, 2] - x_min, boxes[:, 3] - y_min + + 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_raw_anchor( + outputs: list[np.ndarray], + width: int, + height: int, + score_threshold: float, + nms_threshold: float, + columns: int | None = None, +) -> np.ndarray: + """Decode an anchor-based head: (N, 5+C) rows of cx, cy, w, h, objectness and + class scores in pixels, transposed when `columns` says so.""" + tensor = outputs[0] + + if tensor.ndim < 2 or tensor.size == 0: + return np.zeros((20, 6), np.float32) + + predictions = rows_with_columns(tensor, columns or tensor.shape[-1]) + + 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_raw_nms_in_head( + outputs: list[np.ndarray], + width: int, + height: int, + score_threshold: float, +) -> np.ndarray: + """Decode a head that already ran NMS: (N, 6) rows of corner box, + score and class in pixels; only the score filter is left.""" + tensor = outputs[0] + + if tensor.ndim < 2 or tensor.size == 0: + return np.zeros((20, 6), np.float32) + + predictions = rows_with_columns(tensor, NMS_IN_HEAD_COLUMNS) + + 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: str = Field( + default="", + title="DEEPX device", + description="Which NPU this detector binds to, as PCIe:. " + "Empty selects the first NPU.", + ) + + @field_validator("device") + @classmethod + def validate_device(cls, value: str) -> str: + """Reject an unusable device at startup rather than at first inference.""" + resolve_device(value) + return value + + +class DeepxDetector(DetectionApi): + """DEEPX NPU detector: DX-RT session plus the decoder the model needs.""" + + type_key = DETECTOR_KEY + runtime_manifest = DEEPX_MANIFEST + supported_models = [ModelTypeEnum.yologeneric, ModelTypeEnum.yolox] + + def __init__(self, config: DeepxDetectorConfig): + # before the runtime is installed or the socket is looked for: this + # needs no hardware, and ssd, the default, would otherwise be decoded + # as YOLO and return nonsense rather than an error + if config.model.model_type not in self.supported_models: + supported = ", ".join(t.value for t in self.supported_models) + raise ValueError( + f"model_type '{config.model.model_type.value}' is not supported " + f"by the DEEPX detector. Set model.model_type to one of: " + f"{supported}." + ) + + self.activate_dependencies() + + # skip DX-RT's abstract-socket attempt, which cannot leave the container + if not os.environ.get(DXRT_IPC_ENDPOINT_ENV): + os.environ[DXRT_IPC_ENDPOINT_ENV] = DXRT_IPC_SOCKET + endpoint = os.environ[DXRT_IPC_ENDPOINT_ENV] + + if not endpoint.startswith("@") and not os.path.exists(endpoint): + # DX-RT reports this as a bare connect error several layers down + logger.warning( + "No dxrtd socket at %s. Rerun the DEEPX installation script on " + "the host so dxrt.service listens there, mount /run/dxrt into " + "the container, or set %s to where the daemon actually listens", + endpoint, + DXRT_IPC_ENDPOINT_ENV, + ) + + try: + from dx_engine import Configuration, InferenceEngine, InferenceOption + except ModuleNotFoundError: + raise ImportError( + "The DX-RT python bindings are not installed. Frigate installs " + "them at startup when a DEEPX detector is configured; check the " + "startup log for errors." + ) from None + + # SERVICE gates only DX-RT's /proc scan for dxrtd, which a container + # cannot do for the host; the IPC client still needs the socket + Configuration().set_enable(Configuration.ITEM.SERVICE, False) + + super().__init__(config) + + self.model_type = config.model.model_type + 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." + ) + + device = resolve_device(config.device) + logger.info("Loading DEEPX model %s on device %s", model_path, device) + + options = InferenceOption() + options.devices = [device] + options.bound_option = InferenceOption.BOUND_OPTION.NPU_ALL + + self.session = InferenceEngine(str(model_path), options) + self.output = self.inspect_model(config) + if self.output.layout is YoloLayout.yolox: + self.calculate_grids_strides() + self.logged_layout = False + self.ppu_unsupported_scale_reported = False + + self.ppu_layout: PpuLayout | None = None + self.ppu_strides: np.ndarray | None = None + if self.output.layout is YoloLayout.ppu: + self.ppu_layout = self.inspect_ppu_head(model_path) + self.ppu_strides = np.asarray( + self.ppu_layout.strides(self.width), np.float32 + ) + + def inspect_ppu_head(self, model_path: str) -> PpuLayout: + """Read the PPU head layout DX-COM writes into the model file.""" + layout = read_ppu_layout(model_path) + + if layout is None: + raise ValueError( + f"Could not read the PPU head layout from {model_path}; the " + "file is damaged or was compiled before DX-COM 2.4.0. Compile " + "PPU models with DX-COM 2.4.0 or later." + ) + + if layout.anchor_based is None: + raise ValueError( + f"The PPU head in {model_path} names no kind Frigate can decode " + "(anchor-based or anchor-free YOLO); face and pose PPU models are " + "not supported, and PPU models must be compiled with DX-COM 2.4.0 " + "or later." + ) + + if ( + layout.anchor_based is False + and layout.scale_count == 1 + and layout.centre_boxes is None + ): + # both readings are plausible geometry, so only the graph can settle it + raise ValueError( + f"Could not tell from {model_path} whether its PPU head writes " + "boxes as a centre and size or as two corners. The compiled " + "graph that says so is missing from the file or builds its " + "boxes in a way Frigate does not recognise. Compile PPU models " + "with DX-COM 2.4.0 or later." + ) + + logger.info( + "DEEPX PPU head from the model: %s, %d scale(s), grids %s%s", + "anchor-based" if layout.anchor_based else "anchor-free", + layout.scale_count, + layout.grids, + "" + if layout.centre_boxes is None + else ( + ", boxes as a centre and size" + if layout.centre_boxes + else ", boxes as two corners" + ), + ) + return layout + + def inspect_model(self, config: DeepxDetectorConfig) -> YoloOutput: + """Pick the decoder from what the runtime reports, failing here rather than + on the first frame.""" + shapes = [ + tuple(info["shape"]) for info in self.session.get_output_tensors_info() + ] + + try: + num_classes = class_count(config.model.merged_labelmap) + + if self.model_type == ModelTypeEnum.yolox and not self.session.is_ppu(): + # a YOLOX compiled with PPU support is read as PPU below + columns = validate_yolox_outputs( + shapes, num_classes, self.width, self.height + ) + output = YoloOutput(YoloLayout.yolox, columns) + else: + output = infer_yolo_layout( + shapes, + num_classes, + self.session.is_ppu(), + self.session.has_dynamic_output(), + ) + except ValueError as err: + raise ValueError( + f"Cannot decode DEEPX model '{config.model.path}': {err}" + ) from None + + logger.info( + "DEEPX decoding %s output as %s with %d classes", + self.model_type.value, + output.layout.value, + num_classes, + ) + return output + + def decode(self, outputs: list[np.ndarray]) -> np.ndarray: + """Decode the model output according to its type and detected layout.""" + match self.output.layout: + case YoloLayout.ppu: + return self.decode_ppu(outputs) + case YoloLayout.yolox: + return self.decode_yolox(outputs) + case YoloLayout.anchor: + return decode_raw_anchor( + outputs, + self.width, + self.height, + SCORE_THRESHOLD, + NMS_THRESHOLD, + self.output.columns, + ) + case YoloLayout.anchor_free: + rows = rows_with_columns(outputs[0], self.output.columns) + return post_process_yolo([rows], self.width, self.height) + case YoloLayout.nms_in_head: + return decode_raw_nms_in_head( + outputs, self.width, self.height, SCORE_THRESHOLD + ) + case _: + return post_process_yolo(outputs, self.width, self.height) + + def decode_yolox(self, outputs: list[np.ndarray]) -> np.ndarray: + """Decode YOLOX's raw head with Frigate's shared decoder.""" + rows = rows_with_columns(outputs[0], self.output.columns) + predictions = np.array(rows, dtype=np.float32).reshape(1, -1, rows.shape[-1]) + + return post_process_yolox( + predictions, self.width, self.height, self.grids, self.expanded_strides + ) + + def decode_ppu(self, outputs: list[np.ndarray]) -> np.ndarray: + """Decode PPU records against the head layout read from the model.""" + layout = self.ppu_layout + assert layout is not None + + records = ppu_records(outputs) + + if ( + layout.anchor_based + and layout.scale_count not in PPU_ANCHORS_BY_SCALES + and not self.ppu_unsupported_scale_reported + ): + self.ppu_unsupported_scale_reported = True + logger.error( + "This PPU model's anchor-based head reports %d detection " + "scales, which Frigate has no anchor table for (supported: " + "%s); its detections cannot be decoded", + layout.scale_count, + ", ".join(str(n) for n in sorted(PPU_ANCHORS_BY_SCALES)), + ) + + return decode_ppu( + outputs, + self.width, + self.height, + SCORE_THRESHOLD, + NMS_THRESHOLD, + strides=self.ppu_strides, + anchor_based=bool(layout.anchor_based), + centre_boxes=bool(layout.centre_boxes), + records=records, + ) + + def detect_raw(self, tensor_input): + """Run inference and decode the model output.""" + outputs = self.session.run([tensor_input]) + + if not isinstance(outputs, list): + outputs = [outputs] + + if not self.logged_layout: + # before decoding, so a decode failure still leaves the shapes + self.logged_layout = True + self.log_layout(tensor_input, outputs) + + detections = self.decode(outputs) + + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "DEEPX decoded %d detections, best score %.4f", + int((detections[:, 1] > 0).sum()), + float(detections[:, 1].max()), + ) + + return detections + + def log_layout(self, tensor_input, outputs: list[np.ndarray]) -> None: + """Log the real tensor layout once, since a wrong decode returns empty + detections rather than failing.""" + tensors = [("input", tensor_input)] + tensors += [(f"output[{i}]", out) for i, out in enumerate(outputs)] + + for name, tensor in tensors: + array = np.asarray(tensor) + logger.info( + "DEEPX %s: shape=%s dtype=%s min=%s max=%s", + name, + array.shape, + array.dtype, + array.min() if array.size else "n/a", + array.max() if array.size else "n/a", + ) diff --git a/frigate/test/test_deepx.py b/frigate/test/test_deepx.py new file mode 100644 index 0000000000..458f4733fd --- /dev/null +++ b/frigate/test/test_deepx.py @@ -0,0 +1,896 @@ +"""Tests for the DEEPX detector.""" + +import json +import os +import struct +import sys +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +import numpy as np +from pydantic import ValidationError + +from frigate.detectors.detector_config import ModelConfig, ModelTypeEnum +from frigate.detectors.device import ( + DeviceParseError, + build_detector_config, + parse_device, +) +from frigate.detectors.plugins.deepx import ( + DEEPX_MANIFEST, + DXRT_VERSION, + PPU_RECORD_SIZE, + DeepxDetector, + DeepxDetectorConfig, + PpuLayout, + YoloLayout, + class_count, + decode_raw_anchor, + decode_raw_nms_in_head, + infer_yolo_layout, + read_ppu_layout, + resolve_device, + validate_yolox_outputs, +) + + +def model_with_type(model_type) -> ModelConfig: + return ModelConfig( + model_type=model_type, + labelmap_path=None, + labelmap={79: "toothbrush"}, + width=640, + height=640, + ) + + +def build_ppu_record(box, score=0.9, label=0, grid=(7, 9, 2, 2)) -> np.ndarray: + 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 + 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.reshape(1, 1, PPU_RECORD_SIZE) + + +# compile_config.ppu as DX-COM writes it for the two head kinds +ANCHOR_BASED_PPU = {"type": 0, "num_classes": 80, "activation": "Sigmoid"} +ANCHOR_FREE_PPU = {"type": 1, "num_classes": 80} + +PPU_BBOX_NODE = "/head/Mul_2" + +# (grid_w, grid_h, entries) per scale, finest first; the grids give the +# strides at a 640 input +THREE_SCALE_ANCHORS = [(80, 80, 3), (40, 40, 3), (20, 20, 3)] +TWO_SCALE_ANCHORS = [(40, 40, 3), (20, 20, 3)] +FOUR_SCALE_ANCHORS = [(160, 160, 3), (80, 80, 3), (40, 40, 3), (20, 20, 3)] +THREE_SCALE_FREE = [(80, 80, 1), (40, 40, 1), (20, 20, 1)] +ONE_SCALE_FREE = [(100, 84, 1)] + + +def proto_varint(value: int) -> bytes: + out = bytearray() + while True: + byte = value & 0x7F + value >>= 7 + out.append(byte | (0x80 if value else 0)) + if not value: + return bytes(out) + + +def proto_bytes(field: int, payload: bytes) -> bytes: + return proto_varint(field << 3 | 2) + proto_varint(len(payload)) + payload + + +def proto_number(field: int, value: int) -> bytes: + return proto_varint(field << 3) + proto_varint(value) + + +def onnx_node(op_type: str, name: str, inputs: list, outputs: list) -> bytes: + body = b"".join(proto_bytes(1, tensor.encode()) for tensor in inputs) + body += b"".join(proto_bytes(2, tensor.encode()) for tensor in outputs) + return body + proto_bytes(3, name.encode()) + proto_bytes(4, op_type.encode()) + + +def onnx_box_graph(box_format: str) -> bytes: + if box_format == "broken": + return proto_varint(1 << 3 | 3) + + unnamed = box_format == "unnamed" + + def graph_node(op_type: str, name: str, inputs: list, outputs: list) -> bytes: + return onnx_node(op_type, "" if unnamed else name, inputs, outputs) + + nodes = [ + graph_node("Split", "dfl_split", ["dfl", "sizes"], ["lt", "rb"]), + graph_node("Sub", "corner_min", ["anchors", "lt"], ["x1y1"]), + graph_node("Add", "corner_max", ["rb", "anchors"], ["x2y2"]), + ] + if box_format in ("centre", "unnamed"): + nodes += [ + graph_node("Add", "corner_sum", ["x1y1", "x2y2"], ["sum"]), + graph_node("Mul", "corner_mean", ["sum", "half"], ["cxy"]), + graph_node("Sub", "corner_span", ["x2y2", "x1y1"], ["wh"]), + graph_node("Concat", "box_concat", ["cxy", "wh"], ["box"]), + ] + elif box_format == "corner": + nodes.append(graph_node("Concat", "box_concat", ["x1y1", "x2y2"], ["box"])) + else: + nodes.append(graph_node("Concat", "box_concat", ["x1y1", "x1y1"], ["box"])) + + box = "box" + if unnamed: + box = "box_flat" + nodes.append(graph_node("Reshape", "box_reshape", ["shape", "box"], [box])) + + nodes.append(onnx_node("Mul", PPU_BBOX_NODE, [box, "strides"], ["bbox_out"])) + graph = b"".join(proto_bytes(1, node) for node in nodes) + graph += proto_bytes(5, b"weights") + return ( + proto_number(1, 10) # ir_version + + proto_bytes(2, b"onnx_frontend_compiler") # producer_name + + proto_bytes(7, graph) + ) + + +def write_dxnn( + directory, ppu, layers, name="model.dxnn", table=True, box_format=None +) -> str: + """A minimal .dxnn as DX-RT's parsers read it: the container header, a + compile_config carrying `ppu`, and either the PPU tensor table with one + entry per (layer, anchor) as a v8 file has, or with `table` False only + the rmap_info listing of the PPU output tensors, as a v7 file has. + `layers` is (grid_w, grid_h, entries) per scale, finest first; an + anchor-free scale has one entry, and grid_h 1 means a flattened + (1, cells, channels) tensor. `box_format`, "centre" or "corner", adds + the compiled graph that says how the head writes its boxes, along with + the compile_config layer naming the node the PPU reads them from.""" + if box_format is not None and "layer" not in ppu: + ppu = dict(ppu, layer=[{"bbox": PPU_BBOX_NODE, "cls_conf": "/head/Sigmoid"}]) + + compile_config = json.dumps({"compile_version": "2.4.0", "ppu": ppu}).encode() + graph = onnx_box_graph(box_format) if box_format is not None else b"" + + if table: + part = bytearray(struct.pack(" 1: + name_ += f"_anchor_{anchor}" + shape = [1, grid_w, 127] if grid_h == 1 else [1, grid_h, grid_w, 128] + outputs.append({"name": name_, "shape": shape, "layout": "PPU_YOLO"}) + part = json.dumps({"inputs": [], "outputs": outputs}).encode() + + data = { + "compile_config": { + "type": "str", + "offset": 0, + "size": len(compile_config), + }, + "compiled_data": { + "M1A_4K": { + "npu_0": { + "rmap": {"type": "bytes", "offset": 0, "size": 0}, + "ppu" if table else "rmap_info": { + "type": "bytes" if table else "str", + "offset": len(compile_config), + "size": len(part), + }, + } + } + }, + } + if graph: + data["vis_npu_models"] = { + "npu_0": { + "type": "bytes", + "offset": len(compile_config) + len(part), + "size": len(graph), + } + } + + index = json.dumps( + { + "version": 8 if table else 7, + "signature": "DXNN", + "size": 8192, + "data": data, + } + ).encode() + + path = os.path.join(directory, name) + with open(path, "wb") as model: + model.write(b"DXNN" + struct.pack(" YoloLayout: + return infer_yolo_layout(shapes, num_classes, ppu, dynamic_output).layout + + +class TestDeepxModelFile(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + + def layout(self, ppu, layers, **kwargs) -> PpuLayout | None: + return read_ppu_layout(write_dxnn(self.tmp.name, ppu, layers, **kwargs)) + + def test_the_head_kind_and_one_grid_per_scale_are_read(self): + cases = { + "anchor-based: three anchors per scale collapse to one grid each": ( + (ANCHOR_BASED_PPU, THREE_SCALE_ANCHORS, {}), + PpuLayout(anchor_based=True, grids=((80, 80), (40, 40), (20, 20))), + ), + "anchor-free, one scale flattened into a single tensor": ( + (ANCHOR_FREE_PPU, ONE_SCALE_FREE, {"box_format": "centre"}), + PpuLayout(anchor_based=False, grids=((100, 84),), centre_boxes=True), + ), + "a head kind the compiler does not name still yields the scales": ( + ({"type": 7}, [(80, 80, 3), (40, 40, 3)], {}), + PpuLayout(anchor_based=None, grids=((80, 80), (40, 40))), + ), + "no table: the per-anchor split says anchor-based": ( + ({"num_classes": 80}, THREE_SCALE_ANCHORS, {"table": False}), + PpuLayout(anchor_based=True, grids=((80, 80), (40, 40), (20, 20))), + ), + "no table: compile_config places the scales ahead of the names": ( + ( + { + "type": 1, + "outputs": { + f"PPU_Transpose_Output_{i}": {"conv_idx": 2 - i} + for i in range(3) + }, + }, + [(20, 20, 1), (40, 40, 1), (80, 80, 1)], + {"table": False}, + ), + PpuLayout(anchor_based=False, grids=((80, 80), (40, 40), (20, 20))), + ), + "no table: every scale in one (1, cells, channels) tensor": ( + (ANCHOR_FREE_PPU, [(8400, 1, 1)], {"table": False}), + PpuLayout(anchor_based=False, grids=((8400, 1),)), + ), + } + + for head, ((ppu, layers, kwargs), expected) in cases.items(): + with self.subTest(head=head): + self.assertEqual(self.layout(ppu, layers, **kwargs), expected) + + def test_the_box_format_comes_from_the_compiled_graph(self): + for box_format, centre in ( + ("centre", True), + ("corner", False), + ("unnamed", True), + ): + with self.subTest(box_format=box_format): + self.assertEqual( + self.layout(ANCHOR_FREE_PPU, ONE_SCALE_FREE, box_format=box_format), + PpuLayout( + anchor_based=False, grids=((100, 84),), centre_boxes=centre + ), + ) + + def test_a_box_format_the_graph_does_not_answer_is_left_open(self): + cases = { + "no compiled graph at all": (ANCHOR_FREE_PPU, ONE_SCALE_FREE, {}), + "a graph without the node compile_config names": ( + dict(ANCHOR_FREE_PPU, layer=[{"bbox": "/head/Missing"}]), + ONE_SCALE_FREE, + {"box_format": "centre"}, + ), + "a graph that builds its boxes from neither shape": ( + ANCHOR_FREE_PPU, + ONE_SCALE_FREE, + {"box_format": "neither"}, + ), + "a graph section the reader cannot make sense of": ( + ANCHOR_FREE_PPU, + ONE_SCALE_FREE, + {"box_format": "broken"}, + ), + "a grid-decoded head, where the question does not arise": ( + ANCHOR_FREE_PPU, + THREE_SCALE_FREE, + {"box_format": "centre"}, + ), + } + + for graph, (ppu, layers, kwargs) in cases.items(): + with self.subTest(graph=graph): + self.assertIsNone(self.layout(ppu, layers, **kwargs).centre_boxes) + + def test_nothing_is_read_from_a_model_without_ppu_metadata(self): + self.assertIsNone(read_ppu_layout(os.path.join(self.tmp.name, "missing"))) + + path = write_dxnn(self.tmp.name, None, [(80, 80, 3)]) + self.assertIsNone(read_ppu_layout(path)) + + with open(path, "wb") as model: + model.write(b"ONNX" + b"\0" * 100) + self.assertIsNone(read_ppu_layout(path)) + + path = write_dxnn(self.tmp.name, ANCHOR_BASED_PPU, [(80, 80, 3), (40, 40, 3)]) + os.truncate(path, os.path.getsize(path) - 40) + self.assertIsNone(read_ppu_layout(path)) + + +class TestDeepxLayoutInference(unittest.TestCase): + def test_a_shape_and_class_count_pick_one_layout(self): + cases = { + "anchor-free, four columns ahead of the classes": ( + [(1, 84, 8400)], + 80, + YoloLayout.anchor_free, + 84, + ), + "anchor-free, row-major": ([(1, 8400, 84)], 80, YoloLayout.anchor_free, 84), + "anchor-based, an objectness column as well": ( + [(1, 25200, 85)], + 80, + YoloLayout.anchor, + 85, + ), + "anchor-based, channel-major": ( + [(1, 85, 25200)], + 80, + YoloLayout.anchor, + 85, + ), + "NMS in the head, a fixed run of corner records": ( + [(1, 300, 6)], + 80, + YoloLayout.nms_in_head, + None, + ), + # only the label map can tell these two apart + "85 columns with 81 classes is anchor-free": ( + [(1, 8400, 85)], + 81, + YoloLayout.anchor_free, + 85, + ), + "85 columns with 80 classes is anchor-based": ( + [(1, 8400, 85)], + 80, + YoloLayout.anchor, + 85, + ), + # 6 columns is all three layouts, told apart by the row count + "6 columns and thousands of rows, one class": ( + [(1, 25200, 6)], + 1, + YoloLayout.anchor, + 6, + ), + "6 columns and thousands of rows, two classes": ( + [(1, 8400, 6)], + 2, + YoloLayout.anchor_free, + 6, + ), + "6 columns and a few hundred rows, one class": ( + [(1, 300, 6)], + 1, + YoloLayout.nms_in_head, + None, + ), + "7 columns with two classes is only anchor-based": ( + [(1, 8400, 7)], + 2, + YoloLayout.anchor, + 7, + ), + # a square output matches the same width on both axes, which must + # not read as two candidate layouts + "a square output is not ambiguous with itself": ( + [(1, 85, 85)], + 80, + YoloLayout.anchor, + 85, + ), + "three NCHW maps with 255 channels are feature maps": ( + [(1, 255, 80, 80), (1, 255, 40, 40), (1, 255, 20, 20)], + 80, + YoloLayout.multipart, + None, + ), + } + + for head, (shapes, num_classes, layout, columns) in cases.items(): + with self.subTest(head=head): + output = infer_yolo_layout(shapes, num_classes, False, False) + + self.assertIs(output.layout, layout) + if columns is not None: + self.assertEqual(output.columns, columns) + + def test_the_runtime_flags_outrank_the_shapes(self): + self.assertIs(layout_of([(8400,)], 80, ppu=True), YoloLayout.ppu) + self.assertIs( + layout_of([(1, -1, 6)], 80, dynamic_output=True), YoloLayout.nms_in_head + ) + + def test_a_shape_no_layout_fits_is_refused_with_the_reason(self): + cases = { + "fits two layouts": ([(1, 84, 6)], 80), + "labelmap_path": ([(1, 8400, 84)], 91), + "no output tensor": ([], 80), + "255 channels": ( + [(1, 80, 80, 255), (1, 40, 40, 255), (1, 20, 20, 255)], + 80, + ), + "feature maps": ([(1, 8400, 80), (1, 8400, 4)], 80), + "80-class": ([(1, 24, 80, 80), (1, 24, 40, 40), (1, 24, 20, 20)], 3), + } + + for reason, (shapes, num_classes) in cases.items(): + with ( + self.subTest(reason=reason), + self.assertRaisesRegex(ValueError, reason), + ): + layout_of(shapes, num_classes) + + +class TestDeepxOutputValidation(unittest.TestCase): + def test_the_raw_yolox_head_is_read_for_the_configured_input(self): + for shapes, size in ( + ([(1, 8400, 85)], 640), + ([(1, 85, 8400)], 640), + # 52*52 + 26*26 + 13*13 cells at 416 + ([(1, 3549, 85)], 416), + ): + with self.subTest(shapes=shapes, size=size): + self.assertEqual(validate_yolox_outputs(shapes, 80, size, size), 85) + + def test_another_head_under_yolox_is_refused_with_the_reason(self): + cases = { + "width and height": ([(1, 8400, 85)], 80, 416), + "labelmap_path": ([(1, 8400, 85)], 91, 640), + "yolo-generic": ([(1, 8400, 80), (1, 8400, 4)], 80, 640), + } + + for reason, (shapes, num_classes, size) in cases.items(): + with ( + self.subTest(reason=reason), + self.assertRaisesRegex(ValueError, reason), + ): + validate_yolox_outputs(shapes, num_classes, size, size) + + def test_the_label_map_bounds_the_class_count(self): + self.assertEqual(class_count({0: "person", 79: "toothbrush"}), 80) + + with self.assertRaisesRegex(ValueError, "labelmap_path"): + class_count({}) + + +class TestDeepxRawDecode(unittest.TestCase): + def anchor_rows(self, rows) -> list: + out = np.zeros((1, len(rows), 85), dtype=np.float32) + + for i, (cx, cy, w, h, obj, label, score) in enumerate(rows): + out[0, i, 0:4] = [cx, cy, w, h] + out[0, i, 4] = obj + out[0, i, 5 + label] = score + + return [out] + + def test_an_anchor_based_head_becomes_normalized_corners(self): + outputs = self.anchor_rows([(320.0, 160.0, 64.0, 32.0, 0.8, 3, 0.5)]) + + detections = decode_raw_anchor(outputs, 640, 640, 0.25, 0.45) + + self.assertEqual(detections[0][0], 3) + self.assertAlmostEqual(detections[0][1], 0.4, places=5) + 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_a_channel_major_export_is_read_by_column_count(self): + outputs = self.anchor_rows([(320.0, 160.0, 64.0, 32.0, 1.0, 3, 0.9)]) + + detections = decode_raw_anchor( + [np.swapaxes(outputs[0], 1, 2)], 640, 640, 0.25, 0.45, columns=85 + ) + + self.assertEqual(detections[0][0], 3) + self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) + + def test_rows_below_the_combined_threshold_are_dropped(self): + # 0.4 * 0.5 = 0.2, under the threshold both parts clear on their own + outputs = self.anchor_rows([(320.0, 320.0, 40.0, 80.0, 0.4, 3, 0.5)]) + + self.assertTrue(np.all(decode_raw_anchor(outputs, 640, 640, 0.25, 0.45) == 0)) + + def test_at_most_twenty_detections_are_returned(self): + rows = [(20.0 + 24 * i, 320.0, 16.0, 16.0, 1.0, i % 80, 0.9) for i in range(25)] + + detections = decode_raw_anchor(self.anchor_rows(rows), 640, 640, 0.25, 0.45) + + self.assertEqual(detections.shape, (20, 6)) + self.assertEqual(int((detections[:, 1] > 0).sum()), 20) + + def test_an_nms_in_head_output_is_read_without_running_nms(self): + 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], + [300.0, 300.0, 400.0, 400.0, 0.1, 5.0], + ] + ], + dtype=np.float32, + ) + + detections = decode_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) + self.assertTrue(np.all(detections[2] == 0)) + + empty = np.zeros((1, 0, 6), dtype=np.float32) + self.assertTrue(np.all(decode_raw_nms_in_head([empty], 640, 640, 0.25) == 0)) + + +class TestDeepxConfig(unittest.TestCase): + def test_a_device_string_resolves_to_an_npu_index(self): + for configured, index in (("PCIe:1", 1), ("2", 2), ("", 0)): + with self.subTest(device=configured): + self.assertEqual(resolve_device(configured), index) + + def test_a_device_that_is_not_an_index_is_rejected(self): + """The whole string has to be an index. Reading only the tail would + take the 1 out of "PCIe:0,PCIe:1" and bind to an NPU the config never + named, and several NPUs are configured as separate devices entries.""" + for configured in ( + "PCIe:the-fast-one", + "PCIe:0,PCIe:1", + "0,1", + "PCIe:0 PCIe:1", + "PCIe:-1", + "PCIe:", + ): + with self.subTest(device=configured): + with self.assertRaises(ValueError): + resolve_device(configured) + + with self.assertRaises(ValidationError): + DeepxDetectorConfig(type="deepx", device=configured) + + def test_a_bad_device_is_refused_where_the_config_is_parsed(self): + """parse_device builds the detector config to surface a bad device at + startup, so the comma-separated form fails there rather than binding a + detector process to the wrong NPU.""" + self.assertEqual(parse_device("deepx:PCIe:1").device, "PCIe:1") + + for raw in ("deepx:PCIe:0,PCIe:1", "deepx:the-fast-one"): + with self.subTest(raw=raw), self.assertRaises(DeviceParseError): + parse_device(raw) + + def test_a_device_string_builds_this_detector_config(self): + """Frigate turns a `deepx:PCIe:0` entry into the detector config with + the model already attached, which is the path app.py takes; a bare + constructor call does not exercise it.""" + config = build_detector_config( + parse_device("deepx:PCIe:0"), model_with_type(ModelTypeEnum.yologeneric) + ) + + self.assertIsInstance(config, DeepxDetectorConfig) + self.assertEqual(config.device, "PCIe:0") + self.assertEqual(config.model.model_type, ModelTypeEnum.yologeneric) + + def test_the_runtime_manifest_pins_its_wheels_to_the_version(self): + """A PyPI path carries a per-file digest, so bumping DXRT_VERSION has + to rewrite the whole URL; a stale one installs the old wheel and fails + the sha256 on every user's first start.""" + self.assertEqual(DEEPX_MANIFEST.version, DXRT_VERSION) + + for artifact in DEEPX_MANIFEST.artifacts: + with self.subTest(url=artifact.url): + self.assertIn(f"dx_engine-{DXRT_VERSION}-", artifact.url) + + def test_a_model_less_config_still_validates(self): + config = DeepxDetectorConfig(type="deepx") + + self.assertIsNone(config.model) + + +class DeepxDetectorTestCase(unittest.TestCase): + def detector( + self, + model_type=ModelTypeEnum.yologeneric, + outputs_info=None, + ppu=False, + dynamic=False, + model_path="/nonexistent/model.dxnn", + ) -> DeepxDetector: + dx_engine = MagicMock() + dx_engine.Configuration.ITEM.SERVICE = object() + session = dx_engine.InferenceEngine.return_value + session.get_output_tensors_info.return_value = ( + [{"shape": [8400]}] if ppu else outputs_info or [] + ) + session.is_ppu.return_value = ppu + session.has_dynamic_output.return_value = dynamic + + config = DeepxDetectorConfig(type="deepx") + config.model = model_with_type(model_type) + config.model.path = model_path + + with ( + # the detector writes the endpoint into the environment, which the + # rest of the suite shares when it runs in one process + patch.dict(os.environ), + patch.dict(sys.modules, {"dx_engine": dx_engine}), + patch.object(DeepxDetector, "activate_dependencies"), + patch("os.path.isfile", return_value=True), + ): + return DeepxDetector(config) + + def ppu_detector( + self, ppu, layers, model_type=ModelTypeEnum.yologeneric, box_format=None + ) -> DeepxDetector: + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + + return self.detector( + model_type, + ppu=True, + model_path=write_dxnn(tmp.name, ppu, layers, box_format=box_format), + ) + + def detect(self, detector, outputs) -> np.ndarray: + detector.session.run.return_value = outputs + return detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + +class TestDeepxModelType(DeepxDetectorTestCase): + def test_a_model_type_with_no_decoder_is_rejected(self): + for model_type in (ModelTypeEnum.ssd, ModelTypeEnum.dfine): + with ( + self.subTest(model_type=model_type), + self.assertRaisesRegex(ValueError, model_type.value), + ): + self.detector(model_type) + + def test_supported_model_types_are_accepted(self): + for model_type, outputs_info in ( + (ModelTypeEnum.yologeneric, [{"shape": [1, 84, 8400]}]), + (ModelTypeEnum.yolox, [{"shape": [1, 8400, 85]}]), + ): + with self.subTest(model_type=model_type): + detector = self.detector(model_type, outputs_info) + + self.assertEqual(detector.model_type, model_type) + + +class TestDeepxDetectorLoad(DeepxDetectorTestCase): + def test_the_layout_is_settled_at_load(self): + detector = self.detector(ModelTypeEnum.yologeneric, [{"shape": [1, 84, 8400]}]) + self.assertIs(detector.output.layout, YoloLayout.anchor_free) + self.assertEqual(detector.output.columns, 84) + + detector = self.detector(ModelTypeEnum.yolox, [{"shape": [1, 8400, 85]}]) + self.assertIs(detector.output.layout, YoloLayout.yolox) + + for model_type in (ModelTypeEnum.yologeneric, ModelTypeEnum.yolox): + with self.subTest(model_type=model_type): + detector = self.ppu_detector( + ANCHOR_FREE_PPU, THREE_SCALE_FREE, model_type=model_type + ) + + self.assertIs(detector.output.layout, YoloLayout.ppu) + self.assertEqual(detector.ppu_layout.scale_count, 3) + + def test_a_model_the_detector_cannot_decode_is_refused_at_load(self): + cases = { + "Cannot decode DEEPX model": lambda: self.detector( + ModelTypeEnum.yologeneric, [{"shape": [1, 8400, 7]}] + ), + "DX-COM 2.4.0": lambda: self.detector(ModelTypeEnum.yologeneric, ppu=True), + "face and pose": lambda: self.ppu_detector({"type": 2}, [(80, 80, 1)]), + "centre and size or as two corners": lambda: self.ppu_detector( + ANCHOR_FREE_PPU, ONE_SCALE_FREE + ), + } + + for reason, load in cases.items(): + with ( + self.subTest(reason=reason), + self.assertRaisesRegex(ValueError, reason), + ): + load() + + +class TestDeepxDetectRaw(DeepxDetectorTestCase): + def test_a_ppu_record_decodes_by_the_head_in_the_model(self): + cases = { + "anchor-based, layer 2 of 3 at stride 32, the 373x326 anchor": ( + (ANCHOR_BASED_PPU, THREE_SCALE_ANCHORS, None), + build_ppu_record((0.6, 0.4, 0.3, 0.7), label=5), + (5, (0.0, 0.380094, 0.864187, 0.589906)), + ), + "anchor-based, layer 0 of 3 at stride 8, the 16x30 anchor": ( + (ANCHOR_BASED_PPU, THREE_SCALE_ANCHORS, None), + build_ppu_record((0.6, 0.4, 0.3, 0.7), grid=(7, 9, 1, 0)), + (0, (None, 0.116750, None, None)), + ), + "anchor-based, two scales: layer 0 is stride 16, not stride 8": ( + (ANCHOR_BASED_PPU, TWO_SCALE_ANCHORS, None), + build_ppu_record((0.6, 0.4, 0.3, 0.7), grid=(7, 9, 1, 0)), + (0, (0.141156, 0.236031, 0.223844, 0.248969)), + ), + "anchor-free, three scales: cell (10, 9) of stride 32": ( + (ANCHOR_FREE_PPU, THREE_SCALE_FREE, None), + build_ppu_record((1.2, 0.5, 1.0, 0.5), grid=(9, 10, 0, 2), label=7), + (7, (0.433782, 0.492043, 0.516218, 0.627957)), + ), + "anchor-free, one scale: the box fields are already pixels": ( + (ANCHOR_FREE_PPU, ONE_SCALE_FREE, "centre"), + build_ppu_record((320.0, 160.0, 64.0, 32.0), label=3), + (3, (144 / 640, 288 / 640, 176 / 640, 352 / 640)), + ), + "anchor-free, one scale: a sub-pixel box stays sub-pixel": ( + (ANCHOR_FREE_PPU, ONE_SCALE_FREE, "centre"), + build_ppu_record((0.6, 0.4, 0.3, 0.7)), + (0, (None, 0.45 / 640, None, None)), + ), + "a corner-format head reads the record as two corners": ( + (ANCHOR_FREE_PPU, ONE_SCALE_FREE, "corner"), + build_ppu_record((100.0, 50.0, 300.0, 250.0), label=2), + (2, (50 / 640, 100 / 640, 250 / 640, 300 / 640)), + ), + "a centre-format head reads it as a centre and size": ( + (ANCHOR_FREE_PPU, ONE_SCALE_FREE, "centre"), + build_ppu_record((100.0, 50.0, 300.0, 250.0), label=2), + (2, (0.0, 0.0, 175 / 640, 250 / 640)), + ), + } + + for head, ((ppu, layers, box_format), record, (label, box)) in cases.items(): + with self.subTest(head=head): + detector = self.ppu_detector(ppu, layers, box_format=box_format) + + detections = self.detect(detector, [record]) + + self.assertEqual(detections[0][0], label) + for i, expected in enumerate(box, start=2): + if expected is not None: + self.assertAlmostEqual(detections[0][i], expected, places=5) + + def test_the_strides_come_from_the_grids_in_the_model(self): + detector = self.ppu_detector( + ANCHOR_FREE_PPU, [(40, 40, 1), (20, 20, 1), (10, 10, 1)] + ) + + detections = self.detect( + detector, + [build_ppu_record((1.2, 0.5, 1.0, 0.5), grid=(9, 10, 0, 0), label=7)], + ) + + # layer 0 is stride 16: centre (11.2, 9.5) * 16, size e * 16 x sqrt(e) * 16 + self.assertEqual(detections[0][0], 7) + self.assertAlmostEqual(detections[0][2], (152 - np.exp(0.5) * 8) / 640, 5) + self.assertAlmostEqual(detections[0][3], (179.2 - np.exp(1.0) * 8) / 640, 5) + + def test_the_head_stays_what_the_model_said_across_frames(self): + detector = self.ppu_detector(ANCHOR_BASED_PPU, THREE_SCALE_ANCHORS) + + first = self.detect( + detector, [build_ppu_record((0.6, 0.4, 0.3, 0.7), grid=(7, 9, 1, 0))] + ) + # layer 0 of 3: stride 8, anchor 16x30 + self.assertAlmostEqual(first[0][3], 0.116750, places=5) + + second = self.detect( + detector, [build_ppu_record((0.5, 0.5, 0.4, 0.4), grid=(3, 4, 0, 1))] + ) + # layer 1 of 3: stride 16, anchor 30x61, not layer 1 of 2 + self.assertAlmostEqual(second[0][3], 0.0975, places=5) + + detector = self.ppu_detector( + ANCHOR_FREE_PPU, ONE_SCALE_FREE, box_format="centre" + ) + record = [build_ppu_record((100.0, 50.0, 300.0, 250.0), label=2)] + # centre (100, 50), size 300 x 250: the right edge lands at 250 + for frame in range(2): + with self.subTest(frame=frame): + self.assertAlmostEqual( + self.detect(detector, record)[0][5], 250 / 640, places=5 + ) + + def test_records_the_head_cannot_place_come_back_empty(self): + cases = { + "a level or box the anchor table does not carry": ( + THREE_SCALE_ANCHORS, + [build_ppu_record((0.6, 0.4, 0.3, 0.7), grid=(7, 9, 2, 5))], + ), + "a scale count Frigate has no anchor table for": ( + FOUR_SCALE_ANCHORS, + [build_ppu_record((0.6, 0.4, 0.3, 0.7))], + ), + "a record below the score threshold": ( + THREE_SCALE_ANCHORS, + [build_ppu_record((0.6, 0.4, 0.3, 0.7), score=0.1)], + ), + "an unexpected record width": ( + THREE_SCALE_ANCHORS, + [np.zeros((1, 3, 16), dtype=np.uint8)], + ), + **{ + f"no records at all, shaped {shape}": ( + THREE_SCALE_ANCHORS, + [np.zeros(shape, dtype=np.uint8)], + ) + for shape in ((1, 0, PPU_RECORD_SIZE), (0, PPU_RECORD_SIZE), (0,)) + }, + } + + for output, (layers, outputs) in cases.items(): + with self.subTest(output=output): + detector = self.ppu_detector(ANCHOR_BASED_PPU, layers) + + self.assertTrue(np.all(self.detect(detector, outputs) == 0)) + + def test_a_yolox_raw_head_is_decoded_through_the_grid(self): + detector = self.detector(ModelTypeEnum.yolox, [{"shape": [1, 8400, 85]}]) + + # cell (x 10, y 5) of the stride-8 grid is row 5 * 80 + 10 + tensor = np.zeros((1, 8400, 85), np.float32) + tensor[0, 410, :5] = [0.5, 0.5, np.log(4.0), np.log(2.0), 0.9] + tensor[0, 410, 5 + 7] = 0.8 + + detections = self.detect(detector, [tensor]) + + # centre (84, 44), size 32 x 16 + self.assertEqual(detections[0][0], 7) + self.assertAlmostEqual(detections[0][1], 0.72, places=5) + self.assertAlmostEqual(detections[0][2], 36 / 640, places=5) + self.assertAlmostEqual(detections[0][3], 68 / 640, places=5) + self.assertAlmostEqual(detections[0][4], 52 / 640, places=5) + self.assertAlmostEqual(detections[0][5], 100 / 640, places=5) + + detector = self.detector(ModelTypeEnum.yolox, [{"shape": [1, 85, 8400]}]) + detections = self.detect(detector, [np.swapaxes(tensor, 1, 2)]) + self.assertAlmostEqual(detections[0][5], 100 / 640, places=5) + + def test_a_raw_head_comes_back_as_frigates_detection_rows(self): + detector = self.detector(ModelTypeEnum.yologeneric, [{"shape": [1, 84, 8400]}]) + output = np.zeros((1, 84, 8400), dtype=np.float32) + output[0, 0:4, 0] = [320.0, 160.0, 64.0, 32.0] + output[0, 4 + 2, 0] = 0.9 + + detections = self.detect(detector, [output]) + + self.assertEqual(detections.shape, (20, 6)) + self.assertEqual(detections[0][0], 2) + self.assertAlmostEqual(detections[0][1], 0.9, places=5) + self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) diff --git a/frigate/test/test_detector_hardware.py b/frigate/test/test_detector_hardware.py index e3d0c87fd7..888ff1ff69 100644 --- a/frigate/test/test_detector_hardware.py +++ b/frigate/test/test_detector_hardware.py @@ -184,6 +184,26 @@ class TestAccelerators(HardwareProbeTestCase): ) self.assertFalse(memryx.unlimited) + def test_each_deepx_node_is_a_unit(self): + write(os.path.join(self.dev_root, "dxrt0")) + write(os.path.join(self.dev_root, "dxrt1")) + + deepx = self.probe()["deepx"] + + self.assertEqual( + [unit.device for unit in deepx.units], + ["deepx:PCIe:0", "deepx:PCIe:1"], + ) + + def test_a_deepx_npu_is_unlimited(self): + # the host daemon multiplexes, so one module takes several processes + write(os.path.join(self.dev_root, "dxrt0")) + + self.assertTrue(self.probe()["deepx"].unlimited) + + def test_no_deepx_is_reported_without_a_node(self): + self.assertNotIn("deepx", self.probe()) + def test_a_supported_rockchip_soc_is_reported(self): write( os.path.join(self.proc_root, "device-tree", "compatible"),