diff --git a/DeepDataMiningLearning/ngperception/README.md b/DeepDataMiningLearning/ngperception/README.md index 92fbb1e..8a40526 100644 --- a/DeepDataMiningLearning/ngperception/README.md +++ b/DeepDataMiningLearning/ngperception/README.md @@ -11,7 +11,7 @@ see the accuracy trade-offs, and then **fine-tune**. | **Depth / distance** | [depth/](depth/) | ✅ built | MiDaS/DPT → Depth-Anything-V2 → ZoeDepth (metric) | AbsRel / RMSE / δ1 | | **3D occupancy** | [occupancy/](occupancy/) | ✅ baseline | depth→voxel lift (ViPOcc-style) → learned occ nets (planned) | mIoU / geo-IoU (Occ3D) | | Segmentation | `segmentation/` | planned | DeepLabv3 → SegFormer → Mask2Former → OneFormer/SAM2 | mIoU / PQ | -| Tracking (MOT) | `tracking/` | planned | SORT → DeepSORT → ByteTrack → OC-SORT/BoT-SORT | MOTA / IDF1 / HOTA | +| **Tracking (MOT)** | [tracking/](tracking/) | ✅ baseline | **SORT** (motion-only, CPU) → DeepSORT → ByteTrack → OC-SORT/BoT-SORT (planned) | MOTA / IDF1 / ID-switches | | Lane detection | `lane/` | planned | UFLD → CLRNet → CLRerNet | F1 (TuSimple/CULane) | It reuses ngdet's datasets/taxonomy/detectors for **composable fusion** (e.g. detection → diff --git a/DeepDataMiningLearning/ngperception/tracking/README.md b/DeepDataMiningLearning/ngperception/tracking/README.md new file mode 100644 index 0000000..f2fa05e --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/README.md @@ -0,0 +1,109 @@ +# ngperception/tracking — multi-object tracking on top of `ngdet` detections + +The **basic** tier of the tracking task: SORT, the motion-only baseline every later +method is measured against. It runs on **numpy + scipy alone** — no network, no weights, +no GPU, no dataset download — so the whole arm is reproducible by anyone in seconds, +and a later appearance-based backend has a like-for-like reference to beat. + +The module mirrors the three-layer shape the rest of the suite uses +(**trackers / datasets / evaluator**), and consumes `ngdet.Detection` rather than +detecting anything itself — so the same sequence can be re-tracked under a different +detector without touching the tracker, which is what makes the ablation fair. + +## Layout + +``` +tracking/ +├── trackers/base.py # TrackResult + BaseTracker + TRACKER_REGISTRY + build_tracker + iou_matrix +├── trackers/sort.py # @register("sort") — Kalman (cx,cy,area,aspect) + Hungarian on IoU +├── datasets.py # MOTChallenge csv reader (+ a dataset-free synthetic generator) +├── evaluator.py # CLEAR-MOT: MOTA / MOTP / IDF1 / ID-switches, MOTChallenge aggregation +├── run_eval.py # CLI, same shape as depth/run_eval.py +└── tests/ # 26 tests, CPU only, ~1.5 s +``` + +## Quick start — no dataset needed + +```bash +python -m DeepDataMiningLearning.ngperception.tracking.run_eval \ + --trackers sort --synthetic --min-hits 1 +``` + +``` + tracker MOTA MOTP IDF1 IDSW FP FN + sort +1.0000 1.0000 1.0000 0 0 0 +``` + +Dial in the failure you want the tracker to survive: + +| condition | MOTA | MOTP | IDF1 | IDSW | FN | +|---|---:|---:|---:|---:|---:| +| clean | +1.0000 | 1.0000 | 1.0000 | 0 | 0 | +| 6 px detector jitter | +1.0000 | 0.9187 | 1.0000 | 0 | 0 | +| 10 % missed detections | +0.9000 | 1.0000 | 0.9474 | 0 | 24 | +| 15 % missed + 8 px jitter | +0.8583 | 0.8914 | 0.7175 | 0 | 34 | + +*(`--min-hits 1 --max-age 3`, 2 × 30 frames × 4 objects.)* Jitter degrades **MOTP** — +localisation — while leaving MOTA's detection terms alone; missed detections cost +**MOTA** in proportion. That separation is the point of reporting both. + +## On a MOTChallenge-style dataset + +```bash +python -m DeepDataMiningLearning.ngperception.tracking.run_eval \ + --trackers sort --root /data/MOT17 --sequences MOT17-02 MOT17-04 +``` + +`datasets.py` reads `//{gt/gt.txt,det/det.txt}` and does the two conversions +that format needs, in one place, tested: frames are **1-indexed on disk, 0-indexed in the +API**, and boxes are **xywh on disk, xyxy everywhere in `ngdet`/`ngperception`**. Rows +whose `conf` is `0` are MOTChallenge's ignore flag and are dropped by default. + +## Algorithm + +1. a **constant-velocity Kalman filter** per track over `[cx, cy, area, aspect]`, with + aspect held constant — SORT's own simplification, kept so the baseline is the + published one and not a private variant; +2. **Hungarian assignment** on the IoU between each track's *predicted* box and each + detection. Matches below `--track-iou` are rejected **after** the assignment, not + before: filtering first would let a leftover pair win an assignment the optimal + solution had given to a better one. + +## What this baseline does *not* do + +Stated so the numbers are not over-read: + +- **no re-identification** — an object that leaves and returns gets a new id, and the + sequence takes an ID-switch. Recovering the old id is the appearance family's job + (DeepSORT, BoT-SORT); +- **no occlusion reasoning** beyond coasting for `--max-age` frames; +- it inherits every miss of the detector it is given — which is why `run_eval.py` feeds + every backend the *same* detections. + +## Adding a backend + +```python +from .base import BaseTracker, TrackResult, register + +@register("bytetrack") +class ByteTrack(BaseTracker): + family = "bytetrack" + def update(self, detection) -> TrackResult: ... + def reset(self) -> None: ... +``` + +Heavy imports (torch, a re-id model) belong **inside** the subclass `__init__`, never at +module top level, so `import ngperception.tracking` stays cheap and a missing optional +dependency only breaks the backend that needs it — the same rule `ngdet.detectors` uses. + +## Tests + +```bash +python -m pytest DeepDataMiningLearning/ngperception/tracking/tests -q +``` + +26 tests, ~1.5 s, no GPU and no dataset. They cover the IoU edge cases (identical, half, +disjoint, degenerate, empty), id stability and id reuse, survival across a gap shorter +than `max_age` and retirement past it, the index-alignment contract, the MOTChallenge +conversions, and the CLEAR-MOT counters — including that a gap is **not** scored as an +ID switch and that an id swap costs exactly two. diff --git a/DeepDataMiningLearning/ngperception/tracking/__init__.py b/DeepDataMiningLearning/ngperception/tracking/__init__.py new file mode 100644 index 0000000..adf0aa7 --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/__init__.py @@ -0,0 +1 @@ +"""ngperception.tracking — multi-object tracking on top of ngdet detections.""" diff --git a/DeepDataMiningLearning/ngperception/tracking/datasets.py b/DeepDataMiningLearning/ngperception/tracking/datasets.py new file mode 100644 index 0000000..d27d25f --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/datasets.py @@ -0,0 +1,199 @@ +""" +ngperception.tracking.datasets +============================== + +Tracking sequences in a model-agnostic form: a sequence yields, per frame, the +detections a tracker consumes and the ground truth an evaluator scores against. + +The on-disk format is **MOTChallenge**, which every MOT benchmark (MOT16/17/20, +DanceTrack, KITTI-MOT via its converter) either uses or exports to. One CSV row per +object per frame: + + frame, id, bb_left, bb_top, bb_width, bb_height, conf, x, y, z + +Two conversions happen here and both are places bugs hide, so they are done once, in +one place, and tested: + +* **frames are 1-indexed** in the file and 0-indexed in this API; +* **boxes are xywh** (left, top, width, height) in the file and **xyxy** everywhere in + `ngdet`/`ngperception`, so ``x2 = left + width`` and ``y2 = top + height`` — + *not* ``left + width - 1``. The MOTChallenge devkit treats the box as a continuous + rectangle, not an inclusive pixel range, and an off-by-one here shifts every IoU. + +`gt.txt` additionally uses `conf` as a **flag**: 0 marks an ignored/distractor box that +must not be scored. Those rows are dropped by default (`keep_ignored=False`), because +counting them would turn correct behaviour into false positives. + +No dataset ships with this repository. `synthetic_sequence()` produces a sequence with +the same structure so the module, the tracker and the evaluator can all be exercised +without downloading anything — that is what the self-test below and the unit tests use. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from typing import Dict, Iterator, List, Optional, Tuple + +import numpy as np + +DEFAULT_MOT_ROOT = "/mnt/e/Shared/Dataset/MOT17/" + + +@dataclass +class TrackingFrame: + """One frame of a sequence. All arrays are index-aligned. + + `det_boxes` is what the tracker sees; `gt_boxes`/`gt_ids` are what the evaluator + scores against. A frame may have either side empty. + """ + + frame_id: int + det_boxes: np.ndarray = field(default_factory=lambda: np.zeros((0, 4), np.float32)) + det_scores: np.ndarray = field(default_factory=lambda: np.zeros((0,), np.float32)) + gt_boxes: np.ndarray = field(default_factory=lambda: np.zeros((0, 4), np.float32)) + gt_ids: np.ndarray = field(default_factory=lambda: np.zeros((0,), np.int64)) + + +def xywh_to_xyxy(boxes: np.ndarray) -> np.ndarray: + """MOTChallenge (left, top, w, h) -> (x1, y1, x2, y2). Width is a length, not a span.""" + b = np.asarray(boxes, np.float32).reshape(-1, 4) + out = np.empty_like(b) + out[:, 0] = b[:, 0] + out[:, 1] = b[:, 1] + out[:, 2] = b[:, 0] + b[:, 2] + out[:, 3] = b[:, 1] + b[:, 3] + return out + + +def read_mot_csv(path: str, keep_ignored: bool = False + ) -> Dict[int, Tuple[np.ndarray, np.ndarray, np.ndarray]]: + """Read a MOTChallenge csv -> {frame_index: (boxes_xyxy, ids, confs)}. + + Frame indices in the returned dict are **0-based**. Rows whose `conf` is 0 are + dropped unless `keep_ignored` is set. + """ + per_frame: Dict[int, List[Tuple[float, ...]]] = {} + with open(path, "r", encoding="utf-8") as fh: + for raw in fh: + raw = raw.strip() + if not raw or raw.startswith("#"): + continue + parts = raw.split(",") + if len(parts) < 7: + raise ValueError(f"{path}: expected >=7 columns, got {len(parts)}: {raw!r}") + frame = int(float(parts[0])) - 1 # file is 1-indexed + oid = int(float(parts[1])) + l, t, w, h = (float(x) for x in parts[2:6]) + conf = float(parts[6]) + if not keep_ignored and conf == 0.0: + continue + per_frame.setdefault(frame, []).append((l, t, w, h, conf, oid)) + + out: Dict[int, Tuple[np.ndarray, np.ndarray, np.ndarray]] = {} + for frame, rows in per_frame.items(): + arr = np.array([r[:4] for r in rows], np.float32) + out[frame] = (xywh_to_xyxy(arr), + np.array([r[5] for r in rows], np.int64), + np.array([r[4] for r in rows], np.float32)) + return out + + +class MOTSequence: + """One MOTChallenge sequence directory: `//{gt/gt.txt,det/det.txt}`. + + Iterating yields `TrackingFrame` in frame order, including frames where one side + is empty — a tracker must see those to age its tracks correctly. + """ + + def __init__(self, root: str = DEFAULT_MOT_ROOT, name: str = "MOT17-02", + keep_ignored: bool = False): + self.root, self.name = root, name + seq_dir = os.path.join(root, name) + gt_path = os.path.join(seq_dir, "gt", "gt.txt") + det_path = os.path.join(seq_dir, "det", "det.txt") + if not os.path.isdir(seq_dir): + raise FileNotFoundError( + f"Sequence directory not found: {seq_dir}. Point --root at a " + "MOTChallenge-style dataset, or use synthetic_sequence() for a " + "dataset-free run.") + self._gt = read_mot_csv(gt_path, keep_ignored) if os.path.exists(gt_path) else {} + self._det = read_mot_csv(det_path, keep_ignored=True) if os.path.exists(det_path) else {} + frames = set(self._gt) | set(self._det) + self._frames = sorted(frames) + + def __len__(self) -> int: + return len(self._frames) + + def __iter__(self) -> Iterator[TrackingFrame]: + for f in self._frames: + gb, gi, _ = self._gt.get(f, (np.zeros((0, 4), np.float32), + np.zeros((0,), np.int64), + np.zeros((0,), np.float32))) + db, _, dc = self._det.get(f, (np.zeros((0, 4), np.float32), + np.zeros((0,), np.int64), + np.zeros((0,), np.float32))) + yield TrackingFrame(frame_id=f, det_boxes=db, det_scores=dc, + gt_boxes=gb, gt_ids=gi) + + +def synthetic_sequence(n_frames: int = 20, n_objects: int = 3, miss_rate: float = 0.0, + jitter: float = 0.0, seed: int = 0) -> List[TrackingFrame]: + """A dataset-free sequence with known ground truth. + + Objects move linearly. `miss_rate` drops detections (never ground truth), and + `jitter` perturbs detection corners in pixels — so a caller can dial in exactly + the failure the tracker is supposed to survive. + """ + rng = np.random.default_rng(seed) + starts = np.array([[40 + 130 * i, 40 + 60 * i, 100 + 130 * i, 120 + 60 * i] + for i in range(n_objects)], np.float32) + vels = np.array([[7 - 3 * i, 2 + i, 7 - 3 * i, 2 + i] for i in range(n_objects)], + np.float32) + + frames: List[TrackingFrame] = [] + for f in range(n_frames): + gt = starts + vels * f + keep = rng.random(n_objects) >= miss_rate + det = gt[keep].copy() + if jitter and len(det): + det += rng.uniform(-jitter, jitter, det.shape).astype(np.float32) + frames.append(TrackingFrame( + frame_id=f, + det_boxes=det.astype(np.float32), + det_scores=np.ones(len(det), np.float32), + gt_boxes=gt.astype(np.float32), + gt_ids=np.arange(1, n_objects + 1, dtype=np.int64), + )) + return frames + + +# =========================================================================== +# HOW TO TEST / RUN THIS FILE +# python -m DeepDataMiningLearning.ngperception.tracking.datasets +# Round-trips a MOTChallenge csv through the reader and prints a synthetic sequence. +# =========================================================================== +if __name__ == "__main__": + import tempfile + + csv = ("1,1,10,20,30,40,1,-1,-1,-1\n" + "1,2,100,100,50,50,1,-1,-1,-1\n" + "2,1,12,22,30,40,1,-1,-1,-1\n" + "2,9,500,500,10,10,0,-1,-1,-1\n") # conf=0 -> ignored + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "gt.txt") + with open(p, "w", encoding="utf-8") as fh: + fh.write(csv) + parsed = read_mot_csv(p) + + print("frames parsed:", sorted(parsed)) # 0-based -> [0, 1] + boxes, ids, _ = parsed[0] + print("frame 0 ids:", ids.tolist(), "boxes:", boxes.tolist()) + assert sorted(parsed) == [0, 1], "frames must be 0-based" + assert boxes[0].tolist() == [10.0, 20.0, 40.0, 60.0], boxes[0].tolist() + assert 9 not in parsed[1][1].tolist(), "conf=0 row must be dropped" + + seq = synthetic_sequence(n_frames=4, n_objects=2) + for fr in seq: + print(f" frame {fr.frame_id}: {len(fr.det_boxes)} det, {len(fr.gt_boxes)} gt") + print("OK") diff --git a/DeepDataMiningLearning/ngperception/tracking/evaluator.py b/DeepDataMiningLearning/ngperception/tracking/evaluator.py new file mode 100644 index 0000000..ef4a300 --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/evaluator.py @@ -0,0 +1,187 @@ +""" +ngperception.tracking.evaluator +=============================== + +CLEAR-MOT metrics (Bernardin & Stiefelhagen, 2008) plus IDF1 (Ristani et al., ECCV 2016 +workshop) — the standard MOT scoring set, computed per sequence and aggregated the way +MOTChallenge does it: **sum the counts over all frames and sequences, then form the +ratio once**. Averaging per-frame MOTA is a different (and wrong) number, so the +counters are public and the ratios are derived only in `summarize()`. + +Definitions, spelled out because the sign conventions differ between papers: + + MOTA = 1 - (FN + FP + IDSW) / GT [higher better; can be negative] + MOTP = mean IoU over matched pairs [higher better, IoU convention] + IDF1 = 2*IDTP / (2*IDTP + IDFP + IDFN) [higher better] + Recall = TP / GT + Precision = TP / (TP + FP) + +Matching is per frame, by Hungarian assignment on IoU, with pairs below +`iou_threshold` rejected after the assignment (identical to the association rule in +`trackers/base.py`, so the evaluator does not silently use a stricter or looser gate +than the tracker it scores). + +An **ID switch** is counted when a ground-truth object that was previously matched to +tracker id *a* is matched to a different id *b*. Following CLEAR-MOT, the previous +association survives a gap in which the object is unmatched — otherwise every +occlusion would be scored as a switch. + +IDF1 is computed globally over the sequence, not per frame: the optimal one-to-one +mapping between ground-truth ids and tracker ids is found by Hungarian assignment on +the number of frames in which each pair co-occurs, and IDTP is the total of the +matched cells. A per-frame approximation would flatter a tracker that swaps ids. +""" + +from __future__ import annotations + +from typing import Dict, List, Optional, Sequence, Tuple + +import numpy as np + +from .trackers.base import iou_matrix + + +def _match(gt_boxes: np.ndarray, tr_boxes: np.ndarray, iou_threshold: float + ) -> Tuple[List[Tuple[int, int]], np.ndarray]: + """Hungarian match on IoU, thresholded after assignment. Returns (pairs, iou).""" + if len(gt_boxes) == 0 or len(tr_boxes) == 0: + return [], np.zeros((len(gt_boxes), len(tr_boxes)), np.float32) + + from scipy.optimize import linear_sum_assignment + + iou = iou_matrix(gt_boxes, tr_boxes) + rows, cols = linear_sum_assignment(-iou) + pairs = [(int(r), int(c)) for r, c in zip(rows, cols) + if iou[r, c] >= iou_threshold] + return pairs, iou + + +class MOTEvaluator: + """Accumulates CLEAR-MOT counters across frames and sequences. + + Usage + ----- + ev = MOTEvaluator() + ev.new_sequence() # resets the id-association memory + for frame in sequence: + ev.add(gt_ids, gt_boxes, track_ids, track_boxes) + ev.summarize() + """ + + def __init__(self, iou_threshold: float = 0.5): + self.iou_threshold = float(iou_threshold) + self.gt = 0 + self.tp = 0 + self.fp = 0 + self.fn = 0 + self.idsw = 0 + self._iou_sum = 0.0 + self._last_match: Dict[int, int] = {} + self._cooccur: Dict[Tuple[int, int], int] = {} + self._gt_count: Dict[int, int] = {} + self._tr_count: Dict[int, int] = {} + + def new_sequence(self) -> None: + """Start a new sequence: ids are only comparable within one.""" + self._last_match = {} + + def add(self, gt_ids: Sequence[int], gt_boxes: np.ndarray, + track_ids: Sequence[int], track_boxes: np.ndarray) -> None: + gt_ids = list(gt_ids) + track_ids = list(track_ids) + gt_boxes = np.asarray(gt_boxes, np.float32).reshape(-1, 4) + track_boxes = np.asarray(track_boxes, np.float32).reshape(-1, 4) + if len(gt_ids) != len(gt_boxes) or len(track_ids) != len(track_boxes): + raise ValueError("ids and boxes must be index-aligned") + + pairs, iou = _match(gt_boxes, track_boxes, self.iou_threshold) + + self.gt += len(gt_boxes) + self.tp += len(pairs) + self.fp += len(track_boxes) - len(pairs) + self.fn += len(gt_boxes) - len(pairs) + + for g, t in pairs: + gid, tid = gt_ids[g], track_ids[t] + self._iou_sum += float(iou[g, t]) + prev = self._last_match.get(gid) + if prev is not None and prev != tid: + self.idsw += 1 + self._last_match[gid] = tid + self._cooccur[(gid, tid)] = self._cooccur.get((gid, tid), 0) + 1 + + for gid in gt_ids: + self._gt_count[gid] = self._gt_count.get(gid, 0) + 1 + for tid in track_ids: + self._tr_count[tid] = self._tr_count.get(tid, 0) + 1 + + # -- IDF1 --------------------------------------------------------------- + def _idf1(self) -> Tuple[float, int]: + """Global one-to-one id mapping; returns (idf1, idtp).""" + if not self._cooccur: + return 0.0, 0 + from scipy.optimize import linear_sum_assignment + + gids = sorted({g for g, _ in self._cooccur}) + tids = sorted({t for _, t in self._cooccur}) + m = np.zeros((len(gids), len(tids)), np.float64) + for (g, t), c in self._cooccur.items(): + m[gids.index(g), tids.index(t)] = c + + rows, cols = linear_sum_assignment(-m) + idtp = int(m[rows, cols].sum()) + idfn = sum(self._gt_count.values()) - idtp + idfp = sum(self._tr_count.values()) - idtp + denom = 2 * idtp + idfp + idfn + return (2.0 * idtp / denom if denom else 0.0), idtp + + def summarize(self, verbose: bool = True) -> Dict[str, float]: + idf1, idtp = self._idf1() + out = { + "MOTA": 1.0 - (self.fn + self.fp + self.idsw) / self.gt if self.gt else 0.0, + "MOTP": self._iou_sum / self.tp if self.tp else 0.0, + "IDF1": idf1, + "recall": self.tp / self.gt if self.gt else 0.0, + "precision": self.tp / (self.tp + self.fp) if (self.tp + self.fp) else 0.0, + "GT": float(self.gt), "TP": float(self.tp), "FP": float(self.fp), + "FN": float(self.fn), "IDSW": float(self.idsw), "IDTP": float(idtp), + } + if verbose: + print(f" MOTA {out['MOTA']:+.4f} MOTP {out['MOTP']:.4f} IDF1 {out['IDF1']:.4f}" + f" | GT {self.gt} TP {self.tp} FP {self.fp} FN {self.fn}" + f" IDSW {self.idsw}") + return out + + +# =========================================================================== +# HOW TO TEST / RUN THIS FILE +# python -m DeepDataMiningLearning.ngperception.tracking.evaluator +# A perfect tracker must score MOTA 1.0 / IDF1 1.0 / IDSW 0; swapping two ids +# midway must cost exactly two switches and drop IDF1 without touching MOTA's +# detection terms. +# =========================================================================== +if __name__ == "__main__": + boxes_a = np.array([[0, 0, 10, 10]], np.float32) + boxes_b = np.array([[50, 50, 60, 60]], np.float32) + frames = [(np.concatenate([boxes_a, boxes_b]), [1, 2]) for _ in range(6)] + + ev = MOTEvaluator() + ev.new_sequence() + for boxes, gids in frames: + ev.add(gids, boxes, gids, boxes) # tracker == ground truth + print("perfect tracker:") + perfect = ev.summarize() + + ev2 = MOTEvaluator() + ev2.new_sequence() + for i, (boxes, gids) in enumerate(frames): + tids = gids if i < 3 else [2, 1] # ids swapped from frame 3 + ev2.add(gids, boxes, tids, boxes) + print("ids swapped at frame 3:") + swapped = ev2.summarize() + + assert perfect["MOTA"] == 1.0 and perfect["IDF1"] == 1.0 and perfect["IDSW"] == 0 + assert swapped["IDSW"] == 2, swapped["IDSW"] + assert swapped["FP"] == 0 and swapped["FN"] == 0 + assert swapped["IDF1"] < perfect["IDF1"] + print("OK") diff --git a/DeepDataMiningLearning/ngperception/tracking/run_eval.py b/DeepDataMiningLearning/ngperception/tracking/run_eval.py new file mode 100644 index 0000000..a2a8ec0 --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/run_eval.py @@ -0,0 +1,155 @@ +""" +ngperception.tracking.run_eval +============================== + +Compare tracker backends on the same sequences under identical settings — the +tracking analogue of `ngperception/depth/run_eval.py`. + +Every backend sees the **same detections**, so a difference in MOTA/IDF1 is a +difference in association, not in detection quality. That is the only comparison the +suite claims to make; swapping in a stronger detector moves every row at once. + +Examples +-------- + # dataset-free smoke run: synthetic sequences, no download, seconds + python -m DeepDataMiningLearning.ngperception.tracking.run_eval \\ + --trackers sort --synthetic + + # sweep the association gate on synthetic data with detector noise + python -m DeepDataMiningLearning.ngperception.tracking.run_eval \\ + --trackers sort --synthetic --jitter 6 --miss-rate 0.1 + + # MOTChallenge-style directory + python -m DeepDataMiningLearning.ngperception.tracking.run_eval \\ + --trackers sort --root /data/MOT17 --sequences MOT17-02 MOT17-04 +""" + +from __future__ import annotations + +import argparse +import json +import os +from typing import Dict, List + +import numpy as np + +from DeepDataMiningLearning.ngdet.detectors.base import Detection + +from .datasets import DEFAULT_MOT_ROOT, MOTSequence, TrackingFrame, synthetic_sequence +from .evaluator import MOTEvaluator +from .trackers.base import build_tracker + +DEFAULT_OUT_DIR = "DeepDataMiningLearning/ngperception/output/tracking" + + +def _as_detection(frame: TrackingFrame) -> Detection: + n = len(frame.det_boxes) + return Detection( + boxes=frame.det_boxes, + scores=(frame.det_scores if len(frame.det_scores) == n + else np.ones(n, np.float32)), + labels=np.zeros(n, np.int64), + names=["object"] * n, + ) + + +def evaluate(tracker_spec: str, sequences: Dict[str, List[TrackingFrame]], + eval_iou: float, **tracker_kwargs) -> Dict[str, float]: + """Run one backend over every sequence and return the aggregated metrics. + + Counters accumulate across sequences and the ratios are formed once at the end, + which is the MOTChallenge convention -- averaging per-sequence MOTA weights a + 10-frame clip the same as a 1000-frame one. + + `eval_iou` is the *scoring* gate; the tracker's own association gate arrives in + `tracker_kwargs` as `iou_threshold`. They are deliberately different names: the + two thresholds mean different things and calling both `iou_threshold` collides. + """ + tracker = build_tracker(tracker_spec, **tracker_kwargs) + ev = MOTEvaluator(iou_threshold=eval_iou) + + for _, frames in sequences.items(): + tracker.reset() + ev.new_sequence() + for fr in frames: + res = tracker.update(_as_detection(fr)) + ev.add(fr.gt_ids.tolist(), fr.gt_boxes, + res.track_ids.tolist(), res.boxes) + return ev.summarize(verbose=False) + + +def main() -> None: + ap = argparse.ArgumentParser(description="ngperception tracking comparison.") + ap.add_argument("--trackers", nargs="+", default=["sort"], + help='backend specs, e.g. "sort" (see TRACKER_REGISTRY)') + ap.add_argument("--root", default=DEFAULT_MOT_ROOT) + ap.add_argument("--sequences", nargs="*", default=None, + help="sequence directory names under --root") + ap.add_argument("--synthetic", action="store_true", + help="ignore --root and score on generated sequences (no dataset)") + ap.add_argument("--synthetic-frames", type=int, default=30) + ap.add_argument("--synthetic-objects", type=int, default=4) + ap.add_argument("--miss-rate", type=float, default=0.0, + help="synthetic only: fraction of detections dropped per frame") + ap.add_argument("--jitter", type=float, default=0.0, + help="synthetic only: uniform pixel noise on detection corners") + ap.add_argument("--seed", type=int, default=0) + ap.add_argument("--iou-threshold", type=float, default=0.5, + help="evaluation gate (MOTChallenge uses 0.5)") + ap.add_argument("--track-iou", type=float, default=0.3, + help="association gate inside the tracker") + ap.add_argument("--max-age", type=int, default=1) + ap.add_argument("--min-hits", type=int, default=3) + ap.add_argument("--out-dir", default=DEFAULT_OUT_DIR) + args = ap.parse_args() + + if args.synthetic: + sequences = { + f"synthetic-{i}": synthetic_sequence( + n_frames=args.synthetic_frames, n_objects=args.synthetic_objects, + miss_rate=args.miss_rate, jitter=args.jitter, seed=args.seed + i) + for i in range(2) + } + source = (f"synthetic ({args.synthetic_frames} frames x " + f"{args.synthetic_objects} objects, miss={args.miss_rate}, " + f"jitter={args.jitter}, seed={args.seed})") + else: + names = args.sequences or [] + if not names: + ap.error("give --sequences NAME [NAME ...] or use --synthetic") + sequences = {n: list(MOTSequence(args.root, n)) for n in names} + source = f"{args.root} ({', '.join(names)})" + + n_frames = sum(len(v) for v in sequences.values()) + print(f"source : {source}") + print(f"frames : {n_frames} across {len(sequences)} sequence(s)") + print(f"eval : IoU>={args.iou_threshold} tracker: IoU>={args.track_iou} " + f"max_age={args.max_age} min_hits={args.min_hits}") + print() + print(f" {'tracker':<12s} {'MOTA':>9s} {'MOTP':>7s} {'IDF1':>7s} " + f"{'IDSW':>6s} {'FP':>7s} {'FN':>7s}") + + results: Dict[str, Dict[str, float]] = {} + for spec in args.trackers: + m = evaluate(spec, sequences, args.iou_threshold, + iou_threshold=args.track_iou, max_age=args.max_age, + min_hits=args.min_hits) + results[spec] = m + print(f" {spec:<12s} {m['MOTA']:>+9.4f} {m['MOTP']:>7.4f} {m['IDF1']:>7.4f} " + f"{int(m['IDSW']):>6d} {int(m['FP']):>7d} {int(m['FN']):>7d}") + + os.makedirs(args.out_dir, exist_ok=True) + out_path = os.path.join(args.out_dir, "tracking_metrics.json") + with open(out_path, "w", encoding="utf-8") as fh: + json.dump({"source": source, "settings": vars(args), "results": results}, + fh, indent=2) + print(f"\nwrote {out_path}") + + +# =========================================================================== +# HOW TO TEST / RUN THIS FILE +# python -m DeepDataMiningLearning.ngperception.tracking.run_eval --trackers sort --synthetic +# Expected: a clean synthetic sequence scores MOTA close to 1.0 with 0 ID switches. +# =========================================================================== +if __name__ == "__main__": + main() diff --git a/DeepDataMiningLearning/ngperception/tracking/tests/__init__.py b/DeepDataMiningLearning/ngperception/tracking/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/DeepDataMiningLearning/ngperception/tracking/tests/test_tracking.py b/DeepDataMiningLearning/ngperception/tracking/tests/test_tracking.py new file mode 100644 index 0000000..28e1e01 --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/tests/test_tracking.py @@ -0,0 +1,430 @@ +"""Tests for ngperception.tracking — CPU only, no model, no dataset, seconds to run.""" + +from __future__ import annotations + +import numpy as np +import pytest + +from DeepDataMiningLearning.ngdet.detectors.base import Detection +from DeepDataMiningLearning.ngperception.tracking.evaluator import MOTEvaluator +from DeepDataMiningLearning.ngperception.tracking.trackers import sort as _sort # noqa: F401 +from DeepDataMiningLearning.ngperception.tracking.trackers.base import ( + TRACKER_REGISTRY, TrackResult, build_tracker, iou_matrix, +) + + +def det(boxes, scores=None, labels=None, names=None): + b = np.asarray(boxes, np.float32).reshape(-1, 4) + n = len(b) + return Detection( + boxes=b, + scores=np.asarray(scores if scores is not None else [1.0] * n, np.float32), + labels=np.asarray(labels if labels is not None else [0] * n, np.int64), + names=list(names) if names is not None else ["car"] * n, + ) + + +# ------------------------------------------------------------------ geometry +def test_iou_identical_half_and_disjoint(): + a = np.array([[0, 0, 10, 10]], np.float32) + b = np.array([[0, 0, 10, 10], [5, 0, 15, 10], [20, 20, 30, 30]], np.float32) + got = iou_matrix(a, b)[0] + assert got[0] == pytest.approx(1.0) + assert got[1] == pytest.approx(1 / 3, rel=1e-4) # 50 inter / 150 union + assert got[2] == pytest.approx(0.0) + + +def test_iou_degenerate_box_is_zero_not_nan(): + a = np.array([[0, 0, 10, 0]], np.float32) # zero height + b = np.array([[0, 0, 10, 10]], np.float32) + got = iou_matrix(a, b) + assert np.isfinite(got).all() and got[0, 0] == 0.0 + + +def test_iou_empty_inputs_keep_their_shape(): + assert iou_matrix(np.zeros((0, 4)), np.zeros((3, 4))).shape == (0, 3) + assert iou_matrix(np.zeros((2, 4)), np.zeros((0, 4))).shape == (2, 0) + + +# ------------------------------------------------------------------ registry +def test_sort_is_registered_and_buildable(): + assert "sort" in TRACKER_REGISTRY + t = build_tracker("sort", iou_threshold=0.25, max_age=5) + assert t.iou_threshold == 0.25 and t.max_age == 5 + + +def test_unknown_backend_names_the_registered_ones(): + with pytest.raises(KeyError, match="sort"): + build_tracker("deepsort") + + +# ------------------------------------------------------------------ identity +def test_ids_are_stable_while_objects_move(): + t = build_tracker("sort", min_hits=1) + ids_per_frame = [] + for f in range(8): + ids_per_frame.append(t.update(det([ + [10 + 12 * f, 20, 60 + 12 * f, 80], + [200 - 6 * f, 100, 250 - 6 * f, 160], + ])).track_ids.tolist()) + assert all(len(ids) == 2 for ids in ids_per_frame) + assert len({tuple(ids) for ids in ids_per_frame}) == 1, ids_per_frame + + +def test_new_object_gets_a_fresh_id_and_keeps_it(): + t = build_tracker("sort", min_hits=1) + seen = [] + for f in range(6): + boxes = [[10 + 12 * f, 20, 60 + 12 * f, 80]] + if f >= 3: + boxes.append([300, 300 + 5 * (f - 3), 340, 350 + 5 * (f - 3)]) + seen.append(t.update(det(boxes)).track_ids.tolist()) + assert len(seen[2]) == 1 and len(seen[3]) == 2 + newcomer = seen[3][1] + assert newcomer not in seen[2] + assert all(newcomer in ids for ids in seen[3:]) + + +def test_track_survives_a_gap_shorter_than_max_age(): + """A miss inside `max_age` must not end the track, and the id must survive it. + + Without this the suite passes even when `max_age` is ignored entirely -- a + mutation that retires every unmatched track immediately went undetected until + this case was added. + """ + t = build_tracker("sort", min_hits=1, max_age=3) + first = t.update(det([[0, 0, 20, 20]])).track_ids.tolist() + assert first == [1] + for _ in range(2): # two missed frames < max_age + assert len(t.update(det(np.zeros((0, 4), np.float32)))) == 0 + again = t.update(det([[2, 2, 22, 22]])).track_ids.tolist() + assert again == first, f"track should have survived the gap, got {again}" + + +def test_track_dies_after_a_gap_longer_than_max_age(): + t = build_tracker("sort", min_hits=1, max_age=2) + first = t.update(det([[0, 0, 20, 20]])).track_ids.tolist() + for _ in range(4): # longer than max_age + t.update(det(np.zeros((0, 4), np.float32))) + again = t.update(det([[0, 0, 20, 20]])).track_ids.tolist() + assert again != first, "track should have been retired past max_age" + + +def test_ids_are_never_reused_after_a_track_dies(): + t = build_tracker("sort", min_hits=1, max_age=1) + first = t.update(det([[0, 0, 20, 20]])).track_ids.tolist() + for _ in range(4): # object disappears + t.update(det(np.zeros((0, 4), np.float32))) + again = t.update(det([[0, 0, 20, 20]])).track_ids.tolist() + assert again and again[0] not in first + + +def test_reset_restarts_the_sequence(): + t = build_tracker("sort", min_hits=1) + a = t.update(det([[0, 0, 20, 20]])).track_ids.tolist() + t.reset() + b = t.update(det([[0, 0, 20, 20]])).track_ids.tolist() + assert a == b == [1] + + +# ------------------------------------------------------------------ contract +def test_track_result_stays_index_aligned(): + t = build_tracker("sort", min_hits=1) + res = t.update(det([[0, 0, 10, 10], [50, 50, 70, 70]], + scores=[0.9, 0.4], labels=[3, 7], names=["car", "bus"])) + res.check_aligned() + assert len(res) == 2 + assert set(res.labels.tolist()) == {3, 7} + assert set(res.names) == {"car", "bus"} + + +def test_check_aligned_catches_a_broken_result(): + bad = TrackResult(track_ids=np.array([1, 2]), + boxes=np.zeros((2, 4), np.float32), + scores=np.array([1.0], np.float32), # one short + labels=np.zeros(2, np.int64), names=["a", "b"]) + with pytest.raises(ValueError, match="misaligned"): + bad.check_aligned() + + +def test_empty_detection_yields_empty_result(): + t = build_tracker("sort", min_hits=1) + res = t.update(det(np.zeros((0, 4), np.float32))) + assert len(res) == 0 + res.check_aligned() + + +# ------------------------------------------------------------------ evaluator +def test_perfect_tracker_scores_one(): + boxes = np.array([[0, 0, 10, 10], [50, 50, 60, 60]], np.float32) + ev = MOTEvaluator() + ev.new_sequence() + for _ in range(6): + ev.add([1, 2], boxes, [1, 2], boxes) + s = ev.summarize(verbose=False) + assert s["MOTA"] == pytest.approx(1.0) + assert s["IDF1"] == pytest.approx(1.0) + assert s["IDSW"] == 0 and s["FP"] == 0 and s["FN"] == 0 + + +def test_id_swap_costs_exactly_two_switches(): + boxes = np.array([[0, 0, 10, 10], [50, 50, 60, 60]], np.float32) + ev = MOTEvaluator() + ev.new_sequence() + for i in range(6): + ev.add([1, 2], boxes, [1, 2] if i < 3 else [2, 1], boxes) + s = ev.summarize(verbose=False) + assert s["IDSW"] == 2 + assert s["FP"] == 0 and s["FN"] == 0 # detection is untouched + assert s["MOTA"] == pytest.approx(1 - 2 / 12) + assert s["IDF1"] < 1.0 + + +def test_a_gap_is_not_scored_as_an_id_switch(): + """CLEAR-MOT keeps the association across frames where the object is unmatched.""" + box = np.array([[0, 0, 10, 10]], np.float32) + ev = MOTEvaluator() + ev.new_sequence() + ev.add([1], box, [7], box) + ev.add([1], box, [], np.zeros((0, 4), np.float32)) # missed this frame + ev.add([1], box, [7], box) # same id returns + s = ev.summarize(verbose=False) + assert s["IDSW"] == 0 + assert s["FN"] == 1 + + +def test_false_positive_and_miss_are_counted_separately(): + gt = np.array([[0, 0, 10, 10]], np.float32) + tr = np.array([[900, 900, 910, 910]], np.float32) # nowhere near + ev = MOTEvaluator() + ev.new_sequence() + ev.add([1], gt, [1], tr) + s = ev.summarize(verbose=False) + assert s["FP"] == 1 and s["FN"] == 1 and s["TP"] == 0 + assert s["MOTA"] == pytest.approx(-1.0) # MOTA may go negative + + +# ------------------------------------------------------------------ end to end +def test_sort_tracks_a_clean_sequence_at_mota_one(): + """Feed the tracker its own ground truth; it must reproduce it exactly.""" + t = build_tracker("sort", min_hits=1) + ev = MOTEvaluator() + ev.new_sequence() + for f in range(10): + boxes = np.array([[10 + 8 * f, 20, 60 + 8 * f, 80], + [400 - 7 * f, 200, 460 - 7 * f, 260]], np.float32) + res = t.update(det(boxes)) + ev.add([1, 2], boxes, res.track_ids.tolist(), res.boxes) + s = ev.summarize(verbose=False) + assert s["IDSW"] == 0 + assert s["MOTA"] > 0.99, s + assert s["IDF1"] > 0.99, s + + +# ------------------------------------------------------------------ datasets +def test_xywh_to_xyxy_treats_width_as_a_length(): + from DeepDataMiningLearning.ngperception.tracking.datasets import xywh_to_xyxy + got = xywh_to_xyxy([[10, 20, 30, 40]])[0].tolist() + assert got == [10.0, 20.0, 40.0, 60.0] # not 39/59: w is a length, not a span + + +def test_mot_csv_is_one_indexed_on_disk_and_zero_indexed_in_the_api(tmp_path): + from DeepDataMiningLearning.ngperception.tracking.datasets import read_mot_csv + p = tmp_path / "gt.txt" + p.write_text("1,1,10,20,30,40,1,-1,-1,-1\n" + "2,1,12,22,30,40,1,-1,-1,-1\n", encoding="utf-8") + parsed = read_mot_csv(str(p)) + assert sorted(parsed) == [0, 1] + + +def test_mot_csv_drops_ignored_rows_unless_asked(tmp_path): + from DeepDataMiningLearning.ngperception.tracking.datasets import read_mot_csv + p = tmp_path / "gt.txt" + p.write_text("1,1,10,20,30,40,1,-1,-1,-1\n" + "1,9,500,500,10,10,0,-1,-1,-1\n", encoding="utf-8") # conf=0 + assert read_mot_csv(str(p))[0][1].tolist() == [1] + assert sorted(read_mot_csv(str(p), keep_ignored=True)[0][1].tolist()) == [1, 9] + + +def test_mot_csv_rejects_a_short_row(tmp_path): + from DeepDataMiningLearning.ngperception.tracking.datasets import read_mot_csv + p = tmp_path / "gt.txt" + p.write_text("1,1,10,20\n", encoding="utf-8") + with pytest.raises(ValueError, match="columns"): + read_mot_csv(str(p)) + + +def test_missing_sequence_directory_says_what_to_do(tmp_path): + from DeepDataMiningLearning.ngperception.tracking.datasets import MOTSequence + with pytest.raises(FileNotFoundError, match="synthetic_sequence"): + MOTSequence(root=str(tmp_path), name="NOPE-01") + + +def test_synthetic_sequence_is_reproducible_and_shaped(): + from DeepDataMiningLearning.ngperception.tracking.datasets import synthetic_sequence + a = synthetic_sequence(n_frames=5, n_objects=3, jitter=4, seed=7) + b = synthetic_sequence(n_frames=5, n_objects=3, jitter=4, seed=7) + assert len(a) == 5 + for fa, fb in zip(a, b): + assert np.array_equal(fa.det_boxes, fb.det_boxes) + assert len(fa.gt_boxes) == 3 and len(fa.gt_ids) == 3 + + +def test_miss_rate_drops_detections_but_never_ground_truth(): + from DeepDataMiningLearning.ngperception.tracking.datasets import synthetic_sequence + frames = synthetic_sequence(n_frames=40, n_objects=4, miss_rate=0.5, seed=3) + assert all(len(f.gt_boxes) == 4 for f in frames) + total_det = sum(len(f.det_boxes) for f in frames) + assert 0 < total_det < 40 * 4 + + +# ------------------------------------------------------ MOTSequence end to end +def _write_seq(root, name, gt_rows, det_rows): + import os + d = root / name + (d / "gt").mkdir(parents=True) + (d / "det").mkdir(parents=True) + (d / "gt" / "gt.txt").write_text("".join(gt_rows), encoding="utf-8") + (d / "det" / "det.txt").write_text("".join(det_rows), encoding="utf-8") + return str(root) + + +def test_mot_sequence_reads_gt_and_det_in_frame_order(tmp_path): + from DeepDataMiningLearning.ngperception.tracking.datasets import MOTSequence + root = _write_seq( + tmp_path, "SEQ-01", + ["1,1,10,20,30,40,1,-1,-1,-1\n", "2,1,12,22,30,40,1,-1,-1,-1\n", + "3,1,14,24,30,40,1,-1,-1,-1\n"], + ["1,-1,11,21,30,40,0.9,-1,-1,-1\n", "3,-1,15,25,30,40,0.8,-1,-1,-1\n"]) + frames = list(MOTSequence(root, "SEQ-01")) + assert [f.frame_id for f in frames] == [0, 1, 2] + assert len(MOTSequence(root, "SEQ-01")) == 3 + assert [len(f.gt_boxes) for f in frames] == [1, 1, 1] + # frame 1 has ground truth but no detection -- the tracker must still see it + assert [len(f.det_boxes) for f in frames] == [1, 0, 1] + assert frames[0].gt_boxes[0].tolist() == [10.0, 20.0, 40.0, 60.0] + assert frames[0].det_scores[0] == pytest.approx(0.9) + + +def test_mot_sequence_survives_a_missing_det_file(tmp_path): + from DeepDataMiningLearning.ngperception.tracking.datasets import MOTSequence + d = tmp_path / "SEQ-02" / "gt" + d.mkdir(parents=True) + (d / "gt.txt").write_text("1,1,0,0,10,10,1,-1,-1,-1\n", encoding="utf-8") + frames = list(MOTSequence(str(tmp_path), "SEQ-02")) + assert len(frames) == 1 and len(frames[0].det_boxes) == 0 + + +def test_mot_csv_skips_blank_and_comment_lines(tmp_path): + from DeepDataMiningLearning.ngperception.tracking.datasets import read_mot_csv + p = tmp_path / "gt.txt" + p.write_text("# header\n\n1,1,10,20,30,40,1,-1,-1,-1\n\n", encoding="utf-8") + assert sorted(read_mot_csv(str(p))) == [0] + + +def test_sequence_runs_end_to_end_through_tracker_and_evaluator(tmp_path): + """The path a user actually takes: read a sequence, track it, score it.""" + from DeepDataMiningLearning.ngperception.tracking.datasets import MOTSequence + gt, det = [], [] + for f in range(1, 9): # 1-indexed on disk + for oid, x0 in ((1, 10), (2, 200)): + x = x0 + 6 * (f - 1) * (1 if oid == 1 else -1) + row = f"{f},{oid},{x},{50 * oid},40,40,1,-1,-1,-1\n" + gt.append(row) + det.append(f"{f},-1,{x},{50 * oid},40,40,0.9,-1,-1,-1\n") + root = _write_seq(tmp_path, "SEQ-03", gt, det) + + tracker = build_tracker("sort", min_hits=1, max_age=2) + ev = MOTEvaluator() + ev.new_sequence() + for fr in MOTSequence(root, "SEQ-03"): + res = tracker.update(det_from_frame(fr)) + ev.add(fr.gt_ids.tolist(), fr.gt_boxes, res.track_ids.tolist(), res.boxes) + s = ev.summarize(verbose=False) + assert s["GT"] == 16 and s["IDSW"] == 0 + assert s["MOTA"] > 0.99 and s["IDF1"] > 0.99 + + +def det_from_frame(frame): + n = len(frame.det_boxes) + return Detection(boxes=frame.det_boxes, scores=frame.det_scores, + labels=np.zeros(n, np.int64), names=["object"] * n) + + +# ------------------------------------------------------------ error branches +def test_evaluator_rejects_misaligned_input(): + ev = MOTEvaluator() + with pytest.raises(ValueError, match="index-aligned"): + ev.add([1, 2], np.zeros((1, 4), np.float32), [1], np.zeros((1, 4), np.float32)) + + +def test_duplicate_track_ids_in_one_frame_are_rejected(): + bad = TrackResult(track_ids=np.array([5, 5]), boxes=np.zeros((2, 4), np.float32), + scores=np.zeros(2, np.float32), labels=np.zeros(2, np.int64), + names=["a", "b"]) + with pytest.raises(ValueError, match="duplicate"): + bad.check_aligned() + + +def test_base_tracker_is_abstract(): + from DeepDataMiningLearning.ngperception.tracking.trackers.base import BaseTracker + b = BaseTracker() + with pytest.raises(NotImplementedError): + b.update(det([[0, 0, 1, 1]])) + with pytest.raises(NotImplementedError): + b.reset() + + +def test_sort_rejects_an_unknown_variant(): + with pytest.raises(ValueError, match="variant"): + build_tracker("sort:turbo") + + +def test_summarize_verbose_prints_without_crashing(capsys): + ev = MOTEvaluator() + ev.new_sequence() + box = np.array([[0, 0, 10, 10]], np.float32) + ev.add([1], box, [1], box) + ev.summarize(verbose=True) + assert "MOTA" in capsys.readouterr().out + + +def test_a_far_detection_starts_a_new_track_instead_of_hijacking_one(): + """The IoU gate is what stops the assignment matching unrelated boxes. + + Hungarian assignment on its own will happily pair a track with the only + detection available, however far away it is. Without this case the suite + passes even when the gate is removed entirely -- a mutation that let any + detection claim any track went unnoticed until this was added. + """ + t = build_tracker("sort", min_hits=1, max_age=1) + first = t.update(det([[0, 0, 40, 40]])).track_ids.tolist() + assert first == [1] + # A detection on the far side of the image: IoU with the track is 0. + res = t.update(det([[900, 900, 940, 940]])) + ids = res.track_ids.tolist() + assert 1 not in ids, f"track 1 was hijacked by an unrelated detection: {ids}" + assert len(ids) == 1 and ids[0] != 1 + assert res.boxes[0][0] > 800, res.boxes[0].tolist() + + +def test_a_collapsing_box_keeps_a_finite_positive_size(): + """Exercises the area clamp in _KalmanBoxTracker.predict. + + A box shrinking fast gives the filter a negative area velocity; without the + clamp the predicted area goes through zero and the width/height come back + as nan (sqrt of a negative), which then poisons every later IoU. + """ + t = build_tracker("sort", min_hits=1, max_age=40) + for side in (400, 240, 120, 40, 8, 2): # collapsing fast + t.update(det([[100, 100, 100 + side, 100 + side]])) + last = None + for _ in range(12): # coast on prediction alone + t.update(det(np.zeros((0, 4), np.float32))) + assert t._tracks, "track retired too early for this check" + last = t._tracks[0].box + assert np.isfinite(last).all(), f"box went non-finite: {last.tolist()}" + w, h = last[2] - last[0], last[3] - last[1] + # With the clamp this stays ~240 px wide; without it the area passes through + # zero and every corner collapses onto the same point (w = h = 0). + assert w > 1.0 and h > 1.0, f"box collapsed to a point: w={w} h={h}" diff --git a/DeepDataMiningLearning/ngperception/tracking/trackers/__init__.py b/DeepDataMiningLearning/ngperception/tracking/trackers/__init__.py new file mode 100644 index 0000000..b2b19a1 --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/trackers/__init__.py @@ -0,0 +1 @@ +"""Tracker adapters. Import a backend module to register it.""" diff --git a/DeepDataMiningLearning/ngperception/tracking/trackers/base.py b/DeepDataMiningLearning/ngperception/tracking/trackers/base.py new file mode 100644 index 0000000..0ef7726 --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/trackers/base.py @@ -0,0 +1,218 @@ +""" +ngperception.tracking.trackers.base +=================================== + +The pluggable multi-object-tracker contract — the tracking analogue of +`ngdet.detectors.base` and `ngperception.depth.estimators.base`. Every backend (SORT, +DeepSORT, ByteTrack, OC-SORT, ...) is wrapped in a small adapter that subclasses +`BaseTracker` and registers itself with `@register("name")`. + +A tracker's ONLY job is: + per-frame `ngdet.Detection` -> `TrackResult` (the same boxes, now carrying ids) + +It does **not** detect. It consumes whatever `ngdet` produced for the frame, so the same +sequence can be re-tracked under a different detector without re-running the tracker's +own logic — which is what makes the detector/tracker ablation in `run_eval.py` fair. + +Two families exist, and the difference matters for what a backend may import: + +* **motion-only** trackers (SORT, ByteTrack, OC-SORT) associate on geometry alone — + IoU plus a motion model. They are pure numpy/scipy: no network, no weights, no GPU. +* **appearance** trackers (DeepSORT, BoT-SORT) add a re-identification embedding and + therefore need a model. Their heavy imports belong INSIDE the subclass `__init__`, + never at module top level, so that `import ngperception.tracking` stays cheap and a + missing optional dependency only breaks the backend that needs it. + +Identity semantics every adapter must honour: + +* an id is issued once and never reused within a sequence; +* the same physical object keeps its id across frames while it is tracked; +* a track that has not been matched for longer than the backend's `max_age` is retired, + and if the object reappears it is a *new* id — recovering the old one is re-identification, + which is a different (appearance) family. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional, Type + +import numpy as np + + +@dataclass +class TrackResult: + """Confirmed tracks for ONE frame. + + All fields are aligned by index (the i-th box has the i-th id/score/label/name), + the same contract `ngdet.Detection` uses. Boxes are absolute pixel coordinates in + xyxy (x_min, y_min, x_max, y_max) in the original image frame, so a `TrackResult` + can be overlaid on the source image exactly like a `Detection`. + + Attributes + ---------- + track_ids : np.ndarray + int64, one id per row of `boxes`. Unique within a sequence. + boxes : np.ndarray + Nx4 float32, xyxy. For a matched track this is the filtered (smoothed) estimate, + not the raw detection — that is the point of the motion model. + scores : np.ndarray + float32, carried through from the detection that updated the track. A track + coasting on prediction alone is not reported, so every row has a real score. + labels : np.ndarray + int64 unified-taxonomy ids, carried through from the detection. + names : list of str + Human-readable class names, index-aligned with `labels`. + """ + + track_ids: np.ndarray = field(default_factory=lambda: np.zeros((0,), np.int64)) + boxes: np.ndarray = field(default_factory=lambda: np.zeros((0, 4), np.float32)) + scores: np.ndarray = field(default_factory=lambda: np.zeros((0,), np.float32)) + labels: np.ndarray = field(default_factory=lambda: np.zeros((0,), np.int64)) + names: List[str] = field(default_factory=list) + + def __len__(self) -> int: + return len(self.boxes) + + def check_aligned(self) -> None: + """Raise if the index-alignment contract is broken. + + Cheap enough to call in tests and in a backend's own `__main__` block; an + adapter that filters `boxes` but forgets `labels` is exactly the bug this + catches, and it is silent otherwise. + """ + n = len(self.boxes) + for name, arr in (("track_ids", self.track_ids), ("scores", self.scores), + ("labels", self.labels), ("names", self.names)): + if len(arr) != n: + raise ValueError( + f"TrackResult is misaligned: {n} boxes but {len(arr)} {name}") + if len(set(self.track_ids.tolist())) != n: + raise ValueError("TrackResult contains duplicate track ids in one frame") + + +class BaseTracker: + """Abstract base for all tracker adapters. + + Subclasses implement `update`. A tracker is stateful across a sequence, so + `reset()` must return it to the state it had at construction — `run_eval.py` reuses + one instance across sequences and relies on that. + + Parameters + ---------- + iou_threshold : float + Minimum IoU for a detection to be accepted as the continuation of a track. + max_age : int + Frames a track may go unmatched before it is retired. + min_hits : int + Matches required before a track is reported. Suppresses one-frame false + positives at the cost of a short delay on genuinely new objects. + """ + + #: human-readable backend family name (set by subclass) + family: str = "base" + #: True for trackers that need an appearance/re-id model (and therefore a device) + needs_appearance: bool = False + + def __init__(self, iou_threshold: float = 0.3, max_age: int = 1, + min_hits: int = 3, **kwargs): + self.iou_threshold = float(iou_threshold) + self.max_age = int(max_age) + self.min_hits = int(min_hits) + + def update(self, detection) -> TrackResult: + """Advance the tracker by one frame and return the confirmed tracks. + + `detection` is an `ngdet.detectors.base.Detection`. Must be called once per + frame, in order; skipping frames breaks the motion model. + """ + raise NotImplementedError + + def reset(self) -> None: + """Forget all tracks and restart id numbering for a new sequence.""" + raise NotImplementedError + + +# --------------------------------------------------------------------------- +# Geometry shared by every motion-only backend. +# --------------------------------------------------------------------------- +def iou_matrix(a: np.ndarray, b: np.ndarray) -> np.ndarray: + """Pairwise IoU between two sets of xyxy boxes -> (len(a), len(b)) float32. + + Degenerate boxes (zero width or height) give 0 rather than a division warning, + and an empty input gives a correctly-shaped empty matrix so callers do not have + to special-case it. + """ + a = np.asarray(a, np.float32).reshape(-1, 4) + b = np.asarray(b, np.float32).reshape(-1, 4) + if len(a) == 0 or len(b) == 0: + return np.zeros((len(a), len(b)), np.float32) + + x1 = np.maximum(a[:, None, 0], b[None, :, 0]) + y1 = np.maximum(a[:, None, 1], b[None, :, 1]) + x2 = np.minimum(a[:, None, 2], b[None, :, 2]) + y2 = np.minimum(a[:, None, 3], b[None, :, 3]) + inter = np.clip(x2 - x1, 0.0, None) * np.clip(y2 - y1, 0.0, None) + + area_a = np.clip(a[:, 2] - a[:, 0], 0, None) * np.clip(a[:, 3] - a[:, 1], 0, None) + area_b = np.clip(b[:, 2] - b[:, 0], 0, None) * np.clip(b[:, 3] - b[:, 1], 0, None) + union = area_a[:, None] + area_b[None, :] - inter + return np.where(union > 0, inter / np.maximum(union, 1e-12), 0.0).astype(np.float32) + + +# --------------------------------------------------------------------------- +# Registry: short key -> adapter class, so a CLI can spell "--trackers sort" and +# resolve the class by the part before the colon (identical convention to +# ngdet.detectors and ngperception.depth.estimators). +# --------------------------------------------------------------------------- +TRACKER_REGISTRY: Dict[str, Type[BaseTracker]] = {} + + +def register(name: str) -> Callable[[Type[BaseTracker]], Type[BaseTracker]]: + def deco(cls: Type[BaseTracker]) -> Type[BaseTracker]: + TRACKER_REGISTRY[name] = cls + return cls + return deco + + +def build_tracker(spec: str, **kwargs) -> BaseTracker: + """Instantiate a tracker from a "key" or "key:variant" spec string. + + Examples + -------- + build_tracker("sort") + build_tracker("sort", iou_threshold=0.2, max_age=5) + """ + from . import sort # noqa: F401 (side effect: register backends) + + key, variant = (spec.split(":", 1) + [None])[:2] if ":" in spec else (spec, None) + if key not in TRACKER_REGISTRY: + raise KeyError( + f"Unknown tracker backend '{key}'. Registered: {list(TRACKER_REGISTRY)}") + if variant is not None: + kwargs.setdefault("variant", variant) + return TRACKER_REGISTRY[key](**kwargs) + + +# =========================================================================== +# HOW TO TEST / RUN THIS FILE +# python -m DeepDataMiningLearning.ngperception.tracking.trackers.base +# Expected: prints the registered tracker backends and an IoU sanity check. +# =========================================================================== +if __name__ == "__main__": + # Under `python -m`, this file is loaded as `__main__`; importing an adapter + # loads it a *second* time under its real name, and `@register` populates that + # copy's registry, not this one. Read the canonical module so the self-test + # reports what a normal `import` would see rather than an empty list. + from DeepDataMiningLearning.ngperception.tracking.trackers import ( # noqa: F401 + base as _canonical, sort, + ) + print("Registered tracker backends:", list(_canonical.TRACKER_REGISTRY)) + + a = np.array([[0, 0, 10, 10]], np.float32) + b = np.array([[0, 0, 10, 10], # identical -> 1.0 + [5, 0, 15, 10], # half overlap -> 1/3 + [20, 20, 30, 30], # disjoint -> 0.0 + [0, 0, 10, 0]], np.float32) # degenerate -> 0.0 + print("IoU row:", np.round(iou_matrix(a, b)[0], 4).tolist(), + "(expected [1.0, 0.3333, 0.0, 0.0])") diff --git a/DeepDataMiningLearning/ngperception/tracking/trackers/sort.py b/DeepDataMiningLearning/ngperception/tracking/trackers/sort.py new file mode 100644 index 0000000..f843623 --- /dev/null +++ b/DeepDataMiningLearning/ngperception/tracking/trackers/sort.py @@ -0,0 +1,225 @@ +""" +ngperception.tracking.trackers.sort +=================================== + +SORT — Simple Online and Realtime Tracking (Bewley et al., ICIP 2016). +The **basic** tier of the tracking task: the baseline every later method is measured +against, and the one that shows how far geometry alone gets you. + +Two components, both classical: + +1. a **constant-velocity Kalman filter** per track, on the state + ``[cx, cy, s, r, vx, vy, vs]`` where ``s = w*h`` (area) and ``r = w/h`` (aspect). + Aspect is treated as constant — SORT's own simplification, kept here so the + baseline is the published one rather than a private variant; +2. **Hungarian assignment** (``scipy.optimize.linear_sum_assignment``) on the IoU + between each track's predicted box and each detection, with matches below + ``iou_threshold`` rejected after the assignment rather than before, so the + assignment stays globally optimal. + +No network, no weights, no GPU: numpy and scipy only. That is deliberate — it means +the tracking arm of the suite can be run and reproduced by anyone, and a later +appearance-based backend has a like-for-like reference to beat. + +Known limits of the baseline, stated so the numbers are not over-read: SORT has no +re-identification, so an object that leaves and returns gets a new id and the +sequence takes an ID-switch; it has no occlusion handling beyond ``max_age`` coasting; +and it inherits every miss of the detector it is given. +""" + +from __future__ import annotations + +from typing import List, Optional, Tuple + +import numpy as np + +from .base import BaseTracker, TrackResult, iou_matrix, register + + +class _KalmanBoxTracker: + """One tracked object: a constant-velocity Kalman filter over ``[cx, cy, s, r]``.""" + + def __init__(self, box: np.ndarray, score: float, label: int, name: str, track_id: int): + self.id = int(track_id) + self.score = float(score) + self.label = int(label) + self.name = str(name) + + self.time_since_update = 0 + self.hits = 1 + self.age = 0 + + # State transition: position advances by its velocity, aspect is constant. + self._F = np.eye(7, dtype=np.float64) + for i in range(3): + self._F[i, 4 + i] = 1.0 + # We observe the box, not the velocities. + self._H = np.zeros((4, 7), dtype=np.float64) + self._H[:4, :4] = np.eye(4) + + self._P = np.eye(7, dtype=np.float64) * 10.0 + self._P[4:, 4:] *= 1000.0 # velocities start highly uncertain + self._Q = np.eye(7, dtype=np.float64) * 0.01 + self._Q[4:, 4:] *= 0.01 + self._R = np.eye(4, dtype=np.float64) + self._R[2:, 2:] *= 10.0 # area/aspect are noisier than the centre + + self._x = np.zeros(7, dtype=np.float64) + self._x[:4] = self._to_z(box) + + # -- box <-> state ------------------------------------------------------ + @staticmethod + def _to_z(box: np.ndarray) -> np.ndarray: + """xyxy -> [cx, cy, area, aspect].""" + w = max(float(box[2]) - float(box[0]), 1e-6) + h = max(float(box[3]) - float(box[1]), 1e-6) + return np.array([float(box[0]) + w / 2.0, float(box[1]) + h / 2.0, w * h, w / h]) + + @staticmethod + def _to_box(z: np.ndarray) -> np.ndarray: + """[cx, cy, area, aspect] -> xyxy.""" + area = max(float(z[2]), 1e-9) + aspect = max(float(z[3]), 1e-9) + w = float(np.sqrt(area * aspect)) + h = area / max(w, 1e-9) + return np.array([z[0] - w / 2.0, z[1] - h / 2.0, + z[0] + w / 2.0, z[1] + h / 2.0], dtype=np.float32) + + # -- filter ------------------------------------------------------------- + def predict(self) -> np.ndarray: + """Advance one frame and return the predicted box.""" + # A shrinking box can drive the area negative; clamp before it does. + if self._x[2] + self._x[6] <= 0: + self._x[6] = 0.0 + self._x = self._F @ self._x + self._P = self._F @ self._P @ self._F.T + self._Q + self.age += 1 + self.time_since_update += 1 + return self._to_box(self._x[:4]) + + def update(self, box: np.ndarray, score: float, label: int, name: str) -> None: + z = self._to_z(box) + y = z - self._H @ self._x + S = self._H @ self._P @ self._H.T + self._R + K = self._P @ self._H.T @ np.linalg.inv(S) + self._x = self._x + K @ y + self._P = (np.eye(7) - K @ self._H) @ self._P + + self.time_since_update = 0 + self.hits += 1 + self.score = float(score) + self.label = int(label) + self.name = str(name) + + @property + def box(self) -> np.ndarray: + return self._to_box(self._x[:4]) + + +@register("sort") +class SortTracker(BaseTracker): + """Motion-only baseline tracker. See the module docstring for the algorithm.""" + + family = "sort" + needs_appearance = False + + def __init__(self, iou_threshold: float = 0.3, max_age: int = 1, + min_hits: int = 3, variant: Optional[str] = None, **kwargs): + super().__init__(iou_threshold=iou_threshold, max_age=max_age, + min_hits=min_hits, **kwargs) + if variant not in (None, "", "default"): + raise ValueError(f"SortTracker has no variant '{variant}'") + self.reset() + + def reset(self) -> None: + self._tracks: List[_KalmanBoxTracker] = [] + self._next_id = 1 + self._frame = 0 + + # -- association -------------------------------------------------------- + def _associate(self, predicted: np.ndarray, boxes: np.ndarray + ) -> Tuple[List[Tuple[int, int]], List[int]]: + """Return (matches, unmatched_detection_indices). + + The Hungarian solution is computed on the full IoU matrix and only then + filtered by `iou_threshold`. Filtering first would let a greedy leftover win + an assignment the optimal solution had given to a better pair. + """ + if len(predicted) == 0 or len(boxes) == 0: + return [], list(range(len(boxes))) + + from scipy.optimize import linear_sum_assignment + + iou = iou_matrix(predicted, boxes) + rows, cols = linear_sum_assignment(-iou) + matches = [(int(r), int(c)) for r, c in zip(rows, cols) + if iou[r, c] >= self.iou_threshold] + matched_dets = {c for _, c in matches} + return matches, [d for d in range(len(boxes)) if d not in matched_dets] + + def update(self, detection) -> TrackResult: + self._frame += 1 + + boxes = np.asarray(detection.boxes, np.float32).reshape(-1, 4) + n = len(boxes) + scores = (np.asarray(detection.scores, np.float32) if len(detection.scores) == n + else np.ones(n, np.float32)) + labels = (np.asarray(detection.labels, np.int64) if len(detection.labels) == n + else np.zeros(n, np.int64)) + names = (list(detection.names) if len(detection.names) == n + else [""] * n) + + predicted = np.array([t.predict() for t in self._tracks], np.float32) \ + if self._tracks else np.zeros((0, 4), np.float32) + + matches, unmatched = self._associate(predicted, boxes) + for ti, di in matches: + self._tracks[ti].update(boxes[di], scores[di], labels[di], names[di]) + for di in unmatched: + self._tracks.append(_KalmanBoxTracker( + boxes[di], scores[di], labels[di], names[di], self._next_id)) + self._next_id += 1 + + self._tracks = [t for t in self._tracks if t.time_since_update <= self.max_age] + + # Report only tracks updated this frame. During the first `min_hits` frames a + # new track is reported immediately, otherwise a sequence would start empty + # and every object would take an avoidable miss -- this is SORT's own rule. + live = [t for t in self._tracks + if t.time_since_update == 0 + and (t.hits >= self.min_hits or self._frame <= self.min_hits)] + + out = TrackResult( + track_ids=np.array([t.id for t in live], np.int64), + boxes=(np.array([t.box for t in live], np.float32) if live + else np.zeros((0, 4), np.float32)), + scores=np.array([t.score for t in live], np.float32), + labels=np.array([t.label for t in live], np.int64), + names=[t.name for t in live], + ) + out.check_aligned() + return out + + +# =========================================================================== +# HOW TO TEST / RUN THIS FILE +# python -m DeepDataMiningLearning.ngperception.tracking.trackers.sort +# Two objects crossing plus one appearing late: ids must stay stable and the +# newcomer must get a fresh id. +# =========================================================================== +if __name__ == "__main__": + from DeepDataMiningLearning.ngdet.detectors.base import Detection + + def det(boxes): + b = np.array(boxes, np.float32).reshape(-1, 4) + return Detection(boxes=b, scores=np.ones(len(b), np.float32), + labels=np.zeros(len(b), np.int64), names=["car"] * len(b)) + + tracker = SortTracker(min_hits=1) + for t in range(6): + boxes = [[10 + 12 * t, 20, 60 + 12 * t, 80], + [200 - 6 * t, 100, 250 - 6 * t, 160]] + if t >= 3: + boxes.append([300, 300 + 5 * (t - 3), 340, 350 + 5 * (t - 3)]) + res = tracker.update(det(boxes)) + print(f"frame {t}: ids={res.track_ids.tolist()}")