Tracking

Track module for detection tracking functionality.

This module provides tracking utilities and classes for object detection.

ReClass

ReClass(
    num_frames: int = 25,
    threshold: float = 0.75,
    model: str = "rtdetr",
    weights: str = "x",
    device: str = "auto",
    default_class: int = 0,
    match_class: list | None = None,
)

ReClass is responsible for re-classifying object tracks in video frames using detection results.

Attributes:

Name	Type	Description
detector	Detector	The detection model used for re-classification.
num_frames	int	Number of frames to consider for re-classification.
threshold	float	Threshold for matching detections to tracks.
default_class	int	Default class to assign if no match is found.
match_class	list	List of classes to match during re-classification.

Methods:

Name	Description
match_mmv	Matches a track to detections and computes the average score.
re_classify	tracks: pd.DataFrame, input_video: str, track_ids: list = None, out_file: str = None, verbose: bool = True
) -> pd.DataFrame	Re-classifies tracks and returns a DataFrame with results.

Re-classify tracks based on detection results.

Parameters:

Name	Type	Description	Default
num_frames	int	Number of frames to consider for re-classification, default 25	25
threshold	float	Threshold for matching, default 0.75	0.75
model	str	Detection model to use, default 'rtdetr'	'rtdetr'
weights	str	Weights for the detection model, default 'x'	'x'
device	str	Device to use for detection, default 'auto'	'auto'
default_class	int	Default class to assign if no match found, default 0 (pedestrian)	0
match_class	list	List of classes to match, default [1, 36] (bicycle, skateboard/scooter)	None

Source code in src/dnt/track/re_class.py

def __init__(
    self,
    num_frames: int = 25,
    threshold: float = 0.75,
    model: str = "rtdetr",
    weights: str = "x",
    device: str = "auto",
    default_class: int = 0,
    match_class: list | None = None,
) -> None:
    """Re-classify tracks based on detection results.

    Parameters
    ----------
    num_frames : int
        Number of frames to consider for re-classification, default 25
    threshold : float
        Threshold for matching, default 0.75
    model : str
        Detection model to use, default 'rtdetr'
    weights : str
        Weights for the detection model, default 'x'
    device : str
        Device to use for detection, default 'auto'
    default_class : int
        Default class to assign if no match found, default 0 (pedestrian)
    match_class : list
        List of classes to match, default [1, 36] (bicycle, skateboard/scooter)

    """
    self.detector = Detector(model=model, device=device)
    self.num_frames = num_frames
    self.threshold = threshold
    self.default_class = default_class
    self.match_class = match_class if match_class is not None else [1, 36]

match_mmv

match_mmv(
    track: DataFrame, dets: DataFrame
) -> tuple[bool, float]

Match track bboxes to detection bboxes and compute average overlap score.

Parameters:

Name	Type	Description	Default
track	DataFrame	DataFrame containing track data with columns [x, y, w, h, frame].	required
dets	DataFrame	DataFrame containing detection data with columns [x, y, w, h, frame, class].	required

Returns:

Type	Description
tuple[bool, float]	A tuple (hit, avg_score) where: - hit : bool True if average overlap score meets threshold, False otherwise. - avg_score : float Average Intersection over Box (IoB) score across all matched detections.

Notes

Only frames present in both track and detection datasets are considered. The matching uses IoB metric from the engine.iob module.

Source code in src/dnt/track/re_class.py

def match_mmv(self, track: pd.DataFrame, dets: pd.DataFrame) -> tuple[bool, float]:
    """Match track bboxes to detection bboxes and compute average overlap score.

    Parameters
    ----------
    track : pd.DataFrame
        DataFrame containing track data with columns [x, y, w, h, frame].
    dets : pd.DataFrame
        DataFrame containing detection data with columns [x, y, w, h, frame, class].

    Returns
    -------
    tuple[bool, float]
        A tuple (hit, avg_score) where:
        - hit : bool
            True if average overlap score meets threshold, False otherwise.
        - avg_score : float
            Average Intersection over Box (IoB) score across all matched detections.

    Notes
    -----
    Only frames present in both track and detection datasets are considered.
    The matching uses IoB metric from the engine.iob module.

    """
    if track.empty or dets.empty:
        return False, 0.0

    score = 0.0
    cnt = 0
    for _, row in track.iterrows():
        bboxes = row[["x", "y", "w", "h"]].values.reshape(1, -1)
        det = dets[dets["frame"] == row["frame"]]
        if len(det) > 0:
            match_bboxes = det[["x", "y", "w", "h"]].values
            _, overlaps_mmv = iobs(bboxes, match_bboxes)
            if overlaps_mmv.size > 0:
                max_overlap = np.max(overlaps_mmv)
                if max_overlap >= self.threshold:
                    score += max_overlap
                    cnt += 1

    avg_score = score / cnt if cnt > 0 else 0.0
    hit = avg_score >= self.threshold

    return hit, avg_score

re_classify

re_classify(
    tracks: DataFrame,
    input_video: str,
    track_ids: list | None = None,
    out_file: str | None = None,
    verbose: bool = True,
) -> pd.DataFrame

Re-classify tracks using detection matching against reference image frame samples.

For each track, extracts the top N largest frames (by area), runs detection on those frames, and matches detections against the track bboxes using IoB metric. Assigns the highest-scoring match class if confidence exceeds self.threshold.

Parameters:

Name	Type	Description	Default
tracks	DataFrame	Input tracks DataFrame with required columns: track, x, y, w, h, frame. Additional columns are preserved in output.	required
input_video	str	Path to source video file from which to extract frame samples.	required
track_ids	list | None	List of track IDs to re-classify. If None (default), all tracks in the input are re-classified.	None
out_file	str | None	Path to save re-classified results as CSV. If None (default), results are not saved to file.	None
verbose	bool	If True (default), display progress bar during re-classification.	True

Returns:

Type	Description
DataFrame	Output DataFrame with columns [track, cls, avg_score] where: - track : int Track ID from input tracks. - cls : int Re-classified class ID. Set to default_class if no match found. - avg_score : float Maximum IoB score among matched detections, rounded to 2 decimals.

Raises:

Type	Description
ValueError	If input tracks DataFrame is empty.
FileNotFoundError	If input_video does not exist.

Notes

The method considers only the top N frames (self.num_frames) by bounding box area for computational efficiency. It matches detections from match_class list against track bboxes and selects the class with highest average score.

Examples:

>>> import pandas as pd
>>> from .re_class import ReClass
>>> tracks = pd.DataFrame({
...     'frame': [0, 1, 2],
...     'track': [1, 1, 1],
...     'x': [100, 102, 104],
...     'y': [50, 52, 54],
...     'w': [50, 50, 50],
...     'h': [100, 100, 100],
... })
>>> rc = ReClass(num_frames=2, threshold=0.75, match_class=[1, 36])
>>> result = rc.re_classify(tracks, 'video.mp4')
>>> print(result)  # DataFrame with [track, cls, avg_score]

Source code in src/dnt/track/re_class.py

def re_classify(
    self,
    tracks: pd.DataFrame,
    input_video: str,
    track_ids: list | None = None,
    out_file: str | None = None,
    verbose: bool = True,
) -> pd.DataFrame:
    """Re-classify tracks using detection matching against reference image frame samples.

    For each track, extracts the top N largest frames (by area), runs detection on
    those frames, and matches detections against the track bboxes using IoB metric.
    Assigns the highest-scoring match class if confidence exceeds self.threshold.

    Parameters
    ----------
    tracks : pd.DataFrame
        Input tracks DataFrame with required columns: track, x, y, w, h, frame.
        Additional columns are preserved in output.
    input_video : str
        Path to source video file from which to extract frame samples.
    track_ids : list | None, optional
        List of track IDs to re-classify. If None (default), all tracks in
        the input are re-classified.
    out_file : str | None, optional
        Path to save re-classified results as CSV. If None (default), results
        are not saved to file.
    verbose : bool, optional
        If True (default), display progress bar during re-classification.

    Returns
    -------
    pd.DataFrame
        Output DataFrame with columns [track, cls, avg_score] where:
        - track : int
            Track ID from input tracks.
        - cls : int
            Re-classified class ID. Set to default_class if no match found.
        - avg_score : float
            Maximum IoB score among matched detections, rounded to 2 decimals.

    Raises
    ------
    ValueError
        If input tracks DataFrame is empty.
    FileNotFoundError
        If input_video does not exist.

    Notes
    -----
    The method considers only the top N frames (self.num_frames) by bounding
    box area for computational efficiency. It matches detections from match_class
    list against track bboxes and selects the class with highest average score.

    Examples
    --------
    >>> import pandas as pd
    >>> from .re_class import ReClass
    >>> tracks = pd.DataFrame({
    ...     'frame': [0, 1, 2],
    ...     'track': [1, 1, 1],
    ...     'x': [100, 102, 104],
    ...     'y': [50, 52, 54],
    ...     'w': [50, 50, 50],
    ...     'h': [100, 100, 100],
    ... })
    >>> rc = ReClass(num_frames=2, threshold=0.75, match_class=[1, 36])
    >>> result = rc.re_classify(tracks, 'video.mp4')
    >>> print(result)  # DataFrame with [track, cls, avg_score]

    """
    if tracks.empty:
        raise ValueError("Input tracks DataFrame is empty.")
    if not Path(input_video).exists():
        raise FileNotFoundError(f"Video file not found: {input_video}")

    if track_ids is None:
        track_ids = tracks["track"].unique().tolist()

    results = []
    if verbose:
        pbar = tqdm(total=len(track_ids), unit="track", desc="Re-classifying tracks")
    for track_id in track_ids:
        target_track = tracks[tracks["track"] == track_id].copy()
        target_track["area"] = target_track["w"] * target_track["h"]
        target_track.sort_values(by="area", inplace=True, ascending=False)

        top_frames = target_track.head(self.num_frames) if len(target_track) >= self.num_frames else target_track

        dets = self.detector.detect_frames(input_video, top_frames["frame"].values.tolist())

        matched = []
        for cls in self.match_class:
            match_dets = dets[dets["class"] == cls]
            hit, avg_score = self.match_mmv(top_frames, match_dets)
            if hit:
                matched.append((cls, avg_score))

        if len(matched) > 0:
            cls, avg_score = max(matched, key=lambda x: x[1])
        else:
            cls = self.default_class
            avg_score = 0

        results.append([track_id, cls, round(avg_score, 2)])
        if verbose:
            pbar.update()
    if verbose:
        pbar.close()

    df = pd.DataFrame(results, columns=["track", "cls", "avg_score"])
    if out_file:
        df.to_csv(out_file, index=False)

    return df

BoostTrackConfig dataclass

BoostTrackConfig(
    model: MOTModels = MOTModels.BOOSTTRACK,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    reid_weights: ReIDWeights
    | str
    | None = ReIDWeights.OSNET_X1_0_MSMT17,
    det_thresh: float = 0.3,
    max_age: int = 30,
    min_hits: int = 3,
    iou_threshold: float = 0.3,
    asso_func: str = "iou",
)

Bases: MOTBaseConfig

BoostTrack-specific parameters.

Attributes:

Name	Type	Description
reid_weights	ReIDWeights | str | None	Optional ReID weights path (default: "osnet_x1_0_msmt17.pt"). Options: same built-in .pt names listed in BoTSORTParams.reid_weights.
det_thresh	float	Detection confidence threshold (default: 0.3). Increasing this keeps only higher-confidence detections.
max_age	int	Maximum age of unmatched tracks (default: 30). Increasing this keeps tracks alive longer when unmatched.
min_hits	int	Minimum hits before track confirmation (default: 3). Increasing this delays confirmation and reduces short noisy tracks.
iou_threshold	float	IoU threshold for association (default: 0.3). Increasing this demands tighter overlap to match detections.
asso_func	str	Association function name (default: "iou"). Typical options: "iou", "giou", "diou", "ciou", "centroid".

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

BoTSORTConfig dataclass

BoTSORTConfig(
    model: MOTModels = MOTModels.BOTSORT,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    reid_weights: ReIDWeights
    | str
    | None = ReIDWeights.OSNET_X1_0_MSMT17,
    track_high_thresh: float = 0.5,
    track_low_thresh: float = 0.1,
    new_track_thresh: float = 0.6,
    match_thresh: float = 0.8,
    track_buffer: int = 30,
    with_reid: bool = True,
    proximity_thresh: float = 0.5,
    appearance_thresh: float = 0.25,
)

Bases: MOTBaseConfig

BoTSORT-specific parameters.

Attributes:

Name	Type	Description
reid_weights	ReIDWeights | str | None	Optional ReID weights path (default: "osnet_x1_0_msmt17.pt"). Options: None, file name (auto-resolved), or absolute path. Built-in downloadable options: 'resnet50_market1501.pt', 'resnet50_dukemtmcreid.pt', 'resnet50_msmt17.pt', 'resnet50_fc512_market1501.pt', 'resnet50_fc512_dukemtmcreid.pt', 'resnet50_fc512_msmt17.pt', 'mlfn_market1501.pt', 'mlfn_dukemtmcreid.pt', 'mlfn_msmt17.pt', 'hacnn_market1501.pt', 'hacnn_dukemtmcreid.pt', 'hacnn_msmt17.pt', 'mobilenetv2_x1_0_market1501.pt', 'mobilenetv2_x1_0_dukemtmcreid.pt', 'mobilenetv2_x1_0_msmt17.pt', 'mobilenetv2_x1_4_market1501.pt', 'mobilenetv2_x1_4_dukemtmcreid.pt', 'mobilenetv2_x1_4_msmt17.pt', 'osnet_x1_0_market1501.pt', 'osnet_x1_0_dukemtmcreid.pt', 'osnet_x1_0_msmt17.pt', 'osnet_x0_75_market1501.pt', 'osnet_x0_75_dukemtmcreid.pt', 'osnet_x0_75_msmt17.pt', 'osnet_x0_5_market1501.pt', 'osnet_x0_5_dukemtmcreid.pt', 'osnet_x0_5_msmt17.pt', 'osnet_x0_25_market1501.pt', 'osnet_x0_25_dukemtmcreid.pt', 'osnet_x0_25_msmt17.pt', 'osnet_ibn_x1_0_msmt17.pt', 'osnet_ain_x1_0_msmt17.pt', 'lmbn_n_duke.pt', 'lmbn_n_market.pt', 'lmbn_n_cuhk03_d.pt', 'clip_market1501.pt', 'clip_duke.pt', 'clip_veri.pt', 'clip_vehicleid.pt'. Suggestions: use "osnet_x1_0_msmt17.pt" for pedestrians or "clip_vehicleid.pt" / "clip_veri.pt" for vehicles.
track_high_thresh	float	High score threshold for first association (default: 0.5). Increasing this is stricter and may reduce false matches but miss tracks.
track_low_thresh	float	Lower score threshold for second association (default: 0.1). Increasing this keeps fewer low-confidence detections.
new_track_thresh	float	Threshold to initialize new tracks (default: 0.6). Increasing this creates fewer new tracks and can reduce false positives.
match_thresh	float	Matching threshold for association (default: 0.8). Increasing this makes association more permissive.
track_buffer	int	Number of frames to keep lost tracks (default: 30). Increasing this preserves IDs longer through occlusion, but may cause stale tracks to survive longer.
with_reid	bool	Whether to enable ReID-assisted association (default: True). Options: True, False. Disabling this speeds up tracking but may increase ID switches.
proximity_thresh	float	Proximity threshold for ReID matching (default: 0.5). Increasing this requires stronger geometric overlap before ReID is used.
appearance_thresh	float	Appearance similarity threshold for ReID matching (default: 0.25). Increasing this requires closer appearance match and is more conservative.

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

ByteTrackConfig dataclass

ByteTrackConfig(
    model: MOTModels = MOTModels.BYTE_TRACK,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    track_thresh: float = 0.5,
    match_thresh: float = 0.8,
    track_buffer: int = 30,
    frame_rate: int = 30,
)

Bases: MOTBaseConfig

ByteTrack-specific parameters.

Attributes:

Name	Type	Description
track_thresh	float	Detection confidence threshold (default: 0.5). Increasing this filters more weak detections.
match_thresh	float	Threshold for matching detections to tracks (default: 0.8). Increasing this generally allows looser matching.
track_buffer	int	Number of frames to keep lost tracks (default: 30). Increasing this keeps unmatched tracks longer.
frame_rate	int	Source video frame rate used by the tracker (default: 30). Set this close to real FPS for best temporal behavior.

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

DeepOCSORTConfig dataclass

DeepOCSORTConfig(
    model: MOTModels = MOTModels.DEEPOCSORT,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    reid_weights: ReIDWeights
    | str
    | None = ReIDWeights.OSNET_X1_0_MSMT17,
    det_thresh: float = 0.3,
    max_age: int = 30,
    min_hits: int = 3,
    iou_threshold: float = 0.3,
    asso_func: str = "iou",
    delta_t: int = 3,
    inertia: float = 0.2,
)

Bases: MOTBaseConfig

DeepOCSORT-specific parameters.

Attributes:

Name	Type	Description
reid_weights	ReIDWeights | str | None	Optional ReID weights path (default: "osnet_x1_0_msmt17.pt"). Options: same built-in .pt names listed in BoTSORTParams.reid_weights.
det_thresh	float	Detection confidence threshold (default: 0.3). Increasing this keeps fewer low-confidence detections.
max_age	int	Maximum age of unmatched tracks (default: 30). Increasing this keeps unmatched tracks alive longer.
min_hits	int	Minimum hits before track confirmation (default: 3). Increasing this delays track confirmation.
iou_threshold	float	IoU threshold for association (default: 0.3). Increasing this requires tighter overlap.
asso_func	str	Association function name (default: "iou"). Typical options: "iou", "giou", "diou", "ciou", "centroid".
delta_t	int	Time gap used by motion compensation (default: 3). Increasing this smooths over longer temporal windows.
inertia	float	Motion inertia weight (default: 0.2). Increasing this emphasizes velocity continuity.

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

HybridSORTConfig dataclass

HybridSORTConfig(
    model: MOTModels = MOTModels.HYBRIDSORT,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    reid_weights: ReIDWeights
    | str
    | None = ReIDWeights.OSNET_X1_0_MSMT17,
    det_thresh: float = 0.3,
    max_age: int = 30,
    min_hits: int = 3,
    iou_threshold: float = 0.3,
    asso_func: str = "iou",
)

Bases: MOTBaseConfig

HybridSORT-specific parameters.

Attributes:

Name	Type	Description
reid_weights	ReIDWeights | str | None	Optional ReID weights path (default: "osnet_x1_0_msmt17.pt"). Options: same built-in .pt names listed in BoTSORTParams.reid_weights.
det_thresh	float	Detection confidence threshold (default: 0.3). Increasing this reduces weak detections.
max_age	int	Maximum age of unmatched tracks (default: 30). Increasing this keeps tracks longer during occlusion.
min_hits	int	Minimum hits before track confirmation (default: 3). Increasing this reduces early noisy tracks.
iou_threshold	float	IoU threshold for association (default: 0.3). Increasing this makes IoU matching stricter.
asso_func	str	Association function name (default: "iou"). Typical options: "iou", "giou", "diou", "ciou", "centroid".

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

MOTBaseConfig dataclass

MOTBaseConfig(
    model: MOTModels = MOTModels.BOTSORT,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
)

Common configuration fields for BoxMOT tracker creation.

Attributes:

Name	Type	Description
model	MOTModels	BoxMOT tracker backend for this parameter bundle (default: MOTModels.BOTSORT).
per_class	bool	Whether to run tracking independently per class (default: False). Options: True, False. Setting True reduces cross-class ID switches but can create more tracks.
extra_kwargs	dict[str, Any]	Additional kwargs merged into tracker construction arguments (default: {}). Use this for BoxMOT arguments not explicitly represented in dataclasses.

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

MOTModels

Bases: StrEnum

Supported tracker backends exposed by BoxMOT.

Attributes:

Name	Type	Description
BOTSORT	str	BoT-SORT tracker name used by BoxMOT. Good default when you want motion + appearance matching.
BOOSTTRACK	str	BoostTrack tracker name used by BoxMOT. Usually improves association under difficult motion/crowding.
BYTE_TRACK	str	ByteTrack tracker name used by BoxMOT. Faster and simpler; does not require ReID weights.
OCSORT	str	OCSORT tracker name used by BoxMOT. Motion-centric tracker; useful when appearance features are unreliable.
STRONGSORT	str	StrongSORT tracker name used by BoxMOT. Appearance-heavy tracker; typically more robust to long occlusions.
DEEPOCSORT	str	DeepOCSORT tracker name used by BoxMOT. OCSORT variant enhanced with appearance features.
HYBRIDSORT	str	HybridSORT tracker name used by BoxMOT. Hybrid strategy between motion and appearance matching.
SFSORT	str	SFSort tracker name used by BoxMOT. Lightweight motion-centric tracking for real-time pipelines.

OCSORTConfig dataclass

OCSORTConfig(
    model: MOTModels = MOTModels.OCSORT,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    det_thresh: float = 0.3,
    max_age: int = 30,
    min_hits: int = 3,
    iou_threshold: float = 0.3,
    asso_func: str = "iou",
    delta_t: int = 3,
    inertia: float = 0.2,
)

Bases: MOTBaseConfig

OCSORT-specific parameters.

Attributes:

Name	Type	Description
det_thresh	float	Detection confidence threshold (default: 0.3). Increasing this reduces low-confidence detections.
max_age	int	Maximum age of unmatched tracks (default: 30). Increasing this keeps tracks alive through longer gaps.
min_hits	int	Minimum hits before track confirmation (default: 3). Increasing this reduces short-lived false tracks.
iou_threshold	float	IoU threshold for association (default: 0.3). Increasing this makes matching stricter.
asso_func	str	Association function name (default: "iou"). Typical options: "iou", "giou", "diou", "ciou", "centroid".
delta_t	int	Time gap used by motion compensation (default: 3). Increasing this smooths longer motion history, but may lag quick turns.
inertia	float	Motion inertia weight (default: 0.2). Increasing this trusts previous velocity more.

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

ReIDWeights

Bases: StrEnum

Built-in BoxMOT ReID weight file names.

Use these enum values for reid_weights in tracker parameter dataclasses.

SFSORTConfig dataclass

SFSORTConfig(
    model: MOTModels = MOTModels.SFSORT,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    det_thresh: float = 0.3,
    max_age: int = 30,
    min_hits: int = 3,
    iou_threshold: float = 0.3,
    asso_func: str = "iou",
)

Bases: MOTBaseConfig

SFSORT-specific parameters.

Attributes:

Name	Type	Description
det_thresh	float	Detection confidence threshold (default: 0.3). Increasing this reduces weak detections.
max_age	int	Maximum age of unmatched tracks (default: 30). Increasing this keeps tracks longer through brief misses.
min_hits	int	Minimum hits before track confirmation (default: 3). Increasing this reduces short-lived noisy tracks.
iou_threshold	float	IoU threshold for association (default: 0.3). Increasing this requires tighter overlap for matching.
asso_func	str	Association function name (default: "iou"). Typical options: "iou", "giou", "diou", "ciou", "centroid".

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

StrongSORTConfig dataclass

StrongSORTConfig(
    model: MOTModels = MOTModels.STRONGSORT,
    per_class: bool = False,
    extra_kwargs: dict[str, Any] = dict(),
    reid_weights: ReIDWeights
    | str
    | None = ReIDWeights.OSNET_X1_0_MSMT17,
    max_dist: float = 0.2,
    max_iou_dist: float = 0.7,
    max_age: int = 70,
    n_init: int = 3,
    nn_budget: int = 100,
    ema_alpha: float = 0.9,
    mc_lambda: float = 0.995,
)

Bases: MOTBaseConfig

StrongSORT-specific parameters.

Attributes:

Name	Type	Description
reid_weights	ReIDWeights | str | None	Optional ReID weights path (default: "osnet_x1_0_msmt17.pt"). Options: same built-in .pt names listed in BoTSORTParams.reid_weights.
max_dist	float	Maximum cosine distance for appearance matching (default: 0.2). Increasing this allows less similar appearance matches.
max_iou_dist	float	Maximum IoU distance for geometric matching (default: 0.7). Increasing this allows looser geometric matches.
max_age	int	Maximum age of unmatched tracks (default: 70). Increasing this retains tracks through longer occlusions.
n_init	int	Minimum hits before track confirmation (default: 3). Increasing this delays confirmation and reduces unstable IDs.
nn_budget	int	Maximum size of appearance feature gallery (default: 100). Increasing this improves long-term matching memory at higher memory cost.
ema_alpha	float	EMA factor for appearance embeddings (default: 0.9). Increasing this smooths features more and reduces noise.
mc_lambda	float	Motion compensation blending factor (default: 0.995). Increasing this gives more weight to motion compensation.

to_kwargs

to_kwargs() -> dict[str, Any]

Convert dataclass fields to keyword arguments for BoxMOT tracker creation.

Source code in src/dnt/track/tracker.py

def to_kwargs(self) -> dict[str, Any]:
    """Convert dataclass fields to keyword arguments for BoxMOT tracker creation."""
    kwargs = asdict(self)
    kwargs.pop("model", None)
    kwargs.pop("extra_kwargs", None)
    kwargs.update(self.extra_kwargs)
    return kwargs

to_dict

to_dict() -> dict[str, Any]

Return dataclass values as a serializable dictionary.

Source code in src/dnt/track/tracker.py

def to_dict(self) -> dict[str, Any]:
    """Return dataclass values as a serializable dictionary."""
    return self._yaml_safe(asdict(self))

from_dict classmethod

from_dict(data: dict[str, Any]) -> MOTBaseConfig

Build a parameter object from a dictionary.

Unknown keys are stored in extra_kwargs.

Source code in src/dnt/track/tracker.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MOTBaseConfig":
    """Build a parameter object from a dictionary.

    Unknown keys are stored in `extra_kwargs`.
    """
    valid_fields = {f.name for f in fields(cls)}
    known = {k: v for k, v in data.items() if k in valid_fields}
    unknown = {k: v for k, v in data.items() if k not in valid_fields}

    if "model" in known and not isinstance(known["model"], MOTModels):
        known["model"] = MOTModels(str(known["model"]))

    params = cls(**known)
    if unknown:
        params.extra_kwargs.update(unknown)
    return params

export_yaml

export_yaml(yaml_file: str) -> None

Export parameters to a YAML file.

Source code in src/dnt/track/tracker.py

def export_yaml(self, yaml_file: str) -> None:
    """Export parameters to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(self.to_dict(), f, sort_keys=False)

import_yaml classmethod

import_yaml(yaml_file: str) -> MOTBaseConfig

Import parameters from a YAML file.

Source code in src/dnt/track/tracker.py

@classmethod
def import_yaml(cls, yaml_file: str) -> "MOTBaseConfig":
    """Import parameters from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)
    return cls.from_dict(data)

Tracker

Tracker(
    config: BoxMOTModelParams | None = None,
    config_yaml: str | None = None,
    device: str = "auto",
    half: bool = False,
    output_score_cls: bool = True,
    boxmot_verbose: bool = False,
)

Unified interface for BoxMOT tracking and track post-processing.

This class runs BoxMOT tracking given a detection file and source video. It also provides post-processing utilities to infill missing frames, split tracks by large gaps, and drop short tracks.

Attributes:

Name	Type	Description
TRACK_FIELDS	list[str]	Standard output columns for tracking and post-processing utilities (default: class constant).
device	str	Device string used by deep trackers (default: "auto"). Options: "auto", "cpu", "cuda", "mps".
half	bool	Whether half precision is enabled for deep trackers (default: False). Options: True, False. Enabling can improve speed on supported GPUs.
boxmot_model	MOTModels	Selected BoxMOT tracker backend (default: MOTModels.BOTSORT). Options: MOTModels.BOTSORT, MOTModels.BOOSTTRACK, MOTModels.BYTE_TRACK, MOTModels.OCSORT, MOTModels.STRONGSORT, MOTModels.DEEPOCSORT, MOTModels.HYBRIDSORT, MOTModels.SFSORT.
boxmot_config	BoxMOTModelConfig	Configuration dataclass instance for BoxMOT tracker creation (default: model-specific defaults).
boxmot_verbose	bool	If False, suppress BoxMOT INFO/SUCCESS logging output.
output_score_cls	bool	Whether to include tracker score and cls values in outputs. If False, both fields are exported as -1 to keep file schema stable.
REID_WEIGHTS_DIR	Path	Directory where relative ReID weights are resolved and stored.
DEFAULT_REID_WEIGHT	str	Fallback ReID weight file name used when a model expects ReID and no weight is explicitly set.

Initialize the tracker.

Parameters:

Name	Type	Description	Default
config	BoxMOTModelConfig	Configuration bundle for BoxMOT tracker creation. Tracker backend is selected from config.model.	None
config_yaml	str	YAML file containing model-aware config. When provided, values loaded from YAML override config input.	None
device	str	Device string used by deep trackers (default: "auto", "cpu", "cuda", "mps").	'auto'
half	bool	Whether half precision is enabled for deep trackers (default: False).	False
output_score_cls	bool	If True, output tracker confidence and class values in score and cls columns. If False, export -1 for both fields.	True
boxmot_verbose	bool	If False, suppress BoxMOT INFO/SUCCESS logging output.	False

Source code in src/dnt/track/tracker.py

def __init__(
    self,
    config: BoxMOTModelParams | None = None,
    config_yaml: str | None = None,
    device: str = "auto",
    half: bool = False,
    output_score_cls: bool = True,
    boxmot_verbose: bool = False,
) -> None:
    """Initialize the tracker.

    Parameters
    ----------
    config : BoxMOTModelConfig, optional
        Configuration bundle for BoxMOT tracker creation. Tracker backend
        is selected from `config.model`.
    config_yaml : str, optional
        YAML file containing model-aware config. When provided,
        values loaded from YAML override `config` input.
    device : str, optional
        Device string used by deep trackers (default: `"auto"`, "cpu", "cuda", "mps").
    half : bool, optional
        Whether half precision is enabled for deep trackers (default: `False`).
    output_score_cls : bool, optional
        If True, output tracker confidence and class values in `score` and
        `cls` columns. If False, export `-1` for both fields.
    boxmot_verbose : bool, optional
        If False, suppress BoxMOT INFO/SUCCESS logging output.

    """
    self.device = device
    self.half = half
    self.boxmot_verbose = boxmot_verbose
    self.output_score_cls = output_score_cls
    yaml_path = config_yaml
    resolved_config = config or config
    self.model_config_yaml = yaml_path

    if yaml_path:
        resolved_config = self.import_config_from_yaml(yaml_path)

    if resolved_config is None:
        resolved_config = self._default_boxmot_config()

    self.boxmot_model = resolved_config.model
    self.boxmot_config = resolved_config

track

track(
    det_file: str,
    out_file: str,
    video_file: str | None = None,
    show: bool = False,
    video_index: int | None = None,
    video_tot: int | None = None,
    message: str | None = None,
) -> pd.DataFrame

Run tracking on a single detection file using BoxMOT.

Parameters:

Name	Type	Description	Default
det_file	str	Path to detection file in DNT detection format (frame, -, x, y, width, height, confidence, class_id).	required
out_file	str	Path to write tracking results. If empty string, results are not saved.	required
video_file	str	Path to source video file. Required for BoxMOT tracker.	None
show	bool	If True (default: False), display live tracking preview with bounding boxes and track IDs. Press 's' to toggle preview, 'ESC' to hide, 'q' to stop tracking early.	False
video_index	int	Index of current video in batch (for progress bar display).	None
video_tot	int	Total number of videos in batch (for complete progress context).	None
message	str | None	Optional progress text shown in the progress bar. If None, the video stem is used (default: None).	None

Returns:

Type	Description
DataFrame	Tracking results with columns: frame, track, x, y, w, h, score, cls, r3, r4 Each row represents one detected object per frame.

Raises:

Type	Description
FileNotFoundError	If det_file or video_file does not exist.
ValueError	If video_file is None.

Notes

The tracker processes detections frame-by-frame, maintaining track IDs across frames. Detection coordinates are converted from (x1, y1, x2, y2) to (x, y, width, height) format for BoxMOT.

Track IDs are persistent across frame sequences and reused if tracks are lost and then re-acquired within track_buffer frames.

Source code in src/dnt/track/tracker.py

def track(
    self,
    det_file: str,
    out_file: str,
    video_file: str | None = None,
    show: bool = False,
    video_index: int | None = None,
    video_tot: int | None = None,
    message: str | None = None,
) -> pd.DataFrame:
    """Run tracking on a single detection file using BoxMOT.

    Parameters
    ----------
    det_file : str
        Path to detection file in DNT detection format
        (frame, -, x, y, width, height, confidence, class_id).
    out_file : str
        Path to write tracking results. If empty string, results are not saved.
    video_file : str, optional
        Path to source video file. Required for BoxMOT tracker.
    show : bool, optional
        If True (default: False), display live tracking preview with bounding
        boxes and track IDs. Press 's' to toggle preview, 'ESC' to hide,
        'q' to stop tracking early.
    video_index : int, optional
        Index of current video in batch (for progress bar display).
    video_tot : int, optional
        Total number of videos in batch (for complete progress context).
    message : str | None, optional
        Optional progress text shown in the progress bar. If None, the
        video stem is used (default: None).

    Returns
    -------
    pd.DataFrame
        Tracking results with columns: frame, track, x, y, w, h, score, cls, r3, r4
        Each row represents one detected object per frame.

    Raises
    ------
    FileNotFoundError
        If det_file or video_file does not exist.
    ValueError
        If video_file is None.

    Notes
    -----
    The tracker processes detections frame-by-frame, maintaining track IDs
    across frames. Detection coordinates are converted from (x1, y1, x2, y2)
    to (x, y, width, height) format for BoxMOT.

    Track IDs are persistent across frame sequences and reused if tracks
    are lost and then re-acquired within track_buffer frames.

    """
    if not Path(det_file).exists():
        msg = f"Detection file not found: {det_file}"
        raise FileNotFoundError(msg)

    if video_file is None:
        msg = "Video file required for BoxMOT tracking but not provided."
        raise ValueError(msg)
    if not Path(video_file).exists():
        msg = f"Video file not found: {video_file}"
        raise FileNotFoundError(msg)

    return self._track_boxmot(
        video_file=video_file,
        det_file=det_file,
        out_file=out_file,
        show=show,
        video_index=video_index,
        video_tot=video_tot,
        message=message,
    )

track_batch

track_batch(
    det_files: list[str] | None = None,
    video_files: list[str] | None = None,
    output_path: str | None = None,
    is_overwrite: bool = False,
    is_report: bool = True,
    message: str | None = None,
) -> list[str]

Run tracking on multiple detection files sequentially.

Parameters:

Name	Type	Description	Default
det_files	list[str] | None	List of detection file paths. Each file should contain frame-level detections in CSV format. If None (default), returns empty list.	None
video_files	list[str] | None	List of corresponding source video file paths for each detection file. Length should match det_files. Required for BoxMOT tracking.	None
output_path	str | None	Directory to save tracking results. Track files are named based on input filename with '_track.txt' suffix. If None (default), tracking still runs but results are not persisted.	None
is_overwrite	bool	If False (default), skip tracking for videos with existing output files.	False
is_report	bool	If True (default), include skipped files in returned list.	True
message	str | None	Optional progress text shown in each tracking progress bar. If None (default), each video's stem is used.	None

Returns:

Type	Description
list[str]	List of output track file paths. Includes both newly created and existing files (if is_report=True). Empty list if det_files is None.

Notes

Processing is sequential (not parallel). Each detection file is tracked in order with progress display showing "Tracking X of Y".

Files matching between det_files and video_files by index position. If video_files is shorter than det_files, missing videos are left None and those detections are skipped.

Source code in src/dnt/track/tracker.py

def track_batch(
    self,
    det_files: list[str] | None = None,
    video_files: list[str] | None = None,
    output_path: str | None = None,
    is_overwrite: bool = False,
    is_report: bool = True,
    message: str | None = None,
) -> list[str]:
    """Run tracking on multiple detection files sequentially.

    Parameters
    ----------
    det_files : list[str] | None, optional
        List of detection file paths. Each file should contain frame-level
        detections in CSV format. If None (default), returns empty list.
    video_files : list[str] | None, optional
        List of corresponding source video file paths for each detection file.
        Length should match det_files. Required for BoxMOT tracking.
    output_path : str | None, optional
        Directory to save tracking results. Track files are named based on
        input filename with '_track.txt' suffix. If None (default),
        tracking still runs but results are not persisted.
    is_overwrite : bool, optional
        If False (default), skip tracking for videos with existing output files.
    is_report : bool, optional
        If True (default), include skipped files in returned list.
    message : str | None, optional
        Optional progress text shown in each tracking progress bar.
        If None (default), each video's stem is used.

    Returns
    -------
    list[str]
        List of output track file paths. Includes both newly created and
        existing files (if is_report=True). Empty list if det_files is None.

    Notes
    -----
    Processing is sequential (not parallel). Each detection file is tracked
    in order with progress display showing "Tracking X of Y".

    Files matching between det_files and video_files by index position.
    If video_files is shorter than det_files, missing videos are left None
    and those detections are skipped.

    """
    if det_files is None:
        return []

    results: list[str] = []
    total_videos = len(det_files)

    for idx, det_file in enumerate(det_files, start=1):
        base_filename = os.path.splitext(os.path.basename(det_file))[0].replace("_iou", "")

        track_file = None
        if output_path:
            os.makedirs(output_path, exist_ok=True)
            track_file = os.path.join(output_path, base_filename + "_track.txt")

        if track_file and not is_overwrite and os.path.exists(track_file):
            if is_report:
                results.append(track_file)
            continue

        # BoxMOT requires a matching source video.
        video_file = None
        if video_files is not None:  # noqa: SIM102
            if idx - 1 < len(video_files):
                video_file = video_files[idx - 1]

        # run tracking
        self.track(
            det_file=det_file,
            out_file=track_file if track_file else "",  # track() expects a path
            video_file=video_file,
            video_index=idx,
            video_tot=total_videos,
            message=message,
        )

        if track_file:
            results.append(track_file)

    return results

export_config_to_yaml staticmethod

export_config_to_yaml(
    yaml_file: str, config: BoxMOTModelParams
) -> None

Export model-aware BoxMOT config to a YAML file.

Source code in src/dnt/track/tracker.py

@staticmethod
def export_config_to_yaml(
    yaml_file: str,
    config: BoxMOTModelParams,
) -> None:
    """Export model-aware BoxMOT config to a YAML file."""
    out_path = Path(yaml_file)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8") as f:
        yaml.safe_dump(config.to_dict(), f, sort_keys=False)

import_config_from_yaml staticmethod

import_config_from_yaml(
    yaml_file: str,
) -> BoxMOTModelParams

Import model-aware BoxMOT config from a YAML file.

Source code in src/dnt/track/tracker.py

@staticmethod
def import_config_from_yaml(yaml_file: str) -> BoxMOTModelParams:
    """Import model-aware BoxMOT config from a YAML file."""
    with Path(yaml_file).open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f) or {}
    if not isinstance(data, dict):
        msg = f"Invalid YAML content in {yaml_file}: expected a mapping."
        raise ValueError(msg)

    model = MOTModels(str(data.get("model", MOTModels.BOTSORT.value)))
    param_cls = Tracker._params_class_for_model(model)
    return param_cls.from_dict(data)

export_params_to_yaml staticmethod

export_params_to_yaml(
    yaml_file: str, params: BoxMOTModelParams
) -> None

Export model-aware BoxMOT params to a YAML file (backward-compatible wrapper).

Source code in src/dnt/track/tracker.py

@staticmethod
def export_params_to_yaml(
    yaml_file: str,
    params: BoxMOTModelParams,
) -> None:
    """Export model-aware BoxMOT params to a YAML file (backward-compatible wrapper)."""
    Tracker.export_config_to_yaml(yaml_file=yaml_file, config=params)

import_params_from_yaml staticmethod

import_params_from_yaml(
    yaml_file: str,
) -> BoxMOTModelParams

Import model-aware BoxMOT params from a YAML file (backward-compatible wrapper).

Source code in src/dnt/track/tracker.py

@staticmethod
def import_params_from_yaml(yaml_file: str) -> BoxMOTModelParams:
    """Import model-aware BoxMOT params from a YAML file (backward-compatible wrapper)."""
    return Tracker.import_config_from_yaml(yaml_file=yaml_file)

export_current_config_to_yaml

export_current_config_to_yaml(yaml_file: str) -> None

Export this tracker's active model and config to YAML.

Source code in src/dnt/track/tracker.py

def export_current_config_to_yaml(self, yaml_file: str) -> None:
    """Export this tracker's active model and config to YAML."""
    self.export_config_to_yaml(
        yaml_file=yaml_file,
        config=self.boxmot_config,
    )

export_current_params_to_yaml

export_current_params_to_yaml(yaml_file: str) -> None

Export this tracker's active model and config to YAML (backward-compatible wrapper).

Source code in src/dnt/track/tracker.py

def export_current_params_to_yaml(self, yaml_file: str) -> None:
    """Export this tracker's active model and config to YAML (backward-compatible wrapper)."""
    self.export_current_config_to_yaml(yaml_file)

interpolate_tracks_rts

interpolate_tracks_rts(
    tracks: DataFrame | None = None,
    track_file: str | None = None,
    output_file: str | None = None,
    col_names: list[str] | None = None,
    fill_gaps_only: bool = True,
    smooth_existing: bool = False,
    process_var: float = 10.0,
    meas_var_pos: float = 25.0,
    meas_var_size: float = 16.0,
    min_track_len: int = 2,
    max_gap: int = 30,
    add_interp_flag: bool = True,
    interp_col: str = "interp",
    verbose: bool = True,
    video_index: int | None = None,
    video_tot: int | None = None,
) -> pd.DataFrame

Interpolate trajectory gaps in each track chain using RTS smoothing.

Applies a constant-velocity Kalman filter per track on bounding box center and size states, then runs Rauch-Tung-Striebel (RTS) smoothing from FilterPy to produce smooth, continuous trajectories. Missing frames are interpolated with velocity estimates.

Parameters:

Name	Type	Description	Default
tracks	DataFrame	Input track data with columns at minimum: frame, track, x, y, w, h. May also contain cls, score, and other columns which are preserved. If None, track_file is used.	None
track_file	str	CSV file path to read tracks from when tracks is None.	None
output_file	str	CSV file path to write the interpolated results.	None
col_names	list[str]	Column names to apply when input columns are positional integers. Default is ["frame","track","x","y","w","h","score","cls","r3","r4"].	None
fill_gaps_only	bool	If True (default), only interpolate frames without observations. If False, also smooth observed frames.	True
smooth_existing	bool	If True, apply smoothed state to observed frames. Only used when fill_gaps_only is True. Default is False.	False
process_var	float	Process noise variance for Kalman filter. Controls model uncertainty. Default is 10.0.	10.0
meas_var_pos	float	Measurement noise variance for position (cx, cy). Default is 25.0.	25.0
meas_var_size	float	Measurement noise variance for size (w, h). Default is 16.0.	16.0
min_track_len	int	Minimum track length to apply interpolation. Tracks shorter than this are returned as-is. Default is 2.	2
max_gap	int	Maximum number of consecutive missing frames allowed to interpolate within a track chain. Gaps larger than this value are not filled. Default is 30.	30
add_interp_flag	bool	If True (default), add column with interpolation flags (0=observed, 1=interpolated).	True
interp_col	str	Name of the interpolation flag column. Default is "interp".	'interp'
verbose	bool	If True, show tqdm progress bar over tracks. Default is True.	True
video_index	int	Current video index for progress description. Default is None.	None
video_tot	int	Total videos for progress description. Default is None.	None

Returns:

Type	Description
DataFrame	Output tracks with interpolated frames. Columns include all input columns plus interp_col if add_interp_flag is True. Frame indices are continuous within each track after interpolation.

Raises:

Type	Description
ValueError	If tracks has fewer than 6 columns (when columns are not named).

Notes

The Kalman filter uses an 8-state constant-velocity model: [cx, vx, cy, vy, w, vw, h, vh] where (cx, cy) is bounding box center, (w, h) is size, and (vx, vy, vw, vh) are their velocities.

Input coordinates assume [x, y, w, h] format where x, y is top-left corner. These are converted to center coordinates for Kalman processing.

Frame gaps within tracks are filled by interpolation. If a track has missing frames between observations, the filter predicts values for those frames based on velocity estimates from nearby observations.

Examples:

>>> import pandas as pd
>>> import numpy as np
>>> # Create sample track with gaps
>>> tracks = pd.DataFrame({
...     'frame': [0, 1, 5, 6],
...     'track': [1, 1, 1, 1],
...     'x': [10.0, 12.0, 20.0, 22.0],
...     'y': [20.0, 22.0, 30.0, 32.0],
...     'w': [100.0, 100.0, 100.0, 100.0],
...     'h': [50.0, 50.0, 50.0, 50.0],
... })
>>> result = interpolate_tracks_rts(tracks, fill_gaps_only=True)
>>> print(result[['frame', 'track', 'interp']])  # Shows interpolated frames

Source code in src/dnt/track/post_process.py

def interpolate_tracks_rts(
    tracks: pd.DataFrame | None = None,
    track_file: str | None = None,
    output_file: str | None = None,
    col_names: list[str] | None = None,
    fill_gaps_only: bool = True,
    smooth_existing: bool = False,
    process_var: float = 10.0,
    meas_var_pos: float = 25.0,
    meas_var_size: float = 16.0,
    min_track_len: int = 2,
    max_gap: int = 30,
    add_interp_flag: bool = True,
    interp_col: str = "interp",
    verbose: bool = True,
    video_index: int | None = None,
    video_tot: int | None = None,
) -> pd.DataFrame:
    """Interpolate trajectory gaps in each track chain using RTS smoothing.

    Applies a constant-velocity Kalman filter per track on bounding box center
    and size states, then runs Rauch-Tung-Striebel (RTS) smoothing from FilterPy
    to produce smooth, continuous trajectories. Missing frames are interpolated
    with velocity estimates.

    Parameters
    ----------
    tracks : pd.DataFrame, optional
        Input track data with columns at minimum: frame, track, x, y, w, h.
        May also contain cls, score, and other columns which are preserved.
        If None, ``track_file`` is used.
    track_file : str, optional
        CSV file path to read tracks from when ``tracks`` is None.
    output_file : str, optional
        CSV file path to write the interpolated results.
    col_names : list[str], optional
        Column names to apply when input columns are positional integers.
        Default is ["frame","track","x","y","w","h","score","cls","r3","r4"].
    fill_gaps_only : bool, optional
        If True (default), only interpolate frames without observations.
        If False, also smooth observed frames.
    smooth_existing : bool, optional
        If True, apply smoothed state to observed frames. Only used when
        fill_gaps_only is True. Default is False.
    process_var : float, optional
        Process noise variance for Kalman filter. Controls model uncertainty.
        Default is 10.0.
    meas_var_pos : float, optional
        Measurement noise variance for position (cx, cy). Default is 25.0.
    meas_var_size : float, optional
        Measurement noise variance for size (w, h). Default is 16.0.
    min_track_len : int, optional
        Minimum track length to apply interpolation. Tracks shorter than this
        are returned as-is. Default is 2.
    max_gap : int, optional
        Maximum number of consecutive missing frames allowed to interpolate
        within a track chain. Gaps larger than this value are not filled.
        Default is 30.
    add_interp_flag : bool, optional
        If True (default), add column with interpolation flags (0=observed, 1=interpolated).
    interp_col : str, optional
        Name of the interpolation flag column. Default is "interp".
    verbose : bool, optional
        If True, show tqdm progress bar over tracks. Default is True.
    video_index : int, optional
        Current video index for progress description. Default is None.
    video_tot : int, optional
        Total videos for progress description. Default is None.

    Returns
    -------
    pd.DataFrame
        Output tracks with interpolated frames. Columns include all input
        columns plus interp_col if add_interp_flag is True. Frame indices are
        continuous within each track after interpolation.

    Raises
    ------
    ValueError
        If tracks has fewer than 6 columns (when columns are not named).

    Notes
    -----
    The Kalman filter uses an 8-state constant-velocity model:
        [cx, vx, cy, vy, w, vw, h, vh]
    where (cx, cy) is bounding box center, (w, h) is size, and
    (vx, vy, vw, vh) are their velocities.

    Input coordinates assume [x, y, w, h] format where x, y is top-left corner.
    These are converted to center coordinates for Kalman processing.

    Frame gaps within tracks are filled by interpolation. If a track has
    missing frames between observations, the filter predicts values for those
    frames based on velocity estimates from nearby observations.

    Examples
    --------
    >>> import pandas as pd
    >>> import numpy as np
    >>> # Create sample track with gaps
    >>> tracks = pd.DataFrame({
    ...     'frame': [0, 1, 5, 6],
    ...     'track': [1, 1, 1, 1],
    ...     'x': [10.0, 12.0, 20.0, 22.0],
    ...     'y': [20.0, 22.0, 30.0, 32.0],
    ...     'w': [100.0, 100.0, 100.0, 100.0],
    ...     'h': [50.0, 50.0, 50.0, 50.0],
    ... })
    >>> result = interpolate_tracks_rts(tracks, fill_gaps_only=True)
    >>> print(result[['frame', 'track', 'interp']])  # Shows interpolated frames

    """
    from filterpy.common import Q_discrete_white_noise
    from filterpy.kalman import KalmanFilter, rts_smoother

    if col_names is None:
        col_names = ["frame", "track", "x", "y", "w", "h", "score", "cls", "r3", "r4"]

    if tracks is None:
        if not track_file:
            raise ValueError("Either `tracks` or `track_file` must be provided.")
        tracks = pd.read_csv(track_file)

    if len(tracks) == 0:
        out = tracks.copy()
        if output_file:
            out.to_csv(output_file, index=False)
        return out

    df = tracks.copy()

    # Support both positional and named-column track tables.
    required = ["frame", "track", "x", "y", "w", "h"]
    if all(c in df.columns for c in required):
        work = df.copy()
    else:
        if len(df.columns) < len(required):
            raise ValueError("tracks must include at least frame/track/x/y/w/h columns.")
        renamed = col_names[: len(df.columns)]
        work = df.copy()
        work.columns = renamed

    work = work.sort_values(["track", "frame"]).reset_index(drop=True)
    output_rows: list[dict] = []
    grouped = list(work.groupby("track", sort=False))

    pbar = tqdm(total=len(grouped), unit=" tracks", disable=not verbose)
    if verbose:
        if video_index is not None and video_tot is not None:
            pbar.set_description_str(f"RTS interpolate {video_index} of {video_tot}")
        else:
            pbar.set_description_str("RTS interpolate")

    for track_id, g in grouped:
        g = g.sort_values("frame").drop_duplicates("frame", keep="first").reset_index(drop=True)
        if len(g) < min_track_len:
            rows = g.to_dict("records")
            if add_interp_flag:
                for r in rows:
                    if "r3" in g.columns:
                        r["r3"] = 0
                    else:
                        r[interp_col] = 0
            output_rows.extend(rows)
            pbar.update(1)
            continue

        frames_obs = g["frame"].astype(int).to_numpy()
        frame_start = int(frames_obs.min())
        frame_end = int(frames_obs.max())
        frames_full = np.arange(frame_start, frame_end + 1, dtype=int)
        observed_set = set(frames_obs.tolist())
        fillable_missing: set[int] = set()
        for f0, f1 in pairwise(frames_obs):
            gap = int(f1 - f0 - 1)
            if 0 < gap <= max_gap:
                fillable_missing.update(range(int(f0) + 1, int(f1)))

        cx = (g["x"].astype(float) + (g["w"].astype(float) / 2.0)).to_numpy()
        cy = (g["y"].astype(float) + (g["h"].astype(float) / 2.0)).to_numpy()
        ww = g["w"].astype(float).to_numpy()
        hh = g["h"].astype(float).to_numpy()
        z_map = {int(f): np.array([cx[i], cy[i], ww[i], hh[i]], dtype=float) for i, f in enumerate(frames_obs)}
        row_map = {int(row["frame"]): row for row in g.to_dict("records")}

        # State: [cx, vx, cy, vy, w, vw, h, vh]
        kf = KalmanFilter(dim_x=8, dim_z=4)
        kf.F = np.array(
            [
                [1, 1, 0, 0, 0, 0, 0, 0],
                [0, 1, 0, 0, 0, 0, 0, 0],
                [0, 0, 1, 1, 0, 0, 0, 0],
                [0, 0, 0, 1, 0, 0, 0, 0],
                [0, 0, 0, 0, 1, 1, 0, 0],
                [0, 0, 0, 0, 0, 1, 0, 0],
                [0, 0, 0, 0, 0, 0, 1, 1],
                [0, 0, 0, 0, 0, 0, 0, 1],
            ],
            dtype=float,
        )
        kf.H = np.array(
            [
                [1, 0, 0, 0, 0, 0, 0, 0],
                [0, 0, 1, 0, 0, 0, 0, 0],
                [0, 0, 0, 0, 1, 0, 0, 0],
                [0, 0, 0, 0, 0, 0, 1, 0],
            ],
            dtype=float,
        )
        q2 = Q_discrete_white_noise(dim=2, dt=1.0, var=process_var)
        kf.Q = np.zeros((8, 8), dtype=float)
        for i in range(4):
            i0 = i * 2
            kf.Q[i0 : i0 + 2, i0 : i0 + 2] = q2
        kf.R = np.diag([meas_var_pos, meas_var_pos, meas_var_size, meas_var_size]).astype(float)
        kf.P = np.eye(8, dtype=float) * 100.0
        z0 = z_map[frame_start]
        kf.x = np.array([z0[0], 0.0, z0[1], 0.0, z0[2], 0.0, z0[3], 0.0], dtype=float)

        xs, ps, fs, qs = [], [], [], []
        for f in frames_full:
            kf.predict()
            z = z_map.get(int(f))
            if z is not None:
                kf.update(z)
            xs.append(kf.x.copy())
            ps.append(kf.P.copy())
            fs.append(kf.F.copy())
            qs.append(kf.Q.copy())

        xs_s, _, _, _ = rts_smoother(np.asarray(xs), np.asarray(ps), np.asarray(fs), np.asarray(qs))

        if "cls" in g.columns and len(g["cls"].dropna()) > 0:
            cls_mode = g["cls"].mode()
            cls_fill = float(cls_mode.iloc[0]) if len(cls_mode) > 0 else -1
        else:
            cls_fill = -1
        score_fill = float(g["score"].mean()) if "score" in g.columns and len(g["score"].dropna()) > 0 else -1.0

        for i, frame in enumerate(frames_full.tolist()):
            sm_cx = float(xs_s[i, 0])
            sm_cy = float(xs_s[i, 2])
            sm_w = max(1.0, float(xs_s[i, 4]))
            sm_h = max(1.0, float(xs_s[i, 6]))
            sm_x = sm_cx - (sm_w / 2.0)
            sm_y = sm_cy - (sm_h / 2.0)

            if frame in observed_set:
                row = dict(row_map[frame])
                if smooth_existing or (not fill_gaps_only):
                    row["x"] = sm_x
                    row["y"] = sm_y
                    row["w"] = sm_w
                    row["h"] = sm_h
                if add_interp_flag:
                    if "r3" in g.columns:
                        row["r3"] = 0
                    else:
                        row[interp_col] = 0
                output_rows.append(row)
            else:
                if frame not in fillable_missing:
                    continue
                row = {c: np.nan for c in g.columns}
                row["frame"] = frame
                row["track"] = track_id
                row["x"] = sm_x
                row["y"] = sm_y
                row["w"] = sm_w
                row["h"] = sm_h
                if "cls" in g.columns:
                    row["cls"] = cls_fill
                if "score" in g.columns:
                    row["score"] = score_fill
                if add_interp_flag:
                    if "r3" in g.columns:
                        row["r3"] = 1
                    else:
                        row[interp_col] = 1
                output_rows.append(row)
        pbar.update(1)

    pbar.close()

    out = pd.DataFrame(output_rows)
    if "r3" in out.columns:
        cols = list(out.columns)
        idx = cols.index("r3")
        out.rename(columns={"r3": interp_col}, inplace=True)
        cols[idx] = interp_col
        out = out[cols]

    # Keep compatibility with legacy track file readers that enforce integer dtypes.
    int_cols = ["frame", "track", "x", "y", "w", "h", "cls", "r4", interp_col]
    for c in int_cols:
        if c in out.columns:
            out[c] = out[c].fillna(-1).round().astype(int)
    if "score" in out.columns:
        out["score"] = out["score"].fillna(-1).astype(float)

    out = out.sort_values(["frame", "track"]).reset_index(drop=True)
    if output_file:
        out.to_csv(output_file, index=False, header=False)
    return out

link_tracklets

link_tracklets(
    tracks: DataFrame | None = None,
    track_file: str | None = None,
    output_file: str | None = None,
    col_names: list[str] | None = None,
    max_gap: int = 20,
    vel_frames: int = 5,
    size_ratio_max: float = 2.0,
    dist_mult: float = 2.5,
    iou_min: float = 0.05,
    w_d: float = 1.0,
    w_iou: float = 1.0,
    w_s: float = 0.3,
    verbose: bool = True,
    video_index: int | None = None,
    video_tot: int | None = None,
) -> pd.DataFrame

Reconnect broken tracklets using global optimal 1-to-1 matching.

Links tracklets (short track segments) by computing a cost matrix based on spatial proximity, appearance similarity (IoU), and size consistency. Uses linear sum assignment (Hungarian algorithm) to find optimal matches, then merges tracklets via union-find to handle transitive connections.

Parameters:

Name	Type	Description	Default
tracks	DataFrame | None	Input track data with columns: frame, track, x, y, w, h, and optionally score, cls, interp, r4. If None (default), track_file is used.	None
track_file	str | None	CSV file path to read tracks from when tracks is None.	None
output_file	str | None	CSV file path to write linked results. If None (default), results are not saved to file.	None
col_names	list[str] | None	Column names to apply when input has positional integer columns. Default: ["frame","track","x","y","w","h","score","cls","interp","r4"].	None
max_gap	int	Maximum frame gap between tracklet end and start to attempt linking. Default is 20.	20
vel_frames	int	Number of recent frames to use for velocity estimation (polynomial fit). Default is 5.	5
size_ratio_max	float	Maximum allowed width/height ratio between tracklet end and start. Default is 2.0. Values outside [1/ratio_max, ratio_max] are rejected.	2.0
dist_mult	float	Distance threshold multiplier: distance_threshold = dist_mult * sqrt(area). Default is 2.5. Larger values allow more spatial flexibility.	2.5
iou_min	float	Minimum Intersection over Union (IoU) between predicted and actual start box. Default is 0.05. Range [0.0, 1.0].	0.05
w_d	float	Weight for normalized distance cost in weighted sum. Default is 1.0.	1.0
w_iou	float	Weight for (1 - IoU) cost in weighted sum. Default is 1.0.	1.0
w_s	float	Weight for size inconsistency cost (log ratio) in weighted sum. Default is 0.3 (smaller weight for size).	0.3
verbose	bool	If True (default), display tqdm progress bar over tracklets.	True
video_index	int | None	Current video index for progress description. Default is None.	None
video_tot	int | None	Total number of videos for progress description. Default is None.	None

Returns:

Type	Description
DataFrame	Output tracks with linked IDs. Same columns as input. Track IDs are remapped so that all frames belonging to a logical track share the same ID. Frame and track are sorted in output.

Raises:

Type	Description
ValueError	If tracks has fewer than 6 columns and no named columns provided.
FileNotFoundError	If track_file path does not exist.

Notes

Algorithm Overview:

1. Extract descriptor for each track: endpoints, velocity, bounding boxes, class
2. Build cost matrix using spatial (distance, IoU), appearance (class), and size (width/height ratio) metrics with weighted combination
3. Solve linear sum assignment problem (Hungarian algorithm) to find optimal 1-to-1 tracklet pairings with minimum total cost
4. Use Union-Find (Disjoint Set Union) to handle transitive merges: if tracklet A links to B and B links to C, they all get merged to same group
5. Remap all track IDs according to merged components

Cost Function Details:

• Velocity is estimated using polynomial fit (1st order) on recent observed frames
• Predicted next tracklet start = end_position + velocity * temporal_gap
• Distance is normalized by sqrt(bounding_box_area) for scale invariance
• Only considers tracklets from same class (if class info available)
• Skips linking if temporal gap, size ratio, or distance threshold exceeded

Input Requirements:

• Requires "frame", "track", "x", "y", "w", "h" columns minimum
• If "interp" column exists, uses only rows with interp==0 for velocity estimation
• If "cls" column exists, only links tracklets with same class

Examples:

>>> import pandas as pd
>>> # Create sample tracklets
>>> tracks = pd.DataFrame({
...     'frame': [0, 1, 10, 11, 20, 21],
...     'track': [1, 1, 2, 2, 3, 3],
...     'x': [10, 12, 25, 27, 40, 42],
...     'y': [20, 22, 35, 37, 50, 52],
...     'w': [50, 50, 50, 50, 50, 50],
...     'h': [100, 100, 100, 100, 100, 100],
...     'cls': [1, 1, 1, 1, 1, 1],
... })
>>> linked = link_tracklets(tracks, max_gap=15, verbose=False)
>>> # Track IDs may now be remapped: e.g., [1, 1, 1, 1, 1, 1]
>>> print(linked['track'].unique())  # All in same track if linked

Source code in src/dnt/track/post_process.py

def link_tracklets(
    tracks: pd.DataFrame | None = None,
    track_file: str | None = None,
    output_file: str | None = None,
    col_names: list[str] | None = None,
    max_gap: int = 20,
    vel_frames: int = 5,
    size_ratio_max: float = 2.0,
    dist_mult: float = 2.5,
    iou_min: float = 0.05,
    w_d: float = 1.0,
    w_iou: float = 1.0,
    w_s: float = 0.3,
    verbose: bool = True,
    video_index: int | None = None,
    video_tot: int | None = None,
) -> pd.DataFrame:
    """Reconnect broken tracklets using global optimal 1-to-1 matching.

    Links tracklets (short track segments) by computing a cost matrix based on
    spatial proximity, appearance similarity (IoU), and size consistency.
    Uses linear sum assignment (Hungarian algorithm) to find optimal matches,
    then merges tracklets via union-find to handle transitive connections.

    Parameters
    ----------
    tracks : pd.DataFrame | None, optional
        Input track data with columns: frame, track, x, y, w, h, and optionally
        score, cls, interp, r4. If None (default), ``track_file`` is used.
    track_file : str | None, optional
        CSV file path to read tracks from when ``tracks`` is None.
    output_file : str | None, optional
        CSV file path to write linked results. If None (default), results
        are not saved to file.
    col_names : list[str] | None, optional
        Column names to apply when input has positional integer columns.
        Default: ["frame","track","x","y","w","h","score","cls","interp","r4"].
    max_gap : int, optional
        Maximum frame gap between tracklet end and start to attempt linking.
        Default is 20.
    vel_frames : int, optional
        Number of recent frames to use for velocity estimation (polynomial fit).
        Default is 5.
    size_ratio_max : float, optional
        Maximum allowed width/height ratio between tracklet end and start.
        Default is 2.0. Values outside [1/ratio_max, ratio_max] are rejected.
    dist_mult : float, optional
        Distance threshold multiplier: distance_threshold = dist_mult * sqrt(area).
        Default is 2.5. Larger values allow more spatial flexibility.
    iou_min : float, optional
        Minimum Intersection over Union (IoU) between predicted and actual start box.
        Default is 0.05. Range [0.0, 1.0].
    w_d : float, optional
        Weight for normalized distance cost in weighted sum. Default is 1.0.
    w_iou : float, optional
        Weight for (1 - IoU) cost in weighted sum. Default is 1.0.
    w_s : float, optional
        Weight for size inconsistency cost (log ratio) in weighted sum.
        Default is 0.3 (smaller weight for size).
    verbose : bool, optional
        If True (default), display tqdm progress bar over tracklets.
    video_index : int | None, optional
        Current video index for progress description. Default is None.
    video_tot : int | None, optional
        Total number of videos for progress description. Default is None.

    Returns
    -------
    pd.DataFrame
        Output tracks with linked IDs. Same columns as input. Track IDs are
        remapped so that all frames belonging to a logical track share the same ID.
        Frame and track are sorted in output.

    Raises
    ------
    ValueError
        If tracks has fewer than 6 columns and no named columns provided.
    FileNotFoundError
        If track_file path does not exist.

    Notes
    -----
    **Algorithm Overview:**

    1. Extract descriptor for each track: endpoints, velocity, bounding boxes, class
    2. Build cost matrix using spatial (distance, IoU), appearance (class), and
       size (width/height ratio) metrics with weighted combination
    3. Solve linear sum assignment problem (Hungarian algorithm) to find optimal
       1-to-1 tracklet pairings with minimum total cost
    4. Use Union-Find (Disjoint Set Union) to handle transitive merges:
       if tracklet A links to B and B links to C, they all get merged to same group
    5. Remap all track IDs according to merged components

    **Cost Function Details:**

    - Velocity is estimated using polynomial fit (1st order) on recent observed frames
    - Predicted next tracklet start = end_position + velocity * temporal_gap
    - Distance is normalized by sqrt(bounding_box_area) for scale invariance
    - Only considers tracklets from same class (if class info available)
    - Skips linking if temporal gap, size ratio, or distance threshold exceeded

    **Input Requirements:**

    - Requires "frame", "track", "x", "y", "w", "h" columns minimum
    - If "interp" column exists, uses only rows with interp==0 for velocity estimation
    - If "cls" column exists, only links tracklets with same class

    Examples
    --------
    >>> import pandas as pd
    >>> # Create sample tracklets
    >>> tracks = pd.DataFrame({
    ...     'frame': [0, 1, 10, 11, 20, 21],
    ...     'track': [1, 1, 2, 2, 3, 3],
    ...     'x': [10, 12, 25, 27, 40, 42],
    ...     'y': [20, 22, 35, 37, 50, 52],
    ...     'w': [50, 50, 50, 50, 50, 50],
    ...     'h': [100, 100, 100, 100, 100, 100],
    ...     'cls': [1, 1, 1, 1, 1, 1],
    ... })
    >>> linked = link_tracklets(tracks, max_gap=15, verbose=False)
    >>> # Track IDs may now be remapped: e.g., [1, 1, 1, 1, 1, 1]
    >>> print(linked['track'].unique())  # All in same track if linked

    """

    def _iou_xywh(a: tuple[float, float, float, float], b: tuple[float, float, float, float]) -> float:
        """Calculate Intersection over Union (IoU) between two bounding boxes.

        Parameters
        ----------
        a : tuple[float, float, float, float]
            Bounding box A as (x, y, width, height).
        b : tuple[float, float, float, float]
            Bounding box B as (x, y, width, height).

        Returns
        -------
        float
            IoU value in range [0.0, 1.0].

        Notes
        -----
        Uses standard IoU formula: intersection / union.
        Coordinates are in (x, y, width, height) format where (x, y) is top-left.

        """
        ax, ay, aw, ah = a
        bx, by, bw, bh = b
        ax2, ay2 = ax + aw, ay + ah
        bx2, by2 = bx + bw, by + bh
        ix1, iy1 = max(ax, bx), max(ay, by)
        ix2, iy2 = min(ax2, bx2), min(ay2, by2)
        iw, ih = max(0.0, ix2 - ix1), max(0.0, iy2 - iy1)
        inter = iw * ih
        union = max(aw * ah + bw * bh - inter, 1e-6)
        return inter / union

    def _estimate_velocity(frames: np.ndarray, cx: np.ndarray, cy: np.ndarray, k: int) -> tuple[float, float]:
        """Estimate velocity using recent observations via polynomial fitting.

        Parameters
        ----------
        frames : np.ndarray
            Array of frame numbers (timestamps) where observations occur.
        cx : np.ndarray
            Array of center x-coordinates corresponding to frames.
        cy : np.ndarray
            Array of center y-coordinates corresponding to frames.
        k : int
            Number of recent frames to use for velocity estimation.
            Uses last k points if available, otherwise uses all points.

        Returns
        -------
        tuple[float, float]
            Velocity (vx, vy) as pixels per frame.
            Returns (0.0, 0.0) if fewer than 2 observations available.

        Notes
        -----
        Uses 1st-order polynomial (linear) fit via np.polyfit for robust
        velocity estimation. Falls back to simple difference (cx[-1]-cx[-2])/dt
        if fitting fails or insufficient unique frame times.

        """
        n = len(frames)
        if n < 2:
            return 0.0, 0.0
        s = max(0, n - k)
        t = frames[s:].astype(float)
        x = cx[s:].astype(float)
        y = cy[s:].astype(float)
        if len(t) < 2 or np.allclose(t, t[0]):
            dt = float(max(frames[-1] - frames[-2], 1))
            return float((cx[-1] - cx[-2]) / dt), float((cy[-1] - cy[-2]) / dt)
        vx = float(np.polyfit(t, x, 1)[0])
        vy = float(np.polyfit(t, y, 1)[0])
        return vx, vy

    class _DSU:
        """Disjoint Set Union (Union-Find) data structure for tracklet merging.

        Efficiently tracks which tracklet IDs belong to the same connected component
        using path compression and union by root heuristics.

        Attributes
        ----------
        parent : dict[int, int]
            Parent map where parent[x] points to parent node. If parent[x] == x,
            then x is a root (representative) of its component.

        Methods
        -------
        find(x: int) -> int
            Find the root representative of x's component with path compression.
        union(a: int, b: int) -> None
            Merge components containing a and b under a's root representative.

        Examples
        --------
        >>> dsu = _DSU([1, 2, 3, 4])
        >>> dsu.union(1, 2)  # Merge components
        >>> dsu.union(2, 3)  # Also connects 1 and 3
        >>> dsu.find(1) == dsu.find(3)  # Both have same root
        True

        """

        def __init__(self, elems: list[int]) -> None:
            """Initialize DSU with elements in separate components.

            Parameters
            ----------
            elems : list[int]
                List of element IDs to initialize. Each starts in its own component.

            """
            self.parent = {e: e for e in elems}

        def find(self, x: int) -> int:
            """Find root representative of x's component with path compression.

            Parameters
            ----------
            x : int
                Element ID to find.

            Returns
            -------
            int
                Root representative (parent[root] == root).

            """
            p = self.parent[x]
            if p != x:
                self.parent[x] = self.find(p)
            return self.parent[x]

        def union(self, a: int, b: int) -> None:
            """Merge components containing a and b under a's root representative.

            Parameters
            ----------
            a : int
                Element in first component.
            b : int
                Element in second component.

            Notes
            -----
            Updates parent[root_b] = root_a so all members of b's component
            now point to a's root as their ultimate parent.

            """
            ra, rb = self.find(a), self.find(b)
            if ra != rb:
                self.parent[rb] = ra

    if col_names is None:
        col_names = ["frame", "track", "x", "y", "w", "h", "score", "cls", "interp", "r4"]

    if tracks is None:
        if not track_file:
            raise ValueError("Either `tracks` or `track_file` must be provided.")
        tracks = pd.read_csv(track_file, header=None)

    if len(tracks) == 0:
        out = tracks.copy()
        if output_file:
            out.to_csv(output_file, index=False, header=False)
        return out

    df = tracks.copy()
    required = ["frame", "track", "x", "y", "w", "h"]
    if not all(c in df.columns for c in required):
        if len(df.columns) < 6:
            raise ValueError("tracks must include at least frame/track/x/y/w/h columns.")
        df.columns = col_names[: len(df.columns)]
    if "cls" not in df.columns and "class" in df.columns:
        df = df.rename(columns={"class": "cls"})
    if "interp" not in df.columns:
        df["interp"] = 0
    else:
        df["interp"] = pd.to_numeric(df["interp"], errors="coerce").fillna(0).astype(int)

    df = df.sort_values(["frame", "track"]).reset_index(drop=True)
    df["cx"] = df["x"].astype(float) + (df["w"].astype(float) / 2.0)
    df["cy"] = df["y"].astype(float) + (df["h"].astype(float) / 2.0)
    df["area"] = df["w"].astype(float) * df["h"].astype(float)

    descriptors: dict[int, dict] = {}
    grouped = list(df.groupby("track", sort=False))
    pbar = tqdm(total=len(grouped), unit=" tracklets", disable=not verbose)
    if verbose:
        if video_index is not None and video_tot is not None:
            pbar.set_description_str(f"Link tracklets {video_index} of {video_tot}")
        else:
            pbar.set_description_str("Link tracklets")

    for tid, g in grouped:
        # Treat only interp==1 as synthesized points; 0/-1 are real detections.
        g_real = g[g["interp"] != 1].sort_values("frame")
        if len(g_real) < 2:
            descriptors[int(tid)] = {"stitchable": False}
            pbar.update(1)
            continue
        t_start = int(g_real["frame"].iloc[0])
        t_end = int(g_real["frame"].iloc[-1])
        start_row = g_real.iloc[0]
        end_row = g_real.iloc[-1]
        vx, vy = _estimate_velocity(
            g_real["frame"].to_numpy(),
            g_real["cx"].to_numpy(),
            g_real["cy"].to_numpy(),
            vel_frames,
        )
        descriptors[int(tid)] = {
            "stitchable": True,
            "track": int(tid),
            "cls": int(end_row["cls"]) if "cls" in g_real.columns else -1,
            "t_start": t_start,
            "t_end": t_end,
            "start_c": (float(start_row["cx"]), float(start_row["cy"])),
            "end_c": (float(end_row["cx"]), float(end_row["cy"])),
            "start_box": (
                float(start_row["x"]),
                float(start_row["y"]),
                float(start_row["w"]),
                float(start_row["h"]),
            ),
            "end_box": (
                float(end_row["x"]),
                float(end_row["y"]),
                float(end_row["w"]),
                float(end_row["h"]),
            ),
            "area_end": max(float(end_row["area"]), 1.0),
            "vx": vx,
            "vy": vy,
        }
        pbar.update(1)
    pbar.close()

    stitchable = [d for d in descriptors.values() if d.get("stitchable", False)]
    if len(stitchable) <= 1:
        out = df.drop(columns=["cx", "cy", "area"])
        if output_file:
            out.to_csv(output_file, index=False, header=False)
        return out

    ends = sorted(stitchable, key=lambda d: (d["t_end"], d["track"]))
    starts = sorted(stitchable, key=lambda d: (d["t_start"], d["track"]))
    n_end, n_start = len(ends), len(starts)
    inf = 1e9
    cost = np.full((n_end, n_start), inf, dtype=float)

    for i, a in enumerate(ends):
        for j, b in enumerate(starts):
            if a["track"] == b["track"]:
                continue
            dt = b["t_start"] - a["t_end"]
            if dt < 1 or dt > max_gap:
                continue
            if a["cls"] != b["cls"]:
                continue
            wi, hi = max(a["end_box"][2], 1.0), max(a["end_box"][3], 1.0)
            wj, hj = max(b["start_box"][2], 1.0), max(b["start_box"][3], 1.0)
            w_ratio, h_ratio = wj / wi, hj / hi
            if not (1.0 / size_ratio_max <= w_ratio <= size_ratio_max):
                continue
            if not (1.0 / size_ratio_max <= h_ratio <= size_ratio_max):
                continue

            pred_cx = a["end_c"][0] + a["vx"] * dt
            pred_cy = a["end_c"][1] + a["vy"] * dt
            sx, sy = b["start_c"]
            dist = float(np.hypot(pred_cx - sx, pred_cy - sy))
            dist_thr = dist_mult * np.sqrt(a["area_end"]) * (1.0 + (0.03 * dt))
            if dist >= dist_thr:
                continue

            pred_box = (pred_cx - (wi / 2.0), pred_cy - (hi / 2.0), wi, hi)
            iou = _iou_xywh(pred_box, b["start_box"])
            if iou < iou_min:
                continue

            dist_norm = dist / (np.sqrt(a["area_end"]) + 1e-6)
            iou_cost = 1.0 - iou
            size_cost = abs(np.log(max(w_ratio, 1e-6))) + abs(np.log(max(h_ratio, 1e-6)))
            c = (w_d * dist_norm) + (w_iou * iou_cost) + (w_s * size_cost)
            cost[i, j] = c

    matches: list[tuple[int, int]] = []
    try:
        from scipy.optimize import linear_sum_assignment

        ri, ci = linear_sum_assignment(cost)
        for r, c in zip(ri, ci, strict=True):
            if cost[r, c] < inf:
                matches.append((r, c))
    except Exception:
        used_r: set[int] = set()
        used_c: set[int] = set()
        finite_pairs = np.argwhere(cost < inf)
        finite_pairs = sorted(finite_pairs, key=lambda rc: float(cost[rc[0], rc[1]]))
        for r, c in finite_pairs:
            r_i, c_i = int(r), int(c)
            if r_i in used_r or c_i in used_c:
                continue
            used_r.add(r_i)
            used_c.add(c_i)
            matches.append((r_i, c_i))

    dsu = _DSU([int(d["track"]) for d in stitchable])
    for r, c in matches:
        a_tid = int(ends[r]["track"])
        b_tid = int(starts[c]["track"])
        dsu.union(a_tid, b_tid)

    comps: dict[int, list[int]] = {}
    for d in stitchable:
        tid = int(d["track"])
        root = dsu.find(tid)
        comps.setdefault(root, []).append(tid)

    tstart_by_tid = {int(d["track"]): int(d["t_start"]) for d in stitchable}
    rep_map: dict[int, int] = {}
    for members in comps.values():
        rep = min(members, key=lambda t: (tstart_by_tid.get(t, 10**9), t))
        for t in members:
            rep_map[t] = rep

    all_tids = df["track"].astype(int).unique().tolist()
    for tid in all_tids:
        rep_map.setdefault(int(tid), int(tid))

    out = df.copy()
    out["track"] = out["track"].astype(int).map(rep_map).astype(int)
    out = out.drop(columns=["cx", "cy", "area"]).sort_values(["frame", "track"]).reset_index(drop=True)

    if output_file:
        out.to_csv(output_file, index=False, header=False)
    return out