"""Ensemble — reconcile per-technique results into a single NavResult.
The ensemble is the single point in the pipeline where multiple per-technique
estimates become one offset. Every step is honest:
1. Drop ``spurious=True`` results.
2. Drop ``at_edge=True`` results unless dropping them would empty the set.
3. Select the consensus subset: every result sponsors the set of results
that agree with it pairwise (Mahalanobis distance or pixel floor,
both measured only in the intersection of the two results' observable
subspaces, so a rank-1 flat-ring measurement and a full-rank absolute
fix that agree radially can group and fuse); the subset with the
highest summed confidence wins.
4. Exclude results outside the consensus from the combine. A lone
dissenter against a multi-technique consensus is outlier-rejected;
only a genuine alternative can force the conflicted branch: a
runner-up consensus with at least two members (absolute
summed-confidence gap test), or a lone-vs-lone standoff where the
dissenter's confidence is comparable to the winner's (gap measured
relative to the winner, so an already-excluded low-confidence
dissenter retains no veto over a clearly stronger singleton).
Vote mass, quorum, and the agreement boost count only *corroborating*
members: a pass-2 result whose prior was seeded by a technique in the
same subset re-observed its own seed, so it refines the offset but
casts no independent vote.
5. Combine offsets within the winning subset via precision-weighted
(Kalman-style) merging.
6. Apply optional disagreement / conflict penalties.
7. Cap the confidence tier at ``medium`` when the fused covariance is
rank-deficient (one translation axis unobservable) or when every
combined member is a single-star solution (a one-star unique match
or one-inlier refine) — neither an assumed axis nor an
un-cross-checked star is ``high``-tier evidence, however tight the
localization sigma.
8. Emit a NavResult carrying the excluded technique names.
The ensemble is tested in isolation against synthetic per-technique results;
correctness here is what makes the rest of the pipeline trustworthy.
"""
import copy
import math
from collections.abc import Mapping
from dataclasses import dataclass, field, fields
from typing import Any
import numpy as np
from spindoctor.annotation import Annotations
from spindoctor.config import IMAGE_LOGGER
from spindoctor.feature.constants import (
AGREEMENT_FACTOR_CAP,
COMBINED_CONFIDENCE_CAP,
)
from spindoctor.nav_orchestrator.ensemble_consensus import (
consensus_selection,
corroborating_confidence,
corroborating_members,
result_param_vector,
)
from spindoctor.nav_orchestrator.ensemble_observability import (
mixed_scale_pinvh,
observable_basis,
)
from spindoctor.nav_orchestrator.feature_summary import NavFeatureSummary
from spindoctor.nav_orchestrator.image_classifier_result import NavImageClassifierResult
from spindoctor.nav_orchestrator.nav_result import ConfidenceRank, NavResult
from spindoctor.nav_orchestrator.provenance import Provenance
from spindoctor.nav_technique.diagnostics import (
StarRefineDiagnostics,
StarUniqueMatchDiagnostics,
)
from spindoctor.nav_technique.nav_technique import technique_tier
from spindoctor.nav_technique.technique_result import NavTechniqueResult
from spindoctor.support.exceptions import NavContractError
from spindoctor.support.status_reason import NavStatusReason
from spindoctor.support.types import NDArrayFloatType
__all__ = [
'EnsembleConfig',
'derive_confidence_rank',
'ensemble',
]
# Default constants used by the ensemble. Configurable via ``EnsembleConfig``.
DEFAULT_AGREEMENT_SIGMA = 2.0
DEFAULT_AGREEMENT_PIXEL_FLOOR = 5.0
DEFAULT_AGREEMENT_GAP = 0.5
DEFAULT_DISAGREEMENT_PENALTY = 0.7
DEFAULT_CONFLICTED_CONFIDENCE_MULTIPLIER = 0.3
DEFAULT_MIN_CONFIDENCE = 0.35
DEFAULT_PINVH_RCOND = 1.0e-9
DEFAULT_MAX_ALLOWED_ROTATION_DEG = 5.0
# Tier confidence boundaries are sim-anchored (fitted 2026-07-18 on the
# recalibrated simulated-scene campaign, with the #210 covariance
# floors live and the ring truth vocabulary in the scene cohort): the
# smallest confidence at which each tier's sigma-gated subset of the
# campaign's fused results achieves a 0.9 success rate against the
# tier's error budget. The medium tier's
# sigma gate carries the discrimination by itself, so its confidence
# boundary rests at the same 0.35 floor as the low tier and the final
# gate (the tiers are sigma-differentiated). Mirrored in
# config_540_orchestrator.yaml, which carries the full provenance note.
DEFAULT_TIER_THRESHOLDS: dict[str, dict[str, float | None]] = {
'high': {'min_confidence': 0.85, 'max_sigma_px': 0.5},
'medium': {'min_confidence': 0.35, 'max_sigma_px': 2.0},
'low': {'min_confidence': 0.35, 'max_sigma_px': None},
}
[docs]
@dataclass(frozen=True)
class EnsembleConfig:
"""Tunable parameters of the ensemble combine.
Defaults match ``config_540_orchestrator.yaml``.
Parameters:
agreement_sigma: Mahalanobis-distance threshold for grouping.
agreement_pixel_floor: Translation-distance fallback grouping
threshold in pixels. Two results are grouped when *either*
their Mahalanobis distance is at most ``agreement_sigma``
*or* their Euclidean translation distance is at most this
many pixels. The pixel floor exists because per-technique
covariances are CRLB-tight (FFT subpixel localization for
``BodyDiscCorrelateNav``, M-estimator information for the
DT-fit techniques), well below the actual position
uncertainty driven by model error and pointing residuals;
without the floor, results agreeing visually to a few px
register as hundreds of sigmas apart and never group.
Set to ``0.0`` to disable the floor.
agreement_gap: Minimum summed-confidence gap between best and
runner-up groups before declaring a conflict. Applied as an
absolute gap when the runner-up has a quorum (two or more
members); in a lone-vs-lone standoff (singleton best against
a singleton excluded dissenter) the gap is measured relative
to the winner — conflict fires only when
``best - runner_up < agreement_gap * best`` — so a dissenter
with well under ``(1 - agreement_gap)`` of the winner's
confidence cannot veto it.
disagreement_penalty: Multiplier on combined confidence when more
than one group existed.
conflicted_confidence_multiplier: Additional multiplier when the
conflicted branch fires.
min_confidence: Final-result threshold below which the ensemble
returns NavResult.failed instead of NavResult.ok.
pinvh_rcond: rcond for ``scipy.linalg.pinvh``.
max_allowed_rotation_deg: Maximum magnitude (in degrees) a 3-DoF
result's rotation may take before the ensemble rejects it.
The rotation parameter is combined as a small angle (circular
mean of ``(dv, du, theta)`` triples); this bound enforces the
small-angle assumption that every contributing technique fits
against (every DT/star technique clamps its rotation to
``+-max_rotation_deg``, default 5 deg). A 3-DoF result
arriving with ``abs(rotation_rad)`` at or above this bound is a
programming error upstream and raises ``NavContractError``.
tier_thresholds: Mapping ``rank -> {min_confidence, max_sigma_px}``;
see ``derive_confidence_rank``.
"""
agreement_sigma: float = DEFAULT_AGREEMENT_SIGMA
agreement_pixel_floor: float = DEFAULT_AGREEMENT_PIXEL_FLOOR
agreement_gap: float = DEFAULT_AGREEMENT_GAP
disagreement_penalty: float = DEFAULT_DISAGREEMENT_PENALTY
conflicted_confidence_multiplier: float = DEFAULT_CONFLICTED_CONFIDENCE_MULTIPLIER
min_confidence: float = DEFAULT_MIN_CONFIDENCE
pinvh_rcond: float = DEFAULT_PINVH_RCOND
max_allowed_rotation_deg: float = DEFAULT_MAX_ALLOWED_ROTATION_DEG
tier_thresholds: dict[str, dict[str, float | None]] = field(
default_factory=lambda: copy.deepcopy(DEFAULT_TIER_THRESHOLDS)
)
[docs]
@classmethod
def from_mapping(cls, mapping: Mapping[str, Any] | None) -> 'EnsembleConfig':
"""Build an ``EnsembleConfig`` from a YAML-derived mapping.
Parameters:
mapping: The ``orchestrator.ensemble`` config section. ``None``
or empty returns the code defaults. Scalar keys override
their field; a ``tier_thresholds`` block is merged tier-by-
tier over the defaults, so a partial override (for example
only ``low.min_confidence``) leaves the other tiers intact.
Returns:
An ``EnsembleConfig`` with mapping values overriding the code
defaults field-by-field.
Raises:
ValueError: If the mapping contains an unknown key, an unknown
tier name, or an unknown tier-threshold key.
"""
if not mapping:
return cls()
data = dict(mapping)
tiers_raw = data.pop('tier_thresholds', None)
scalar_fields = {f.name for f in fields(cls)} - {'tier_thresholds'}
unknown = set(data) - scalar_fields
if unknown:
raise ValueError(f'Unknown orchestrator.ensemble config keys: {sorted(unknown)}')
kwargs: dict[str, Any] = {}
for key, value in data.items():
try:
kwargs[key] = float(value)
except (TypeError, ValueError) as exc:
raise ValueError(
f'orchestrator.ensemble.{key} must be a number; got {value!r}'
) from exc
if tiers_raw is not None:
if not isinstance(tiers_raw, Mapping):
raise ValueError('orchestrator.ensemble.tier_thresholds must be a mapping')
tiers = copy.deepcopy(DEFAULT_TIER_THRESHOLDS)
for tier, spec in tiers_raw.items():
if tier not in tiers:
raise ValueError(
f'Unknown tier {tier!r} in tier_thresholds; valid tiers: {sorted(tiers)}'
)
if not isinstance(spec, Mapping):
raise ValueError(f'tier_thresholds[{tier!r}] must be a mapping')
for key, value in spec.items():
if key not in ('min_confidence', 'max_sigma_px'):
raise ValueError(f'Unknown tier_thresholds[{tier!r}] key {key!r}')
try:
tiers[tier][key] = None if value is None else float(value)
except (TypeError, ValueError) as exc:
raise ValueError(
f'orchestrator.ensemble.tier_thresholds[{tier!r}].{key} '
f'must be a number or null; got {value!r}'
) from exc
kwargs['tier_thresholds'] = tiers
return cls(**kwargs)
def _source_bodies(result: NavTechniqueResult) -> frozenset[str]:
"""Return the set of body names a result was computed against.
Reads the structured ``NavTechniqueResult.source_bodies`` populated by
the body-feature techniques from each consumed feature's
``NavFeature.body_name``. Ring and star techniques leave it empty, so
they are naturally excluded from the body-only fallback-supersession
filter. This replaces the earlier ``feature_ids`` string-parsing, which
silently broke if the feature-id format changed.
"""
return result.source_bodies
def _drop_superseded_fallbacks(
results: list[NavTechniqueResult],
) -> list[NavTechniqueResult]:
"""Drop fallback-tier results superseded by a non-spurious primary.
A fallback result is superseded when at least one of its source
bodies appears in the source-body set of any non-spurious primary
result. Fallback results whose source body has no primary
coverage stay in the set so a scene with only a fallback (e.g.,
a body too small for limb fitting, where only BodyBlob ran) still
produces an offset.
Returns a new list preserving input ordering; the input list is
not mutated.
"""
primary_success_bodies: set[str] = set()
for r in results:
if r.spurious:
continue
if technique_tier(r.technique_name) != 'primary':
continue
primary_success_bodies.update(_source_bodies(r))
if not primary_success_bodies:
return list(results)
kept: list[NavTechniqueResult] = []
for r in results:
if technique_tier(r.technique_name) == 'fallback':
superseded = bool(_source_bodies(r) & primary_success_bodies)
if superseded:
IMAGE_LOGGER.info(
'Dropping fallback %s for body %s: superseded by a '
'non-spurious primary result on the same body',
r.technique_name,
', '.join(sorted(_source_bodies(r))) or '(unknown)',
)
continue
kept.append(r)
return kept
@dataclass(frozen=True)
class _CombinedEstimate:
"""Output of :func:`_combine_precision_weighted`.
Parameters:
offset_px: Combined ``(dv, du)`` translation.
rotation_rad: Combined rotation in radians; ``None`` when every
input was 2-DoF.
covariance_px2: Combined covariance, ``(2, 2)`` or ``(3, 3)``
matching the input parameter dimensionality.
is_rank_deficient: True when the *translation block* of the
combined covariance has an unobservable direction (a null or
sentinel axis). An unobservable rotation alone — every
contributing technique reporting the rotation sentinel — does
not make the translation fix rank-deficient.
"""
offset_px: tuple[float, float]
rotation_rad: float | None
covariance_px2: NDArrayFloatType
is_rank_deficient: bool
def _combine_precision_weighted(
group: list[NavTechniqueResult], *, rcond: float, max_allowed_rotation_deg: float
) -> _CombinedEstimate:
"""Information-form combine of a group of agreeing results.
Implements the Kalman-style information-form merge for the translation
components:
Sigma_combined = pinvh( sum_i pinvh(Sigma_i) )
mu_combined = Sigma_combined @ sum_i ( pinvh(Sigma_i) @ mu_i )
Parameter vectors carry rotation as a third component when every input
is 3-DoF; the resulting combined estimate then carries a non-``None``
``rotation_rad`` field. The rotation parameter is *not* averaged as a
plain Euclidean coordinate (which would wrap incorrectly near
``+-pi``); instead it is combined on the circle as the precision-weighted
circular mean ``atan2(sum_i w_i sin theta_i, sum_i w_i cos theta_i)``,
with ``w_i`` the rotation-component information ``pinvh(Sigma_i)[2, 2]``.
The translation components and the full combined covariance are produced
exactly as before by the information-form merge.
Parameters:
group: Non-empty list of agreeing results. Every member must
share the same parameter dimensionality (all 2-DoF or all
3-DoF — :func:`_agreement_groups` already enforces this).
rcond: rcond for ``pinvh``.
max_allowed_rotation_deg: Small-angle bound; each 3-DoF input's
``rotation_rad`` magnitude must stay strictly below this many
degrees (in radians) for the circular-mean combine to be valid.
Returns:
:class:`_CombinedEstimate`.
Raises:
ValueError: if ``group`` is empty (defensive; the orchestrator must
ensure non-emptiness before calling), if total weight is zero,
or if the group contains a mix of 2-DoF and 3-DoF results.
NavContractError: if a 3-DoF input's rotation magnitude is at or
above the small-angle bound (an upstream programming error).
"""
if not group:
raise ValueError('empty group passed to _combine_precision_weighted')
info_sum: NDArrayFloatType | None = None
info_mu_sum: NDArrayFloatType | None = None
n_params: int | None = None
max_rotation_rad = math.radians(max_allowed_rotation_deg)
rot_w_sin = 0.0
rot_w_cos = 0.0
for res in group:
cov = np.asarray(res.covariance_px2, np.float64)
if n_params is None:
n_params = cov.shape[0]
elif cov.shape[0] != n_params:
raise ValueError(
f'mixed-DoF group passed to _combine_precision_weighted: '
f'expected {n_params}-DoF, got {cov.shape[0]} from {res.technique_name}'
)
info = mixed_scale_pinvh(cov, rcond=rcond)
mu = result_param_vector(res)
if cov.shape[0] == 3:
theta = float(mu[2])
# Small-angle assumption: every contributing technique clamps its
# rotation fit to +-max_rotation_deg, so a result arriving outside
# that bound is an upstream programming error, not data the
# circular-mean combine should silently absorb. Raised as
# NavContractError (not a bare assert) so the check survives
# python -O and is never swallowed as an ordinary technique
# failure.
if abs(theta) >= max_rotation_rad:
raise NavContractError(
f'{res.technique_name}: rotation {math.degrees(theta):.3f} deg '
f'violates small-angle bound +-{max_allowed_rotation_deg:.3f} deg'
)
# Rotation weight: the scalar rotation-information element
# ``info[2, 2]`` only. The circular mean needs one scalar weight
# per angle, so the rotation-translation cross-information
# ``info[:2, 2]`` (which the information-form translation merge
# below does retain) cannot enter this average. Dropping it
# approximates each input's rotation as independent of its
# translation, i.e. it uses the conditional rotation information
# instead of the marginal ``1 / Sigma[2, 2]``; under the
# small-angle clamp enforced above the rotations differ by at
# most a few degrees, so the reweighting error this introduces
# in the combined angle is far below the reported rotation
# sigma. The combined covariance itself is exact: it comes from
# the full information-form merge, cross-terms included.
w_theta = float(info[2, 2])
rot_w_sin += w_theta * math.sin(theta)
rot_w_cos += w_theta * math.cos(theta)
if info_sum is None:
info_sum = info.copy()
info_mu_sum = info @ mu
else:
info_sum = info_sum + info
assert info_mu_sum is not None # narrowed for mypy
info_mu_sum = info_mu_sum + info @ mu
assert info_sum is not None
assert info_mu_sum is not None
cov_combined = mixed_scale_pinvh(info_sum, rcond=rcond)
mu_combined = cov_combined @ info_mu_sum
# Rank-deficiency is a property of the translation fix: the combined
# covariance's (v, u) block has an unobservable direction (exact null
# or sentinel-sized variance). Judging the full parameter matrix
# instead would let an unobservable *rotation* (every input carrying
# the rotation sentinel) mislabel a fully-constrained 2-D offset as
# rank-1.
trans_basis = observable_basis(np.ascontiguousarray(cov_combined[:2, :2]))
is_rank_deficient = trans_basis.shape[1] < 2
rotation: float | None = None
if n_params == 3:
# Combine the rotation on the circle so angles near +-pi do not cancel
# to a spurious ~0; fall back to the information-form estimate when the
# rotation weight is degenerate (every input rotation-unobservable).
if rot_w_sin == 0.0 and rot_w_cos == 0.0:
rotation = float(mu_combined[2])
else:
rotation = float(math.atan2(rot_w_sin, rot_w_cos))
return _CombinedEstimate(
offset_px=(float(mu_combined[0]), float(mu_combined[1])),
rotation_rad=rotation,
covariance_px2=cov_combined,
is_rank_deficient=is_rank_deficient,
)
def _combine_confidence(
group: list[NavTechniqueResult],
*,
rcond: float,
disagreement_penalty: float,
apply_disagreement_penalty: bool,
) -> float:
"""Precision-weighted combine of per-result confidences.
Weights are ``trace(pinvh(Sigma_i[:2, :2]))`` -- the positional (v, u)
precision in ``px^-2``, with any rotation axis marginalised out so the
weight is not skewed by the unrelated ``rad^-2`` rotation precision of a
3-DoF result. Tighter covariances contribute more to the combined
confidence than loose ones. The boosted combined
confidence reflects the number of significant contributors, capped per
``AGREEMENT_FACTOR_CAP`` and ``COMBINED_CONFIDENCE_CAP``.
Parameters:
group: Non-empty list of agreeing results.
rcond: rcond for ``pinvh``.
disagreement_penalty: Multiplier applied if other groups existed.
apply_disagreement_penalty: True if more than one group existed
before this combine.
Returns:
Combined confidence in ``[0, 1]``; never above
``COMBINED_CONFIDENCE_CAP``.
Raises:
ValueError: if every input covariance shares a null direction
(W = 0); the orchestrator's caller routes this to
``unobservable_offset`` failure.
"""
weights = []
for res in group:
cov = np.asarray(res.covariance_px2, np.float64)
# Weight by the *positional* precision only. Taking trace(pinvh(cov))
# over a full 3-DoF covariance would add the v, u precisions (px^-2) to
# the rotation precision (rad^-2), so a star-field result whose rotation
# is tightly pinned would dominate the confidence average on an
# arbitrary, unit-mixed scale. Marginalising rotation out and tracing
# the 2x2 translation block keeps every weight in px^-2; the additive
# trace (rather than det(info)^(1/p)) stays well-defined for the rank-1
# ring-edge covariances whose unobservable axis carries zero precision.
# The 2-DoF path is unchanged: cov[:2, :2] is then the whole matrix.
info_xy = mixed_scale_pinvh(np.ascontiguousarray(cov[:2, :2]), rcond=rcond)
weights.append(float(np.trace(info_xy)))
w_total = sum(weights)
if w_total <= 0.0:
raise ValueError('precision-weighted combine: total weight is zero; offset is unobservable')
weighted_avg = sum(w * r.confidence for w, r in zip(weights, group, strict=True))
weighted_avg /= w_total
significant_threshold = 0.1 * max(weights)
# The agreement boost reflects independent corroboration, so a
# prior-descendant whose seeding technique is in the group does not
# count toward it (issue #222) — its agreement re-observes its own seed. Its
# precision still participates in the weighted average.
corroborating = set(corroborating_members(group))
n_significant = sum(
1
for w, r in zip(weights, group, strict=True)
if w > significant_threshold and r in corroborating
)
if n_significant <= 1:
agreement_factor = 1.0
else:
agreement_factor = 1.0 + 0.5 * math.log2(n_significant)
agreement_factor = min(agreement_factor, AGREEMENT_FACTOR_CAP)
combined = min(weighted_avg * agreement_factor, COMBINED_CONFIDENCE_CAP)
if apply_disagreement_penalty:
combined *= disagreement_penalty
return combined
def _is_single_star_result(res: NavTechniqueResult) -> bool:
"""Return True when a result rests on a single star with no cross-check.
A one-star ``StarUniqueMatchNav`` match and a one-inlier
``StarRefineNav`` refine each localize the offset from a single
detection: nothing in the solution corroborates the identification
itself. The ensemble caps the tier of a consensus built solely from
such results (see :func:`ensemble`). Multi-star solutions and every
non-star technique return False.
"""
diag = res.diagnostics
if isinstance(diag, StarUniqueMatchDiagnostics):
return diag.mode == 'one_star'
if isinstance(diag, StarRefineDiagnostics):
return diag.n_stars_used <= 1
return False
[docs]
def derive_confidence_rank(
*,
confidence: float,
sigma_px: tuple[float, float] | None,
tier_thresholds: dict[str, dict[str, float | None]] | None = None,
) -> ConfidenceRank:
"""Derive the five-bucket confidence rank from confidence + sigma.
``max_sigma_px`` compares ``max(sigma_dv, sigma_du)`` only.
``high`` / ``medium`` / ``low`` tiers require both confidence and sigma
constraints; ``conflicted`` and ``failed`` are status-driven and not
chosen here.
Parameters:
confidence: Combined confidence in ``[0, 1]``.
sigma_px: Per-axis 1sigma marginal uncertainty (use ``None`` to mean
"unknown / not applicable").
tier_thresholds: Mapping ``rank -> {min_confidence, max_sigma_px}``
with ``max_sigma_px`` allowed to be ``None``.
Returns:
``'high'``, ``'medium'``, or ``'low'`` if any tier matches; else
``'failed'``.
"""
thresholds = tier_thresholds or DEFAULT_TIER_THRESHOLDS
max_sigma = max(sigma_px) if sigma_px is not None else None
if max_sigma is None:
# No covariance was supplied, so every sigma-constrained tier
# (high / medium) is unreachable and the result can only earn the
# best sigma-free tier (low). Surface this: a missing covariance is
# almost always an upstream technique failing to populate it, not a
# legitimately unconstrained fit, and would otherwise cap the rank
# silently.
IMAGE_LOGGER.warning(
'derive_confidence_rank: sigma_px is None (no covariance); '
'sigma-constrained tiers are unreachable and the rank caps at the '
'best sigma-free tier'
)
ranks: tuple[ConfidenceRank, ...] = ('high', 'medium', 'low')
for rank in ranks:
spec = thresholds[rank]
min_conf = spec['min_confidence']
max_allowed = spec['max_sigma_px']
assert min_conf is not None # min_confidence is always set
if confidence < min_conf:
continue
if max_allowed is not None and (max_sigma is None or max_sigma > max_allowed):
continue
return rank
return 'failed'
[docs]
def ensemble(
results: list[NavTechniqueResult],
*,
feature_inventory: list[NavFeatureSummary],
image_classifier: NavImageClassifierResult,
provenance: Provenance,
config: EnsembleConfig | None = None,
model_metadata: dict[str, dict[str, Any]] | None = None,
annotations: Annotations | None = None,
) -> NavResult:
"""Reconcile per-technique results into a single NavResult.
Parameters:
results: Per-technique results from one or both passes.
feature_inventory: Feature inventory (kept + gated entries).
image_classifier: Image-quality classifier verdict.
provenance: Reproducibility envelope.
config: Optional ``EnsembleConfig`` overrides.
model_metadata: Optional per-NavModel diagnostic dict map.
annotations: Optional pre-built annotation collection from the
orchestrator's ``_collect_annotations`` pass.
Returns:
A single NavResult — ok / conflicted / failed.
"""
cfg = config or EnsembleConfig()
md = model_metadata if model_metadata is not None else {}
ann = annotations if annotations is not None else Annotations()
if not results:
return NavResult.failed(
status_reason=NavStatusReason.NO_FEASIBLE_TECHNIQUES,
image_classifier=image_classifier,
provenance=provenance,
model_metadata=md,
annotations=ann,
)
viable = [r for r in results if not r.spurious]
if not viable:
IMAGE_LOGGER.info(
'All %d technique result(s) returned spurious=True: %s',
len(results),
', '.join(r.technique_name for r in results),
)
return NavResult.failed(
status_reason=NavStatusReason.ALL_TECHNIQUES_SPURIOUS,
image_classifier=image_classifier,
provenance=provenance,
per_technique=results,
feature_inventory=feature_inventory,
model_metadata=md,
annotations=ann,
)
# Drop fallback-tier results superseded by a non-spurious primary
# for the same body (e.g., BodyTerminatorNav / BodyBlobNav when
# BodyLimbNav or BodyDiscCorrelateNav succeeded on the same body).
# The full ``results`` list is preserved on the NavResult for
# diagnostics; only the ensemble math sees the filtered set.
viable = _drop_superseded_fallbacks(viable)
interior = [r for r in viable if not r.at_edge]
if interior:
viable = interior
selection = consensus_selection(
viable,
agreement_sigma=cfg.agreement_sigma,
agreement_pixel_floor=cfg.agreement_pixel_floor,
rcond=cfg.pinvh_rcond,
max_allowed_rotation_deg=cfg.max_allowed_rotation_deg,
)
best_group = selection.best
excluded = selection.excluded
excluded_names = [r.technique_name for r in excluded]
consensus_names = [r.technique_name for r in best_group]
# Vote mass and quorum count only corroborating members: a pass-2
# prior-descendant refines the offset but casts no independent vote
# (issue #222).
best_summed_conf = corroborating_confidence(best_group)
best_has_quorum = len(corroborating_members(best_group)) >= 2
apply_disagreement_penalty = bool(excluded)
try:
combined = _combine_precision_weighted(
best_group,
rcond=cfg.pinvh_rcond,
max_allowed_rotation_deg=cfg.max_allowed_rotation_deg,
)
combined_confidence = _combine_confidence(
best_group,
rcond=cfg.pinvh_rcond,
disagreement_penalty=cfg.disagreement_penalty,
apply_disagreement_penalty=apply_disagreement_penalty,
)
except ValueError:
# Total weight zero — offset unobservable in every contributing
# input.
IMAGE_LOGGER.info(
'Combined precision-weighted offset is unobservable: every input '
'covariance shares one null direction (%d input(s))',
len(best_group),
)
return NavResult.failed(
status_reason=NavStatusReason.UNOBSERVABLE_OFFSET,
image_classifier=image_classifier,
provenance=provenance,
per_technique=results,
feature_inventory=feature_inventory,
model_metadata=md,
annotations=ann,
)
# Conflict check. Excluded results force the conflicted branch only when
# they constitute a genuine alternative. A runner-up consensus with at
# least two members contests via the absolute summed-confidence gap. In
# a lone-vs-lone standoff (the winning subset has no quorum either) the
# gap is measured *relative to the winner*: the consensus selection has
# already excluded the dissenter as an outlier, so it retains veto power
# only while its confidence is comparable to the winner's -- a dissenter
# at half the winner's confidence is outlier-rejected, not a conflict
# (issue #258). A single dissenter against a multi-technique consensus
# is always outlier-rejected -- reported via excluded_from_consensus and
# the disagreement penalty, not as a conflict.
if excluded:
gap = best_summed_conf - selection.runner_up_confidence
if selection.runner_up_has_quorum:
is_conflict = gap < cfg.agreement_gap
elif not best_has_quorum:
is_conflict = gap < cfg.agreement_gap * best_summed_conf
else:
is_conflict = False
if is_conflict:
conflicted_confidence = combined_confidence * cfg.conflicted_confidence_multiplier
effective_gap_threshold = (
cfg.agreement_gap
if selection.runner_up_has_quorum
else cfg.agreement_gap * best_summed_conf
)
IMAGE_LOGGER.info(
'Conflicted: best-vs-runner-up summed-confidence gap %.3f is '
'below the effective agreement_gap threshold %.3f '
'(best %.3f, runner-up %.3f, runner-up quorum: %s); '
'conflicted confidence = %.3f '
'(combined %.3f x conflicted_multiplier %.3f)',
gap,
effective_gap_threshold,
best_summed_conf,
selection.runner_up_confidence,
selection.runner_up_has_quorum,
conflicted_confidence,
combined_confidence,
cfg.conflicted_confidence_multiplier,
)
return NavResult.conflicted(
offset_px=combined.offset_px,
covariance_px2=combined.covariance_px2,
confidence=conflicted_confidence,
per_technique=results,
feature_inventory=feature_inventory,
image_classifier=image_classifier,
provenance=provenance,
excluded_from_consensus=excluded_names,
consensus_techniques=consensus_names,
model_metadata=md,
annotations=ann,
)
IMAGE_LOGGER.info(
'Excluded %d result(s) from consensus as outlier(s): %s '
'(consensus %d member(s) summing %.3f; excluded runner-up %.3f, '
'quorum: %s)',
len(excluded),
', '.join(excluded_names),
len(best_group),
best_summed_conf,
selection.runner_up_confidence,
selection.runner_up_has_quorum,
)
if combined_confidence < cfg.min_confidence:
IMAGE_LOGGER.info(
'Combined confidence %.3f is below the min_confidence threshold %.3f',
combined_confidence,
cfg.min_confidence,
)
return NavResult.failed(
status_reason=NavStatusReason.FINAL_CONFIDENCE_BELOW_THRESHOLD,
image_classifier=image_classifier,
provenance=provenance,
per_technique=results,
feature_inventory=feature_inventory,
model_metadata=md,
annotations=ann,
)
cov = combined.covariance_px2
sigma_dv = float(math.sqrt(max(cov[0, 0], 0.0)))
sigma_du = float(math.sqrt(max(cov[1, 1], 0.0)))
sigma_along_unobservable_px = float('inf') if combined.is_rank_deficient else None
rank = derive_confidence_rank(
confidence=combined_confidence,
sigma_px=(sigma_dv, sigma_du),
tier_thresholds=cfg.tier_thresholds,
)
if rank == 'high' and combined.is_rank_deficient:
# A fused result with an unobservable translation axis reports one
# axis as an assumption, not a measurement. However precise the
# observable axis is, that is not a 'high'-tier absolute fix
# (issue #221);
# ``sigma_along_unobservable_px`` carries the same verdict to the
# metadata.
IMAGE_LOGGER.info(
'Tier capped at medium: the fused covariance is rank-deficient '
'(one translation axis unobservable)'
)
rank = 'medium'
if rank == 'high' and all(_is_single_star_result(r) for r in best_group):
# A single-star solution's confidence is deliberately capped to say
# "weak, no cross-check" (the one-inlier refine caps at exactly the
# high tier's confidence boundary), and its localization sigma is
# CRLB-tight, so it would otherwise clear the high gates. A result
# whose every combined member rests on one star tops out at medium
# (issue #263).
IMAGE_LOGGER.info(
'Tier capped at medium: every consensus member is a single-star '
'solution with no independent cross-check (%s)',
', '.join(r.technique_name for r in best_group),
)
rank = 'medium'
if rank == 'failed':
# Confidence + sigma combination doesn't earn any tier. Distinguish the
# two causes: if the combined confidence cleared the *lowest* tier's
# ``min_confidence`` yet still earned no tier, the offset was confident
# but too imprecise (sigma exceeded every tier's ``max_sigma_px``);
# otherwise the confidence itself was below threshold.
lowest_min_conf = min(
float(spec['min_confidence'] or 0.0) for spec in cfg.tier_thresholds.values()
)
if combined_confidence >= lowest_min_conf:
failed_reason = NavStatusReason.FINAL_SIGMA_ABOVE_THRESHOLD
else:
failed_reason = NavStatusReason.FINAL_CONFIDENCE_BELOW_THRESHOLD
IMAGE_LOGGER.info(
'No tier earned (%s): combined confidence %.3f, sigma (dv, du) = '
'(%.3f, %.3f) px (max %.3f); tier thresholds = %s',
failed_reason.value,
combined_confidence,
sigma_dv,
sigma_du,
max(sigma_dv, sigma_du),
cfg.tier_thresholds,
)
return NavResult.failed(
status_reason=failed_reason,
image_classifier=image_classifier,
provenance=provenance,
per_technique=results,
feature_inventory=feature_inventory,
model_metadata=md,
annotations=ann,
)
status_reason = (
NavStatusReason.RANK_1_ONLY if combined.is_rank_deficient else NavStatusReason.OK
)
sigma_rotation_rad: float | None = None
if combined.rotation_rad is not None and cov.shape == (3, 3):
sigma_rotation_rad = float(math.sqrt(max(cov[2, 2], 0.0)))
return NavResult.success(
offset_px=combined.offset_px,
covariance_px2=combined.covariance_px2,
confidence=combined_confidence,
confidence_rank=rank,
status_reason=status_reason,
per_technique=results,
feature_inventory=feature_inventory,
image_classifier=image_classifier,
provenance=provenance,
sigma_along_unobservable_px=sigma_along_unobservable_px,
excluded_from_consensus=excluded_names,
consensus_techniques=consensus_names,
model_metadata=md,
annotations=ann,
rotation_rad=combined.rotation_rad,
sigma_rotation_rad=sigma_rotation_rad,
)