provenance

sleap_nn.inference.provenance

Provenance metadata utilities for inference outputs.

This module provides utilities for building and managing provenance metadata that is stored in SLP files produced during inference. Provenance metadata helps track where predictions came from and how they were generated.

Functions:

Name	Description
build_inference_provenance	Build provenance metadata dictionary for inference output.
build_tracking_only_provenance	Build provenance metadata for tracking-only pipeline.
carry_tracking_record	Keep the tracking record apply_tracking left on the labels.
merge_provenance	Merge additional provenance fields into base provenance.

build_inference_provenance(model_paths=None, model_type=None, start_time=None, end_time=None, input_labels=None, input_path=None, frames_processed=None, frames_total=None, frame_selection_method=None, inference_params=None, tracking_params=None, device=None, cli_args=None, include_system_info=True, input_provenance=None)

Build provenance metadata dictionary for inference output.

This function creates a comprehensive provenance dictionary that captures all relevant metadata about an inference run, enabling reproducibility and tracking of prediction origins.

Parameters:

Name	Type	Description	Default
model_paths	Optional[list[str]]	List of paths to model checkpoints used for inference.	None
model_type	Optional[str]	Type of model used (e.g., "top_down", "bottom_up", "single_instance").	None
start_time	Optional[datetime]	Datetime when inference started.	None
end_time	Optional[datetime]	Datetime when inference finished.	None
input_labels	Optional[Labels]	Input Labels object if inference was run on an SLP file. The provenance from this object will be preserved.	None
input_path	Optional[Union[str, Path]]	Path to input file (SLP or video).	None
frames_processed	Optional[int]	Number of frames that were processed.	None
frames_total	Optional[int]	Total number of frames in the input.	None
frame_selection_method	Optional[str]	Method used to select frames (e.g., "all", "labeled", "suggested", "range").	None
inference_params	Optional[dict[str, Any]]	Dictionary of inference parameters (peak_threshold, integral_refinement, batch_size, etc.).	None
tracking_params	Optional[dict[str, Any]]	Dictionary of tracking parameters if tracking was run.	None
device	Optional[str]	Device used for inference (e.g., "cuda:0", "cpu", "mps").	None
cli_args	Optional[dict[str, Any]]	Command-line arguments if available.	None
include_system_info	bool	If True, include detailed system information. Set to False for lighter-weight provenance.	True
input_provenance	Optional[dict[str, Any]]	The input's provenance as it was BEFORE tracking; it takes precedence over input_labels.provenance. Tracking works in place and replaces (or removes) the input's "tracking" record, so a caller that tracks input_labels first snapshots this beforehand.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing provenance metadata suitable for storing in Labels.provenance.

Example

from datetime import datetime provenance = build_inference_provenance( ... model_paths=["/path/to/model.ckpt"], ... model_type="top_down", ... start_time=datetime.now(), ... end_time=datetime.now(), ... device="cuda:0", ... ) labels.provenance = provenance labels.save("predictions.slp")

Source code in sleap_nn/inference/provenance.py

def build_inference_provenance(
    model_paths: Optional[list[str]] = None,
    model_type: Optional[str] = None,
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
    input_labels: Optional[sio.Labels] = None,
    input_path: Optional[Union[str, Path]] = None,
    frames_processed: Optional[int] = None,
    frames_total: Optional[int] = None,
    frame_selection_method: Optional[str] = None,
    inference_params: Optional[dict[str, Any]] = None,
    tracking_params: Optional[dict[str, Any]] = None,
    device: Optional[str] = None,
    cli_args: Optional[dict[str, Any]] = None,
    include_system_info: bool = True,
    input_provenance: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    """Build provenance metadata dictionary for inference output.

    This function creates a comprehensive provenance dictionary that captures
    all relevant metadata about an inference run, enabling reproducibility
    and tracking of prediction origins.

    Args:
        model_paths: List of paths to model checkpoints used for inference.
        model_type: Type of model used (e.g., "top_down", "bottom_up",
            "single_instance").
        start_time: Datetime when inference started.
        end_time: Datetime when inference finished.
        input_labels: Input Labels object if inference was run on an SLP file.
            The provenance from this object will be preserved.
        input_path: Path to input file (SLP or video).
        frames_processed: Number of frames that were processed.
        frames_total: Total number of frames in the input.
        frame_selection_method: Method used to select frames (e.g., "all",
            "labeled", "suggested", "range").
        inference_params: Dictionary of inference parameters (peak_threshold,
            integral_refinement, batch_size, etc.).
        tracking_params: Dictionary of tracking parameters if tracking was run.
        device: Device used for inference (e.g., "cuda:0", "cpu", "mps").
        cli_args: Command-line arguments if available.
        include_system_info: If True, include detailed system information.
            Set to False for lighter-weight provenance.
        input_provenance: The input's provenance as it was BEFORE tracking; it
            takes precedence over ``input_labels.provenance``. Tracking works in
            place and replaces (or removes) the input's ``"tracking"`` record, so
            a caller that tracks ``input_labels`` first snapshots this beforehand.

    Returns:
        Dictionary containing provenance metadata suitable for storing in
        Labels.provenance.

    Example:
        >>> from datetime import datetime
        >>> provenance = build_inference_provenance(
        ...     model_paths=["/path/to/model.ckpt"],
        ...     model_type="top_down",
        ...     start_time=datetime.now(),
        ...     end_time=datetime.now(),
        ...     device="cuda:0",
        ... )
        >>> labels.provenance = provenance
        >>> labels.save("predictions.slp")
    """
    provenance: dict[str, Any] = {}

    # Timestamps
    if start_time is not None:
        provenance["inference_start_timestamp"] = start_time.isoformat()
    if end_time is not None:
        provenance["inference_end_timestamp"] = end_time.isoformat()
    if start_time is not None and end_time is not None:
        runtime_seconds = (end_time - start_time).total_seconds()
        provenance["inference_runtime_seconds"] = runtime_seconds

    # Version information
    provenance["sleap_nn_version"] = sleap_nn.__version__
    provenance["sleap_io_version"] = sio.__version__

    # Model information
    if model_paths is not None:
        # Store as absolute POSIX paths for cross-platform compatibility
        provenance["model_paths"] = [
            Path(p).resolve().as_posix() if isinstance(p, (str, Path)) else str(p)
            for p in model_paths
        ]
    if model_type is not None:
        provenance["model_type"] = model_type

    # Input data lineage
    if input_path is not None:
        provenance["source_file"] = (
            Path(input_path).resolve().as_posix()
            if isinstance(input_path, (str, Path))
            else str(input_path)
        )

    # Preserve input provenance if available
    input_prov = _input_provenance(input_labels, input_provenance)
    if input_prov:
        provenance["input_provenance"] = input_prov
        # Also set source_labels for compatibility with sleap-io conventions
        if "filename" in input_prov:
            provenance["source_labels"] = input_prov["filename"]

    # Frame selection information
    if frames_processed is not None or frames_total is not None:
        frame_info: dict[str, Any] = {}
        if frame_selection_method is not None:
            frame_info["method"] = frame_selection_method
        if frames_processed is not None:
            frame_info["frames_processed"] = frames_processed
        if frames_total is not None:
            frame_info["frames_total"] = frames_total
        if frame_info:
            provenance["frame_selection"] = frame_info

    # Inference parameters
    if inference_params is not None:
        # Filter out None values and convert paths
        clean_params = {}
        for key, value in inference_params.items():
            if value is not None:
                if isinstance(value, Path):
                    clean_params[key] = value.as_posix()
                else:
                    clean_params[key] = value
        if clean_params:
            provenance["inference_config"] = clean_params

    # Tracking parameters
    if tracking_params is not None:
        clean_tracking = {k: v for k, v in tracking_params.items() if v is not None}
        if clean_tracking:
            provenance["tracking_config"] = clean_tracking

    # Device information
    if device is not None:
        provenance["device"] = device

    # CLI arguments
    if cli_args is not None:
        # Filter out None values
        clean_cli = {k: v for k, v in cli_args.items() if v is not None}
        if clean_cli:
            provenance["cli_args"] = clean_cli

    # System information (can be disabled for lighter provenance)
    if include_system_info:
        try:
            system_info = get_system_info_dict()
            # Extract key fields for provenance (avoid excessive nesting)
            provenance["system_info"] = {
                "python_version": system_info.get("python_version"),
                "platform": system_info.get("platform"),
                "pytorch_version": system_info.get("pytorch_version"),
                "cuda_version": system_info.get("cuda_version"),
                "accelerator": system_info.get("accelerator"),
                "gpu_count": system_info.get("gpu_count"),
            }
            # Include GPU names if available
            if system_info.get("gpus"):
                provenance["system_info"]["gpus"] = [
                    gpu.get("name") for gpu in system_info["gpus"]
                ]
        except Exception:
            # Don't fail inference if system info collection fails
            pass

    return provenance

build_tracking_only_provenance(input_labels=None, input_path=None, start_time=None, end_time=None, tracking_params=None, frames_processed=None, include_system_info=True, input_provenance=None)

Build provenance metadata for tracking-only pipeline.

This is a simplified version of build_inference_provenance for when only tracking is run without model inference.

Parameters:

Name	Type	Description	Default
input_labels	Optional[Labels]	Input Labels object with existing predictions.	None
input_path	Optional[Union[str, Path]]	Path to input SLP file.	None
start_time	Optional[datetime]	Datetime when tracking started.	None
end_time	Optional[datetime]	Datetime when tracking finished.	None
tracking_params	Optional[dict[str, Any]]	Dictionary of tracking parameters.	None
frames_processed	Optional[int]	Number of frames that were tracked.	None
include_system_info	bool	If True, include system information.	True
input_provenance	Optional[dict[str, Any]]	The input's provenance as it was BEFORE tracking (see :func:build_inference_provenance); takes precedence over input_labels.provenance, which tracking in place has changed.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing provenance metadata.

Source code in sleap_nn/inference/provenance.py

def build_tracking_only_provenance(
    input_labels: Optional[sio.Labels] = None,
    input_path: Optional[Union[str, Path]] = None,
    start_time: Optional[datetime] = None,
    end_time: Optional[datetime] = None,
    tracking_params: Optional[dict[str, Any]] = None,
    frames_processed: Optional[int] = None,
    include_system_info: bool = True,
    input_provenance: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    """Build provenance metadata for tracking-only pipeline.

    This is a simplified version of build_inference_provenance for when
    only tracking is run without model inference.

    Args:
        input_labels: Input Labels object with existing predictions.
        input_path: Path to input SLP file.
        start_time: Datetime when tracking started.
        end_time: Datetime when tracking finished.
        tracking_params: Dictionary of tracking parameters.
        frames_processed: Number of frames that were tracked.
        include_system_info: If True, include system information.
        input_provenance: The input's provenance as it was BEFORE tracking (see
            :func:`build_inference_provenance`); takes precedence over
            ``input_labels.provenance``, which tracking in place has changed.

    Returns:
        Dictionary containing provenance metadata.
    """
    provenance: dict[str, Any] = {}

    # Timestamps
    if start_time is not None:
        provenance["tracking_start_timestamp"] = start_time.isoformat()
    if end_time is not None:
        provenance["tracking_end_timestamp"] = end_time.isoformat()
    if start_time is not None and end_time is not None:
        runtime_seconds = (end_time - start_time).total_seconds()
        provenance["tracking_runtime_seconds"] = runtime_seconds

    # Version information
    provenance["sleap_nn_version"] = sleap_nn.__version__
    provenance["sleap_io_version"] = sio.__version__

    # Note that this is tracking-only
    provenance["pipeline_type"] = "tracking_only"

    # Input data lineage
    if input_path is not None:
        provenance["source_file"] = (
            Path(input_path).resolve().as_posix()
            if isinstance(input_path, (str, Path))
            else str(input_path)
        )

    # Preserve input provenance if available
    input_prov = _input_provenance(input_labels, input_provenance)
    if input_prov:
        provenance["input_provenance"] = input_prov
        if "filename" in input_prov:
            provenance["source_labels"] = input_prov["filename"]

    # Frame information
    if frames_processed is not None:
        provenance["frames_processed"] = frames_processed

    # Tracking parameters
    if tracking_params is not None:
        clean_tracking = {k: v for k, v in tracking_params.items() if v is not None}
        if clean_tracking:
            provenance["tracking_config"] = clean_tracking

    # System information
    if include_system_info:
        try:
            system_info = get_system_info_dict()
            provenance["system_info"] = {
                "python_version": system_info.get("python_version"),
                "platform": system_info.get("platform"),
                "pytorch_version": system_info.get("pytorch_version"),
                "accelerator": system_info.get("accelerator"),
            }
        except Exception:
            pass

    return provenance

carry_tracking_record(provenance, tracked_labels, tracker_config=None)

Keep the tracking record apply_tracking left on the labels.

The provenance builders return a FRESH dict that callers assign to labels.provenance, which would drop the "tracking" record a MOT tracker (or identity linking) wrote there. This copies it into provenance, only when tracker_config is such a run (apply_tracking removes a stale record an input file carried, so any record found is this run's). Callers snapshot the input's provenance BEFORE tracking (the builders' input_provenance), which keeps the input's own record as lineage; a caller that copied it after tracking would find this run's record in input_provenance instead, and it is removed there.

Parameters:

Name	Type	Description	Default
provenance	dict[str, Any]	The freshly built provenance dict (modified and returned).	required
tracked_labels	Any	The labels apply_tracking tracked.	required
tracker_config	Any	The run's TrackerConfig (None: no tracking).	None

Returns:

Type	Description
dict[str, Any]	provenance, with "tracking" set when this run recorded one.

Source code in sleap_nn/inference/provenance.py

def carry_tracking_record(
    provenance: dict[str, Any],
    tracked_labels: Any,
    tracker_config: Any = None,
) -> dict[str, Any]:
    """Keep the tracking record ``apply_tracking`` left on the labels.

    The provenance builders return a FRESH dict that callers assign to
    ``labels.provenance``, which would drop the ``"tracking"`` record a MOT
    tracker (or identity linking) wrote there. This copies it into
    ``provenance``, only when ``tracker_config`` is such a run (``apply_tracking``
    removes a stale record an input file carried, so any record found is this
    run's). Callers snapshot the input's provenance BEFORE tracking (the
    builders' ``input_provenance``), which keeps the input's own record as
    lineage; a caller that copied it after tracking would find this run's record
    in ``input_provenance`` instead, and it is removed there.

    Args:
        provenance: The freshly built provenance dict (modified and returned).
        tracked_labels: The labels ``apply_tracking`` tracked.
        tracker_config: The run's ``TrackerConfig`` (``None``: no tracking).

    Returns:
        ``provenance``, with ``"tracking"`` set when this run recorded one.
    """
    if tracker_config is None or not (
        getattr(tracker_config, "is_mot", False)
        or getattr(tracker_config, "link_tracklets", False)
    ):
        return provenance
    source = getattr(tracked_labels, "provenance", None)
    record = source.get(TRACKING_RECORD_KEY) if isinstance(source, dict) else None
    if record is None:
        return provenance
    provenance[TRACKING_RECORD_KEY] = record
    input_prov = provenance.get("input_provenance")
    if isinstance(input_prov, dict) and input_prov.get(TRACKING_RECORD_KEY) is record:
        input_prov = {k: v for k, v in input_prov.items() if k != TRACKING_RECORD_KEY}
        if input_prov:
            provenance["input_provenance"] = input_prov
        else:
            provenance.pop("input_provenance")
    return provenance

merge_provenance(base_provenance, additional, overwrite=True)

Merge additional provenance fields into base provenance.

Parameters:

Name	Type	Description	Default
base_provenance	dict[str, Any]	Base provenance dictionary.	required
additional	dict[str, Any]	Additional fields to merge.	required
overwrite	bool	If True, additional fields overwrite base fields. If False, base fields take precedence.	True

Returns:

Type	Description
dict[str, Any]	Merged provenance dictionary.

Source code in sleap_nn/inference/provenance.py

def merge_provenance(
    base_provenance: dict[str, Any],
    additional: dict[str, Any],
    overwrite: bool = True,
) -> dict[str, Any]:
    """Merge additional provenance fields into base provenance.

    Args:
        base_provenance: Base provenance dictionary.
        additional: Additional fields to merge.
        overwrite: If True, additional fields overwrite base fields.
            If False, base fields take precedence.

    Returns:
        Merged provenance dictionary.
    """
    result = dict(base_provenance)
    for key, value in additional.items():
        if key not in result or overwrite:
            result[key] = value
    return result