feat: add engine auto-recompilation, mortality detection, multi-execution batching, and dashboard updates

- Add engine_utils for TensorRT compatibility verification, auto-recompilation from .pt models, and YAML auto-updates
- Add mortality detection pipeline (mortality.py, test_run_mortality.sh, mortality_config.yaml)
- Add multi-execution batch modes (parallel_processes, tensor_batching, hybrid) in batch_runner.py
- Add daily test run automation scripts and video processing runners
- Add dashboard REST API, live stream endpoints, and web UI templates
- Clean up git tracking by ignoring __pycache__, .pyc, and build artifacts
This commit is contained in:
proitlab committed 2026-08-19 10:17:17 +07:00
1 parent c05f3d98d0
commit 010f6e1493
97 files changed
+3361 -643

No files matched your search

Binary file not shown.
+112
View File
@@ -0,0 +1,112 @@
"""Tests for CountingZone validation and double-count suppression."""
from __future__ import annotations
import sys
import unittest
from unittest.mock import MagicMock
# counting.py imports cv2; stub it so unit tests run without OpenCV installed.
if "cv2" not in sys.modules:
cv2_stub = MagicMock()
cv2_stub.pointPolygonTest = MagicMock(return_value=1.0)
sys.modules["cv2"] = cv2_stub
import numpy as np
from chicken_counter.config import GateConfig, RoiConfig
from chicken_counter.counting import CountingZone
from chicken_counter.types import TrackObservation
def _zone(**kwargs) -> CountingZone:
defaults = dict(
roi=RoiConfig(
points=[(0, 0), (200, 0), (200, 200), (0, 200)],
min_overlap_ratio=0.0,
),
gate=GateConfig(),
trail_length=10,
track_buffer=75,
min_box_area_px=0,
validate_while_inside=True,
dedupe_radius_px=64,
dedupe_frames=40,
)
defaults.update(kwargs)
return CountingZone(**defaults)
def _track(
track_id: int,
centroid: tuple[int, int] = (100, 100),
bbox: tuple[int, int, int, int] | None = None,
) -> TrackObservation:
cx, cy = centroid
if bbox is None:
bbox = (cx - 20, cy - 20, cx + 20, cy + 20)
return TrackObservation(
track_id=track_id,
class_id=0,
confidence=0.9,
bbox_xyxy=bbox,
centroid=centroid,
)
class CountingDedupeTests(unittest.TestCase):
def setUp(self) -> None:
# Treat every centroid as inside the ROI polygon for these unit tests.
sys.modules["cv2"].pointPolygonTest = MagicMock(return_value=1.0)
def test_counts_unique_track_once(self) -> None:
zone = _zone()
events = zone.update([_track(1)], frame_index=1)
self.assertEqual(len(events), 1)
self.assertEqual(zone.total_entered_count, 1)
events = zone.update([_track(1)], frame_index=2)
self.assertEqual(len(events), 0)
self.assertEqual(zone.total_entered_count, 1)
def test_suppresses_nearby_id_switch_double_count(self) -> None:
zone = _zone(dedupe_radius_px=64, dedupe_frames=40)
events = zone.update([_track(1, (100, 100))], frame_index=10)
self.assertEqual(len(events), 1)
self.assertEqual(zone.total_entered_count, 1)
# New track ID near the prior count should reuse sequence, not increment.
events = zone.update([_track(2, (110, 105))], frame_index=20)
self.assertEqual(len(events), 0)
self.assertEqual(zone.total_entered_count, 1)
self.assertTrue(zone.is_validated(2))
self.assertEqual(zone.sequence_number_for(2), 1)
def test_allows_distant_second_bird(self) -> None:
zone = _zone(dedupe_radius_px=64, dedupe_frames=40)
zone.update([_track(1, (40, 40))], frame_index=10)
events = zone.update([_track(2, (160, 160))], frame_index=15)
self.assertEqual(len(events), 1)
self.assertEqual(zone.total_entered_count, 2)
def test_allows_recount_after_dedupe_window(self) -> None:
zone = _zone(dedupe_radius_px=64, dedupe_frames=10)
zone.update([_track(1, (100, 100))], frame_index=10)
events = zone.update([_track(2, (100, 100))], frame_index=30)
self.assertEqual(len(events), 1)
self.assertEqual(zone.total_entered_count, 2)
def test_dedupe_disabled_when_radius_zero(self) -> None:
zone = _zone(dedupe_radius_px=0, dedupe_frames=40)
zone.update([_track(1, (100, 100))], frame_index=10)
events = zone.update([_track(2, (100, 100))], frame_index=12)
self.assertEqual(len(events), 1)
self.assertEqual(zone.total_entered_count, 2)
def test_counting_polygon_is_numpy_array(self) -> None:
zone = _zone()
self.assertIsInstance(zone._counting_polygon, np.ndarray)
if __name__ == "__main__":
unittest.main()
-40
View File
@@ -1,40 +0,0 @@
"""Tests for mask polygon handling in DetectionTracker."""
from __future__ import annotations
import unittest
import numpy as np
from chicken_counter.types import TrackObservation
class TrackObservationMaskTests(unittest.TestCase):
def test_track_observation_accepts_mask_polygon(self) -> None:
poly = np.array([[10.0, 20.0], [30.0, 20.0], [30.0, 40.0]], dtype=np.float64)
track = TrackObservation(
track_id=1,
class_id=0,
confidence=0.9,
bbox_xyxy=(10, 20, 30, 40),
centroid=(20, 30),
mask_polygon_xy=poly,
)
self.assertIsNotNone(track.mask_polygon_xy)
self.assertEqual(track.mask_polygon_xy.shape, (3, 2))
def test_crop_offset_translation_pattern(self) -> None:
"""Mirrors tracking.infer crop offset applied to mask polygons."""
offset_x, offset_y = 100, 50
local = np.array([[0.0, 0.0], [10.0, 0.0], [10.0, 10.0]], dtype=np.float64)
full = local.copy()
full[:, 0] += offset_x
full[:, 1] += offset_y
self.assertAlmostEqual(float(full[0, 0]), 100.0)
self.assertAlmostEqual(float(full[0, 1]), 50.0)
self.assertAlmostEqual(float(full[2, 0]), 110.0)
self.assertAlmostEqual(float(full[2, 1]), 60.0)
if __name__ == "__main__":
unittest.main()