Data Association Tutorial
This tutorial demonstrates data association algorithms for matching measurements to tracked targets in a cluttered, multi-target scenario.
Topics covered:
Global Nearest Neighbor (GNN) – greedy assignment
Optimal assignment via the Hungarian algorithm (
scipy.optimize.linear_sum_assignment)Cost-matrix formulation with gating
Simple track initiation/deletion logic driven by association outcome
Scenario Setup
The tutorial simulates three targets moving on linear trajectories, with clutter measurements mixed in at each time step. Each step gives every target a 95% detection probability and adds Poisson(2) clutter measurements drawn uniformly over the surveillance region:
import numpy as np
np.random.seed(42)
n_steps = 50
n_targets = 3
# Target trajectories: x, y, vx, vy
targets = np.array(
[
[10.0, 10.0, 1.0, 0.5],
[15.0, 5.0, -0.5, 1.5],
[5.0, 15.0, 0.8, -0.8],
]
)
meas_list = []
for k in range(n_steps):
targets[:, 0] += targets[:, 2] * 0.1
targets[:, 1] += targets[:, 3] * 0.1
step_meas = []
for i in range(n_targets):
if np.random.rand() < 0.95:
step_meas.append(targets[i, :2] + np.random.randn(2) * 0.5)
for _ in range(np.random.poisson(2)):
step_meas.append(np.random.uniform(0, 20, 2))
meas_list.append(np.array(step_meas) if step_meas else np.zeros((0, 2)))
Global Nearest Neighbor
GNN greedily assigns the lowest-cost (track, measurement) pair under a 5 m gate, repeating until no pair is left, then ages out tracks that go too long without a hit and never reached a confirmation threshold:
# Track fields: x, y, vx, vy, age, confidence
tracks = [[z[0], z[1], 0, 0, 0, 3] for z in meas_list[0][:n_targets]]
for k in range(n_steps):
measurements = meas_list[k]
n_tracks = len(tracks)
n_meas = len(measurements)
if n_tracks > 0 and n_meas > 0:
cost_matrix = np.zeros((n_tracks, n_meas))
for i in range(n_tracks):
track_pos = np.array([tracks[i][0], tracks[i][1]])
for j in range(n_meas):
cost_matrix[i, j] = np.linalg.norm(track_pos - measurements[j])
assignments = {}
used_meas = set()
for _ in range(min(n_tracks, n_meas)):
best_track, best_meas, min_cost = -1, -1, np.inf
for i in range(n_tracks):
for j in range(n_meas):
if j not in used_meas and cost_matrix[i, j] < min_cost:
min_cost, best_track, best_meas = cost_matrix[i, j], i, j
if best_track >= 0 and min_cost < 5.0: # gate
assignments[best_track] = best_meas
used_meas.add(best_meas)
for i, track in enumerate(tracks):
if i in assignments:
z = measurements[assignments[i]]
track[0] = 0.7 * track[0] + 0.3 * z[0]
track[1] = 0.7 * track[1] + 0.3 * z[1]
track[4] += 1
track[5] = min(track[5] + 1, 5)
else:
track[4] += 1
track[5] = max(track[5] - 1, 0)
for j in range(n_meas):
if j not in used_meas:
tracks.append([measurements[j][0], measurements[j][1], 0, 0, 0, 0])
tracks = [t for t in tracks if t[4] < 10 or t[5] >= 3]
n_gnn_tracks = len(tracks)
GNN is cheap but can commit to a locally optimal pair that blocks a better overall assignment when tracks are close together.
Optimal Assignment (Hungarian Algorithm)
The Hungarian algorithm instead finds the assignment that minimizes total cost across all track-measurement pairs simultaneously, over the same measurement stream:
from scipy.optimize import linear_sum_assignment
tracks_h = [[z[0], z[1], 0, 0, 0, 3] for z in meas_list[0][:n_targets]]
for k in range(n_steps):
measurements = meas_list[k]
n_tracks = len(tracks_h)
n_meas = len(measurements)
if n_tracks > 0 and n_meas > 0:
cost_matrix = np.zeros((n_tracks, n_meas))
for i in range(n_tracks):
track_pos = np.array([tracks_h[i][0], tracks_h[i][1]])
for j in range(n_meas):
d = np.linalg.norm(track_pos - measurements[j])
# Ungated pairs get a large penalty cost instead of being
# excluded outright.
cost_matrix[i, j] = d if d < 5.0 else 1000.0
track_idx, meas_idx = linear_sum_assignment(cost_matrix)
assignments = {
t: m for t, m in zip(track_idx, meas_idx) if cost_matrix[t, m] < 1000.0
}
for i, track in enumerate(tracks_h):
if i in assignments:
z = measurements[assignments[i]]
track[0] = 0.7 * track[0] + 0.3 * z[0]
track[1] = 0.7 * track[1] + 0.3 * z[1]
track[4] += 1
track[5] = min(track[5] + 1, 5)
else:
track[4] += 1
track[5] = max(track[5] - 1, 0)
used_meas = set(assignments.values())
for j in range(n_meas):
if j not in used_meas:
tracks_h.append([measurements[j][0], measurements[j][1], 0, 0, 0, 0])
tracks_h = [t for t in tracks_h if t[4] < 10 or t[5] >= 3]
n_hungarian_tracks = len(tracks_h)
Track Maintenance
Both approaches share the same simple track lifecycle, already applied
inside each loop above: an associated measurement pulls the track’s
position; a missed association ages the track down; and a track that goes
too long without a hit and never reached a confirmation threshold is
dropped – the same t[4] < 10 or t[5] >= 3 rule filters both
tracks (GNN) and tracks_h (Hungarian):
print(f"GNN confirmed tracks: {n_gnn_tracks}")
print(f"Hungarian confirmed tracks: {n_hungarian_tracks}")
Next Steps
See Multi-Target Tracking Tutorial for a full tracker built around association output (track confirmation, deletion, OSPA scoring)
See Assignment Algorithms for the library’s GNN, JPDA, and Hungarian-based association routines (this tutorial reimplements a minimal version of both for illustration)
See Trackers for the tracker classes that wrap association with state estimation