qc

sleap.qc

Label Quality Control module for SLEAP.

This module provides tools to detect annotation errors in pose labeling data.

Example usage

import sleap_io as sio from sleap.qc import LabelQCDetector, QCConfig

Load labels

labels = sio.load_file("labels.slp")

Create detector with default config

detector = LabelQCDetector()

Fit on labels (uses all instances for training)

detector.fit(labels)

Get results

results = detector.score(labels)

Get flagged instances above threshold

flagged = results.get_flagged(threshold=0.7)

Modules:

Name	Description
config	Configuration for Label QC detector.
detector	Main Label QC Detector class.
features	Feature extraction for Label QC.
frame_level	Frame-level quality checks: instance count, duplicate detection.
gmm	Gaussian Mixture Model for anomaly detection.
insample_prediction	In-sample model prediction: labelable-but-unlabeled points (Tier-2).
results	Result classes for Label QC.

Classes:

Name	Description
LabelQCDetector	Main detection interface for Label QC.
QCConfig	Configuration for QC detector.
QCFlag	Single flagged instance with explanation.
QCResults	Container for all QC results.

LabelQCDetector

Main detection interface for Label QC.

This class provides the primary API for detecting annotation errors in pose labeling data.

Example

detector = LabelQCDetector() detector.fit(labels) results = detector.score(labels) flagged = results.get_flagged(threshold=0.7)

Attributes:

Name	Type	Description
config		Configuration for the detector.
skeleton_analyzer	Optional[SkeletonAnalyzer]	Analyzer for skeleton properties.
baseline_extractor	Optional[BaselineFeatureExtractor]	Baseline feature extractor.
gmm_detector	Optional[GMMDetector]	GMM-based anomaly detector.
zscore_detector	Optional[ZScoreDetector]	Fallback z-score detector.
visibility_model	Optional[VisibilityModel]	Visibility pattern model.
nn_scorer	Optional[NearestNeighborScorer]	Nearest neighbor scorer.
instance_count_checker	Optional[InstanceCountChecker]	Frame-level instance count checker.
use_gmm	bool	Whether GMM is being used (vs fallback).
feature_names	list[str]	Combined list of feature names.

Methods:

Name	Description
__init__	Initialize detector with optional config.
fit	Fit detector on labels (uses user-labeled instances).
flag	Return list of flagged instances above threshold.
score	Score all instances and return results.

Source code in sleap/qc/detector.py

class LabelQCDetector:
    """Main detection interface for Label QC.

    This class provides the primary API for detecting annotation errors
    in pose labeling data.

    Example:
        detector = LabelQCDetector()
        detector.fit(labels)
        results = detector.score(labels)
        flagged = results.get_flagged(threshold=0.7)

    Attributes:
        config: Configuration for the detector.
        skeleton_analyzer: Analyzer for skeleton properties.
        baseline_extractor: Baseline feature extractor.
        gmm_detector: GMM-based anomaly detector.
        zscore_detector: Fallback z-score detector.
        visibility_model: Visibility pattern model.
        nn_scorer: Nearest neighbor scorer.
        instance_count_checker: Frame-level instance count checker.
        use_gmm: Whether GMM is being used (vs fallback).
        feature_names: Combined list of feature names.
    """

    def __init__(self, config: Optional[QCConfig] = None):
        """Initialize detector with optional config.

        Args:
            config: Configuration for the detector. If None, uses defaults.
        """
        self.config = config or QCConfig()

        # These will be set during fit()
        self.skeleton_analyzer: Optional[SkeletonAnalyzer] = None
        self.baseline_extractor: Optional[BaselineFeatureExtractor] = None
        self.gmm_detector: Optional[GMMDetector] = None
        self.zscore_detector: Optional[ZScoreDetector] = None
        self.visibility_model: Optional[VisibilityModel] = None
        self.nn_scorer: Optional[NearestNeighborScorer] = None
        self.instance_count_checker: Optional[InstanceCountChecker] = None

        self.use_gmm: bool = True
        self.feature_names: list[str] = []

        # Cache for computed statistics
        self._hull_stats: Optional[dict] = None

        # B1 detector fit-time state (set in fit(), consumed in _extract_features
        # and score()). Initialized empty so _extract_features is safe even if
        # called before fit() sets them.
        self._chirality_model: Optional[dict] = None
        self._symmetry_pairs: list[tuple[int, int]] = []
        self._axis_nodes: Optional[tuple[int, int]] = None
        self._midline_nodes: list[int] = []
        self._ordering_chains: list[list[int]] = []
        self._adjacency: Optional[dict[int, list[int]]] = None
        self._co_visibility: Optional[np.ndarray] = None

        # B2 appearance-channel fit-time state (set in fit() when
        # use_appearance is on, consumed in score()). None = no appearance model.
        self._appearance_model: Optional[dict] = None

    def fit(
        self,
        labels: "sio.Labels",
        progress_callback: Optional[ProgressCallback] = None,
    ) -> "LabelQCDetector":
        """Fit detector on labels (uses user-labeled instances).

        Args:
            labels: Labels object containing annotated instances.
            progress_callback: Optional callback for progress updates.
                Called with (step_name, progress_fraction, detail_message).

        Returns:
            Self for chaining.
        """

        def _report(step: str, progress: float, detail: str = None):
            if progress_callback:
                progress_callback(step, progress, detail)

        if not labels.skeletons:
            raise ValueError("Labels must have at least one skeleton")

        skeleton = labels.skeletons[0]
        self.skeleton_analyzer = SkeletonAnalyzer(skeleton)

        # Collect all instances as arrays
        _report("Collecting instances", 0.0, None)
        instances = self._collect_instances(labels)
        if len(instances) == 0:
            raise ValueError("No instances found in labels")
        _report("Collecting instances", 0.05, f"{len(instances)} instances")

        # Fit baseline feature extractor
        _report("Fitting feature extractors", 0.05, "Baseline features")
        self.baseline_extractor = BaselineFeatureExtractor(
            edges=self.skeleton_analyzer.edges,
            n_nodes=self.skeleton_analyzer.n_nodes,
            symmetry_pairs=self.skeleton_analyzer.symmetry_pairs,
        )
        self.baseline_extractor.fit(instances)

        # Fit visibility model
        _report("Fitting feature extractors", 0.08, "Visibility model")
        visibility_masks = self._get_visibility_masks(instances)
        self.visibility_model = VisibilityModel()
        self.visibility_model.fit(visibility_masks)

        # Fit NN scorer
        _report("Fitting feature extractors", 0.10, "Nearest neighbor scorer")
        self.nn_scorer = NearestNeighborScorer(normalize=True)
        self.nn_scorer.fit(np.array(instances))

        # Compute leave-one-out NN distances for training using fast KD-tree method
        # (so training features are comparable to test features)
        _report("Computing nearest neighbors", 0.12, "Building KD-tree")
        self._training_nn_distances = self._compute_loo_nn_distances_fast(instances)
        _report("Computing nearest neighbors", 0.15, "Done")

        # Compute hull statistics for z-scoring
        _report("Computing hull statistics", 0.15, None)
        hull_areas = []
        for inst in instances:
            hull = compute_convex_hull(inst)
            if hull["hull_area"] > 0:
                hull_areas.append(hull["hull_area"])
        self._hull_stats = {
            "mean": np.mean(hull_areas) if hull_areas else 1.0,
            "std": np.std(hull_areas) if hull_areas else 1.0,
        }

        # B1 fit-time setup. These MUST exist before _extract_all_features runs,
        # since _extract_features reads them while building the feature matrix.
        _report("Fitting feature extractors", 0.16, "B1 detectors")
        sa = self.skeleton_analyzer
        self._symmetry_pairs = list(sa.symmetry_pairs) or infer_symmetry_pairs_by_name(
            sa.node_names
        )
        # Chirality measures each symmetric pair against the LOCAL tangent of the
        # body midline near that pair, so the midline must be the full ORDERED
        # set of non-symmetric nodes (nose -> tail). Two failure modes to avoid:
        #   * a single STRAIGHT axis (nose->tail chord) misjudges the side of a
        #     pair whenever the animal curls, producing false L/R-flip flags;
        #   * ``sa.spine`` (the skeleton's longest graph path) drops midline
        #     nodes that hang off a hub on a star topology (e.g. Neck/Trunk),
        #     and can even end at a side leaf, biasing the axis to one side.
        # So take ALL non-symmetric nodes and order them by their mean PCA
        # projection, which recovers nose->tail robustly across topologies.
        _sym_idxs = {i for pair in self._symmetry_pairs for i in pair}
        _midline_unordered = [i for i in range(sa.n_nodes) if i not in _sym_idxs]
        self._midline_nodes = order_midline_by_pca(instances, _midline_unordered)
        # Two-node anchor fallback for instances where < 2 midline nodes are
        # visible (compute_chirality then uses these, else a PCA axis).
        if len(self._midline_nodes) >= 2:
            self._axis_nodes = (self._midline_nodes[0], self._midline_nodes[-1])
        elif len(sa.spine) >= 2:
            self._axis_nodes = (sa.spine[0], sa.spine[-1])
        else:
            self._axis_nodes = None
        self._adjacency = sa.get_adjacency()
        self._ordering_chains = resolve_chains(
            sa.node_names, self.config.ordered_chains or None, sa.get_curvature_chains()
        )
        self._co_visibility = self.visibility_model.co_visibility_matrix
        if self.config.should_use_chirality(len(self._symmetry_pairs) >= 1):
            self._chirality_model = fit_chirality(
                instances,
                self._symmetry_pairs,
                self._midline_nodes,
                axis_node_indices=self._axis_nodes,
            )

        # B2 appearance channel (experimental, default-OFF): build a per-node
        # appearance model from the labeled frames. Guarded by use_appearance so
        # the default path never touches (potentially expensive) video decoding.
        # Each labeled frame is decoded ONCE; undecodable frames are skipped.
        if self.config.use_appearance:
            _report("Fitting feature extractors", 0.18, "Appearance model")
            appearance_pairs = []
            for video in labels.videos:
                for lf in [lf for lf in labels if lf.video == video]:
                    try:
                        frame = video[lf.frame_idx]
                    except Exception:
                        continue
                    for inst in lf.user_instances:
                        appearance_pairs.append(
                            (frame, inst.numpy(invisible_as_nan=True))
                        )
            self._appearance_model = fit_appearance(
                appearance_pairs,
                n_nodes=self.skeleton_analyzer.n_nodes,
                patch_size=self.config.appearance_patch_size,
                min_samples=self.config.appearance_min_samples,
            )

        # Build feature matrix (use LOO NN distances for training)
        _report("Extracting features", 0.20, f"0/{len(instances)}")
        self.feature_names = self._get_feature_names()  # Set first, needed by extract
        feature_matrix = self._extract_all_features(
            instances, use_loo_nn=True, progress_callback=progress_callback
        )

        # Decide between GMM and fallback
        n_samples = len(instances)
        if n_samples >= self.config.gmm_min_samples and self.config.use_gmm:
            _report("Fitting detection model", 0.70, "GMM with EM algorithm")
            self.use_gmm = True
            self.gmm_detector = GMMDetector(
                n_components=self.config.gmm_n_components,
                percentile_threshold=self.config.gmm_percentile_threshold,
            )
            self.gmm_detector.fit(feature_matrix, self.feature_names)
        else:
            _report("Fitting detection model", 0.70, "Z-score fallback")
            self.use_gmm = False
            self.zscore_detector = ZScoreDetector(threshold=3.0)
            self.zscore_detector.fit(feature_matrix)
        _report("Fitting detection model", 0.75, "Done")

        # Fit instance count checker
        _report("Fitting frame-level checkers", 0.75, None)
        frame_counts, video_ids = self._collect_frame_counts(labels)
        self.instance_count_checker = InstanceCountChecker(per_video=True)
        self.instance_count_checker.fit(frame_counts, video_ids)
        _report("Fitting complete", 0.80, None)

        return self

    def score(
        self,
        labels: "sio.Labels",
        progress_callback: Optional[ProgressCallback] = None,
    ) -> QCResults:
        """Score all instances and return results.

        Args:
            labels: Labels object to score.
            progress_callback: Optional callback for progress updates.
                Called with (step_name, progress_fraction, detail_message).

        Returns:
            QCResults containing instance scores, frame results, and
            feature contributions.
        """

        def _report(step: str, progress: float, detail: str = None):
            if progress_callback:
                progress_callback(step, progress, detail)

        if self.baseline_extractor is None:
            raise ValueError("Detector not fitted. Call fit() first.")

        results = QCResults(feature_names=self.feature_names)

        # Count total instances for progress
        total_instances = sum(len(lf.user_instances) for lf in labels)
        instance_count = 0

        # Score all instances
        _report("Scoring instances", 0.80, f"0/{total_instances}")
        for video_idx, video in enumerate(labels.videos):
            video_id = self._video_id(video, video_idx)
            labeled_frames = [lf for lf in labels if lf.video == video]

            for lf in labeled_frames:
                frame_idx = lf.frame_idx

                # Decode the frame ONCE per labeled frame for the appearance
                # channel (experimental). Hoisted out of the instance loop so a
                # frame is never decoded more than once; undecodable -> None.
                appearance_frame = None
                if self.config.use_appearance and self._appearance_model is not None:
                    try:
                        appearance_frame = lf.video[frame_idx]
                    except Exception:
                        appearance_frame = None

                # Collect instances for this frame
                frame_instances = []
                for inst_idx, inst in enumerate(lf.user_instances):
                    points = self._instance_to_array(inst)
                    frame_instances.append(points)

                    # Score instance
                    key = InstanceKey(video_idx, frame_idx, inst_idx)
                    features = self._extract_features(points)
                    score, contributions = self._score_instance(features)

                    # Pop the forced-issue marker before contributions are
                    # stored, so feature_contributions stays pure floats.
                    forced_issue = contributions.pop("_forced_top_issue", None)

                    results.instance_scores[key] = score
                    results.feature_contributions[key] = contributions

                    if forced_issue is not None:
                        results.forced_issues[key] = forced_issue

                    # Missing-node channel (experimental): scored separately from
                    # the GMM and merged in QCResults.get_flagged.
                    if (
                        self.config.use_missing_node_check
                        and self._co_visibility is not None
                    ):
                        _vmask = ~np.isnan(points).any(axis=1)
                        _mn = score_missing_nodes(
                            _vmask,
                            self._co_visibility,
                            self.skeleton_analyzer.edges,
                            threshold=self.config.missing_node_prob_threshold,
                        )
                        if _mn["missing_node_score"] > 0:
                            results.channel_scores.setdefault("missing_node", {})[
                                key
                            ] = _mn["missing_node_score"]

                    # Appearance channel (experimental): scored against the
                    # per-node appearance model using the once-decoded frame.
                    if (
                        self.config.use_appearance
                        and self._appearance_model is not None
                        and appearance_frame is not None
                    ):
                        _ap = score_appearance(
                            appearance_frame, points, self._appearance_model
                        )
                        if _ap["appearance_outlier_score"] > 0:
                            results.channel_scores.setdefault("appearance", {})[key] = (
                                _ap["appearance_outlier_score"]
                            )

                    # Progress update (every 500 instances)
                    instance_count += 1
                    if instance_count % 500 == 0:
                        progress = 0.80 + 0.18 * (instance_count / total_instances)
                        msg = f"{instance_count}/{total_instances}"
                        _report("Scoring instances", progress, msg)

                # Frame-level checks
                frame_key = FrameKey(video_idx, frame_idx)
                frame_qc = self._check_frame(
                    frame_instances, video_id, is_negative=lf.is_negative
                )
                results.frame_results[frame_key] = frame_qc

        # In-sample model-prediction channel (experimental, Tier-2 missing-node):
        # ONE batched inference over ALL labeled frames, run after the per-instance
        # loop completes. run_insample_prediction self-skips (returns an empty
        # instance_scores) when the model path is falsy, so guarding only on
        # use_insample_prediction is safe and avoids real inference by default.
        if self.config.use_insample_prediction:
            out = run_insample_prediction(
                labels,
                model_path=self.config.insample_model_path or "",
                peak_threshold=self.config.insample_peak_threshold,
                min_confidence=self.config.insample_min_confidence,
                device=self.config.insample_device,
                progress_callback=progress_callback,
            )
            for (v_idx, f_idx, i_idx), s in out["instance_scores"].items():
                results.channel_scores.setdefault("prediction", {})[
                    InstanceKey(v_idx, f_idx, i_idx)
                ] = s

        _report("Complete", 1.0, f"{instance_count} instances scored")
        return results

    def flag(self, labels: "sio.Labels", threshold: Optional[float] = None) -> list:
        """Return list of flagged instances above threshold.

        Args:
            labels: Labels object to check.
            threshold: Score threshold. If None, uses config default.

        Returns:
            List of QCFlag objects.
        """
        threshold = threshold or self.config.instance_threshold
        results = self.score(labels)
        return results.get_flagged(threshold)

    def _collect_instances(self, labels: "sio.Labels") -> list[np.ndarray]:
        """Collect all instances as numpy arrays."""
        instances = []
        for lf in labels:
            for inst in lf.user_instances:
                points = self._instance_to_array(inst)
                instances.append(points)
        return instances

    def _instance_to_array(self, instance: "sio.Instance") -> np.ndarray:
        """Convert instance to (n_nodes, 2) array with invisible points as NaN.

        Explicitly passes ``invisible_as_nan=True`` instead of relying on the
        sleap-io default. Invisible (``visible=False``) nodes must never
        contribute their stored coordinates to QC geometry features: those
        coordinates are display-only placeholders (the GUI has to draw an
        invisible node *somewhere*), and older sleap-io versions defaulted to
        returning them, which leaked far-off invisible-node coordinates into
        the edge/angle/distance/hull statistics (see #2753).

        Feature extractors treat NaN as "missing" and skip those nodes, while
        the downstream visibility mask (``~np.isnan(...)``) still records that
        the node is invisible, so the visibility-pattern features keep working.
        """
        return instance.numpy(invisible_as_nan=True)

    @staticmethod
    def _video_id(video: "sio.Video", video_idx: int) -> str:
        """Return a stable, hashable identifier for a video.

        ``Video.filename`` is a list of paths for image-sequence backends
        (e.g. ``ImageVideo``, as produced by CVAT/COCO imports). A list is
        unhashable, so it cannot be used as a dict key for the per-video
        grouping in the frame-level checks. Fall back to the video index,
        which is unique and stable across ``fit``/``score``.

        Args:
            video: The video to identify.
            video_idx: Index of the video within ``labels.videos``.

        Returns:
            The filename when it is a non-empty string, otherwise the video
            index as a string.
        """
        filename = getattr(video, "filename", None)
        if isinstance(filename, str) and filename:
            return filename
        return str(video_idx)

    def _get_visibility_masks(self, instances: list[np.ndarray]) -> np.ndarray:
        """Get visibility masks for all instances."""
        masks = []
        for inst in instances:
            mask = ~np.isnan(inst).any(axis=1)
            masks.append(mask)
        return np.array(masks)

    def _extract_features(
        self, points: np.ndarray, nn_distance: Optional[float] = None
    ) -> np.ndarray:
        """Extract combined feature vector for a single instance.

        Args:
            points: (N_nodes, 2) array of coordinates.
            nn_distance: Optional precomputed NN distance (skips slow NN query).
        """
        # Baseline features
        baseline = self.baseline_extractor.extract(points)

        # V3 features
        v3_features = []

        # Curvature
        if self.config.should_use_curvature(self.skeleton_analyzer.max_chain_length):
            chains = self.skeleton_analyzer.get_curvature_chains()
            if chains:
                curv = compute_curvature(points, chains[0])
                v3_features.extend([curv["max_curvature"], curv["curvature_std"]])
            else:
                v3_features.extend([0.0, 0.0])
        else:
            v3_features.extend([0.0, 0.0])

        # Visibility pattern
        vis_mask = ~np.isnan(points).any(axis=1)
        vis_result = self.visibility_model.score(vis_mask)
        v3_features.append(vis_result["pattern_score"])

        # NN distance (use precomputed if available)
        if nn_distance is not None:
            v3_features.append(nn_distance)
        else:
            nn_result = self.nn_scorer.score(points)
            v3_features.append(nn_result["nn_distance"])

        # Hull features
        hull = compute_convex_hull(points)
        hull_area_z = (hull["hull_area"] - self._hull_stats["mean"]) / max(
            self._hull_stats["std"], 1e-6
        )
        v3_features.extend([hull_area_z, hull["compactness"]])

        # --- B1 detectors. Each block ALWAYS appends a fixed number of values
        # (emitting 0.0 defaults when its flag is off), so the feature-vector
        # width is identical at fit and score time. The append order MUST match
        # V3_FEATURE_NAMES exactly. ---

        # (c) chirality / whole-instance L/R flip
        if self._chirality_model is not None:
            v3_features.append(
                compute_chirality(
                    points,
                    self._symmetry_pairs,
                    self._midline_nodes,
                    self._chirality_model,
                    axis_node_indices=self._axis_nodes,
                )["chirality_wrong_fraction"]
            )
        else:
            v3_features.append(0.0)

        # (d) chimera / pose-split — log1p to tame the unbounded dynamic range
        # before the GMM
        if self.config.use_split_detection:
            _ps = compute_pose_split(
                points,
                self._adjacency,
                self.baseline_extractor.stats.edge_means,
                self.baseline_extractor.stats.edge_stds,
            )["split_score"]
            v3_features.append(float(np.log1p(max(_ps, 0.0))))
        else:
            v3_features.append(0.0)

        # (b) chain ordering (experimental)
        if (
            self.config.should_use_chain_ordering(
                self.skeleton_analyzer.max_chain_length
            )
            and self._ordering_chains
        ):
            _ord = compute_chain_ordering(
                points,
                self._ordering_chains,
                max_turn_angle=np.deg2rad(self.config.chain_turn_angle_deg),
            )
            v3_features.extend(
                [_ord["order_inversion_rate"], float(_ord["chain_intersection_count"])]
            )
        else:
            v3_features.extend([0.0, 0.0])

        return np.concatenate([baseline, np.array(v3_features)])

    def _extract_all_features(
        self,
        instances: list[np.ndarray],
        use_loo_nn: bool = False,
        progress_callback: Optional[ProgressCallback] = None,
    ) -> np.ndarray:
        """Extract features for all instances.

        Uses batch NN scoring for O(n log n) performance instead of O(n²).

        Args:
            instances: List of pose arrays.
            use_loo_nn: If True, use leave-one-out NN distances (for training).
            progress_callback: Optional callback for progress updates.
        """

        def _report(step: str, progress: float, detail: str = None):
            if progress_callback:
                progress_callback(step, progress, detail)

        n = len(instances)

        # Pre-compute all NN distances in batch (fast KD-tree query)
        if use_loo_nn and hasattr(self, "_training_nn_distances"):
            # Use precomputed LOO distances for training
            nn_distances = self._training_nn_distances
        else:
            # Batch query for scoring (not LOO)
            _report("Computing NN distances", 0.20, f"Batch query for {n} instances")
            nn_distances = self.nn_scorer.score_batch(np.array(instances))

        # Extract features with precomputed NN distances
        features = []
        for i, inst in enumerate(instances):
            feat = self._extract_features(inst, nn_distance=nn_distances[i])
            features.append(feat)

            # Progress update (every 1000 instances)
            if (i + 1) % 1000 == 0:
                progress = 0.20 + 0.50 * ((i + 1) / n)
                _report("Extracting features", progress, f"{i + 1}/{n}")

        return np.array(features)

    def _compute_loo_nn_distances_fast(
        self, instances: list[np.ndarray]
    ) -> list[float]:
        """Compute leave-one-out nearest neighbor distances using KD-tree.

        Uses sklearn's NearestNeighbors with k=2 to efficiently find
        each instance's nearest neighbor (excluding itself).

        This is O(n log n) vs O(n^2) for the naive approach.

        For each instance, finds distance to nearest OTHER instance.

        Args:
            instances: List of (n_nodes, 2) pose arrays.

        Returns:
            List of LOO NN distances.
        """
        from sklearn.neighbors import NearestNeighbors

        # Normalize poses
        normalized = [normalize_pose(inst) for inst in instances]

        # Flatten and impute NaN with 0 for KD-tree
        # (NaN handling is approximate but maintains rank ordering)
        flattened = []
        for norm in normalized:
            flat = norm.flatten()
            flat = np.nan_to_num(flat, nan=0.0)
            flattened.append(flat)
        X = np.array(flattened)

        # Use KD-tree with k=2 (self + nearest other)
        nn = NearestNeighbors(n_neighbors=2, algorithm="auto", metric="euclidean")
        nn.fit(X)
        distances, _ = nn.kneighbors(X)

        # distances[:,0] is distance to self (0)
        # distances[:,1] is distance to nearest neighbor
        return distances[:, 1].tolist()

    def _compute_loo_nn_distances(self, instances: list[np.ndarray]) -> list[float]:
        """Compute leave-one-out nearest neighbor distances (naive O(n^2)).

        For each instance, finds distance to nearest OTHER instance.

        Note: For datasets > 1000 instances, use _compute_loo_nn_distances_fast
        instead which uses KD-tree for O(n log n) performance.
        """
        from sleap.qc.features.reference import pose_distance

        n = len(instances)
        normalized = [normalize_pose(inst) for inst in instances]
        loo_distances = []

        for i in range(n):
            min_dist = float("inf")
            for j in range(n):
                if i == j:
                    continue
                dist = pose_distance(normalized[i], normalized[j], method="euclidean")
                if dist < min_dist:
                    min_dist = dist
            loo_distances.append(min_dist if np.isfinite(min_dist) else 0.0)

        return loo_distances

    def _get_feature_names(self) -> list[str]:
        """Get combined feature names."""
        return BASELINE_FEATURE_NAMES + V3_FEATURE_NAMES

    def _score_instance(self, features: np.ndarray) -> tuple[float, dict[str, float]]:
        """Score an instance and return contributions."""
        # Handle NaN in features
        features_clean = np.nan_to_num(features, nan=0.0, posinf=10.0, neginf=-10.0)

        if self.use_gmm:
            result = self.gmm_detector.score(features_clean)
            score = result["normalized_score"]
        else:
            scores = self.zscore_detector.score_batch(features_clean.reshape(1, -1))
            score = scores[0] if len(scores) > 0 else 0.0

        score = float(score) if np.isfinite(score) else 0.0

        # Build contributions dict (raw feature values keyed by name).
        contributions = {}
        for i, name in enumerate(self.feature_names):
            contributions[name] = float(features[i]) if i < len(features) else 0.0

        # Raise-only hard-rule overrides. These never lower the GMM score; they
        # only force it up (and record a human-readable issue) when an
        # unambiguous structural error is present. The chimera (d) detector gets
        # NO hard rule for now — it relies on its GMM feature
        # (pose_split_score), which is why there is no clause for it here.
        forced = None
        if (
            self._chirality_model is not None
            and contributions.get("chirality_wrong_fraction", 0.0)
            >= self.config.chirality_flip_threshold
        ):
            forced = (
                max(0.9, contributions["chirality_wrong_fraction"]),
                "Whole-instance L/R flip",
            )
        elif self.config.should_use_chain_ordering(
            self.skeleton_analyzer.max_chain_length
        ) and (
            contributions.get("chain_intersection_count", 0.0) >= 1
            or contributions.get("order_inversion_rate", 0.0)
            >= self.config.order_inversion_threshold
        ):
            forced = (0.9, "Wrong keypoint order along chain")

        if forced is not None:
            score = max(score, forced[0])
            contributions["_forced_top_issue"] = forced[1]

        return score, contributions

    def _check_frame(
        self,
        instances: list[np.ndarray],
        video_id: str,
        is_negative: bool = False,
    ) -> FrameQC:
        """Check frame-level quality."""
        frame_qc = FrameQC()

        # Instance count check
        count_result = self.instance_count_checker.check(len(instances), video_id)
        frame_qc.is_incomplete = count_result["is_incomplete"]
        frame_qc.expected_instance_count = int(count_result["expected_count"])
        frame_qc.actual_instance_count = len(instances)

        # Negative (background) frames should have no instances.
        frame_qc.is_negative_with_instances = check_negative_frame(
            is_negative, len(instances)
        )

        # Duplicate detection
        if len(instances) >= 2:
            if self.config.use_duplicate_score:
                duplicates = detect_duplicates(
                    instances,
                    iou_threshold=self.config.duplicate_iou_threshold,
                    node_distance_threshold=(
                        self.config.duplicate_node_distance_threshold
                    ),
                    node_overlap_ratio=self.config.duplicate_node_overlap_ratio,
                    edge_means=self.baseline_extractor.stats.edge_means,
                    duplicate_score_threshold=self.config.duplicate_score_threshold,
                )
            else:
                # Keep current behavior: IOU + node-overlap only. An
                # unreachable score threshold (> the clamped [0, 1] max) keeps
                # the always-computed split-duplicate signal from ever firing.
                duplicates = detect_duplicates(
                    instances,
                    iou_threshold=self.config.duplicate_iou_threshold,
                    node_distance_threshold=(
                        self.config.duplicate_node_distance_threshold
                    ),
                    node_overlap_ratio=self.config.duplicate_node_overlap_ratio,
                    duplicate_score_threshold=float("inf"),
                )
            for dup in duplicates:
                frame_qc.duplicate_pairs.append((dup["index_a"], dup["index_b"]))
                frame_qc.duplicate_reasons.append(dup["reason"])
                frame_qc.duplicate_scores.append(dup.get("duplicate_score", 1.0))

        return frame_qc

    def _collect_frame_counts(
        self, labels: "sio.Labels"
    ) -> tuple[list[int], list[str]]:
        """Collect instance counts per frame."""
        counts = []
        video_ids = []
        for video_idx, video in enumerate(labels.videos):
            video_id = self._video_id(video, video_idx)
            labeled_frames = [lf for lf in labels if lf.video == video]

            for lf in labeled_frames:
                counts.append(len(lf.user_instances))
                video_ids.append(video_id)

        return counts, video_ids

__init__(config=None)

Initialize detector with optional config.

Parameters:

Name	Type	Description	Default
config	Optional[QCConfig]	Configuration for the detector. If None, uses defaults.	None

Source code in sleap/qc/detector.py

def __init__(self, config: Optional[QCConfig] = None):
    """Initialize detector with optional config.

    Args:
        config: Configuration for the detector. If None, uses defaults.
    """
    self.config = config or QCConfig()

    # These will be set during fit()
    self.skeleton_analyzer: Optional[SkeletonAnalyzer] = None
    self.baseline_extractor: Optional[BaselineFeatureExtractor] = None
    self.gmm_detector: Optional[GMMDetector] = None
    self.zscore_detector: Optional[ZScoreDetector] = None
    self.visibility_model: Optional[VisibilityModel] = None
    self.nn_scorer: Optional[NearestNeighborScorer] = None
    self.instance_count_checker: Optional[InstanceCountChecker] = None

    self.use_gmm: bool = True
    self.feature_names: list[str] = []

    # Cache for computed statistics
    self._hull_stats: Optional[dict] = None

    # B1 detector fit-time state (set in fit(), consumed in _extract_features
    # and score()). Initialized empty so _extract_features is safe even if
    # called before fit() sets them.
    self._chirality_model: Optional[dict] = None
    self._symmetry_pairs: list[tuple[int, int]] = []
    self._axis_nodes: Optional[tuple[int, int]] = None
    self._midline_nodes: list[int] = []
    self._ordering_chains: list[list[int]] = []
    self._adjacency: Optional[dict[int, list[int]]] = None
    self._co_visibility: Optional[np.ndarray] = None

    # B2 appearance-channel fit-time state (set in fit() when
    # use_appearance is on, consumed in score()). None = no appearance model.
    self._appearance_model: Optional[dict] = None

fit(labels, progress_callback=None)

Fit detector on labels (uses user-labeled instances).

Parameters:

Name	Type	Description	Default
labels	'sio.Labels'	Labels object containing annotated instances.	required
progress_callback	Optional[ProgressCallback]	Optional callback for progress updates. Called with (step_name, progress_fraction, detail_message).	None

Returns:

Type	Description
'LabelQCDetector'	Self for chaining.

Source code in sleap/qc/detector.py

def fit(
    self,
    labels: "sio.Labels",
    progress_callback: Optional[ProgressCallback] = None,
) -> "LabelQCDetector":
    """Fit detector on labels (uses user-labeled instances).

    Args:
        labels: Labels object containing annotated instances.
        progress_callback: Optional callback for progress updates.
            Called with (step_name, progress_fraction, detail_message).

    Returns:
        Self for chaining.
    """

    def _report(step: str, progress: float, detail: str = None):
        if progress_callback:
            progress_callback(step, progress, detail)

    if not labels.skeletons:
        raise ValueError("Labels must have at least one skeleton")

    skeleton = labels.skeletons[0]
    self.skeleton_analyzer = SkeletonAnalyzer(skeleton)

    # Collect all instances as arrays
    _report("Collecting instances", 0.0, None)
    instances = self._collect_instances(labels)
    if len(instances) == 0:
        raise ValueError("No instances found in labels")
    _report("Collecting instances", 0.05, f"{len(instances)} instances")

    # Fit baseline feature extractor
    _report("Fitting feature extractors", 0.05, "Baseline features")
    self.baseline_extractor = BaselineFeatureExtractor(
        edges=self.skeleton_analyzer.edges,
        n_nodes=self.skeleton_analyzer.n_nodes,
        symmetry_pairs=self.skeleton_analyzer.symmetry_pairs,
    )
    self.baseline_extractor.fit(instances)

    # Fit visibility model
    _report("Fitting feature extractors", 0.08, "Visibility model")
    visibility_masks = self._get_visibility_masks(instances)
    self.visibility_model = VisibilityModel()
    self.visibility_model.fit(visibility_masks)

    # Fit NN scorer
    _report("Fitting feature extractors", 0.10, "Nearest neighbor scorer")
    self.nn_scorer = NearestNeighborScorer(normalize=True)
    self.nn_scorer.fit(np.array(instances))

    # Compute leave-one-out NN distances for training using fast KD-tree method
    # (so training features are comparable to test features)
    _report("Computing nearest neighbors", 0.12, "Building KD-tree")
    self._training_nn_distances = self._compute_loo_nn_distances_fast(instances)
    _report("Computing nearest neighbors", 0.15, "Done")

    # Compute hull statistics for z-scoring
    _report("Computing hull statistics", 0.15, None)
    hull_areas = []
    for inst in instances:
        hull = compute_convex_hull(inst)
        if hull["hull_area"] > 0:
            hull_areas.append(hull["hull_area"])
    self._hull_stats = {
        "mean": np.mean(hull_areas) if hull_areas else 1.0,
        "std": np.std(hull_areas) if hull_areas else 1.0,
    }

    # B1 fit-time setup. These MUST exist before _extract_all_features runs,
    # since _extract_features reads them while building the feature matrix.
    _report("Fitting feature extractors", 0.16, "B1 detectors")
    sa = self.skeleton_analyzer
    self._symmetry_pairs = list(sa.symmetry_pairs) or infer_symmetry_pairs_by_name(
        sa.node_names
    )
    # Chirality measures each symmetric pair against the LOCAL tangent of the
    # body midline near that pair, so the midline must be the full ORDERED
    # set of non-symmetric nodes (nose -> tail). Two failure modes to avoid:
    #   * a single STRAIGHT axis (nose->tail chord) misjudges the side of a
    #     pair whenever the animal curls, producing false L/R-flip flags;
    #   * ``sa.spine`` (the skeleton's longest graph path) drops midline
    #     nodes that hang off a hub on a star topology (e.g. Neck/Trunk),
    #     and can even end at a side leaf, biasing the axis to one side.
    # So take ALL non-symmetric nodes and order them by their mean PCA
    # projection, which recovers nose->tail robustly across topologies.
    _sym_idxs = {i for pair in self._symmetry_pairs for i in pair}
    _midline_unordered = [i for i in range(sa.n_nodes) if i not in _sym_idxs]
    self._midline_nodes = order_midline_by_pca(instances, _midline_unordered)
    # Two-node anchor fallback for instances where < 2 midline nodes are
    # visible (compute_chirality then uses these, else a PCA axis).
    if len(self._midline_nodes) >= 2:
        self._axis_nodes = (self._midline_nodes[0], self._midline_nodes[-1])
    elif len(sa.spine) >= 2:
        self._axis_nodes = (sa.spine[0], sa.spine[-1])
    else:
        self._axis_nodes = None
    self._adjacency = sa.get_adjacency()
    self._ordering_chains = resolve_chains(
        sa.node_names, self.config.ordered_chains or None, sa.get_curvature_chains()
    )
    self._co_visibility = self.visibility_model.co_visibility_matrix
    if self.config.should_use_chirality(len(self._symmetry_pairs) >= 1):
        self._chirality_model = fit_chirality(
            instances,
            self._symmetry_pairs,
            self._midline_nodes,
            axis_node_indices=self._axis_nodes,
        )

    # B2 appearance channel (experimental, default-OFF): build a per-node
    # appearance model from the labeled frames. Guarded by use_appearance so
    # the default path never touches (potentially expensive) video decoding.
    # Each labeled frame is decoded ONCE; undecodable frames are skipped.
    if self.config.use_appearance:
        _report("Fitting feature extractors", 0.18, "Appearance model")
        appearance_pairs = []
        for video in labels.videos:
            for lf in [lf for lf in labels if lf.video == video]:
                try:
                    frame = video[lf.frame_idx]
                except Exception:
                    continue
                for inst in lf.user_instances:
                    appearance_pairs.append(
                        (frame, inst.numpy(invisible_as_nan=True))
                    )
        self._appearance_model = fit_appearance(
            appearance_pairs,
            n_nodes=self.skeleton_analyzer.n_nodes,
            patch_size=self.config.appearance_patch_size,
            min_samples=self.config.appearance_min_samples,
        )

    # Build feature matrix (use LOO NN distances for training)
    _report("Extracting features", 0.20, f"0/{len(instances)}")
    self.feature_names = self._get_feature_names()  # Set first, needed by extract
    feature_matrix = self._extract_all_features(
        instances, use_loo_nn=True, progress_callback=progress_callback
    )

    # Decide between GMM and fallback
    n_samples = len(instances)
    if n_samples >= self.config.gmm_min_samples and self.config.use_gmm:
        _report("Fitting detection model", 0.70, "GMM with EM algorithm")
        self.use_gmm = True
        self.gmm_detector = GMMDetector(
            n_components=self.config.gmm_n_components,
            percentile_threshold=self.config.gmm_percentile_threshold,
        )
        self.gmm_detector.fit(feature_matrix, self.feature_names)
    else:
        _report("Fitting detection model", 0.70, "Z-score fallback")
        self.use_gmm = False
        self.zscore_detector = ZScoreDetector(threshold=3.0)
        self.zscore_detector.fit(feature_matrix)
    _report("Fitting detection model", 0.75, "Done")

    # Fit instance count checker
    _report("Fitting frame-level checkers", 0.75, None)
    frame_counts, video_ids = self._collect_frame_counts(labels)
    self.instance_count_checker = InstanceCountChecker(per_video=True)
    self.instance_count_checker.fit(frame_counts, video_ids)
    _report("Fitting complete", 0.80, None)

    return self

flag(labels, threshold=None)

Return list of flagged instances above threshold.

Parameters:

Name	Type	Description	Default
labels	'sio.Labels'	Labels object to check.	required
threshold	Optional[float]	Score threshold. If None, uses config default.	None

Returns:

Type	Description
list	List of QCFlag objects.

Source code in sleap/qc/detector.py

def flag(self, labels: "sio.Labels", threshold: Optional[float] = None) -> list:
    """Return list of flagged instances above threshold.

    Args:
        labels: Labels object to check.
        threshold: Score threshold. If None, uses config default.

    Returns:
        List of QCFlag objects.
    """
    threshold = threshold or self.config.instance_threshold
    results = self.score(labels)
    return results.get_flagged(threshold)

score(labels, progress_callback=None)

Score all instances and return results.

Parameters:

Name	Type	Description	Default
labels	'sio.Labels'	Labels object to score.	required
progress_callback	Optional[ProgressCallback]	Optional callback for progress updates. Called with (step_name, progress_fraction, detail_message).	None

Returns:

Type	Description
QCResults	QCResults containing instance scores, frame results, and feature contributions.

Source code in sleap/qc/detector.py

def score(
    self,
    labels: "sio.Labels",
    progress_callback: Optional[ProgressCallback] = None,
) -> QCResults:
    """Score all instances and return results.

    Args:
        labels: Labels object to score.
        progress_callback: Optional callback for progress updates.
            Called with (step_name, progress_fraction, detail_message).

    Returns:
        QCResults containing instance scores, frame results, and
        feature contributions.
    """

    def _report(step: str, progress: float, detail: str = None):
        if progress_callback:
            progress_callback(step, progress, detail)

    if self.baseline_extractor is None:
        raise ValueError("Detector not fitted. Call fit() first.")

    results = QCResults(feature_names=self.feature_names)

    # Count total instances for progress
    total_instances = sum(len(lf.user_instances) for lf in labels)
    instance_count = 0

    # Score all instances
    _report("Scoring instances", 0.80, f"0/{total_instances}")
    for video_idx, video in enumerate(labels.videos):
        video_id = self._video_id(video, video_idx)
        labeled_frames = [lf for lf in labels if lf.video == video]

        for lf in labeled_frames:
            frame_idx = lf.frame_idx

            # Decode the frame ONCE per labeled frame for the appearance
            # channel (experimental). Hoisted out of the instance loop so a
            # frame is never decoded more than once; undecodable -> None.
            appearance_frame = None
            if self.config.use_appearance and self._appearance_model is not None:
                try:
                    appearance_frame = lf.video[frame_idx]
                except Exception:
                    appearance_frame = None

            # Collect instances for this frame
            frame_instances = []
            for inst_idx, inst in enumerate(lf.user_instances):
                points = self._instance_to_array(inst)
                frame_instances.append(points)

                # Score instance
                key = InstanceKey(video_idx, frame_idx, inst_idx)
                features = self._extract_features(points)
                score, contributions = self._score_instance(features)

                # Pop the forced-issue marker before contributions are
                # stored, so feature_contributions stays pure floats.
                forced_issue = contributions.pop("_forced_top_issue", None)

                results.instance_scores[key] = score
                results.feature_contributions[key] = contributions

                if forced_issue is not None:
                    results.forced_issues[key] = forced_issue

                # Missing-node channel (experimental): scored separately from
                # the GMM and merged in QCResults.get_flagged.
                if (
                    self.config.use_missing_node_check
                    and self._co_visibility is not None
                ):
                    _vmask = ~np.isnan(points).any(axis=1)
                    _mn = score_missing_nodes(
                        _vmask,
                        self._co_visibility,
                        self.skeleton_analyzer.edges,
                        threshold=self.config.missing_node_prob_threshold,
                    )
                    if _mn["missing_node_score"] > 0:
                        results.channel_scores.setdefault("missing_node", {})[
                            key
                        ] = _mn["missing_node_score"]

                # Appearance channel (experimental): scored against the
                # per-node appearance model using the once-decoded frame.
                if (
                    self.config.use_appearance
                    and self._appearance_model is not None
                    and appearance_frame is not None
                ):
                    _ap = score_appearance(
                        appearance_frame, points, self._appearance_model
                    )
                    if _ap["appearance_outlier_score"] > 0:
                        results.channel_scores.setdefault("appearance", {})[key] = (
                            _ap["appearance_outlier_score"]
                        )

                # Progress update (every 500 instances)
                instance_count += 1
                if instance_count % 500 == 0:
                    progress = 0.80 + 0.18 * (instance_count / total_instances)
                    msg = f"{instance_count}/{total_instances}"
                    _report("Scoring instances", progress, msg)

            # Frame-level checks
            frame_key = FrameKey(video_idx, frame_idx)
            frame_qc = self._check_frame(
                frame_instances, video_id, is_negative=lf.is_negative
            )
            results.frame_results[frame_key] = frame_qc

    # In-sample model-prediction channel (experimental, Tier-2 missing-node):
    # ONE batched inference over ALL labeled frames, run after the per-instance
    # loop completes. run_insample_prediction self-skips (returns an empty
    # instance_scores) when the model path is falsy, so guarding only on
    # use_insample_prediction is safe and avoids real inference by default.
    if self.config.use_insample_prediction:
        out = run_insample_prediction(
            labels,
            model_path=self.config.insample_model_path or "",
            peak_threshold=self.config.insample_peak_threshold,
            min_confidence=self.config.insample_min_confidence,
            device=self.config.insample_device,
            progress_callback=progress_callback,
        )
        for (v_idx, f_idx, i_idx), s in out["instance_scores"].items():
            results.channel_scores.setdefault("prediction", {})[
                InstanceKey(v_idx, f_idx, i_idx)
            ] = s

    _report("Complete", 1.0, f"{instance_count} instances scored")
    return results

QCConfig dataclass

Configuration for QC detector.

Attributes:

Name	Type	Description
use_gmm	bool	Whether to use GMM-based anomaly detection.
use_curvature	Literal['auto'] | bool	Whether to compute curvature features. If "auto", enables when skeleton has chains >= 5 nodes.
use_symmetry	Literal['auto'] | bool	Whether to compute symmetry features. If "auto", enables when skeleton has symmetry pairs defined.
use_anatomical	bool	Whether to compute anatomical features (signed angles).
use_chirality	Literal['auto'] | bool	Whether to compute the whole-instance left/right mirror-flip (chirality) feature. If "auto", enables when the skeleton has symmetry pairs (defined or inferred by name). Reliable detector, default-ON.
use_split_detection	bool	Whether to compute the pose-split (chimera) feature that flags a single instance spanning two animals. Reliable detector, default-ON.
use_duplicate_score	bool	Whether to fold the complementary split-duplicate signal into frame-level duplicate detection. Reliable detector, default-ON.
use_chain_ordering	Literal['auto'] | bool	Whether to compute the keypoint chain-ordering feature (wrong order along an ordered chain). If "auto", enables when the longest chain has >= 4 nodes. Experimental, default-OFF.
use_missing_node_check	bool	Whether to run the missing-node check (a node a instance's peers usually keep is absent). Experimental, default-OFF.
use_appearance	bool	Whether to run the appearance-outlier channel (a node placed on visually-wrong pixels, e.g. on bedding instead of fur). Needs decoded image frames; scored outside the GMM as the "appearance" channel. Experimental, default-OFF.
appearance_patch_size	int	Side length (pixels) of the square image patch cut around each node for the appearance descriptor.
appearance_min_samples	int	Minimum number of patch samples a node needs at fit time before the appearance model has an opinion on it.
use_insample_prediction	bool	Whether to run the in-sample model-prediction channel (Tier-2 missing-node): run a trained sleap-nn model on the labeled frames and flag unlabeled nodes the model confidently localizes. Scored outside the GMM as the "prediction" channel. EXPENSIVE (full model inference). Experimental, default-OFF.
insample_model_path	str	Path to a trained sleap-nn model directory for the in-sample prediction channel. Empty disables the channel (no-op).
insample_peak_threshold	float	Peak-finding confidence threshold passed to the in-sample model inference (lower = more candidate peaks).
insample_min_confidence	float	Confidence at/above which a model prediction at an unlabeled node counts as a disagreement (gates the channel score).
insample_device	str	Torch device for the in-sample inference ("auto"/"cpu"/"cuda"/"mps").
instance_threshold	float	Threshold for flagging instances (0-1). Higher = fewer flags, lower = more flags.
frame_threshold	float	Threshold for frame-level checks.
duplicate_iou_threshold	float	IOU threshold for duplicate detection.
duplicate_node_overlap_ratio	float	Node overlap ratio for partial duplicates.
chirality_flip_threshold	float	chirality_wrong_fraction at/above which an instance is force-flagged as a whole-instance L/R flip.
duplicate_score_threshold	float	Combined duplicate-score at/above which a pair is flagged as a duplicate at the frame level.
chain_turn_angle_deg	float	Per-interior-node turning angle (degrees) above which a chain node counts as an ordering inversion.
order_inversion_threshold	float	order_inversion_rate at/above which an instance is force-flagged as having wrong keypoint order.
missing_node_prob_threshold	float	Minimum expected-visibility probability for a missing node to be flagged as suspicious.
ordered_chains	list	User-defined ordered chains as lists of node names (ground truth ordering for the chain-ordering detector). Empty = fall back to auto-detected skeleton chains.
gmm_n_components	int	Number of GMM components.
gmm_min_samples	int	Minimum samples required for GMM fitting. Below this, falls back to z-score thresholding.
gmm_percentile_threshold	float	Percentile below which instances are anomalies.
auto_calibrate	bool	Whether to auto-calibrate threshold from data.
calibration_percentile	float	Percentile for auto-calibration.

Methods:

Name	Description
should_use_chain_ordering	Determine if the chain-ordering feature should be used.
should_use_chirality	Determine if the chirality (L/R mirror-flip) feature should be used.
should_use_curvature	Determine if curvature features should be used.
should_use_symmetry	Determine if symmetry features should be used.

Source code in sleap/qc/config.py

@dataclass
class QCConfig:
    """Configuration for QC detector.

    Attributes:
        use_gmm: Whether to use GMM-based anomaly detection.
        use_curvature: Whether to compute curvature features.
            If "auto", enables when skeleton has chains >= 5 nodes.
        use_symmetry: Whether to compute symmetry features.
            If "auto", enables when skeleton has symmetry pairs defined.
        use_anatomical: Whether to compute anatomical features (signed angles).
        use_chirality: Whether to compute the whole-instance left/right
            mirror-flip (chirality) feature. If "auto", enables when the
            skeleton has symmetry pairs (defined or inferred by name). Reliable
            detector, default-ON.
        use_split_detection: Whether to compute the pose-split (chimera) feature
            that flags a single instance spanning two animals. Reliable
            detector, default-ON.
        use_duplicate_score: Whether to fold the complementary split-duplicate
            signal into frame-level duplicate detection. Reliable detector,
            default-ON.
        use_chain_ordering: Whether to compute the keypoint chain-ordering
            feature (wrong order along an ordered chain). If "auto", enables
            when the longest chain has >= 4 nodes. Experimental, default-OFF.
        use_missing_node_check: Whether to run the missing-node check (a node a
            instance's peers usually keep is absent). Experimental, default-OFF.
        use_appearance: Whether to run the appearance-outlier channel (a node
            placed on visually-wrong pixels, e.g. on bedding instead of fur).
            Needs decoded image frames; scored outside the GMM as the
            ``"appearance"`` channel. Experimental, default-OFF.
        appearance_patch_size: Side length (pixels) of the square image patch
            cut around each node for the appearance descriptor.
        appearance_min_samples: Minimum number of patch samples a node needs at
            fit time before the appearance model has an opinion on it.
        use_insample_prediction: Whether to run the in-sample model-prediction
            channel (Tier-2 missing-node): run a trained sleap-nn model on the
            labeled frames and flag unlabeled nodes the model confidently
            localizes. Scored outside the GMM as the ``"prediction"`` channel.
            EXPENSIVE (full model inference). Experimental, default-OFF.
        insample_model_path: Path to a trained sleap-nn model directory for the
            in-sample prediction channel. Empty disables the channel (no-op).
        insample_peak_threshold: Peak-finding confidence threshold passed to the
            in-sample model inference (lower = more candidate peaks).
        insample_min_confidence: Confidence at/above which a model prediction at
            an unlabeled node counts as a disagreement (gates the channel score).
        insample_device: Torch device for the in-sample inference
            (``"auto"``/``"cpu"``/``"cuda"``/``"mps"``).
        instance_threshold: Threshold for flagging instances (0-1).
            Higher = fewer flags, lower = more flags.
        frame_threshold: Threshold for frame-level checks.
        duplicate_iou_threshold: IOU threshold for duplicate detection.
        duplicate_node_overlap_ratio: Node overlap ratio for partial duplicates.
        chirality_flip_threshold: ``chirality_wrong_fraction`` at/above which an
            instance is force-flagged as a whole-instance L/R flip.
        duplicate_score_threshold: Combined duplicate-score at/above which a
            pair is flagged as a duplicate at the frame level.
        chain_turn_angle_deg: Per-interior-node turning angle (degrees) above
            which a chain node counts as an ordering inversion.
        order_inversion_threshold: ``order_inversion_rate`` at/above which an
            instance is force-flagged as having wrong keypoint order.
        missing_node_prob_threshold: Minimum expected-visibility probability for
            a missing node to be flagged as suspicious.
        ordered_chains: User-defined ordered chains as lists of node *names*
            (ground truth ordering for the chain-ordering detector). Empty =
            fall back to auto-detected skeleton chains.
        gmm_n_components: Number of GMM components.
        gmm_min_samples: Minimum samples required for GMM fitting.
            Below this, falls back to z-score thresholding.
        gmm_percentile_threshold: Percentile below which instances are anomalies.
        auto_calibrate: Whether to auto-calibrate threshold from data.
        calibration_percentile: Percentile for auto-calibration.
    """

    # Feature selection
    use_gmm: bool = True
    use_curvature: Literal["auto"] | bool = "auto"
    use_symmetry: Literal["auto"] | bool = "auto"
    use_anatomical: bool = False
    # New detectors: reliable ones default-ON, experimental ones default-OFF.
    use_chirality: Literal["auto"] | bool = "auto"  # (c)
    use_split_detection: bool = True  # (d)
    use_duplicate_score: bool = True  # (a)
    use_chain_ordering: Literal["auto"] | bool = False  # (b) experimental
    use_missing_node_check: bool = False  # (f, Tier-1) experimental
    # B2 non-GMM channels (default-OFF / experimental).
    use_appearance: bool = False  # (e) appearance outlier / wrong-object
    use_insample_prediction: bool = False  # (f, Tier-2) in-sample model prediction

    # Thresholds (validated in v4 investigation)
    instance_threshold: float = 0.7  # Default: balanced
    frame_threshold: float = 0.5
    duplicate_iou_threshold: float = 0.5
    duplicate_node_overlap_ratio: float = 0.8
    duplicate_node_distance_threshold: float = 10.0

    # New-detector thresholds.
    chirality_flip_threshold: float = 0.5
    duplicate_score_threshold: float = 0.5
    chain_turn_angle_deg: float = 60.0
    order_inversion_threshold: float = 0.3
    missing_node_prob_threshold: float = 0.9

    # B2 appearance-outlier channel settings.
    appearance_patch_size: int = 7
    appearance_min_samples: int = 20

    # B2 in-sample model-prediction channel settings.
    insample_model_path: str = ""
    insample_peak_threshold: float = 0.2
    insample_min_confidence: float = 0.5
    insample_device: str = "auto"

    # User-defined ordered chains (lists of node NAMES) for chain-ordering.
    ordered_chains: list = field(default_factory=list)

    # GMM settings
    gmm_n_components: int = 5
    gmm_min_samples: int = 50
    gmm_percentile_threshold: float = 5.0

    # Calibration (reserved: NOT consumed anywhere yet — no auto-calibration is
    # implemented. Flagging uses the fixed instance_threshold / GUI slider value.)
    auto_calibrate: bool = True
    calibration_percentile: float = 95.0

    def should_use_curvature(self, max_chain_length: int) -> bool:
        """Determine if curvature features should be used."""
        if isinstance(self.use_curvature, bool):
            return self.use_curvature
        # Auto mode: enable for chains >= 5 nodes
        return max_chain_length >= 5

    def should_use_symmetry(self, has_symmetry: bool) -> bool:
        """Determine if symmetry features should be used."""
        if isinstance(self.use_symmetry, bool):
            return self.use_symmetry
        # Auto mode: enable if skeleton has symmetry pairs
        return has_symmetry

    def should_use_chirality(self, has_symmetry: bool) -> bool:
        """Determine if the chirality (L/R mirror-flip) feature should be used."""
        if isinstance(self.use_chirality, bool):
            return self.use_chirality
        # Auto mode: enable if skeleton has symmetry pairs (defined or inferred).
        return has_symmetry

    def should_use_chain_ordering(self, max_chain_length: int) -> bool:
        """Determine if the chain-ordering feature should be used."""
        if isinstance(self.use_chain_ordering, bool):
            return self.use_chain_ordering
        # Auto mode: enable for chains >= 4 nodes (need an interior turning angle).
        return max_chain_length >= 4

should_use_chain_ordering(max_chain_length)

Determine if the chain-ordering feature should be used.

Source code in sleap/qc/config.py

def should_use_chain_ordering(self, max_chain_length: int) -> bool:
    """Determine if the chain-ordering feature should be used."""
    if isinstance(self.use_chain_ordering, bool):
        return self.use_chain_ordering
    # Auto mode: enable for chains >= 4 nodes (need an interior turning angle).
    return max_chain_length >= 4

should_use_chirality(has_symmetry)

Determine if the chirality (L/R mirror-flip) feature should be used.

Source code in sleap/qc/config.py

def should_use_chirality(self, has_symmetry: bool) -> bool:
    """Determine if the chirality (L/R mirror-flip) feature should be used."""
    if isinstance(self.use_chirality, bool):
        return self.use_chirality
    # Auto mode: enable if skeleton has symmetry pairs (defined or inferred).
    return has_symmetry

should_use_curvature(max_chain_length)

Determine if curvature features should be used.

Source code in sleap/qc/config.py

def should_use_curvature(self, max_chain_length: int) -> bool:
    """Determine if curvature features should be used."""
    if isinstance(self.use_curvature, bool):
        return self.use_curvature
    # Auto mode: enable for chains >= 5 nodes
    return max_chain_length >= 5

should_use_symmetry(has_symmetry)

Determine if symmetry features should be used.

Source code in sleap/qc/config.py

def should_use_symmetry(self, has_symmetry: bool) -> bool:
    """Determine if symmetry features should be used."""
    if isinstance(self.use_symmetry, bool):
        return self.use_symmetry
    # Auto mode: enable if skeleton has symmetry pairs
    return has_symmetry

QCFlag dataclass

Single flagged instance with explanation.

Attributes:

Name	Type	Description
frame_idx	int	Frame index.
instance_idx	int	Instance index within the frame.
video_idx	int	Video index.

Source code in sleap/qc/results.py

@dataclass
class QCFlag:
    """Single flagged instance with explanation."""

    instance_key: InstanceKey
    score: float
    confidence: str  # "low", "medium", "high"
    top_issue: str
    feature_contributions: dict[str, float]
    explanation: str

    @property
    def video_idx(self) -> int:
        """Video index."""
        return self.instance_key.video_idx

    @property
    def frame_idx(self) -> int:
        """Frame index."""
        return self.instance_key.frame_idx

    @property
    def instance_idx(self) -> int:
        """Instance index within the frame."""
        return self.instance_key.instance_idx

frame_idx property

Frame index.

instance_idx property

Instance index within the frame.

video_idx property

Video index.

QCResults dataclass

Container for all QC results.

Attributes:

Name	Type	Description
instance_scores	dict[InstanceKey, float]	Mapping from instance key to anomaly score (0-1).
frame_results	dict[FrameKey, FrameQC]	Mapping from frame key to frame-level QC results.
feature_contributions	dict[InstanceKey, dict[str, float]]	Mapping from instance key to per-feature scores.
feature_names	list[str]	List of feature names used.
forced_issues	dict[InstanceKey, str]	Mapping from instance key to a forced top-issue label set by a detector hard rule (e.g. "Whole-instance L/R flip"). Takes precedence over inferred/channel issues in get_flagged.
channel_scores	dict[str, dict[InstanceKey, float]]	Mapping from channel name (e.g. "missing_node") to a {InstanceKey: float} of per-instance scores produced outside the GMM. Merged with the GMM score in get_flagged.

Methods:

Name	Description
get_explanation	Get human-readable explanation for instance.
get_flagged	Get instances flagged above threshold.
get_frame_issues	Get frames with issues (incomplete, duplicates, or bad negatives).
to_dataframe	Export results as DataFrame.

Source code in sleap/qc/results.py

@dataclass
class QCResults:
    """Container for all QC results.

    Attributes:
        instance_scores: Mapping from instance key to anomaly score (0-1).
        frame_results: Mapping from frame key to frame-level QC results.
        feature_contributions: Mapping from instance key to per-feature scores.
        feature_names: List of feature names used.
        forced_issues: Mapping from instance key to a forced top-issue label set
            by a detector hard rule (e.g. "Whole-instance L/R flip"). Takes
            precedence over inferred/channel issues in ``get_flagged``.
        channel_scores: Mapping from channel name (e.g. "missing_node") to a
            ``{InstanceKey: float}`` of per-instance scores produced outside the
            GMM. Merged with the GMM score in ``get_flagged``.
    """

    instance_scores: dict[InstanceKey, float] = field(default_factory=dict)
    frame_results: dict[FrameKey, FrameQC] = field(default_factory=dict)
    feature_contributions: dict[InstanceKey, dict[str, float]] = field(
        default_factory=dict
    )
    feature_names: list[str] = field(default_factory=list)
    forced_issues: dict[InstanceKey, str] = field(default_factory=dict)
    channel_scores: dict[str, dict[InstanceKey, float]] = field(default_factory=dict)

    def get_flagged(self, threshold: float = 0.7) -> list[QCFlag]:
        """Get instances flagged above threshold.

        Merges the GMM-based ``instance_scores`` with any per-channel scores
        (e.g. the missing-node channel) scored outside the GMM: the final score
        for an instance is the max of its GMM score and its best channel score.
        An instance can therefore be flagged purely on a channel even if it had
        no GMM score (its ``feature_contributions`` may be absent, in which case
        an empty dict is used safely).

        Args:
            threshold: Score threshold (0-1). Instances with a final score
                >= threshold are flagged.

        Returns:
            List of QCFlag objects, sorted by score descending.
        """
        # Union of all keys: GMM-scored instances plus any channel-only ones.
        keys = set(self.instance_scores)
        for channel in self.channel_scores.values():
            keys.update(channel)

        flagged = []
        for key in keys:
            gmm_score = self.instance_scores.get(key, 0.0)

            # Best channel score (and which channel won) for this key.
            chan = float("-inf")
            winning_channel = None
            for channel_name, channel in self.channel_scores.items():
                value = channel.get(key)
                if value is not None and value > chan:
                    chan = value
                    winning_channel = channel_name

            final = max(gmm_score, chan)
            if final < threshold:
                continue

            contributions = self.feature_contributions.get(key, {})

            # Top-issue precedence: a forced hard-rule issue wins; otherwise, if
            # a channel out-scored the GMM, use that channel's label; otherwise
            # infer from feature contributions.
            forced = self.forced_issues.get(key)
            if forced is not None:
                top_issue = forced
            elif winning_channel is not None and chan > gmm_score:
                top_issue = CHANNEL_ISSUE_LABELS.get(
                    winning_channel, f"High {winning_channel}"
                )
            else:
                top_issue = self._infer_top_issue(contributions)

            confidence = self._get_confidence(final, contributions)
            explanation = self._generate_explanation(final, top_issue, contributions)

            flagged.append(
                QCFlag(
                    instance_key=key,
                    score=final,
                    confidence=confidence,
                    top_issue=top_issue,
                    feature_contributions=contributions,
                    explanation=explanation,
                )
            )

        # Sort by score descending
        flagged.sort(key=lambda f: f.score, reverse=True)
        return flagged

    def get_frame_issues(self) -> list[tuple[FrameKey, FrameQC]]:
        """Get frames with issues (incomplete, duplicates, or bad negatives)."""
        issues = []
        for key, frame_qc in self.frame_results.items():
            if (
                frame_qc.is_incomplete
                or frame_qc.duplicate_pairs
                or frame_qc.is_negative_with_instances
            ):
                issues.append((key, frame_qc))
        return issues

    def get_explanation(self, instance_key: InstanceKey) -> str:
        """Get human-readable explanation for instance."""
        score = self.instance_scores.get(instance_key)
        if score is None:
            return "Instance not found in results."

        contributions = self.feature_contributions.get(instance_key, {})
        top_issue = self._infer_top_issue(contributions)
        return self._generate_explanation(score, top_issue, contributions)

    def to_dataframe(self) -> "pd.DataFrame":
        """Export results as DataFrame.

        Returns:
            DataFrame with columns: video_idx, frame_idx, instance_idx, score,
            confidence, top_issue, and one column per feature.
        """
        import pandas as pd

        rows = []
        for key, score in self.instance_scores.items():
            contributions = self.feature_contributions.get(key, {})
            row = {
                "video_idx": key.video_idx,
                "frame_idx": key.frame_idx,
                "instance_idx": key.instance_idx,
                "score": score,
                "confidence": self._get_confidence(score, contributions),
                "top_issue": self._infer_top_issue(contributions),
            }
            row.update(contributions)
            rows.append(row)

        return pd.DataFrame(rows)

    def _infer_top_issue(self, contributions: dict[str, float]) -> str:
        """Infer the most likely issue from feature contributions.

        Normalizes contributions to comparable scales before finding the
        dominant feature, since z-score features (~0-5) and raw distance
        features (~0-100+) have different magnitudes.
        """
        if not contributions:
            return "Unknown"

        # Normalize contributions to comparable scales
        # Z-score features are already ~0-5 range, raw features need scaling
        scale_factors = {
            # Raw distance features - scale to ~0-5 range
            "max_centroid_distance": 30.0,
            "centroid_distance_std": 10.0,
            "nn_distance": 10.0,
            # Curvature is typically 0-3
            "max_curvature": 1.0,
            "curvature_std": 1.0,
            # Rate features (0-1 range) - scale up to be comparable
            "visibility_rate": 0.3,
            "visibility_pattern_score": 0.3,
            "has_isolated_invisible": 0.3,
            # Symmetry: only meaningful if skeleton has symmetry defined
            # Value of 1.0 usually means no symmetry info, so scale down
            "min_symmetry_consistency": 5.0,
        }

        normalized = {}
        for feat, val in contributions.items():
            scale = scale_factors.get(feat, 1.0)
            # Skip features with default/uninformative values
            if feat == "min_symmetry_consistency" and val == 1.0:
                normalized[feat] = 0.0  # Ignore if no symmetry data
            else:
                normalized[feat] = val / scale

        # Find the feature with highest normalized contribution
        top_feature = max(normalized, key=normalized.get)

        # Map feature names to issue descriptions
        issue_map = {
            "max_edge_zscore": "Unusual edge length",
            "mean_edge_zscore": "Unusual proportions",
            "max_angle_zscore": "Unusual joint angle",
            "mean_angle_zscore": "Unusual pose structure",
            "max_pairwise_zscore": "Unusual node spacing",
            "mean_pairwise_zscore": "Unusual scale",
            "bbox_area_zscore": "Unusual scale",
            "max_centroid_distance": "Isolated node",
            "centroid_distance_std": "Inconsistent spacing",
            "min_symmetry_consistency": "Likely L/R swap",
            "visibility_rate": "Unusual visibility",
            "has_isolated_invisible": "Isolated invisible node",
            "visibility_pattern_score": "Unusual visibility pattern",
            "nn_distance": "Unusual pose shape",
            "max_curvature": "Unusual curvature",
            "hull_area_zscore": "Unusual pose extent",
        }

        return issue_map.get(top_feature, f"High {top_feature}")

    def _get_confidence(self, score: float, contributions: dict[str, float]) -> str:
        """Determine confidence level."""
        if score > 0.8:
            return "high"
        elif score > 0.5:
            return "medium"
        return "low"

    def _generate_explanation(
        self, score: float, top_issue: str, contributions: dict[str, float]
    ) -> str:
        """Generate human-readable explanation."""
        lines = [f"Anomaly score: {score:.2f}", f"Primary issue: {top_issue}"]

        if contributions:
            # Get top 3 contributing features
            sorted_features = sorted(
                contributions.items(), key=lambda x: x[1], reverse=True
            )[:3]
            lines.append("Top contributing features:")
            for feature, value in sorted_features:
                lines.append(f"  - {feature}: {value:.3f}")

        return "\n".join(lines)

get_explanation(instance_key)

Get human-readable explanation for instance.

Source code in sleap/qc/results.py

def get_explanation(self, instance_key: InstanceKey) -> str:
    """Get human-readable explanation for instance."""
    score = self.instance_scores.get(instance_key)
    if score is None:
        return "Instance not found in results."

    contributions = self.feature_contributions.get(instance_key, {})
    top_issue = self._infer_top_issue(contributions)
    return self._generate_explanation(score, top_issue, contributions)

get_flagged(threshold=0.7)

Get instances flagged above threshold.

Merges the GMM-based instance_scores with any per-channel scores (e.g. the missing-node channel) scored outside the GMM: the final score for an instance is the max of its GMM score and its best channel score. An instance can therefore be flagged purely on a channel even if it had no GMM score (its feature_contributions may be absent, in which case an empty dict is used safely).

Parameters:

Name	Type	Description	Default
threshold	float	Score threshold (0-1). Instances with a final score = threshold are flagged.	0.7

Returns:

Type	Description
list[QCFlag]	List of QCFlag objects, sorted by score descending.

Source code in sleap/qc/results.py

def get_flagged(self, threshold: float = 0.7) -> list[QCFlag]:
    """Get instances flagged above threshold.

    Merges the GMM-based ``instance_scores`` with any per-channel scores
    (e.g. the missing-node channel) scored outside the GMM: the final score
    for an instance is the max of its GMM score and its best channel score.
    An instance can therefore be flagged purely on a channel even if it had
    no GMM score (its ``feature_contributions`` may be absent, in which case
    an empty dict is used safely).

    Args:
        threshold: Score threshold (0-1). Instances with a final score
            >= threshold are flagged.

    Returns:
        List of QCFlag objects, sorted by score descending.
    """
    # Union of all keys: GMM-scored instances plus any channel-only ones.
    keys = set(self.instance_scores)
    for channel in self.channel_scores.values():
        keys.update(channel)

    flagged = []
    for key in keys:
        gmm_score = self.instance_scores.get(key, 0.0)

        # Best channel score (and which channel won) for this key.
        chan = float("-inf")
        winning_channel = None
        for channel_name, channel in self.channel_scores.items():
            value = channel.get(key)
            if value is not None and value > chan:
                chan = value
                winning_channel = channel_name

        final = max(gmm_score, chan)
        if final < threshold:
            continue

        contributions = self.feature_contributions.get(key, {})

        # Top-issue precedence: a forced hard-rule issue wins; otherwise, if
        # a channel out-scored the GMM, use that channel's label; otherwise
        # infer from feature contributions.
        forced = self.forced_issues.get(key)
        if forced is not None:
            top_issue = forced
        elif winning_channel is not None and chan > gmm_score:
            top_issue = CHANNEL_ISSUE_LABELS.get(
                winning_channel, f"High {winning_channel}"
            )
        else:
            top_issue = self._infer_top_issue(contributions)

        confidence = self._get_confidence(final, contributions)
        explanation = self._generate_explanation(final, top_issue, contributions)

        flagged.append(
            QCFlag(
                instance_key=key,
                score=final,
                confidence=confidence,
                top_issue=top_issue,
                feature_contributions=contributions,
                explanation=explanation,
            )
        )

    # Sort by score descending
    flagged.sort(key=lambda f: f.score, reverse=True)
    return flagged

get_frame_issues()

Get frames with issues (incomplete, duplicates, or bad negatives).

Source code in sleap/qc/results.py

def get_frame_issues(self) -> list[tuple[FrameKey, FrameQC]]:
    """Get frames with issues (incomplete, duplicates, or bad negatives)."""
    issues = []
    for key, frame_qc in self.frame_results.items():
        if (
            frame_qc.is_incomplete
            or frame_qc.duplicate_pairs
            or frame_qc.is_negative_with_instances
        ):
            issues.append((key, frame_qc))
    return issues

to_dataframe()

Export results as DataFrame.

Returns:

Type	Description
'pd.DataFrame'	DataFrame with columns: video_idx, frame_idx, instance_idx, score, confidence, top_issue, and one column per feature.

Source code in sleap/qc/results.py

def to_dataframe(self) -> "pd.DataFrame":
    """Export results as DataFrame.

    Returns:
        DataFrame with columns: video_idx, frame_idx, instance_idx, score,
        confidence, top_issue, and one column per feature.
    """
    import pandas as pd

    rows = []
    for key, score in self.instance_scores.items():
        contributions = self.feature_contributions.get(key, {})
        row = {
            "video_idx": key.video_idx,
            "frame_idx": key.frame_idx,
            "instance_idx": key.instance_idx,
            "score": score,
            "confidence": self._get_confidence(score, contributions),
            "top_issue": self._infer_top_issue(contributions),
        }
        row.update(contributions)
        rows.append(row)

    return pd.DataFrame(rows)