From 720f5f076cebf07073f0e2955db9243ef6f2802d Mon Sep 17 00:00:00 2001 From: Egemen Tuncarslan Date: Tue, 1 Sep 2026 15:39:02 +0300 Subject: [PATCH] ngperception/tracking: SORT baseline, evaluator and dataset-free harness The README's task table listed Tracking (MOT) as planned with SORT as the basic tier; the folder did not exist. This adds it, following the same three-layer shape the rest of the suite uses (trackers / datasets / evaluator) and consuming ngdet.Detection rather than detecting anything itself, so a sequence can be re-tracked under a different detector without touching the tracker. Contents trackers/base.py TrackResult + BaseTracker + TRACKER_REGISTRY + build_tracker + iou_matrix, mirroring ngdet.detectors.base and depth.estimators.base trackers/sort.py @register("sort") -- constant-velocity Kalman filter over (cx, cy, area, aspect) plus Hungarian assignment on IoU. Matches below the gate are rejected AFTER the assignment so it stays globally optimal datasets.py MOTChallenge csv reader plus a synthetic generator, so the module runs with no dataset at all evaluator.py CLEAR-MOT: MOTA / MOTP / IDF1 / ID-switches, counters summed across sequences and the ratios formed once (the MOTChallenge convention) run_eval.py CLI in the same shape as depth/run_eval.py numpy and scipy only: no network, no weights, no GPU, nothing to download. That is deliberate -- the tracking arm stays reproducible by anyone, and a later appearance-based backend has a like-for-like reference to beat. Measured, 2 x 30 frames x 4 objects, --min-hits 1 --max-age 3: 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 +0.8583 0.8914 0.7175 0 34 Jitter costs MOTP and leaves MOTA's detection terms alone; missed detections cost MOTA in proportion. Reporting both keeps those separable. Tests: 37, ~1.2 s, no GPU and no dataset. Line coverage is 100% of the importable code in all four modules (the only uncovered lines are the `if __name__` self-test blocks, which each run under `python -m`). They were checked against 18 seeded mutations -- gate removal, assignment sign, id reuse, area/aspect swap, the area clamp, IoU union, the ID-switch counter, MOTA's terms, both MOTChallenge conversions -- and all 18 turn the suite red. Two limits are stated in the README rather than left implicit: SORT has no re-identification, so an object that leaves and returns takes an ID-switch; and it inherits every miss of the detector it is given. Co-Authored-By: Claude Opus 5 --- DeepDataMiningLearning/ngperception/README.md | 2 +- .../ngperception/tracking/README.md | 109 +++++ .../ngperception/tracking/__init__.py | 1 + .../ngperception/tracking/datasets.py | 199 ++++++++ .../ngperception/tracking/evaluator.py | 187 ++++++++ .../ngperception/tracking/run_eval.py | 155 +++++++ .../ngperception/tracking/tests/__init__.py | 0 .../tracking/tests/test_tracking.py | 430 ++++++++++++++++++ .../tracking/trackers/__init__.py | 1 + .../ngperception/tracking/trackers/base.py | 218 +++++++++ .../ngperception/tracking/trackers/sort.py | 225 +++++++++ 11 files changed, 1526 insertions(+), 1 deletion(-) create mode 100644 DeepDataMiningLearning/ngperception/tracking/README.md create mode 100644 DeepDataMiningLearning/ngperception/tracking/__init__.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/datasets.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/evaluator.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/run_eval.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/tests/__init__.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/tests/test_tracking.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/trackers/__init__.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/trackers/base.py create mode 100644 DeepDataMiningLearning/ngperception/tracking/trackers/sort.py diff --git a/DeepDataMiningLearning/ngperception/README.md b/DeepDataMiningLearning/ngperception/README.md index 92fbb1e9..8a40526b 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 00000000..f2fa05e7 --- /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 00000000..adf0aa71 --- /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 00000000..d27d25f0 --- /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 00000000..ef4a300f --- /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 00000000..a2a8ec01 --- /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 00000000..e69de29b diff --git a/DeepDataMiningLearning/ngperception/tracking/tests/test_tracking.py b/DeepDataMiningLearning/ngperception/tracking/tests/test_tracking.py new file mode 100644 index 00000000..28e1e015 --- /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 00000000..b2b19a1b --- /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 00000000..0ef7726a --- /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 00000000..f843623f --- /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()}")