diff --git a/docker/deepx/user_installation.sh b/docker/deepx/user_installation.sh index 9938e4444e..ff332172f7 100755 --- a/docker/deepx/user_installation.sh +++ b/docker/deepx/user_installation.sh @@ -5,8 +5,10 @@ # 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 (v3.4.0) and NPU firmware (v2.7.4) must agree or inference -# hangs instead of failing at startup; 3.4.0 needs driver 2.5.0, firmware 2.7.0. +# 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). @@ -28,11 +30,13 @@ if ! lspci -d 1ff4: | grep -q .; then fi # fetch the pinned commit rather than cloning the tag, so a retag cannot swap -# in different source -mkdir dx_rt_npu_linux_driver +# 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 add origin https://github.com/DEEPX-AI/dx_rt_npu_linux_driver.git +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 @@ -93,8 +97,24 @@ 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 --now dxrt.service +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" diff --git a/docs/data/object_detectors_models.yaml b/docs/data/object_detectors_models.yaml index 78fe3e3d3c..3621ae0264 100644 --- a/docs/data/object_detectors_models.yaml +++ b/docs/data/object_detectors_models.yaml @@ -988,37 +988,9 @@ memryx: deepx: title: DEEPX NPU models: - - key: yolo - label: YOLO (anchor / anchor-free / NMS-in-head) - recommended: true - download: No model is bundled with Frigate. Download a pre-compiled `.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. - 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/model.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` | - - The width and height must match the resolution the `.dxnn` file was compiled for. 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/model.dxnn - labelmap_path: /labelmap/coco-80.txt - model_type: yolo-generic - width: 640 - height: 640 - key: damoyolo label: DAMO-YOLO (TinyNAS-L20T / S / M / L) - recommended: false + recommended: true download: No model is bundled with Frigate. Download a pre-compiled DAMO-YOLO `.dxnn` model (TinyNAS-L20T, S, M, or L size) from the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo), then bind-mount it into the container and point the model's `path` at it. All four sizes share one detection head, so nothing else has to be told apart. 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: @@ -1044,6 +1016,34 @@ deepx: model_type: damo-yolo width: 640 height: 640 + - key: yolo + label: YOLO + recommended: false + 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 output layout is read from the compiled model, so anchor-based, anchor-free, NMS-in-head and anchor-free PPU models need no extra configuration. PPU models compiled from an anchor-based head are not supported. + 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/model.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` | + + The width and height must match the resolution the `.dxnn` file was compiled for. 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/model.dxnn + labelmap_path: /labelmap/coco-80.txt + model_type: yolo-generic + 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 7c457eab90..7b75523f8a 100644 --- a/docs/docs/configuration/object_detectors.md +++ b/docs/docs/configuration/object_detectors.md @@ -655,64 +655,12 @@ The DX-RT Python bindings are not part of the Frigate image. They are downloaded -Frigate does not bundle a model for this detector. Models must be compiled to DEEPX's `.dxnn` format. +Frigate does not bundle a model for this detector. Models must be compiled to DEEPX's `.dxnn` format. Two model types are supported: -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 object detection models alongside a JSON file describing how each was compiled: YOLO variants (`model_type: yolo-generic`) and DAMO-YOLO (`model_type: damo-yolo`). Pick a model, download the `.dxnn`, bind-mount it into the container, and point the model's `path` at it. Note which one you picked, since it determines the `model_type` value below. `model_type` has to be set: Frigate defaults it to `ssd`, which no DEEPX decoder reads, so a model that leaves it unset is rejected at startup. +- `damo-yolo` for DAMO-YOLO models, the recommended choice. All four sizes (TinyNAS-L20T, S, M, L) are supported. +- `yolo-generic` for YOLO object detection models. 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 anchor-free models compiled with DEEPX's Post-Processing Unit (PPU) support. PPU models compiled from an anchor-based head are not supported; compile those without PPU support. -```yaml -models: - - devices: - - deepx:PCIe:0 - path: /config/model_cache/deepx/model.dxnn - labelmap_path: /labelmap/coco-80.txt - model_type: yolo-generic - width: 640 - height: 640 -``` - -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:1 -``` - -Alternatively, compile your own model with the DX-COM compiler. - -`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. - -#### Label maps - -Every supported model from the DEEPX ModelZoo is 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. - -#### Decoder options {#deepx-decoder-options} - -`model_format`, `ppu`, and `anchors` belong to the DEEPX detector rather than to a model. A `models` entry carries no per-detector options, so there is currently no way to set them and every DEEPX detector runs with their defaults, `model_format: auto` and `ppu: false`. In practice a `.dxnn` file therefore has to be one the `auto` inference can read: an anchor-free `yolo-generic` head, or DAMO-YOLO, which takes no layout at all. The sections below describe what each option selects, for when the config grows a place to set them. - -#### YOLO model formats {#deepx-yolo-model-formats} - -For `model_type: yolo-generic`, DX-COM preserves the detection head of the model it compiles, so the head's tensor layout determines how the output has to be read. `model_format` names the layout the `.dxnn` file was compiled with, across three decode families: - -| `model_format` | Output layout | PPU support | -| -------------- | -------------- | ----------- | -| `anchor` | Anchor-based, with a separate objectness score | Yes | -| `anchor_free` | Anchor-free, no objectness, boxes already in pixels | Yes | -| `nms_in_head` | NMS runs in the detection head, output is final corner boxes | No | - -The `auto` default infers the layout from the output shape. That inference cannot distinguish an anchor-based head from an anchor-free one, so it is only reliable for anchor-free heads. - -Two mistakes an explicit layout prevents: - -- Decoding an anchor-free model with the anchor formula squares the box dimensions, producing boxes hundreds of thousands of pixels wide that clip to the entire frame. -- Decoding an anchor-based model as anchor-free reads the objectness column as a class score, so confidences and class IDs are both wrong. - -`nms_in_head` models run NMS inside the head, so the detector's `nms_threshold` does not apply to them. - -#### DAMO-YOLO models - -For `model_type: damo-yolo` no layout is needed. All four sizes (TinyNAS-L20T, S, M, L) share one detection head, differing only in backbone depth/width, and the decoder reads that one layout directly: +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 and DAMO-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. ```yaml models: @@ -725,21 +673,22 @@ models: height: 640 ``` -#### PPU models +`model_type` must be set to `damo-yolo` or `yolo-generic` to match the model. Frigate defaults it to `ssd`, which this detector does not support, so a model that leaves it unset is rejected at startup. -Models compiled with Post-Processing Unit (PPU) support perform candidate selection on the NPU itself and emit a list of surviving detections rather than raw feature maps. This leaves only the box decode and NMS for the host, which lowers CPU usage. PPU support is only available for `model_type: yolo-generic` — there is no confirmed PPU record layout for DAMO-YOLO yet, and combining the two is rejected at startup. +`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. -`ppu` and `model_format` are independent: `ppu` says whether candidate selection already happened on the NPU, and `model_format` says how to read the boxes that come back. Both must match how the model was compiled. +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: -`model_format` is required whenever `ppu` is set. A PPU record has the same fixed-width layout for every yolo-generic variant, so unlike the raw output path there is no shape to infer the layout from — leaving it at `auto` is rejected at startup rather than risking a silent anchor-free/anchor-based mismatch. Since neither option can currently be set, PPU-compiled models cannot be used. +```yaml +models: + - devices: + - deepx:PCIe:0 + - deepx:PCIe:0 +``` -DX-COM cannot compile `nms_in_head` models with PPU support, since their heads already produce final detections. Combining the two is rejected at startup. +#### Label maps -##### Anchors {#deepx-ppu-anchors} - -`model_format: anchor` with `ppu` also needs `anchors`. A PPU record names its anchor by index and says nothing about the table behind that index, and anchor tables differ between models that share this output layout, so there is nothing safe to default to — the wrong table rescales every box by the wrong amount, and the boxes still look plausible enough that nothing reports an error. It is required rather than defaulted for the same reason `model_format` is. - -The table is copied out of the config the model was trained with. It is one group of `width,height` pairs per feature-map layer, groups separated by `;`, in stride order 8, 16, 32, for example `10,13,16,30,33,23;30,61,62,45,59,119;116,90,156,198,373,326`. The nested list form a model config usually declares them in is accepted too. Every group must hold the same number of pairs, since a record's anchor index is a single number that has to mean the same thing on every layer. `anchors` is rejected on any other combination, because no other decode path reads it: `anchor_free` and `nms_in_head` models carry their box geometry in the output, and the non-PPU `anchor` path receives boxes already in pixels. +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. --- diff --git a/docs/docs/frigate/hardware.md b/docs/docs/frigate/hardware.md index 1bed9babe2..90ae8fd9db 100644 --- a/docs/docs/frigate/hardware.md +++ b/docs/docs/frigate/hardware.md @@ -267,7 +267,7 @@ Frigate supports the DEEPX NPU in both of its form factors: the **DX-M1** M.2 mo DEEPX NPU support in Frigate is developed and maintained by [Sixfab](https://sixfab.com). -The NPU runs models compiled to the `.dxnn` format with DEEPX's DX-COM compiler. Pre-compiled YOLO and DAMO-YOLO models can be downloaded from the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo), and YOLO models compiled with Post-Processing Unit (PPU) support move candidate selection onto the NPU, which reduces host CPU usage. +The NPU runs models compiled to the `.dxnn` format with DEEPX's DX-COM compiler. Pre-compiled YOLO and DAMO-YOLO models can be downloaded from the [DEEPX ModelZoo](https://developer.deepx.ai/modelzoo). Anchor-free YOLO models compiled with Post-Processing Unit (PPU) support move candidate selection onto the NPU, which reduces host CPU usage; PPU models compiled from an anchor-based head are not supported. The DEEPX kernel driver, the DX-RT runtime, and the `dxrtd` daemon all run on the Docker host rather than inside the Frigate container, and have to be installed there before the NPU can be used. Frigate connects to the daemon over its socket, so the NPU stays available to other programs on the host at the same time. See the [installation docs](installation.md#deepx-npu) for the setup steps. diff --git a/docs/docs/frigate/installation.md b/docs/docs/frigate/installation.md index b522f0192d..f3adc8003e 100644 --- a/docs/docs/frigate/installation.md +++ b/docs/docs/frigate/installation.md @@ -426,10 +426,11 @@ Confirm the NPU is visible before continuing: ls /dev/dxrt* ``` -Then confirm the daemon is running: +Then confirm the daemon is running and listening in `/run/dxrt`: ```bash systemctl is-active dxrt.service +ls /run/dxrt/ ``` #### Setup @@ -438,7 +439,7 @@ To set up Frigate, follow the default installation instructions, for example: `g #### Docker configuration -Frigate needs the NPU device node and the daemon's socket: +Frigate needs the NPU device node and the directory holding the daemon's socket: ```yaml services: @@ -446,24 +447,26 @@ services: devices: - /dev/dxrt0:/dev/dxrt0 volumes: - - /tmp/dxrt_dynamic_ipc.sock:/tmp/dxrt_dynamic_ipc.sock + - /run/dxrt:/run/dxrt ``` -If you can't use Docker Compose, add `--device /dev/dxrt0:/dev/dxrt0 -v /tmp/dxrt_dynamic_ipc.sock:/tmp/dxrt_dynamic_ipc.sock` to your `docker run` command. +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` and `/etc/profile.d/dxrt.sh` for exactly that, the second so the host's own `dxrt-cli` and `dxtop` look in the same place. + :::note -Restart `dxrt.service` before starting the container, not after. A bind-mounted socket pins the inode it saw at container start, so recreating the socket afterwards leaves the container holding a deleted one. +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 daemon publishes a shared memory segment for device monitoring, which `dxtop` and device status read, but inference does not. - -`dxrtd` listens on both an abstract socket and `/tmp/dxrt_dynamic_ipc.sock`. Only the second crosses into a container, so Frigate names it through `DXRT_DYNAMIC_IPC_ENDPOINT` on your behalf. Set that variable yourself only if you moved the socket, in which case set it for the daemon too, through a systemd drop-in. +`/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`. diff --git a/frigate/detectors/plugins/deepx.py b/frigate/detectors/plugins/deepx.py index 4da5e973ff..f8cd19480e 100644 --- a/frigate/detectors/plugins/deepx.py +++ b/frigate/detectors/plugins/deepx.py @@ -1,10 +1,8 @@ """DEEPX NPU detector running compiled .dxnn models via the DX-RT runtime.""" -import atexit -import glob import logging import os -import subprocess +from dataclasses import dataclass from enum import Enum from typing import Literal @@ -21,84 +19,45 @@ logger = logging.getLogger(__name__) DETECTOR_KEY = "deepx" -# Installed at first start rather than shipped in the image. DEEPX's PyPI -# wheels are byte-identical to the ones libdxrt-bin carries, so the host -# runtime and the container bindings come from one vendor build. +# Installed at first start; DEEPX's PyPI wheels match the libdxrt-bin build. DXRT_VERSION = "3.4.0" -# Where dxrtd, running on the host, accepts client connections. The abstract -# socket it also listens on cannot be reached from a container (verified). +# Where the host dxrtd accepts clients. Its abstract socket is unreachable +# from a container, and a mounted socket file pins a stale inode after a +# daemon restart, so the directory is what gets mounted. DXRT_IPC_ENDPOINT_ENV = "DXRT_DYNAMIC_IPC_ENDPOINT" -DXRT_IPC_SOCKET = "/tmp/dxrt_dynamic_ipc.sock" +DXRT_IPC_SOCKET = "/run/dxrt/dxrt_dynamic_ipc.sock" -# DX-RT decides whether dxrtd runs by scanning /proc for a cmdline holding -# "dxrtd" (checkService, lib/device_pool/service_util.cpp). The daemon lives on -# the host, and sharing its PID namespace is closed to us because s6-overlay -# demands PID 1, so a placeholder carrying that name answers the scan while the -# real work goes to the host over DXRT_IPC_SOCKET. -_SERVICE_PLACEHOLDER: subprocess.Popen | None = None +# Pre-NMS filter, matching Frigate's shared YOLO decoder; per-object +# min_score and threshold are applied by Frigate afterwards. +SCORE_THRESHOLD = 0.4 +NMS_THRESHOLD = 0.4 + +SUPPORTED_MODEL_TYPES = (ModelTypeEnum.yologeneric, ModelTypeEnum.damoyolo) + +# 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_SCORE_BYTES = (20, 24) +PPU_LABEL_BYTES = (24, 28) + +# A nms-in-head output is (N, 6) rows of x_min, y_min, x_max, y_max, score, +# class, with N decided per frame and capped in the head (300 for the +# ModelZoo exports). A raw head has thousands of rows, which is what tells +# the two apart when a 1 or 2 class model also has 6 columns. +NMS_IN_HEAD_COLUMNS = 6 +NMS_IN_HEAD_MAX_ROWS = 1000 + +# Frigate's shared multipart decoder reads exactly three NCHW feature maps +# with 3 anchors x (5 + 80) channels each. +MULTIPART_OUTPUTS = 3 +MULTIPART_CLASSES = 80 +MULTIPART_CHANNELS = 3 * (5 + MULTIPART_CLASSES) -def service_is_visible() -> bool: - """Whether a process named dxrtd is visible in this PID namespace.""" - for path in glob.glob("/proc/[0-9]*/cmdline"): - try: - with open(path, "rb") as f: - if b"dxrtd" in f.read(): - return True - except OSError: - continue - - return False - - -def satisfy_service_check(socket_path: str) -> None: - """Make DX-RT's liveness scan succeed when the daemon runs on the host. - - A no-op when a dxrtd is visible, or when `socket_path` is absent so a - genuinely stopped daemon still reports itself. - """ - global _SERVICE_PLACEHOLDER - - if _SERVICE_PLACEHOLDER is not None and _SERVICE_PLACEHOLDER.poll() is None: - return - - if service_is_visible(): - return - - if not os.path.exists(socket_path): - logger.warning( - "No dxrtd socket at %s. Start dxrt.service on the host and mount " - "the socket into the container.", - socket_path, - ) - return - - try: - _SERVICE_PLACEHOLDER = subprocess.Popen( - ["dxrtd", "infinity"], - executable="/bin/sleep", - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - except OSError as err: - logger.error("Could not satisfy the DX-RT service check: %s", err) - return - - atexit.register(_stop_service_placeholder) - logger.debug( - "Standing in for the host dxrtd in this namespace as pid %d", - _SERVICE_PLACEHOLDER.pid, - ) - - -def _stop_service_placeholder() -> None: - if _SERVICE_PLACEHOLDER is not None and _SERVICE_PLACEHOLDER.poll() is None: - _SERVICE_PLACEHOLDER.terminate() - - -# PyPI paths carry a per-file digest, so each URL is written out in full; a -# version bump has to replace it, which the manifest tests check for. +# PyPI URLs carry a per-file digest, so a version bump rewrites them whole. DEEPX_MANIFEST = RuntimeManifest( name=DETECTOR_KEY, version=DXRT_VERSION, @@ -128,144 +87,6 @@ DEEPX_MANIFEST = RuntimeManifest( ) -class DeepxModelTypeEnum(str, Enum): - """The subset of Frigate's model types the DEEPX decoders can read. - - `model_type` sits on the root `model` block, where an omitted value - defaults to `ssd`; a model validator rejects anything not listed here so a - type with no DEEPX decoder fails at startup rather than being decoded as - yolo-generic. `ssd` is left out on purpose: DX-COM compiles no SSD head - this detector can read. - """ - - # PPU-compiled models still use yolo-generic; the `ppu` option is what - # selects the decode path, since there is no PPU-specific model type. - yologeneric = ModelTypeEnum.yologeneric.value - damoyolo = ModelTypeEnum.damoyolo.value - - -class ModelFormatEnum(str, Enum): - """How a .dxnn file's output is decoded. DX-COM preserves the source - model's head, so the layout has to be named; a model_validator keeps each - value with its own `model_type`. - """ - - # The default. yolo-generic infers the layout from the output shape and - # damo-yolo takes none. A real member rather than None so the config form - # round-trips a value -- a null is written back as a deletion. - auto = "auto" - # model_type=yolo-generic: which detection-head tensor layout the .dxnn - # was compiled with. - anchor = "anchor" - anchor_free = "anchor_free" - nms_in_head = "nms_in_head" - - -# Anchor-based heads. Raw output is (N, 5+C) with a separate objectness column, -# and the box columns hold sigmoid activations that must be rescaled against the -# grid cell and anchor that produced them. -ANCHOR_FORMATS = frozenset( - { - ModelFormatEnum.anchor, - } -) - -# Anchor-free heads. Raw output is (4+C, N) channel-major with no objectness -# column, and the box columns are already in input pixels. -ANCHOR_FREE_FORMATS = frozenset( - { - ModelFormatEnum.anchor_free, - } -) - -# Heads that run NMS on-device and emit final boxes as (N, 6) corner records. -# DX-COM cannot compile these with PPU support. -NMS_IN_HEAD_FORMATS = frozenset( - { - ModelFormatEnum.nms_in_head, - } -) - -# The feature-map strides a PPU record's layer index selects between. -PPU_STRIDES = np.array([8, 16, 32], dtype=np.float32) -PPU_ANCHOR_LAYERS = len(PPU_STRIDES) - -# Byte layout of the fixed-width detection record the PPU emits. -PPU_RECORD_SIZE = 32 -PPU_BOX_BYTES = (0, 16) -PPU_GRID_BYTES = (16, 20) -PPU_SCORE_BYTES = (20, 24) -PPU_LABEL_BYTES = (24, 28) - - -# DAMO-YOLO's ZeroHead is anchor-free at these three strides regardless of -# size (TinyNAS-L20T/S/M/L differ only in backbone depth/width). -DAMOYOLO_STRIDES = (8, 16, 32) - - -def _flatten(group): - """Yield a nested sequence one level flat, leaving a flat one alone.""" - for item in group: - if isinstance(item, (list, tuple)): - yield from item - else: - yield item - - -def parse_anchors(value: str) -> dict[int, np.ndarray]: - """Parse the anchor table an anchor-based PPU model was trained with. - - One group per layer separated by ";" in PPU_STRIDES order, each a flat run - of width,height pairs, keyed by stride the way decode_ppu_anchor reads it. - """ - groups = [group for group in value.split(";") if group.strip()] - - if len(groups) != PPU_ANCHOR_LAYERS: - strides = ", ".join(str(int(stride)) for stride in PPU_STRIDES) - raise ValueError( - f"expected {PPU_ANCHOR_LAYERS} anchor groups separated by ';', one " - f"per feature-map layer for strides {strides}, got {len(groups)}" - ) - - parsed = [] - - for group in groups: - try: - numbers = [float(part) for part in group.split(",") if part.strip()] - except ValueError: - raise ValueError( - f'anchor group "{group.strip()}" is not a comma-separated list ' - "of numbers" - ) from None - - if not numbers or len(numbers) % 2: - raise ValueError( - f'anchor group "{group.strip()}" is not an even-length list of ' - "width,height pairs" - ) - - if any(number <= 0 for number in numbers): - raise ValueError( - f'anchor group "{group.strip()}" has a width or height that is ' - "not positive" - ) - - parsed.append(np.array(numbers, dtype=np.float32).reshape(-1, 2)) - - # a record picks its anchor with a bare index and its layer separately, so - # that index only means one thing if every layer offers the same count - if len({len(anchors) for anchors in parsed}) != 1: - counts = ", ".join(str(len(anchors)) for anchors in parsed) - raise ValueError( - f"every anchor group must hold the same number of pairs, got {counts}" - ) - - return { - int(stride): anchors - for stride, anchors in zip(PPU_STRIDES, parsed, strict=True) - } - - def resolve_device(configured: str) -> 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. @@ -283,50 +104,16 @@ def resolve_device(configured: str) -> int: ) from None -def reinterpret(tensor: np.ndarray, byte_range: tuple[int, int], dtype) -> np.ndarray: - """Read a byte column range of a uint8 PPU tensor as `dtype`, copied - first because numpy will not reinterpret non-contiguous memory.""" - lo, hi = byte_range - return np.ascontiguousarray(tensor[:, lo:hi]).view(dtype) - - -def ppu_records(outputs: list[np.ndarray]) -> np.ndarray | None: - """Return the fixed-width record array a PPU model emits, or None.""" - if not outputs or outputs[0].ndim < 2: - return None - - # drop the batch dimension, leaving one uint8 record per candidate - records = outputs[0][0] - - if ( - records.ndim != 2 - or records.shape[1] != PPU_RECORD_SIZE - or records.shape[0] == 0 - ): - logger.debug("Unexpected PPU output shape %s, skipping frame", records.shape) - return None - - return records - - -def _softmax(x: np.ndarray) -> np.ndarray: - """Numerically stable softmax over the last axis.""" - shifted = x - np.max(x, axis=-1, keepdims=True) - exp = np.exp(shifted) - return exp / np.sum(exp, axis=-1, keepdims=True) - - def _split_box_and_class_outputs( outputs: list[np.ndarray], ) -> tuple[np.ndarray, np.ndarray] | None: - """Split a 2-tensor raw output into (box tensor, class tensor) by shape, - for model types where DX-COM's order is not confirmed: a box-like (..., 4) - or (..., 4, bins) wins, None when ambiguous as with 4 classes.""" + """Split a 2-tensor raw output into (box tensor, class tensor) by shape: + the (1, N, 4) tensor is the boxes, None when ambiguous as with 4 classes.""" if len(outputs) != 2: return None def is_box_shaped(t: np.ndarray) -> bool: - return (t.ndim == 3 and t.shape[-1] == 4) or (t.ndim == 4 and t.shape[-2] == 4) + return t.ndim == 3 and t.shape[-1] == 4 first, second = outputs[0], outputs[1] first_is_box, second_is_box = is_box_shaped(first), is_box_shaped(second) @@ -382,8 +169,6 @@ def run_nms( ) -> np.ndarray: """Return the indices surviving NMS, in descending score order.""" if logger.isEnabledFor(logging.DEBUG): - # Separates "the decode produced nothing" from "every score fell under - # the threshold", identical from the empty detection array alone. logger.debug( "NMS input: %d boxes, best score %.4f, %d over threshold %.2f", len(scores), @@ -403,105 +188,242 @@ def run_nms( return np.array(indices).reshape(-1) -def decode_ppu_anchor( - outputs: list[np.ndarray], - anchors: dict[int, np.ndarray], - width: int, - height: int, - score_threshold: float, - nms_threshold: float, -) -> np.ndarray: - """Decode PPU output from an anchor-based head into Frigate's (20, 6) rows. +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.""" - The NPU has already dropped sub-threshold candidates, leaving the anchor - decode and NMS. A record names its anchor by index, so `anchors` has to be - the table the model was trained with. - """ - records = ppu_records(outputs) + ppu = "ppu" # fixed-width records from the NPU's post-processing unit + 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 - 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() +@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.""" - grid = records[:, PPU_GRID_BYTES[0] : PPU_GRID_BYTES[1]] - grid_y = grid[:, 0].astype(np.float32) - grid_x = grid[:, 1].astype(np.float32) - anchor_idx = grid[:, 2] - layer_idx = grid[:, 3] + layout: YoloLayout + columns: int | None = None - # Both indices come off the wire and either could read past the tables - # below, so a malformed record skips the frame rather than crashing. - anchors_per_layer = len(next(iter(anchors.values()))) - if np.any(layer_idx >= PPU_ANCHOR_LAYERS) or np.any( - anchor_idx >= anchors_per_layer - ): - logger.debug( - "PPU record indexes layer %d / anchor %d, outside the %d layers and " - "%d anchors per layer configured, skipping frame", - layer_idx.max(), - anchor_idx.max(), - PPU_ANCHOR_LAYERS, - anchors_per_layer, +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]: + """Drop the batch and any other size-1 axis, leaving rows x columns.""" + 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. The + label map's class count tells a 4+C anchor-free head from a 5+C + anchor-based one; raises ValueError when no layout fits.""" + 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: + # a single row keeps only its column count + 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." ) - return np.zeros((20, 6), np.float32) - stride = PPU_STRIDES[layer_idx] - anchor_w = np.zeros(len(boxes), dtype=np.float32) - anchor_h = np.zeros(len(boxes), dtype=np.float32) + widths = {d for d, _ in hits} - for layer_stride, layer_anchors in anchors.items(): - mask = stride == layer_stride + 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" + ) - if np.any(mask): - anchor_w[mask] = layer_anchors[anchor_idx[mask], 0] - anchor_h[mask] = layer_anchors[anchor_idx[mask], 1] + width = widths.pop() + rows = next(d for d in dims if d != width) if dims[0] != dims[1] else dims[0] + layouts = [layout for _, layout in hits] - # Anchor-based decode: the model emits sigmoid outputs that are rescaled - # against the grid cell and its anchor to recover pixel geometry. - center_x = (boxes[:, 0] * 2.0 - 0.5 + grid_x) * stride - center_y = (boxes[:, 1] * 2.0 - 0.5 + grid_y) * stride - box_w = (boxes[:, 2] ** 2 * 4.0) * anchor_w - box_h = (boxes[:, 3] ** 2 * 4.0) * anchor_h + if len(layouts) == 1: + return YoloOutput(layouts[0], width) - x_min = center_x - box_w * 0.5 - y_min = center_y - box_h * 0.5 + # 6 columns is both NMS-in-head and a 1 or 2 class raw head; a raw head + # has thousands of candidate rows, a head with NMS has a few hundred + 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) - order = run_nms(x_min, y_min, box_w, box_h, scores, score_threshold, nms_threshold) + return YoloOutput(YoloLayout.nms_in_head, width) - return fill_detections( - x_min, - y_min, - x_min + box_w, - y_min + box_h, - scores, - labels, - width, - height, - order, + +def _infer_multipart(shapes: list[tuple[int, ...]], num_classes: int) -> YoloOutput: + """Accept only what the shared multipart decoder reads.""" + 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}). " + "If this is a DAMO-YOLO model set model_type to damo-yolo; " + "otherwise export the model with its detection head included." + ) + + return YoloOutput(YoloLayout.multipart) + + +def validate_damoyolo_outputs(shapes: list[tuple[int, ...]], num_classes: int) -> None: + """Fail unless the model emits DAMO-YOLO's (1, N, num_classes) scores + and (1, N, 4) boxes.""" + dims = [_significant_dims(shape) for shape in shapes] + box_like = [d for d in dims if len(d) == 2 and d[1] == 4] + class_like = [d for d in dims if len(d) == 2 and d[1] == num_classes] + + if ( + len(shapes) == 2 + and len(box_like) == 1 + and len(class_like) == 1 + and box_like[0][0] == class_like[0][0] + ): + return + + raise ValueError( + f"output shapes {[tuple(shape) for shape in shapes]} are not DAMO-YOLO's " + f"(1, N, {num_classes}) scores and (1, N, 4) boxes. Check that " + "labelmap_path matches the model, usually /labelmap/coco-80.txt, and " + "that model_type matches the compiled model; a YOLO model needs " + "model_type yolo-generic." ) -def decode_ppu_anchor_free( +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 + + +def ppu_boxes_are_anchor_based(boxes: np.ndarray) -> bool | None: + """An anchor-based head leaves raw sigmoids in the box fields, an + anchor-free head pixels. A value above 1 settles anchor-free, a frame + strictly inside 0..1 settles anchor-based, anything else is None.""" + if len(boxes) == 0: + return None + + if np.any(boxes > 1.0): + return False + + if np.all((boxes > 0.0) & (boxes <= 1.0)): + return True + + return None + + +def ppu_layout_is_anchor_based(outputs: list[np.ndarray]) -> bool | None: + """Whether this PPU model's records are anchor-based, or None when the + frame cannot tell; the caller keeps the first conclusive answer.""" + records = ppu_records(outputs) + + if records is None: + return None + + boxes = reinterpret(records, PPU_BOX_BYTES, np.float32).reshape(-1, 4) + + return ppu_boxes_are_anchor_based(boxes) + + +def decode_ppu( outputs: list[np.ndarray], width: int, height: int, score_threshold: float, nms_threshold: float, -) -> np.ndarray: - """Decode PPU output from an anchor-free head: no anchors and no - objectness, so the box arrives as (cx, cy, w, h) pixels. Running the - anchor formula over the unused grid columns is what yields absurd boxes. - """ + anchor_based: bool | None = None, +) -> np.ndarray | None: + """Decode anchor-free PPU records (pixel cx, cy, w, h, score, label), + leaving only NMS. None for anchor-based records, whose anchor table is + not in the model; `anchor_based` skips the heuristic once settled.""" 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) + + if anchor_based is None: + anchor_based = ppu_boxes_are_anchor_based(boxes) + + if anchor_based: + return None + + if anchor_based is None: + # nothing conclusive in this frame; no pixel box to draw either + return np.zeros((20, 6), np.float32) + scores = reinterpret(records, PPU_SCORE_BYTES, np.float32).flatten() labels = reinterpret(records, PPU_LABEL_BYTES, np.uint32).flatten() @@ -531,23 +453,17 @@ def decode_raw_anchor( height: int, score_threshold: float, nms_threshold: float, + columns: int | None = None, ) -> np.ndarray: - """Decode raw output from an anchor-based head: (N, 5+C) rows of - [cx, cy, w, h, objectness, class scores...] in pixels. Confidence is - objectness times the best class score, which is what parts this from the - anchor-free layout, where column 4 is already a class. - """ + """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. Confidence is objectness times the best class score.""" tensor = outputs[0] - if tensor.ndim < 2 or tensor.shape[-1] == 0: + if tensor.ndim < 2 or tensor.size == 0: return np.zeros((20, 6), np.float32) - # squeeze would collapse the row axis too when a single candidate is - # returned, so fold every leading axis into rows instead - predictions = tensor.reshape(-1, tensor.shape[-1]) - - if predictions.shape[0] == 0: - return np.zeros((20, 6), np.float32) + predictions = rows_with_columns(tensor, columns or tensor.shape[-1]) objectness = predictions[:, 4] class_scores = predictions[:, 5:] @@ -574,82 +490,20 @@ def decode_raw_anchor( ) -def decode_raw_anchor_free( - outputs: list[np.ndarray], - width: int, - height: int, - score_threshold: float, - nms_threshold: float, -) -> np.ndarray: - """Decode raw output from an anchor-free head, or an unconfirmed layout. - - One tensor with no objectness column, (4+C, N) or (N, 4+C), boxes as - (cx, cy, w, h) pixels. `model_format: auto` lands here too, reading every - column past the box as a class, which only holds for an anchor-free head. - """ - predictions = outputs[0] - - # Drop the batch axis without a blind squeeze, which would also collapse - # the box count on a single-candidate frame. Fold leading axes into rows. - if predictions.ndim == 3 and predictions.shape[0] == 1: - predictions = predictions[0] - - if predictions.ndim != 2: - return np.zeros((20, 6), np.float32) - - # DX-COM may export channel-major (4+C, N); transpose to (N, 4+C) so - # rows are boxes, matching every other decoder in this file. - if predictions.shape[0] < predictions.shape[1]: - predictions = predictions.T - - if predictions.shape[0] == 0 or predictions.shape[1] <= 4: - return np.zeros((20, 6), np.float32) - - class_scores = predictions[:, 4:] - labels = np.argmax(class_scores, axis=1) - scores = 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 raw output from a head that already ran NMS: (N, 6) rows of - [x_min, y_min, x_max, y_max, score, class] in pixels, sorted by score, so - only the score filter and normalization are left. - """ + """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.shape[-1] == 0: + if tensor.ndim < 2 or tensor.size == 0: return np.zeros((20, 6), np.float32) - # squeeze would collapse the row axis too when a single candidate is - # returned, so fold every leading axis into rows instead - predictions = tensor.reshape(-1, tensor.shape[-1]) - - if predictions.shape[0] == 0: - return np.zeros((20, 6), np.float32) + predictions = rows_with_columns(tensor, NMS_IN_HEAD_COLUMNS) scores = predictions[:, 4] order = np.flatnonzero(scores >= score_threshold) @@ -667,123 +521,53 @@ def decode_raw_nms_in_head( ) -def _dfl_integral(distribution: np.ndarray) -> np.ndarray: - """Reduce a DFL distribution to an expected distance via weighted sum. - From DAMO-YOLO's Integral module (Apache-2.0): each box side spans - `reg_max + 1` bins, reduced to one distance in stride units.""" - reg_max = distribution.shape[-1] - 1 - project = np.arange(reg_max + 1, dtype=np.float32) - return distribution @ project - - -def _damoyolo_center_priors( - width: int, height: int, strides: tuple[int, ...] -) -> np.ndarray: - """One (center_x, center_y, stride) row per grid cell across all scales, in - pixels. From DAMO-YOLO's ZeroHead.get_single_level_center_priors.""" - rows = [] - - for stride in strides: - grid_h, grid_w = height // stride, width // stride - row_idx, col_idx = np.mgrid[0:grid_h, 0:grid_w] - centers_x = col_idx.ravel().astype(np.float32) * stride - centers_y = row_idx.ravel().astype(np.float32) * stride - strides_col = np.full(centers_x.shape, stride, dtype=np.float32) - rows.append(np.stack([centers_x, centers_y, strides_col], axis=1)) - - return np.concatenate(rows, axis=0) - - def decode_damoyolo_raw( outputs: list[np.ndarray], width: int, height: int, score_threshold: float, nms_threshold: float, - strides: tuple[int, ...] = DAMOYOLO_STRIDES, -) -> np.ndarray: - """Decode DAMO-YOLO's raw output: sigmoid class scores + a DFL box head. +) -> np.ndarray | None: + """Decode DAMO-YOLO's output: DX-COM folds the DFL box head into the + model, so it emits (1, N, num_classes) sigmoid class scores and (1, N, 4) + boxes as x_min, y_min, x_max, y_max in input pixels, in either order. + Anchor-free with no objectness column, leaving the score filter and NMS. - From DAMO-YOLO's ZeroHead (Apache-2.0). Anchor-free, but boxes are per-side - DFL histograms, not the pixel distances decode_raw_anchor_free expects. - Takes two tensors in either order: (1, N, num_classes) sigmoid scores, and - boxes as (1, N, 4, reg_max + 1) raw DFL decoded here, or (1, N, 4) pixel - corners should DX-COM fold it in. Unverified against a compiled model. + Returns None when the outputs are not that pair at all, which is a + mismatch between the model and the config rather than an empty frame, and + stays wrong for every frame after it. A frame that simply held nothing + over the threshold comes back as the empty detection array. """ split = _split_box_and_class_outputs(outputs) if split is None: logger.debug( "Could not identify DAMO-YOLO box/class outputs in %d tensor(s) with " - "shapes %s, skipping frame", + "shapes %s", len(outputs), _shapes(outputs), ) - return np.zeros((20, 6), np.float32) + return None box_output, cls_scores = split if cls_scores.ndim != 3 or cls_scores.shape[0] == 0 or cls_scores.shape[1] == 0: - return np.zeros((20, 6), np.float32) + logger.debug("Unexpected DAMO-YOLO class output shape %s", cls_scores.shape) + return None cls_scores = cls_scores[0] + boxes = box_output[0] num_priors = cls_scores.shape[0] - if box_output.ndim == 4: - # raw per-side DFL: softmax the bins, reduce to a distance in stride - # units, then scale by each prior's stride against its own center - distances = _dfl_integral(_softmax(box_output[0])) - - if distances.shape[0] != num_priors: - logger.debug( - "DAMO-YOLO box count %d does not match class count %d, skipping frame", - distances.shape[0], - num_priors, - ) - return np.zeros((20, 6), np.float32) - - priors = _damoyolo_center_priors(width, height, strides) - - if priors.shape[0] != num_priors: - logger.debug( - "DAMO-YOLO prior count %d does not match model output %d, " - "skipping frame", - priors.shape[0], - num_priors, - ) - return np.zeros((20, 6), np.float32) - - center_x, center_y, stride = priors[:, 0], priors[:, 1], priors[:, 2] - distances_px = distances * stride[:, None] - - x_min = center_x - distances_px[:, 0] - y_min = center_y - distances_px[:, 1] - x_max = center_x + distances_px[:, 2] - y_max = center_y + distances_px[:, 3] - elif box_output.ndim == 3 and box_output.shape[-1] == 4: - boxes = box_output[0] - - if boxes.shape[0] != num_priors: - logger.debug( - "DAMO-YOLO box count %d does not match class count %d, skipping frame", - boxes.shape[0], - num_priors, - ) - return np.zeros((20, 6), np.float32) - - x_min, y_min, x_max, y_max = ( - boxes[:, 0], - boxes[:, 1], - boxes[:, 2], - boxes[:, 3], - ) - else: + if boxes.shape[0] != num_priors: logger.debug( - "Unexpected DAMO-YOLO box output shape %s, skipping frame", - box_output.shape, + "DAMO-YOLO box count %d does not match class count %d", + boxes.shape[0], + num_priors, ) - return np.zeros((20, 6), np.float32) + return None + x_min, y_min, x_max, y_max = boxes[:, 0], boxes[:, 1], boxes[:, 2], boxes[:, 3] labels = np.argmax(cls_scores, axis=1) scores = cls_scores[np.arange(num_priors), labels] @@ -814,43 +598,6 @@ class DeepxDetectorConfig(BaseDetectorConfig): description="Which NPU this detector binds to, as PCIe:. " "Empty selects the first NPU.", ) - ppu: bool = Field( - default=False, - title="PPU-compiled model", - description="Set to true when the model was compiled with DEEPX's " - "Post-Processing Unit support, which moves candidate selection onto " - "the NPU and changes the output format. Only supported for " - "model_type yolo-generic.", - ) - model_format: ModelFormatEnum = Field( - default=ModelFormatEnum.auto, - title="Detection head layout the model was compiled from", - description="Selects how the model's output is decoded. Only used " - 'for model_type yolo-generic: leave as "auto" to infer the layout ' - "from the output shape, which cannot tell an anchor-based head from " - "an anchor-free one and is only reliable for anchor-free heads. " - "Required when ppu is true, since a PPU record's layout is identical " - "for every yolo-generic variant and cannot be inferred. damo-yolo " - "reads one fixed layout and takes no value here.", - ) - anchors: str = Field( - default="", - title="Anchor sizes an anchor-based PPU model was trained with", - description="The model's own anchor table, as one group of " - "comma-separated width,height pairs per feature-map layer, groups " - 'separated by ";" in stride order 8, 16, 32. Required when ppu is ' - "true and model_format is anchor, and not used otherwise: a PPU " - "record names its anchor by index only, so the table has to come " - "from the model. Copy it from the config the model was trained with.", - ) - score_threshold: float = Field( - default=0.25, - title="Score threshold used when decoding detector output", - ) - nms_threshold: float = Field( - default=0.45, - title="NMS IoU threshold used when decoding detector output", - ) @field_validator("device") @classmethod @@ -859,170 +606,50 @@ class DeepxDetectorConfig(BaseDetectorConfig): resolve_device(value) return value - @field_validator("anchors", mode="before") - @classmethod - def coerce_anchors(cls, value): - """Accept the nested list a model's config declares anchors in, flat - row or explicit pairs, as well as the string the form writes.""" - if value is None: - return "" - - if isinstance(value, (list, tuple)): - return ";".join( - ",".join(str(number) for number in _flatten(group)) for group in value - ) - - return value - - @field_validator("anchors") - @classmethod - def validate_anchors(cls, value: str) -> str: - """Reject an anchor table that can't be read, so the error surfaces at - config load rather than on the first frame.""" - if not value.strip(): - return "" - - try: - parse_anchors(value) - except ValueError as err: - raise ValueError(f"anchors could not be read: {err}") from None - - return value - - @field_validator("model_format", mode="before") - @classmethod - def coerce_model_format(cls, value): - """A blank or absent entry means "no explicit layout".""" - if value is None or (isinstance(value, str) and not value.strip()): - return ModelFormatEnum.auto - - return value - - @property - def resolved_model_type(self) -> DeepxModelTypeEnum | None: - """The model type the decoders will actually run, read off the root - model block. None while the model is still unresolved, which is how a - device string is checked on its own before any model is attached.""" - if self.model is None: - return None - - return DeepxModelTypeEnum(self.model.model_type.value) - @model_validator(mode="after") def validate_model_type_is_supported(self): - """Reject a model type no DEEPX decoder reads. Frigate defaults - model_type to ssd, which would otherwise fall through to the - yolo-generic decode path and return nonsense instead of an error.""" + """Reject a model type with no decoder here (Frigate defaults it to + ssd). The model is None while a bare device string is validated.""" if self.model is None: return self - try: - DeepxModelTypeEnum(self.model.model_type.value) - except ValueError: - supported = ", ".join(member.value for member in DeepxModelTypeEnum) + if self.model.model_type not in SUPPORTED_MODEL_TYPES: + supported = ", ".join(t.value for t in SUPPORTED_MODEL_TYPES) raise ValueError( - f"model_type '{self.model.model_type.value}' has no DEEPX " - f"decoder. Set model.model_type to one of: {supported}." - ) from None - - return self - - @model_validator(mode="after") - def validate_ppu_is_supported(self): - """Reject a PPU pairing DX-COM cannot produce, or that has no - confirmed PPU record layout yet.""" - if not self.ppu: - return self - - if self.model_format in NMS_IN_HEAD_FORMATS: - raise ValueError( - f"model_format '{self.model_format.value}' runs NMS in the " - "detection head and cannot be compiled with PPU support by " - "DX-COM. Set ppu to false." - ) - - if self.resolved_model_type is DeepxModelTypeEnum.damoyolo: - raise ValueError( - 'PPU is not yet supported for model_type "damo-yolo". Set ppu to false.' - ) - - return self - - @model_validator(mode="after") - def validate_ppu_requires_model_format(self): - """Reject an unset model_format when ppu is enabled. Every - yolo-generic variant shares one PPU record layout, so there is no - shape to infer from, and defaulting to the anchor-based decoder - corrupts anchor-free detections rather than failing loudly.""" - if self.ppu and self.model_format is ModelFormatEnum.auto: - raise ValueError( - "model_format must be set when ppu is true. The PPU output " - "layout can't be inferred, and guessing wrong silently " - "corrupts detections." - ) - - return self - - @model_validator(mode="after") - def validate_anchors_match_the_decode_path(self): - """Require an anchor table for the one decode that reads it, and - reject one anywhere else. A record names its anchor by index only and - the table behind it differs per model, so a built-in default would - silently rescale every box. Refuse to guess.""" - reads_anchors = self.ppu and self.model_format in ANCHOR_FORMATS - - if reads_anchors and not self.anchors: - raise ValueError( - "anchors must be set when ppu is true and model_format is " - f"'{self.model_format.value}'. Copy the anchor table from the " - "config the model was trained with, one group per feature-map " - 'layer separated by ";", in stride order 8, 16, 32.' - ) - - if self.anchors and not reads_anchors: - raise ValueError( - "anchors is only read when ppu is true and model_format is " - f"'{ModelFormatEnum.anchor.value}'; every other decode path " - "takes its box geometry from the model output. Remove it." - ) - - return self - - @model_validator(mode="after") - def validate_model_format_matches_model_type(self): - """Reject a model_format on a model_type that has no layout to pick: - only yolo-generic has variants the decoder cannot tell apart, so a - value on DAMO-YOLO's one fixed layout is a mistake, not a choice.""" - model_type = self.resolved_model_type - - if model_type is None or model_type is DeepxModelTypeEnum.yologeneric: - return self - - if self.model_format is not ModelFormatEnum.auto: - raise ValueError( - "model_format is not used for model_type " - f"{model_type.value} -- its output layout is fixed. " - 'Leave it as "auto".' + f"model_type '{self.model.model_type.value}' is not supported " + f"by the DEEPX detector. Set model.model_type to one of: {supported}." ) return self class DeepxDetector(DetectionApi): + """DEEPX NPU detector: DX-RT session plus the decoder the model needs.""" + type_key = DETECTOR_KEY runtime_manifest = DEEPX_MANIFEST def __init__(self, config: DeepxDetectorConfig): self.activate_dependencies() - # DX-RT tries an abstract socket first, which never crosses a network - # namespace, and only falls back after half a second of retries. - # Naming the filesystem one up front skips that; dxrtd listens on both. - os.environ.setdefault(DXRT_IPC_ENDPOINT_ENV, DXRT_IPC_SOCKET) - satisfy_service_check(os.environ[DXRT_IPC_ENDPOINT_ENV]) + # 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, + # so name the missing socket while there is still context for it + 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 InferenceEngine, InferenceOption + from dx_engine import Configuration, InferenceEngine, InferenceOption except ModuleNotFoundError: raise ImportError( "The DX-RT python bindings are not installed. Frigate installs " @@ -1030,12 +657,14 @@ class DeepxDetector(DetectionApi): "startup log for errors." ) from None + # DX-RT checks for dxrtd by scanning /proc, which cannot see the host + # daemon from a container. SERVICE only gates that scan; the IPC + # client still uses the socket and fails loudly if dxrtd is down. + Configuration().set_enable(Configuration.ITEM.SERVICE, False) + super().__init__(config) - # model_type sits on the model, not the detector; a validator has - # already rejected any value with no DEEPX decoder behind it. - self.model_type = ModelTypeEnum(config.model.model_type.value) - + self.model_type = config.model.model_type model_path = config.model.path if not model_path or not os.path.isfile(model_path): @@ -1053,110 +682,146 @@ class DeepxDetector(DetectionApi): options.bound_option = InferenceOption.BOUND_OPTION.NPU_ALL self.session = InferenceEngine(str(model_path), options) - self.ppu = config.ppu - self.model_format = config.model_format - # validated above, and only non-empty for the decode that reads it - self.anchors = parse_anchors(config.anchors) if config.anchors else {} - self.score_threshold = config.score_threshold - self.nms_threshold = config.nms_threshold - + self.output = self.inspect_model(config) self.logged_layout = False + # settled on the first frame that carries a record, then kept + self.ppu_anchor_based: bool | None = None + self.ppu_anchor_based_reported = False + self.damoyolo_outputs_reported = False + + def inspect_model(self, config: DeepxDetectorConfig) -> YoloOutput | None: + """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.damoyolo: + if self.session.is_ppu(): + raise ValueError( + "DAMO-YOLO models compiled with PPU support are not supported" + ) + + validate_damoyolo_outputs(shapes, num_classes) + return None + + 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 for model_type %s as %s", - "PPU" if self.ppu else "raw", - self.model_type.value, - "auto-detected" - if self.model_format is ModelFormatEnum.auto - else self.model_format.value, + "DEEPX decoding yolo-generic output as %s with %d classes", + output.layout.value, + num_classes, ) + return output def decode(self, outputs: list[np.ndarray]) -> np.ndarray: - """Decode raw or PPU output according to the configured model type - and detection head layout.""" + """Decode the model output according to its type and detected layout.""" if self.model_type == ModelTypeEnum.damoyolo: - return decode_damoyolo_raw( - outputs, - self.width, - self.height, - self.score_threshold, - self.nms_threshold, - ) + return self.decode_damoyolo(outputs) - # model_type is yolo-generic from here on - if self.ppu: - # Config validation guarantees an anchor or anchor-free format - # whenever ppu is true; PPU records give no other way to tell. - if self.model_format in ANCHOR_FREE_FORMATS: - return decode_ppu_anchor_free( + match self.output.layout: + case YoloLayout.ppu: + return self.decode_ppu(outputs) + case YoloLayout.anchor: + return decode_raw_anchor( outputs, self.width, self.height, - self.score_threshold, - self.nms_threshold, + SCORE_THRESHOLD, + NMS_THRESHOLD, + self.output.columns, + ) + case YoloLayout.anchor_free: + # the shared decoder reads (N, 4+C) with the same thresholds; + # hand it rows so its own orientation guess is never needed + 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 _: + # per-scale feature maps, same thresholds as above + return post_process_yolo(outputs, self.width, self.height) + + def decode_ppu(self, outputs: list[np.ndarray]) -> np.ndarray: + """Decode PPU records, settling the head on the first conclusive + frame.""" + if self.ppu_anchor_based is None: + self.ppu_anchor_based = ppu_layout_is_anchor_based(outputs) + + if self.ppu_anchor_based: + if not self.ppu_anchor_based_reported: + self.ppu_anchor_based_reported = True + logger.error( + "This PPU model was compiled from an anchor-based head, whose " + "records cannot be decoded without the model's anchor table. " + "Compile it without PPU support, or use an anchor-free model" ) - return decode_ppu_anchor( - outputs, - self.anchors, - self.width, - self.height, - self.score_threshold, - self.nms_threshold, - ) + return np.zeros((20, 6), np.float32) - if self.model_format in NMS_IN_HEAD_FORMATS: - return decode_raw_nms_in_head( - outputs, self.width, self.height, self.score_threshold - ) - - # Only single-output anchor heads need the dedicated decoder; a - # multi-part output is feature maps the shared helper already reads. - if self.model_format in ANCHOR_FORMATS and len(outputs) == 1: - return decode_raw_anchor( - outputs, - self.width, - self.height, - self.score_threshold, - self.nms_threshold, - ) - - # `auto` and an explicit `anchor_free` both read one tensor with no - # objectness column, and the dedicated decoder keeps them honoring - # this detector's own score_threshold and nms_threshold - if len(outputs) == 1 and ( - self.model_format in ANCHOR_FREE_FORMATS - or self.model_format is ModelFormatEnum.auto - ): - return decode_raw_anchor_free( - outputs, - self.width, - self.height, - self.score_threshold, - self.nms_threshold, - ) - - return post_process_yolo( + # None here means no frame has carried a record yet, which leaves the + # heuristic to run again on the next one + return decode_ppu( outputs, self.width, self.height, - self.score_threshold, - self.nms_threshold, + SCORE_THRESHOLD, + NMS_THRESHOLD, + anchor_based=self.ppu_anchor_based, ) + def decode_damoyolo(self, outputs: list[np.ndarray]) -> np.ndarray: + """Decode DAMO-YOLO output, reporting once if a frame lacks the + expected tensor pair.""" + detections = decode_damoyolo_raw( + outputs, self.width, self.height, SCORE_THRESHOLD, NMS_THRESHOLD + ) + + if detections is None: + if not self.damoyolo_outputs_reported: + self.damoyolo_outputs_reported = True + logger.error( + "This model does not emit the pair of tensors the DAMO-YOLO " + "decoder reads, one (1, N, 4) of boxes and one (1, N, " + "num_classes) of scores, but %d tensor(s) with shapes %s. " + "Check that model_type matches the compiled model; a YOLO " + "model needs model_type yolo-generic", + len(outputs), + _shapes(outputs), + ) + + return np.zeros((20, 6), np.float32) + + return detections + def detect_raw(self, tensor_input): - """Run inference and decode the DEEPX model output.""" + """Run inference and decode the model output.""" outputs = self.session.run([tensor_input]) if not isinstance(outputs, list): outputs = [outputs] - detections = self.decode(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", @@ -1167,11 +832,8 @@ class DeepxDetector(DetectionApi): return detections def log_layout(self, tensor_input, outputs: list[np.ndarray]) -> None: - """Log the model's real input and output layout, once per start. - - A wrong guess at the compiled layout returns an empty detection array - rather than failing, and these shapes are what tell those apart. - """ + """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)] diff --git a/frigate/test/test_deepx.py b/frigate/test/test_deepx.py index 0a870e2ef9..8e830bb19a 100644 --- a/frigate/test/test_deepx.py +++ b/frigate/test/test_deepx.py @@ -1,46 +1,35 @@ -"""Tests for the DEEPX detector's per-format output decoding.""" +"""Tests for the DEEPX detector.""" import os import sys import unittest -from unittest.mock import patch +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.plugins.deepx import ( - ANCHOR_FORMATS, - ANCHOR_FREE_FORMATS, DEEPX_MANIFEST, DXRT_IPC_ENDPOINT_ENV, DXRT_IPC_SOCKET, DXRT_VERSION, - NMS_IN_HEAD_FORMATS, - PPU_ANCHOR_LAYERS, PPU_RECORD_SIZE, DeepxDetector, DeepxDetectorConfig, - DeepxModelTypeEnum, - ModelFormatEnum, + YoloLayout, + class_count, decode_damoyolo_raw, - decode_ppu_anchor, - decode_ppu_anchor_free, + decode_ppu, decode_raw_anchor, - decode_raw_anchor_free, decode_raw_nms_in_head, - parse_anchors, + infer_yolo_layout, resolve_device, - satisfy_service_check, - service_is_visible, + rows_with_columns, + validate_damoyolo_outputs, ) from frigate.util.runtime_deps import ArtifactKind -# Deliberately not any published model's anchor table: a decoder that fell back -# to a built-in set instead of reading the configured one would not match these. -TEST_ANCHOR_STRING = "11,17,23,29,31,37;41,47,53,59,61,67;71,73,79,83,89,97" -TEST_ANCHORS = parse_anchors(TEST_ANCHOR_STRING) - def model_with_type(model_type) -> ModelConfig: """The root model block a device is validated against. model_type lives @@ -50,367 +39,321 @@ def model_with_type(model_type) -> ModelConfig: return ModelConfig(model_type=model_type, labelmap_path=None) -def build_ppu_record( - box: tuple[float, float, float, float], - grid_y: int, - grid_x: int, - anchor_idx: int, - layer_idx: int, - score: float, - label: int, -) -> np.ndarray: - """Build a single fixed-width detection record as the PPU emits it.""" +def build_ppu_record(box, score=0.9, label=0, grid=(7, 9, 2, 2)) -> np.ndarray: + """One fixed-width record as the PPU emits it, in a (1, 1, 32) output.""" record = np.zeros(PPU_RECORD_SIZE, dtype=np.uint8) record[0:16] = np.array(box, dtype=np.float32).view(np.uint8) - record[16:20] = [grid_y, grid_x, anchor_idx, layer_idx] + record[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 + return record.reshape(1, 1, PPU_RECORD_SIZE) class TestDeepxPpuDecode(unittest.TestCase): - def test_decodes_a_single_detection(self): - # stride 8 (layer 0) with the first anchor of that layer - outputs = [ - build_ppu_record( - box=(0.5, 0.5, 0.5, 0.5), - grid_y=10, - grid_x=10, - anchor_idx=0, - layer_idx=0, - score=0.9, - label=2, - ).reshape(1, 1, PPU_RECORD_SIZE) - ] - - detections = decode_ppu_anchor(outputs, TEST_ANCHORS, 640, 640, 0.25, 0.45) - - anchor_w, anchor_h = TEST_ANCHORS[8][0] - # center = (0.5 * 2 - 0.5 + 10) * 8, size = (0.5 ** 2 * 4) * anchor - center = (0.5 * 2.0 - 0.5 + 10) * 8 - box_w = 0.5**2 * 4.0 * anchor_w - box_h = 0.5**2 * 4.0 * anchor_h - - self.assertEqual(detections[0][0], 2) - self.assertAlmostEqual(detections[0][1], 0.9, places=5) - self.assertAlmostEqual(detections[0][2], (center - box_h / 2) / 640, places=5) - self.assertAlmostEqual(detections[0][3], (center - box_w / 2) / 640, places=5) - self.assertAlmostEqual(detections[0][4], (center + box_h / 2) / 640, places=5) - self.assertAlmostEqual(detections[0][5], (center + box_w / 2) / 640, places=5) - - # remaining slots stay empty - self.assertTrue(np.all(detections[1:] == 0)) - - def test_uses_the_anchor_for_the_records_layer(self): - # the same box on stride 32 (layer 2) must decode to a larger box - def decode(layer_idx: int) -> np.ndarray: - outputs = [ - build_ppu_record( - box=(0.5, 0.5, 0.5, 0.5), - grid_y=1, - grid_x=1, - anchor_idx=0, - layer_idx=layer_idx, - score=0.9, - label=0, - ).reshape(1, 1, PPU_RECORD_SIZE) - ] - return decode_ppu_anchor(outputs, TEST_ANCHORS, 640, 640, 0.25, 0.45) - - small = decode(0) - large = decode(2) - - small_width = small[0][5] - small[0][3] - large_width = large[0][5] - large[0][3] - - self.assertGreater(large_width, small_width) - - def test_drops_detections_below_the_score_threshold(self): - outputs = [ - build_ppu_record( - box=(0.5, 0.5, 0.5, 0.5), - grid_y=10, - grid_x=10, - anchor_idx=0, - layer_idx=0, - score=0.1, - label=2, - ).reshape(1, 1, PPU_RECORD_SIZE) - ] - - detections = decode_ppu_anchor(outputs, TEST_ANCHORS, 640, 640, 0.25, 0.45) - - self.assertTrue(np.all(detections == 0)) - - def test_caps_output_at_twenty_detections(self): - records = np.stack( - [ - build_ppu_record( - box=(0.5, 0.5, 0.5, 0.5), - grid_y=i, - grid_x=i, - anchor_idx=0, - layer_idx=0, - score=0.9, - label=1, - ) - for i in range(40) - ] - ) - - detections = decode_ppu_anchor( - [records.reshape(1, 40, PPU_RECORD_SIZE)], - TEST_ANCHORS, + def test_reads_anchor_free_box_bytes_as_pixel_geometry(self): + """The grid columns carry no meaning for an anchor-free head and must + not influence the result.""" + detections = decode_ppu( + [build_ppu_record((320.0, 160.0, 64.0, 32.0), label=3)], 640, 640, 0.25, 0.45, ) - self.assertEqual(detections.shape, (20, 6)) + self.assertEqual(detections[0][0], 3) + self.assertAlmostEqual(detections[0][1], 0.9, places=5) + # x: 320 +/- 32 -> 288..352, y: 160 +/- 16 -> 144..176 + self.assertAlmostEqual(detections[0][2], 144 / 640, places=5) + self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) + self.assertAlmostEqual(detections[0][4], 176 / 640, places=5) + self.assertAlmostEqual(detections[0][5], 352 / 640, places=5) - def test_returns_empty_detections_for_no_output(self): - self.assertTrue( - np.all(decode_ppu_anchor([], TEST_ANCHORS, 640, 640, 0.25, 0.45) == 0) + def test_anchor_based_records_are_reported_not_misdecoded(self): + """An anchor-based head leaves raw sigmoids in the box fields; read + as pixels they would give a sub-pixel box in the top-left corner.""" + self.assertIsNone( + decode_ppu([build_ppu_record((0.6, 0.4, 0.3, 0.7))], 640, 640, 0.25, 0.45) ) + def test_drops_records_below_the_score_threshold(self): + detections = decode_ppu( + [build_ppu_record((320.0, 160.0, 64.0, 32.0), score=0.1)], + 640, + 640, + 0.25, + 0.45, + ) + + self.assertTrue(np.all(detections == 0)) + + def test_returns_empty_detections_for_no_records(self): + # a frame with no surviving candidate comes back with an empty + # leading axis, as seen on hardware + for shape in ((1, 0, PPU_RECORD_SIZE), (0, PPU_RECORD_SIZE), (0,)): + with self.subTest(shape=shape): + out = np.zeros(shape, dtype=np.uint8) + + self.assertTrue(np.all(decode_ppu([out], 640, 640, 0.25, 0.45) == 0)) + def test_returns_empty_detections_for_an_unexpected_record_width(self): - # a non-PPU model would produce a different record width - outputs = [np.zeros((1, 4, 16), dtype=np.uint8)] + out = np.zeros((1, 3, 16), dtype=np.uint8) - self.assertTrue( - np.all(decode_ppu_anchor(outputs, TEST_ANCHORS, 640, 640, 0.25, 0.45) == 0) + self.assertTrue(np.all(decode_ppu([out], 640, 640, 0.25, 0.45) == 0)) + + +def layout_of(shapes, num_classes, ppu=False, dynamic_output=False) -> YoloLayout: + return infer_yolo_layout(shapes, num_classes, ppu, dynamic_output).layout + + +class TestDeepxLayoutInference(unittest.TestCase): + """The output layout is read off the model rather than configured, so + every head DX-COM can compile has to be told apart from the others.""" + + def test_ppu_wins_over_every_shape(self): + self.assertIs(layout_of([(8400,)], 80, ppu=True), YoloLayout.ppu) + + def test_an_anchor_free_head_has_four_columns_ahead_of_the_classes(self): + for shape in ((1, 84, 8400), (1, 8400, 84)): + with self.subTest(shape=shape): + output = infer_yolo_layout([shape], 80, False, False) + + self.assertIs(output.layout, YoloLayout.anchor_free) + self.assertEqual(output.columns, 84) + + def test_an_anchor_based_head_has_an_objectness_column(self): + for shape in ((1, 25200, 85), (1, 85, 25200)): + with self.subTest(shape=shape): + output = infer_yolo_layout([shape], 80, False, False) + + self.assertIs(output.layout, YoloLayout.anchor) + self.assertEqual(output.columns, 85) + + def test_a_six_column_or_dynamic_output_ran_nms_in_the_head(self): + self.assertIs(layout_of([(1, 300, 6)], 80), YoloLayout.nms_in_head) + self.assertIs( + layout_of([(1, -1, 6)], 80, dynamic_output=True), YoloLayout.nms_in_head ) + def test_the_label_map_settles_anchor_against_anchor_free(self): + # 85 columns is anchor-free with 81 classes and anchor-based with 80; + # only the label map can tell + shape = [(1, 8400, 85)] + self.assertIs(layout_of(shape, 81), YoloLayout.anchor_free) + self.assertIs(layout_of(shape, 80), YoloLayout.anchor) -class TestDeepxAnchors(unittest.TestCase): - """The anchor table comes from the model, not from this file. A PPU record - names its anchor by index alone, and that index means different box sizes - for different models, so there is nothing safe to default to.""" + def test_six_columns_on_a_small_model_is_told_apart_by_row_count(self): + """One class puts an anchor-based head at 6 columns and two classes + put an anchor-free one there, both the width of an nms-in-head + output. A raw head has thousands of rows, NMS in the head a few + hundred.""" + self.assertIs(layout_of([(1, 25200, 6)], 1), YoloLayout.anchor) + self.assertIs(layout_of([(1, 8400, 6)], 2), YoloLayout.anchor_free) + self.assertIs(layout_of([(1, 300, 6)], 1), YoloLayout.nms_in_head) + self.assertIs(layout_of([(1, 300, 6)], 2), YoloLayout.nms_in_head) - def test_parses_a_group_per_stride_keyed_by_stride(self): - table = parse_anchors("11,17,23,29;41,47,53,59;71,73,79,83") + def test_a_shape_matching_two_layouts_on_different_axes_is_refused(self): + # 84 rows by 6 columns fits anchor-free transposed and nms-in-head + with self.assertRaisesRegex(ValueError, "fits two layouts"): + layout_of([(1, 84, 6)], 80) - self.assertEqual(sorted(table), [8, 16, 32]) - np.testing.assert_array_equal(table[8], [[11, 17], [23, 29]]) - np.testing.assert_array_equal(table[32], [[71, 73], [79, 83]]) + def test_a_small_label_map_still_reads_an_unambiguous_shape(self): + # 2 classes: 7 columns is only anchor-based + self.assertIs(layout_of([(1, 8400, 7)], 2), YoloLayout.anchor) - def test_accepts_the_nested_list_a_model_config_declares(self): - config = DeepxDetectorConfig( - type="deepx", - ppu=True, - model_format=ModelFormatEnum.anchor, - anchors=[[11, 17, 23, 29], [41, 47, 53, 59], [71, 73, 79, 83]], + def test_a_mismatched_label_map_is_reported(self): + # the image's default 91-class label map against an 80-class model + with self.assertRaisesRegex(ValueError, "labelmap_path"): + layout_of([(1, 8400, 84)], 91) + + def test_a_missing_output_is_reported(self): + with self.assertRaisesRegex(ValueError, "no output tensor"): + layout_of([], 80) + + +class TestDeepxMultipartInference(unittest.TestCase): + """The shared multipart decoder reads one exact layout, so anything else + with several outputs has to fail at load rather than on the first frame.""" + + def test_three_nchw_maps_with_255_channels_are_feature_maps(self): + shapes = [(1, 255, 80, 80), (1, 255, 40, 40), (1, 255, 20, 20)] + + self.assertIs(layout_of(shapes, 80), YoloLayout.multipart) + + def test_nhwc_maps_are_refused(self): + shapes = [(1, 80, 80, 255), (1, 40, 40, 255), (1, 20, 20, 255)] + + with self.assertRaisesRegex(ValueError, "255 channels"): + layout_of(shapes, 80) + + def test_a_damoyolo_pair_under_yolo_generic_is_pointed_at_damo_yolo(self): + with self.assertRaisesRegex(ValueError, "damo-yolo"): + layout_of([(1, 8400, 80), (1, 8400, 4)], 80) + + def test_a_non_coco_class_count_is_refused(self): + shapes = [(1, 24, 80, 80), (1, 24, 40, 40), (1, 24, 20, 20)] + + with self.assertRaisesRegex(ValueError, "80-class"): + layout_of(shapes, 3) + + +class TestDeepxDamoyoloValidation(unittest.TestCase): + def test_the_modelzoo_pair_passes_in_either_order(self): + validate_damoyolo_outputs([(1, 8400, 80), (1, 8400, 4)], 80) + validate_damoyolo_outputs([(1, 8400, 4), (1, 8400, 80)], 80) + + def test_a_wrong_class_count_names_the_label_map(self): + with self.assertRaisesRegex(ValueError, "labelmap_path"): + validate_damoyolo_outputs([(1, 8400, 80), (1, 8400, 4)], 91) + + def test_a_single_yolo_tensor_names_yolo_generic(self): + with self.assertRaisesRegex(ValueError, "yolo-generic"): + validate_damoyolo_outputs([(1, 84, 8400)], 80) + + def test_mismatched_row_counts_are_refused(self): + with self.assertRaises(ValueError): + validate_damoyolo_outputs([(1, 8400, 80), (1, 8000, 4)], 80) + + +class TestDeepxClassCount(unittest.TestCase): + def test_the_highest_id_bounds_the_count(self): + self.assertEqual(class_count({0: "person", 79: "toothbrush"}), 80) + + def test_an_empty_label_map_is_reported(self): + with self.assertRaisesRegex(ValueError, "labelmap_path"): + class_count({}) + + +class TestDeepxRowOrientation(unittest.TestCase): + def test_a_channel_major_tensor_is_transposed_by_its_column_count(self): + tensor = np.arange(2 * 85).reshape(1, 85, 2).astype(np.float32) + + rows = rows_with_columns(tensor, 85) + + self.assertEqual(rows.shape, (2, 85)) + np.testing.assert_array_equal(rows[1], tensor[0, :, 1]) + + def test_a_row_major_tensor_is_left_alone(self): + tensor = np.zeros((1, 300, 6), np.float32) + + self.assertEqual(rows_with_columns(tensor, 6).shape, (300, 6)) + + def test_an_empty_dynamic_output_keeps_its_columns(self): + self.assertEqual(rows_with_columns(np.zeros((1, 0, 6)), 6).shape, (0, 6)) + + +class TestDeepxRawAnchorDecode(unittest.TestCase): + def test_a_channel_major_export_is_read_by_column_count(self): + """A (1, 85, N) export decodes the same as (1, N, 85) once the + decoder is told the row width, instead of reading N-wide rows.""" + rows = np.zeros((2, 85), np.float32) + rows[0, :5] = [320.0, 160.0, 64.0, 32.0, 1.0] + rows[0, 5 + 3] = 0.9 + detections = decode_raw_anchor( + [rows.T[np.newaxis]], 640, 640, 0.25, 0.45, columns=85 ) - np.testing.assert_array_equal( - parse_anchors(config.anchors)[8], [[11, 17], [23, 29]] + self.assertEqual(detections[0][0], 3) + self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) + + def test_at_most_twenty_detections_are_returned(self): + # 25 well separated confident boxes, all of which survive NMS + rows = np.zeros((25, 85), np.float32) + for i in range(25): + rows[i, :5] = [20.0 + 24 * i, 320.0, 16.0, 16.0, 1.0] + rows[i, 5 + (i % 80)] = 0.9 + + detections = decode_raw_anchor([rows[np.newaxis]], 640, 640, 0.25, 0.45) + + self.assertEqual(detections.shape, (20, 6)) + self.assertEqual(int((detections[:, 1] > 0).sum()), 20) + + def build_raw_anchor_output(self, rows): + """Build an (1, N, 5+C) tensor with 80 classes.""" + out = np.zeros((1, len(rows), 85), dtype=np.float32) + + for i, (cx, cy, w, h, obj, label, cls_score) in enumerate(rows): + out[0, i, 0:4] = [cx, cy, w, h] + out[0, i, 4] = obj + out[0, i, 5 + label] = cls_score + + return [out] + + def test_confidence_is_objectness_times_class_score(self): + outputs = self.build_raw_anchor_output( + [(320.0, 320.0, 40.0, 80.0, 0.8, 3, 0.5)] ) - def test_accepts_explicit_pairs(self): - config = DeepxDetectorConfig( - type="deepx", - ppu=True, - model_format=ModelFormatEnum.anchor, - anchors=[[[11, 17], [23, 29]], [[41, 47], [53, 59]], [[71, 73], [79, 83]]], + 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) + + def test_drops_rows_whose_combined_score_is_below_threshold(self): + # 0.4 * 0.5 = 0.2, under a 0.25 threshold even though both parts are + # individually above it + outputs = self.build_raw_anchor_output( + [(320.0, 320.0, 40.0, 80.0, 0.4, 3, 0.5)] ) - np.testing.assert_array_equal( - parse_anchors(config.anchors)[8], [[11, 17], [23, 29]] + detections = decode_raw_anchor(outputs, 640, 640, 0.25, 0.45) + + self.assertTrue(np.all(detections == 0)) + + def test_converts_center_form_to_normalized_corners(self): + outputs = self.build_raw_anchor_output( + [(320.0, 160.0, 64.0, 32.0, 1.0, 0, 1.0)] ) - def test_anchor_ppu_without_anchors_is_rejected(self): - """Guessing the table silently rescales every box, so refuse to.""" - with self.assertRaises(ValidationError): - DeepxDetectorConfig( - type="deepx", ppu=True, model_format=ModelFormatEnum.anchor - ) + detections = decode_raw_anchor(outputs, 640, 640, 0.25, 0.45) - def test_anchors_on_a_decode_that_ignores_them_is_rejected(self): - for kwargs in ( - dict(ppu=True, model_format=ModelFormatEnum.anchor_free), - dict(model_format=ModelFormatEnum.anchor), - dict(), - ): - with self.subTest(**kwargs): - with self.assertRaises(ValidationError): - DeepxDetectorConfig( - type="deepx", anchors=TEST_ANCHOR_STRING, **kwargs - ) + 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_group_per_stride_is_required(self): - with self.assertRaises(ValidationError): - DeepxDetectorConfig( - type="deepx", - ppu=True, - model_format=ModelFormatEnum.anchor, - anchors="11,17;41,47", - ) - def test_groups_must_hold_equal_length_pairs(self): - for bad in ("11,17,23;41,47,53;71,73,79", "11,17;41,47,53,59;71,73"): - with self.subTest(anchors=bad): - with self.assertRaises(ValidationError): - DeepxDetectorConfig( - type="deepx", - ppu=True, - model_format=ModelFormatEnum.anchor, - anchors=bad, - ) - - def test_decode_uses_the_supplied_table(self): - """Two tables over one record must give two different box sizes.""" - outputs = [ - build_ppu_record( - box=(0.5, 0.5, 0.5, 0.5), - grid_y=10, - grid_x=10, - anchor_idx=0, - layer_idx=0, - score=0.9, - label=2, - ).reshape(1, 1, PPU_RECORD_SIZE) - ] - - narrow = decode_ppu_anchor( - outputs, parse_anchors("4,4;8,8;16,16"), 640, 640, 0.25, 0.45 - ) - wide = decode_ppu_anchor( - outputs, parse_anchors("40,40;80,80;160,160"), 640, 640, 0.25, 0.45 - ) - - self.assertGreater(wide[0][5] - wide[0][3], narrow[0][5] - narrow[0][3]) - - def test_an_out_of_range_index_skips_the_frame(self): - """Both indices come off the wire, and reading past the table would - take the detection process down rather than drop one frame.""" - for anchor_idx, layer_idx in ((9, 0), (0, PPU_ANCHOR_LAYERS)): - with self.subTest(anchor_idx=anchor_idx, layer_idx=layer_idx): - outputs = [ - build_ppu_record( - box=(0.5, 0.5, 0.5, 0.5), - grid_y=10, - grid_x=10, - anchor_idx=anchor_idx, - layer_idx=layer_idx, - score=0.9, - label=2, - ).reshape(1, 1, PPU_RECORD_SIZE) +class TestDeepxRawNmsInHeadDecode(unittest.TestCase): + def test_reads_corner_records_without_running_nms(self): + """Two heavily overlapping boxes both survive: the head already ran NMS.""" + out = np.array( + [ + [ + [100.0, 100.0, 200.0, 200.0, 0.9, 2.0], + [102.0, 102.0, 202.0, 202.0, 0.8, 2.0], ] - - detections = decode_ppu_anchor( - outputs, TEST_ANCHORS, 640, 640, 0.25, 0.45 - ) - - self.assertTrue(np.all(detections == 0)) - - -class TestDeepxModelFormats(unittest.TestCase): - def test_every_yolo_format_belongs_to_exactly_one_family(self): - families = [ANCHOR_FORMATS, ANCHOR_FREE_FORMATS, NMS_IN_HEAD_FORMATS] - yolo_formats = ANCHOR_FORMATS | ANCHOR_FREE_FORMATS | NMS_IN_HEAD_FORMATS - - for fmt in ModelFormatEnum: - if fmt not in yolo_formats: - continue - - matches = [family for family in families if fmt in family] - self.assertEqual( - len(matches), 1, f"{fmt.value} must be in exactly one family" - ) - - def test_families_do_not_overlap(self): - self.assertFalse(ANCHOR_FORMATS & ANCHOR_FREE_FORMATS) - self.assertFalse(ANCHOR_FORMATS & NMS_IN_HEAD_FORMATS) - self.assertFalse(ANCHOR_FREE_FORMATS & NMS_IN_HEAD_FORMATS) - - def test_every_format_belongs_to_a_yolo_family(self): - # apart from the `auto` sentinel, every ModelFormatEnum member is a - # yolo-generic head shape -- nothing should be unclassified - yolo_formats = ANCHOR_FORMATS | ANCHOR_FREE_FORMATS | NMS_IN_HEAD_FORMATS - - for fmt in ModelFormatEnum: - if fmt is ModelFormatEnum.auto: - continue - - self.assertIn(fmt, yolo_formats, f"{fmt.value} is not classified") - - def test_auto_is_in_no_family(self): - """`auto` means "no explicit layout", so every decode branch keyed off - a family must fall through to shape-based inference.""" - self.assertNotIn(ModelFormatEnum.auto, ANCHOR_FORMATS) - self.assertNotIn(ModelFormatEnum.auto, ANCHOR_FREE_FORMATS) - self.assertNotIn(ModelFormatEnum.auto, NMS_IN_HEAD_FORMATS) - - -class TestDeepxDetectorConfig(unittest.TestCase): - def test_ppu_without_model_format_is_rejected(self): - """PPU records are fixed-width regardless of variant, so an unset - model_format can't be inferred and must not silently fall back to the - anchor decoder.""" - with self.assertRaises(ValidationError): - DeepxDetectorConfig(type="deepx", ppu=True) - - def test_ppu_with_anchor_free_model_format_is_accepted(self): - config = DeepxDetectorConfig( - type="deepx", ppu=True, model_format=ModelFormatEnum.anchor_free + ], + dtype=np.float32, ) - self.assertEqual(config.model_format, ModelFormatEnum.anchor_free) + detections = decode_raw_nms_in_head([out], 640, 640, 0.25) - def test_raw_without_model_format_is_still_accepted(self): - # the raw path falls back to shape-based inference, so this remains - # optional when ppu is not set - config = DeepxDetectorConfig(type="deepx") + 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.assertIs(config.model_format, ModelFormatEnum.auto) - - def test_an_omitted_model_format_folds_to_auto(self): - """The config form round-trips a value for every field, so a blank or - null entry has to land on the sentinel rather than stay None.""" - for value in (None, "", " "): - with self.subTest(value=value): - config = DeepxDetectorConfig(type="deepx", model_format=value) - - self.assertIs(config.model_format, ModelFormatEnum.auto) - - def test_ppu_with_damoyolo_model_type_is_rejected(self): - """No confirmed PPU record layout exists for DAMO-YOLO yet. The type - comes off the model block, so the check has to read it from there - rather than from a detector field the runtime never uses.""" - with self.assertRaises(ValidationError): - DeepxDetectorConfig( - type="deepx", - model=model_with_type(ModelTypeEnum.damoyolo), - ppu=True, - model_format=ModelFormatEnum.anchor_free, - ) - - def test_ppu_with_yolo_generic_model_type_is_accepted(self): - config = DeepxDetectorConfig( - type="deepx", - model=model_with_type(ModelTypeEnum.yologeneric), - ppu=True, - model_format=ModelFormatEnum.anchor_free, + def test_applies_the_score_threshold(self): + out = np.array( + [ + [ + [100.0, 100.0, 200.0, 200.0, 0.9, 2.0], + [300.0, 300.0, 400.0, 400.0, 0.1, 5.0], + ] + ], + dtype=np.float32, ) - self.assertTrue(config.ppu) + detections = decode_raw_nms_in_head([out], 640, 640, 0.25) - def test_damoyolo_model_type_rejects_a_model_format(self): - with self.assertRaises(ValidationError): - DeepxDetectorConfig( - type="deepx", - model=model_with_type(ModelTypeEnum.damoyolo), - model_format=ModelFormatEnum.anchor_free, - ) + self.assertAlmostEqual(detections[0][1], 0.9, places=5) + self.assertTrue(np.all(detections[1] == 0)) - def test_damoyolo_model_type_without_model_format_is_accepted(self): - config = DeepxDetectorConfig( - type="deepx", - model=model_with_type(ModelTypeEnum.damoyolo), - ) + def test_returns_empty_detections_for_an_empty_output(self): + out = np.zeros((1, 0, 6), dtype=np.float32) - self.assertIs(config.model_format, ModelFormatEnum.auto) + self.assertTrue(np.all(decode_raw_nms_in_head([out], 640, 640, 0.25) == 0)) class TestDeepxDeviceSelection(unittest.TestCase): @@ -475,340 +418,118 @@ class TestDeepxRuntimeManifest(unittest.TestCase): self.assertFalse(DEEPX_MANIFEST.needs_ld_library_path) -class TestDeepxPpuAnchorFreeDecode(unittest.TestCase): - def test_reads_the_box_bytes_as_pixel_geometry(self): - """An anchor-free PPU record holds cx, cy, w, h directly in pixels.""" - record = build_ppu_record( - box=(320.0, 160.0, 64.0, 32.0), - # grid columns carry no meaning for an anchor-free head, and must - # not influence the result - grid_y=7, - grid_x=9, - anchor_idx=2, - layer_idx=2, - score=0.9, - label=0, - ) - outputs = [record.reshape(1, 1, PPU_RECORD_SIZE)] - - detections = decode_ppu_anchor_free(outputs, 640, 640, 0.25, 0.45) - - # x: 320 +/- 32 -> 288..352, y: 160 +/- 16 -> 144..176 - self.assertAlmostEqual(detections[0][2], 144 / 640, places=5) - self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) - self.assertAlmostEqual(detections[0][4], 176 / 640, places=5) - self.assertAlmostEqual(detections[0][5], 352 / 640, places=5) - - def test_anchor_formula_would_blow_the_box_up(self): - """Guards the bug this decoder exists to prevent.""" - record = build_ppu_record( - box=(320.0, 160.0, 64.0, 32.0), - grid_y=7, - grid_x=9, - anchor_idx=2, - layer_idx=2, - score=0.9, - label=0, - ) - outputs = [record.reshape(1, 1, PPU_RECORD_SIZE)] - - anchor_free = decode_ppu_anchor_free(outputs, 640, 640, 0.25, 0.45) - anchor = decode_ppu_anchor(outputs, TEST_ANCHORS, 640, 640, 0.25, 0.45) - - # the anchor decode squares the width, saturating the clip to the frame - self.assertEqual(anchor[0][5], 1.0) - self.assertLess(anchor_free[0][5], 1.0) - - -class TestDeepxRawAnchorDecode(unittest.TestCase): - def build_raw_anchor_output(self, rows): - """Build an (1, N, 5+C) tensor with 80 classes.""" - out = np.zeros((1, len(rows), 85), dtype=np.float32) - - for i, (cx, cy, w, h, obj, label, cls_score) in enumerate(rows): - out[0, i, 0:4] = [cx, cy, w, h] - out[0, i, 4] = obj - out[0, i, 5 + label] = cls_score - - return [out] - - def test_confidence_is_objectness_times_class_score(self): - outputs = self.build_raw_anchor_output( - [(320.0, 320.0, 40.0, 80.0, 0.8, 3, 0.5)] - ) - - detections = decode_raw_anchor(outputs, 640, 640, 0.25, 0.45) - - self.assertEqual(detections[0][0], 3) - self.assertAlmostEqual(detections[0][1], 0.4, places=5) - - def test_drops_rows_whose_combined_score_is_below_threshold(self): - # 0.4 * 0.5 = 0.2, under a 0.25 threshold even though both parts are - # individually above it - outputs = self.build_raw_anchor_output( - [(320.0, 320.0, 40.0, 80.0, 0.4, 3, 0.5)] - ) - - detections = decode_raw_anchor(outputs, 640, 640, 0.25, 0.45) - - self.assertTrue(np.all(detections == 0)) - - def test_converts_center_form_to_normalized_corners(self): - outputs = self.build_raw_anchor_output( - [(320.0, 160.0, 64.0, 32.0, 1.0, 0, 1.0)] - ) - - detections = decode_raw_anchor(outputs, 640, 640, 0.25, 0.45) - - self.assertAlmostEqual(detections[0][2], 144 / 640, places=5) - self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) - self.assertAlmostEqual(detections[0][4], 176 / 640, places=5) - self.assertAlmostEqual(detections[0][5], 352 / 640, places=5) - - -class TestDeepxRawAnchorFreeDecode(unittest.TestCase): - def build_raw_anchor_free_output(self, rows, num_classes=3, total_rows=10): - """Build an (1, N, 4+C) tensor -- already row-per-box -- with no - objectness column, padded with all-zero rows so N comfortably - outnumbers the channel count (4+C). - - The decoder tells a channel-major export apart from a row-major one - purely by comparing those two sizes, the same way it would on a real - grid of thousands of anchors -- and that comparison only comes out - right when boxes outnumber channels, so a test with too few rows - would trip the same ambiguity `auto` accepts by design. - """ - total_rows = max(total_rows, len(rows)) - out = np.zeros((1, total_rows, 4 + num_classes), dtype=np.float32) - - for i, (cx, cy, w, h, label, cls_score) in enumerate(rows): - out[0, i, 0:4] = [cx, cy, w, h] - out[0, i, 4 + label] = cls_score - - return [out] - - def test_keeps_a_detection_between_the_deepx_and_frigate_default_thresholds(self): - """Regression test: post_process_yolo hardcodes a 0.4 score - threshold, which would silently drop this 0.3-confidence detection - even though it clears the DEEPX decoder's own 0.25 default.""" - outputs = self.build_raw_anchor_free_output( - [(320.0, 320.0, 40.0, 80.0, 2, 0.3)] - ) - - detections = decode_raw_anchor_free(outputs, 640, 640, 0.25, 0.45) - - self.assertEqual(detections[0][0], 2) - self.assertAlmostEqual(detections[0][1], 0.3, places=5) - - def test_drops_detections_below_the_score_threshold(self): - outputs = self.build_raw_anchor_free_output( - [(320.0, 320.0, 40.0, 80.0, 2, 0.1)] - ) - - detections = decode_raw_anchor_free(outputs, 640, 640, 0.25, 0.45) - - self.assertTrue(np.all(detections == 0)) - - def test_transposes_a_channel_major_output(self): - """DX-COM may export (4+C, N) instead of (N, 4+C); both must decode - to the same result.""" - row_major = self.build_raw_anchor_free_output( - [(320.0, 320.0, 40.0, 80.0, 2, 0.9)] - ) - channel_major = [np.transpose(row_major[0], (0, 2, 1))] - - row_major_detections = decode_raw_anchor_free(row_major, 640, 640, 0.25, 0.45) - channel_major_detections = decode_raw_anchor_free( - channel_major, 640, 640, 0.25, 0.45 - ) - - np.testing.assert_array_almost_equal( - row_major_detections, channel_major_detections - ) - - def test_converts_center_form_to_normalized_corners(self): - outputs = self.build_raw_anchor_free_output( - [(320.0, 160.0, 64.0, 32.0, 0, 1.0)] - ) - - detections = decode_raw_anchor_free(outputs, 640, 640, 0.25, 0.45) - - self.assertAlmostEqual(detections[0][2], 144 / 640, places=5) - self.assertAlmostEqual(detections[0][3], 288 / 640, places=5) - self.assertAlmostEqual(detections[0][4], 176 / 640, places=5) - self.assertAlmostEqual(detections[0][5], 352 / 640, places=5) - - def test_returns_empty_detections_for_no_output(self): - out = np.zeros((1, 0, 84), dtype=np.float32) - - self.assertTrue( - np.all(decode_raw_anchor_free([out], 640, 640, 0.25, 0.45) == 0) - ) - - -class TestDeepxRawNmsInHeadDecode(unittest.TestCase): - def test_reads_corner_records_without_running_nms(self): - """Two heavily overlapping boxes both survive: the head already ran NMS.""" - out = np.array( - [ - [ - [100.0, 100.0, 200.0, 200.0, 0.9, 2.0], - [102.0, 102.0, 202.0, 202.0, 0.8, 2.0], - ] - ], - dtype=np.float32, - ) - - detections = decode_raw_nms_in_head([out], 640, 640, 0.25) - - self.assertEqual(detections[0][0], 2) - self.assertAlmostEqual(detections[0][1], 0.9, places=5) - self.assertAlmostEqual(detections[0][3], 100 / 640, places=5) - self.assertEqual(detections[1][0], 2) - self.assertAlmostEqual(detections[1][1], 0.8, places=5) - - def test_applies_the_score_threshold(self): - out = np.array( - [ - [ - [100.0, 100.0, 200.0, 200.0, 0.9, 2.0], - [300.0, 300.0, 400.0, 400.0, 0.1, 5.0], - ] - ], - dtype=np.float32, - ) - - detections = decode_raw_nms_in_head([out], 640, 640, 0.25) - - self.assertAlmostEqual(detections[0][1], 0.9, places=5) - self.assertTrue(np.all(detections[1] == 0)) - - def test_returns_empty_detections_for_an_empty_output(self): - out = np.zeros((1, 0, 6), dtype=np.float32) - - self.assertTrue(np.all(decode_raw_nms_in_head([out], 640, 640, 0.25) == 0)) - - class TestDeepxDamoyoloDecode(unittest.TestCase): - def total_priors(self, width, height, strides): - return sum((height // s) * (width // s) for s in strides) + """DX-COM emits (1, N, C) sigmoid scores and (1, N, 4) pixel corners, as + the compiled ModelZoo models report.""" - def test_decodes_a_detection_from_already_decoded_boxes(self): - """box_output shaped (1, N, 4): DX-COM's export is assumed to have - already applied the full distance-to-box decode, so these are final - (x_min, y_min, x_max, y_max) pixel coordinates, used as-is.""" - width = height = 640 - num_priors = 100 - prior_index = 5 + def build_output(self, rows, num_classes=80, boxes_first=False): + scores = np.zeros((1, len(rows), num_classes), dtype=np.float32) + boxes = np.zeros((1, len(rows), 4), dtype=np.float32) - cls_scores = np.zeros((1, num_priors, 80), dtype=np.float32) - cls_scores[0, prior_index, 7] = 0.9 + for i, (x_min, y_min, x_max, y_max, label, score) in enumerate(rows): + boxes[0, i] = [x_min, y_min, x_max, y_max] + scores[0, i, label] = score - box_output = np.zeros((1, num_priors, 4), dtype=np.float32) - box_output[0, prior_index] = [100.0, 120.0, 140.0, 160.0] + return [boxes, scores] if boxes_first else [scores, boxes] + def test_decodes_a_detection_in_either_output_order(self): + rows = [(288.0, 144.0, 352.0, 176.0, 3, 0.9)] + + for boxes_first in (False, True): + with self.subTest(boxes_first=boxes_first): + detections = decode_damoyolo_raw( + self.build_output(rows, boxes_first=boxes_first), + 640, + 640, + 0.25, + 0.45, + ) + + self.assertEqual(detections[0][0], 3) + self.assertAlmostEqual(detections[0][1], 0.9, 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_clips_boxes_that_run_off_the_frame(self): + # DX-COM's decoded corners can fall outside the input detections = decode_damoyolo_raw( - [cls_scores, box_output], width, height, 0.25, 0.45 + self.build_output([(-40.0, 10.0, 700.0, 100.0, 0, 0.9)]), + 640, + 640, + 0.25, + 0.45, ) - self.assertEqual(detections[0][0], 7) - self.assertAlmostEqual(detections[0][1], 0.9, places=5) - self.assertAlmostEqual(detections[0][2], 120 / height, places=5) - self.assertAlmostEqual(detections[0][3], 100 / width, places=5) - self.assertAlmostEqual(detections[0][4], 160 / height, places=5) - self.assertAlmostEqual(detections[0][5], 140 / width, places=5) - - def test_decodes_a_detection_from_a_dfl_distribution(self): - """box_output shaped (1, N, 4, bins): a raw per-side DFL histogram, - needing softmax + integral + stride-scaled distance-to-box here.""" - width = height = 640 - strides = (8, 16, 32) - grid_w = width // 8 - prior_row, prior_col = 10, 10 - prior_index = prior_row * grid_w + prior_col - num_priors = self.total_priors(width, height, strides) - - cls_scores = np.zeros((1, num_priors, 80), dtype=np.float32) - cls_scores[0, prior_index, 9] = 0.8 - - # reg_max=1 (2 bins): put ~all softmax mass on bin index 1, so the - # integral (weighted sum against [0, 1]) comes out to ~1.0 stride - # unit on every side - box_output = np.zeros((1, num_priors, 4, 2), dtype=np.float32) - box_output[0, prior_index, :, 0] = -30.0 - box_output[0, prior_index, :, 1] = 30.0 - - detections = decode_damoyolo_raw( - [cls_scores, box_output], width, height, 0.25, 0.45, strides - ) - - # stride-8 grid cell (10, 10) centers at pixel (80, 80); ~1 - # stride-unit distance on every side -> an 8px box around it - self.assertEqual(detections[0][0], 9) - self.assertAlmostEqual(detections[0][1], 0.8, places=5) - self.assertAlmostEqual(detections[0][2], 72 / height, places=3) - self.assertAlmostEqual(detections[0][3], 72 / width, places=3) - self.assertAlmostEqual(detections[0][4], 88 / height, places=3) - self.assertAlmostEqual(detections[0][5], 88 / width, places=3) + self.assertEqual(detections[0][3], 0.0) + self.assertEqual(detections[0][5], 1.0) def test_returns_empty_detections_below_score_threshold(self): - num_priors = 10 - cls_scores = np.zeros((1, num_priors, 80), dtype=np.float32) - cls_scores[0, 0, 1] = 0.1 - - box_output = np.zeros((1, num_priors, 4), dtype=np.float32) - box_output[0, 0] = [10.0, 10.0, 20.0, 20.0] - - detections = decode_damoyolo_raw([cls_scores, box_output], 640, 640, 0.5, 0.45) - - self.assertTrue(np.all(detections == 0)) - - def test_returns_empty_detections_when_dfl_prior_count_does_not_match_the_grid( - self, - ): - # a DFL-shaped box output with far fewer priors than the (8, 16, 32) - # strides actually produce for a 640x640 input - num_priors = 5 - cls_scores = np.zeros((1, num_priors, 80), dtype=np.float32) - cls_scores[0, 0, 1] = 0.9 - box_output = np.zeros((1, num_priors, 4, 2), dtype=np.float32) - detections = decode_damoyolo_raw( - [cls_scores, box_output], 640, 640, 0.25, 0.45, (8, 16, 32) + self.build_output([(288.0, 144.0, 352.0, 176.0, 3, 0.1)]), + 640, + 640, + 0.25, + 0.45, ) self.assertTrue(np.all(detections == 0)) - def test_returns_empty_detections_when_box_and_class_counts_disagree(self): - cls_scores = np.zeros((1, 10, 80), dtype=np.float32) - cls_scores[0, 0, 1] = 0.9 - box_output = np.zeros((1, 7, 4), dtype=np.float32) + def test_a_box_class_count_mismatch_is_reported_not_read_as_empty(self): + """A structural mismatch stays wrong on every later frame, so it comes + back as None and the caller reports it, rather than looking like a + frame in which nothing cleared the threshold.""" + scores = np.zeros((1, 5, 80), dtype=np.float32) + boxes = np.zeros((1, 4, 4), dtype=np.float32) - detections = decode_damoyolo_raw([cls_scores, box_output], 640, 640, 0.25, 0.45) + self.assertIsNone(decode_damoyolo_raw([scores, boxes], 640, 640, 0.25, 0.45)) - self.assertTrue(np.all(detections == 0)) + def test_a_wrong_output_count_is_reported_not_read_as_empty(self): + out = np.zeros((1, 8400, 84), dtype=np.float32) - def test_returns_empty_detections_for_wrong_output_count(self): - outputs = [np.zeros((1, 10, 80), dtype=np.float32)] - - detections = decode_damoyolo_raw(outputs, 640, 640, 0.25, 0.45, (8, 16, 32)) - - self.assertTrue(np.all(detections == 0)) + self.assertIsNone(decode_damoyolo_raw([out], 640, 640, 0.25, 0.45)) -if __name__ == "__main__": - unittest.main() +class TestDeepxCompiledModelShapes(unittest.TestCase): + """The output shapes DX-RT 3.4 reports for the compiled 640x640, 80-class + models in the DEEPX ModelZoo, each pinned to the decoder it has to land + on. Named by the head that produced them, since every model compiled from + the same head reports the same shape and the decoder only sees the shape. + """ + + def test_modelzoo_shapes(self): + cases = { + "anchor-based, three scales flattened into one tensor": ( + [(1, 25200, 85)], + YoloLayout.anchor, + ), + "anchor-free, channel-major, no objectness column": ( + [(1, 84, 8400)], + YoloLayout.anchor_free, + ), + "NMS in the head, a fixed run of corner records": ( + [(1, 300, 6)], + YoloLayout.nms_in_head, + ), + } + + for head, (shapes, layout) in cases.items(): + with self.subTest(head=head): + self.assertIs(layout_of(shapes, 80), layout) class TestDeepxModelType(unittest.TestCase): def test_a_model_type_with_no_decoder_is_rejected(self): """Frigate defaults model_type to ssd, and every unsupported type - would otherwise fall through to the yolo-generic decode path and - return nonsense rather than an error.""" + would otherwise be decoded as YOLO and return nonsense rather than + an error.""" for model_type in (ModelTypeEnum.ssd, ModelTypeEnum.dfine): - with self.subTest(model_type=model_type): - with self.assertRaises(ValidationError): - DeepxDetectorConfig(type="deepx", model=model_with_type(model_type)) + with ( + self.subTest(model_type=model_type), + self.assertRaises(ValidationError), + ): + DeepxDetectorConfig(type="deepx", model=model_with_type(model_type)) def test_supported_model_types_are_accepted(self): for model_type in (ModelTypeEnum.yologeneric, ModelTypeEnum.damoyolo): @@ -817,21 +538,12 @@ class TestDeepxModelType(unittest.TestCase): type="deepx", model=model_with_type(model_type) ) - self.assertEqual(config.resolved_model_type.value, model_type.value) + self.assertEqual(config.model.model_type, model_type) def test_an_unresolved_model_defers_the_check(self): """A device string is validated on its own before any model is attached, so an absent model must not fail.""" - config = DeepxDetectorConfig(type="deepx") - - self.assertIsNone(config.resolved_model_type) - - def test_model_type_values_match_frigate_model_types(self): - """The detector reads model.model_type back as a ModelTypeEnum, so - every value has to round-trip.""" - for value in DeepxModelTypeEnum: - with self.subTest(value=value): - self.assertEqual(ModelTypeEnum(value.value).value, value.value) + DeepxDetectorConfig(type="deepx") class TestDeepxIpcEndpoint(unittest.TestCase): @@ -841,9 +553,11 @@ class TestDeepxIpcEndpoint(unittest.TestCase): def _construct(self): # dx_engine is not installed in the test environment, and forcing the # import to fail keeps this test honest on a machine where it is - with patch.dict(sys.modules, {"dx_engine": None}): - with self.assertRaises(ImportError): - DeepxDetector(DeepxDetectorConfig(type="deepx")) + with ( + patch.dict(sys.modules, {"dx_engine": None}), + self.assertRaises(ImportError), + ): + DeepxDetector(DeepxDetectorConfig(type="deepx")) def test_the_filesystem_socket_is_named_when_nothing_else_is(self): with patch.dict(os.environ): @@ -851,48 +565,317 @@ class TestDeepxIpcEndpoint(unittest.TestCase): self._construct() self.assertEqual(os.environ[DXRT_IPC_ENDPOINT_ENV], DXRT_IPC_SOCKET) + def test_a_blank_endpoint_counts_as_unset(self): + # an empty variable would otherwise be handed to DX-RT as the path + with patch.dict(os.environ, {DXRT_IPC_ENDPOINT_ENV: ""}): + self._construct() + self.assertEqual(os.environ[DXRT_IPC_ENDPOINT_ENV], DXRT_IPC_SOCKET) + + def test_an_abstract_endpoint_has_no_file_to_check(self): + with ( + patch.dict(os.environ, {DXRT_IPC_ENDPOINT_ENV: "@dxrt_dynamic_ipc.sock"}), + self.assertNoLogs("frigate.detectors.plugins.deepx", level="WARNING"), + ): + self._construct() + def test_an_operator_supplied_endpoint_is_left_alone(self): with patch.dict(os.environ, {DXRT_IPC_ENDPOINT_ENV: "/run/dxrt/custom.sock"}): self._construct() self.assertEqual(os.environ[DXRT_IPC_ENDPOINT_ENV], "/run/dxrt/custom.sock") + def test_a_missing_socket_is_named_before_the_runtime_buries_it(self): + """DX-RT reports an absent socket as a bare connect error several + layers down, with nothing saying which path it tried.""" + with ( + patch.dict(os.environ, {DXRT_IPC_ENDPOINT_ENV: "/run/dxrt/absent.sock"}), + self.assertLogs("frigate.detectors.plugins.deepx", level="WARNING") as logs, + ): + self._construct() + + self.assertTrue( + any("/run/dxrt/absent.sock" in r.getMessage() for r in logs.records) + ) + + def test_the_socket_lives_in_a_mountable_directory(self): + # a single socket file bind-mounted on its own pins the inode dxrtd + # had at container start, so the directory has to be what is mounted + self.assertEqual(os.path.dirname(DXRT_IPC_SOCKET), "/run/dxrt") + class TestDeepxServiceCheck(unittest.TestCase): """DX-RT scans /proc for a dxrtd process to decide the daemon is up, which - a container cannot satisfy from the host. See satisfy_service_check.""" + a container cannot satisfy from the host. The runtime's own SERVICE toggle + turns that scan off, and the IPC client keeps using the socket.""" - def setUp(self): - deepx = sys.modules["frigate.detectors.plugins.deepx"] - self.deepx = deepx - deepx._SERVICE_PLACEHOLDER = None - self.addCleanup(setattr, deepx, "_SERVICE_PLACEHOLDER", None) + def test_the_process_scan_is_disabled_before_the_model_is_loaded(self): + dx_engine = MagicMock() + dx_engine.Configuration.ITEM.SERVICE = object() + order = [] + dx_engine.Configuration.return_value.set_enable.side_effect = lambda *a: ( + order.append("set_enable") + ) + session = MagicMock() + session.get_output_tensors_info.return_value = [ + {"shape": [1, 8400, 80]}, + {"shape": [1, 8400, 4]}, + ] + session.is_ppu.return_value = False - def test_nothing_is_started_when_a_daemon_is_already_visible(self): - with patch.object(self.deepx, "service_is_visible", return_value=True): - with patch.object(self.deepx.subprocess, "Popen") as popen: - satisfy_service_check("/tmp/dxrt_dynamic_ipc.sock") + def engine(*args): + order.append("engine") + return session - popen.assert_not_called() + dx_engine.InferenceEngine.side_effect = engine - def test_nothing_is_started_when_the_socket_is_absent(self): - # a genuinely stopped daemon must still report itself - with patch.object(self.deepx, "service_is_visible", return_value=False): - with patch.object(self.deepx.subprocess, "Popen") as popen: - satisfy_service_check("/nonexistent/dxrt.sock") + config = DeepxDetectorConfig( + type="deepx", + model=ModelConfig( + model_type=ModelTypeEnum.damoyolo, + labelmap_path=None, + labelmap={79: "toothbrush"}, + ), + ) + config.model.path = "/nonexistent/model.dxnn" - popen.assert_not_called() + with ( + patch.dict(sys.modules, {"dx_engine": dx_engine}), + patch.object(DeepxDetector, "activate_dependencies"), + patch("os.path.isfile", return_value=True), + ): + DeepxDetector(config) - def test_a_placeholder_named_dxrtd_is_started_otherwise(self): - with patch.object(self.deepx, "service_is_visible", return_value=False): - with patch.object(self.deepx.os.path, "exists", return_value=True): - with patch.object(self.deepx.subprocess, "Popen") as popen: - popen.return_value.poll.return_value = None - satisfy_service_check("/tmp/dxrt_dynamic_ipc.sock") + dx_engine.Configuration.return_value.set_enable.assert_called_once_with( + dx_engine.Configuration.ITEM.SERVICE, False + ) + self.assertEqual(order, ["set_enable", "engine"]) - args, kwargs = popen.call_args - self.assertEqual(args[0][0], "dxrtd") - self.assertEqual(kwargs["executable"], "/bin/sleep") - def test_the_scan_matches_the_runtime_own_check(self): - # our own cmdline does not name dxrtd, so a bare test run sees none - self.assertIsInstance(service_is_visible(), bool) +class TestDeepxDetectorLayout(unittest.TestCase): + """The detector asks the runtime about the loaded model and keeps the + answer; the per-frame decode only dispatches on it.""" + + def _detector(self, model_type, outputs_info, ppu=False, dynamic=False): + dx_engine = MagicMock() + dx_engine.Configuration.ITEM.SERVICE = object() + session = dx_engine.InferenceEngine.return_value + session.get_output_tensors_info.return_value = outputs_info + session.is_ppu.return_value = ppu + session.has_dynamic_output.return_value = dynamic + + config = DeepxDetectorConfig( + type="deepx", + model=ModelConfig( + model_type=model_type, labelmap_path=None, labelmap={79: "toothbrush"} + ), + ) + config.model.path = "/nonexistent/model.dxnn" + + with ( + patch.dict(sys.modules, {"dx_engine": dx_engine}), + patch.object(DeepxDetector, "activate_dependencies"), + patch("os.path.isfile", return_value=True), + ): + return DeepxDetector(config), dx_engine + + def test_the_layout_is_read_from_the_model_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) + + def test_a_ppu_model_is_recognized_by_the_runtime_flag(self): + # the shape DX-RT 3.4 reports for the compiled PPU model + detector, _ = self._detector( + ModelTypeEnum.yologeneric, [{"shape": [8400]}], ppu=True + ) + + self.assertIs(detector.output.layout, YoloLayout.ppu) + + def test_an_anchor_based_ppu_model_is_reported_once(self): + detector, _ = self._detector( + ModelTypeEnum.yologeneric, [{"shape": [8400]}], ppu=True + ) + detector.session.run.return_value = [build_ppu_record((0.6, 0.4, 0.3, 0.7))] + detector.width, detector.height = 640, 640 + + with self.assertLogs("frigate.detectors.plugins.deepx", level="ERROR") as logs: + first = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + second = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + self.assertTrue(np.all(first == 0) and np.all(second == 0)) + self.assertEqual(len([r for r in logs.records if r.levelname == "ERROR"]), 1) + + def test_the_ppu_head_is_settled_once_and_kept(self): + """Which head a PPU model was compiled from is a property of the + model. A later frame whose boxes all happen to fall under 1.0 must not + re-open the question and start dropping detections.""" + detector, _ = self._detector( + ModelTypeEnum.yologeneric, [{"shape": [8400]}], ppu=True + ) + detector.width, detector.height = 640, 640 + detector.session.run.return_value = [ + build_ppu_record((320.0, 160.0, 64.0, 32.0), label=3) + ] + + first = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + self.assertFalse(detector.ppu_anchor_based) + + detector.session.run.return_value = [build_ppu_record((0.6, 0.4, 0.3, 0.7))] + second = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + self.assertEqual(first[0][0], 3) + self.assertFalse(detector.ppu_anchor_based_reported) + # decoded as pixels, which is a sub-pixel box in the corner + self.assertAlmostEqual(second[0][1], 0.9, places=5) + + def test_an_empty_or_all_zero_ppu_frame_leaves_the_head_undecided(self): + """Only a box value above 1 (anchor-free) or a frame of values + strictly inside 0..1 (anchor-based) is evidence; an empty frame or a + degenerate all-zero record is neither and must not freeze the model + as anchor-based.""" + detector, _ = self._detector( + ModelTypeEnum.yologeneric, [{"shape": [8400]}], ppu=True + ) + detector.width, detector.height = 640, 640 + + for frame in ( + np.zeros((1, 0, PPU_RECORD_SIZE), np.uint8), + build_ppu_record((0.0, 0.0, 0.0, 0.0)), + ): + detector.session.run.return_value = [frame] + detections = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + self.assertTrue(np.all(detections == 0)) + self.assertIsNone(detector.ppu_anchor_based) + + detector.session.run.return_value = [ + build_ppu_record((320.0, 160.0, 64.0, 32.0), label=3) + ] + detections = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + self.assertFalse(detector.ppu_anchor_based) + self.assertEqual(detections[0][0], 3) + + def test_damoyolo_is_checked_against_the_model_at_load(self): + """A YOLO model or a wrong label map under damo-yolo fails at load + with the reason, rather than returning nothing on every frame.""" + with self.assertRaisesRegex(ValueError, "yolo-generic"): + self._detector(ModelTypeEnum.damoyolo, [{"shape": [1, 84, 8400]}]) + + with self.assertRaisesRegex(ValueError, "PPU"): + self._detector( + ModelTypeEnum.damoyolo, + [{"shape": [1, 8400, 80]}, {"shape": [1, 8400, 4]}], + ppu=True, + ) + + def test_damoyolo_frames_that_lose_the_expected_pair_are_reported_once(self): + """The load-time check passed; a frame that still does not carry + the pair says so once rather than only under debug logging.""" + detector, _ = self._detector( + ModelTypeEnum.damoyolo, [{"shape": [1, 8400, 80]}, {"shape": [1, 8400, 4]}] + ) + detector.width, detector.height = 640, 640 + detector.session.run.return_value = [np.zeros((1, 8400, 84), dtype=np.float32)] + + with self.assertLogs("frigate.detectors.plugins.deepx", level="ERROR") as logs: + first = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + second = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + self.assertTrue(np.all(first == 0) and np.all(second == 0)) + self.assertEqual(len([r for r in logs.records if r.levelname == "ERROR"]), 1) + + def test_damoyolo_needs_no_layout(self): + detector, _ = self._detector( + ModelTypeEnum.damoyolo, + [{"shape": [1, 8400, 80]}, {"shape": [1, 8400, 4]}], + ) + + self.assertIsNone(detector.output) + + def test_an_empty_label_map_fails_at_load_with_the_deepx_context(self): + dx_engine = MagicMock() + dx_engine.Configuration.ITEM.SERVICE = object() + dx_engine.InferenceEngine.return_value.get_output_tensors_info.return_value = [ + {"shape": [1, 84, 8400]} + ] + config = DeepxDetectorConfig( + type="deepx", + model=ModelConfig(model_type=ModelTypeEnum.yologeneric, labelmap_path=None), + ) + config.model.path = "/nonexistent/model.dxnn" + + with ( + patch.dict(sys.modules, {"dx_engine": dx_engine}), + patch.object(DeepxDetector, "activate_dependencies"), + patch("os.path.isfile", return_value=True), + self.assertRaisesRegex( + ValueError, "Cannot decode DEEPX model.*labelmap_path" + ), + ): + DeepxDetector(config) + + def test_the_tensor_layout_is_logged_before_the_first_decode(self): + """A decode that raises must still leave the runtime shapes in the + log, which is what tells a wrong layout guess apart afterwards.""" + detector, _ = self._detector( + ModelTypeEnum.yologeneric, [{"shape": [1, 84, 8400]}] + ) + detector.width, detector.height = 640, 640 + detector.session.run.return_value = [np.zeros((1, 84, 8400), np.float32)] + + with ( + patch.object(detector, "decode", side_effect=RuntimeError("boom")), + self.assertLogs("frigate.detectors.plugins.deepx", level="INFO") as logs, + self.assertRaises(RuntimeError), + ): + detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + self.assertTrue( + any( + "output[0]: shape=(1, 84, 8400)" in r.getMessage() for r in logs.records + ) + ) + + def test_an_undecodable_model_fails_at_load(self): + with self.assertRaisesRegex(ValueError, "Cannot decode DEEPX model"): + self._detector(ModelTypeEnum.yologeneric, [{"shape": [1, 8400, 7]}]) + + def test_detect_raw_runs_the_model_output_through_the_detected_layout(self): + """End to end on a mocked session: a compiled channel-major + anchor-free shape is oriented by its column count and handed to the + shared decoder, coming back as Frigate's (20, 6) rows.""" + 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 + detector.session.run.return_value = [output] + detector.width, detector.height = 640, 640 + + detections = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + 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) + + def test_detect_raw_decodes_damoyolo_without_a_layout(self): + detector, _ = self._detector( + ModelTypeEnum.damoyolo, [{"shape": [1, 8400, 80]}, {"shape": [1, 8400, 4]}] + ) + scores = np.zeros((1, 8400, 80), dtype=np.float32) + boxes = np.zeros((1, 8400, 4), dtype=np.float32) + scores[0, 0, 5] = 0.8 + boxes[0, 0] = [100.0, 100.0, 200.0, 300.0] + detector.session.run.return_value = [scores, boxes] + detector.width, detector.height = 640, 640 + + detections = detector.detect_raw(np.zeros((1, 640, 640, 3), np.uint8)) + + self.assertEqual(detections[0][0], 5) + self.assertAlmostEqual(detections[0][1], 0.8, places=5) + self.assertAlmostEqual(detections[0][4], 300 / 640, places=5) diff --git a/frigate/test/test_post_processing.py b/frigate/test/test_post_processing.py index 6fd33783a7..9290c0c152 100644 --- a/frigate/test/test_post_processing.py +++ b/frigate/test/test_post_processing.py @@ -81,21 +81,9 @@ class TestYoloNmsPostProcess(unittest.TestCase): self.assertEqual(len(detections), 1) np.testing.assert_allclose(detections[0], [0, 0.90, *A_ROW], atol=2e-3) - def test_honors_caller_supplied_thresholds(self): - """A detector with its own thresholds must not be held to the 0.4 - default, which would drop this 0.3 confidence box.""" - output = self._single_output([[A_CX, A_CY, A_W, A_H, 0.30, 0.0]]) - - self.assertEqual(len(kept(post_process_yolo(output, WIDTH, HEIGHT))), 0) - - detections = kept(post_process_yolo(output, WIDTH, HEIGHT, 0.25, 0.45)) - - self.assertEqual(len(detections), 1) - np.testing.assert_allclose(detections[0], [0, 0.30, *A_ROW], atol=2e-3) - class TestMultipartYoloPostProcess(unittest.TestCase): - def _multipart_output(self, a_conf=0.95, b_conf=0.90) -> list[np.ndarray]: + def _multipart_output(self) -> list[np.ndarray]: """Build a 3-scale anchor-based YOLO output containing boxes A and B, both decoded through anchor 0 of the stride-32 scale.""" outputs = [ @@ -106,8 +94,8 @@ class TestMultipartYoloPostProcess(unittest.TestCase): stride, (anchor_w, anchor_h) = 32, (142, 110) for cx, cy, w, h, conf, class_channel in [ - (A_CX, A_CY, A_W, A_H, a_conf, 5), # class 0 - (B_CX, B_CY, B_W, B_H, b_conf, 6), # class 1 + (A_CX, A_CY, A_W, A_H, 0.95, 5), # class 0 + (B_CX, B_CY, B_W, B_H, 0.90, 6), # class 1 ]: cell_x, cell_y = int(cx // stride), int(cy // stride) dx = (cx / stride - cell_x + 0.5) / 2 @@ -139,19 +127,6 @@ class TestMultipartYoloPostProcess(unittest.TestCase): self.assertEqual(len(detections), 0) - def test_honors_caller_supplied_thresholds(self): - """Same as the single-tensor case: the multi-part branch must filter - at the caller's threshold, not the 0.4 default.""" - outputs = self._multipart_output(a_conf=0.35, b_conf=0.30) - - self.assertEqual(len(kept(post_process_yolo(outputs, WIDTH, HEIGHT))), 0) - - detections = kept(post_process_yolo(outputs, WIDTH, HEIGHT, 0.25, 0.45)) - - self.assertEqual(len(detections), 2) - np.testing.assert_allclose(detections[0], [0, 0.35, *A_ROW], atol=2e-3) - np.testing.assert_allclose(detections[1], [1, 0.30, *B_ROW], atol=2e-3) - class TestYoloxPostProcess(unittest.TestCase): def test_keeps_separated_objects_far_from_origin(self): diff --git a/frigate/util/model.py b/frigate/util/model.py index e0d1d2f961..80f0d29a04 100644 --- a/frigate/util/model.py +++ b/frigate/util/model.py @@ -134,8 +134,6 @@ def __post_process_multipart_yolo( output_list, width, height, - score_threshold=0.4, - nms_threshold=0.4, ): anchors = [ [(12, 16), (19, 36), (40, 28)], @@ -168,7 +166,7 @@ def __post_process_multipart_yolo( class_conf = class_probs[class_id] conf = class_conf * pred[4] - if conf < score_threshold: + if conf < 0.4: continue dx = pred[0] @@ -193,8 +191,8 @@ def __post_process_multipart_yolo( indices = cv2.dnn.NMSBoxes( bboxes=xyxy_to_xywh_for_nms(all_boxes), scores=all_scores, - score_threshold=score_threshold, - nms_threshold=nms_threshold, + score_threshold=0.4, + nms_threshold=0.4, ) results = np.zeros((20, 6), np.float32) @@ -216,13 +214,7 @@ def __post_process_multipart_yolo( return results -def __post_process_nms_yolo( - predictions: np.ndarray, - width, - height, - score_threshold=0.4, - nms_threshold=0.4, -) -> np.ndarray: +def __post_process_nms_yolo(predictions: np.ndarray, width, height) -> np.ndarray: predictions = np.squeeze(predictions) # transpose the output so it has order (inferences, class_ids) @@ -230,8 +222,8 @@ def __post_process_nms_yolo( predictions = predictions.T scores = np.max(predictions[:, 4:], axis=1) - predictions = predictions[scores > score_threshold, :] - scores = scores[scores > score_threshold] + predictions = predictions[scores > 0.4, :] + scores = scores[scores > 0.4] class_ids = np.argmax(predictions[:, 4:], axis=1) # Rescale box @@ -245,10 +237,7 @@ def __post_process_nms_yolo( # run NMS indices = cv2.dnn.NMSBoxes( - xyxy_to_xywh_for_nms(boxes), - scores, - score_threshold=score_threshold, - nms_threshold=nms_threshold, + xyxy_to_xywh_for_nms(boxes), scores, score_threshold=0.4, nms_threshold=0.4 ) detections = np.zeros((20, 6), np.float32) for i, (bbox, confidence, class_id) in enumerate( @@ -269,21 +258,11 @@ def __post_process_nms_yolo( return detections -def post_process_yolo( - output: list[np.ndarray], - width: int, - height: int, - score_threshold: float = 0.4, - nms_threshold: float = 0.4, -) -> np.ndarray: +def post_process_yolo(output: list[np.ndarray], width: int, height: int) -> np.ndarray: if len(output) > 1: - return __post_process_multipart_yolo( - output, width, height, score_threshold, nms_threshold - ) + return __post_process_multipart_yolo(output, width, height) else: - return __post_process_nms_yolo( - output[0], width, height, score_threshold, nms_threshold - ) + return __post_process_nms_yolo(output[0], width, height) def post_process_yolox(