predictor

sleap_nn.inference.predictor

Predictor — high-level orchestrator for the inference stack.

Composes an :class:InferenceLayer (or composed layer like :class:TopDownLayer) with a :class:Provider source and a :class:FilterPipeline post-processor.

Three usage tiers:

• :meth:predict — synchronous, returns sio.Labels (or a list of Outputs if make_labels=False). Loads everything into memory; use for short videos / interactive sessions.
• :meth:predict_streaming — yields one Outputs per batch as a generator; nothing is retained across batches, so memory stays O(batch) (with paf_workers>0, O(in-flight window)). Tracking is not supported here (it needs the full frame list); use :meth:predict.
• :meth:predict_to_file — buffered write of a .slp via :class:IncrementalLabelsWriter. Heavy tensors are dropped per batch; the (slimmed) LabeledFrames accumulate until finalize (see that class for the memory note).

Classes:

Name	Description
Predictor	High-level orchestrator: layer + source dispatch + filter pipeline.

Predictor

High-level orchestrator: layer + source dispatch + filter pipeline.

Parameters:

Name	Type	Description	Default
layer		Any object exposing predict(image) -> Outputs. Includes every :class:InferenceLayer subclass plus composed layers like :class:TopDownLayer.	required
skeleton		Optional sio.Skeleton resolved from the training config. Populated automatically by :meth:from_model_paths. Used as the default for predict(make_labels=True) and predict_to_file() when no explicit skeleton kwarg is passed.	required
batch_size		Default batch size for auto-constructed providers when predict / predict_streaming receive an sio.Video or sio.Labels instead of a pre-built Provider.	required
filter_config		Optional post-inference filter config. Default is the no-op identity.	required
paf_workers		Number of CPU worker processes for the bottom-up PAF grouping stage. 0 (default) runs grouping inline in the main process — the parity path. >0 is only honored when layer is a :class:BottomUpLayer; for any other layer type the value is ignored (with a logged warning, since other model types have no equivalent pipelined CPU stage yet).	required
tracker_config		Optional :class:TrackerConfig. When set, :meth:predict runs the tracker on the resulting sio.Labels (requires make_labels=True) before returning.	required

Notes

Keeps no state across calls — same predictor can be reused on multiple sources safely.

Methods:

Name	Description
__attrs_post_init__	Warn (once) if paf_workers was set on a layer that can't use it.
from_export_dir	Build a :class:Predictor from an exported ONNX/TensorRT directory.
from_model_paths	Build a :class:Predictor from one or more checkpoint paths.
predict	Run inference on a source.
predict_streaming	Yield one Outputs per batch from source.
predict_to_file	Run inference and write results to a .slp file.
retrack	Retrack an existing sio.Labels without running inference.
to_labels	Concatenate per-batch Outputs into a single sio.Labels.

Attributes:

Name	Type	Description
filter_pipeline	FilterPipeline	Build a fresh FilterPipeline from the config (cheap).

Source code in sleap_nn/inference/predictor.py

@attrs.define
class Predictor:
    """High-level orchestrator: layer + source dispatch + filter pipeline.

    Args:
        layer: Any object exposing ``predict(image) -> Outputs``. Includes
            every :class:`InferenceLayer` subclass plus composed layers
            like :class:`TopDownLayer`.
        skeleton: Optional ``sio.Skeleton`` resolved from the training
            config. Populated automatically by
            :meth:`from_model_paths`.
            Used as the default for ``predict(make_labels=True)`` and
            ``predict_to_file()`` when no explicit ``skeleton`` kwarg is
            passed.
        batch_size: Default batch size for auto-constructed providers when
            ``predict`` / ``predict_streaming`` receive an ``sio.Video``
            or ``sio.Labels`` instead of a pre-built ``Provider``.
        filter_config: Optional post-inference filter config. Default is
            the no-op identity.
        paf_workers: Number of CPU worker processes for the bottom-up
            PAF grouping stage. ``0`` (default) runs grouping inline in
            the main process — the parity path. ``>0`` is only honored
            when ``layer`` is a :class:`BottomUpLayer`; for any other
            layer type the value is ignored (with a logged warning, since
            other model types have no equivalent pipelined CPU stage yet).
        tracker_config: Optional :class:`TrackerConfig`. When set,
            :meth:`predict` runs the tracker on the resulting
            ``sio.Labels`` (requires ``make_labels=True``) before
            returning.

    Notes:
        Keeps no state across calls — same predictor can be reused on
        multiple sources safely.
    """

    layer: Any  # TODO: unify layer types under a common Protocol
    skeleton: Optional["sio.Skeleton"] = None
    batch_size: int = 4
    filter_config: FilterConfig = attrs.Factory(FilterConfig)
    paf_workers: int = 0
    tracker_config: Optional[TrackerConfig] = None
    # Provenance metadata (populated by the factories; attached to the saved /
    # returned Labels by predict() so .slp files carry inference lineage —
    # #530 gap: the new flow previously wrote no provenance).
    model_paths: Optional[List[str]] = None
    device: Optional[str] = None
    # Centroid-only output representation: "instance" (default; single-node
    # PredictedInstance), "centroid" (sio.PredictedCentroid), or "both".
    emit_centroid: str = "instance"

    def __attrs_post_init__(self) -> None:
        """Warn (once) if ``paf_workers`` was set on a layer that can't use it.

        ``paf_workers > 0`` only pipelines the CPU-bound grouping stage for
        plain bottom-up (:class:`BottomUpLayer`); every other layer type
        (including :class:`BottomUpMultiClassLayer`, which has its own
        CPU-bound Hungarian-matching identity step) silently ignores the
        setting today. Without this, that's an easy-to-miss no-op — a user
        expecting a speedup gets none, with no signal why.
        """
        if self.paf_workers > 0 and not self._can_pipeline():
            logger.warning(
                f"paf_workers={self.paf_workers} was set, but pipelined CPU "
                f"grouping is only implemented for plain bottom-up models. "
                f"layer={type(self.layer).__name__} will run the inline "
                "(unpipelined) path; paf_workers has no effect here."
            )

    @property
    def filter_pipeline(self) -> FilterPipeline:
        """Build a fresh ``FilterPipeline`` from the config (cheap)."""
        return FilterPipeline(self.filter_config)

    # ──────────────────────────────────────────────────────────────────
    # Factory classmethods
    # ──────────────────────────────────────────────────────────────────

    @classmethod
    def from_model_paths(
        cls,
        model_paths: List[str],
        *,
        device: str = "cpu",
        batch_size: int = 4,
        backbone_ckpt_path: Optional[str] = None,
        head_ckpt_path: Optional[str] = None,
        peak_threshold: Union[float, List[float]] = 0.2,
        integral_refinement: str = "integral",
        integral_patch_size: int = 5,
        max_instances: Optional[int] = None,
        return_confmaps: bool = False,
        preprocess_config: Optional[Any] = None,
        anchor_part: Optional[str] = None,
        filter_config: Optional["FilterConfig"] = None,
        paf_workers: int = 0,
        tracker_config: Optional["TrackerConfig"] = None,
        centroid_only: bool = False,
        emit_centroid: str = "instance",
        max_edge_length_ratio: float = 0.25,
        dist_penalty_weight: float = 1.0,
        n_points: int = 10,
        min_instance_peaks: Union[int, float] = 0,
        min_line_scores: float = 0.25,
        fg_threshold: float = 0.5,
        min_mask_area: int = 0,
        center_nms_kernel: int = 3,
        mask_cleanup: bool = False,
        mask_cleanup_radius: int = 0,
        distance_gate_alpha: Optional[float] = None,
        merge_fragments: bool = False,
        merge_method: str = "greedy",
        merge_thresholds: tuple = (0.85, 0.6, 0.4),
        merge_w_valley: float = 1.0,
        merge_w_offset: float = 0.25,
        merge_dilate: int = 1,
        full_res_masks: bool = False,
        mask_output: str = "mask",
        polygon_epsilon: float = 0.01,
    ) -> "Predictor":
        """Build a :class:`Predictor` from one or more checkpoint paths.

        Args:
            model_paths: Trained model directories containing
                ``training_config.{yaml,json}`` + ``best.ckpt``. Each entry may
                alternatively be a path to that ``best.ckpt`` or
                ``training_config.{yaml,json}`` file; all forms resolve to the
                model directory and load ``best.ckpt`` (#575). For top-down, pass
                two paths (centroid + centered-instance) in either order.
            device: ``"cpu"``, ``"cuda"``, ``"mps"``, or ``"cuda:N"``.
            batch_size: Default batch size for auto-constructed providers.
            backbone_ckpt_path: Override backbone weights with this ``.ckpt``.
            head_ckpt_path: Override head weights.
            peak_threshold: Default peak threshold. ``List[float]`` for
                top-down (``[centroid_thresh, keypoint_thresh]``). Can be
                overridden per-call via ``predict(peak_threshold=...)``.
            integral_refinement: ``"integral"`` or ``"none"``.
            integral_patch_size: Refinement patch size.
            max_instances: Cap on instances per frame.
            return_confmaps: Return confidence maps on Outputs.
            preprocess_config: OmegaConf overrides for preprocessing.
            anchor_part: Override centroid anchor node name.
            filter_config: Post-inference :class:`FilterConfig`.
            paf_workers: CPU workers for bottom-up PAF grouping.
            tracker_config: :class:`TrackerConfig` for tracking.
            centroid_only: Force centroid-only output even when a
                centered-instance model is among ``model_paths``.
            emit_centroid: Centroid-only output representation: ``"instance"``
                (default; single-node ``PredictedInstance``), ``"centroid"``
                (``sio.PredictedCentroid``), or ``"both"``. Honored only for
                centroid-only layers.
            max_edge_length_ratio: Bottom-up PAF max edge length ratio.
            dist_penalty_weight: Bottom-up PAF distance penalty weight.
            n_points: Bottom-up PAF line integration sample count.
            min_instance_peaks: Bottom-up min peaks for a valid instance.
            min_line_scores: Bottom-up per-edge match threshold. (These five
                are applied only to plain bottom-up models.)
            fg_threshold: Foreground probability threshold for binarizing the
                segmentation map (bottom-up segmentation only).
            min_mask_area: Minimum predicted-mask area in original-image pixels;
                smaller masks are dropped to suppress over-segmentation. ``0``
                disables it (bottom-up segmentation only).
            center_nms_kernel: Odd window size for center-peak NMS; larger merges
                nearby duplicate centers (bottom-up segmentation only).
            mask_cleanup: Keep-largest-CC + hole-fill per mask (bottom-up
                segmentation only).
            mask_cleanup_radius: Morphological open->close radius (output-stride
                pixels) applied during ``mask_cleanup``; ``0`` keeps keep-largest
                + fill only (bottom-up segmentation only).
            distance_gate_alpha: Adaptive distance-gate strength; ``None``
                (default) keeps the byte-for-byte argmin grouping (bottom-up
                segmentation only).
            merge_fragments: Enable the RAG fragment-merge to re-fuse
                over-segmented animal halves; ``False`` (default) is byte-for-byte
                today (bottom-up segmentation only).
            merge_method: ``"greedy"`` (default) or ``"multicut"`` agglomeration;
                inert when ``merge_fragments=False`` (bottom-up segmentation only).
            merge_thresholds: Greedy-merge decreasing affinity thresholds; inert
                when ``merge_fragments=False`` (bottom-up segmentation only).
            merge_w_valley: Center-valley merge-term weight; inert when off
                (bottom-up segmentation only).
            merge_w_offset: Offset-agreement merge-term weight; inert when off
                (bottom-up segmentation only).
            merge_dilate: Merge contact-test dilation iterations; inert when off
                (bottom-up segmentation only).
            full_res_masks: Encode masks at full original resolution instead of
                the model output-stride grid (default ``False``: stride encoding
                is ~stride^2 smaller and lossless at model resolution; bottom-up
                segmentation only).
            mask_output: Mask output representation: ``"mask"`` (default),
                ``"polygon"`` (Douglas-Peucker ``sio.PredictedROI`` only), or
                ``"both"`` (bottom-up segmentation only).
            polygon_epsilon: Douglas-Peucker tolerance (fraction of perimeter)
                for ``mask_output`` polygon/both (bottom-up segmentation only).
        """
        from sleap_nn.inference.loaders import load_model_assets
        from sleap_nn.system_info import get_startup_info_string

        logger.info(get_startup_info_string())

        loaded, model_types = load_model_assets(
            model_paths,
            device=device,
            backbone_ckpt_path=backbone_ckpt_path,
            head_ckpt_path=head_ckpt_path,
            peak_threshold=peak_threshold,
            integral_refinement=integral_refinement,
            integral_patch_size=integral_patch_size,
            max_instances=max_instances,
            return_confmaps=return_confmaps,
            preprocess_config=preprocess_config,
            anchor_part=anchor_part,
            max_edge_length_ratio=max_edge_length_ratio,
            dist_penalty_weight=dist_penalty_weight,
            n_points=n_points,
            min_instance_peaks=min_instance_peaks,
            min_line_scores=min_line_scores,
            fg_threshold=fg_threshold,
            min_mask_area=min_mask_area,
            center_nms_kernel=center_nms_kernel,
            mask_cleanup=mask_cleanup,
            mask_cleanup_radius=mask_cleanup_radius,
            distance_gate_alpha=distance_gate_alpha,
            merge_fragments=merge_fragments,
            merge_method=merge_method,
            merge_thresholds=merge_thresholds,
            merge_w_valley=merge_w_valley,
            merge_w_offset=merge_w_offset,
            merge_dilate=merge_dilate,
            full_res_masks=full_res_masks,
            mask_output=mask_output,
            polygon_epsilon=polygon_epsilon,
        )

        if centroid_only:
            if "centroid" not in model_types:
                raise ValueError(
                    "centroid_only=True requires a centroid model in model_paths; "
                    f"detected types: {model_types}."
                )
            layer = _build_centroid_layer(
                loaded.inference_model.centroid_crop,
                device,
                assets=loaded,
            )
        else:
            layer = _select_layer(loaded, model_types, device)

        skeleton = loaded.skeletons[0] if loaded.skeletons else None
        kwargs: dict = {
            "layer": layer,
            "skeleton": skeleton,
            "batch_size": batch_size,
            "paf_workers": paf_workers,
            "model_paths": [str(p) for p in model_paths],
            "device": device,
            "emit_centroid": emit_centroid,
        }
        if filter_config is not None:
            kwargs["filter_config"] = filter_config
        if tracker_config is not None:
            kwargs["tracker_config"] = tracker_config

        # Spin-up log: a one-line record of *what* model is running on *what*,
        # so a run starts with a legible header instead of silence (#610).
        n_nodes = len(skeleton.nodes) if skeleton is not None else None
        spec = [
            f"type={'+'.join(model_types)}",
            f"backbone={loaded.backbone_type}",
            f"nodes={n_nodes}",
            f"device={device}",
            f"batch_size={batch_size}",
            f"peak_threshold={peak_threshold}",
            f"max_instances={max_instances}",
            f"integral_refinement={integral_refinement}",
            f"paf_workers={paf_workers}",
        ]
        if "bottomup_segmentation" in model_types:
            spec.append(f"fg_threshold={fg_threshold}")
            spec.append(f"min_mask_area={min_mask_area}")
            spec.append(f"distance_gate_alpha={distance_gate_alpha}")
            spec.append(f"merge_fragments={merge_fragments}")
            spec.append(f"full_res_masks={full_res_masks}")
            spec.append(f"mask_output={mask_output}")
        if "centered_instance_segmentation" in model_types:
            spec.append(f"fg_threshold={fg_threshold}")
            spec.append(f"mask_output={mask_output}")
        if "semantic_segmentation" in model_types:
            spec.append(f"fg_threshold={fg_threshold}")
            spec.append(f"min_mask_area={min_mask_area}")
            spec.append(f"full_res_masks={full_res_masks}")
            spec.append(f"mask_output={mask_output}")
        logger.info("Loaded inference model | " + " | ".join(spec))

        return cls(**kwargs)

    @classmethod
    def from_export_dir(
        cls,
        export_dir: Union[str, Any],
        *,
        runtime: str = "auto",
        device: str = "auto",
        batch_size: int = 4,
        return_confmaps: bool = False,
        filter_config: Optional["FilterConfig"] = None,
        paf_workers: int = 0,
        tracker_config: Optional["TrackerConfig"] = None,
        max_instances: Optional[int] = None,
        min_instance_peaks: float = 0,
        min_line_scores: float = 0.25,
        peak_conf_threshold: Optional[float] = None,
        emit_centroid: str = "instance",
    ) -> "Predictor":
        """Build a :class:`Predictor` from an exported ONNX/TensorRT directory.

        Args:
            export_dir: Directory containing ``export_metadata.json`` +
                ``model.onnx`` or ``model.trt``.
            runtime: ``"auto"`` (prefer TRT), ``"onnx"``, or ``"tensorrt"``.
            device: Device string.
            batch_size: Default batch size.
            return_confmaps: Return confidence maps on Outputs.
            filter_config: Post-inference :class:`FilterConfig`.
            paf_workers: CPU workers for bottom-up PAF grouping.
            tracker_config: :class:`TrackerConfig` for tracking.
            max_instances: Cap on instances per frame (bottom-up).
            min_instance_peaks: Min peaks for a valid instance (bottom-up).
            min_line_scores: Per-edge match threshold (bottom-up).
            peak_conf_threshold: Runtime peak-confidence threshold for the
                exported bottom-up path. Gates PAF candidate connections by the
                src/dst peak confidence (legacy parity). Defaults to the
                threshold baked at export time (``metadata.peak_threshold``).
                Note the wrapper already bakes a peak threshold during peak
                finding, so this can only *tighten* beyond the baked value.
            emit_centroid: Centroid-only output representation for an exported
                standalone centroid model: ``"instance"`` (default; single-node
                ``PredictedInstance``, frontend-compatible), ``"centroid"``
                (``sio.PredictedCentroid``), or ``"both"``. Honored only for
                ``ExportedCentroidLayer``; mirrors ``from_model_paths`` so the
                exported runtime matches the checkpoint path.
        """
        from sleap_nn.export.metadata import ExportMetadata

        export_dir = Path(export_dir)

        metadata_path = export_dir / "export_metadata.json"
        if not metadata_path.exists():
            raise FileNotFoundError(
                f"export_metadata.json not found at {metadata_path}. "
                f"Pass a directory written by `sleap_nn export`."
            )
        metadata = ExportMetadata.load(metadata_path)

        runtime, model_path = _resolve_export_runtime(export_dir, runtime)
        backend = _build_export_backend(runtime, model_path, device)

        # Default the runtime peak-confidence threshold to the value baked at
        # export time, falling back to legacy's 0.2 when the metadata carries no
        # baked threshold (matches legacy export inference, #582).
        if peak_conf_threshold is not None:
            resolved_peak_conf = peak_conf_threshold
        else:
            meta_thr = getattr(metadata, "peak_threshold", None)
            resolved_peak_conf = meta_thr if meta_thr is not None else 0.2
        layer = _select_export_layer(
            metadata=metadata,
            backend=backend,
            return_confmaps=return_confmaps,
            max_instances=max_instances,
            min_instance_peaks=min_instance_peaks,
            min_line_scores=min_line_scores,
            peak_conf_threshold=resolved_peak_conf,
        )

        skeleton = _skeleton_from_export(export_dir, metadata)
        kwargs: dict = {
            "layer": layer,
            "skeleton": skeleton,
            "batch_size": batch_size,
            "paf_workers": paf_workers,
            "model_paths": [str(export_dir)],
            "device": device,
            "emit_centroid": emit_centroid,
        }
        if filter_config is not None:
            kwargs["filter_config"] = filter_config
        if tracker_config is not None:
            kwargs["tracker_config"] = tracker_config
        return cls(**kwargs)

    # ──────────────────────────────────────────────────────────────────
    # Source dispatch: sio.Video / sio.Labels / str / Provider
    # ──────────────────────────────────────────────────────────────────

    def _needs_gt_instances(self) -> bool:
        """Whether the configured layer consumes ground-truth instances.

        GT-fallback layers (``CentroidLayer(use_gt_centroids=True)`` /
        ``CenteredInstanceLayer(use_gt_peaks=True)``) require frames that carry
        user instances, so a ``.slp`` source must be restricted to labeled
        frames. The normal (real-model) path predicts ALL frames, matching the
        legacy ``LabelsReader`` default (#530 audit: new flow wrongly defaulted
        to labeled-only).
        """
        layer = self.layer
        candidates = [layer]
        for sub in ("centroid_layer", "centered_instance_layer"):
            inner = getattr(layer, sub, None)
            if inner is not None:
                candidates.append(inner)
        return any(
            getattr(c, "use_gt_centroids", False) or getattr(c, "use_gt_peaks", False)
            for c in candidates
        )

    def _preprocess_provenance_params(self) -> dict:
        """The scale/crop_size actually used for this run, for provenance.

        Best-effort and defensive -- provenance must never break inference, so
        every lookup falls back to omitting the field rather than raising.
        Topdown-family layers (``TopDownLayer`` and its ``TopDownSegmentation``/
        ``TopDownMultiClass`` subclasses) have two independently-scaled stages,
        so both are recorded distinctly rather than collapsing to one shared
        "scale" the way the training-config's own baked-in value might suggest.
        """
        layer = getattr(self.layer, "inner", self.layer)  # unwrap Tiled* wrappers
        centroid_layer = getattr(layer, "centroid_layer", None)
        instance_layer = getattr(layer, "centered_instance_layer", None)
        if centroid_layer is None and instance_layer is None:
            return {
                "scale": getattr(
                    getattr(layer, "preprocess_config", None), "scale", None
                )
            }
        params: dict = {}
        if centroid_layer is not None:
            params["centroid_scale"] = getattr(
                getattr(centroid_layer, "preprocess_config", None), "scale", None
            )
        if instance_layer is not None:
            params["instance_scale"] = getattr(
                getattr(instance_layer, "preprocess_config", None), "scale", None
            )
        crop_size = getattr(layer, "crop_size", None)
        if crop_size is not None:
            params["crop_size"] = crop_size
        return params

    def _build_inference_provenance(
        self,
        *,
        source: Any,
        start_time: Any,
        end_time: Any,
        n_frames: int,
        inference_params: dict,
    ) -> dict:
        """Provenance dict for the saved/returned Labels (#530 gap fix)."""
        from sleap_nn.inference.provenance import build_inference_provenance

        tracking_params = (
            attrs.asdict(self.tracker_config)
            if self.tracker_config is not None
            else None
        )
        return build_inference_provenance(
            model_paths=self.model_paths,
            model_type=type(self.layer).__name__,
            device=self.device,
            start_time=start_time,
            end_time=end_time,
            input_path=source if isinstance(source, str) else None,
            frames_processed=n_frames,
            inference_params=inference_params,
            tracking_params=tracking_params,
        )

    @staticmethod
    def _describe_source(source: Any) -> str:
        """Best-effort human label for a prediction source (#610)."""
        if isinstance(source, str):
            return source
        filename = getattr(source, "filename", None)
        if filename:
            return str(filename)
        return type(source).__name__

    def _log_inference_start(
        self, source: Any, provider: "Provider", videos: Optional[list]
    ) -> None:
        """Log a one-line spin-up record of the source being processed (#610)."""
        n_frames = _safe_num_frames(provider)
        parts = [
            f"source={self._describe_source(source)}",
            f"frames={n_frames if n_frames >= 0 else '?'}",
            f"videos={len(videos) if videos else 1}",
        ]
        sio_video = getattr(provider, "_sio_video", None)
        if sio_video is not None:
            try:
                shape = tuple(sio_video.shape)  # (N, H, W, C)
                if len(shape) == 4:
                    parts.append(f"shape={shape[1]}x{shape[2]}x{shape[3]}")
            except Exception:  # pragma: no cover — metadata best-effort
                pass
            fps = getattr(sio_video, "fps", None)
            if fps:
                parts.append(f"fps={fps}")
        parts.append(f"tracking={self.tracker_config is not None}")
        logger.info("Starting inference | " + " | ".join(parts))

    def _log_filter_config(self) -> None:
        """Log which post-inference filters are active, with their values.

        Matches legacy ``run_inference``'s per-filter confirmation messages
        -- useful for confirming a filter flag actually took effect (silent
        no-ops here have bitten us before, see #715/#716/#717).
        """
        cfg = self.filter_config
        if cfg.min_visible_nodes > 0 or cfg.min_visible_node_fraction > 0.0:
            logger.info(
                f"Filtered instances by node count: "
                f"min_visible_nodes={cfg.min_visible_nodes}, "
                f"min_visible_node_fraction={cfg.min_visible_node_fraction}"
            )
        if cfg.min_mean_node_score > 0.0 or cfg.min_instance_score > 0.0:
            logger.info(
                f"Filtered instances by confidence: "
                f"min_mean_node_score={cfg.min_mean_node_score}, "
                f"min_instance_score={cfg.min_instance_score}"
            )
        if cfg.overlapping:
            logger.info(
                f"Filtered overlapping instances with "
                f"{cfg.overlapping_method.upper()} threshold: "
                f"{cfg.overlapping_threshold}"
            )

    def _log_inference_summary(
        self,
        *,
        n_frames: int,
        elapsed_s: float,
        output: Optional[str] = None,
        n_objects: Optional[int] = None,
        object_label: str = "instances",
    ) -> None:
        """Log a one-line post-run summary (#610).

        ``n_objects`` (instances or masks) is optional — the streaming path
        drops per-frame objects, so it reports frames/throughput only.
        """
        fps = n_frames / elapsed_s if elapsed_s > 0 else 0.0
        parts = [f"frames={n_frames}"]
        if n_objects is not None:
            mean = n_objects / n_frames if n_frames > 0 else 0.0
            parts.append(f"{object_label}={n_objects} ({mean:.2f}/frame)")
        parts += [
            f"elapsed={elapsed_s:.1f}s",
            f"throughput={fps:.1f} fps",
            f"tracking={self.tracker_config is not None}",
        ]
        if output:
            parts.append(f"output={output}")
        logger.info("Inference complete | " + " | ".join(parts))

    def _make_provider(
        self,
        source: Any,
        frames: Optional[List[int]] = None,
        **provider_kwargs: Any,
    ) -> tuple["Provider", Optional[List["sio.Video"]]]:
        """Wrap a source into a ``Provider`` + extract videos for label packaging.

        Returns ``(provider, videos)`` where ``videos`` is a list of
        ``sio.Video`` when derivable from the source, else ``None``.
        """
        import sleap_io as sio

        from sleap_nn.inference.providers import (
            LabelsProvider,
            NumpyProvider,
            VideoProvider,
        )

        if isinstance(source, (np.ndarray, torch.Tensor)):
            # In-memory frame stack ``(N, H, W, C)`` — the batch-oriented analog
            # of the realtime ``layer.predict`` path. Route to ``NumpyProvider``;
            # a raw array is not a video file, so ``VideoProvider`` raised
            # "Unknown video file type", and a bare tensor would otherwise be
            # mistaken for a ``Provider`` by the ``__iter__`` branch below.
            # ``frames`` (if given) labels the per-frame indices on the output.
            provider = NumpyProvider(
                images=source,
                batch_size=self.batch_size,
                frame_indices=(
                    np.asarray(frames, dtype=np.int64) if frames is not None else None
                ),
                **provider_kwargs,
            )
            return provider, None

        if isinstance(source, str):
            if source.endswith(".slp"):
                # Load once so we can both build the provider AND attach the
                # real videos to the output Labels — legacy parity: predicted
                # frames must reference the source video, not be dropped
                # (#530 audit: .slp path returned videos=None).
                labels = sio.load_slp(source)
                provider_kwargs.setdefault(
                    "only_labeled_frames", self._needs_gt_instances()
                )
                provider = LabelsProvider(
                    labels=labels,
                    batch_size=self.batch_size,
                    frames=frames,
                    **provider_kwargs,
                )
                return provider, (list(labels.videos) if labels.videos else None)
            video = sio.Video(source)
            provider = VideoProvider(
                video=source,
                batch_size=self.batch_size,
                frames=frames,
                **provider_kwargs,
            )
            return provider, [video]

        if isinstance(source, sio.Video):
            provider = VideoProvider(
                video=source,
                batch_size=self.batch_size,
                frames=frames,
                **provider_kwargs,
            )
            return provider, [source]

        if isinstance(source, sio.Labels):
            provider_kwargs.setdefault(
                "only_labeled_frames", self._needs_gt_instances()
            )
            provider = LabelsProvider(
                labels=source,
                batch_size=self.batch_size,
                frames=frames,
                **provider_kwargs,
            )
            videos = list(source.videos) if source.videos else None
            return provider, videos

        if isinstance(source, (list, tuple)):
            # Multi-source input: predict([v1, v2, v3]) -> one merged Labels with
            # monotonically increasing per-frame video_indices (#582). Recurse to
            # reuse per-type dispatch, then concatenate. Per-source frame
            # selection is not expressible for a flat list, so `frames` is not
            # applied here (build providers explicitly if you need it).
            from sleap_nn.inference.providers import MultiVideoProvider

            if len(source) == 0:
                raise ValueError("predict() received an empty list of sources.")
            sub_providers: list = []
            all_videos: list = []
            video_offsets: list = []
            for sub in source:
                sub_provider, sub_videos = self._make_provider(sub, **provider_kwargs)
                sub_providers.append(sub_provider)
                # Each source starts at the current end of the merged video
                # list; its own (possibly multi-video) indices are offset by
                # this. Substitute a placeholder when a sub-source has no
                # derivable video so frames never reference a None video
                # (unserializable) — matches the writer's placeholder.
                video_offsets.append(len(all_videos))
                if sub_videos:
                    all_videos.extend(sub_videos)
                else:
                    all_videos.append(sio.Video(filename="unknown", backend=None))
            return (
                MultiVideoProvider(
                    providers=sub_providers, video_offsets=video_offsets
                ),
                all_videos,
            )

        if hasattr(source, "__iter__"):
            # A pre-built Provider. Recover its source videos (when it exposes
            # them) so predicted frames reference the real Video, matching the
            # str/sio.Video/sio.Labels branches above (#530). Without this the
            # output Labels carries a None-placeholder video and sio.Labels.save
            # crashes walking video.backend — hit whenever the CLI wraps the
            # input in a LabelsProvider for --only_suggested_frames /
            # --exclude_user_labeled / --video_index etc. (#699).
            provider_videos = getattr(source, "videos", None)
            return source, (list(provider_videos) if provider_videos else None)

        raise TypeError(
            f"Unsupported source type: {type(source).__name__}. "
            f"Pass an sio.Video, sio.Labels, file path string, or a Provider."
        )

    @staticmethod
    def retrack(
        labels: "sio.Labels",
        tracker_config: TrackerConfig,
        clean_empty_frames: bool = False,
        progress_callback: Optional[Callable[[int, int], None]] = None,
    ) -> "sio.Labels":
        """Retrack an existing ``sio.Labels`` without running inference.

        Pure tracking -- useful when you already have predicted instances
        in a ``.slp`` and just want to (re)apply a tracker. The tracker
        runs once over the full LabeledFrame list; post-tracking cleanup
        (cull / connect-single-breaks) is applied per ``tracker_config``.

        Args:
            labels: A ``sio.Labels`` whose ``predicted_instances`` are
                tracked in-place semantics — this returns a new
                ``Labels`` with tracked instances.
            tracker_config: :class:`TrackerConfig` to drive the tracker.
            clean_empty_frames: When ``True``, drop empty frames from
                the result (matches ``--no_empty_frames``).
            progress_callback: Optional ``(processed_frames, total_frames)``
                callback invoked after each frame is tracked.

        Returns:
            New ``sio.Labels`` with tracks attached.
        """
        out = apply_tracking(labels, tracker_config, progress_callback)
        if clean_empty_frames:
            out.clean(frames=True, skeletons=False)
        return out

    # ──────────────────────────────────────────────────────────────────
    # Synchronous: returns Outputs list or sio.Labels
    # ──────────────────────────────────────────────────────────────────

    def predict(
        self,
        source: Any,
        *,
        make_labels: bool = True,
        frames: Optional[List[int]] = None,
        skeleton: Optional["sio.Skeleton"] = None,
        videos: Optional[List["sio.Video"]] = None,
        clean_empty_frames: bool = False,
        progress_callback: Optional[Callable[[int, int], None]] = None,
        tracking_progress_callback: Optional[Callable[[int, int], None]] = None,
        peak_threshold: Optional[float] = None,
        centroid_threshold: Optional[float] = None,
        keypoint_threshold: Optional[float] = None,
        max_instances: Optional[int] = None,
        integral_refinement: Optional[str] = None,
        integral_patch_size: Optional[int] = None,
        return_confmaps: Optional[bool] = None,
        return_crops: Optional[bool] = None,
        return_pafs: Optional[bool] = None,
        return_paf_graph: Optional[bool] = None,
        return_class_maps: Optional[bool] = None,
        return_class_vectors: Optional[bool] = None,
    ) -> Union[List[Outputs], "sio.Labels"]:
        """Run inference on a source.

        Args:
            source: ``sio.Video``, ``sio.Labels``, video path string, or
                a pre-built :class:`Provider`. When a non-Provider source
                is given, a provider is auto-constructed using
                ``self.batch_size``.
            make_labels: When ``True`` (the default), return a
                ``sio.Labels``. Set to ``False`` for a raw
                ``List[Outputs]``.
            frames: Frame indices to predict on. Only used when ``source``
                is an ``sio.Video`` or video path.
            skeleton: ``sio.Skeleton`` for label conversion. Falls back to
                ``self.skeleton`` when ``None``.
            videos: Optional list of ``sio.Video`` indexed by
                ``video_indices`` for label conversion. Auto-derived from
                the source when possible.
            clean_empty_frames: When ``True`` and ``make_labels=True``,
                drop ``LabeledFrame``s with no instances from the
                returned ``sio.Labels``.
            progress_callback: Optional ``(processed_frames, total_frames)``
                callback invoked after each batch. Counts are in frames
                (batch-size-invariant); ``total_frames`` is ``-1`` when the
                provider can't report its length up front.
            tracking_progress_callback: Optional
                ``(processed_frames, total_frames)`` callback invoked
                after each frame during tracking.
            peak_threshold: Override peak threshold for all stages. For
                per-stage control on top-down models, use
                ``centroid_threshold`` / ``keypoint_threshold`` instead.
            centroid_threshold: Override peak threshold for the centroid
                stage only (top-down models).
            keypoint_threshold: Override peak threshold for the centered-
                instance stage only (top-down models).
            max_instances: Override max instances per frame.
            integral_refinement: ``"integral"`` or ``"none"``.
            integral_patch_size: Override integral refinement patch size.
            return_confmaps: Override whether to return confidence maps.
            return_crops: Override whether to return per-instance crops
                (top-down only).
            return_pafs: Override whether to return part-affinity fields
                (bottom-up).
            return_paf_graph: Override whether to return the PAF graph
                (bottom-up).
            return_class_maps: Override whether to return class maps
                (multi-class bottom-up).
            return_class_vectors: Override whether to return class vectors
                (multi-class top-down).

        Returns:
            ``sio.Labels`` (default) or ``List[Outputs]`` (when
            ``make_labels=False``).
        """
        from datetime import datetime

        # Tracking operates on sio.PredictedInstance objects; centroid emission
        # (emit_centroid != 'instance') puts predictions in LabeledFrame.centroids
        # as sio.PredictedCentroid, which the tracker would silently drop. Fail
        # fast rather than lose data.
        if self.tracker_config is not None and self.emit_centroid != "instance":
            raise ValueError(
                "Tracking is incompatible with emit_centroid="
                f"{self.emit_centroid!r}: the tracker operates on "
                "sio.PredictedInstance objects, but this mode emits "
                "sio.PredictedCentroid objects (in LabeledFrame.centroids) that "
                "would be dropped. Use emit_centroid='instance' (the default) "
                "for tracking."
            )

        # Segmentation emits sio.PredictedSegmentationMask objects to
        # LabeledFrame.masks; apply_tracking auto-detects the mask-only labels
        # and tracks them by pixel mask-IoU (#619), so tracking is supported.

        provider, auto_videos = self._make_provider(source, frames=frames)
        if videos is None:
            videos = auto_videos

        self._log_inference_start(source, provider, videos)
        self._log_filter_config()
        _prov_start = datetime.now()
        layer = self._scoped_postprocess_layer(
            peak_threshold=peak_threshold,
            centroid_threshold=centroid_threshold,
            keypoint_threshold=keypoint_threshold,
            max_instances=max_instances,
            integral_refinement=integral_refinement,
            integral_patch_size=integral_patch_size,
            return_confmaps=return_confmaps,
            return_crops=return_crops,
            return_pafs=return_pafs,
            return_paf_graph=return_paf_graph,
            return_class_maps=return_class_maps,
            return_class_vectors=return_class_vectors,
        )
        outputs_list = list(self._batch_iter(provider, progress_callback, layer=layer))
        _prov_end = datetime.now()

        if not make_labels:
            if self.tracker_config is not None:
                raise ValueError(
                    "tracker_config requires make_labels=True; the tracker "
                    "operates on sio.PredictedInstance objects."
                )
            return outputs_list
        if skeleton is not None:
            self.skeleton = skeleton
        # An `embedding` model emits appearance vectors, not poses or masks, so
        # nothing packages them into a `sio.Labels` -- `make_labels=True` returned
        # an EMPTY Labels and looked like a model that predicted nothing. The CLI
        # routes this to the .h5 writer; say so here too, since the Python API can
        # reach it directly.
        if self._is_embedding_layer():
            raise ValueError(
                "make_labels=True is not supported for an `embedding` (re-ID) "
                "model: it predicts appearance vectors, and this path has nothing "
                "to attach them to. Use "
                "`sleap_nn.inference.embedding.predict_embeddings_to_slp(...)` "
                "(or `sleap-nn predict --save_embeddings slp`), which attaches each "
                "vector to its source detection via `sio.Embedding`, or pass "
                "`make_labels=False` for the raw `Outputs`."
            )
        if self.skeleton is None and not self._is_segmentation_layer():
            raise ValueError(
                "make_labels=True requires a skeleton. Either pass "
                "`skeleton=...` or build the Predictor via Predictor.from_model_paths() "
                "which sets it automatically from the training config."
            )
        labels = self.to_labels(
            outputs_list,
            videos=videos,
            keep_empty_frames=True,
        )
        if self.tracker_config is not None:
            labels = apply_tracking(
                labels, self.tracker_config, tracking_progress_callback
            )
        if clean_empty_frames:
            labels.clean(frames=True, skeletons=False)
        labels.provenance = self._build_inference_provenance(
            source=source,
            start_time=_prov_start,
            end_time=_prov_end,
            n_frames=len(labels.labeled_frames),
            inference_params={
                "peak_threshold": peak_threshold,
                "centroid_threshold": centroid_threshold,
                "keypoint_threshold": keypoint_threshold,
                "max_instances": max_instances,
                "integral_refinement": integral_refinement,
                "integral_patch_size": integral_patch_size,
                "batch_size": self.batch_size,
                **self._preprocess_provenance_params(),
            },
        )

        # Post-run summary (#610). Segmentation emits masks (LabeledFrame.masks);
        # everything else emits instances (LabeledFrame.instances).
        n_lf = len(labels.labeled_frames)
        if self._is_segmentation_layer():
            n_objects = sum(
                len(getattr(lf, "masks", None) or []) for lf in labels.labeled_frames
            )
            object_label = "masks"
        else:
            n_objects = sum(len(lf.instances) for lf in labels.labeled_frames)
            object_label = "instances"
        self._log_inference_summary(
            n_frames=n_lf,
            elapsed_s=(_prov_end - _prov_start).total_seconds(),
            n_objects=n_objects,
            object_label=object_label,
        )
        return labels

    # ──────────────────────────────────────────────────────────────────
    # Streaming: yields one Outputs at a time
    # ──────────────────────────────────────────────────────────────────

    def predict_streaming(
        self,
        source: Any,
        *,
        frames: Optional[List[int]] = None,
        progress_callback: Optional[Callable[[int, int], None]] = None,
        peak_threshold: Optional[float] = None,
        centroid_threshold: Optional[float] = None,
        keypoint_threshold: Optional[float] = None,
        max_instances: Optional[int] = None,
        integral_refinement: Optional[str] = None,
        integral_patch_size: Optional[int] = None,
        return_confmaps: Optional[bool] = None,
        return_crops: Optional[bool] = None,
        return_pafs: Optional[bool] = None,
        return_paf_graph: Optional[bool] = None,
        return_class_maps: Optional[bool] = None,
        return_class_vectors: Optional[bool] = None,
    ) -> Iterator[Outputs]:
        """Yield one ``Outputs`` per batch from ``source``.

        Args:
            source: ``sio.Video``, ``sio.Labels``, video path string, or
                a pre-built :class:`Provider`.
            frames: Frame indices (only for video sources).
            progress_callback: Optional ``(processed_frames, total_frames)``
                callback.
            peak_threshold: Override peak threshold for all stages.
            centroid_threshold: Override centroid stage threshold (top-down).
            keypoint_threshold: Override centered-instance threshold (top-down).
            max_instances: Override max instances per frame.
            integral_refinement: ``"integral"`` or ``"none"``.
            integral_patch_size: Override integral refinement patch size.
            return_confmaps: Override whether to return confidence maps.
            return_crops: Override whether to return per-instance crops
                (top-down only).
            return_pafs: Override whether to return part-affinity fields
                (bottom-up).
            return_paf_graph: Override whether to return the PAF graph
                (bottom-up).
            return_class_maps: Override whether to return class maps
                (multi-class bottom-up).
            return_class_vectors: Override whether to return class vectors
                (multi-class top-down).
        """
        if self.tracker_config is not None:
            raise ValueError(
                "tracker_config is not supported on predict_streaming / "
                "predict_to_file. End-of-stream tracker cleanup needs the "
                "full LabeledFrame list; use predict() instead."
            )
        provider, _ = self._make_provider(source, frames=frames)
        layer = self._scoped_postprocess_layer(
            peak_threshold=peak_threshold,
            centroid_threshold=centroid_threshold,
            keypoint_threshold=keypoint_threshold,
            max_instances=max_instances,
            integral_refinement=integral_refinement,
            integral_patch_size=integral_patch_size,
            return_confmaps=return_confmaps,
            return_crops=return_crops,
            return_pafs=return_pafs,
            return_paf_graph=return_paf_graph,
            return_class_maps=return_class_maps,
            return_class_vectors=return_class_vectors,
        )
        if self.paf_workers > 0 and self._can_pipeline():
            yield from self._predict_streaming_pipelined(
                provider, progress_callback, layer=layer
            )
            return
        yield from self._batch_iter(provider, progress_callback, layer=layer)

    # ──────────────────────────────────────────────────────────────────
    # Disk-streaming: write to a .slp incrementally
    # ──────────────────────────────────────────────────────────────────

    def predict_to_file(
        self,
        source: Any,
        path: str,
        *,
        frames: Optional[List[int]] = None,
        skeleton: Optional["sio.Skeleton"] = None,
        videos: Optional[List["sio.Video"]] = None,
        write_interval: int = 500,
        progress_callback: Optional[Callable[[int, int], None]] = None,
    ) -> str:
        """Run inference and write results to a ``.slp`` file.

        Each batch's ``Outputs`` is slimmed and converted to LabeledFrames
        immediately, so the heavy intermediate tensors (confmaps, PAFs) are
        dropped right away. The slimmed LabeledFrames accumulate until the
        file is finalized at close (see :class:`IncrementalLabelsWriter` for
        the memory note).

        Args:
            source: ``sio.Video``, ``sio.Labels``, video path string, or
                a pre-built :class:`Provider`.
            path: Destination ``.slp`` path.
            frames: Frame indices (only for video sources).
            skeleton: ``sio.Skeleton`` for instance conversion. Falls back
                to ``self.skeleton`` when ``None``.
            videos: Optional list of ``sio.Video`` indexed by
                ``video_indices`` for the saved labels.
            write_interval: Number of LabeledFrames to buffer before
                a disk flush.
            progress_callback: Optional ``(processed_frames, total_frames)``
                callback invoked after each batch.

        Returns:
            The (resolved) destination path string.
        """
        if skeleton is not None:
            self.skeleton = skeleton
        # Same as `predict(make_labels=True)`: an embedding model has nothing to
        # write into a `.slp`, so this streamed an empty file.
        if self._is_embedding_layer():
            raise ValueError(
                "predict_to_file is not supported for an `embedding` (re-ID) "
                "model: it predicts appearance vectors, and this path has nothing "
                "to attach them to. Use "
                "`sleap_nn.inference.embedding.predict_embeddings_to_slp(...)` "
                "(or `sleap-nn predict --save_embeddings slp`) instead."
            )

        if self.skeleton is None and not self._is_segmentation_layer():
            raise ValueError(
                "predict_to_file requires a skeleton. Either pass "
                "`skeleton=...` or build the Predictor via Predictor.from_model_paths() "
                "which sets it automatically from the training config."
            )
        from datetime import datetime

        from sleap_nn.inference.writer import IncrementalLabelsWriter

        provider, derived = self._make_provider(source, frames=frames)
        # Preserve the real source video(s) on the streamed .slp instead of the
        # 'unknown' placeholder. Mirrors the in-memory predict() path; for a
        # pre-built Provider source `derived` is None so the caller-supplied
        # `videos` (possibly None) is used (#582).
        if videos is None:
            videos = derived
        self._log_inference_start(source, provider, derived)
        self._log_filter_config()
        pkg = self._resolve_centroid_packaging()
        # NOTE: this streaming writer does NOT apply multi-class packaging — neither
        # `tracks` (pre-existing) nor `identities` are threaded into the per-batch
        # `slim.to_labels`. The default `sleap-nn predict` flow uses the in-memory
        # `Predictor.predict`/`to_labels` path (run.py), which DOES emit both. A
        # multi-class model run through this streaming path therefore omits the
        # predicted Track/Identity packaging; route such models through `predict()`.
        writer = IncrementalLabelsWriter(
            path=path,
            skeleton=self.skeleton,
            videos=videos,
            write_interval=write_interval,
            anchor_ind=pkg.anchor_ind,
            collapse_skeleton=pkg.collapse_skeleton,
            emit_centroid=pkg.emit_centroid,
            source=pkg.source,
            mask_output=getattr(self.layer, "mask_output", "mask"),
            polygon_epsilon=getattr(self.layer, "polygon_epsilon", 0.01),
        )
        _prov_start = datetime.now()
        with writer:
            for outputs in self.predict_streaming(
                provider, progress_callback=progress_callback
            ):
                writer.write(outputs)
            # Attach provenance before the context exit finalizes the .slp, so
            # the streamed output carries lineage like the in-memory path (#583).
            _prov_end = datetime.now()
            writer.provenance = self._build_inference_provenance(
                source=source,
                start_time=_prov_start,
                end_time=_prov_end,
                n_frames=writer.frame_count,
                inference_params={
                    "batch_size": self.batch_size,
                    **self._preprocess_provenance_params(),
                },
            )
        # Post-run summary (#610). The streaming path drops per-frame objects to
        # keep memory O(window), so report frames / throughput only.
        self._log_inference_summary(
            n_frames=writer.frame_count,
            elapsed_s=(_prov_end - _prov_start).total_seconds(),
            output=path,
        )
        return path

    # ──────────────────────────────────────────────────────────────────
    # Internals
    # ──────────────────────────────────────────────────────────────────

    def _batch_iter(
        self,
        provider: Provider,
        progress_callback: Optional[Callable[[int, int], None]] = None,
        layer: Optional[Any] = None,
    ) -> Iterator[Outputs]:
        """Run ``layer.predict`` + ``FilterPipeline`` per provider batch.

        ``layer`` defaults to ``self.layer``; callers applying predict-time
        postprocess overrides pass the scoped copy from
        ``_scoped_postprocess_layer`` instead so ``self.layer`` is never
        mutated.
        """
        import inspect

        if layer is None:
            layer = self.layer

        try:
            sig = inspect.signature(layer.predict)
            layer_accepts_instances = "instances" in sig.parameters
        except (TypeError, ValueError):  # pragma: no cover — non-introspectable
            layer_accepts_instances = False

        pipeline = self.filter_pipeline
        total = _safe_num_frames(provider)
        frames_done = 0
        for batch in provider:
            kwargs: dict = {}
            if batch.instances is not None and layer_accepts_instances:
                kwargs["instances"] = (
                    batch.instances
                    if isinstance(batch.instances, torch.Tensor)
                    else torch.from_numpy(batch.instances)
                )
            outputs = layer.predict(batch.images, **kwargs)
            outputs = pipeline(outputs)
            outputs = self._stamp_metadata(outputs, batch)
            yield outputs
            if progress_callback is not None:
                frames_done += int(batch.images.shape[0])
                progress_callback(frames_done, total)
        if progress_callback is not None and total >= 0 and frames_done != total:
            # The provider can yield fewer frames than its upfront `total`
            # estimate (e.g. LabelsProvider skips a synthesized placeholder
            # whose pixels turn out unreadable) -- without this, a caller
            # gating "done" on `processed >= total` (the GUI's JSON progress
            # consumer) never sees a completion signal even though inference
            # succeeded.
            progress_callback(frames_done, frames_done)

    # ──────────────────────────────────────────────────────────────────
    # Pipelined bottom-up: GPU stage in main proc, CPU grouping in pool
    # ──────────────────────────────────────────────────────────────────

    def _can_pipeline(self) -> bool:
        """``True`` iff ``layer`` is a :class:`BottomUpLayer` (not multiclass)."""
        from sleap_nn.inference.layers.bottomup import BottomUpLayer

        return isinstance(self.layer, BottomUpLayer)

    def _predict_streaming_pipelined(
        self,
        provider: Provider,
        progress_callback: Optional[Callable[[int, int], None]] = None,
        layer: Optional[Any] = None,
    ) -> Iterator[Outputs]:
        """Stream ``Outputs`` with the CPU grouping stage in a worker pool.

        ``layer`` defaults to ``self.layer``; see ``_batch_iter`` for why a
        caller applying postprocess overrides passes a scoped copy instead.
        """
        from sleap_nn.inference.streaming import PafGroupingPool

        if layer is None:
            layer = self.layer

        pipeline = self.filter_pipeline
        params = layer.grouping_params()
        total = _safe_num_frames(provider)

        # Bound the number of in-flight batches so memory stays O(window),
        # not O(whole video). Submitting every batch before draining (the old
        # behavior) kept all ScoredBatch payloads + cached batches resident at
        # once (#583). Keep a small multiple of n_workers in flight so workers
        # stay fed; drain the OLDEST completed result (FIFO) before submitting
        # more, preserving submission-order output.
        max_in_flight = max(2 * self.paf_workers, self.paf_workers + 1)
        meta: dict[int, Any] = {}
        frames_done = 0
        with PafGroupingPool(
            n_workers=self.paf_workers, grouping_params=params
        ) as pool:
            for ordinal, batch in enumerate(provider):
                x, info = layer.preprocess(batch.images)
                raw = layer.backend(x)
                scored = layer._score_pafs_on_gpu(raw, info)
                pool.submit(ordinal, scored)
                meta[ordinal] = batch
                while len(pool) >= max_in_flight:
                    done_ordinal, outputs = pool.drain_one()
                    done_batch = meta.pop(done_ordinal)
                    outputs = pipeline(outputs)
                    outputs = self._stamp_metadata(outputs, done_batch)
                    yield outputs
                    if progress_callback is not None:
                        frames_done += int(done_batch.images.shape[0])
                        progress_callback(frames_done, total)
            # Drain the remaining in-flight batches.
            while True:
                result = pool.drain_one()
                if result is None:
                    break
                done_ordinal, outputs = result
                done_batch = meta.pop(done_ordinal)
                outputs = pipeline(outputs)
                outputs = self._stamp_metadata(outputs, done_batch)
                yield outputs
                if progress_callback is not None:
                    frames_done += int(done_batch.images.shape[0])
                    progress_callback(frames_done, total)
        if progress_callback is not None and total >= 0 and frames_done != total:
            # See the matching comment in `_batch_iter`: the provider's
            # upfront `total` can overcount frames that are later skipped
            # as unreadable, so force a final completion signal.
            progress_callback(frames_done, frames_done)

    @staticmethod
    def _stamp_metadata(outputs: Outputs, batch: Any) -> Outputs:
        """Attach ``frame_indices`` / ``video_indices`` from the batch."""
        kwargs: dict = {}
        if batch.frame_indices is not None and outputs.frame_indices is None:
            kwargs["frame_indices"] = (
                batch.frame_indices
                if isinstance(batch.frame_indices, torch.Tensor)
                else torch.from_numpy(np.asarray(batch.frame_indices))
            )
        if batch.video_indices is not None and outputs.video_indices is None:
            kwargs["video_indices"] = (
                batch.video_indices
                if isinstance(batch.video_indices, torch.Tensor)
                else torch.from_numpy(np.asarray(batch.video_indices))
            )
        if not kwargs:
            return outputs
        return attrs.evolve(outputs, **kwargs)

    def to_labels(
        self,
        outputs_list: List[Outputs],
        videos: Optional[List["sio.Video"]] = None,
        keep_empty_frames: bool = False,
    ) -> "sio.Labels":
        """Concatenate per-batch ``Outputs`` into a single ``sio.Labels``.

        Args:
            outputs_list: Per-batch ``Outputs`` to concatenate.
            videos: List of ``sio.Video`` indexed by ``video_indices``.
            keep_empty_frames: Forwarded to :meth:`Outputs.to_labels` -- keep
                zero-detection frames instead of dropping them.
                :meth:`predict` always passes ``True`` here (matching the
                legacy pipeline's default of keeping every processed frame,
                tracking or not -- #714 fixed this for the tracking case,
                #717 for the non-tracking default); ``clean_empty_frames``
                (``--no_empty_frames`` on the CLI) is the opt-in way to drop
                them afterward, applied uniformly regardless of tracking.
        """
        import sleap_io as sio

        skeleton = self.skeleton
        pkg = self._resolve_centroid_packaging()
        tracks = self._multiclass_tracks()
        identities = self._multiclass_identities()
        videos = list(videos) if videos else [None]
        # When a standalone centroid model collapses to a 1-node 'centroid'
        # skeleton, that is the skeleton attached to emitted instances and to
        # Labels.skeletons; otherwise the original training skeleton is kept.
        out_skeleton = (
            pkg.collapse_skeleton if pkg.collapse_skeleton is not None else skeleton
        )
        # Segmentation mask output representation (read off the layer; identity
        # defaults for non-segmentation layers).
        mask_output = getattr(self.layer, "mask_output", "mask")
        polygon_epsilon = getattr(self.layer, "polygon_epsilon", 0.01)
        all_lf: list = []
        used_tracks: list = []
        seen_track_ids: set = set()
        used_identities: list = []
        seen_identity_names: set = set()
        for outputs in outputs_list:
            sub = outputs.to_labels(
                skeleton=skeleton,
                videos=videos,
                anchor_ind=pkg.anchor_ind,
                tracks=tracks,
                identities=identities,
                collapse_skeleton=pkg.collapse_skeleton,
                emit_centroid=pkg.emit_centroid,
                source=pkg.source,
                mask_output=mask_output,
                polygon_epsilon=polygon_epsilon,
                keep_empty_frames=keep_empty_frames,
            )
            all_lf.extend(sub.labeled_frames)
            for trk in sub.tracks:
                if id(trk) not in seen_track_ids:
                    seen_track_ids.add(id(trk))
                    used_tracks.append(trk)
            # Dedup identities by name across batches (the simplified sio.Identity
            # matches by name; the same canonical objects are reused for every frame,
            # so this collapses to the registry).
            for ident in sub.identities:
                if ident.name not in seen_identity_names:
                    seen_identity_names.add(ident.name)
                    used_identities.append(ident)
        valid_videos = [v for v in videos if v is not None]
        labels = sio.Labels(
            labeled_frames=all_lf,
            videos=valid_videos,
            skeletons=[out_skeleton],
        )
        if used_tracks:
            labels.tracks = used_tracks
        if used_identities:
            labels.identities = used_identities
        return labels

    def _multiclass_tracks(self) -> Optional[list["sio.Track"]]:
        """Build the ``sio.Track`` registry for multi-class identity packaging.

        Reads ``class_names`` off the (possibly composed) multi-class layer —
        populated at build time from the training config — and constructs one
        ``sio.Track`` per class, ordered by class index. Returns ``None`` for
        non-multiclass layers. Matches legacy ``predictors.py`` track
        construction (TopDownMultiClass:3808-3811, BottomUpMultiClass:2966).
        """
        import sleap_io as sio

        class_names = getattr(self.layer, "class_names", None)
        if not class_names:
            return None
        return [sio.Track(name=str(name)) for name in class_names]

    def _multiclass_identities(self) -> Optional[list["sio.Identity"]]:
        """Build the canonical ``sio.Identity`` registry for multi-class models.

        Sibling of :meth:`_multiclass_tracks`. Reads ``class_names`` off the (possibly
        composed) multi-class layer and builds one ``sio.Identity(name=<class name>)``
        per class, ordered by class index. The simplified sleap-io ``Identity`` (name +
        metadata, sleap-io #535) matches by NAME across files and retrains, so the class
        name is itself the canonical cross-file identity key — no per-class uuid bridge.

        The identities are built **once** here and the same objects are reused for every
        frame/instance (the canonical-reference contract — ``Identity`` compares by
        object identity, so the writer's ``identity in labels.identities`` registration
        check only passes for the exact objects we register; cross-file joins use
        ``Identity.matches`` on the name). Returns ``None`` for non-multiclass layers.
        """
        import sleap_io as sio

        class_names = getattr(self.layer, "class_names", None)
        if not class_names:
            return None
        # The classes map to a global Identity only when the model declares them
        # as unique individuals (``class_output == "identity"``). A ``"track"``
        # model (the default) emits only the per-video Track — no Identity is
        # fabricated. ``"category"`` (shared types/roles) is not implemented here;
        # the head configs' validator rejects it at config load, so this only fires
        # for a hand-written ``training_config.yaml`` that bypassed them.
        class_output = getattr(self.layer, "class_output", "track")
        if class_output not in ("track", "identity"):
            raise NotImplementedError(
                f"class_output={class_output!r} is not supported; predicted classes "
                "can be emitted as 'track' (default) or 'identity'. Set the "
                "multi-class head's class_output to one of those. (sleap-io does "
                "have a Category data model, but mapping classes onto it is not "
                "implemented.)"
            )
        if class_output != "identity":
            return None
        return [sio.Identity(name=str(name)) for name in class_names]

    def _packaging_anchor_ind(self) -> Optional[int]:
        """Anchor-node slot for centroid-only output packaging."""
        from sleap_nn.inference.layers.exported import ExportedCentroidLayer

        if isinstance(self.layer, (CentroidLayer, ExportedCentroidLayer)):
            return self.layer.anchor_ind
        return None

    def _packaging_centroid_method(self) -> Optional[str]:
        """Resolved centroid method for the ``sio.Centroid.source`` tag (#586)."""
        from sleap_nn.inference.layers.exported import ExportedCentroidLayer

        if isinstance(self.layer, (CentroidLayer, ExportedCentroidLayer)):
            return getattr(self.layer, "centroid_method", None)
        return None

    def _is_centroid_only_layer(self) -> bool:
        """``True`` iff ``layer`` is a standalone centroid layer."""
        from sleap_nn.inference.layers.exported import ExportedCentroidLayer

        return isinstance(self.layer, (CentroidLayer, ExportedCentroidLayer))

    def _is_segmentation_layer(self) -> bool:
        """``True`` iff ``layer`` is a mask-producing segmentation layer.

        Covers bottom-up (:class:`SegmentationLayer`), top-down
        (:class:`TopDownSegmentationLayer`), whole-frame semantic
        (:class:`SemanticSegmentationLayer`, a ``SegmentationLayer`` subclass),
        and their tiled wrappers (:class:`TiledSegmentationLayer` /
        :class:`TiledSemanticSegmentationLayer`, which are NOT ``InferenceLayer``
        subclasses and would otherwise miss this gate — matters for a genuinely
        skeleton-less semantic mask model under tiling). Gates
        tracking/no-skeleton/mask-count behavior — every one of these emits
        ``pred_masks``.
        """
        return isinstance(
            self.layer,
            (
                SegmentationLayer,
                TopDownSegmentationLayer,
                SemanticSegmentationLayer,
                TiledSegmentationLayer,
                TiledSemanticSegmentationLayer,
            ),
        )

    def _is_embedding_layer(self) -> bool:
        """``True`` iff ``layer`` is an appearance-embedding (re-ID) layer.

        Embedding models are skeleton-less (like segmentation): they emit
        ``Outputs.pred_embeddings`` rather than keypoints/masks. Gates the
        no-skeleton path so a bare ``predict()`` on an embedding model does not
        raise the "requires a skeleton" error (the re-ID path goes
        through :func:`sleap_nn.inference.embedding.predict_embeddings_to_slp`).
        """
        return isinstance(self.layer, (EmbeddingLayer, TopDownEmbeddingLayer))

    def _resolve_centroid_packaging(self) -> _CentroidPackaging:
        """Resolve the single-source centroid output-packaging decision.

        For a centroid-only layer trained on a MULTI-node skeleton, the output
        collapses to a 1-node 'centroid' skeleton (``sio.get_centroid_skeleton()``);
        a genuinely 1-node model is left as-is (``collapse_skeleton=None``). The
        ``source`` tag mirrors the #586 anchor-fallback semantics. For
        non-centroid layers, emission is forced to ``"instance"`` and no
        collapse occurs.
        """
        from sleap_nn.inference.centroid_convert import centroid_source_for_anchor

        if not self._is_centroid_only_layer():
            return _CentroidPackaging(
                collapse_skeleton=None,
                anchor_ind=None,
                emit_centroid="instance",
                source=centroid_source_for_anchor(None),
            )
        anchor_ind = self._packaging_anchor_ind()
        node_names = (
            list(self.skeleton.node_names) if self.skeleton is not None else None
        )
        source = centroid_source_for_anchor(
            anchor_ind, node_names, self._packaging_centroid_method()
        )
        collapse_skeleton = None
        if self.skeleton is not None and len(self.skeleton.nodes) > 1:
            import sleap_io as sio

            collapse_skeleton = sio.get_centroid_skeleton()
        return _CentroidPackaging(
            collapse_skeleton=collapse_skeleton,
            anchor_ind=anchor_ind,
            emit_centroid=self.emit_centroid,
            source=source,
        )

    # ──────────────────────────────────────────────────────────────────
    # Prediction-time postprocess overrides
    # ──────────────────────────────────────────────────────────────────

    @staticmethod
    def _collect_postprocess_targets(layer: Any) -> list:
        """Return all sub-layers that own a ``postprocess_config``.

        ``Tiled*`` wrappers (``TiledLayer``/``TiledSegmentationLayer``/
        ``TiledSemanticSegmentationLayer``) hold their wrapped layer's
        ``postprocess_config`` on ``.inner``, not on themselves -- unwrap so
        callers see the real owner instead of concluding (via the `hasattr`
        check below) that there's nothing to override (#712 follow-up).
        """
        from sleap_nn.inference.layers.topdown import TopDownLayer

        if isinstance(layer, TopDownLayer):
            targets = [layer.centroid_layer, layer.centered_instance_layer]
        elif hasattr(layer, "inner"):
            targets = [layer.inner]
        elif hasattr(layer, "postprocess_config"):
            targets = [layer]
        else:
            targets = []
        return targets

    def _scoped_postprocess_layer(
        self,
        peak_threshold: Optional[float] = None,
        centroid_threshold: Optional[float] = None,
        keypoint_threshold: Optional[float] = None,
        max_instances: Optional[int] = None,
        integral_refinement: Optional[str] = None,
        integral_patch_size: Optional[int] = None,
        return_confmaps: Optional[bool] = None,
        return_crops: Optional[bool] = None,
        return_pafs: Optional[bool] = None,
        return_paf_graph: Optional[bool] = None,
        return_class_maps: Optional[bool] = None,
        return_class_vectors: Optional[bool] = None,
    ) -> Any:
        """Return the layer to use for one predict call, with overrides applied.

        Returns ``self.layer`` unchanged when no override is requested (the
        common case). Otherwise returns a *private*, shallow-copied layer
        (and shallow-copied sub-layers, for composed layers like
        :class:`TopDownLayer`) with the overrides baked into the copies'
        ``postprocess_config`` -- ``self.layer`` and its real sub-layers are
        never mutated.

        This matters because ``predict_streaming()`` is a generator: an
        earlier version of this mutated ``self.layer.postprocess_config`` in
        place and restored it when the generator was exhausted/closed. Two
        ``predict_streaming()`` calls on the same ``Predictor`` interleaved
        via alternating ``next()`` shared that same mutable state, so each
        call's overrides could clobber the other's mid-stream. Returning an
        independent copy per call makes concurrent/interleaved calls fully
        isolated from each other, with nothing to restore.

        For top-down layers, ``centroid_threshold`` applies to the centroid
        stage and ``keypoint_threshold`` to the centered-instance stage.
        ``peak_threshold`` sets both when the per-stage kwargs aren't given.
        """
        from sleap_nn.inference.layers.topdown import TopDownLayer

        has_any = any(
            v is not None
            for v in (
                peak_threshold,
                centroid_threshold,
                keypoint_threshold,
                max_instances,
                integral_refinement,
                integral_patch_size,
                return_confmaps,
                return_crops,
                return_pafs,
                return_paf_graph,
                return_class_maps,
                return_class_vectors,
            )
        )
        if not has_any:
            return self.layer

        is_topdown = isinstance(self.layer, TopDownLayer)

        def _copy_with_overrides(target: Any) -> Any:
            old_cfg = target.postprocess_config
            overrides: dict = {}

            # Threshold routing for top-down. Use explicit None checks so an
            # explicit 0.0 override ("accept all peaks") is honored rather
            # than swallowed by a falsy `or` (#584).
            if is_topdown:
                is_centroid = target is self.layer.centroid_layer
                if is_centroid:
                    t = (
                        centroid_threshold
                        if centroid_threshold is not None
                        else peak_threshold
                    )
                else:
                    t = (
                        keypoint_threshold
                        if keypoint_threshold is not None
                        else peak_threshold
                    )
            else:
                t = peak_threshold

            if t is not None:
                overrides["peak_threshold"] = t
            if max_instances is not None and hasattr(old_cfg, "max_instances"):
                overrides["max_instances"] = max_instances
            if integral_refinement is not None:
                overrides["refinement"] = integral_refinement
            if integral_patch_size is not None:
                overrides["integral_patch_size"] = integral_patch_size
            if return_confmaps is not None:
                overrides["return_confmaps"] = return_confmaps
            # The remaining intermediate-tensor toggles all live on
            # PostprocessConfig; guard with hasattr defensively (#583).
            for _name, _val in (
                ("return_pafs", return_pafs),
                ("return_paf_graph", return_paf_graph),
                ("return_class_maps", return_class_maps),
                ("return_class_vectors", return_class_vectors),
            ):
                if _val is not None and hasattr(old_cfg, _name):
                    overrides[_name] = _val

            new_target = copy.copy(target)
            if overrides:
                new_target.postprocess_config = attrs.evolve(old_cfg, **overrides)
            return new_target

        if is_topdown:
            new_layer = copy.copy(self.layer)
            new_layer.centroid_layer = _copy_with_overrides(self.layer.centroid_layer)
            new_layer.centered_instance_layer = _copy_with_overrides(
                self.layer.centered_instance_layer
            )
            # return_crops lives on TopDownLayer, not on postprocess_config.
            if return_crops is not None:
                new_layer.return_crops = return_crops
            return new_layer

        targets = self._collect_postprocess_targets(self.layer)
        if not targets:
            return self.layer
        # Non-top-down layers with a postprocess_config always have exactly
        # one target: the layer itself, or (for a Tiled* wrapper) its .inner.
        new_target = _copy_with_overrides(targets[0])
        if hasattr(self.layer, "inner"):
            # Rewrap: the caller needs a layer that still tiles, not the bare
            # overridden inner layer on its own.
            new_layer = copy.copy(self.layer)
            new_layer.inner = new_target
            return new_layer
        return new_target

filter_pipeline property

Build a fresh FilterPipeline from the config (cheap).

__attrs_post_init__()

Warn (once) if paf_workers was set on a layer that can't use it.

paf_workers > 0 only pipelines the CPU-bound grouping stage for plain bottom-up (:class:BottomUpLayer); every other layer type (including :class:BottomUpMultiClassLayer, which has its own CPU-bound Hungarian-matching identity step) silently ignores the setting today. Without this, that's an easy-to-miss no-op — a user expecting a speedup gets none, with no signal why.

Source code in sleap_nn/inference/predictor.py

def __attrs_post_init__(self) -> None:
    """Warn (once) if ``paf_workers`` was set on a layer that can't use it.

    ``paf_workers > 0`` only pipelines the CPU-bound grouping stage for
    plain bottom-up (:class:`BottomUpLayer`); every other layer type
    (including :class:`BottomUpMultiClassLayer`, which has its own
    CPU-bound Hungarian-matching identity step) silently ignores the
    setting today. Without this, that's an easy-to-miss no-op — a user
    expecting a speedup gets none, with no signal why.
    """
    if self.paf_workers > 0 and not self._can_pipeline():
        logger.warning(
            f"paf_workers={self.paf_workers} was set, but pipelined CPU "
            f"grouping is only implemented for plain bottom-up models. "
            f"layer={type(self.layer).__name__} will run the inline "
            "(unpipelined) path; paf_workers has no effect here."
        )

from_export_dir(export_dir, *, runtime='auto', device='auto', batch_size=4, return_confmaps=False, filter_config=None, paf_workers=0, tracker_config=None, max_instances=None, min_instance_peaks=0, min_line_scores=0.25, peak_conf_threshold=None, emit_centroid='instance') classmethod

Build a :class:Predictor from an exported ONNX/TensorRT directory.

Parameters:

Name	Type	Description	Default
export_dir	Union[str, Any]	Directory containing export_metadata.json + model.onnx or model.trt.	required
runtime	str	"auto" (prefer TRT), "onnx", or "tensorrt".	'auto'
device	str	Device string.	'auto'
batch_size	int	Default batch size.	4
return_confmaps	bool	Return confidence maps on Outputs.	False
filter_config	Optional['FilterConfig']	Post-inference :class:FilterConfig.	None
paf_workers	int	CPU workers for bottom-up PAF grouping.	0
tracker_config	Optional['TrackerConfig']	:class:TrackerConfig for tracking.	None
max_instances	Optional[int]	Cap on instances per frame (bottom-up).	None
min_instance_peaks	float	Min peaks for a valid instance (bottom-up).	0
min_line_scores	float	Per-edge match threshold (bottom-up).	0.25
peak_conf_threshold	Optional[float]	Runtime peak-confidence threshold for the exported bottom-up path. Gates PAF candidate connections by the src/dst peak confidence (legacy parity). Defaults to the threshold baked at export time (metadata.peak_threshold). Note the wrapper already bakes a peak threshold during peak finding, so this can only tighten beyond the baked value.	None
emit_centroid	str	Centroid-only output representation for an exported standalone centroid model: "instance" (default; single-node PredictedInstance, frontend-compatible), "centroid" (sio.PredictedCentroid), or "both". Honored only for ExportedCentroidLayer; mirrors from_model_paths so the exported runtime matches the checkpoint path.	'instance'

Source code in sleap_nn/inference/predictor.py

@classmethod
def from_export_dir(
    cls,
    export_dir: Union[str, Any],
    *,
    runtime: str = "auto",
    device: str = "auto",
    batch_size: int = 4,
    return_confmaps: bool = False,
    filter_config: Optional["FilterConfig"] = None,
    paf_workers: int = 0,
    tracker_config: Optional["TrackerConfig"] = None,
    max_instances: Optional[int] = None,
    min_instance_peaks: float = 0,
    min_line_scores: float = 0.25,
    peak_conf_threshold: Optional[float] = None,
    emit_centroid: str = "instance",
) -> "Predictor":
    """Build a :class:`Predictor` from an exported ONNX/TensorRT directory.

    Args:
        export_dir: Directory containing ``export_metadata.json`` +
            ``model.onnx`` or ``model.trt``.
        runtime: ``"auto"`` (prefer TRT), ``"onnx"``, or ``"tensorrt"``.
        device: Device string.
        batch_size: Default batch size.
        return_confmaps: Return confidence maps on Outputs.
        filter_config: Post-inference :class:`FilterConfig`.
        paf_workers: CPU workers for bottom-up PAF grouping.
        tracker_config: :class:`TrackerConfig` for tracking.
        max_instances: Cap on instances per frame (bottom-up).
        min_instance_peaks: Min peaks for a valid instance (bottom-up).
        min_line_scores: Per-edge match threshold (bottom-up).
        peak_conf_threshold: Runtime peak-confidence threshold for the
            exported bottom-up path. Gates PAF candidate connections by the
            src/dst peak confidence (legacy parity). Defaults to the
            threshold baked at export time (``metadata.peak_threshold``).
            Note the wrapper already bakes a peak threshold during peak
            finding, so this can only *tighten* beyond the baked value.
        emit_centroid: Centroid-only output representation for an exported
            standalone centroid model: ``"instance"`` (default; single-node
            ``PredictedInstance``, frontend-compatible), ``"centroid"``
            (``sio.PredictedCentroid``), or ``"both"``. Honored only for
            ``ExportedCentroidLayer``; mirrors ``from_model_paths`` so the
            exported runtime matches the checkpoint path.
    """
    from sleap_nn.export.metadata import ExportMetadata

    export_dir = Path(export_dir)

    metadata_path = export_dir / "export_metadata.json"
    if not metadata_path.exists():
        raise FileNotFoundError(
            f"export_metadata.json not found at {metadata_path}. "
            f"Pass a directory written by `sleap_nn export`."
        )
    metadata = ExportMetadata.load(metadata_path)

    runtime, model_path = _resolve_export_runtime(export_dir, runtime)
    backend = _build_export_backend(runtime, model_path, device)

    # Default the runtime peak-confidence threshold to the value baked at
    # export time, falling back to legacy's 0.2 when the metadata carries no
    # baked threshold (matches legacy export inference, #582).
    if peak_conf_threshold is not None:
        resolved_peak_conf = peak_conf_threshold
    else:
        meta_thr = getattr(metadata, "peak_threshold", None)
        resolved_peak_conf = meta_thr if meta_thr is not None else 0.2
    layer = _select_export_layer(
        metadata=metadata,
        backend=backend,
        return_confmaps=return_confmaps,
        max_instances=max_instances,
        min_instance_peaks=min_instance_peaks,
        min_line_scores=min_line_scores,
        peak_conf_threshold=resolved_peak_conf,
    )

    skeleton = _skeleton_from_export(export_dir, metadata)
    kwargs: dict = {
        "layer": layer,
        "skeleton": skeleton,
        "batch_size": batch_size,
        "paf_workers": paf_workers,
        "model_paths": [str(export_dir)],
        "device": device,
        "emit_centroid": emit_centroid,
    }
    if filter_config is not None:
        kwargs["filter_config"] = filter_config
    if tracker_config is not None:
        kwargs["tracker_config"] = tracker_config
    return cls(**kwargs)

from_model_paths(model_paths, *, device='cpu', batch_size=4, backbone_ckpt_path=None, head_ckpt_path=None, peak_threshold=0.2, integral_refinement='integral', integral_patch_size=5, max_instances=None, return_confmaps=False, preprocess_config=None, anchor_part=None, filter_config=None, paf_workers=0, tracker_config=None, centroid_only=False, emit_centroid='instance', max_edge_length_ratio=0.25, dist_penalty_weight=1.0, n_points=10, min_instance_peaks=0, min_line_scores=0.25, fg_threshold=0.5, min_mask_area=0, center_nms_kernel=3, mask_cleanup=False, mask_cleanup_radius=0, distance_gate_alpha=None, merge_fragments=False, merge_method='greedy', merge_thresholds=(0.85, 0.6, 0.4), merge_w_valley=1.0, merge_w_offset=0.25, merge_dilate=1, full_res_masks=False, mask_output='mask', polygon_epsilon=0.01) classmethod

Build a :class:Predictor from one or more checkpoint paths.

Parameters:

Name	Type	Description	Default
model_paths	List[str]	Trained model directories containing training_config.{yaml,json} + best.ckpt. Each entry may alternatively be a path to that best.ckpt or training_config.{yaml,json} file; all forms resolve to the model directory and load best.ckpt (#575). For top-down, pass two paths (centroid + centered-instance) in either order.	required
device	str	"cpu", "cuda", "mps", or "cuda:N".	'cpu'
batch_size	int	Default batch size for auto-constructed providers.	4
backbone_ckpt_path	Optional[str]	Override backbone weights with this .ckpt.	None
head_ckpt_path	Optional[str]	Override head weights.	None
peak_threshold	Union[float, List[float]]	Default peak threshold. List[float] for top-down ([centroid_thresh, keypoint_thresh]). Can be overridden per-call via predict(peak_threshold=...).	0.2
integral_refinement	str	"integral" or "none".	'integral'
integral_patch_size	int	Refinement patch size.	5
max_instances	Optional[int]	Cap on instances per frame.	None
return_confmaps	bool	Return confidence maps on Outputs.	False
preprocess_config	Optional[Any]	OmegaConf overrides for preprocessing.	None
anchor_part	Optional[str]	Override centroid anchor node name.	None
filter_config	Optional['FilterConfig']	Post-inference :class:FilterConfig.	None
paf_workers	int	CPU workers for bottom-up PAF grouping.	0
tracker_config	Optional['TrackerConfig']	:class:TrackerConfig for tracking.	None
centroid_only	bool	Force centroid-only output even when a centered-instance model is among model_paths.	False
emit_centroid	str	Centroid-only output representation: "instance" (default; single-node PredictedInstance), "centroid" (sio.PredictedCentroid), or "both". Honored only for centroid-only layers.	'instance'
max_edge_length_ratio	float	Bottom-up PAF max edge length ratio.	0.25
dist_penalty_weight	float	Bottom-up PAF distance penalty weight.	1.0
n_points	int	Bottom-up PAF line integration sample count.	10
min_instance_peaks	Union[int, float]	Bottom-up min peaks for a valid instance.	0
min_line_scores	float	Bottom-up per-edge match threshold. (These five are applied only to plain bottom-up models.)	0.25
fg_threshold	float	Foreground probability threshold for binarizing the segmentation map (bottom-up segmentation only).	0.5
min_mask_area	int	Minimum predicted-mask area in original-image pixels; smaller masks are dropped to suppress over-segmentation. 0 disables it (bottom-up segmentation only).	0
center_nms_kernel	int	Odd window size for center-peak NMS; larger merges nearby duplicate centers (bottom-up segmentation only).	3
mask_cleanup	bool	Keep-largest-CC + hole-fill per mask (bottom-up segmentation only).	False
mask_cleanup_radius	int	Morphological open->close radius (output-stride pixels) applied during mask_cleanup; 0 keeps keep-largest + fill only (bottom-up segmentation only).	0
distance_gate_alpha	Optional[float]	Adaptive distance-gate strength; None (default) keeps the byte-for-byte argmin grouping (bottom-up segmentation only).	None
merge_fragments	bool	Enable the RAG fragment-merge to re-fuse over-segmented animal halves; False (default) is byte-for-byte today (bottom-up segmentation only).	False
merge_method	str	"greedy" (default) or "multicut" agglomeration; inert when merge_fragments=False (bottom-up segmentation only).	'greedy'
merge_thresholds	tuple	Greedy-merge decreasing affinity thresholds; inert when merge_fragments=False (bottom-up segmentation only).	(0.85, 0.6, 0.4)
merge_w_valley	float	Center-valley merge-term weight; inert when off (bottom-up segmentation only).	1.0
merge_w_offset	float	Offset-agreement merge-term weight; inert when off (bottom-up segmentation only).	0.25
merge_dilate	int	Merge contact-test dilation iterations; inert when off (bottom-up segmentation only).	1
full_res_masks	bool	Encode masks at full original resolution instead of the model output-stride grid (default False: stride encoding is ~stride^2 smaller and lossless at model resolution; bottom-up segmentation only).	False
mask_output	str	Mask output representation: "mask" (default), "polygon" (Douglas-Peucker sio.PredictedROI only), or "both" (bottom-up segmentation only).	'mask'
polygon_epsilon	float	Douglas-Peucker tolerance (fraction of perimeter) for mask_output polygon/both (bottom-up segmentation only).	0.01

Source code in sleap_nn/inference/predictor.py

@classmethod
def from_model_paths(
    cls,
    model_paths: List[str],
    *,
    device: str = "cpu",
    batch_size: int = 4,
    backbone_ckpt_path: Optional[str] = None,
    head_ckpt_path: Optional[str] = None,
    peak_threshold: Union[float, List[float]] = 0.2,
    integral_refinement: str = "integral",
    integral_patch_size: int = 5,
    max_instances: Optional[int] = None,
    return_confmaps: bool = False,
    preprocess_config: Optional[Any] = None,
    anchor_part: Optional[str] = None,
    filter_config: Optional["FilterConfig"] = None,
    paf_workers: int = 0,
    tracker_config: Optional["TrackerConfig"] = None,
    centroid_only: bool = False,
    emit_centroid: str = "instance",
    max_edge_length_ratio: float = 0.25,
    dist_penalty_weight: float = 1.0,
    n_points: int = 10,
    min_instance_peaks: Union[int, float] = 0,
    min_line_scores: float = 0.25,
    fg_threshold: float = 0.5,
    min_mask_area: int = 0,
    center_nms_kernel: int = 3,
    mask_cleanup: bool = False,
    mask_cleanup_radius: int = 0,
    distance_gate_alpha: Optional[float] = None,
    merge_fragments: bool = False,
    merge_method: str = "greedy",
    merge_thresholds: tuple = (0.85, 0.6, 0.4),
    merge_w_valley: float = 1.0,
    merge_w_offset: float = 0.25,
    merge_dilate: int = 1,
    full_res_masks: bool = False,
    mask_output: str = "mask",
    polygon_epsilon: float = 0.01,
) -> "Predictor":
    """Build a :class:`Predictor` from one or more checkpoint paths.

    Args:
        model_paths: Trained model directories containing
            ``training_config.{yaml,json}`` + ``best.ckpt``. Each entry may
            alternatively be a path to that ``best.ckpt`` or
            ``training_config.{yaml,json}`` file; all forms resolve to the
            model directory and load ``best.ckpt`` (#575). For top-down, pass
            two paths (centroid + centered-instance) in either order.
        device: ``"cpu"``, ``"cuda"``, ``"mps"``, or ``"cuda:N"``.
        batch_size: Default batch size for auto-constructed providers.
        backbone_ckpt_path: Override backbone weights with this ``.ckpt``.
        head_ckpt_path: Override head weights.
        peak_threshold: Default peak threshold. ``List[float]`` for
            top-down (``[centroid_thresh, keypoint_thresh]``). Can be
            overridden per-call via ``predict(peak_threshold=...)``.
        integral_refinement: ``"integral"`` or ``"none"``.
        integral_patch_size: Refinement patch size.
        max_instances: Cap on instances per frame.
        return_confmaps: Return confidence maps on Outputs.
        preprocess_config: OmegaConf overrides for preprocessing.
        anchor_part: Override centroid anchor node name.
        filter_config: Post-inference :class:`FilterConfig`.
        paf_workers: CPU workers for bottom-up PAF grouping.
        tracker_config: :class:`TrackerConfig` for tracking.
        centroid_only: Force centroid-only output even when a
            centered-instance model is among ``model_paths``.
        emit_centroid: Centroid-only output representation: ``"instance"``
            (default; single-node ``PredictedInstance``), ``"centroid"``
            (``sio.PredictedCentroid``), or ``"both"``. Honored only for
            centroid-only layers.
        max_edge_length_ratio: Bottom-up PAF max edge length ratio.
        dist_penalty_weight: Bottom-up PAF distance penalty weight.
        n_points: Bottom-up PAF line integration sample count.
        min_instance_peaks: Bottom-up min peaks for a valid instance.
        min_line_scores: Bottom-up per-edge match threshold. (These five
            are applied only to plain bottom-up models.)
        fg_threshold: Foreground probability threshold for binarizing the
            segmentation map (bottom-up segmentation only).
        min_mask_area: Minimum predicted-mask area in original-image pixels;
            smaller masks are dropped to suppress over-segmentation. ``0``
            disables it (bottom-up segmentation only).
        center_nms_kernel: Odd window size for center-peak NMS; larger merges
            nearby duplicate centers (bottom-up segmentation only).
        mask_cleanup: Keep-largest-CC + hole-fill per mask (bottom-up
            segmentation only).
        mask_cleanup_radius: Morphological open->close radius (output-stride
            pixels) applied during ``mask_cleanup``; ``0`` keeps keep-largest
            + fill only (bottom-up segmentation only).
        distance_gate_alpha: Adaptive distance-gate strength; ``None``
            (default) keeps the byte-for-byte argmin grouping (bottom-up
            segmentation only).
        merge_fragments: Enable the RAG fragment-merge to re-fuse
            over-segmented animal halves; ``False`` (default) is byte-for-byte
            today (bottom-up segmentation only).
        merge_method: ``"greedy"`` (default) or ``"multicut"`` agglomeration;
            inert when ``merge_fragments=False`` (bottom-up segmentation only).
        merge_thresholds: Greedy-merge decreasing affinity thresholds; inert
            when ``merge_fragments=False`` (bottom-up segmentation only).
        merge_w_valley: Center-valley merge-term weight; inert when off
            (bottom-up segmentation only).
        merge_w_offset: Offset-agreement merge-term weight; inert when off
            (bottom-up segmentation only).
        merge_dilate: Merge contact-test dilation iterations; inert when off
            (bottom-up segmentation only).
        full_res_masks: Encode masks at full original resolution instead of
            the model output-stride grid (default ``False``: stride encoding
            is ~stride^2 smaller and lossless at model resolution; bottom-up
            segmentation only).
        mask_output: Mask output representation: ``"mask"`` (default),
            ``"polygon"`` (Douglas-Peucker ``sio.PredictedROI`` only), or
            ``"both"`` (bottom-up segmentation only).
        polygon_epsilon: Douglas-Peucker tolerance (fraction of perimeter)
            for ``mask_output`` polygon/both (bottom-up segmentation only).
    """
    from sleap_nn.inference.loaders import load_model_assets
    from sleap_nn.system_info import get_startup_info_string

    logger.info(get_startup_info_string())

    loaded, model_types = load_model_assets(
        model_paths,
        device=device,
        backbone_ckpt_path=backbone_ckpt_path,
        head_ckpt_path=head_ckpt_path,
        peak_threshold=peak_threshold,
        integral_refinement=integral_refinement,
        integral_patch_size=integral_patch_size,
        max_instances=max_instances,
        return_confmaps=return_confmaps,
        preprocess_config=preprocess_config,
        anchor_part=anchor_part,
        max_edge_length_ratio=max_edge_length_ratio,
        dist_penalty_weight=dist_penalty_weight,
        n_points=n_points,
        min_instance_peaks=min_instance_peaks,
        min_line_scores=min_line_scores,
        fg_threshold=fg_threshold,
        min_mask_area=min_mask_area,
        center_nms_kernel=center_nms_kernel,
        mask_cleanup=mask_cleanup,
        mask_cleanup_radius=mask_cleanup_radius,
        distance_gate_alpha=distance_gate_alpha,
        merge_fragments=merge_fragments,
        merge_method=merge_method,
        merge_thresholds=merge_thresholds,
        merge_w_valley=merge_w_valley,
        merge_w_offset=merge_w_offset,
        merge_dilate=merge_dilate,
        full_res_masks=full_res_masks,
        mask_output=mask_output,
        polygon_epsilon=polygon_epsilon,
    )

    if centroid_only:
        if "centroid" not in model_types:
            raise ValueError(
                "centroid_only=True requires a centroid model in model_paths; "
                f"detected types: {model_types}."
            )
        layer = _build_centroid_layer(
            loaded.inference_model.centroid_crop,
            device,
            assets=loaded,
        )
    else:
        layer = _select_layer(loaded, model_types, device)

    skeleton = loaded.skeletons[0] if loaded.skeletons else None
    kwargs: dict = {
        "layer": layer,
        "skeleton": skeleton,
        "batch_size": batch_size,
        "paf_workers": paf_workers,
        "model_paths": [str(p) for p in model_paths],
        "device": device,
        "emit_centroid": emit_centroid,
    }
    if filter_config is not None:
        kwargs["filter_config"] = filter_config
    if tracker_config is not None:
        kwargs["tracker_config"] = tracker_config

    # Spin-up log: a one-line record of *what* model is running on *what*,
    # so a run starts with a legible header instead of silence (#610).
    n_nodes = len(skeleton.nodes) if skeleton is not None else None
    spec = [
        f"type={'+'.join(model_types)}",
        f"backbone={loaded.backbone_type}",
        f"nodes={n_nodes}",
        f"device={device}",
        f"batch_size={batch_size}",
        f"peak_threshold={peak_threshold}",
        f"max_instances={max_instances}",
        f"integral_refinement={integral_refinement}",
        f"paf_workers={paf_workers}",
    ]
    if "bottomup_segmentation" in model_types:
        spec.append(f"fg_threshold={fg_threshold}")
        spec.append(f"min_mask_area={min_mask_area}")
        spec.append(f"distance_gate_alpha={distance_gate_alpha}")
        spec.append(f"merge_fragments={merge_fragments}")
        spec.append(f"full_res_masks={full_res_masks}")
        spec.append(f"mask_output={mask_output}")
    if "centered_instance_segmentation" in model_types:
        spec.append(f"fg_threshold={fg_threshold}")
        spec.append(f"mask_output={mask_output}")
    if "semantic_segmentation" in model_types:
        spec.append(f"fg_threshold={fg_threshold}")
        spec.append(f"min_mask_area={min_mask_area}")
        spec.append(f"full_res_masks={full_res_masks}")
        spec.append(f"mask_output={mask_output}")
    logger.info("Loaded inference model | " + " | ".join(spec))

    return cls(**kwargs)

predict(source, *, make_labels=True, frames=None, skeleton=None, videos=None, clean_empty_frames=False, progress_callback=None, tracking_progress_callback=None, peak_threshold=None, centroid_threshold=None, keypoint_threshold=None, max_instances=None, integral_refinement=None, integral_patch_size=None, return_confmaps=None, return_crops=None, return_pafs=None, return_paf_graph=None, return_class_maps=None, return_class_vectors=None)

Run inference on a source.

Parameters:

Name	Type	Description	Default
source	Any	sio.Video, sio.Labels, video path string, or a pre-built :class:Provider. When a non-Provider source is given, a provider is auto-constructed using self.batch_size.	required
make_labels	bool	When True (the default), return a sio.Labels. Set to False for a raw List[Outputs].	True
frames	Optional[List[int]]	Frame indices to predict on. Only used when source is an sio.Video or video path.	None
skeleton	Optional['sio.Skeleton']	sio.Skeleton for label conversion. Falls back to self.skeleton when None.	None
videos	Optional[List['sio.Video']]	Optional list of sio.Video indexed by video_indices for label conversion. Auto-derived from the source when possible.	None
clean_empty_frames	bool	When True and make_labels=True, drop LabeledFrames with no instances from the returned sio.Labels.	False
progress_callback	Optional[Callable[[int, int], None]]	Optional (processed_frames, total_frames) callback invoked after each batch. Counts are in frames (batch-size-invariant); total_frames is -1 when the provider can't report its length up front.	None
tracking_progress_callback	Optional[Callable[[int, int], None]]	Optional (processed_frames, total_frames) callback invoked after each frame during tracking.	None
peak_threshold	Optional[float]	Override peak threshold for all stages. For per-stage control on top-down models, use centroid_threshold / keypoint_threshold instead.	None
centroid_threshold	Optional[float]	Override peak threshold for the centroid stage only (top-down models).	None
keypoint_threshold	Optional[float]	Override peak threshold for the centered- instance stage only (top-down models).	None
max_instances	Optional[int]	Override max instances per frame.	None
integral_refinement	Optional[str]	"integral" or "none".	None
integral_patch_size	Optional[int]	Override integral refinement patch size.	None
return_confmaps	Optional[bool]	Override whether to return confidence maps.	None
return_crops	Optional[bool]	Override whether to return per-instance crops (top-down only).	None
return_pafs	Optional[bool]	Override whether to return part-affinity fields (bottom-up).	None
return_paf_graph	Optional[bool]	Override whether to return the PAF graph (bottom-up).	None
return_class_maps	Optional[bool]	Override whether to return class maps (multi-class bottom-up).	None
return_class_vectors	Optional[bool]	Override whether to return class vectors (multi-class top-down).	None

Returns:

Type	Description
Union[List[Outputs], 'sio.Labels']	sio.Labels (default) or List[Outputs] (when make_labels=False).

Source code in sleap_nn/inference/predictor.py

def predict(
    self,
    source: Any,
    *,
    make_labels: bool = True,
    frames: Optional[List[int]] = None,
    skeleton: Optional["sio.Skeleton"] = None,
    videos: Optional[List["sio.Video"]] = None,
    clean_empty_frames: bool = False,
    progress_callback: Optional[Callable[[int, int], None]] = None,
    tracking_progress_callback: Optional[Callable[[int, int], None]] = None,
    peak_threshold: Optional[float] = None,
    centroid_threshold: Optional[float] = None,
    keypoint_threshold: Optional[float] = None,
    max_instances: Optional[int] = None,
    integral_refinement: Optional[str] = None,
    integral_patch_size: Optional[int] = None,
    return_confmaps: Optional[bool] = None,
    return_crops: Optional[bool] = None,
    return_pafs: Optional[bool] = None,
    return_paf_graph: Optional[bool] = None,
    return_class_maps: Optional[bool] = None,
    return_class_vectors: Optional[bool] = None,
) -> Union[List[Outputs], "sio.Labels"]:
    """Run inference on a source.

    Args:
        source: ``sio.Video``, ``sio.Labels``, video path string, or
            a pre-built :class:`Provider`. When a non-Provider source
            is given, a provider is auto-constructed using
            ``self.batch_size``.
        make_labels: When ``True`` (the default), return a
            ``sio.Labels``. Set to ``False`` for a raw
            ``List[Outputs]``.
        frames: Frame indices to predict on. Only used when ``source``
            is an ``sio.Video`` or video path.
        skeleton: ``sio.Skeleton`` for label conversion. Falls back to
            ``self.skeleton`` when ``None``.
        videos: Optional list of ``sio.Video`` indexed by
            ``video_indices`` for label conversion. Auto-derived from
            the source when possible.
        clean_empty_frames: When ``True`` and ``make_labels=True``,
            drop ``LabeledFrame``s with no instances from the
            returned ``sio.Labels``.
        progress_callback: Optional ``(processed_frames, total_frames)``
            callback invoked after each batch. Counts are in frames
            (batch-size-invariant); ``total_frames`` is ``-1`` when the
            provider can't report its length up front.
        tracking_progress_callback: Optional
            ``(processed_frames, total_frames)`` callback invoked
            after each frame during tracking.
        peak_threshold: Override peak threshold for all stages. For
            per-stage control on top-down models, use
            ``centroid_threshold`` / ``keypoint_threshold`` instead.
        centroid_threshold: Override peak threshold for the centroid
            stage only (top-down models).
        keypoint_threshold: Override peak threshold for the centered-
            instance stage only (top-down models).
        max_instances: Override max instances per frame.
        integral_refinement: ``"integral"`` or ``"none"``.
        integral_patch_size: Override integral refinement patch size.
        return_confmaps: Override whether to return confidence maps.
        return_crops: Override whether to return per-instance crops
            (top-down only).
        return_pafs: Override whether to return part-affinity fields
            (bottom-up).
        return_paf_graph: Override whether to return the PAF graph
            (bottom-up).
        return_class_maps: Override whether to return class maps
            (multi-class bottom-up).
        return_class_vectors: Override whether to return class vectors
            (multi-class top-down).

    Returns:
        ``sio.Labels`` (default) or ``List[Outputs]`` (when
        ``make_labels=False``).
    """
    from datetime import datetime

    # Tracking operates on sio.PredictedInstance objects; centroid emission
    # (emit_centroid != 'instance') puts predictions in LabeledFrame.centroids
    # as sio.PredictedCentroid, which the tracker would silently drop. Fail
    # fast rather than lose data.
    if self.tracker_config is not None and self.emit_centroid != "instance":
        raise ValueError(
            "Tracking is incompatible with emit_centroid="
            f"{self.emit_centroid!r}: the tracker operates on "
            "sio.PredictedInstance objects, but this mode emits "
            "sio.PredictedCentroid objects (in LabeledFrame.centroids) that "
            "would be dropped. Use emit_centroid='instance' (the default) "
            "for tracking."
        )

    # Segmentation emits sio.PredictedSegmentationMask objects to
    # LabeledFrame.masks; apply_tracking auto-detects the mask-only labels
    # and tracks them by pixel mask-IoU (#619), so tracking is supported.

    provider, auto_videos = self._make_provider(source, frames=frames)
    if videos is None:
        videos = auto_videos

    self._log_inference_start(source, provider, videos)
    self._log_filter_config()
    _prov_start = datetime.now()
    layer = self._scoped_postprocess_layer(
        peak_threshold=peak_threshold,
        centroid_threshold=centroid_threshold,
        keypoint_threshold=keypoint_threshold,
        max_instances=max_instances,
        integral_refinement=integral_refinement,
        integral_patch_size=integral_patch_size,
        return_confmaps=return_confmaps,
        return_crops=return_crops,
        return_pafs=return_pafs,
        return_paf_graph=return_paf_graph,
        return_class_maps=return_class_maps,
        return_class_vectors=return_class_vectors,
    )
    outputs_list = list(self._batch_iter(provider, progress_callback, layer=layer))
    _prov_end = datetime.now()

    if not make_labels:
        if self.tracker_config is not None:
            raise ValueError(
                "tracker_config requires make_labels=True; the tracker "
                "operates on sio.PredictedInstance objects."
            )
        return outputs_list
    if skeleton is not None:
        self.skeleton = skeleton
    # An `embedding` model emits appearance vectors, not poses or masks, so
    # nothing packages them into a `sio.Labels` -- `make_labels=True` returned
    # an EMPTY Labels and looked like a model that predicted nothing. The CLI
    # routes this to the .h5 writer; say so here too, since the Python API can
    # reach it directly.
    if self._is_embedding_layer():
        raise ValueError(
            "make_labels=True is not supported for an `embedding` (re-ID) "
            "model: it predicts appearance vectors, and this path has nothing "
            "to attach them to. Use "
            "`sleap_nn.inference.embedding.predict_embeddings_to_slp(...)` "
            "(or `sleap-nn predict --save_embeddings slp`), which attaches each "
            "vector to its source detection via `sio.Embedding`, or pass "
            "`make_labels=False` for the raw `Outputs`."
        )
    if self.skeleton is None and not self._is_segmentation_layer():
        raise ValueError(
            "make_labels=True requires a skeleton. Either pass "
            "`skeleton=...` or build the Predictor via Predictor.from_model_paths() "
            "which sets it automatically from the training config."
        )
    labels = self.to_labels(
        outputs_list,
        videos=videos,
        keep_empty_frames=True,
    )
    if self.tracker_config is not None:
        labels = apply_tracking(
            labels, self.tracker_config, tracking_progress_callback
        )
    if clean_empty_frames:
        labels.clean(frames=True, skeletons=False)
    labels.provenance = self._build_inference_provenance(
        source=source,
        start_time=_prov_start,
        end_time=_prov_end,
        n_frames=len(labels.labeled_frames),
        inference_params={
            "peak_threshold": peak_threshold,
            "centroid_threshold": centroid_threshold,
            "keypoint_threshold": keypoint_threshold,
            "max_instances": max_instances,
            "integral_refinement": integral_refinement,
            "integral_patch_size": integral_patch_size,
            "batch_size": self.batch_size,
            **self._preprocess_provenance_params(),
        },
    )

    # Post-run summary (#610). Segmentation emits masks (LabeledFrame.masks);
    # everything else emits instances (LabeledFrame.instances).
    n_lf = len(labels.labeled_frames)
    if self._is_segmentation_layer():
        n_objects = sum(
            len(getattr(lf, "masks", None) or []) for lf in labels.labeled_frames
        )
        object_label = "masks"
    else:
        n_objects = sum(len(lf.instances) for lf in labels.labeled_frames)
        object_label = "instances"
    self._log_inference_summary(
        n_frames=n_lf,
        elapsed_s=(_prov_end - _prov_start).total_seconds(),
        n_objects=n_objects,
        object_label=object_label,
    )
    return labels

predict_streaming(source, *, frames=None, progress_callback=None, peak_threshold=None, centroid_threshold=None, keypoint_threshold=None, max_instances=None, integral_refinement=None, integral_patch_size=None, return_confmaps=None, return_crops=None, return_pafs=None, return_paf_graph=None, return_class_maps=None, return_class_vectors=None)

Yield one Outputs per batch from source.

Parameters:

Name	Type	Description	Default
source	Any	sio.Video, sio.Labels, video path string, or a pre-built :class:Provider.	required
frames	Optional[List[int]]	Frame indices (only for video sources).	None
progress_callback	Optional[Callable[[int, int], None]]	Optional (processed_frames, total_frames) callback.	None
peak_threshold	Optional[float]	Override peak threshold for all stages.	None
centroid_threshold	Optional[float]	Override centroid stage threshold (top-down).	None
keypoint_threshold	Optional[float]	Override centered-instance threshold (top-down).	None
max_instances	Optional[int]	Override max instances per frame.	None
integral_refinement	Optional[str]	"integral" or "none".	None
integral_patch_size	Optional[int]	Override integral refinement patch size.	None
return_confmaps	Optional[bool]	Override whether to return confidence maps.	None
return_crops	Optional[bool]	Override whether to return per-instance crops (top-down only).	None
return_pafs	Optional[bool]	Override whether to return part-affinity fields (bottom-up).	None
return_paf_graph	Optional[bool]	Override whether to return the PAF graph (bottom-up).	None
return_class_maps	Optional[bool]	Override whether to return class maps (multi-class bottom-up).	None
return_class_vectors	Optional[bool]	Override whether to return class vectors (multi-class top-down).	None

Source code in sleap_nn/inference/predictor.py

def predict_streaming(
    self,
    source: Any,
    *,
    frames: Optional[List[int]] = None,
    progress_callback: Optional[Callable[[int, int], None]] = None,
    peak_threshold: Optional[float] = None,
    centroid_threshold: Optional[float] = None,
    keypoint_threshold: Optional[float] = None,
    max_instances: Optional[int] = None,
    integral_refinement: Optional[str] = None,
    integral_patch_size: Optional[int] = None,
    return_confmaps: Optional[bool] = None,
    return_crops: Optional[bool] = None,
    return_pafs: Optional[bool] = None,
    return_paf_graph: Optional[bool] = None,
    return_class_maps: Optional[bool] = None,
    return_class_vectors: Optional[bool] = None,
) -> Iterator[Outputs]:
    """Yield one ``Outputs`` per batch from ``source``.

    Args:
        source: ``sio.Video``, ``sio.Labels``, video path string, or
            a pre-built :class:`Provider`.
        frames: Frame indices (only for video sources).
        progress_callback: Optional ``(processed_frames, total_frames)``
            callback.
        peak_threshold: Override peak threshold for all stages.
        centroid_threshold: Override centroid stage threshold (top-down).
        keypoint_threshold: Override centered-instance threshold (top-down).
        max_instances: Override max instances per frame.
        integral_refinement: ``"integral"`` or ``"none"``.
        integral_patch_size: Override integral refinement patch size.
        return_confmaps: Override whether to return confidence maps.
        return_crops: Override whether to return per-instance crops
            (top-down only).
        return_pafs: Override whether to return part-affinity fields
            (bottom-up).
        return_paf_graph: Override whether to return the PAF graph
            (bottom-up).
        return_class_maps: Override whether to return class maps
            (multi-class bottom-up).
        return_class_vectors: Override whether to return class vectors
            (multi-class top-down).
    """
    if self.tracker_config is not None:
        raise ValueError(
            "tracker_config is not supported on predict_streaming / "
            "predict_to_file. End-of-stream tracker cleanup needs the "
            "full LabeledFrame list; use predict() instead."
        )
    provider, _ = self._make_provider(source, frames=frames)
    layer = self._scoped_postprocess_layer(
        peak_threshold=peak_threshold,
        centroid_threshold=centroid_threshold,
        keypoint_threshold=keypoint_threshold,
        max_instances=max_instances,
        integral_refinement=integral_refinement,
        integral_patch_size=integral_patch_size,
        return_confmaps=return_confmaps,
        return_crops=return_crops,
        return_pafs=return_pafs,
        return_paf_graph=return_paf_graph,
        return_class_maps=return_class_maps,
        return_class_vectors=return_class_vectors,
    )
    if self.paf_workers > 0 and self._can_pipeline():
        yield from self._predict_streaming_pipelined(
            provider, progress_callback, layer=layer
        )
        return
    yield from self._batch_iter(provider, progress_callback, layer=layer)

predict_to_file(source, path, *, frames=None, skeleton=None, videos=None, write_interval=500, progress_callback=None)

Run inference and write results to a .slp file.

Each batch's Outputs is slimmed and converted to LabeledFrames immediately, so the heavy intermediate tensors (confmaps, PAFs) are dropped right away. The slimmed LabeledFrames accumulate until the file is finalized at close (see :class:IncrementalLabelsWriter for the memory note).

Parameters:

Name	Type	Description	Default
source	Any	sio.Video, sio.Labels, video path string, or a pre-built :class:Provider.	required
path	str	Destination .slp path.	required
frames	Optional[List[int]]	Frame indices (only for video sources).	None
skeleton	Optional['sio.Skeleton']	sio.Skeleton for instance conversion. Falls back to self.skeleton when None.	None
videos	Optional[List['sio.Video']]	Optional list of sio.Video indexed by video_indices for the saved labels.	None
write_interval	int	Number of LabeledFrames to buffer before a disk flush.	500
progress_callback	Optional[Callable[[int, int], None]]	Optional (processed_frames, total_frames) callback invoked after each batch.	None

Returns:

Type	Description
str	The (resolved) destination path string.

Source code in sleap_nn/inference/predictor.py

def predict_to_file(
    self,
    source: Any,
    path: str,
    *,
    frames: Optional[List[int]] = None,
    skeleton: Optional["sio.Skeleton"] = None,
    videos: Optional[List["sio.Video"]] = None,
    write_interval: int = 500,
    progress_callback: Optional[Callable[[int, int], None]] = None,
) -> str:
    """Run inference and write results to a ``.slp`` file.

    Each batch's ``Outputs`` is slimmed and converted to LabeledFrames
    immediately, so the heavy intermediate tensors (confmaps, PAFs) are
    dropped right away. The slimmed LabeledFrames accumulate until the
    file is finalized at close (see :class:`IncrementalLabelsWriter` for
    the memory note).

    Args:
        source: ``sio.Video``, ``sio.Labels``, video path string, or
            a pre-built :class:`Provider`.
        path: Destination ``.slp`` path.
        frames: Frame indices (only for video sources).
        skeleton: ``sio.Skeleton`` for instance conversion. Falls back
            to ``self.skeleton`` when ``None``.
        videos: Optional list of ``sio.Video`` indexed by
            ``video_indices`` for the saved labels.
        write_interval: Number of LabeledFrames to buffer before
            a disk flush.
        progress_callback: Optional ``(processed_frames, total_frames)``
            callback invoked after each batch.

    Returns:
        The (resolved) destination path string.
    """
    if skeleton is not None:
        self.skeleton = skeleton
    # Same as `predict(make_labels=True)`: an embedding model has nothing to
    # write into a `.slp`, so this streamed an empty file.
    if self._is_embedding_layer():
        raise ValueError(
            "predict_to_file is not supported for an `embedding` (re-ID) "
            "model: it predicts appearance vectors, and this path has nothing "
            "to attach them to. Use "
            "`sleap_nn.inference.embedding.predict_embeddings_to_slp(...)` "
            "(or `sleap-nn predict --save_embeddings slp`) instead."
        )

    if self.skeleton is None and not self._is_segmentation_layer():
        raise ValueError(
            "predict_to_file requires a skeleton. Either pass "
            "`skeleton=...` or build the Predictor via Predictor.from_model_paths() "
            "which sets it automatically from the training config."
        )
    from datetime import datetime

    from sleap_nn.inference.writer import IncrementalLabelsWriter

    provider, derived = self._make_provider(source, frames=frames)
    # Preserve the real source video(s) on the streamed .slp instead of the
    # 'unknown' placeholder. Mirrors the in-memory predict() path; for a
    # pre-built Provider source `derived` is None so the caller-supplied
    # `videos` (possibly None) is used (#582).
    if videos is None:
        videos = derived
    self._log_inference_start(source, provider, derived)
    self._log_filter_config()
    pkg = self._resolve_centroid_packaging()
    # NOTE: this streaming writer does NOT apply multi-class packaging — neither
    # `tracks` (pre-existing) nor `identities` are threaded into the per-batch
    # `slim.to_labels`. The default `sleap-nn predict` flow uses the in-memory
    # `Predictor.predict`/`to_labels` path (run.py), which DOES emit both. A
    # multi-class model run through this streaming path therefore omits the
    # predicted Track/Identity packaging; route such models through `predict()`.
    writer = IncrementalLabelsWriter(
        path=path,
        skeleton=self.skeleton,
        videos=videos,
        write_interval=write_interval,
        anchor_ind=pkg.anchor_ind,
        collapse_skeleton=pkg.collapse_skeleton,
        emit_centroid=pkg.emit_centroid,
        source=pkg.source,
        mask_output=getattr(self.layer, "mask_output", "mask"),
        polygon_epsilon=getattr(self.layer, "polygon_epsilon", 0.01),
    )
    _prov_start = datetime.now()
    with writer:
        for outputs in self.predict_streaming(
            provider, progress_callback=progress_callback
        ):
            writer.write(outputs)
        # Attach provenance before the context exit finalizes the .slp, so
        # the streamed output carries lineage like the in-memory path (#583).
        _prov_end = datetime.now()
        writer.provenance = self._build_inference_provenance(
            source=source,
            start_time=_prov_start,
            end_time=_prov_end,
            n_frames=writer.frame_count,
            inference_params={
                "batch_size": self.batch_size,
                **self._preprocess_provenance_params(),
            },
        )
    # Post-run summary (#610). The streaming path drops per-frame objects to
    # keep memory O(window), so report frames / throughput only.
    self._log_inference_summary(
        n_frames=writer.frame_count,
        elapsed_s=(_prov_end - _prov_start).total_seconds(),
        output=path,
    )
    return path

retrack(labels, tracker_config, clean_empty_frames=False, progress_callback=None) staticmethod

Retrack an existing sio.Labels without running inference.

Pure tracking -- useful when you already have predicted instances in a .slp and just want to (re)apply a tracker. The tracker runs once over the full LabeledFrame list; post-tracking cleanup (cull / connect-single-breaks) is applied per tracker_config.

Parameters:

Name	Type	Description	Default
labels	'sio.Labels'	A sio.Labels whose predicted_instances are tracked in-place semantics — this returns a new Labels with tracked instances.	required
tracker_config	TrackerConfig	:class:TrackerConfig to drive the tracker.	required
clean_empty_frames	bool	When True, drop empty frames from the result (matches --no_empty_frames).	False
progress_callback	Optional[Callable[[int, int], None]]	Optional (processed_frames, total_frames) callback invoked after each frame is tracked.	None

Returns:

Type	Description
'sio.Labels'	New sio.Labels with tracks attached.

Source code in sleap_nn/inference/predictor.py

@staticmethod
def retrack(
    labels: "sio.Labels",
    tracker_config: TrackerConfig,
    clean_empty_frames: bool = False,
    progress_callback: Optional[Callable[[int, int], None]] = None,
) -> "sio.Labels":
    """Retrack an existing ``sio.Labels`` without running inference.

    Pure tracking -- useful when you already have predicted instances
    in a ``.slp`` and just want to (re)apply a tracker. The tracker
    runs once over the full LabeledFrame list; post-tracking cleanup
    (cull / connect-single-breaks) is applied per ``tracker_config``.

    Args:
        labels: A ``sio.Labels`` whose ``predicted_instances`` are
            tracked in-place semantics — this returns a new
            ``Labels`` with tracked instances.
        tracker_config: :class:`TrackerConfig` to drive the tracker.
        clean_empty_frames: When ``True``, drop empty frames from
            the result (matches ``--no_empty_frames``).
        progress_callback: Optional ``(processed_frames, total_frames)``
            callback invoked after each frame is tracked.

    Returns:
        New ``sio.Labels`` with tracks attached.
    """
    out = apply_tracking(labels, tracker_config, progress_callback)
    if clean_empty_frames:
        out.clean(frames=True, skeletons=False)
    return out

to_labels(outputs_list, videos=None, keep_empty_frames=False)

Concatenate per-batch Outputs into a single sio.Labels.

Parameters:

Name	Type	Description	Default
outputs_list	List[Outputs]	Per-batch Outputs to concatenate.	required
videos	Optional[List['sio.Video']]	List of sio.Video indexed by video_indices.	None
keep_empty_frames	bool	Forwarded to :meth:Outputs.to_labels -- keep zero-detection frames instead of dropping them. :meth:predict always passes True here (matching the legacy pipeline's default of keeping every processed frame, tracking or not -- #714 fixed this for the tracking case, 717 for the non-tracking default); clean_empty_frames (--no_empty_frames on the CLI) is the opt-in way to drop them afterward, applied uniformly regardless of tracking.	False

Source code in sleap_nn/inference/predictor.py

def to_labels(
    self,
    outputs_list: List[Outputs],
    videos: Optional[List["sio.Video"]] = None,
    keep_empty_frames: bool = False,
) -> "sio.Labels":
    """Concatenate per-batch ``Outputs`` into a single ``sio.Labels``.

    Args:
        outputs_list: Per-batch ``Outputs`` to concatenate.
        videos: List of ``sio.Video`` indexed by ``video_indices``.
        keep_empty_frames: Forwarded to :meth:`Outputs.to_labels` -- keep
            zero-detection frames instead of dropping them.
            :meth:`predict` always passes ``True`` here (matching the
            legacy pipeline's default of keeping every processed frame,
            tracking or not -- #714 fixed this for the tracking case,
            #717 for the non-tracking default); ``clean_empty_frames``
            (``--no_empty_frames`` on the CLI) is the opt-in way to drop
            them afterward, applied uniformly regardless of tracking.
    """
    import sleap_io as sio

    skeleton = self.skeleton
    pkg = self._resolve_centroid_packaging()
    tracks = self._multiclass_tracks()
    identities = self._multiclass_identities()
    videos = list(videos) if videos else [None]
    # When a standalone centroid model collapses to a 1-node 'centroid'
    # skeleton, that is the skeleton attached to emitted instances and to
    # Labels.skeletons; otherwise the original training skeleton is kept.
    out_skeleton = (
        pkg.collapse_skeleton if pkg.collapse_skeleton is not None else skeleton
    )
    # Segmentation mask output representation (read off the layer; identity
    # defaults for non-segmentation layers).
    mask_output = getattr(self.layer, "mask_output", "mask")
    polygon_epsilon = getattr(self.layer, "polygon_epsilon", 0.01)
    all_lf: list = []
    used_tracks: list = []
    seen_track_ids: set = set()
    used_identities: list = []
    seen_identity_names: set = set()
    for outputs in outputs_list:
        sub = outputs.to_labels(
            skeleton=skeleton,
            videos=videos,
            anchor_ind=pkg.anchor_ind,
            tracks=tracks,
            identities=identities,
            collapse_skeleton=pkg.collapse_skeleton,
            emit_centroid=pkg.emit_centroid,
            source=pkg.source,
            mask_output=mask_output,
            polygon_epsilon=polygon_epsilon,
            keep_empty_frames=keep_empty_frames,
        )
        all_lf.extend(sub.labeled_frames)
        for trk in sub.tracks:
            if id(trk) not in seen_track_ids:
                seen_track_ids.add(id(trk))
                used_tracks.append(trk)
        # Dedup identities by name across batches (the simplified sio.Identity
        # matches by name; the same canonical objects are reused for every frame,
        # so this collapses to the registry).
        for ident in sub.identities:
            if ident.name not in seen_identity_names:
                seen_identity_names.add(ident.name)
                used_identities.append(ident)
    valid_videos = [v for v in videos if v is not None]
    labels = sio.Labels(
        labeled_frames=all_lf,
        videos=valid_videos,
        skeletons=[out_skeleton],
    )
    if used_tracks:
        labels.tracks = used_tracks
    if used_identities:
        labels.identities = used_identities
    return labels