diff --git a/frigate/detectors/plugins/deepx.py b/frigate/detectors/plugins/deepx.py index f8286a7b02..3e28450304 100644 --- a/frigate/detectors/plugins/deepx.py +++ b/frigate/detectors/plugins/deepx.py @@ -464,6 +464,68 @@ 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. + + The exported head emits a single tensor with no objectness column, + (4+C, N) channel-major or already transposed to (N, 4+C), and box columns + already in input pixels as (center_x, center_y, width, height). This is + also what `model_format: auto` falls back to: with only one output + tensor there is no objectness column to tell an anchor-based head from + an anchor-free one apart, so every column past the box is read as a + class score, which is only correct for an anchor-free head -- matching + `auto` being "only reliable for anchor-free heads" per its config + description. + """ + predictions = outputs[0] + + # Drop the batch axis without a blind squeeze, which would also collapse + # the box-count axis on the single-candidate frame case. Fold every + # leading axis into rows instead. + 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, @@ -1016,6 +1078,22 @@ class Deepx(DetectionApi): self.nms_threshold, ) + # `auto` (the default) and an explicit `anchor_free` both read a + # single tensor with no objectness column. Routing these through 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(outputs, self.width, self.height) def detect_raw(self, tensor_input): diff --git a/frigate/test/test_deepx.py b/frigate/test/test_deepx.py index d2ecb55f6b..00ee72ae44 100644 --- a/frigate/test/test_deepx.py +++ b/frigate/test/test_deepx.py @@ -20,6 +20,7 @@ from frigate.detectors.plugins.deepx import ( decode_ppu_anchor, decode_ppu_anchor_free, decode_raw_anchor, + decode_raw_anchor_free, decode_raw_nms_in_head, decode_ssd_raw, parse_device_ids, @@ -404,6 +405,84 @@ class TestDeepxRawAnchorDecode(unittest.TestCase): 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."""