Tracking hand-off

DetectionFrame, TrackHandoff and the reference nearest_neighbour_associator that pass timestamped point clouds to a tracker.

Radar does not include a tracker. These records pass timestamped detections to one you supply, with a small reference associator so the contract has a runnable example.

from witwin.radar.processing import DetectionFrame, TrackHandoff, nearest_neighbour_associator

DetectionFrame#

class DetectionFrame(time_s, xyz, velocity_mps, energy, frame_index)

One frame of detections with a timestamp.

FieldMeaning
time_sFrame time, s.
xyz[N, 3] positions, m, point-cloud frame.
velocity_mps[N] radial velocity, m/s, closing positive.
energy[N] energy, dB.
frame_indexIndex of the frame in your sequence.

len(frame) is N.

DetectionFrame.from_point_cloud#

DetectionFrame.from_point_cloud(cloud, *, time_s, frame_index) -> DetectionFrame

Wrap a PointCloud without recomputing anything.

DetectionFrame.as_fixed_size#

DetectionFrame.as_fixed_size(size, *, generator=None) -> torch.Tensor

A [size, 6] tensor in POINT_CLOUD_COLUMNS order, for a learned model that needs a fixed input size.

  • No detections: all zeros.
  • Fewer than size: every detection appears once, and the remaining slots are filled by sampling with replacement.
  • More than size: size detections sampled without replacement.
sizeintrequired

Number of rows.

generatortorch.Generator | Nonedefault None

Random generator for the sampling, on the detections' device. Pass one for reproducible batches.

batch = frame.as_fixed_size(128, generator=torch.Generator(device="cuda").manual_seed(0))

TrackHandoff#

class TrackHandoff(associator=None)

Stores detection frames in order and calls an associator on each new frame. It has no motion model, gate or track lifetime of its own.

associatorCallable | Nonedefault None

associator(previous, current, elapsed_s) returns an [N] int64 tensor: for each current detection, the index of the matched previous detection, or -1 to open a new track. previous is None for the first frame. None uses nearest_neighbour_associator.

MemberMeaning
push(frame)Append a DetectionFrame and return its [N] int64 track IDs. Frame times must increase.
track_countNumber of tracks opened so far.
track(track_id)[(time_s, xyz), ...] for one track, in frame order.
frames, assignmentsThe stored frames and their track IDs.

Raises from push: ValueError if the frame time is not after the previous one or the associator returns the wrong number of entries; TypeError if the frame is not a DetectionFrame or the associator does not return an int64 tensor.

handoff = TrackHandoff()
for k, frame_result in enumerate(radar.stream(scene, targets, times=times)):
    cloud = frame_result.frame(0).points(max_points=64)
    handoff.push(DetectionFrame.from_point_cloud(cloud, time_s=times[k], frame_index=k))
print(handoff.track_count, handoff.track(0)[:3])

nearest_neighbour_associator#

nearest_neighbour_associator(previous, current, elapsed_s, *, gate_m=1.0) -> torch.Tensor

Reference constant-velocity association. Each previous detection is moved forward by elapsed_s along its radial velocity, and each current detection is matched greedily to the nearest prediction within gate_m. One prediction claims at most one detection. It has no track initiation logic, no track deletion and no covariance, and it is not differentiable.

previousDetectionFrame | Nonerequired

The previous frame, or None for the first one.

currentDetectionFramerequired

The new frame.

elapsed_sfloatrequired

Time since the previous frame, s.

gate_mfloatdefault 1.0

Largest match distance, m.

Returns [N] int64 indices into previous, -1 for unmatched detections.

To change the gate, wrap it:

from functools import partial
 
handoff = TrackHandoff(associator=partial(nearest_neighbour_associator, gate_m=0.5))