feat: freeze ROI mode, deadband+sticky jitter fix, explicit fixed/frozen ROI flags

- zone_mode=freeze: TruckROITracker locks box after 3 consecutive truck
  detections; geometry stays fixed while presence stays live (truck
  departure -> confidence 0 -> batch transitions keep working)
- jitter: 10px deadband skips EMA on tiny bbox noise; sticky main-truck
  pick (IoU>=0.3) prevents ROI jumps between parked trucks
- TruckROI.fixed/frozen flags replace dashboard confidence>=1.0 heuristic
- gate truck_centroid on truck_present in both pipelines
- +22 tests (test_truck_roi.py, test_freeze_plumbing.py); 127 pass
This commit is contained in:
jetson committed 2026-09-28 14:36:21 +07:00
1 parent 9a276b21f2
commit 4bd9de58e1
8 files changed
+463 -27

No files matched your search

+6 -2
View File
@@ -67,7 +67,7 @@ class DashboardOverlay:
return frame
def _draw_roi(self, frame: np.ndarray, roi: TruckROI) -> None:
if roi.confidence >= 1.0:
if roi.fixed:
# Fixed zone — dashed cyan border on all edges
dash_len, gap_len = 12, 8
for edge in [
@@ -87,8 +87,12 @@ class DashboardOverlay:
cv2.rectangle(
frame, (roi.x1, roi.y1), (roi.x2, roi.y2), ORANGE, 2,
)
if roi.frozen:
label = f"ROI FROZEN ({roi.confidence:.0%})"
else:
label = f"TRUCK ROI ({roi.confidence:.0%})"
cv2.putText(
frame, f"TRUCK ROI ({roi.confidence:.0%})",
frame, label,
(roi.x1, roi.y1 - 8),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, ORANGE, 1,
)
+2
View File
@@ -209,6 +209,7 @@ class JobQueue:
class_filter=class_filter,
truck_model_config=truck_model_config,
fixed_zone=job.fixed_zone,
freeze_roi=(job.zone_mode == "freeze"),
preview_path=job.preview_path,
preview_enabled=job.preview_enabled,
preview_every_n=2,
@@ -256,6 +257,7 @@ class JobQueue:
class_filters=job.class_filters,
truck_model_config=truck_model_config,
fixed_zone=job.fixed_zone,
freeze_roi=(job.zone_mode == "freeze"),
preview_path=job.preview_path,
preview_enabled=job.preview_enabled,
preview_every_n=2,
+17 -5
View File
@@ -108,6 +108,7 @@ def run_pipeline(
preview_max_dim: int = 480,
preview_jpeg_quality: int = 60,
live_stats_callback: Callable[[dict], None] | None = None,
freeze_roi: bool = False,
) -> PipelineResult:
"""Process a video file through the counting pipeline.
@@ -128,6 +129,8 @@ def run_pipeline(
preview_jpeg_quality: JPEG quality for preview frames (1-100).
live_stats_callback: Optional fn(dict) called every 10 frames with
{fps, loading, unloading, frame} for live UI stats.
freeze_roi: Lock the auto-detected truck ROI after 3 consecutive
detections (zone_mode "freeze"); False = plain auto-detect.
Returns:
PipelineResult with counting summary.
@@ -176,7 +179,9 @@ def run_pipeline(
tracker = ByteTrackTracker(shared_model, conf=sack_conf)
stabilizer = BboxStabilizer()
roi_tracker = TruckROITracker(frame_width=w, frame_height=h)
roi_tracker = TruckROITracker(
frame_width=w, frame_height=h, freeze_after=(3 if freeze_roi else None)
)
counter = MultiClassLineCounter(
line_y=int(h * 0.50),
line_x_start=int(w * 0.38),
@@ -235,7 +240,7 @@ def run_pipeline(
# Batch lifecycle
truck_centroid = None
if roi is not None:
if truck_present:
truck_centroid = ((roi.x1 + roi.x2) / 2.0, (roi.y1 + roi.y2) / 2.0)
if frame_idx % truck_det_interval == 0:
@@ -398,8 +403,13 @@ def run_merged_pipeline(
preview_max_dim: int = 480,
preview_jpeg_quality: int = 60,
live_stats_callback: Callable[[dict], None] | None = None,
freeze_roi: bool = False,
) -> MergedPipelineResult:
"""Run multiple models on the same video, merge detections into one output."""
"""Run multiple models on the same video, merge detections into one output.
freeze_roi: Lock the auto-detected truck ROI after 3 consecutive
detections (zone_mode "freeze"); False = plain auto-detect.
"""
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
raise RuntimeError(f"Cannot open video: {video_path}")
@@ -441,7 +451,9 @@ def run_merged_pipeline(
tracker_model = YOLO(model_configs[0].path)
tracker = ByteTrackTracker(tracker_model, conf=sack_conf)
stabilizer = BboxStabilizer()
roi_tracker = TruckROITracker(frame_width=w, frame_height=h)
roi_tracker = TruckROITracker(
frame_width=w, frame_height=h, freeze_after=(3 if freeze_roi else None)
)
counter = MultiClassLineCounter(
line_y=int(h * 0.50),
line_x_start=int(w * 0.38),
@@ -494,7 +506,7 @@ def run_merged_pipeline(
counter.line_x_end = roi.x2
truck_centroid = None
if roi is not None:
if truck_present:
truck_centroid = ((roi.x1 + roi.x2) / 2.0, (roi.y1 + roi.y2) / 2.0)
if frame_idx % truck_det_interval == 0:
+119 -20
View File
@@ -3,10 +3,18 @@
Uses exponential moving average (EMA) to smooth the bounding box across
frames, preventing jitter from frame-to-frame detection variance.
For y2-based counting (bottom edge of sack bbox), the counting line is
placed `LINE_OFFSET_PX` pixels relative to the truck bottom edge (`roi.y2`).
With `offset = +20`, the line sits at `roi.y2 + 20` (~620px), cleanly
separating sacks on the ground (`y2 > 650`) from loaded sacks (`y2 < 580`).
A deadband (`deadband_px`) skips the EMA entirely when every bbox edge
moved less than that many pixels, so the ROI and counting line stay still
under tiny detection jitter.
The counting line is placed `LINE_OFFSET_PX` pixels relative to the truck
top edge (`roi.y1`), i.e. the line sits at `roi.y1 + LINE_OFFSET_PX`.
With `offset = +20` the line sits 20px below the truck top edge, into
the truck body.
Optional freeze (`freeze_after`): after that many consecutive detections
the smoothed box is locked; geometry never moves again, while presence
(confidence) stays live.
"""
from __future__ import annotations
@@ -20,6 +28,9 @@ from src.interfaces import Detection
# Negative = above truck top edge (towards the camera).
LINE_OFFSET_PX = 0
# Minimum IoU for the sticky main-truck pick to keep the previous truck.
STICKY_IOU = 0.3
@dataclass
class TruckROI:
@@ -31,6 +42,8 @@ class TruckROI:
y2: int
line_y: int # counting line Y position (pixels)
confidence: float
fixed: bool = False # True for zones built from zones.json (never moves)
frozen: bool = False # True when auto ROI is locked after freeze_after hits
@property
def width(self) -> int:
@@ -49,10 +62,13 @@ class TruckROITracker:
"""Tracks the main truck and provides a smoothed ROI + counting line.
Main truck = largest truck detection whose center X falls in the
expected lane (center region of the frame).
expected lane (center region of the frame). The pick is sticky: a
detection overlapping the previous main truck (IoU >= 0.3) wins over
the largest-in-lane rule, so the ROI does not jump between parked
trucks.
The counting line is placed `line_offset` pixels below the truck
bottom edge (`roi.y2`).
The counting line is placed `line_offset` pixels relative to the truck
top edge (`roi.y1`).
"""
def __init__(
@@ -63,6 +79,8 @@ class TruckROITracker:
lane_x_max: float = 0.80,
ema_alpha: float = 0.15,
line_offset: int = LINE_OFFSET_PX,
deadband_px: float = 10.0,
freeze_after: int | None = None,
) -> None:
self._fw = frame_width
self._fh = frame_height
@@ -70,6 +88,8 @@ class TruckROITracker:
self._lane_x_max = int(lane_x_max * frame_width)
self._alpha = ema_alpha
self._line_offset = line_offset
self._deadband = deadband_px
self._freeze_after = freeze_after
# Smoothed bbox (None until first detection)
self._sx1: float | None = None
@@ -77,6 +97,14 @@ class TruckROITracker:
self._sx2: float | None = None
self._sy2: float | None = None
# Sticky pick: raw bbox of the last chosen main truck
self._prev_raw_bbox: tuple[float, float, float, float] | None = None
# Freeze state
self._hit_streak = 0
self._frozen = False
self._frozen_box: tuple[float, float, float, float] | None = None
self._last_roi: TruckROI | None = None
self._frames_without_truck = 0
@@ -85,37 +113,60 @@ class TruckROITracker:
main = self._pick_main_truck(truck_detections)
if main is None:
self._hit_streak = 0
self._frames_without_truck += 1
if self._frozen:
# Frozen: geometry never resets, only presence goes away.
if self._frames_without_truck > 5:
# >5 misses: signal truck departure, keep frozen geometry.
self._last_roi = self._build_roi(self._frozen_box, 0.0)
return self._last_roi
return self._last_roi # hold last known confidence briefly
if self._frames_without_truck > 5: # Clear ROI if truck is missing for >5 updates (~3 seconds)
self.reset()
return None
return self._last_roi # hold last known ROI briefly
self._frames_without_truck = 0
self._hit_streak += 1
if self._frozen:
# Geometry locked; presence stays live.
self._last_roi = self._build_roi(self._frozen_box, main.confidence)
return self._last_roi
x1, y1, x2, y2 = main.bbox
# EMA smoothing
# EMA smoothing (skipped while all edges moved < deadband)
if self._sx1 is None:
self._sx1, self._sy1 = float(x1), float(y1)
self._sx2, self._sy2 = float(x2), float(y2)
else:
elif (
abs(x1 - self._sx1) >= self._deadband
or abs(y1 - self._sy1) >= self._deadband
or abs(x2 - self._sx2) >= self._deadband
or abs(y2 - self._sy2) >= self._deadband
):
a = self._alpha
self._sx1 = a * x1 + (1 - a) * self._sx1
self._sy1 = a * y1 + (1 - a) * self._sy1
self._sx2 = a * x2 + (1 - a) * self._sx2
self._sy2 = a * y2 + (1 - a) * self._sy2
# Build ROI — line placed at truck top edge
roi_x1 = max(0, int(self._sx1))
roi_y1 = max(0, int(self._sy1))
roi_x2 = min(self._fw, int(self._sx2))
roi_y2 = min(self._fh, int(self._sy2))
line_y = roi_y1 + self._line_offset
if (
self._freeze_after is not None
and self._hit_streak >= self._freeze_after
and not self._frozen
):
self._frozen = True
self._frozen_box = (self._sx1, self._sy1, self._sx2, self._sy2)
self._last_roi = TruckROI(
x1=roi_x1, y1=roi_y1, x2=roi_x2, y2=roi_y2,
line_y=line_y, confidence=main.confidence,
)
if self._frozen:
self._last_roi = self._build_roi(self._frozen_box, main.confidence)
else:
self._last_roi = self._build_roi(
(self._sx1, self._sy1, self._sx2, self._sy2), main.confidence
)
return self._last_roi
@property
@@ -130,11 +181,40 @@ class TruckROITracker:
self._sx1 = self._sy1 = self._sx2 = self._sy2 = None
self._last_roi = None
self._frames_without_truck = 0
self._prev_raw_bbox = None
self._hit_streak = 0
self._frozen = False
self._frozen_box = None
def _build_roi(
self,
box: tuple[float, float, float, float],
confidence: float,
) -> TruckROI:
"""Build a TruckROI from a smoothed/frozen box (line at top edge)."""
sx1, sy1, sx2, sy2 = box
roi_x1 = max(0, int(sx1))
roi_y1 = max(0, int(sy1))
roi_x2 = min(self._fw, int(sx2))
roi_y2 = min(self._fh, int(sy2))
return TruckROI(
x1=roi_x1, y1=roi_y1, x2=roi_x2, y2=roi_y2,
line_y=roi_y1 + self._line_offset,
confidence=confidence,
frozen=self._frozen,
)
def _pick_main_truck(
self, detections: list[Detection]
) -> Detection | None:
"""Select the largest truck whose center X is in the expected lane."""
"""Select the main truck: sticky overlap with previous, else largest in lane."""
# Sticky: keep the previous main truck if any detection still overlaps it.
if self._prev_raw_bbox is not None:
for det in detections:
if _iou(det.bbox, self._prev_raw_bbox) >= STICKY_IOU:
self._prev_raw_bbox = tuple(det.bbox)
return det
best: Detection | None = None
best_area = 0
@@ -148,4 +228,23 @@ class TruckROITracker:
best = det
best_area = area
if best is not None:
self._prev_raw_bbox = tuple(best.bbox)
return best
def _iou(
a: tuple[float, float, float, float],
b: tuple[float, float, float, float],
) -> float:
"""Intersection-over-union of two (x1, y1, x2, y2) boxes."""
ax1, ay1, ax2, ay2 = a
bx1, by1, bx2, by2 = b
ix1, iy1 = max(ax1, bx1), max(ay1, by1)
ix2, iy2 = min(ax2, bx2), min(ay2, by2)
iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1)
inter = iw * ih
if inter <= 0.0:
return 0.0
union = (ax2 - ax1) * (ay2 - ay1) + (bx2 - bx1) * (by2 - by1) - inter
return inter / union if union > 0.0 else 0.0
+1
View File
@@ -59,6 +59,7 @@ def build_static_roi(zone: dict[str, Any], video_w: int, video_h: int) -> TruckR
x1=x1, y1=y1, x2=x2, y2=y2,
line_y=line_y,
confidence=1.0,
fixed=True,
)
+4
View File
@@ -406,6 +406,10 @@
<input type="radio" name="zone_mode" value="fixed">
Fixed Zone
</label>
<label>
<input type="radio" name="zone_mode" value="freeze">
Auto-detect + Freeze
</label>
</div>
<div class="zone-fixed-config" id="zone-fixed-config">
<div class="zone-preset-row">
+168
View File
@@ -0,0 +1,168 @@
# tests/test_freeze_plumbing.py
"""Wiring tests for zone_mode "freeze" plumbing (pipeline, job, template)."""
import inspect
import pytest
from src.pipeline import run_merged_pipeline, run_pipeline
# ── Pipeline signatures ──────────────────────────────────────────────────
def test_run_pipeline_accepts_freeze_roi():
"""run_pipeline exposes freeze_roi keyword (default off = dynamic)."""
sig = inspect.signature(run_pipeline)
assert "freeze_roi" in sig.parameters
assert sig.parameters["freeze_roi"].default is False
def test_run_merged_pipeline_accepts_freeze_roi():
"""run_merged_pipeline exposes freeze_roi keyword (default off = dynamic)."""
sig = inspect.signature(run_merged_pipeline)
assert "freeze_roi" in sig.parameters
assert sig.parameters["freeze_roi"].default is False
# ── Job worker passes freeze_roi from zone_mode ──────────────────────────
def _fake_pipeline_result(tmp_path):
from src.pipeline import PipelineResult
return PipelineResult(
output_path=str(tmp_path / "out.mp4"),
frame_count=10,
loading_count=1,
unloading_count=0,
batch_count=1,
duration_seconds=1.0,
model_name="a.pt",
class_filter=None,
)
def _fake_merged_result(tmp_path):
from src.pipeline import MergedPipelineResult
return MergedPipelineResult(
output_path=str(tmp_path / "merged.mp4"),
frame_count=10,
loading_count=1,
unloading_count=0,
batch_count=1,
duration_seconds=1.0,
model_names=["a.pt", "b.pt"],
class_filter=None,
)
def _run_single_zone_mode(tmp_path, monkeypatch, zone_mode: str) -> dict:
import src.job as job_mod
from src.job import Job, JobQueue, JobStatus
from src.model_registry import ModelConfig
captured: dict = {}
def fake_run_pipeline(**kwargs):
captured.update(kwargs)
return _fake_pipeline_result(tmp_path)
monkeypatch.setattr(job_mod, "run_pipeline", fake_run_pipeline)
q = JobQueue(output_dir=str(tmp_path))
job = Job(
job_id=f"job-freeze-{zone_mode}",
video_path="/tmp/test.mp4",
model_configs=[
ModelConfig(
filename="a.pt",
path=str(tmp_path / "a.pt"),
stem="a",
known_classes=["sack", "box", "truck"],
)
],
output_dir=str(tmp_path / f"job-freeze-{zone_mode}"),
zone_mode=zone_mode,
)
q._jobs[job.job_id] = job
q._run_job(job.job_id)
assert job.status == JobStatus.COMPLETED
return captured
def _run_merged_zone_mode(tmp_path, monkeypatch, zone_mode: str) -> dict:
import src.job as job_mod
from src.job import Job, JobQueue, JobStatus
from src.model_registry import ModelConfig
captured: dict = {}
def fake_run_merged_pipeline(**kwargs):
captured.update(kwargs)
return _fake_merged_result(tmp_path)
monkeypatch.setattr(job_mod, "run_merged_pipeline", fake_run_merged_pipeline)
q = JobQueue(output_dir=str(tmp_path))
job = Job(
job_id=f"job-freeze-merged-{zone_mode}",
video_path="/tmp/test.mp4",
model_configs=[
ModelConfig(
filename="a.pt",
path=str(tmp_path / "a.pt"),
stem="a",
known_classes=["sack", "box", "truck"],
),
ModelConfig(
filename="b.pt",
path=str(tmp_path / "b.pt"),
stem="b",
known_classes=["sack", "box"],
),
],
output_dir=str(tmp_path / f"job-freeze-merged-{zone_mode}"),
zone_mode=zone_mode,
)
q._jobs[job.job_id] = job
q._run_job(job.job_id)
assert job.status == JobStatus.COMPLETED
return captured
@pytest.mark.parametrize(
"zone_mode,expected",
[("freeze", True), ("dynamic", False), ("fixed", False)],
)
def test_job_single_passes_freeze_roi(tmp_path, monkeypatch, zone_mode, expected):
"""Single-model worker forwards freeze_roi=True only for zone_mode "freeze"."""
kwargs = _run_single_zone_mode(tmp_path, monkeypatch, zone_mode)
assert kwargs["freeze_roi"] is expected
assert kwargs["fixed_zone"] is None
@pytest.mark.parametrize(
"zone_mode,expected",
[("freeze", True), ("dynamic", False), ("fixed", False)],
)
def test_job_merged_passes_freeze_roi(tmp_path, monkeypatch, zone_mode, expected):
"""Merged worker forwards freeze_roi=True only for zone_mode "freeze"."""
kwargs = _run_merged_zone_mode(tmp_path, monkeypatch, zone_mode)
assert kwargs["freeze_roi"] is expected
# ── Template ─────────────────────────────────────────────────────────────
def test_index_html_has_freeze_radio():
"""Upload form offers the third zone mode radio."""
from pathlib import Path
html = Path(__file__).resolve().parent.parent.joinpath(
"templates", "index.html"
).read_text(encoding="utf-8")
assert '<input type="radio" name="zone_mode" value="freeze">' in html
+146
View File
@@ -0,0 +1,146 @@
import pytest
from src.interfaces import Detection
from src.truck_roi import LINE_OFFSET_PX, TruckROI, TruckROITracker
def truck(bbox, conf=0.9):
return Detection(
bbox=tuple(bbox), confidence=conf, class_id=0, class_name="truck"
)
def make_tracker(**kw):
return TruckROITracker(frame_width=640, frame_height=480, **kw)
def geom(roi):
return (roi.x1, roi.y1, roi.x2, roi.y2, roi.line_y)
def test_first_detection_initializes_box_and_line():
t = make_tracker(line_offset=20)
roi = t.update([truck((300, 100, 500, 400))])
assert roi is not None
assert (roi.x1, roi.y1, roi.x2, roi.y2) == (300, 100, 500, 400)
assert roi.line_y == roi.y1 + 20 == 120
assert roi.frozen is False
assert t.roi is roi
def test_line_offset_defaults_to_top_edge():
t = make_tracker()
roi = t.update([truck((300, 100, 500, 400))])
assert roi.line_y == roi.y1 + LINE_OFFSET_PX
def test_deadband_small_jitter_leaves_line_y_unchanged():
t = make_tracker() # deadband_px = 10
first = t.update([truck((300, 100, 500, 400))])
jittered = t.update([truck((305, 105, 505, 405))]) # 5px per edge < 10
assert geom(jittered) == geom(first)
assert jittered.line_y == first.line_y == 100
def test_deadband_large_move_applies_ema():
t = make_tracker()
t.update([truck((300, 100, 500, 400))])
moved = t.update([truck((350, 150, 550, 450))]) # 50px per edge >= 10
# EMA: 0.15*150 + 0.85*100 = 107.5 -> int 107
assert moved.y1 == 107
assert moved.line_y == 107
assert moved.y1 != 100
def test_sticky_pick_prefers_previous_truck_over_largest():
a1 = truck((280, 100, 460, 400), conf=0.95) # area 54000, center 370 in lane
a2 = truck((300, 150, 440, 350), conf=0.95) # area 28000, IoU with a1 ~0.52
b2 = truck((450, 100, 560, 450), conf=0.40) # area 38500 > a2, center 505 in lane
t = make_tracker()
t.update([a1])
roi = t.update([a2, b2]) # b2 is now largest, but a2 overlaps prev pick
assert roi.confidence == 0.95 # a2 picked, not b2
# y1 moved toward a2 (150); picking b2 (y1=100) would leave y1 at 100
assert roi.y1 == 107
# EMA of a2 x1: 0.15*300 + 0.85*280 = 283
assert roi.x1 == 283
def test_freeze_locks_geometry_after_hits():
t = make_tracker(freeze_after=3)
t.update([truck((300, 100, 500, 400))])
t.update([truck((305, 110, 505, 410))])
third = t.update([truck((310, 120, 510, 420))])
assert third.frozen is True
fourth = t.update([truck((350, 200, 550, 500))])
assert fourth.frozen is True
assert geom(fourth) == geom(third)
def test_freeze_counts_deadband_skips_as_hits():
t = make_tracker(freeze_after=2)
t.update([truck((300, 100, 500, 400))])
second = t.update([truck((304, 104, 504, 404))]) # < deadband: geometry skips
assert second.frozen is True
assert second.y1 == 100 # geometry unchanged by deadband skip
def test_freeze_truck_lost_keeps_roi_then_zero_confidence():
t = make_tracker(freeze_after=1)
first = t.update([truck((300, 100, 500, 400))])
assert first.frozen is True
frozen_geom = geom(first)
for _ in range(5): # within >5 miss threshold: hold last confidence
held = t.update([])
assert held is not None
assert held.confidence == pytest.approx(0.9)
assert held.frozen is True
assert geom(held) == frozen_geom
lost = t.update([]) # 6th miss: signal departure, keep frozen geometry
assert lost is not None
assert lost.confidence == 0.0
assert lost.frozen is True
assert geom(lost) == frozen_geom
still_lost = t.update([]) # stays dead, still not None
assert still_lost is not None
assert still_lost.confidence == 0.0
assert geom(still_lost) == frozen_geom
def test_never_frozen_truck_gone_returns_none():
t = make_tracker() # freeze_after=None
roi = t.update([truck((300, 100, 500, 400))])
assert roi.frozen is False
for _ in range(5):
assert t.update([]) is not None # hold behavior
assert t.update([]) is None # >5 misses: cleared
assert t.roi is None
assert t.frames_without_truck == 0
def test_reset_clears_freeze_and_sticky_state():
t = make_tracker(freeze_after=2)
t.update([truck((300, 100, 500, 400))])
second = t.update([truck((305, 110, 505, 410))])
assert second.frozen is True
t.reset()
assert t.roi is None
assert t.frames_without_truck == 0
after = t.update([truck((300, 100, 500, 400))])
assert after.frozen is False # streak restarted, 1 < 2
assert (after.x1, after.y1, after.x2, after.y2) == (300, 100, 500, 400)
def test_truck_roi_fixed_defaults_false():
roi = TruckROI(x1=0, y1=0, x2=10, y2=10, line_y=0, confidence=1.0)
assert roi.fixed is False
assert roi.frozen is False