High-level API

Auto-generated reference for the high-level Python surface: the Boxmot facade, the composable track(...) and evaluate(...) entry points, and the structured result objects returned by every workflow.

Boxmot facade

Source code in boxmot/api/_facade.py

class Boxmot:
    def __init__(
        self,
        detector: Any = _UNSET,
        reid: Any = _UNSET,
        tracker: Any = _UNSET,
        classes: Any = None,
        project: str | Path = BOXMOT_DEFAULTS.track.project,
    ) -> None:
        self._detector_explicit = detector is not _UNSET and detector is not None
        self._reid_explicit = reid is not _UNSET and reid is not None
        self._tracker_explicit = tracker is not _UNSET and tracker is not None

        self.detector = BOXMOT_DEFAULTS.shared.detector if detector is _UNSET else detector
        self.reid = BOXMOT_DEFAULTS.shared.reid if reid is _UNSET else reid
        self.tracker = BOXMOT_DEFAULTS.track.tracker if tracker is _UNSET else tracker
        self.classes = support.normalize_classes(classes)
        self.project = Path(project)

    def _detector_path(self, required: bool = True) -> Path | None:
        return support.detector_path_from_spec(self.detector, required=required)

    def _reid_path(self, required: bool = True) -> Path | None:
        return support.reid_path_from_spec(self.reid, required=required)

    def _tracker_name(self, required: bool = True) -> str | None:
        return support.tracker_name_from_spec(self.tracker, required=required)

    def _tracker_backend(self, required: bool = True) -> str | None:
        return support.tracker_backend_from_spec(self.tracker, required=required)

    def _tracker_config_from_spec(self) -> dict[str, Any] | None:
        return support.tracker_config_from_spec(self.tracker)

    def _base_eval_args(
        self,
        benchmark: str | Path,
        *,
        imgsz=None,
        conf=None,
        iou: float = BOXMOT_DEFAULTS.eval.iou,
        device: str = BOXMOT_DEFAULTS.eval.device,
        half: bool = BOXMOT_DEFAULTS.eval.half,
        project: str | Path | None = None,
        verbose: bool = BOXMOT_DEFAULTS.eval.verbose,
        show_progress: bool = True,
        postprocessing: str = BOXMOT_DEFAULTS.eval.postprocessing,
        tracker_backend: str | None = None,
        tracking_backend: str = "thread",
    ):
        return adapters.build_eval_args(
            self,
            benchmark,
            imgsz=imgsz,
            conf=conf,
            iou=iou,
            device=device,
            half=half,
            project=project,
            verbose=verbose,
            show_progress=show_progress,
            postprocessing=postprocessing,
            tracker_backend=tracker_backend,
            tracking_backend=tracking_backend,
        )

    def track(
        self,
        *,
        source: Any,
        imgsz=None,
        conf=None,
        iou: float = BOXMOT_DEFAULTS.track.iou,
        device: str = BOXMOT_DEFAULTS.track.device,
        half: bool = BOXMOT_DEFAULTS.track.half,
        save: bool = BOXMOT_DEFAULTS.track.save,
        save_txt: bool = BOXMOT_DEFAULTS.track.save_txt,
        show: bool = BOXMOT_DEFAULTS.track.show,
        drawer=None,
        show_trajectories: bool = False,
        verbose: bool = BOXMOT_DEFAULTS.track.verbose,
        tracker_backend: str | None = None,
    ) -> TrackRunResult:
        args = adapters.build_track_args(
            self,
            source=source,
            imgsz=imgsz,
            conf=conf,
            iou=iou,
            device=device,
            half=half,
            save=save,
            save_txt=save_txt,
            show=show,
            verbose=verbose,
            tracker_backend=tracker_backend,
        )
        return tracker_module.run_track(
            args,
            detector_spec=self.detector,
            reid_spec=self.reid,
            tracker_spec=self.tracker,
            classes=self.classes,
            drawer=drawer,
            show_trajectories=show_trajectories,
        )

    def generate(
        self,
        *,
        benchmark: str | Path | None = None,
        source: str | Path | None = None,
        imgsz=None,
        conf=None,
        iou: float = BOXMOT_DEFAULTS.generate.iou,
        device: str = BOXMOT_DEFAULTS.generate.device,
        half: bool = BOXMOT_DEFAULTS.generate.half,
        project: str | Path | None = None,
        verbose: bool = BOXMOT_DEFAULTS.generate.verbose,
        batch_size: int = BOXMOT_DEFAULTS.generate.batch_size,
        auto_batch: bool = BOXMOT_DEFAULTS.generate.auto_batch,
        resume: bool = BOXMOT_DEFAULTS.generate.resume,
        n_threads: int = BOXMOT_DEFAULTS.generate.n_threads,
    ) -> GenerateResult:
        _validate_generate_inputs(benchmark=benchmark, source=source)
        args = adapters.build_generate_args(
            self,
            benchmark=benchmark,
            source=source,
            imgsz=imgsz,
            conf=conf,
            iou=iou,
            device=device,
            half=half,
            project=project,
            verbose=verbose,
            batch_size=batch_size,
            auto_batch=auto_batch,
            resume=resume,
            n_threads=n_threads,
        )
        timing_stats = cache_module.run_generate(args)
        return GenerateResult(
            benchmark=str(getattr(args, "benchmark", None) or getattr(args, "data", None) or "") or None,
            source=Path(args.source) if getattr(args, "source", None) else None,
            cache_dir=_cache_dir_from_args(args),
            detectors=tuple(Path(detector) for detector in args.detector),
            reid_models=tuple(Path(reid_model) for reid_model in args.reid),
            timings=timing_summary_from_stats(timing_stats),
            args=args,
        )

    def val(
        self,
        *,
        benchmark: str | Path,
        imgsz=None,
        conf=None,
        iou: float = BOXMOT_DEFAULTS.eval.iou,
        device: str = BOXMOT_DEFAULTS.eval.device,
        half: bool = BOXMOT_DEFAULTS.eval.half,
        project: str | Path | None = None,
        verbose: bool = BOXMOT_DEFAULTS.eval.verbose,
        postprocessing: str = BOXMOT_DEFAULTS.eval.postprocessing,
        tracker_backend: str | None = None,
        tracking_backend: str = "thread",
    ) -> ValidationResult:
        args = self._base_eval_args(
            benchmark,
            imgsz=imgsz,
            conf=conf,
            iou=iou,
            device=device,
            half=half,
            project=project,
            verbose=verbose,
            show_progress=True,
            postprocessing=postprocessing,
            tracker_backend=tracker_backend,
            tracking_backend=tracking_backend,
        )
        from boxmot.utils.rich.eval_reporting import EvalWorkflowReporter

        evaluator_module._normalize_eval_models(args)
        pipeline = EvalWorkflowReporter(args).pipeline()
        with pipeline:
            metrics = evaluator_module.run_eval(
                args,
                evolve_config=self._tracker_config_from_spec(),
                pipeline=pipeline,
            )
            metrics.workflow_rendered = True
            return metrics

    def tune(
        self,
        *,
        benchmark: str | Path,
        n_trials: int = BOXMOT_DEFAULTS.tune.n_trials,
        imgsz=None,
        conf=None,
        iou: float = BOXMOT_DEFAULTS.eval.iou,
        device: str = BOXMOT_DEFAULTS.eval.device,
        half: bool = BOXMOT_DEFAULTS.eval.half,
        project: str | Path | None = None,
        maximize: Sequence[str] = BOXMOT_DEFAULTS.tune.maximize,
        minimize: Sequence[str] = BOXMOT_DEFAULTS.tune.minimize,
        verbose: bool = BOXMOT_DEFAULTS.eval.verbose,
        tracker_backend: str | None = None,
        tracking_backend: str = "thread",
        seed: int = 0,
    ) -> TuneResult:
        args = adapters.build_tune_args(
            self,
            benchmark,
            n_trials=n_trials,
            imgsz=imgsz,
            conf=conf,
            iou=iou,
            device=device,
            half=half,
            project=project,
            maximize=maximize,
            minimize=minimize,
            verbose=verbose,
            tracker_backend=tracker_backend,
            tracking_backend=tracking_backend,
            seed=seed,
        )
        args.compare_to_first_trial = True
        tune_results = tuner_module.run_tune(
            args,
            baseline_config=self._tracker_config_from_spec(),
        )
        tune_results.workflow_rendered = True
        return tune_results

    def research(
        self,
        *,
        benchmark: str | Path,
        project: str | Path | None = None,
        verbose: bool = BOXMOT_DEFAULTS.research.verbose,
        proposal_model: str = BOXMOT_DEFAULTS.research.proposal_model,
        proposal_api_key: str | None = BOXMOT_DEFAULTS.research.proposal_api_key,
        proposal_api_key_env: str | None = BOXMOT_DEFAULTS.research.proposal_api_key_env,
        max_metric_calls: int = BOXMOT_DEFAULTS.research.max_metric_calls,
        eval_timeout: float = BOXMOT_DEFAULTS.research.eval_timeout,
        keep_workspace: bool = BOXMOT_DEFAULTS.research.keep_workspace,
        idf1_penalty: float = BOXMOT_DEFAULTS.research.idf1_penalty,
        mota_penalty: float = BOXMOT_DEFAULTS.research.mota_penalty,
        idf1_tolerance: float = BOXMOT_DEFAULTS.research.idf1_tolerance,
        mota_tolerance: float = BOXMOT_DEFAULTS.research.mota_tolerance,
        tracker_backend: str | None = None,
        tracking_backend: str = "thread",
    ) -> research_module.ResearchResult:
        args = adapters.build_research_args(
            self,
            benchmark,
            project=project,
            verbose=verbose,
            proposal_model=proposal_model,
            proposal_api_key=proposal_api_key,
            proposal_api_key_env=proposal_api_key_env,
            max_metric_calls=max_metric_calls,
            eval_timeout=eval_timeout,
            keep_workspace=keep_workspace,
            idf1_penalty=idf1_penalty,
            mota_penalty=mota_penalty,
            idf1_tolerance=idf1_tolerance,
            mota_tolerance=mota_tolerance,
            tracker_backend=tracker_backend,
            tracking_backend=tracking_backend,
        )
        return research_module.run_research(args)

    def export(
        self,
        *,
        include: Sequence[str] = BOXMOT_DEFAULTS.export.include,
        device: str = BOXMOT_DEFAULTS.export.device,
        half: bool = BOXMOT_DEFAULTS.export.half,
        optimize: bool = BOXMOT_DEFAULTS.export.optimize,
        dynamic: bool = BOXMOT_DEFAULTS.export.dynamic,
        simplify: bool = BOXMOT_DEFAULTS.export.simplify,
        opset: int = BOXMOT_DEFAULTS.export.opset,
        workspace: int = BOXMOT_DEFAULTS.export.workspace,
        verbose: bool = False,
        batch_size: int = BOXMOT_DEFAULTS.export.batch_size,
        imgsz=None,
    ) -> ExportResult:
        args = adapters.build_export_args(
            self,
            include=include,
            device=device,
            half=half,
            optimize=optimize,
            dynamic=dynamic,
            simplify=simplify,
            opset=opset,
            workspace=workspace,
            verbose=verbose,
            batch_size=batch_size,
            imgsz=imgsz,
        )
        return export_module.run_export(args)

    def train(
        self,
        *,
        model: str = BOXMOT_DEFAULTS.train.model,
        dataset: str = BOXMOT_DEFAULTS.train.dataset,
        data_dir: str | Path | None = BOXMOT_DEFAULTS.train.data_dir,
        loss: str = BOXMOT_DEFAULTS.train.loss,
        preprocess: str = BOXMOT_DEFAULTS.train.preprocess,
        imgsz=None,
        batch_size: int = BOXMOT_DEFAULTS.train.batch_size,
        lr: float = BOXMOT_DEFAULTS.train.lr,
        weight_decay: float = BOXMOT_DEFAULTS.train.weight_decay,
        epochs: int = BOXMOT_DEFAULTS.train.epochs,
        warmup_epochs: int = BOXMOT_DEFAULTS.train.warmup_epochs,
        eval_interval: int = BOXMOT_DEFAULTS.train.eval_interval,
        p_ids: int = BOXMOT_DEFAULTS.train.p_ids,
        k_instances: int = BOXMOT_DEFAULTS.train.k_instances,
        margin: float = BOXMOT_DEFAULTS.train.margin,
        label_smooth: float = BOXMOT_DEFAULTS.train.label_smooth,
        center_loss_weight: float = BOXMOT_DEFAULTS.train.center_loss_weight,
        pretrained: bool = BOXMOT_DEFAULTS.train.pretrained,
        device: str = BOXMOT_DEFAULTS.train.device,
        project: str | Path | None = None,
        name: str = BOXMOT_DEFAULTS.train.name,
        num_workers: int = BOXMOT_DEFAULTS.train.num_workers,
        seed: int = BOXMOT_DEFAULTS.train.seed,
        eval_datasets: Sequence[str] = BOXMOT_DEFAULTS.train.eval_datasets,
        ema_decay: float | None = BOXMOT_DEFAULTS.train.ema_decay,
        gaussian_blur: bool = BOXMOT_DEFAULTS.train.gaussian_blur,
        random_grayscale: float = BOXMOT_DEFAULTS.train.random_grayscale,
        color_jitter: bool = BOXMOT_DEFAULTS.train.color_jitter,
        resume: str | Path | None = None,
    ):
        train_project = project if project is not None else BOXMOT_DEFAULTS.train.project
        args = build_mode_namespace(
            "train",
            {
                "model": model,
                "dataset": dataset,
                "data_dir": None if data_dir is None else str(data_dir),
                "loss": loss,
                "preprocess": preprocess,
                "imgsz": imgsz,
                "batch_size": int(batch_size),
                "lr": float(lr),
                "weight_decay": float(weight_decay),
                "epochs": int(epochs),
                "warmup_epochs": int(warmup_epochs),
                "eval_interval": int(eval_interval),
                "p_ids": int(p_ids),
                "k_instances": int(k_instances),
                "margin": float(margin),
                "label_smooth": float(label_smooth),
                "center_loss_weight": float(center_loss_weight),
                "pretrained": bool(pretrained),
                "device": device,
                "project": Path(train_project),
                "name": name,
                "num_workers": int(num_workers),
                "seed": int(seed),
                "eval_datasets": list(eval_datasets),
                "ema_decay": ema_decay,
                "gaussian_blur": bool(gaussian_blur),
                "random_grayscale": float(random_grayscale),
                "color_jitter": bool(color_jitter),
                "resume": None if resume is None else str(resume),
            },
        )
        return reid_trainer_module.main(args)

    def eval_reid(
        self,
        *,
        weights: str | Path,
        dataset: str,
        data_dir: str | Path,
        model: str | None = None,
        device: str = "cpu",
        batch_size: int = 64,
        num_workers: int = 4,
        output: str | Path | None = None,
    ) -> dict[str, Any]:
        args = SimpleNamespace(
            weights=str(weights),
            model=model,
            dataset=dataset,
            data_dir=str(data_dir),
            device=device,
            batch_size=int(batch_size),
            num_workers=int(num_workers),
            output=None if output is None else str(output),
        )
        return reid_evaluator_module.main(args)

Composable entry points

Create a lazy streaming tracking result iterator.

Source code in boxmot/api/_facade.py

def track(source, detector, reid=None, tracker=None, *, verbose: bool = True, drawer=None) -> Results:
    """Create a lazy streaming tracking result iterator."""
    if tracker is None:
        raise ValueError("A tracker instance is required.")
    return Results(source, detector, reid, tracker, verbose=verbose, drawer=drawer)

Aggregate run metrics over one or more tracking results.

Source code in boxmot/api/_facade.py

def evaluate(
    data,
    detector=None,
    reid=None,
    tracker=None,
    *,
    metrics: bool = True,
    speed: bool = True,
    verbose: bool = False,
) -> dict[str, Any]:
    """Aggregate run metrics over one or more tracking results."""
    runs = _coerce_results(
        data,
        detector=detector,
        reid=reid,
        tracker=tracker,
        verbose=verbose,
        track_fn=track,
    )
    summaries = [run.summary() for run in runs]

    total_frames = sum(summary["frames"] for summary in summaries)
    total_detections = sum(summary["detections"] for summary in summaries)
    total_tracks = sum(summary["tracks"] for summary in summaries)
    total_det_ms = sum(summary["timings_ms"]["det"] for summary in summaries)
    total_reid_ms = sum(summary["timings_ms"]["reid"] for summary in summaries)
    total_track_ms = sum(summary["timings_ms"]["track"] for summary in summaries)
    total_ms = sum(summary["timings_ms"]["total"] for summary in summaries)

    response: dict[str, Any] = {
        "sources": len(summaries),
        "runs": summaries,
    }

    if metrics:
        response["metrics"] = {
            "frames": total_frames,
            "detections": total_detections,
            "tracks": total_tracks,
            "avg_tracks_per_frame": (total_tracks / total_frames) if total_frames else 0.0,
        }

    if speed:
        response["speed"] = {
            "det_ms": total_det_ms,
            "reid_ms": total_reid_ms,
            "track_ms": total_track_ms,
            "total_ms": total_ms,
            "avg_total_ms": (total_ms / total_frames) if total_frames else 0.0,
            "fps": (1000.0 * total_frames / total_ms) if total_ms else 0.0,
        }

    return response

Result objects

Source code in boxmot/engine/workflows/results.py

@dataclass(slots=True)
class TrackRunResult:
    source: Any
    results: Results
    video_path: Path | None
    text_path: Path | None
    _timings: dict[str, Any] = field(default_factory=dict, repr=False)
    _summary: dict[str, Any] = field(default_factory=dict, repr=False)

    @property
    def timings(self) -> dict[str, Any]:
        self.refresh()
        return self._timings

    @property
    def summary(self) -> dict[str, Any]:
        self.refresh()
        return self._summary

    def __str__(self) -> str:
        return self.render()

    def __repr__(self) -> str:
        self.refresh()
        return (
            f"TrackRunResult(source={self.source!r}, summary={self._summary!r}, "
            f"video_path={self.video_path!r}, text_path={self.text_path!r})"
        )

    def __iter__(self) -> Iterator[FrameResult]:
        for track_result in self.results:
            self.refresh()
            yield track_result
        self.refresh()

    def show(self) -> None:
        self.results.show()
        self.refresh()

    def stop(self, reason: str | None = None) -> None:
        self.results.stop(reason)
        self.refresh()

    def format_summary(self) -> str:
        self.refresh()
        return self.results.format_summary()

    def render(self) -> str:
        return self.format_summary()

    def renderable(self) -> RenderableType:
        self.refresh()
        return _build_tracking_summary_renderable(self._summary)

    def print_summary(self) -> None:
        print_text(self.render())

    def refresh(self) -> None:
        summary_fn = getattr(self.results, "summary", None)
        if callable(summary_fn):
            self._summary = summary_fn()
        else:
            self._summary = _results_summary_snapshot(self.results, self.source)
        self._timings = _track_timings_from_summary(self._summary)

Source code in boxmot/engine/workflows/results.py

@dataclass(slots=True)
class GenerateResult:
    benchmark: str | None
    source: Any
    cache_dir: Path
    detectors: tuple[Path, ...]
    reid_models: tuple[Path, ...]
    timings: dict[str, Any] = field(default_factory=dict)
    args: Any = None

    def __str__(self) -> str:
        return self.render()

    def __repr__(self) -> str:
        return (
            f"GenerateResult(benchmark={self.benchmark!r}, source={self.source!r}, "
            f"cache_dir={self.cache_dir!r}, detectors={self.detectors!r}, reid_models={self.reid_models!r})"
        )

    def render(self) -> str:
        timing_stats = reporting.timing_stats_from_snapshot(self.timings)
        if timing_stats is not None:
            summary = timing_stats.format_summary()
            if summary:
                return summary

        target = self.benchmark or self.source
        lines = ["GENERATE SUMMARY"]
        if target is not None:
            label = "Benchmark" if self.benchmark else "Source"
            lines.append(f"{label}: {target}")
        lines.append(f"Cache dir: {self.cache_dir}")
        return "\n".join(lines)

    def print_summary(self) -> None:
        print_text(self.render())

    def to_dict(self) -> dict[str, Any]:
        return {
            "benchmark": self.benchmark,
            "source": None if self.source is None else str(self.source),
            "cache_dir": str(self.cache_dir),
            "detectors": [str(path) for path in self.detectors],
            "reid_models": [str(path) for path in self.reid_models],
            "timings": dict(self.timings),
        }

Source code in boxmot/engine/workflows/results.py

@dataclass(slots=True)
class ValidationResult:
    benchmark: str
    raw: dict[str, Any]
    summary_label: str
    summary: dict[str, Any]
    exp_dir: Path | None = None
    timings: dict[str, Any] = field(default_factory=dict)
    args: Any = None
    workflow_rendered: bool = False

    def __str__(self) -> str:
        if self.workflow_rendered:
            return ""
        return self.render()

    def __repr__(self) -> str:
        return (
            f"ValidationResult(benchmark={self.benchmark!r}, "
            f"summary={self.summary!r}, exp_dir={self.exp_dir!r})"
        )

    def render(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
    ) -> str:
        return reporting.render_validation_cli_report(
            self.raw,
            args=self.args,
            timings=self.timings,
            title=reporting.CLI_RESULTS_SUMMARY_TITLE if title is None else title,
            include_sequences=include_sequences,
            include_timings=include_timings,
        )

    def renderable(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
        compare_raw: dict[str, Any] | None = None,
        compare_args: Any = None,
    ) -> RenderableType:
        return reporting.build_validation_cli_renderable(
            self.raw,
            args=self.args,
            timings=self.timings,
            title=title,
            include_sequences=include_sequences,
            include_timings=include_timings,
            compare_raw=compare_raw,
            compare_args=compare_args,
        )

    def format_report(self, *, title: str | None = None, include_sequences: bool = True) -> str:
        report_title = reporting.DEFAULT_VALIDATION_REPORT_TITLE if title is None else title
        return reporting.format_validation_report(
            self.raw,
            args=self.args,
            title=report_title,
            include_sequences=include_sequences,
        )

    def print_report(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
    ) -> None:
        reporting.print_validation_cli_report(
            self.raw,
            args=self.args,
            timings=self.timings,
            title=reporting.CLI_RESULTS_SUMMARY_TITLE if title is None else title,
            include_sequences=include_sequences,
            include_timings=include_timings,
        )

    def to_dict(self, *, include_raw: bool = False) -> dict[str, Any]:
        payload = {
            "benchmark": self.benchmark,
            "summary_label": self.summary_label,
            "summary": dict(self.summary),
            "timings": dict(self.timings),
            "exp_dir": None if self.exp_dir is None else str(self.exp_dir),
        }
        if include_raw:
            payload["raw"] = self.raw
        return payload

Source code in boxmot/engine/workflows/results.py

@dataclass(slots=True)
class TuneResult:
    benchmark: str
    tracker: str
    trials: list[TuneTrialResult]
    best: TuneTrialResult
    best_config: dict[str, Any]
    best_yaml: Path
    workflow_rendered: bool = False

    @property
    def summary_label(self) -> str:
        return self.best.summary_label

    @property
    def summary(self) -> dict[str, Any]:
        return self.best.summary

    @property
    def raw(self) -> dict[str, Any]:
        return self.best.raw

    @property
    def timings(self) -> dict[str, Any]:
        return self.best.timings

    @property
    def exp_dir(self) -> Path | None:
        return self.best.exp_dir

    @property
    def args(self) -> Any:
        return self.best.args

    @property
    def baseline(self) -> TuneTrialResult:
        return self.trials[0]

    def __str__(self) -> str:
        if self.workflow_rendered:
            return ""
        return self.render()

    def __repr__(self) -> str:
        return (
            f"TuneResult(benchmark={self.benchmark!r}, tracker={self.tracker!r}, "
            f"summary={self.summary!r}, best_config={self.best_config!r}, best_yaml={self.best_yaml!r})"
        )

    def render(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
    ) -> str:
        return reporting.render_validation_cli_report(
            self.best.raw,
            args=self.best.args,
            timings=self.best.timings,
            title=reporting.CLI_TUNE_BEST_SUMMARY_TITLE if title is None else title,
            include_sequences=include_sequences,
            include_timings=include_timings,
            compare_raw=self.baseline.raw,
            compare_args=self.baseline.args,
        )

    def format_report(self, *, title: str | None = None, include_sequences: bool = True) -> str:
        return self.format_best_report(title=title, include_sequences=include_sequences)

    def print_report(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
    ) -> None:
        print_text(
            self.render(
                title=title,
                include_sequences=include_sequences,
                include_timings=include_timings,
            )
        )

    def format_best_report(self, *, title: str | None = None, include_sequences: bool = True) -> str:
        report_title = reporting.DEFAULT_TUNE_BEST_REPORT_TITLE if title is None else title
        return reporting.format_validation_report(
            self.best.raw,
            args=self.best.args,
            title=report_title,
            include_sequences=include_sequences,
        )

    def print_best_report(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
    ) -> None:
        print_text(
            self.render(
                title=title,
                include_sequences=include_sequences,
                include_timings=include_timings,
            )
        )

    def to_dict(self, *, include_trials: bool = False, include_raw: bool = False) -> dict[str, Any]:
        payload = {
            "benchmark": self.benchmark,
            "tracker": self.tracker,
            "summary_label": self.summary_label,
            "summary": dict(self.summary),
            "best_config": dict(self.best_config),
            "best_yaml": str(self.best_yaml),
            "best": self.best.to_dict(include_raw=include_raw),
        }
        if include_trials:
            payload["trials"] = [trial.to_dict(include_raw=include_raw) for trial in self.trials]
        return payload

Source code in boxmot/engine/workflows/results.py

@dataclass(slots=True)
class TuneTrialResult:
    index: int
    config: dict[str, Any]
    metrics: ValidationResult
    score: tuple[float, ...]

    @property
    def benchmark(self) -> str:
        return self.metrics.benchmark

    @property
    def raw(self) -> dict[str, Any]:
        return self.metrics.raw

    @property
    def summary_label(self) -> str:
        return self.metrics.summary_label

    @property
    def summary(self) -> dict[str, Any]:
        return self.metrics.summary

    @property
    def timings(self) -> dict[str, Any]:
        return self.metrics.timings

    @property
    def exp_dir(self) -> Path | None:
        return self.metrics.exp_dir

    @property
    def args(self) -> Any:
        return self.metrics.args

    def __str__(self) -> str:
        return self.render()

    def __repr__(self) -> str:
        return (
            f"TuneTrialResult(index={self.index}, summary={self.summary!r}, "
            f"config={self.config!r}, exp_dir={self.exp_dir!r})"
        )

    def render(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
    ) -> str:
        return self.metrics.render(
            title=title,
            include_sequences=include_sequences,
            include_timings=include_timings,
        )

    def format_report(self, *, title: str | None = None, include_sequences: bool = True) -> str:
        return self.metrics.format_report(title=title, include_sequences=include_sequences)

    def print_report(
        self,
        *,
        title: str | None = None,
        include_sequences: bool = True,
        include_timings: bool = False,
    ) -> None:
        self.metrics.print_report(
            title=title,
            include_sequences=include_sequences,
            include_timings=include_timings,
        )

    def to_dict(self, *, include_raw: bool = False) -> dict[str, Any]:
        return {
            "index": self.index,
            "config": dict(self.config),
            "score": list(self.score),
            "metrics": self.metrics.to_dict(include_raw=include_raw),
        }

Source code in boxmot/engine/research.py

@dataclass(slots=True)
class ResearchResult:
    tracker: str
    benchmark: str
    proposal_model: str
    run_dir: Path
    best_candidate_dir: Path
    editable_files: tuple[str, ...]
    train_sequences: tuple[str, ...]
    val_sequences: tuple[str, ...]
    baseline_summary: dict[str, int | float]
    best_summary: dict[str, int | float]
    delta_summary: dict[str, float]
    workspace_dir: Path | None = None

    def __str__(self) -> str:
        return self.render()

    def render(self) -> str:
        def _format_metrics(metrics: Mapping[str, int | float], *, signed: bool = False) -> str:
            parts = []
            for metric in RESEARCH_METRICS:
                value = metrics.get(metric)
                if isinstance(value, (int, float)):
                    fmt = f"{float(value):+.3f}" if signed else f"{float(value):.3f}"
                    parts.append(f"{metric}={fmt}")
            return " ".join(parts) if parts else "n/a"

        lines = [
            "RESEARCH SUMMARY",
            f"Tracker: {self.tracker}",
            f"Benchmark: {self.benchmark}",
            f"Proposal model: {self.proposal_model}",
            f"Baseline: {_format_metrics(self.baseline_summary)}",
            f"Best: {_format_metrics(self.best_summary)}",
            f"Delta: {_format_metrics(self.delta_summary, signed=True)}",
            f"Best candidate dir: {self.best_candidate_dir}",
        ]
        if self.workspace_dir is not None:
            lines.append(f"Workspace: {self.workspace_dir}")
        return "\n".join(lines)

    def print_summary(self) -> None:
        print_text(self.render())

    def to_dict(self) -> dict[str, Any]:
        return {
            "tracker": self.tracker,
            "benchmark": self.benchmark,
            "proposal_model": self.proposal_model,
            "run_dir": str(self.run_dir),
            "best_candidate_dir": str(self.best_candidate_dir),
            "editable_files": list(self.editable_files),
            "train_sequences": list(self.train_sequences),
            "val_sequences": list(self.val_sequences),
            "baseline_summary": self.baseline_summary,
            "best_summary": self.best_summary,
            "delta_summary": self.delta_summary,
            "workspace_dir": None if self.workspace_dir is None else str(self.workspace_dir),
        }

Source code in boxmot/engine/workflows/results.py

@dataclass(slots=True)
class ExportResult:
    weights: Path
    files: dict[str, Any]
    parity: dict[str, dict[str, Any]] = field(default_factory=dict)

    @property
    def parity_ok(self) -> bool:
        """True iff every checked format is within tolerance.

        Formats whose parity was skipped (e.g. ``engine``) don't count
        against the result.
        """
        if not self.parity:
            return True
        return all(stats.get("ok", False) for stats in self.parity.values())

parity_ok property

True iff every checked format is within tolerance.

Formats whose parity was skipped (e.g. engine) don't count against the result.

Source code in boxmot/engine/tracking/results.py

class Results:
    def __init__(
        self,
        source,
        detector: Any,
        reid: Any,
        tracker: Any,
        verbose: bool = True,
        drawer: Drawer | None = None,
        progress_callback: Callable[[str], None] | None = None,
    ) -> None:
        if detector is None:
            raise ValueError("A detector instance is required.")
        if tracker is None:
            raise ValueError("A tracker instance is required.")

        self.source = source
        self.detector = detector
        self.reid = reid
        self.tracker = tracker
        self.verbose = bool(verbose)
        self.drawer = drawer
        self._progress_callback = progress_callback
        self._generator: Iterator[FrameResult] | None = None
        self._exhausted = False
        self._interrupted = False
        self._track_ids_seen: set[int] = set()
        self.totals = {
            "det": 0.0,
            "detector_preprocess": 0.0,
            "detector_process": 0.0,
            "detector_postprocess": 0.0,
            "reid": 0.0,
            "reid_preprocess": 0.0,
            "reid_process": 0.0,
            "reid_postprocess": 0.0,
            "track": 0.0,
            "total": 0.0,
            "frames": 0,
            "detections": 0,
            "tracks": 0,
        }

    def __iter__(self):
        if self._exhausted:
            return iter([])
        if self._generator is None:
            self._generator = self._process()
        return self

    def __next__(self) -> FrameResult:
        if self._generator is None:
            self._generator = self._process()
        try:
            return next(self._generator)
        except StopIteration:
            self._exhausted = True
            raise

    @staticmethod
    def _as_2d_array(values: Any, empty_cols: int = 0) -> np.ndarray:
        arr = np.asarray(values, dtype=np.float32)
        if arr.size == 0:
            cols = arr.shape[1] if arr.ndim == 2 else empty_cols
            return np.empty((0, cols), dtype=np.float32)
        if arr.ndim == 1:
            return arr.reshape(1, -1)
        return arr

    @staticmethod
    def _extract_detections(output: Any) -> np.ndarray:
        if isinstance(output, (list, tuple)) and len(output) == 1:
            output = output[0]
        if isinstance(output, Detections):
            cols = output.dets.shape[1] if output.dets.ndim == 2 else (7 if output.is_obb else 6)
            return Results._as_2d_array(output.dets, empty_cols=cols)
        if hasattr(output, "dets"):
            dets = getattr(output, "dets")
            cols = dets.shape[1] if isinstance(dets, np.ndarray) and dets.ndim == 2 else 6
            return Results._as_2d_array(dets, empty_cols=cols)
        if output is None:
            return np.empty((0, 6), dtype=np.float32)
        return Results._as_2d_array(output, empty_cols=6)

    @staticmethod
    def _extract_masks(output: Any) -> np.ndarray | None:
        if isinstance(output, (list, tuple)) and len(output) == 1:
            output = output[0]
        if isinstance(output, Detections) and output.masks is not None:
            return output.masks
        if hasattr(output, "masks"):
            return getattr(output, "masks")
        return None

    def _iter_frames(self):
        source = self.source
        if isinstance(source, (str, Path)):
            source_path = Path(source)
            if source_path.is_dir() and (source_path / "img1").is_dir():
                source = source_path / "img1"
        yield from iter_source(source)

    def _log_frame_timings(self, frame_idx: int, det_ms: float, reid_ms: float, track_ms: float) -> None:
        total_ms = det_ms + reid_ms + track_ms
        if self.reid is None and reid_ms <= 0.0:
            message = f"Frame {frame_idx} | Det: {det_ms:.1f}ms | Track: {track_ms:.1f}ms | Total: {total_ms:.1f}ms"
            if self._progress_callback is not None:
                self._progress_callback(message)
            else:
                LOGGER.info(message)
            return
        message = (
            f"Frame {frame_idx} | Det: {det_ms:.1f}ms | ReID: {reid_ms:.1f}ms | "
            f"Track: {track_ms:.1f}ms | Total: {total_ms:.1f}ms"
        )
        if self._progress_callback is not None:
            self._progress_callback(message)
        else:
            LOGGER.info(message)

    def _tracker_reid_time_ms(self) -> float:
        getter = getattr(self.tracker, "get_last_reid_time_ms", None)
        value = getter() if callable(getter) else getattr(self.tracker, "last_reid_time_ms", 0.0)
        try:
            return max(float(value), 0.0)
        except (TypeError, ValueError):
            return 0.0

    def _tracker_reid_phase_breakdown(self) -> dict[str, float]:
        breakdown: dict[str, float] = {"preprocess": 0.0, "process": 0.0, "postprocess": 0.0}
        for phase in breakdown:
            getter = getattr(self.tracker, f"get_last_reid_{phase}_time_ms", None)
            value = getter() if callable(getter) else getattr(self.tracker, f"last_reid_{phase}_time_ms", 0.0)
            try:
                breakdown[phase] = max(float(value), 0.0)
            except (TypeError, ValueError):
                breakdown[phase] = 0.0
        return breakdown

    def _log_summary(self) -> None:
        self.print_summary()

    def _run_reid(self, frame: np.ndarray, dets: np.ndarray) -> np.ndarray | None:
        if self.reid is None:
            return None
        try:
            return self.reid(frame, boxes=dets)
        except TypeError:
            return self.reid(frame, dets)

    def _add_detector_phase_time(self, phase: str, time_ms: float) -> None:
        phase_key = f"detector_{phase}"
        elapsed_ms = float(time_ms or 0.0)
        self.totals[phase_key] += elapsed_ms
        self.totals["det"] += elapsed_ms

    def _add_reid_phase_time(self, phase: str, time_ms: float) -> None:
        phase_key = f"reid_{phase}"
        elapsed_ms = float(time_ms or 0.0)
        self.totals[phase_key] += elapsed_ms
        self.totals["reid"] += elapsed_ms

    def _run_detector_timed(self, frame: np.ndarray) -> tuple[np.ndarray, np.ndarray | None, float]:
        if all(hasattr(self.detector, attr) for attr in ("preprocess", "process", "postprocess")):
            try:
                preprocess_started = time.perf_counter()
                preprocessed = self.detector.preprocess(frame)
                preprocess_ms = (time.perf_counter() - preprocess_started) * 1000
                self._add_detector_phase_time("preprocess", preprocess_ms)

                process_started = time.perf_counter()
                raw_output = self.detector.process(preprocessed)
                process_ms = (time.perf_counter() - process_started) * 1000
                self._add_detector_phase_time("process", process_ms)

                postprocess_started = time.perf_counter()
                detector_output = self.detector.postprocess(raw_output)
                postprocess_ms = (time.perf_counter() - postprocess_started) * 1000
                self._add_detector_phase_time("postprocess", postprocess_ms)

                dets = self._extract_detections(detector_output)
                masks = self._extract_masks(detector_output)
                return dets, masks, preprocess_ms + process_ms + postprocess_ms
            except NotImplementedError:
                pass

        det_started = time.perf_counter()
        detector_output = self.detector(frame)
        det_ms = (time.perf_counter() - det_started) * 1000
        self._add_detector_phase_time("process", det_ms)
        dets = self._extract_detections(detector_output)
        masks = self._extract_masks(detector_output)
        return dets, masks, det_ms

    def _run_reid_timed(self, frame: np.ndarray, dets: np.ndarray) -> tuple[np.ndarray | None, float]:
        if self.reid is None:
            return None, 0.0

        if all(hasattr(self.reid, attr) for attr in ("preprocess", "process", "postprocess")):
            try:
                preprocess_started = time.perf_counter()
                try:
                    payload = self.reid.preprocess(frame, boxes=dets)
                except TypeError:
                    payload = self.reid.preprocess(frame, dets)
                preprocess_ms = (time.perf_counter() - preprocess_started) * 1000
                self._add_reid_phase_time("preprocess", preprocess_ms)

                process_started = time.perf_counter()
                try:
                    features = self.reid.process(payload, boxes=dets)
                except TypeError:
                    features = self.reid.process(payload, dets)
                process_ms = (time.perf_counter() - process_started) * 1000
                self._add_reid_phase_time("process", process_ms)

                postprocess_started = time.perf_counter()
                try:
                    features = self.reid.postprocess(features, boxes=dets)
                except TypeError:
                    features = self.reid.postprocess(features, dets)
                postprocess_ms = (time.perf_counter() - postprocess_started) * 1000
                self._add_reid_phase_time("postprocess", postprocess_ms)
                return features, preprocess_ms + process_ms + postprocess_ms
            except NotImplementedError:
                pass

        reid_started = time.perf_counter()
        features = self._run_reid(frame, dets)
        reid_ms = (time.perf_counter() - reid_started) * 1000
        self._add_reid_phase_time("process", reid_ms)
        return features, reid_ms

    def _run_tracker(self, dets: np.ndarray, frame: np.ndarray, features: np.ndarray | None, masks: np.ndarray | None = None) -> TrackResults:
        kwargs: dict[str, Any] = {}
        if features is not None:
            kwargs["embs"] = features
        if masks is not None:
            kwargs["masks"] = masks
        if kwargs:
            try:
                result = self.tracker.update(dets, frame, **kwargs)
            except TypeError:
                # Tracker doesn't accept these kwargs; fall back
                if features is not None:
                    try:
                        result = self.tracker.update(dets, frame, features)
                    except TypeError:
                        result = self.tracker.update(dets, frame)
                else:
                    result = self.tracker.update(dets, frame)
        else:
            result = self.tracker.update(dets, frame)
        if isinstance(result, TrackResults):
            return result
        return TrackResults(result)

    @staticmethod
    def _extract_track_ids(tracks: TrackResults) -> set[int]:
        if tracks.size == 0 or tracks.ndim != 2:
            return set()
        return {int(tid) for tid in tracks.id.tolist()}

    def _summary_snapshot(self) -> dict[str, Any]:
        frames = int(self.totals["frames"])
        avg_total = (self.totals["total"] / frames) if frames else 0.0
        return {
            "source": str(self.source),
            "frames": frames,
            "detections": int(self.totals["detections"]),
            "tracks": int(self.totals["tracks"]),
            "unique_tracks": len(self._track_ids_seen),
            "timings_ms": {
                "det": float(self.totals["det"]),
                "detector_preprocess": float(self.totals["detector_preprocess"]),
                "detector_process": float(self.totals["detector_process"]),
                "detector_postprocess": float(self.totals["detector_postprocess"]),
                "reid": float(self.totals["reid"]),
                "reid_preprocess": float(self.totals["reid_preprocess"]),
                "reid_process": float(self.totals["reid_process"]),
                "reid_postprocess": float(self.totals["reid_postprocess"]),
                "track": float(self.totals["track"]),
                "total": float(self.totals["total"]),
                "avg_total": float(avg_total),
            },
        }

    def stop(self, reason: str | None = None) -> None:
        if self._exhausted:
            return

        self._interrupted = True
        if reason:
            if self._progress_callback is not None:
                self._progress_callback(reason)
            else:
                LOGGER.info(reason)

        generator = self._generator
        self._generator = None
        if generator is not None:
            generator.close()
        else:
            self._exhausted = True

    def format_summary(self) -> str:
        summary = self.summary()
        timings = summary["timings_ms"]
        frames = max(int(summary["frames"]), 1)
        width = 86
        breakdown = derive_timing_breakdown(timings, frames, total_time_ms=timings["total"])

        lines = [
            "=" * width,
            f"{'TRACKING SUMMARY':^{width}}",
            "=" * width,
            f"Source:      {summary['source']}",
            f"Frames:      {summary['frames']}",
            f"Detections:  {summary['detections']}",
            f"Track rows:  {summary['tracks']}",
            f"Unique IDs:  {summary.get('unique_tracks', 0)}",
            "-" * width,
            f"{'Stage':<20} {'Total (ms)':>12} {'Avg (ms)':>12} {'FPS':>10}",
            "-" * width,
        ]
        for entry in build_timing_display_rows(
            breakdown,
            frames,
            metadata=dict(getattr(self, "timing_metadata", {})),
            overall_avg_ms=float(timings["avg_total"]),
        ):
            if entry["kind"] == "group":
                lines.append(str(entry["label"]))
                continue
            if entry["kind"] == "note":
                lines.append(str(entry["label"]))
                continue
            lines.append(
                f"{str(entry['label']):<20} {float(entry['total']):>12.1f} {float(entry['avg']):>12.2f} {float(entry['fps']):>10.1f}"
            )
        lines.append("=" * width)
        return "\n".join(lines)

    def print_summary(self) -> None:
        frames = int(self.totals["frames"])
        if frames == 0:
            return
        print_text(self.format_summary())

    def _process(self):
        if hasattr(self.tracker, "reset"):
            self.tracker.reset()

        try:
            for frame_idx, (path, frame) in enumerate(self._iter_frames(), start=1):
                dets, masks, det_ms = self._run_detector_timed(frame)
                features, reid_ms = self._run_reid_timed(frame, dets)

                track_started = time.perf_counter()
                tracks = self._run_tracker(dets, frame, features, masks)
                track_ms = (time.perf_counter() - track_started) * 1000
                if self.reid is None:
                    tracker_reid_ms = min(self._tracker_reid_time_ms(), track_ms)
                    reid_ms += tracker_reid_ms
                    phase_breakdown = self._tracker_reid_phase_breakdown()
                    if any(phase_breakdown.values()) and tracker_reid_ms > 0.0:
                        breakdown_total = sum(phase_breakdown.values())
                        scale = tracker_reid_ms / breakdown_total if breakdown_total > 0.0 else 1.0
                        for phase, value in phase_breakdown.items():
                            scaled = max(float(value) * scale, 0.0)
                            self._add_reid_phase_time(phase, scaled)
                    else:
                        self._add_reid_phase_time("process", tracker_reid_ms)
                    track_ms = max(track_ms - tracker_reid_ms, 0.0)

                total_ms = det_ms + reid_ms + track_ms
                self.totals["track"] += track_ms
                self.totals["total"] += total_ms
                self.totals["frames"] += 1
                self.totals["detections"] += int(dets.shape[0])
                self.totals["tracks"] += int(tracks.shape[0])
                self._track_ids_seen.update(self._extract_track_ids(tracks))

                if self.verbose or self._progress_callback is not None:
                    self._log_frame_timings(frame_idx, det_ms, reid_ms, track_ms)

                yield FrameResult(
                    frame_idx=frame_idx,
                    frame=frame,
                    tracks=tracks,
                    detections=dets,
                    source_path=path,
                    get_drawer=lambda: self.drawer,
                    stop_session=self.stop,
                    embeddings=features,
                    masks=masks,
                )
        except KeyboardInterrupt:
            self._interrupted = True
            if self._progress_callback is not None:
                self._progress_callback("Tracking interrupted by user.")
            else:
                LOGGER.info("Tracking interrupted by user.")
            return
        finally:
            self._exhausted = True
            if self.verbose:
                self._log_summary()

    def save(self, output_path: str | Path) -> Path:
        path = Path(output_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        if path.exists():
            path.unlink()
        for frame_result in self:
            write_mot_results(path, frame_result.to_mot())
        return path

    def save_vid(self, output_path: str | Path, fps: float = 30.0) -> Path:
        """Save annotated tracking video to disk (streaming, not buffered).

        Args:
            output_path: Output .mp4 path.
            fps: Frames per second for the output video.

        Returns:
            Path: The written video path.
        """
        path = Path(output_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        writer: cv2.VideoWriter | None = None
        try:
            for frame_result in self:
                rendered = frame_result.plot()
                if writer is None:
                    height, width = rendered.shape[:2]
                    writer = cv2.VideoWriter(
                        str(path),
                        cv2.VideoWriter_fourcc(*"mp4v"),
                        fps,
                        (width, height),
                    )
                writer.write(rendered)
        finally:
            if writer is not None:
                writer.release()
        return path

    def summary(self) -> dict[str, Any]:
        return self._summary_snapshot()

    def show(self) -> None:
        for track_result in self:
            if not track_result.show():
                break
        cv2.destroyAllWindows()

save_vid(output_path, fps=30.0)

Save annotated tracking video to disk (streaming, not buffered).

Parameters:

Name	Type	Description	Default
output_path	str | Path	Output .mp4 path.	required
fps	float	Frames per second for the output video.	30.0

Returns:

Name	Type	Description
Path	Path	The written video path.

Source code in boxmot/engine/tracking/results.py

def save_vid(self, output_path: str | Path, fps: float = 30.0) -> Path:
    """Save annotated tracking video to disk (streaming, not buffered).

    Args:
        output_path: Output .mp4 path.
        fps: Frames per second for the output video.

    Returns:
        Path: The written video path.
    """
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    writer: cv2.VideoWriter | None = None
    try:
        for frame_result in self:
            rendered = frame_result.plot()
            if writer is None:
                height, width = rendered.shape[:2]
                writer = cv2.VideoWriter(
                    str(path),
                    cv2.VideoWriter_fourcc(*"mp4v"),
                    fps,
                    (width, height),
                )
            writer.write(rendered)
    finally:
        if writer is not None:
            writer.release()
    return path

Per-frame tracking result container.

Bundles the source frame with its TrackResults (returned by the tracker) and adds visualization and frame-aware export. Track data is accessed directly via self.tracks (a TrackResults instance with .id, .conf, .cls, .xyxy, .xywha, etc.).

Attributes:

Name	Type	Description
frame_idx	int	1-based frame index.
frame	ndarray	The source frame (HxWxC BGR).
tracks	TrackResults	Track output array with named accessors.
detections	ndarray | None	Raw detector output for this frame.
embeddings	ndarray | None	ReID embeddings (N, D) for detections, if available.
source_path	str	Path of the source file/stream.

Methods:

Name	Description
plot	Render tracks on the frame and return the annotated image.
show	Display the annotated frame in a window.
save	Save the annotated frame to disk.
save_txt	Append tracks in MOT format to a text file.
save_csv	Append tracks in CSV format to a file.
summary	Return tracks as a list of dicts.
to_json	Return tracks as a JSON string.
to_csv	Return tracks as a CSV string.
verbose	Return a human-readable summary string.

Source code in boxmot/engine/tracking/results.py

class FrameResult:
    """Per-frame tracking result container.

    Bundles the source frame with its TrackResults (returned by the tracker)
    and adds visualization and frame-aware export.  Track data is accessed
    directly via ``self.tracks`` (a TrackResults instance with .id, .conf,
    .cls, .xyxy, .xywha, etc.).

    Attributes:
        frame_idx (int): 1-based frame index.
        frame (np.ndarray): The source frame (HxWxC BGR).
        tracks (TrackResults): Track output array with named accessors.
        detections (np.ndarray | None): Raw detector output for this frame.
        embeddings (np.ndarray | None): ReID embeddings (N, D) for detections, if available.
        source_path (str): Path of the source file/stream.

    Methods:
        plot: Render tracks on the frame and return the annotated image.
        show: Display the annotated frame in a window.
        save: Save the annotated frame to disk.
        save_txt: Append tracks in MOT format to a text file.
        save_csv: Append tracks in CSV format to a file.
        summary: Return tracks as a list of dicts.
        to_json: Return tracks as a JSON string.
        to_csv: Return tracks as a CSV string.
        verbose: Return a human-readable summary string.
    """

    def __init__(
        self,
        frame_idx: int,
        frame: np.ndarray,
        tracks: TrackResults | np.ndarray,
        detections: np.ndarray | None,
        source_path: str,
        get_drawer: Callable[[], Drawer | None],
        stop_session: Callable[[str | None], None] | None = None,
        embeddings: np.ndarray | None = None,
        masks: np.ndarray | None = None,
    ) -> None:
        self.frame_idx = int(frame_idx)
        self.frame = frame
        self.tracks = tracks if isinstance(tracks, TrackResults) else TrackResults(tracks)
        self.source_path = source_path
        self._get_drawer = get_drawer
        self._stop_session = stop_session

        # Reorder detections and embeddings to align with tracks via det_ind
        raw_dets = None if detections is None else self._as_2d_array(detections)
        self.detections, self.embeddings = self._align_to_tracks(raw_dets, embeddings)
        self.masks = self._align_masks(masks)

    @staticmethod
    def _as_2d_array(values: Any) -> np.ndarray:
        arr = np.asarray(values, dtype=np.float32)
        if arr.size == 0:
            cols = arr.shape[1] if arr.ndim == 2 else 0
            return np.empty((0, cols), dtype=np.float32)
        if arr.ndim == 1:
            return arr.reshape(1, -1)
        return arr

    def _align_to_tracks(
        self, dets: np.ndarray | None, embs: np.ndarray | None,
    ) -> tuple[np.ndarray | None, np.ndarray | None]:
        """Reorder detections and embeddings so they align 1-to-1 with tracks.

        Coasting tracks (det_ind == -1) get zero-filled rows.
        """
        if self.tracks.size == 0:
            det_cols = dets.shape[1] if dets is not None and dets.ndim == 2 else 6
            empty_dets = np.empty((0, det_cols), dtype=np.float32) if dets is not None else None
            return empty_dets, None

        det_inds = self.tracks.det_ind
        valid = det_inds >= 0

        aligned_dets: np.ndarray | None = None
        if dets is not None:
            cols = dets.shape[1]
            aligned_dets = np.zeros((len(self.tracks), cols), dtype=np.float32)
            aligned_dets[valid] = dets[det_inds[valid]]

        aligned_embs: np.ndarray | None = None
        if embs is not None:
            embs_arr = np.asarray(embs, dtype=np.float32)
            dim = embs_arr.shape[1]
            aligned_embs = np.zeros((len(self.tracks), dim), dtype=np.float32)
            aligned_embs[valid] = embs_arr[det_inds[valid]]

        return aligned_dets, aligned_embs

    def _align_masks(self, masks: np.ndarray | None) -> np.ndarray | None:
        """Reorder masks to align 1-to-1 with tracks via det_ind."""
        if masks is None or self.tracks.size == 0:
            return None
        det_inds = self.tracks.det_ind
        valid = det_inds >= 0
        h, w = masks.shape[1], masks.shape[2]
        aligned = np.zeros((len(self.tracks), h, w), dtype=masks.dtype)
        aligned[valid] = masks[det_inds[valid]]
        return aligned

    # ------------------------------------------------------------------
    # Convenience
    # ------------------------------------------------------------------

    @property
    def num_tracks(self) -> int:
        """Number of active tracks in this frame."""
        return int(self.tracks.shape[0])

    def __len__(self) -> int:
        return self.num_tracks

    # ------------------------------------------------------------------
    # Visualization
    # ------------------------------------------------------------------

    def _default_draw(self, frame: np.ndarray) -> np.ndarray:
        drawn = frame.copy()
        if self.tracks.size == 0:
            return drawn

        # Draw mask overlays first (beneath boxes)
        if self.masks is not None:
            h_frame, w_frame = drawn.shape[:2]
            for i in range(len(self.tracks)):
                mask = self.masks[i]
                if mask.sum() == 0:
                    continue
                track_id = int(self.tracks.id[i])
                color = _track_color(track_id)
                # Resize mask to frame dimensions if needed
                h_mask, w_mask = mask.shape[:2]
                if (h_mask, w_mask) != (h_frame, w_frame):
                    mask = cv2.resize(mask, (w_frame, h_frame), interpolation=cv2.INTER_NEAREST)
                colored = np.zeros_like(drawn)
                colored[:] = color
                mask_bool = mask.astype(bool)
                drawn[mask_bool] = cv2.addWeighted(drawn, 0.6, colored, 0.4, 0)[mask_bool]

        for i in range(len(self.tracks)):
            track_id = int(self.tracks.id[i])
            conf = float(self.tracks.conf[i])
            if self.tracks.is_obb:
                cx, cy, width, height, angle = self.tracks.xywha[i]
                rect = ((float(cx), float(cy)), (max(float(width), 1.0), max(float(height), 1.0)), float(np.degrees(angle)))
                corners = cv2.boxPoints(rect).astype(np.int32)
                cv2.polylines(drawn, [corners], True, _track_color(track_id), 2)
                label_point = tuple(corners[0])
            else:
                x1, y1, x2, y2 = self.tracks.xyxy[i].round().astype(int)
                cv2.rectangle(drawn, (x1, y1), (x2, y2), _track_color(track_id), 2)
                label_point = (x1, max(0, y1 - 6))

            cv2.putText(
                drawn,
                f"{track_id} {conf:.2f}",
                label_point,
                cv2.FONT_HERSHEY_SIMPLEX,
                0.5,
                _track_color(track_id),
                1,
                cv2.LINE_AA,
            )

        return drawn

    def plot(self) -> np.ndarray:
        """Plot tracks on the frame and return the annotated image."""
        drawer = self._get_drawer()
        if drawer is not None:
            return drawer(self.frame.copy(), self.tracks)
        return self._default_draw(self.frame)

    def render(self) -> np.ndarray:
        """Alias for plot()."""
        return self.plot()

    def show(self, window_name: str = "Tracking") -> bool:
        """Display the annotated frame. Returns False if user pressed 'q'."""
        cv2.imshow(window_name, self.plot())
        key = cv2.waitKey(1) & 0xFF
        should_continue = key not in (ord("q"), 27)
        if not should_continue and self._stop_session is not None:
            self._stop_session("Tracking stopped by user.")
        return should_continue

    def save(self, filename: str | Path | None = None) -> str:
        """Save annotated frame to disk.

        Args:
            filename: Output path. Defaults to 'result_<frame_idx>.jpg'.

        Returns:
            str: Path where the image was saved.
        """
        if filename is None:
            filename = f"result_{self.frame_idx:06d}.jpg"
        path = Path(filename)
        path.parent.mkdir(parents=True, exist_ok=True)
        cv2.imwrite(str(path), self.plot())
        return str(path)

    # ------------------------------------------------------------------
    # Export
    # ------------------------------------------------------------------

    def to_mot(self) -> np.ndarray:
        """Convert tracks to MOT challenge format array."""
        if self.tracks.size == 0:
            return np.empty((0, 0), dtype=np.float32)
        if self.tracks.is_obb:
            return convert_to_mmot_obb_format(self.tracks, self.frame_idx)
        return convert_to_mot_format(self.tracks, self.frame_idx)

    def save_txt(self, path: str | Path) -> None:
        """Append tracks in MOT challenge format to a text file."""
        self.tracks.save_mot(path, frame_id=self.frame_idx)

    def save_csv(self, path: str | Path, header: bool = True) -> None:
        """Append tracks in CSV format to a file."""
        self.tracks.save_csv(path, frame_id=self.frame_idx, header=header)

    def save_vid(self, path: str | Path, fps: float | None = None) -> None:
        """Append annotated frame to a video file (streaming).

        Call once per frame in your loop. The video writer is created on
        the first call and reused for subsequent frames with the same path.
        Call ``FrameResult.close_vid()`` after the loop to finalize.

        Args:
            path: Output .mp4 path.
            fps: Frames per second. If None (default), auto-detected from
                 the source video/camera.
        """
        key = str(Path(path))
        rendered = self.plot()
        if key not in _video_writers:
            if fps is None:
                fps = _resolve_fps(self.source_path)
            Path(path).parent.mkdir(parents=True, exist_ok=True)
            h, w = rendered.shape[:2]
            _video_writers[key] = cv2.VideoWriter(
                key, cv2.VideoWriter_fourcc(*"mp4v"), fps, (w, h),
            )
        _video_writers[key].write(rendered)

    @staticmethod
    def close_vid(path: str | Path | None = None) -> None:
        """Release video writer(s).

        Args:
            path: Release writer for this path only. If None, release all.
        """
        if path is not None:
            key = str(Path(path))
            writer = _video_writers.pop(key, None)
            if writer is not None:
                writer.release()
        else:
            for writer in _video_writers.values():
                writer.release()
            _video_writers.clear()

    def to_csv(self) -> str:
        """Return tracks as a CSV-formatted string."""
        return self.tracks.to_csv(frame_id=self.frame_idx)

    def to_json(self, indent: int | None = None) -> str:
        """Return tracks as a JSON string."""
        return self.tracks.to_json(indent=indent)

    def summary(self) -> list[dict[str, Any]]:
        """Return tracks as a list of dictionaries."""
        return self.tracks.summary()

    def verbose(self) -> str:
        """Return a human-readable summary string for this frame."""
        n = self.num_tracks
        if n == 0:
            return f"Frame {self.frame_idx}: (no tracks)"
        ids = ", ".join(str(i) for i in self.tracks.id[:5])
        suffix = f", ... ({n} total)" if n > 5 else ""
        return f"Frame {self.frame_idx}: {n} tracks [IDs: {ids}{suffix}]"

    def __str__(self) -> str:
        rows = self.to_mot()
        if rows.size == 0:
            return ""
        if rows.ndim == 1:
            rows = rows.reshape(1, -1)

        buffer = io.StringIO()
        if rows.shape[1] == 9:
            np.savetxt(buffer, rows, fmt="%d,%d,%d,%d,%d,%d,%.6f,%d,%d")
        else:
            np.savetxt(buffer, rows, fmt="%g", delimiter=",")
        return buffer.getvalue()

    def __repr__(self) -> str:
        return f"Tracks(frame={self.frame_idx}, n={self.num_tracks}, obb={self.tracks.is_obb})"

num_tracks property

Number of active tracks in this frame.

close_vid(path=None) staticmethod

Release video writer(s).

Parameters:

Name	Type	Description	Default
path	str | Path | None	Release writer for this path only. If None, release all.	None

Source code in boxmot/engine/tracking/results.py

@staticmethod
def close_vid(path: str | Path | None = None) -> None:
    """Release video writer(s).

    Args:
        path: Release writer for this path only. If None, release all.
    """
    if path is not None:
        key = str(Path(path))
        writer = _video_writers.pop(key, None)
        if writer is not None:
            writer.release()
    else:
        for writer in _video_writers.values():
            writer.release()
        _video_writers.clear()

plot()

Plot tracks on the frame and return the annotated image.

Source code in boxmot/engine/tracking/results.py

def plot(self) -> np.ndarray:
    """Plot tracks on the frame and return the annotated image."""
    drawer = self._get_drawer()
    if drawer is not None:
        return drawer(self.frame.copy(), self.tracks)
    return self._default_draw(self.frame)

render()

Alias for plot().

Source code in boxmot/engine/tracking/results.py

def render(self) -> np.ndarray:
    """Alias for plot()."""
    return self.plot()

save(filename=None)

Save annotated frame to disk.

Parameters:

Name	Type	Description	Default
filename	str | Path | None	Output path. Defaults to 'result_.jpg'.	None

Returns:

Name	Type	Description
str	str	Path where the image was saved.

Source code in boxmot/engine/tracking/results.py

def save(self, filename: str | Path | None = None) -> str:
    """Save annotated frame to disk.

    Args:
        filename: Output path. Defaults to 'result_<frame_idx>.jpg'.

    Returns:
        str: Path where the image was saved.
    """
    if filename is None:
        filename = f"result_{self.frame_idx:06d}.jpg"
    path = Path(filename)
    path.parent.mkdir(parents=True, exist_ok=True)
    cv2.imwrite(str(path), self.plot())
    return str(path)

save_csv(path, header=True)

Append tracks in CSV format to a file.

Source code in boxmot/engine/tracking/results.py

def save_csv(self, path: str | Path, header: bool = True) -> None:
    """Append tracks in CSV format to a file."""
    self.tracks.save_csv(path, frame_id=self.frame_idx, header=header)

save_txt(path)

Append tracks in MOT challenge format to a text file.

Source code in boxmot/engine/tracking/results.py

def save_txt(self, path: str | Path) -> None:
    """Append tracks in MOT challenge format to a text file."""
    self.tracks.save_mot(path, frame_id=self.frame_idx)

save_vid(path, fps=None)

Append annotated frame to a video file (streaming).

Call once per frame in your loop. The video writer is created on the first call and reused for subsequent frames with the same path. Call FrameResult.close_vid() after the loop to finalize.

Parameters:

Name	Type	Description	Default
path	str | Path	Output .mp4 path.	required
fps	float | None	Frames per second. If None (default), auto-detected from the source video/camera.	None

Source code in boxmot/engine/tracking/results.py

def save_vid(self, path: str | Path, fps: float | None = None) -> None:
    """Append annotated frame to a video file (streaming).

    Call once per frame in your loop. The video writer is created on
    the first call and reused for subsequent frames with the same path.
    Call ``FrameResult.close_vid()`` after the loop to finalize.

    Args:
        path: Output .mp4 path.
        fps: Frames per second. If None (default), auto-detected from
             the source video/camera.
    """
    key = str(Path(path))
    rendered = self.plot()
    if key not in _video_writers:
        if fps is None:
            fps = _resolve_fps(self.source_path)
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        h, w = rendered.shape[:2]
        _video_writers[key] = cv2.VideoWriter(
            key, cv2.VideoWriter_fourcc(*"mp4v"), fps, (w, h),
        )
    _video_writers[key].write(rendered)

show(window_name='Tracking')

Display the annotated frame. Returns False if user pressed 'q'.

Source code in boxmot/engine/tracking/results.py

def show(self, window_name: str = "Tracking") -> bool:
    """Display the annotated frame. Returns False if user pressed 'q'."""
    cv2.imshow(window_name, self.plot())
    key = cv2.waitKey(1) & 0xFF
    should_continue = key not in (ord("q"), 27)
    if not should_continue and self._stop_session is not None:
        self._stop_session("Tracking stopped by user.")
    return should_continue

summary()

Return tracks as a list of dictionaries.

Source code in boxmot/engine/tracking/results.py

def summary(self) -> list[dict[str, Any]]:
    """Return tracks as a list of dictionaries."""
    return self.tracks.summary()

to_csv()

Return tracks as a CSV-formatted string.

Source code in boxmot/engine/tracking/results.py

def to_csv(self) -> str:
    """Return tracks as a CSV-formatted string."""
    return self.tracks.to_csv(frame_id=self.frame_idx)

to_json(indent=None)

Return tracks as a JSON string.

Source code in boxmot/engine/tracking/results.py

def to_json(self, indent: int | None = None) -> str:
    """Return tracks as a JSON string."""
    return self.tracks.to_json(indent=indent)

to_mot()

Convert tracks to MOT challenge format array.

Source code in boxmot/engine/tracking/results.py

def to_mot(self) -> np.ndarray:
    """Convert tracks to MOT challenge format array."""
    if self.tracks.size == 0:
        return np.empty((0, 0), dtype=np.float32)
    if self.tracks.is_obb:
        return convert_to_mmot_obb_format(self.tracks, self.frame_idx)
    return convert_to_mot_format(self.tracks, self.frame_idx)

verbose()

Return a human-readable summary string for this frame.

Source code in boxmot/engine/tracking/results.py

def verbose(self) -> str:
    """Return a human-readable summary string for this frame."""
    n = self.num_tracks
    if n == 0:
        return f"Frame {self.frame_idx}: (no tracks)"
    ids = ", ".join(str(i) for i in self.tracks.id[:5])
    suffix = f", ... ({n} total)" if n > 5 else ""
    return f"Frame {self.frame_idx}: {n} tracks [IDs: {ids}{suffix}]"