Python API Reference

Auto-generated reference for the public Python surface: the BoxMOT facade, runtime wrappers, explicit workflow helpers, and structured result objects.

BoxMOT facade

Source code in boxmot/pipeline.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,
        *,
        detector_kwargs: Mapping[str, Any] | None = None,
        reid_kwargs: Mapping[str, Any] | None = None,
        tracker_kwargs: Mapping[str, Any] | None = None,
        model: str | None = None,
        recipe: str | None = None,
    ) -> None:
        self.detector_kwargs = _normalize_component_kwargs(detector_kwargs, "detector_kwargs")
        self.reid_kwargs = _normalize_component_kwargs(reid_kwargs, "reid_kwargs")
        self.tracker_kwargs = _normalize_component_kwargs(tracker_kwargs, "tracker_kwargs")

        _reject_kwargs_with_initialized_component("detector", detector, self.detector_kwargs)
        _reject_kwargs_with_initialized_component("reid", reid, self.reid_kwargs)
        _reject_kwargs_with_initialized_component("tracker", tracker, self.tracker_kwargs)

        train_model = model
        train_recipe = recipe
        if train_model is not None and train_recipe is not None:
            raise ValueError("Provide only one of model=... or recipe=... when selecting a training profile.")

        if (
            detector is not _UNSET
            and reid is _UNSET
            and tracker is _UNSET
            and train_model is None
            and train_recipe is None
        ):
            train_spec = _resolve_reid_weight_train_spec(detector)
            if train_spec is not None:
                train_kind, train_name = train_spec
                reid = detector
                if train_kind == "model":
                    train_model = train_name
                else:
                    train_recipe = train_name
                detector = _UNSET
            else:
                train_spec = _resolve_train_model_spec(detector)
                if train_spec is not None:
                    train_kind, train_name = train_spec
                    if train_kind == "model":
                        train_model = train_name
                    else:
                        train_recipe = train_name
                    detector = _UNSET

        if train_model is not None:
            train_spec = _resolve_train_model_spec(train_model)
            if train_spec is not None:
                train_kind, train_name = train_spec
                if train_kind == "model":
                    train_model = train_name
                else:
                    train_model = None
                    train_recipe = train_name

        if train_recipe is not None:
            train_spec = _resolve_train_model_spec(train_recipe)
            if train_spec is not None:
                train_kind, train_name = train_spec
                if train_kind == "recipe":
                    train_recipe = train_name
                elif train_model is None:
                    train_model = train_name
                    train_recipe = None

        if train_model is None and train_recipe is None and reid is not _UNSET and reid is not None:
            train_spec = _resolve_reid_weight_train_spec(reid)
            if train_spec is not None:
                train_kind, train_name = train_spec
                if train_kind == "model":
                    train_model = train_name
                else:
                    train_recipe = train_name

        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._train_model_explicit = train_model is not None
        self._train_recipe_explicit = train_recipe 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.train_model = train_model
        self.train_recipe = train_recipe
        self.classes = _normalize_classes(classes)
        self.project = Path(project)

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

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

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

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

    def _tracker_config_from_spec(self) -> dict[str, Any] | None:
        tracker_config = _workflow_support().tracker_config_from_spec(self.tracker)
        tracker_kwargs = getattr(self, "tracker_kwargs", None)
        if not tracker_kwargs:
            return tracker_config

        if tracker_config is None:
            tracker_config = _workflow_support().default_tracker_config(self.tracker)
        tracker_config.update(tracker_kwargs)
        return tracker_config

    def _base_eval_args(
        self,
        benchmark: str | Path,
        *,
        split: str | None = None,
        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 _api_args().build_eval_args(
            self,
            benchmark,
            split=split,
            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:
        from boxmot.engine.tracking import workflow as tracker_module

        args = _api_args().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,
            detector_kwargs=getattr(self, "detector_kwargs", None),
            reid_kwargs=getattr(self, "reid_kwargs", None),
            tracker_kwargs=getattr(self, "tracker_kwargs", None),
            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)
        from boxmot.engine.eval import cache as cache_module
        from boxmot.engine.workflows.reporting import timing_summary_from_stats
        from boxmot.engine.workflows.results import GenerateResult

        args = _api_args().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,
        split: str | None = None,
        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,
            split=split,
            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.engine.eval import evaluator as evaluator_module
        from boxmot.utils.rich.reporters.eval 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,
        split: str | None = None,
        n_trials: int = BOXMOT_DEFAULTS.tune.n_trials,
        objectives: Sequence[str] | str | None = None,
        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] | str | None = None,
        minimize: Sequence[str] | str | None = None,
        verbose: bool = BOXMOT_DEFAULTS.eval.verbose,
        tracker_backend: str | None = None,
        tracking_backend: str = "thread",
        seed: int = 0,
    ) -> TuneResult:
        from boxmot.engine.tuning import tuner as tuner_module

        args = _api_args().build_tune_args(
            self,
            benchmark,
            split=split,
            n_trials=n_trials,
            objectives=objectives,
            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,
        hota_penalty: float = BOXMOT_DEFAULTS.research.hota_penalty,
        idf1_penalty: float = BOXMOT_DEFAULTS.research.idf1_penalty,
        mota_penalty: float = BOXMOT_DEFAULTS.research.mota_penalty,
        hota_tolerance: float = BOXMOT_DEFAULTS.research.hota_tolerance,
        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",
    ) -> ResearchResult:
        from boxmot.engine import research as research_module

        args = _api_args().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,
            hota_penalty=hota_penalty,
            idf1_penalty=idf1_penalty,
            mota_penalty=mota_penalty,
            hota_tolerance=hota_tolerance,
            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,
        *,
        format: str | Sequence[str] | None = None,
        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 = True,
        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,
        tflite_quantize: str = BOXMOT_DEFAULTS.export.tflite_quantize,
        tflite_calibration_data=None,
        tflite_calibration_samples: int = BOXMOT_DEFAULTS.export.tflite_calibration_samples,
        tflite_calibration_preprocess: str = BOXMOT_DEFAULTS.export.tflite_calibration_preprocess,
        tflite_calibration_seed: int = BOXMOT_DEFAULTS.export.tflite_calibration_seed,
        tflite_calibration_update: str = BOXMOT_DEFAULTS.export.tflite_calibration_update,
        tflite_static_activation_bits: int = BOXMOT_DEFAULTS.export.tflite_static_activation_bits,
    ) -> ExportResult:
        from boxmot.engine.reid import export as export_module

        export_include = _normalize_export_include(format, include)
        args = _api_args().build_export_args(
            self,
            include=export_include,
            device=device,
            half=half,
            optimize=optimize,
            dynamic=dynamic,
            simplify=simplify,
            opset=opset,
            workspace=workspace,
            verbose=verbose,
            batch_size=batch_size,
            imgsz=imgsz,
            tflite_quantize=tflite_quantize,
            tflite_calibration_data=tflite_calibration_data,
            tflite_calibration_samples=tflite_calibration_samples,
            tflite_calibration_preprocess=tflite_calibration_preprocess,
            tflite_calibration_seed=tflite_calibration_seed,
            tflite_calibration_update=tflite_calibration_update,
            tflite_static_activation_bits=tflite_static_activation_bits,
        )
        return export_module.run_export(args)

    def embed(
        self,
        *,
        source: str | Path | Any,
        boxes: Any = None,
        device: str = BOXMOT_DEFAULTS.track.device,
        half: bool = BOXMOT_DEFAULTS.track.half,
        preprocess: str | None = None,
    ):
        from boxmot.reid import ReID
        from boxmot.reid.core.preprocessing import get_preprocess_fn

        if isinstance(self.reid, (str, Path)):
            reid = ReID(
                self._reid_path(required=True),
                device=device,
                half=half,
                preprocess_name=preprocess,
            )
        else:
            reid = _workflow_support().build_reid_from_spec(self.reid, device=device, half=half)
            if preprocess is not None:
                reid.preprocess_name = preprocess
                backend = getattr(reid, "model", None)
                if backend is not None:
                    backend._preprocess_name = preprocess
                    backend.preprocess_fn = get_preprocess_fn(preprocess)

        return reid(source, boxes=boxes)

    def train(
        self,
        *,
        cfg: str | Path | None = None,
        recipe: str | None = None,
        model: str = BOXMOT_DEFAULTS.train.model,
        dataset: str = BOXMOT_DEFAULTS.train.dataset,
        data: str | Path | Sequence[str | Path] | None = None,
        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,
        id_loss_weight: float = BOXMOT_DEFAULTS.train.id_loss_weight,
        metric_loss_weight: float = BOXMOT_DEFAULTS.train.metric_loss_weight,
        early_id_loss_weight: float = BOXMOT_DEFAULTS.train.early_id_loss_weight,
        early_id_loss_epochs: int = BOXMOT_DEFAULTS.train.early_id_loss_epochs,
        center_loss_ramp_start_epoch: int = BOXMOT_DEFAULTS.train.center_loss_ramp_start_epoch,
        center_loss_ramp_end_epoch: int = BOXMOT_DEFAULTS.train.center_loss_ramp_end_epoch,
        metric_feature: str = BOXMOT_DEFAULTS.train.metric_feature,
        inference_feature: str = BOXMOT_DEFAULTS.train.inference_feature,
        feature_fusion: str = BOXMOT_DEFAULTS.train.feature_fusion,
        feat_dim: int = BOXMOT_DEFAULTS.train.feat_dim,
        neck_dim: int = BOXMOT_DEFAULTS.train.neck_dim,
        head_pool: str = BOXMOT_DEFAULTS.train.head_pool,
        head_parts: Sequence[int] = BOXMOT_DEFAULTS.train.head_parts,
        head_type: str = BOXMOT_DEFAULTS.train.head_type,
        part_pooling: str = BOXMOT_DEFAULTS.train.part_pooling,
        num_part_tokens: int = BOXMOT_DEFAULTS.train.num_part_tokens,
        evidence_num_roles: int = BOXMOT_DEFAULTS.train.evidence_num_roles,
        decouple_patterns: bool = BOXMOT_DEFAULTS.train.decouple_patterns,
        pattern_adapter_dim: int = BOXMOT_DEFAULTS.train.pattern_adapter_dim,
        stripe_visibility: bool = BOXMOT_DEFAULTS.train.stripe_visibility,
        branch_aware_metric: bool = BOXMOT_DEFAULTS.train.branch_aware_metric,
        branch_metric_part_weight: float = BOXMOT_DEFAULTS.train.branch_metric_part_weight,
        evidence_alignment_loss_weight: float = BOXMOT_DEFAULTS.train.evidence_alignment_loss_weight,
        evidence_alignment_margin: float = BOXMOT_DEFAULTS.train.evidence_alignment_margin,
        evidence_sinkhorn_iters: int = BOXMOT_DEFAULTS.train.evidence_sinkhorn_iters,
        evidence_sinkhorn_temperature: float = BOXMOT_DEFAULTS.train.evidence_sinkhorn_temperature,
        evidence_rerank_topk: int = BOXMOT_DEFAULTS.train.evidence_rerank_topk,
        evidence_null_loss_weight: float = BOXMOT_DEFAULTS.train.evidence_null_loss_weight,
        evidence_diversity_loss_weight: float = BOXMOT_DEFAULTS.train.evidence_diversity_loss_weight,
        head_warmup_epochs: int = BOXMOT_DEFAULTS.train.head_warmup_epochs,
        head_warmup_lr_mult: float = BOXMOT_DEFAULTS.train.head_warmup_lr_mult,
        gradual_unfreeze: bool = BOXMOT_DEFAULTS.train.gradual_unfreeze,
        gradual_unfreeze_head_epochs: int = BOXMOT_DEFAULTS.train.gradual_unfreeze_head_epochs,
        gradual_unfreeze_stage_epochs: int = BOXMOT_DEFAULTS.train.gradual_unfreeze_stage_epochs,
        gradual_unfreeze_backbone_lr_mult: float = BOXMOT_DEFAULTS.train.gradual_unfreeze_backbone_lr_mult,
        gradual_unfreeze_backbone_lr_epochs: int = BOXMOT_DEFAULTS.train.gradual_unfreeze_backbone_lr_epochs,
        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,
        deterministic: bool = BOXMOT_DEFAULTS.train.deterministic,
        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,
        random_erasing: float = BOXMOT_DEFAULTS.train.random_erasing,
        resume: str | Path | None = None,
    ):
        train_project = project if project is not None else BOXMOT_DEFAULTS.train.project
        from boxmot.configs import build_mode_namespace
        from boxmot.engine.reid import trainer as reid_trainer_module
        from boxmot.engine.reid.data import resolve_reid_train_data

        train_model = model
        train_recipe = recipe if recipe is not None else self.train_recipe
        if model == BOXMOT_DEFAULTS.train.model and self.train_model is not None:
            train_model = self.train_model
        elif recipe is None:
            train_spec = _resolve_train_model_spec(model)
            if train_spec is not None:
                train_kind, train_name = train_spec
                if train_kind == "model":
                    train_model = train_name
                else:
                    train_model = BOXMOT_DEFAULTS.train.model
                    train_recipe = train_name

        if train_recipe is not None:
            train_spec = _resolve_train_model_spec(train_recipe)
            if train_spec is not None and train_spec[0] == "recipe":
                train_recipe = train_spec[1]

        payload = {
            "cfg": None if cfg is None else str(cfg),
            "recipe": train_recipe,
            "model": train_model,
            "dataset": dataset,
            "data": _normalize_reid_data(data),
            "data_dir": None if data_dir is None else str(data_dir),
            "loss": loss,
            "preprocess": preprocess,
            "imgsz": imgsz if imgsz is not None else BOXMOT_DEFAULTS.train.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),
            "id_loss_weight": float(id_loss_weight),
            "metric_loss_weight": float(metric_loss_weight),
            "early_id_loss_weight": float(early_id_loss_weight),
            "early_id_loss_epochs": int(early_id_loss_epochs),
            "center_loss_ramp_start_epoch": int(center_loss_ramp_start_epoch),
            "center_loss_ramp_end_epoch": int(center_loss_ramp_end_epoch),
            "metric_feature": metric_feature,
            "inference_feature": inference_feature,
            "feature_fusion": feature_fusion,
            "feat_dim": int(feat_dim),
            "neck_dim": int(neck_dim),
            "head_pool": head_pool,
            "head_parts": tuple(int(part) for part in head_parts),
            "head_type": head_type,
            "part_pooling": part_pooling,
            "num_part_tokens": int(num_part_tokens),
            "evidence_num_roles": int(evidence_num_roles),
            "decouple_patterns": bool(decouple_patterns),
            "pattern_adapter_dim": int(pattern_adapter_dim),
            "stripe_visibility": bool(stripe_visibility),
            "branch_aware_metric": bool(branch_aware_metric),
            "branch_metric_part_weight": float(branch_metric_part_weight),
            "evidence_alignment_loss_weight": float(evidence_alignment_loss_weight),
            "evidence_alignment_margin": float(evidence_alignment_margin),
            "evidence_sinkhorn_iters": int(evidence_sinkhorn_iters),
            "evidence_sinkhorn_temperature": float(evidence_sinkhorn_temperature),
            "evidence_rerank_topk": int(evidence_rerank_topk),
            "evidence_null_loss_weight": float(evidence_null_loss_weight),
            "evidence_diversity_loss_weight": float(evidence_diversity_loss_weight),
            "head_warmup_epochs": int(head_warmup_epochs),
            "head_warmup_lr_mult": float(head_warmup_lr_mult),
            "gradual_unfreeze": bool(gradual_unfreeze),
            "gradual_unfreeze_head_epochs": int(gradual_unfreeze_head_epochs),
            "gradual_unfreeze_stage_epochs": int(gradual_unfreeze_stage_epochs),
            "gradual_unfreeze_backbone_lr_mult": float(gradual_unfreeze_backbone_lr_mult),
            "gradual_unfreeze_backbone_lr_epochs": int(gradual_unfreeze_backbone_lr_epochs),
            "pretrained": bool(pretrained),
            "device": device,
            "project": Path(train_project),
            "name": name,
            "num_workers": int(num_workers),
            "seed": int(seed),
            "deterministic": bool(deterministic),
            "eval_datasets": list(eval_datasets),
            "ema_decay": ema_decay,
            "gaussian_blur": bool(gaussian_blur),
            "random_grayscale": float(random_grayscale),
            "color_jitter": bool(color_jitter),
            "random_erasing": float(random_erasing),
            "resume": None if resume is None else str(resume),
        }
        explicit_keys = None
        if cfg is not None:
            explicit_keys = {"cfg"}
            if data is not None:
                explicit_keys.add("data")
            if project is not None:
                explicit_keys.add("project")
            if resume is not None:
                explicit_keys.add("resume")
            for key, value in payload.items():
                if (
                    value is not None
                    and key not in {"cfg", "data", "project", "resume"}
                    and not _matches_train_default(key, value)
                ):
                    explicit_keys.add(key)

        args = build_mode_namespace("train", payload, explicit_keys=explicit_keys)
        args = resolve_reid_train_data(args)
        return reid_trainer_module.main(args)

    def eval_reid(
        self,
        *,
        weights: str | Path,
        dataset: str,
        data_dir: str | Path,
        model: str | None = None,
        preprocess: str | None = None,
        imgsz: int | tuple[int, int] | None = None,
        inference_feature: str | None = None,
        flip_tta: bool | 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),
            preprocess=preprocess,
            imgsz=imgsz,
            inference_feature=inference_feature,
            flip_tta=flip_tta,
            device=device,
            batch_size=int(batch_size),
            num_workers=int(num_workers),
            output=None if output is None else str(output),
        )
        from boxmot.engine.reid import evaluator as reid_evaluator_module

        return reid_evaluator_module.main(args)

Runtime wrappers

Bases: Detector

High-level detector runtime with user-facing constructor names.

This wrapper keeps boxmot.detectors.Detector as the lower-level runtime while making from boxmot import Detector fit the public API.

Source code in boxmot/api/runtime.py

class Detector(_Detector):
    """High-level detector runtime with user-facing constructor names.

    This wrapper keeps ``boxmot.detectors.Detector`` as the lower-level runtime
    while making ``from boxmot import Detector`` fit the public API.
    """

    def __init__(
        self,
        model: str | Path,
        *,
        device: str = BOXMOT_DEFAULTS.track.device,
        image_size=None,
        confidence: float | None = None,
        iou: float = BOXMOT_DEFAULTS.track.iou,
        classes: Any = None,
        agnostic_nms: bool = False,
        half: bool = BOXMOT_DEFAULTS.track.half,
        batch: int = 1,
        vid_stride: int = 1,
    ) -> None:
        self.half = bool(half)
        super().__init__(
            path=_ensure_model_path(model),
            device=device,
            imgsz=image_size,
            conf=confidence,
            iou=iou,
            classes=classes,
            agnostic_nms=agnostic_nms,
            batch=batch,
            vid_stride=vid_stride,
        )

    @property
    def image_size(self):
        return self.imgsz

    @property
    def confidence(self) -> float:
        return self.conf

    @staticmethod
    def _unwrap_single(result):
        if isinstance(result, list) and len(result) == 1:
            return result[0]
        return result

    def predict(self, source, **kwargs):
        """Run inference and return ``Detections`` for a single image."""
        kwargs.setdefault("as_detections", True)
        return self._unwrap_single(super().predict(source, **kwargs))

    def __call__(self, source, stream: bool = False, **kwargs):
        kwargs.setdefault("as_detections", True)
        result = super().__call__(source, stream=stream, **kwargs)
        if stream:
            return result
        return self._unwrap_single(result)

predict(source, **kwargs)

Run inference and return Detections for a single image.

Source code in boxmot/api/runtime.py

def predict(self, source, **kwargs):
    """Run inference and return ``Detections`` for a single image."""
    kwargs.setdefault("as_detections", True)
    return self._unwrap_single(super().predict(source, **kwargs))

High-level ReID runtime with embed/export helpers.

Source code in boxmot/api/runtime.py

class ReIDModel:
    """High-level ReID runtime with embed/export helpers."""

    def __init__(
        self,
        weights: str | Path,
        *,
        device: str = BOXMOT_DEFAULTS.track.device,
        half: bool = BOXMOT_DEFAULTS.track.half,
        preprocess: str | None = None,
    ) -> None:
        self.runtime = ReID(
            _ensure_model_path(weights),
            device=device,
            half=half,
            preprocess_name=preprocess,
        )
        self.path = self.runtime.path
        self.weights = self.runtime.weights
        self.device = self.runtime.device
        self.half = self.runtime.half
        self.preprocess = self.runtime.preprocess_name
        self.model = self.runtime.model
        self.export_result = None

    def embed(self, source: str | Path | Any, *, boxes: Any = None) -> np.ndarray:
        """Generate embeddings for image crops or image+box detections."""
        return self.runtime(source, boxes=boxes)

    def __call__(self, source: str | Path | Any, *, boxes: Any = None) -> np.ndarray:
        return self.embed(source, boxes=boxes)

    def get_features(self, boxes: Any, image: Any) -> np.ndarray:
        """Tracker-facing feature extraction contract."""
        return self.embed(image, boxes=boxes)

    def export(
        self,
        *,
        format: str | Sequence[str] | None = None,
        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 = True,
        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,
        tflite_quantize: str = BOXMOT_DEFAULTS.export.tflite_quantize,
        tflite_calibration_data=None,
        tflite_calibration_samples: int = BOXMOT_DEFAULTS.export.tflite_calibration_samples,
        tflite_calibration_preprocess: str = BOXMOT_DEFAULTS.export.tflite_calibration_preprocess,
        tflite_calibration_seed: int = BOXMOT_DEFAULTS.export.tflite_calibration_seed,
        tflite_calibration_update: str = BOXMOT_DEFAULTS.export.tflite_calibration_update,
        tflite_static_activation_bits: int = BOXMOT_DEFAULTS.export.tflite_static_activation_bits,
    ) -> "ReIDModel":
        from boxmot.pipeline import BoxMOT

        result = BoxMOT(reid=self.path).export(
            format=format,
            include=include,
            device=device,
            half=half,
            optimize=optimize,
            dynamic=dynamic,
            simplify=simplify,
            opset=opset,
            workspace=workspace,
            verbose=verbose,
            batch_size=batch_size,
            imgsz=imgsz,
            tflite_quantize=tflite_quantize,
            tflite_calibration_data=tflite_calibration_data,
            tflite_calibration_samples=tflite_calibration_samples,
            tflite_calibration_preprocess=tflite_calibration_preprocess,
            tflite_calibration_seed=tflite_calibration_seed,
            tflite_calibration_update=tflite_calibration_update,
            tflite_static_activation_bits=tflite_static_activation_bits,
        )
        exported = type(self)(
            result.embedding_weights,
            device=device,
            half=result.half,
            preprocess=self.preprocess,
        )
        exported.export_result = result
        return exported

embed(source, *, boxes=None)

Generate embeddings for image crops or image+box detections.

Source code in boxmot/api/runtime.py

def embed(self, source: str | Path | Any, *, boxes: Any = None) -> np.ndarray:
    """Generate embeddings for image crops or image+box detections."""
    return self.runtime(source, boxes=boxes)

get_features(boxes, image)

Tracker-facing feature extraction contract.

Source code in boxmot/api/runtime.py

def get_features(self, boxes: Any, image: Any) -> np.ndarray:
    """Tracker-facing feature extraction contract."""
    return self.embed(image, boxes=boxes)

Workflow helpers

Create a lazy streaming tracking result iterator.

Source code in boxmot/api/functional.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.")

    from boxmot.engine.tracking.results import Results

    return Results(source, detector, reid, tracker, verbose=verbose, drawer=drawer)

Aggregate run metrics over one or more tracking results.

Source code in boxmot/api/functional.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/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:
        return as_2d_array(values, empty_cols=empty_cols)

    @staticmethod
    def _extract_detections(output: Any) -> np.ndarray:
        return extract_detection_array(output)

    @staticmethod
    def _extract_masks(output: Any) -> np.ndarray | None:
        return extract_masks(output)

    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
            label = str(entry["label"])
            total = float(entry["total"])
            avg = float(entry["avg"])
            fps = float(entry["fps"])
            lines.append(
                f"{label:<20} {total:>12.1f} {avg:>12.2f} {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:
        return as_2d_array(values)

    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:
        return draw_tracks(frame, self.tracks, self.masks)

    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.
        """
        append_frame(path, self.plot(), self.source_path, fps=fps)

    @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.
        """
        close_video(path)

    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.
    """
    close_video(path)

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.
    """
    append_frame(path, self.plot(), self.source_path, fps=fps)

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}]"