tracking and motion updates

DEMO.md

@@ -57,9 +57,11 @@ Recommended starting detector settings:
 - Resolution: `640` by default, `960` or `1280` if detections are missing.
 
 Rules automatically add their referenced labels to the detector class list.
-Rules are frame-local for now. Presence, count, near, far, enter, exit, change,
-and cooldown are supported; object identity, movement, speed, direction, and
-trajectories are not tracked yet.
+Rules can use presence, count, near, far, moving, enter, exit, change, and
+cooldown. Moving uses lightweight same-label tracking across sampled frames;
+it requires at least three tracked observations and tolerates one missed sampled
+frame by default. Speed, direction, long-gap re-identification, and trajectory
+paths are not supported yet.
 
 ## Prompts To Try
 
@@ -87,6 +89,12 @@ Simple presence:
 When a laptop is visible, notify me.
 ```
 
+Simple motion:
+
+```text
+If a car is moving, notify me.
+```
+
 ## Expected Flow
 
 1. Upload the video in Detector.

README.md

@@ -86,12 +86,14 @@ rules:
         name: turn on pc
 ```
 
-Initial video conditions include presence, count, near, and far. Near/far use the
-minimum horizontal/vertical gap between detection boxes in normalized frame
-percent. Gates include enabled state and cooldown. Triggers can fire while a
-condition is true, when it becomes true, when it becomes false, or on either
-change. Rules are frame-local: Tiny Trigger does not yet track object identities,
-movement, speed, direction, or trajectories across frames.
+Initial video conditions include presence, count, near, far, and moving. Near/far
+use the minimum horizontal/vertical gap between detection boxes in normalized
+frame percent. Moving uses lightweight same-label centroid tracking across
+sampled frames and tolerates one missed sampled frame by default to avoid repeat
+alerts from detector flicker. Gates include enabled state and cooldown. Triggers
+can fire while a condition is true, when it becomes true, when it becomes false,
+or on either change. Tiny Trigger does not yet support speed, direction, long-gap
+re-identification, or trajectory path rules.
 
 ```yaml
 rules:

frontend/src/lib/types.ts

@@ -5,6 +5,7 @@ export interface Detection {
   confidence: number
   bbox_xyxy: [number, number, number, number]
   bbox_xyxy_norm: [number, number, number, number]
+  track_id: number | null
 }
 
 export interface ActionEvent {

server.py

@@ -61,6 +61,7 @@ def _detection_dict(d: Any) -> dict[str, Any]:
         "confidence": round(d.confidence, 4),
         "bbox_xyxy": [round(v, 1) for v in d.bbox_xyxy],
         "bbox_xyxy_norm": [round(v, 4) for v in d.bbox_xyxy_norm],
+        "track_id": d.track_id,
     }
 
 

tests/test_automation.py

@@ -9,7 +9,13 @@ from tiny_trigger.automation import RuleEngine, document_labels, evaluate_video_
 from tiny_trigger.models import Detection, FrameSample
 
 
-def detection(label: str, box: tuple[float, float, float, float], frame: int = 0, time: float = 0.0) -> Detection:
+def detection(
+    label: str,
+    box: tuple[float, float, float, float],
+    frame: int = 0,
+    time: float = 0.0,
+    track_id: int | None = None,
+) -> Detection:
     return Detection(
         frame_index=frame,
         timestamp_sec=time,
@@ -17,6 +23,7 @@ def detection(label: str, box: tuple[float, float, float, float], frame: int = 0
         confidence=0.9,
         bbox_xyxy=(0.0, 0.0, 10.0, 10.0),
         bbox_xyxy_norm=box,
+        track_id=track_id,
     )
 
 
@@ -122,6 +129,121 @@ def test_rule_labels_include_condition_labels() -> None:
     assert document_labels(document) == ["person", "monitor", "cat", "door"]
 
 
+def test_moving_condition_uses_tracked_centroid_history() -> None:
+    document = load_automation_text(
+        json.dumps(
+            {
+                "rules": [
+                    {
+                        "name": "car-moving",
+                        "when": {
+                            "all": [
+                                {"moving": {"label": "car", "min_displacement_percent": 3, "window_frames": 3}}
+                            ]
+                        },
+                        "then": [{"type": "simulate", "name": "notify"}],
+                    }
+                ]
+            }
+        )
+    )
+    engine = RuleEngine(document.rules)
+
+    first = engine.evaluate_frame(
+        [detection("car", (0.10, 0.10, 0.20, 0.20), frame=0, track_id=1)],
+        frame_index=0,
+        timestamp_sec=0.0,
+    )
+    second = engine.evaluate_frame(
+        [detection("car", (0.11, 0.10, 0.21, 0.20), frame=1, time=1.0, track_id=1)],
+        frame_index=1,
+        timestamp_sec=1.0,
+    )
+    third = engine.evaluate_frame(
+        [detection("car", (0.20, 0.10, 0.30, 0.20), frame=2, time=2.0, track_id=1)],
+        frame_index=2,
+        timestamp_sec=2.0,
+    )
+
+    assert first == []
+    assert second == []
+    assert [event.action for event in third] == ["notify"]
+
+
+def test_moving_condition_migrates_short_window_to_three() -> None:
+    document = load_automation_text(
+        json.dumps(
+            {
+                "rules": [
+                    {
+                        "name": "legacy-moving",
+                        "when": {"all": [{"moving": {"label": "car", "window_frames": 2}}]},
+                        "then": [{"type": "simulate", "name": "notify"}],
+                    }
+                ]
+            }
+        )
+    )
+
+    moving = document.rules[0].when.all_conditions[0].moving
+    assert moving is not None
+    assert moving.window_frames == 3
+
+
+def test_moving_condition_does_not_reenter_after_one_missed_frame() -> None:
+    document = load_automation_text(
+        json.dumps(
+            {
+                "rules": [
+                    {
+                        "name": "car-moving",
+                        "when": {
+                            "all": [
+                                {
+                                    "moving": {
+                                        "label": "car",
+                                        "min_displacement_percent": 3,
+                                        "window_frames": 3,
+                                        "max_missing_frames": 1,
+                                    }
+                                }
+                            ]
+                        },
+                        "then": [{"type": "simulate", "name": "notify"}],
+                    }
+                ]
+            }
+        )
+    )
+    engine = RuleEngine(document.rules)
+
+    assert engine.evaluate_frame(
+        [detection("car", (0.10, 0.10, 0.20, 0.20), frame=0, track_id=1)],
+        frame_index=0,
+        timestamp_sec=0.0,
+    ) == []
+    assert engine.evaluate_frame(
+        [detection("car", (0.14, 0.10, 0.24, 0.20), frame=1, time=1.0, track_id=1)],
+        frame_index=1,
+        timestamp_sec=1.0,
+    ) == []
+    first_alert = engine.evaluate_frame(
+        [detection("car", (0.18, 0.10, 0.28, 0.20), frame=2, time=2.0, track_id=1)],
+        frame_index=2,
+        timestamp_sec=2.0,
+    )
+    missed_frame = engine.evaluate_frame([], frame_index=3, timestamp_sec=3.0)
+    same_track_returns = engine.evaluate_frame(
+        [detection("car", (0.22, 0.10, 0.32, 0.20), frame=4, time=4.0, track_id=1)],
+        frame_index=4,
+        timestamp_sec=4.0,
+    )
+
+    assert [event.action for event in first_alert] == ["notify"]
+    assert missed_frame == []
+    assert same_track_returns == []
+
+
 def test_gate_cooldown_fire_once_per_window() -> None:
     document = load_automation_text(
         json.dumps(

tests/test_detector.py

@@ -1,7 +1,36 @@
 from __future__ import annotations
 
-from tiny_trigger.detector import parse_class_prompt
+from tiny_trigger.detector import parse_class_prompt, suppress_duplicate_detections
+from tiny_trigger.models import Detection
 
 
 def test_parse_class_prompt_splits_and_dedupes() -> None:
     assert parse_class_prompt(" cat, feeder robot\npackage; cat ") == ["cat", "feeder robot", "package"]
+
+
+def detection(label: str, confidence: float, box: tuple[float, float, float, float]) -> Detection:
+    return Detection(
+        frame_index=0,
+        timestamp_sec=0.0,
+        label=label,
+        confidence=confidence,
+        bbox_xyxy=(0.0, 0.0, 10.0, 10.0),
+        bbox_xyxy_norm=box,
+    )
+
+
+def test_suppress_duplicate_detections_keeps_highest_confidence_same_label_box() -> None:
+    detections = [
+        detection("person", 0.62, (0.10, 0.10, 0.50, 0.50)),
+        detection("person", 0.91, (0.11, 0.11, 0.51, 0.51)),
+        detection("person", 0.80, (0.60, 0.10, 0.80, 0.30)),
+        detection("bag", 0.70, (0.11, 0.11, 0.51, 0.51)),
+    ]
+
+    filtered = suppress_duplicate_detections(detections, iou_threshold=0.8)
+
+    assert len(filtered) == 3
+    assert ("person", 0.91) in [(item.label, item.confidence) for item in filtered]
+    assert ("person", 0.62) not in [(item.label, item.confidence) for item in filtered]
+    assert ("person", 0.80) in [(item.label, item.confidence) for item in filtered]
+    assert ("bag", 0.70) in [(item.label, item.confidence) for item in filtered]

tests/test_tracking.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from tiny_trigger.models import Detection
+from tiny_trigger.tracking import SimpleTracker
+
+
+def detection(label: str, box: tuple[float, float, float, float]) -> Detection:
+    return Detection(
+        frame_index=0,
+        timestamp_sec=0.0,
+        label=label,
+        confidence=0.9,
+        bbox_xyxy=(0.0, 0.0, 10.0, 10.0),
+        bbox_xyxy_norm=box,
+    )
+
+
+def test_simple_tracker_keeps_same_label_track_id() -> None:
+    tracker = SimpleTracker(max_match_distance_percent=20)
+
+    first = tracker.update([detection("car", (0.10, 0.10, 0.20, 0.20))])
+    second = tracker.update([detection("car", (0.15, 0.10, 0.25, 0.20))])
+    third = tracker.update([detection("person", (0.15, 0.10, 0.25, 0.20))])
+
+    assert first[0].track_id == second[0].track_id
+    assert third[0].track_id != first[0].track_id

tests/test_video.py

@@ -34,6 +34,65 @@ class FakeDetector:
         ]
 
 
+class MovingDetector:
+    class_names = ["cat"]
+
+    def detect(
+        self,
+        frame: Any,
+        *,
+        frame_index: int,
+        timestamp_sec: float,
+        confidence: float,
+        image_size: int | None = None,
+        max_detections: int | None = None,
+    ) -> list[Detection]:
+        offset = frame_index * 0.01
+        return [
+            Detection(
+                frame_index=frame_index,
+                timestamp_sec=timestamp_sec,
+                label="cat",
+                confidence=0.99,
+                bbox_xyxy=(2.0 + frame_index, 2.0, 12.0 + frame_index, 12.0),
+                bbox_xyxy_norm=(0.1 + offset, 0.1, 0.2 + offset, 0.2),
+            )
+        ]
+
+
+class DuplicateDetector:
+    class_names = ["cat"]
+
+    def detect(
+        self,
+        frame: Any,
+        *,
+        frame_index: int,
+        timestamp_sec: float,
+        confidence: float,
+        image_size: int | None = None,
+        max_detections: int | None = None,
+    ) -> list[Detection]:
+        return [
+            Detection(
+                frame_index=frame_index,
+                timestamp_sec=timestamp_sec,
+                label="cat",
+                confidence=0.62,
+                bbox_xyxy=(2.0, 2.0, 16.0, 16.0),
+                bbox_xyxy_norm=(0.1, 0.1, 0.5, 0.5),
+            ),
+            Detection(
+                frame_index=frame_index,
+                timestamp_sec=timestamp_sec,
+                label="cat",
+                confidence=0.91,
+                bbox_xyxy=(3.0, 3.0, 17.0, 17.0),
+                bbox_xyxy_norm=(0.11, 0.11, 0.51, 0.51),
+            ),
+        ]
+
+
 def test_process_video_with_fake_detector(tmp_path: Path) -> None:
     cv2 = __import__("cv2")
     video_path = _make_video(tmp_path)
@@ -62,6 +121,37 @@ def test_process_video_with_fake_detector(tmp_path: Path) -> None:
         capture.release()
 
 
+def test_process_video_assigns_track_ids(tmp_path: Path) -> None:
+    video_path = _make_video(tmp_path, fps=10.0, frames=4)
+
+    result = process_video(
+        video_path=str(video_path),
+        class_prompt="cat",
+        frame_stride=1,
+        max_frames=3,
+        detector=MovingDetector(),
+        output_dir=str(tmp_path),
+    )
+
+    assert [item.track_id for item in result.detections] == [1, 1, 1]
+
+
+def test_process_video_suppresses_duplicate_same_label_boxes(tmp_path: Path) -> None:
+    video_path = _make_video(tmp_path, fps=10.0, frames=2)
+
+    result = process_video(
+        video_path=str(video_path),
+        class_prompt="cat",
+        frame_stride=1,
+        max_frames=1,
+        detector=DuplicateDetector(),
+        output_dir=str(tmp_path),
+    )
+
+    assert len(result.detections) == 1
+    assert result.detections[0].confidence == 0.91
+
+
 def test_process_video_samples_once_per_second(tmp_path: Path) -> None:
     video_path = _make_video(tmp_path, fps=30.0, frames=95)
 

tiny_trigger/automation.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 from collections import defaultdict
+from math import hypot
 from typing import Any, Literal
 
 from pydantic import AliasChoices, BaseModel, ConfigDict, Field, ValidationError, model_validator
@@ -60,6 +61,25 @@ class FarCondition(BaseModel):
     min_gap_percent: float = Field(default=25.0, ge=0.0)
 
 
+class MovingCondition(BaseModel):
+    label: str
+    min_displacement_percent: float = Field(default=3.0, ge=0.0)
+    window_frames: int = Field(default=3, ge=3)
+    max_missing_frames: int = Field(default=1, ge=0)
+
+    @model_validator(mode="before")
+    @classmethod
+    def migrate_short_window(cls, data: Any) -> Any:
+        if isinstance(data, dict) and data.get("window_frames") is not None:
+            try:
+                window_frames = int(data["window_frames"])
+            except (TypeError, ValueError):
+                return data
+            if window_frames < 3:
+                return {**data, "window_frames": 3}
+        return data
+
+
 class CooldownCondition(BaseModel):
     key: str | None = None
     seconds: float | None = Field(default=None, ge=0.0)
@@ -79,11 +99,12 @@ class ConditionBlock(BaseModel):
     count: CountCondition | None = None
     near: NearCondition | None = None
     far: FarCondition | None = None
+    moving: MovingCondition | None = None
     cooldown: CooldownCondition | None = None
 
     @model_validator(mode="after")
     def exactly_one_condition(self) -> "ConditionBlock":
-        selected = [self.present, self.count, self.near, self.far, self.cooldown]
+        selected = [self.present, self.count, self.near, self.far, self.moving, self.cooldown]
         if sum(item is not None for item in selected) != 1:
             raise ValueError("Each condition block must contain exactly one condition.")
         return self
@@ -195,6 +216,8 @@ def rule_labels(rule: AutomationRule) -> list[str]:
             labels.extend([condition.near.a, condition.near.b])
         if condition.far:
             labels.extend([condition.far.a, condition.far.b])
+        if condition.moving:
+            labels.append(condition.moving.label)
     return _dedupe_labels(labels)
 
 
@@ -217,6 +240,8 @@ class RuleEngine:
         self.rules = rules
         self.last_fired: dict[str, float] = dict(last_fired or {})
         self.last_matched: dict[str, bool] = dict(last_matched or {})
+        self.track_history: dict[int, list[tuple[int, tuple[float, float]]]] = {}
+        self.moving_track_last_seen: dict[tuple[str, int], int] = {}
 
     def evaluate_frame(
         self,
@@ -226,11 +251,12 @@ class RuleEngine:
         timestamp_sec: float,
     ) -> list[ActionEvent]:
         events: list[ActionEvent] = []
+        self._update_track_history(detections)
         for rule in self.rules:
             if not self._gate_allows(rule, timestamp_sec):
                 self.last_matched[rule.name] = False
                 continue
-            matched = self._rule_matches(rule, detections, timestamp_sec)
+            matched = self._rule_matches(rule, detections, frame_index, timestamp_sec)
             previous = self.last_matched.get(rule.name, False)
             edge = _trigger_edge(previous=previous, matched=matched)
             self.last_matched[rule.name] = matched
@@ -262,15 +288,21 @@ class RuleEngine:
                 )
         return events
 
-    def _rule_matches(self, rule: AutomationRule, detections: list[Detection], timestamp_sec: float) -> bool:
+    def _rule_matches(
+        self,
+        rule: AutomationRule,
+        detections: list[Detection],
+        frame_index: int,
+        timestamp_sec: float,
+    ) -> bool:
         all_ok = all(
-            self._condition_matches(condition, detections, rule.name, timestamp_sec)
+            self._condition_matches(condition, detections, frame_index, rule.name, timestamp_sec)
             for condition in rule.when.all_conditions
         )
         any_ok = True
         if rule.when.any_conditions:
             any_ok = any(
-                self._condition_matches(condition, detections, rule.name, timestamp_sec)
+                self._condition_matches(condition, detections, frame_index, rule.name, timestamp_sec)
                 for condition in rule.when.any_conditions
             )
         return all_ok and any_ok
@@ -286,6 +318,7 @@ class RuleEngine:
         self,
         condition: ConditionBlock,
         detections: list[Detection],
+        frame_index: int,
         rule_name: str,
         timestamp_sec: float,
     ) -> bool:
@@ -311,11 +344,54 @@ class RuleEngine:
                 return False
             return _min_box_gap_percent(left, right) >= condition.far.min_gap_percent
 
+        if condition.moving:
+            return self._moving_matches(condition.moving, by_label, frame_index)
+
         if condition.cooldown:
             return self._cooldown_allows(condition.cooldown, rule_name, timestamp_sec)
 
         return False
 
+    def _moving_matches(
+        self,
+        condition: MovingCondition,
+        by_label: dict[str, list[Detection]],
+        frame_index: int,
+    ) -> bool:
+        label = _label(condition.label)
+        label_detections = by_label[label]
+        for detection in label_detections:
+            if detection.track_id is None:
+                continue
+            history = self.track_history.get(detection.track_id, [])
+            if len(history) < condition.window_frames:
+                continue
+            _first_frame, first_centroid = history[-condition.window_frames]
+            _last_frame, last_centroid = history[-1]
+            displacement = hypot(
+                last_centroid[0] - first_centroid[0],
+                last_centroid[1] - first_centroid[1],
+            )
+            if displacement >= condition.min_displacement_percent:
+                self.moving_track_last_seen[(label, detection.track_id)] = detection.frame_index
+                return True
+        if label_detections:
+            return False
+        return any(
+            last_seen_frame <= frame_index
+            and frame_index - last_seen_frame <= condition.max_missing_frames
+            for (track_label, _track_id), last_seen_frame in self.moving_track_last_seen.items()
+            if track_label == label
+        )
+
+    def _update_track_history(self, detections: list[Detection]) -> None:
+        for detection in detections:
+            if detection.track_id is None:
+                continue
+            history = self.track_history.setdefault(detection.track_id, [])
+            history.append((detection.frame_index, _centroid_percent(detection)))
+            del history[:-10]
+
     def _mark_cooldowns(self, rule: AutomationRule, timestamp_sec: float) -> None:
         if rule.gate.cooldown:
             self.last_fired[rule.gate.cooldown.key or rule.name] = timestamp_sec
@@ -380,6 +456,11 @@ def _box_gap_percent(left: Detection, right: Detection) -> float:
     return max(gap_x, gap_y) * 100.0
 
 
+def _centroid_percent(detection: Detection) -> tuple[float, float]:
+    x1, y1, x2, y2 = detection.bbox_xyxy_norm
+    return ((x1 + x2) * 50.0, (y1 + y2) * 50.0)
+
+
 def _trigger_edge(*, previous: bool, matched: bool) -> Literal["enter", "exit", "while", "none"]:
     if matched and not previous:
         return "enter"

tiny_trigger/detector.py

@@ -122,7 +122,25 @@ def detections_from_ultralytics_result(
                 bbox_xyxy_norm=_normalize_box(bbox, width, height),
             )
         )
-    return detections
+    return suppress_duplicate_detections(detections)
+
+
+def suppress_duplicate_detections(
+    detections: list[Detection],
+    *,
+    iou_threshold: float = 0.8,
+) -> list[Detection]:
+    """Keep the highest-confidence same-label box for heavily overlapping detections."""
+    kept: list[Detection] = []
+    for detection in sorted(detections, key=lambda item: item.confidence, reverse=True):
+        duplicate = any(
+            _same_label(detection, existing)
+            and _box_iou(detection.bbox_xyxy_norm, existing.bbox_xyxy_norm) >= iou_threshold
+            for existing in kept
+        )
+        if not duplicate:
+            kept.append(detection)
+    return sorted(kept, key=lambda item: (item.frame_index, item.label, item.bbox_xyxy_norm))
 
 
 def _label_from_names(names: Any, class_id: int, fallback_names: list[str]) -> str:
@@ -149,3 +167,26 @@ def _normalize_box(bbox: list[float], width: int, height: int) -> tuple[float, f
 
 def _clamp01(value: float) -> float:
     return max(0.0, min(1.0, value))
+
+
+def _same_label(left: Detection, right: Detection) -> bool:
+    return left.label.strip().lower() == right.label.strip().lower()
+
+
+def _box_iou(
+    left: tuple[float, float, float, float],
+    right: tuple[float, float, float, float],
+) -> float:
+    ax1, ay1, ax2, ay2 = left
+    bx1, by1, bx2, by2 = right
+    intersection_width = max(0.0, min(ax2, bx2) - max(ax1, bx1))
+    intersection_height = max(0.0, min(ay2, by2) - max(ay1, by1))
+    intersection = intersection_width * intersection_height
+    if intersection <= 0:
+        return 0.0
+    left_area = max(0.0, ax2 - ax1) * max(0.0, ay2 - ay1)
+    right_area = max(0.0, bx2 - bx1) * max(0.0, by2 - by1)
+    union = left_area + right_area - intersection
+    if union <= 0:
+        return 0.0
+    return intersection / union

tiny_trigger/llm.py

@@ -12,7 +12,7 @@ from .automation import AutomationDocument, automation_schema
 SYSTEM_PROMPT = """You compile home automation requests into Tiny Trigger rules.
 Return JSON only. Never return code, markdown, explanations, or tool calls.
 The root object MUST include a non-empty "rules" array.
-Use video conditions in when: present, count, near, far.
+Use video conditions in when: present, count, near, far, moving.
 Use state gates in gate: enabled, cooldown.
 Use trigger.on for edge behavior: while, enter, exit, change.
 Use only these action types: simulate, webhook.
@@ -22,8 +22,8 @@ Use trigger.on="while" only when the user explicitly wants repeated actions whil
 When the request says one object is near, next to, beside, at, by, close to, or in front of another object, you MUST emit a near condition.
 Do not replace a near relation with two present conditions.
 Use max_gap_percent for near/far box-edge distance. It is the largest horizontal/vertical edge gap between boxes in normalized frame percent; touching or overlapping boxes have gap 0.
-Do not generate motion, movement, speed, direction, trajectory, tracking, same-object, or identity rules. Tiny Trigger does not yet track identities across frames.
-For requests like "car moving", "person walks", "object moved", or "same car", return a simple presence/near/far approximation only if the request can still be useful without motion; otherwise return JSON with no rules.
+Use moving for simple same-object centroid displacement across sampled frames, such as "car moving" or "person walks". Use min_displacement_percent, default 3, window_frames, minimum/default 3, and max_missing_frames, default 1.
+Do not generate speed, direction, long-gap re-identification, or trajectory path rules.
 If the user mentions elapsed time since an action or limiting repeat fires, encode it as gate.cooldown.
 If the user asks for one action when a condition starts and another action when it stops, use trigger.on="change" and then.enter / then.exit.
 """
@@ -462,6 +462,24 @@ JSON:
   ]
 }}
 
+User: If a car is moving, notify me.
+JSON:
+{{
+  "rules": [
+    {{
+      "name": "car-moving",
+      "when": {{
+        "all": [
+          {{"moving": {{"label": "car", "min_displacement_percent": 3, "window_frames": 3, "max_missing_frames": 1}}}}
+        ]
+      }},
+      "trigger": {{"on": "enter"}},
+      "gate": {{"enabled": true}},
+      "then": [{{"type": "simulate", "name": "notify me"}}]
+    }}
+  ]
+}}
+
 User: If person is near monitor turn on lights. When they leave, turn off lights.
 JSON:
 {{

tiny_trigger/models.py

@@ -12,6 +12,7 @@ class Detection(BaseModel):
     confidence: float = Field(ge=0.0, le=1.0)
     bbox_xyxy: tuple[float, float, float, float]
     bbox_xyxy_norm: tuple[float, float, float, float]
+    track_id: int | None = None
 
 
 class FrameSample(BaseModel):

tiny_trigger/tracking.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from math import hypot
+
+from .models import Detection
+
+
+def _label(value: str) -> str:
+    return value.strip().lower()
+
+
+def _centroid_percent(detection: Detection) -> tuple[float, float]:
+    x1, y1, x2, y2 = detection.bbox_xyxy_norm
+    return ((x1 + x2) * 50.0, (y1 + y2) * 50.0)
+
+
+@dataclass
+class _Track:
+    id: int
+    label: str
+    centroid: tuple[float, float]
+    missed: int = 0
+
+
+class SimpleTracker:
+    """Small label-aware centroid tracker for sampled video detections."""
+
+    def __init__(self, *, max_match_distance_percent: float = 20.0, max_missed: int = 2) -> None:
+        self.max_match_distance_percent = max_match_distance_percent
+        self.max_missed = max_missed
+        self._next_id = 1
+        self._tracks: dict[int, _Track] = {}
+
+    def update(self, detections: list[Detection]) -> list[Detection]:
+        candidates: list[tuple[float, int, int]] = []
+        detection_centroids = [_centroid_percent(detection) for detection in detections]
+        for detection_index, detection in enumerate(detections):
+            label = _label(detection.label)
+            centroid = detection_centroids[detection_index]
+            for track in self._tracks.values():
+                if track.label != label:
+                    continue
+                distance = hypot(centroid[0] - track.centroid[0], centroid[1] - track.centroid[1])
+                if distance <= self.max_match_distance_percent:
+                    candidates.append((distance, detection_index, track.id))
+
+        assigned_detections: set[int] = set()
+        assigned_tracks: set[int] = set()
+        active_tracks: set[int] = set()
+        detection_track_ids: dict[int, int] = {}
+        for _distance, detection_index, track_id in sorted(candidates):
+            if detection_index in assigned_detections or track_id in assigned_tracks:
+                continue
+            assigned_detections.add(detection_index)
+            assigned_tracks.add(track_id)
+            detection_track_ids[detection_index] = track_id
+
+        tracked: list[Detection] = []
+        for detection_index, detection in enumerate(detections):
+            centroid = detection_centroids[detection_index]
+            track_id = detection_track_ids.get(detection_index)
+            if track_id is None:
+                track_id = self._next_id
+                self._next_id += 1
+                self._tracks[track_id] = _Track(
+                    id=track_id,
+                    label=_label(detection.label),
+                    centroid=centroid,
+                )
+            else:
+                track = self._tracks[track_id]
+                track.centroid = centroid
+                track.missed = 0
+            active_tracks.add(track_id)
+            tracked.append(detection.model_copy(update={"track_id": track_id}))
+
+        for track_id, track in list(self._tracks.items()):
+            if track_id in active_tracks:
+                continue
+            track.missed += 1
+            if track.missed > self.max_missed:
+                del self._tracks[track_id]
+
+        return tracked

tiny_trigger/video.py

@@ -8,8 +8,9 @@ from pathlib import Path
 from typing import Callable
 from uuid import uuid4
 
-from .detector import Detector, UltralyticsYOLOEDetector, parse_class_prompt
+from .detector import Detector, UltralyticsYOLOEDetector, parse_class_prompt, suppress_duplicate_detections
 from .models import ActionEvent, Detection, FrameSample, VideoProcessResult
+from .tracking import SimpleTracker
 
 
 ProgressCallback = Callable[[int, int | None], None]
@@ -75,6 +76,7 @@ def process_video(
 
     detections: list[Detection] = []
     frames: list[FrameSample] = []
+    tracker = SimpleTracker()
     processed_frames = 0
     frame_index = -1
     latest_detections: list[Detection] = []
@@ -89,7 +91,7 @@ def process_video(
                     break
                 timestamp_sec = frame_index / source_fps
                 frames.append(FrameSample(frame_index=frame_index, timestamp_sec=timestamp_sec))
-                latest_detections = detector.detect(
+                frame_detections = detector.detect(
                     frame,
                     frame_index=frame_index,
                     timestamp_sec=timestamp_sec,
@@ -97,6 +99,8 @@ def process_video(
                     image_size=image_size,
                     max_detections=max_detections,
                 )
+                frame_detections = suppress_duplicate_detections(frame_detections)
+                latest_detections = tracker.update(frame_detections)
                 detections.extend(latest_detections)
                 processed_frames += 1
                 if progress:
@@ -294,7 +298,8 @@ def _draw_detections(frame, detections: list[Detection]) -> None:
     for detection in detections:
         x1, y1, x2, y2 = [int(value) for value in detection.bbox_xyxy]
         color = _color_for_label(detection.label)
-        label = f"{detection.label} {detection.confidence:.2f}"
+        track = f" #{detection.track_id}" if detection.track_id is not None else ""
+        label = f"{detection.label}{track} {detection.confidence:.2f}"
         cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
         text_y = max(18, y1 - 8)
         cv2.putText(frame, label, (x1, text_y), cv2.FONT_HERSHEY_SIMPLEX, 0.55, color, 2, cv2.LINE_AA)