"""Adapters from the native pair estimators to the verdict rule of ``calibrex check``.

Each adapter runs one existing estimator on a bag, with the *candidate*
(deployed) transform as its reference, and returns an :class:`EstimatorRun`:
the estimator's artifact(s), per-axis :class:`~calibrex.check.verdict.AxisEstimate`
values and the clock offset. The adapters are looked up in :data:`ESTIMATORS`
so tests can replace them.

Conventions
-----------
Every estimator reports ``T_parent_child`` of its own convention: ``imu-lidar``
reports ``T_lidar_imu``, ``camera-imu`` ``T_cam_imu`` and ``lidar-lidar``
``T_reference_target``. The planner's ``T_first_second`` is converted to that
convention (``imu-lidar`` is inverted) before comparing. Rotation errors are
``rotvec(R_candidate R_estimate^T)`` in degrees, a small rotation about the
parent frame's axes, which is the convention of the estimators' standard
deviations; translation errors are component differences in metres.
"""

from __future__ import annotations

import hashlib
import importlib.util
import math
import tempfile
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeAlias

import numpy as np
from numpy.typing import NDArray

from calibrex.check.cache import EstimatorCache, cached_artifact
from calibrex.check.frame_tree import StaticFrameTree
from calibrex.check.tf_sources import LoadedSource, camchain_entries
from calibrex.check.verdict import (
    RIGID_SCAN_ROTATION_FLOOR_DEG,
    AxisEstimate,
    FocalEstimate,
    FocalRecord,
)
from calibrex.core.calibration_check import (
    CheckAxisName,
    CheckPairRecord,
    CheckReasonCode,
    CheckTimeOffset,
    CheckTopicRecord,
)
from calibrex.core.progress import emit_stage
from calibrex.data.ros_cdr import AUTOWARE_VELOCITY_REPORT_TYPES

if TYPE_CHECKING:
    from calibrex.core.camera_lidar_edge import CameraLidarEdgeArtifact
    from calibrex.data.livox_ros2 import ScanStore
    from calibrex.evaluation.imu_lidar_rotation import ImuLidarRunOptions
    from calibrex.evaluation.visual_rotation import CameraModel

FloatArray: TypeAlias = NDArray[np.float64]
PointTimeEncoding = Literal["offset_s", "absolute_ns", "absolute_s"]
POINT_TIME_FIELD_CANDIDATES: tuple[str, ...] = (
    "offset_time",
    "t",
    "time",
    "timestamp",
    "time_stamp",
    "timestamps",
    "point_time",
    "time_offset",
    "relative_time",
)
POINTCLOUD2_TYPE = "sensor_msgs/msg/PointCloud2"
IMAGE_TYPE = "sensor_msgs/msg/Image"
CAMERA_INFO_TYPE = "sensor_msgs/msg/CameraInfo"
DATASET_FAMILY = "calibrex-check"
DATASET_LICENSE = "user-provided"


class CheckSkipError(Exception):
    """The pair cannot be run: a machine-readable reason and a message."""

    def __init__(self, code: CheckReasonCode, reason: str) -> None:
        super().__init__(reason)
        self.code: CheckReasonCode = code
        self.reason = reason


class SavableArtifact(Protocol):
    """A schema artifact that can be written to YAML or JSON."""

    @property
    def schema_version(self) -> str: ...

    def save(self, path: str | Path) -> None: ...


@dataclass(frozen=True)
class EvidenceArtifact:
    """One estimator artifact to be written to the evidence directory."""

    role: str
    artifact: SavableArtifact
    policy_status: str | None = None
    from_cache: bool | None = None


@dataclass(frozen=True)
class EstimatorRun:
    """What an adapter returns for one pair."""

    estimator: str
    artifacts: tuple[EvidenceArtifact, ...]
    solved: bool
    policy_status: str
    policy_reasons: tuple[str, ...]
    estimates: tuple[AxisEstimate, ...]
    compared: FloatArray
    time_offset: CheckTimeOffset | None = None
    notes: tuple[str, ...] = ()
    deskew: Literal["gyro", "constant_velocity", "none"] | None = None
    """LiDAR pairs: how the scans were deskewed (``none`` = rigid scans)."""
    focal: FocalRecord | None = None
    """camera-focal: the focal-length estimates (the pair is judged on these, not on axes)."""

    @property
    def evidence_from_cache(self) -> bool | None:
        """Whether every estimator artifact came from the cache; ``None`` when none is cacheable."""

        flags = [item.from_cache for item in self.artifacts if item.from_cache is not None]
        return all(flags) if flags else None


@dataclass(frozen=True)
class RunControls:
    """Runtime controls shared by the adapters."""

    max_duration_s: float | None = None
    gnss_max_duration_s: float | None = None
    """Seconds analysed by the GNSS pairs (more than imu-lidar needs); ``None``: max_duration_s."""
    camera: str | None = None
    imu_lidar_translation: bool = True
    acceleration_unit: Literal["mps2", "g"] = "mps2"
    imu_lidar_deskew: Literal["auto", "gyro", "none"] = "auto"
    rigid_scan_rotation_floor_deg: float = RIGID_SCAN_ROTATION_FLOOR_DEG
    lidar_lidar_deskew: Literal["auto", "constant_velocity", "none"] = "auto"
    gnss_lidar_deskew: Literal["auto", "constant_velocity", "none"] = "auto"
    camera_lidar_motion: Literal["auto", "none"] = "auto"
    frame_tree: StaticFrameTree | None = None
    """The candidate tree (static transforms between sensor frames), when known."""
    progress: Callable[[str], None] = field(default=lambda message: None)
    cache: EstimatorCache | None = None
    bag_sha256: str = ""
    scan_store: ScanStore | None = None
    memo: dict[tuple[object, ...], Any] = field(default_factory=dict)
    """Per-run results shared by pairs (the LiDAR odometry of the vehicle pairs)."""
    camera_lidar_start_scan_min_control_t: float | None = None


@dataclass(frozen=True)
class PairContext:
    """Everything an adapter needs for one planned pair."""

    bag: Path
    pair: CheckPairRecord
    topic_types: Mapping[str, str]
    sensor_topics: Mapping[str, tuple[str, ...]]
    candidate: FloatArray
    sources: Sequence[LoadedSource]
    controls: RunControls
    topics: Sequence[CheckTopicRecord] = ()


Adapter = Callable[[PairContext], EstimatorRun]


# ------------------------------------------------------------------ conversions


def transform_matrix(
    translation_m: Sequence[float], quaternion_xyzw: Sequence[float]
) -> FloatArray:
    """Build a 4x4 transform from a translation and an xyzw quaternion."""

    from scipy.spatial.transform import Rotation

    matrix = np.eye(4)
    matrix[:3, :3] = Rotation.from_quat(np.asarray(quaternion_xyzw, dtype=np.float64)).as_matrix()
    matrix[:3, 3] = np.asarray(translation_m, dtype=np.float64)
    return matrix


def invert_transform(matrix: FloatArray) -> FloatArray:
    """Invert a rigid transform."""

    inverse = np.eye(4)
    inverse[:3, :3] = matrix[:3, :3].T
    inverse[:3, 3] = -matrix[:3, :3].T @ matrix[:3, 3]
    return inverse


def rotation_errors_deg(candidate: FloatArray, estimate: FloatArray) -> tuple[float, float, float]:
    """Return ``rotvec(R_candidate R_estimate^T)`` in degrees (about parent-frame axes)."""

    from scipy.spatial.transform import Rotation

    local = Rotation.from_matrix(candidate @ estimate.T).as_rotvec()
    return (
        math.degrees(float(local[0])),
        math.degrees(float(local[1])),
        math.degrees(float(local[2])),
    )


def _estimate(
    name: CheckAxisName,
    unit: Literal["deg", "m"],
    error: float,
    record: Any,
) -> AxisEstimate:
    observable = record is not None and record.status == "estimated"
    excitation = getattr(record, "excitation", None)
    control = getattr(record, "known_bad_control", None)
    undetected = observable and control is not None and not control.detected
    if undetected and control is not None:
        amount = getattr(control, "amount", None)
        control_unit = getattr(control, "unit", None)
        if amount is None:
            amount, control_unit = control.amount_deg, "deg"
        return AxisEstimate(
            name=name,
            unit=unit,
            candidate_error=error,
            std=float(record.std_reported),
            estimated=False,
            unchecked_reason=(
                f"the estimator's known-bad control on this axis ({amount:g} "
                f"{control_unit}) was not detected on held-out data "
                f"(delta chi-square {control.holdout_delta_chi2:.1f}), so the estimate "
                "is not trusted as a yardstick"
            ),
            unchecked_code="control_not_detected",
            excitation=excitation,
        )
    return AxisEstimate(
        name=name,
        unit=unit,
        candidate_error=error,
        std=float(record.std_reported) if record is not None else 0.0,
        estimated=observable,
        unchecked_code=None
        if observable
        else ("unobservable" if record is not None else "no_estimate"),
        unchecked_reason=None
        if observable
        else (
            (
                "constrained only by a declared prior, not by the data"
                if getattr(record, "status", None) == "prior"
                else f"not constrained by the data (reported std {record.std_reported:.4g} {unit})"
            )
            if record is not None
            else "the estimator reported no value"
        ),
        excitation=excitation,
    )


def rotation_axis_estimates(
    records: Sequence[Any],
    candidate_rotation: FloatArray,
    estimate_rotation: FloatArray,
) -> list[AxisEstimate]:
    """Roll/pitch/yaw estimates from DoF records and the two rotations."""

    errors = rotation_errors_deg(candidate_rotation, estimate_rotation)
    by_name = {record.name: record for record in records}
    names: tuple[CheckAxisName, ...] = ("roll", "pitch", "yaw")
    return [
        _estimate(name, "deg", errors[index], by_name.get(name)) for index, name in enumerate(names)
    ]


def translation_axis_estimates(
    records: Sequence[Any],
    candidate_translation: Sequence[float],
    estimate_translation: Sequence[float],
) -> list[AxisEstimate]:
    """x/y/z estimates from axis records and the two translations."""

    by_name = {record.name: record for record in records}
    names: tuple[CheckAxisName, ...] = ("x", "y", "z")
    return [
        _estimate(
            name,
            "m",
            float(candidate_translation[index]) - float(estimate_translation[index]),
            by_name.get(name),
        )
        for index, name in enumerate(names)
    ]


def time_offset_of(records: Sequence[Any], estimate_s: float) -> CheckTimeOffset | None:
    """The clock-offset record of a rotation artifact."""

    for record in records:
        if record.name == "time_offset":
            return CheckTimeOffset(
                estimate_s=float(estimate_s),
                std_s=float(record.std_reported),
                status="estimated" if record.status == "estimated" else "unobservable",
            )
    return None


def _pose_rotation(artifact: Any) -> FloatArray:
    from scipy.spatial.transform import Rotation

    return np.asarray(Rotation.from_quat(artifact.rotation_quat_xyzw).as_matrix(), dtype=np.float64)


# ------------------------------------------------------------ point-time field


def detect_point_time(bag: Path, topic: str) -> tuple[str, PointTimeEncoding] | None:
    """Find a per-point time field of a PointCloud2 topic and how to read it.

    Looks at the first message: picks the first of the usual field names and
    classifies its values against the header stamp as an offset in seconds, an
    absolute time in seconds or an absolute time in nanoseconds. Returns
    ``None`` when there is no usable field.
    """

    from calibrex.data import ros_cdr
    from calibrex.data.rosbag2 import iter_topic_messages

    for _conn, timestamp_ns, payload in iter_topic_messages(bag, topic, limit=1):
        cloud = ros_cdr.decode_ros2_pointcloud2(topic, timestamp_ns, payload)
        names = {point_field.name for point_field in cloud.fields}
        for candidate in POINT_TIME_FIELD_CANDIDATES:
            if candidate not in names:
                continue
            decoded = ros_cdr.decode_ros2_pointcloud2(
                topic, timestamp_ns, payload, point_time_field=candidate
            )
            encoding = classify_point_time(decoded.point_time_offsets_s, decoded.timestamp_ns)
            if encoding is not None:
                return candidate, encoding
        return None
    return None


ABSOLUTE_POINT_TIME_WINDOW_S = 5.0
"""Largest distance from the header stamp at which point times count as absolute."""


def classify_point_time(values: Any, header_ns: int) -> PointTimeEncoding | None:
    """Classify decoded per-point time values against the header stamp (ns)."""

    if values is None:
        return None
    array = np.asarray(values, dtype=np.float64)
    array = array[np.isfinite(array)]
    if array.size == 0:
        return None
    median = float(np.median(array))
    header_s = header_ns * 1.0e-9
    # An absolute point time lies within a sweep (and some latency) of its
    # header.  A wider window misreads offsets as absolute times when the
    # header runs on a sensor's since-boot clock (an Ouster OS1 ~750 s after
    # boot stamps headers near 750 s while its offsets stay below 0.1 s).
    if header_s > 1.0 and abs(median - header_s) < ABSOLUTE_POINT_TIME_WINDOW_S:
        return "absolute_s"
    if header_ns > 0 and abs(median - header_ns) < ABSOLUTE_POINT_TIME_WINDOW_S * 1.0e9:
        return "absolute_ns"
    if float(np.max(np.abs(array))) < 10.0:
        return "offset_s"
    return None


RIGID_SCAN_OBSERVABLE_ROTATION_STD_DEG = 0.3
"""Largest reported rotation std of a rigid-scan axis still called estimated.

The gyro-mode default is 0.1 deg (a fifth of the 0.5 deg floor); the same ratio
to the 1.5 deg rigid-scan floor gives 0.3 deg. Fixed (not tied to the floor
option) so changing the floor does not invalidate the cached estimate.
"""
RIGID_SCAN_CONTROL_DEG = 1.5
"""Size of the estimator's own known-bad rotation control on rigid scans.

The control must detect a shift as large as the tolerance floor, because that is
the smallest error the verdict can resolve there (1 deg in gyro mode, whose floor
is 0.5 deg, but not resolvable on rigid scans).
"""


def imu_lidar_run_options(deskew: Literal["gyro", "none"]) -> ImuLidarRunOptions:
    """Estimator options of the imu-lidar adapter for a deskew mode."""

    from calibrex.evaluation.imu_lidar_rotation import ImuLidarRunOptions
    from calibrex.solvers.imu_lidar_rotation_solver import RotationOptions

    if deskew == "gyro":
        return ImuLidarRunOptions()
    return ImuLidarRunOptions(
        deskew="none",
        rotation_control_deg=RIGID_SCAN_CONTROL_DEG,
        solver=RotationOptions(observable_rotation_std_deg=RIGID_SCAN_OBSERVABLE_ROTATION_STD_DEG),
    )


def select_imu_lidar_deskew(
    bag: Path, topic: str, mode: Literal["auto", "gyro", "none"]
) -> tuple[Literal["gyro", "none"], tuple[str, PointTimeEncoding] | None]:
    """Choose how imu-lidar deskews ``topic``: per-point time (gyro) or rigid scans (none).

    ``auto`` uses the per-point time field whenever the cloud has one and falls
    back to rigid scans only when it has none; ``none`` forces rigid scans;
    ``gyro`` requires the field and skips the pair without it.
    """

    if mode == "none":
        return "none", None
    spec = detect_point_time(bag, topic)
    if spec is not None:
        return "gyro", spec
    if mode == "gyro":
        raise CheckSkipError(
            "unsupported_sensor",
            f"{topic} has no usable per-point time field, so its scans cannot be deskewed "
            "(--imu-lidar-deskew none treats them as rigid scans)",
        )
    return "none", None


def select_odometry_deskew(
    bag: Path,
    topic: str,
    mode: Literal["auto", "constant_velocity", "none"],
    flag: str,
) -> tuple[Literal["constant_velocity", "none"], tuple[str, PointTimeEncoding] | None]:
    """Choose how a LiDAR-odometry pair treats ``topic``: per-point-time deskew or rigid scans.

    ``auto`` uses the per-point time field whenever the cloud has one and falls
    back to rigid scans only when it has none; ``none`` forces rigid scans;
    ``constant_velocity`` requires the field and skips the pair without it.
    """

    if mode == "none":
        return "none", None
    spec = detect_point_time(bag, topic)
    if spec is not None:
        return "constant_velocity", spec
    if mode == "constant_velocity":
        raise CheckSkipError(
            "unsupported_sensor",
            f"{topic} has no usable per-point time field, so its scans cannot be deskewed "
            f"({flag} none treats them as rigid scans)",
        )
    return "none", None


def _require_pointcloud2(ctx: PairContext, topics: Sequence[str]) -> str:
    for topic in topics:
        if ctx.topic_types.get(topic) == POINTCLOUD2_TYPE:
            return topic
    raise CheckSkipError(
        "unsupported_sensor",
        f"LiDAR topic(s) {', '.join(topics)} are not sensor_msgs/PointCloud2 "
        "(Livox CustomMsg is not supported by the check yet)",
    )


# ----------------------------------------------------------------- imu-lidar


def _require_gyro_rate(ctx: PairContext, imu_topic: str) -> None:
    """Skip imu-lidar when the IMU is too slow for the estimator's gyro coverage rule."""

    from calibrex.data.livox_ros2 import LivoxStreamProfile, load_livox_imu
    from calibrex.solvers.imu_lidar_rotation_solver import MAX_GYRO_GAP_S

    profile = LivoxStreamProfile(
        name=f"imu-rate:{imu_topic}",
        point_topic="",
        imu_topic=imu_topic,
        point_time_field=None,
        point_time_encoding="offset_s",
        acceleration_unit=ctx.controls.acceleration_unit,
    )
    store = ctx.controls.scan_store
    samples = store.imu(ctx.bag, profile) if store is not None else load_livox_imu(ctx.bag, profile)
    if len(samples.times_s) < 2:
        return
    interval = float(np.median(np.diff(samples.times_s)))
    if interval > MAX_GYRO_GAP_S:
        raise CheckSkipError(
            "unsupported_sensor",
            f"{imu_topic} runs at about {1.0 / interval:.1f} Hz (median interval "
            f"{interval * 1e3:.0f} ms); imu-lidar compares LiDAR rotation with the gyro over "
            f"each scan interval and needs gyro samples at least every {MAX_GYRO_GAP_S * 1e3:.0f} "
            f"ms (>= {1.0 / MAX_GYRO_GAP_S:.0f} Hz), so a slower IMU covers no interval "
            "(KITTI's OXTS unit is 10 Hz)",
        )


def run_imu_lidar(ctx: PairContext) -> EstimatorRun:
    """Rotation (and optionally lever arm) of ``T_lidar_imu`` from angular rates."""

    from calibrex.core.imu_lidar_rotation import load_imu_lidar_rotation
    from calibrex.core.imu_lidar_translation import load_imu_lidar_translation
    from calibrex.data.livox_ros2 import LivoxStreamProfile
    from calibrex.evaluation.imu_lidar_rotation import (
        rebase_rotation_reference,
        run_livox_imu_lidar_rotation,
    )
    from calibrex.evaluation.imu_lidar_translation import (
        ImuLidarTranslationOptions,
        rebase_translation_reference,
        run_livox_imu_lidar_translation,
    )

    lidar_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("lidar", ()))
    imu_topics = ctx.sensor_topics.get("imu", ())
    if not imu_topics:
        raise CheckSkipError("missing_topic", "no IMU topic for the pair")
    imu_topic = imu_topics[0]
    _require_gyro_rate(ctx, imu_topic)
    deskew, point_time = select_imu_lidar_deskew(
        ctx.bag, lidar_topic, ctx.controls.imu_lidar_deskew
    )
    field_name: str | None = None
    encoding: PointTimeEncoding = "offset_s"
    if point_time is not None:
        field_name, encoding = point_time
    profile = LivoxStreamProfile(
        name=f"ros2-pointcloud2:{lidar_topic}+imu:{imu_topic}",
        point_topic=lidar_topic,
        imu_topic=imu_topic,
        point_time_field=field_name,
        point_time_encoding=encoding,
        acceleration_unit=ctx.controls.acceleration_unit,
    )
    rigid = deskew == "none"
    candidate = invert_transform(ctx.candidate)  # T_lidar_imu
    command = ["calibrex", "check", str(ctx.bag), "imu-lidar", lidar_topic, imu_topic]
    reference = "deployed candidate calibration (T_lidar_imu)"
    ctx.controls.progress(f"imu-lidar: rotation on {lidar_topic} + {imu_topic}")
    # The estimate does not depend on the candidate (it is only compared with it
    # afterwards), so it is cached and the reference fields are rebased on a hit.
    rotation_options = {
        "profile": profile,
        "max_seconds": ctx.controls.max_duration_s,
        "options": imu_lidar_run_options(deskew),
    }
    rotation, rotation_cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "imu_lidar_rotation",
        rotation_options,
        loader=load_imu_lidar_rotation,
        compute=lambda: run_livox_imu_lidar_rotation(
            [ctx.bag],
            profile,
            imu_lidar_run_options(deskew),
            dataset_family=DATASET_FAMILY,
            dataset_license=DATASET_LICENSE,
            reference_rotation=candidate[:3, :3],
            max_seconds=ctx.controls.max_duration_s,
            reference=reference,
            command=command,
            scan_store=ctx.controls.scan_store,
        ),
        rebase=lambda hit: rebase_rotation_reference(hit, candidate[:3, :3]),
    )
    if rigid:
        notes = [
            "deskew none: the cloud has no per-point time field"
            if ctx.controls.imu_lidar_deskew == "auto"
            else "deskew none: forced by --imu-lidar-deskew none",
            "scans are treated as rigid snapshots (no motion correction); rotation axes use a "
            f"floor of {ctx.controls.rigid_scan_rotation_floor_deg:g} deg and the time offset "
            "includes the driver's scan-stamp convention (about half a sweep)",
        ]
    else:
        notes = [f"per-point time: field '{field_name}' read as {encoding}"]
    cache_flag = rotation_cached if ctx.controls.cache is not None else None
    artifacts = [
        EvidenceArtifact("rotation", rotation, rotation.policy_status, cache_flag),
    ]
    estimates: list[AxisEstimate] = []
    solved = rotation.rotation_quat_xyzw is not None and rotation.solver_status == "converged"
    if rotation.rotation_quat_xyzw is not None:
        estimates += rotation_axis_estimates(
            rotation.dofs, candidate[:3, :3], _pose_rotation(rotation)
        )
        if rigid:
            estimates = [
                replace(item, floor=ctx.controls.rigid_scan_rotation_floor_deg)
                for item in estimates
            ]
    else:
        estimates += rotation_axis_estimates(rotation.dofs, candidate[:3, :3], candidate[:3, :3])
        estimates = [_unestimated(item) for item in estimates]
    policy = rotation.policy_status
    reasons = list(rotation.policy_reasons)
    if ctx.controls.imu_lidar_translation and solved:
        ctx.controls.progress("imu-lidar: lever arm (translation) pass")
        # The lever arm is solved from the rotation estimate (not the candidate),
        # so its key is the rotation's key plus the translation options.
        translation_options = {
            **rotation_options,
            "rotation_estimator": "imu_lidar_rotation",
            "translation_options": ImuLidarTranslationOptions(deskew=deskew),
        }
        translation, translation_cached = cached_artifact(
            ctx.controls.cache,
            ctx.controls.bag_sha256,
            "imu_lidar_translation",
            translation_options,
            loader=load_imu_lidar_translation,
            compute=lambda: run_livox_imu_lidar_translation(
                [ctx.bag],
                profile,
                rotation,
                ImuLidarTranslationOptions(deskew=deskew),
                dataset_family=DATASET_FAMILY,
                dataset_license=DATASET_LICENSE,
                reference_translation=candidate[:3, 3],
                max_seconds=ctx.controls.max_duration_s,
                reference=reference,
                command=[*command, "translation"],
                scan_store=ctx.controls.scan_store,
            ),
            rebase=lambda hit: rebase_translation_reference(hit, candidate[:3, 3]),
        )
        artifacts.append(
            EvidenceArtifact(
                "translation",
                translation,
                translation.policy_status,
                translation_cached if ctx.controls.cache is not None else None,
            )
        )
        notes.append(f"translation estimator policy: {translation.policy_status}")
        if translation.translation_m is not None:
            translation_estimates = translation_axis_estimates(
                translation.axes, candidate[:3, 3], translation.translation_m
            )
            if translation.policy_status == "fail":
                reason = "the lever-arm estimate failed its held-out check: " + "; ".join(
                    translation.policy_reasons
                )
                translation_estimates = [
                    _unestimated(item, reason) for item in translation_estimates
                ]
            estimates += translation_estimates
    elif ctx.controls.imu_lidar_translation:
        notes.append("translation not estimated: the rotation was not solved")
    return EstimatorRun(
        estimator=rotation.method,
        artifacts=tuple(artifacts),
        solved=solved,
        policy_status=policy,
        policy_reasons=tuple(reasons),
        estimates=tuple(estimates),
        compared=candidate,
        time_offset=time_offset_of(rotation.dofs, rotation.time_offset_s),
        notes=tuple(notes),
        deskew=deskew,
    )


def _unestimated(
    item: AxisEstimate, reason: str = "the estimator produced no rotation"
) -> AxisEstimate:
    return AxisEstimate(
        item.name, item.unit, item.candidate_error, item.std, False, reason, "no_estimate"
    )


# --------------------------------------------------------------- lidar-lidar


LIDAR_LIDAR_RIGID_ROTATION_FLOOR_DEG = 0.5
"""Rotation floor of lidar-lidar on rigid reference scans.

Measured rigid-minus-deskewed bias on NTU VIRAL tnp_01 was at most 0.07 deg and the
reported stds did not grow, so the observability threshold (0.1 deg) and the
known-bad controls keep their default values. Largest bias + 3 x the largest
rigid std (0.106 deg) = 0.39 deg, rounded up to 0.5 deg, which is the default floor.
"""
LIDAR_LIDAR_RIGID_TRANSLATION_FLOOR_M = 0.03
"""Translation floor of lidar-lidar on rigid reference scans.

Largest measured bias 4.9 mm + 3 x the largest rigid std (8.4 mm) = 0.030 m.
"""


def lidar_lidar_run_options(deskew: Literal["constant_velocity", "none"]) -> Any:
    """Estimator options of the lidar-lidar adapter for a deskew mode."""

    from calibrex.evaluation.lidar_lidar_map import LidarLidarMapOptions

    return LidarLidarMapOptions(deskew=deskew)


def _rigid_note(mode: str, flag: str, topic: str) -> str:
    why = (
        f"the cloud {topic} has no per-point time field"
        if mode == "auto"
        else f"forced by {flag} none"
    )
    return (
        f"deskew none: {why}; scans are treated as rigid snapshots (no motion correction) "
        "and the axes use the rigid-scan floors"
    )


def run_lidar_lidar(ctx: PairContext) -> EstimatorRun:
    """``T_reference_target`` by map registration, started from the candidate."""

    from calibrex.evaluation.lidar_lidar_map import run_ros2_lidar_lidar_map

    reference_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("first", ()))
    target_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("second", ()))
    # Only the reference odometry deskews (the target scans are registered as given),
    # so only the reference cloud needs a per-point time field.
    deskew, ref_time = select_odometry_deskew(
        ctx.bag, reference_topic, ctx.controls.lidar_lidar_deskew, "--lidar-lidar-deskew"
    )
    rigid = deskew == "none"
    ref_field: str | None = None
    ref_encoding: PointTimeEncoding = "offset_s"
    if ref_time is not None:
        ref_field, ref_encoding = ref_time
    options = lidar_lidar_run_options(deskew)
    ctx.controls.progress(f"lidar-lidar: {reference_topic} -> {target_topic}")
    artifact = run_ros2_lidar_lidar_map(
        ctx.bag,
        reference_topic=reference_topic,
        target_topic=target_topic,
        initial=ctx.candidate,
        point_time_field=ref_field,
        point_time_encoding=ref_encoding,
        # the target's own time field is never used: it is read as the reference's
        target_point_time_field=None,
        target_point_time_encoding=ref_encoding,
        options=options,
        dataset_family=DATASET_FAMILY,
        dataset_license=DATASET_LICENSE,
        reference_transform=ctx.candidate,
        reference="deployed candidate calibration (T_reference_target)",
        max_seconds=ctx.controls.max_duration_s,
        command=["calibrex", "check", str(ctx.bag), "lidar-lidar", reference_topic, target_topic],
    )
    notes = [
        _rigid_note(ctx.controls.lidar_lidar_deskew, "--lidar-lidar-deskew", reference_topic)
        if rigid
        else f"per-point time: {reference_topic} field '{ref_field}' ({ref_encoding}) deskews "
        "the reference odometry; the target scans are registered as given",
        "the solver starts at the candidate; a candidate far outside the capture range "
        "of the registration is reported through the held-out residual, not a large delta",
    ]
    estimates: list[AxisEstimate] = []
    solved = artifact.transform is not None and artifact.solver_status != "insufficient_samples"
    if artifact.transform is not None:
        estimate = transform_matrix(
            artifact.transform.translation_m, artifact.transform.rotation_quat_xyzw
        )
        estimates += rotation_axis_estimates(artifact.dofs, ctx.candidate[:3, :3], estimate[:3, :3])
        estimates += translation_axis_estimates(
            [r for r in artifact.dofs if r.name in {"x", "y", "z"}],
            ctx.candidate[:3, 3],
            estimate[:3, 3],
        )
        if rigid:
            estimates = [
                replace(
                    item,
                    floor=LIDAR_LIDAR_RIGID_ROTATION_FLOOR_DEG
                    if item.unit == "deg"
                    else LIDAR_LIDAR_RIGID_TRANSLATION_FLOOR_M,
                )
                for item in estimates
            ]
    if artifact.reference_holdout_delta_chi2 is not None:
        notes.append(
            "held-out chi-square increase when the candidate replaces the estimate: "
            f"{artifact.reference_holdout_delta_chi2:.1f}"
        )
    return EstimatorRun(
        estimator=artifact.method,
        artifacts=(EvidenceArtifact("map_registration", artifact, artifact.policy_status),),
        solved=solved,
        policy_status=artifact.policy_status,
        policy_reasons=tuple(artifact.policy_reasons),
        estimates=tuple(estimates),
        compared=ctx.candidate,
        notes=tuple(notes),
        deskew=deskew,
    )


# ---------------------------------------------------------------- camera-imu


def camera_model_from_camera_info(bag: Path, topic: str, *, rectified: bool = False) -> CameraModel:
    """Pinhole intrinsics and distortion from the first message of a CameraInfo topic.

    With ``rectified`` (the image topic is an ``image_rect*`` stream), the image has
    already been undistorted: its camera is the projection ``P`` without distortion, the
    ROS image_pipeline convention, not ``K`` and ``D``, which describe the raw image.
    """

    from calibrex.data import ros_cdr
    from calibrex.data.rosbag2 import iter_topic_messages
    from calibrex.evaluation.visual_rotation import CameraModel

    for _conn, timestamp_ns, payload in iter_topic_messages(bag, topic, limit=1):
        info = ros_cdr.decode_ros2_camera_info(topic, timestamp_ns, payload)
        k = info.k
        if len(k) != 9 or not k[0] > 0.0 or not k[4] > 0.0:
            raise CheckSkipError(
                "missing_intrinsics", f"CameraInfo {topic} is uncalibrated (K is zero)"
            )
        if rectified:
            p = info.p
            if len(p) != 12 or not p[0] > 0.0 or not p[5] > 0.0:
                raise CheckSkipError(
                    "missing_intrinsics", f"CameraInfo {topic} has no projection P for {topic}"
                )
            return CameraModel(p[0], p[5], p[2], p[6], "none", ())
        coeffs = tuple(float(value) for value in info.d)
        model = info.distortion_model.lower()
        distortion: Literal["equidistant", "radtan", "none"]
        if not coeffs or not any(coeffs):
            distortion = "none"
        elif model in {"equidistant", "fisheye"}:
            distortion = "equidistant"
            coeffs = coeffs[:4]
        elif model in {"plumb_bob", "rational_polynomial", ""}:
            distortion = "radtan"
        else:
            raise CheckSkipError(
                "missing_intrinsics",
                f"CameraInfo {topic}: unsupported model {info.distortion_model!r}",
            )
        return CameraModel(k[0], k[4], k[2], k[5], distortion, coeffs)
    raise CheckSkipError("missing_intrinsics", f"CameraInfo topic {topic} has no messages")


def is_rectified_image_topic(topic: str) -> bool:
    """Whether an image topic carries rectified images (image_pipeline's ``image_rect*``)."""

    return topic.rstrip("/").rsplit("/", 1)[-1].startswith("image_rect")


def _camera_info_topic(ctx: PairContext, image_topic: str) -> str | None:
    folder = image_topic.rsplit("/", 1)[0]
    candidates = sorted(
        topic for topic, message_type in ctx.topic_types.items() if message_type == CAMERA_INFO_TYPE
    )
    for topic in candidates:
        if topic.rsplit("/", 1)[0] == folder:
            return topic
    return None


def resolve_camera_intrinsics(
    ctx: PairContext, image_topic: str, camera_frame: str
) -> tuple[CameraModel, str, str]:
    """Return ``(model, source description, camera key or '')`` or raise a skip."""

    from calibrex.evaluation.visual_rotation import CameraModel

    for source in ctx.sources:
        if source.path is None or source.kind not in {"kalibr_camchain", "rtk_slam_calib"}:
            continue
        entries = camchain_entries(source.path)
        for key, frame, entry in entries:
            if frame == camera_frame and "intrinsics" in entry:
                try:
                    model = CameraModel.from_kalibr(entry)
                except ValueError as exc:
                    raise CheckSkipError(
                        "missing_intrinsics", f"{source.path.name} {key}: {exc}"
                    ) from exc
                return model, f"{source.path.name} {key}", key
    info_topic = _camera_info_topic(ctx, image_topic)
    if info_topic is not None:
        rectified = is_rectified_image_topic(image_topic)
        return (
            camera_model_from_camera_info(ctx.bag, info_topic, rectified=rectified),
            f"CameraInfo {info_topic}" + (" (P: the image is rectified)" if rectified else ""),
            "",
        )
    raise CheckSkipError(
        "missing_intrinsics",
        f"no intrinsics for {image_topic}: give a Kalibr camchain with --tf or record a "
        "CameraInfo topic next to the image topic",
    )


def run_camera_imu(ctx: PairContext) -> EstimatorRun:
    """``T_cam_imu`` rotation from tracked image features against the gyro."""

    if importlib.util.find_spec("cv2") is None:
        raise CheckSkipError(
            "missing_dependency", "camera-imu needs OpenCV: pip install 'calibrex[opencv]'"
        )
    from calibrex.core.imu_lidar_rotation import load_imu_lidar_rotation
    from calibrex.evaluation.camera_imu_rotation import (
        CameraImuRunOptions,
        run_ros2_camera_imu_rotation,
    )
    from calibrex.evaluation.imu_lidar_rotation import rebase_rotation_reference

    camera_topics = ctx.sensor_topics.get("camera", ())
    image_topic = next((t for t in camera_topics if ctx.topic_types.get(t) == IMAGE_TYPE), None)
    if image_topic is None:
        raise CheckSkipError(
            "unsupported_sensor",
            f"camera topic(s) {', '.join(camera_topics)} are not sensor_msgs/Image "
            "(compressed images are not supported by the check yet)",
        )
    imu_topics = ctx.sensor_topics.get("imu", ())
    if not imu_topics:
        raise CheckSkipError("missing_topic", "no IMU topic for the pair")
    camera_frame = ctx.pair.frames[0]
    model, intrinsics_source, _key = resolve_camera_intrinsics(ctx, image_topic, camera_frame)
    ctx.controls.progress(
        f"camera-imu: {image_topic} + {imu_topics[0]} (intrinsics: {intrinsics_source})"
    )
    candidate = ctx.candidate  # T_cam_imu
    run_options = CameraImuRunOptions(max_seconds=ctx.controls.max_duration_s)
    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "camera_imu_rotation",
        {
            "image_topic": image_topic,
            "imu_topic": imu_topics[0],
            "camera": model,
            "acceleration_unit": ctx.controls.acceleration_unit,
            "options": run_options,
        },
        loader=load_imu_lidar_rotation,
        compute=lambda: run_ros2_camera_imu_rotation(
            [ctx.bag],
            image_topic=image_topic,
            imu_topic=imu_topics[0],
            camera=model,
            dataset_family=DATASET_FAMILY,
            dataset_license=DATASET_LICENSE,
            acceleration_unit=ctx.controls.acceleration_unit,
            reference_rotation=candidate[:3, :3],
            reference="deployed candidate calibration (T_cam_imu)",
            options=run_options,
            command=["calibrex", "check", str(ctx.bag), "camera-imu", image_topic, imu_topics[0]],
        ),
        rebase=lambda hit: rebase_rotation_reference(hit, candidate[:3, :3]),
    )
    solved = artifact.rotation_quat_xyzw is not None and artifact.solver_status == "converged"
    if artifact.rotation_quat_xyzw is not None:
        estimates = rotation_axis_estimates(
            artifact.dofs, candidate[:3, :3], _pose_rotation(artifact)
        )
    else:
        estimates = [
            _unestimated(item)
            for item in rotation_axis_estimates(artifact.dofs, candidate[:3, :3], candidate[:3, :3])
        ]
    return EstimatorRun(
        estimator=f"{artifact.method} (camera)",
        artifacts=(
            EvidenceArtifact(
                "rotation",
                artifact,
                artifact.policy_status,
                cached if ctx.controls.cache is not None else None,
            ),
        ),
        solved=solved,
        policy_status=artifact.policy_status,
        policy_reasons=tuple(artifact.policy_reasons),
        estimates=tuple(estimates),
        compared=candidate,
        time_offset=time_offset_of(artifact.dofs, artifact.time_offset_s),
        notes=(f"intrinsics: {intrinsics_source}",),
    )


# --------------------------------------------------------------- camera-focal
#
# camera-focal judges the deployed *intrinsics* (fx, fy), not an extrinsic. The candidate
# is the focal length of the camchain / CameraInfo that camera-imu also uses; the estimate
# is a scale s (f_est = s * f_candidate, the camera/gyro rate ratio measured with features
# normalized by the candidate focal). It consumes this run's camera-imu *estimate* (rotation,
# clock offset, gyro bias), like the standalone `calibrex camera-imu focal` consumes the
# rotation artifact. The candidate extrinsic is not used: a wrong candidate rotation would
# leak rate between camera axes, while the estimated rotation is what the data support. The
# estimate depends on the candidate only through the normalization of the tracked features
# (s absorbs it to first order), so it is cached under the camera model, the camera-imu
# estimate and the options, never under the candidate extrinsic.

FOCAL_RATE_FLOOR_NOTE = (
    "the focal scale is the camera/gyro rate ratio about the camera y (fx) and x (fy) axes; "
    "it also absorbs anything else that scales tracked rotations, so the optical-axis ratio "
    "(which must be 1) is the control"
)
_ROTATION_AXIS_NAMES = ("roll", "pitch", "yaw")
FOCAL_MAX_ROTATION_STD_DEG = 0.5
"""Largest camera-imu rotation std (any axis) for which camera-focal runs.

A rotation error of ``d`` mixes angular rate between camera axes by about ``sin d``
(0.9 % per 0.5 deg at equal rates), the size of the focal-scale floor. Axes are
therefore not required to pass camera-imu's own 0.3 deg observability bound, only
to be known to 0.5 deg; looser rotations make the focal ratio unreliable, so the
pair is skipped instead of judged.
"""


def _camera_imu_dependency(ctx: PairContext) -> tuple[EstimatorRun, Any]:
    """This run's camera-imu result for the same camera and IMU, or a missing_dependency skip."""

    run = ctx.controls.memo.get((PAIR_RUN_KEY, "camera-imu", tuple(ctx.pair.frames)))
    if run is None:
        raise CheckSkipError(
            "missing_dependency",
            "camera-focal reuses this check's camera-imu rotation, clock offset and gyro bias; "
            "camera-imu gave no result for this camera (not selected with --pairs or --camera, "
            "skipped, or failed)",
        )
    rotation = next((i.artifact for i in run.artifacts if i.role == "rotation"), None)
    problems: list[str] = []
    if not run.solved or getattr(rotation, "rotation_quat_xyzw", None) is None:
        problems.append("camera-imu produced no rotation estimate")
    if run.policy_status == "fail":
        problems.append("camera-imu failed its own held-out check")
    loose = [
        f"{item.name} ({item.std:.2f} deg)"
        for item in run.estimates
        if item.name in _ROTATION_AXIS_NAMES
        and not (math.isfinite(item.std) and item.std <= FOCAL_MAX_ROTATION_STD_DEG)
    ]
    if loose:
        problems.append(
            f"camera-imu's rotation std exceeds {FOCAL_MAX_ROTATION_STD_DEG:g} deg about "
            + ", ".join(loose)
        )
    if problems:
        raise CheckSkipError(
            "missing_dependency",
            "camera-focal needs a camera-imu rotation that is solved and known to within "
            f"{FOCAL_MAX_ROTATION_STD_DEG:g} deg about every axis: " + "; ".join(problems),
        )
    return run, rotation


def _rotation_digest(artifact: Any) -> str:
    """SHA-256 of the saved camera-imu rotation artifact (the evidence file's digest)."""

    from calibrex.core.provenance import sha256_path

    with tempfile.TemporaryDirectory() as scratch:
        path = Path(scratch) / "rotation.yaml"
        artifact.save(path)
        digest = sha256_path(path)
    assert digest is not None
    return digest


def run_camera_focal(ctx: PairContext) -> EstimatorRun:
    """Deployed focal lengths against the gyro, on this check's camera-imu estimate."""

    from calibrex.core.camera_focal_scale import (
        CameraFocalScaleArtifact,
        load_camera_focal_scale,
    )
    from calibrex.evaluation.camera_focal import FocalScaleOptions, run_camera_focal_check

    _run, rotation = _camera_imu_dependency(ctx)
    camera_topics = ctx.sensor_topics.get("camera", ())
    image_topic = next((t for t in camera_topics if ctx.topic_types.get(t) == IMAGE_TYPE), None)
    imu_topics = ctx.sensor_topics.get("imu", ())
    if image_topic is None or not imu_topics:
        raise CheckSkipError("missing_topic", "camera-focal needs an Image and an IMU topic")
    model, intrinsics_source, _key = resolve_camera_intrinsics(ctx, image_topic, ctx.pair.frames[0])
    ctx.controls.progress(
        f"camera-focal: {image_topic} + {imu_topics[0]} (intrinsics: {intrinsics_source})"
    )
    focal_options = FocalScaleOptions()
    rotation_digest = _rotation_digest(rotation)
    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "camera_focal_scale",
        {
            "image_topic": image_topic,
            "imu_topic": imu_topics[0],
            "camera": model,
            "frame_stride": 4,
            "max_seconds": ctx.controls.max_duration_s,
            "options": focal_options,
            # the camera-imu estimate, not the candidate: the rotation, clock offset, bias
            "rotation_quat_xyzw": rotation.rotation_quat_xyzw,
            "time_offset_s": rotation.time_offset_s,
            "gyro_bias_rps": rotation.gyro_bias_rps,
        },
        loader=load_camera_focal_scale,
        compute=lambda: run_camera_focal_check(
            str(ctx.bag),
            image_topic=image_topic,
            imu_topic=imu_topics[0],
            camera=model,
            rotation_artifact=rotation,
            rotation_artifact_sha256=rotation_digest,
            dataset_family=DATASET_FAMILY,
            dataset_license=DATASET_LICENSE,
            frame_stride=4,
            max_seconds=ctx.controls.max_duration_s,
            options=focal_options,
            command=["calibrex", "check", str(ctx.bag), "camera-focal", image_topic, imu_topics[0]],
        ),
        rebase=lambda hit: hit.model_copy(
            update={
                "provenance": hit.provenance.model_copy(
                    update={"rotation_artifact_sha256": rotation_digest}
                )
            }
        ),
    )
    assert isinstance(artifact, CameraFocalScaleArtifact)
    return EstimatorRun(
        estimator=f"{artifact.method} (camera focal)",
        artifacts=(
            EvidenceArtifact(
                "focal_scale",
                artifact,
                artifact.policy_status,
                cached if ctx.controls.cache is not None else None,
            ),
        ),
        solved=True,
        policy_status=artifact.policy_status,
        policy_reasons=tuple(artifact.policy_reasons),
        estimates=(),
        compared=ctx.candidate,
        notes=(
            f"intrinsics: {intrinsics_source}",
            "the rotation, clock offset and gyro bias are camera-imu's estimate from this run",
            FOCAL_RATE_FLOOR_NOTE,
        ),
        focal=focal_record_of(artifact, intrinsics_source),
    )


def focal_record_of(artifact: Any, intrinsics_source: str | None = None) -> FocalRecord:
    """The judged quantities of a focal-scale artifact: ``fx`` from the y ratio, ``fy`` from x."""

    bound = float(artifact.options.get("observable_ratio_std", 0.005))
    control_off = artifact.policy_status == "warn"
    ratios = artifact.rate_ratio
    stds = artifact.rate_ratio_std
    components: list[FocalEstimate] = []
    spec: tuple[tuple[Literal["fx", "fy"], int, float, float, float], ...] = (
        ("fx", 1, artifact.fx_used_px, artifact.fx_estimate_px, artifact.fx_std_px),
        ("fy", 0, artifact.fy_used_px, artifact.fy_estimate_px, artifact.fy_std_px),
    )
    for name, index, used, estimate, std_px in spec:
        std = float(stds[index])
        reason: str | None = None
        code: Literal["unobservable", "control_not_detected", "no_estimate"] | None = None
        if control_off:
            code = "control_not_detected"
            reason = (
                "the optical-axis control ratio differs from 1, so something besides the focal "
                "length scales the rotations and the scale is not trusted as a focal length"
            )
        elif std > bound:
            code = "unobservable"
            reason = (
                f"the {'y' if name == 'fx' else 'x'}-axis ratio std {std:.4f} exceeds the "
                f"bound {bound:g}"
            )
        components.append(
            FocalEstimate(
                name=name,
                candidate_px=float(used),
                estimate_px=float(estimate),
                estimate_std_px=float(std_px),
                scale=float(ratios[index]),
                scale_std=std,
                estimated=reason is None,
                unchecked_reason=reason,
                unchecked_code=code,
            )
        )
    return FocalRecord(
        components=tuple(components),
        optical_axis_ratio=float(ratios[2]),
        optical_axis_ratio_std=float(stds[2]),
        intrinsics_source=intrinsics_source,
    )


# ------------------------------------------------------------- vehicle pairs
#
# lidar-vehicle, imu-vehicle, ins-lidar and lidar-wheel_odometry run the estimators
# of the KITTI benchmarks on bag data (see ``vehicle_inputs``). None of them uses
# the candidate: it enters only when the estimate is compared with it, so the
# artifacts are cached. Sweeps are registered as rigid snapshots, like the KITTI
# runners do; a per-point time field is neither required nor used.

ODOMETRY_TYPE = "nav_msgs/msg/Odometry"
TWIST_TYPES = frozenset(
    {
        "geometry_msgs/msg/TwistStamped",
        "geometry_msgs/msg/TwistWithCovarianceStamped",
        *AUTOWARE_VELOCITY_REPORT_TYPES,
    }
)
VEHICLE_PROVENANCE_GENERATOR = "calibrex.check.estimators"
_RIGID_SWEEP_NOTE = (
    "LiDAR sweeps are registered as rigid snapshots; motion within a sweep is not compensated"
)
_PLANAR_NOTE = (
    "The vehicle frame is defined by the motion: it assumes no side slip and no vertical "
    "velocity, and roll needs turns."
)


def _digest_or_topic(ctx: PairContext, *names: str) -> str:
    """The bag digest for artifact provenance (a digest of the topic names without one)."""

    if ctx.controls.bag_sha256:
        return ctx.controls.bag_sha256
    return hashlib.sha256("\0".join(names).encode("utf-8")).hexdigest()


def _lidar_odometry(ctx: PairContext, topic: str) -> Any:
    """LiDAR odometry of a topic, computed once per run and shared by the vehicle pairs."""

    from calibrex.check import vehicle_inputs

    key = ("lidar_odometry", str(ctx.bag), topic, ctx.controls.max_duration_s)
    track = ctx.controls.memo.get(key)
    if track is None:
        ctx.controls.progress(f"vehicle pairs: LiDAR odometry on {topic}")
        track = vehicle_inputs.read_lidar_odometry(
            ctx.bag, topic, max_duration_s=ctx.controls.max_duration_s
        )
        ctx.controls.memo[key] = track
    return track


SEGMENT_BLOCK_STRIDE = 100_000
"""Block-number offset between stream segments, so no block (and no motion) spans two."""


def _lidar_motions(track: Any, *, step: int, block_duration_s: float) -> list[Any]:
    """Blocked LiDAR motions of every odometry segment, pooled as the KITTI runners pool drives."""

    from calibrex.solvers.vehicle_frame_solver import motions_from_poses

    motions: list[Any] = []
    for position, (times, poses) in enumerate(track.segment_slices()):
        motions += motions_from_poses(
            times,
            poses,
            step=step,
            block_duration_s=block_duration_s,
            block_offset=position * SEGMENT_BLOCK_STRIDE,
        )
    return motions


def _segment_note(track: Any) -> tuple[str, ...]:
    count = len(track.segment_slices())
    if count < 2:
        return ()
    return (
        f"the LiDAR stream has {count} segments (gaps over 1 s); odometry and motions are "
        "kept within a segment",
    )


def _rotation_estimates_of(
    records: Sequence[Any], rotation_quat_xyzw: Sequence[float] | None, candidate: FloatArray
) -> list[AxisEstimate]:
    """Roll/pitch/yaw estimates against a candidate rotation of the same convention."""

    if rotation_quat_xyzw is None:
        estimates = rotation_axis_estimates(records, candidate, candidate)
        return [_unestimated(item) for item in estimates]
    from scipy.spatial.transform import Rotation

    estimate = np.asarray(Rotation.from_quat(rotation_quat_xyzw).as_matrix(), dtype=np.float64)
    return rotation_axis_estimates(records, candidate, estimate)


def _vehicle_run(
    artifact: Any,
    *,
    estimator: str,
    cached: bool,
    ctx: PairContext,
    estimates: Sequence[AxisEstimate],
    compared: FloatArray,
    solved: bool,
    time_offset: CheckTimeOffset | None = None,
    notes: Sequence[str] = (),
    role: str = "estimate",
) -> EstimatorRun:
    return EstimatorRun(
        estimator=estimator,
        artifacts=(
            EvidenceArtifact(
                role,
                artifact,
                artifact.policy_status,
                cached if ctx.controls.cache is not None else None,
            ),
        ),
        solved=solved,
        policy_status=artifact.policy_status,
        policy_reasons=tuple(artifact.policy_reasons),
        estimates=tuple(estimates),
        compared=compared,
        time_offset=time_offset,
        notes=tuple(notes),
    )


def run_lidar_vehicle(ctx: PairContext) -> EstimatorRun:
    """``R_vehicle_lidar`` from the non-holonomic motion of the LiDAR odometry."""

    from calibrex import __version__
    from calibrex.core.provenance import git_commit
    from calibrex.core.vehicle_frame_rotation import (
        VehicleFrameProvenance,
        load_vehicle_frame_rotation,
    )
    from calibrex.evaluation.vehicle_frame import (
        VehicleFrameRunOptions,
        build_vehicle_frame_artifact,
        evaluate_vehicle_frame,
    )

    lidar_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("lidar", ()))
    options = VehicleFrameRunOptions()
    candidate = invert_transform(ctx.candidate)  # T_vehicle_lidar

    def compute() -> Any:
        track = _lidar_odometry(ctx, lidar_topic)
        motions = _lidar_motions(
            track, step=options.step_frames, block_duration_s=options.block_duration_s
        )
        evaluation = evaluate_vehicle_frame(motions, options)
        return build_vehicle_frame_artifact(
            evaluation,
            modality="lidar",
            motions=len(motions),
            references=[],
            options=options,
            include_step=True,
            limitations=[
                _PLANAR_NOTE,
                _RIGID_SWEEP_NOTE,
                "The translation is not estimated.",
                *_segment_note(track),
            ],
            provenance=VehicleFrameProvenance(
                generator=VEHICLE_PROVENANCE_GENERATOR,
                generator_version=__version__,
                git_commit=git_commit(),
                command=["calibrex", "check", str(ctx.bag), "lidar-vehicle", lidar_topic],
                dataset_family=DATASET_FAMILY,
                sequence_ids=[ctx.bag.name],
                sensor=lidar_topic,
                input_sha256=_digest_or_topic(ctx, lidar_topic),
                dataset_license=DATASET_LICENSE,
            ),
        )

    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "lidar_vehicle",
        {
            "topic": lidar_topic,
            "max_duration_s": ctx.controls.max_duration_s,
            "options": options,
        },
        loader=load_vehicle_frame_rotation,
        compute=compute,
        rebase=lambda hit: hit,
    )
    return _vehicle_run(
        artifact,
        estimator="vehicle_frame_rotation (lidar)",
        cached=cached,
        ctx=ctx,
        estimates=_rotation_estimates_of(
            artifact.dofs, artifact.rotation_quat_xyzw, candidate[:3, :3]
        ),
        compared=candidate,
        solved=artifact.rotation_quat_xyzw is not None,
        notes=(
            f"LiDAR odometry on {lidar_topic} ({_RIGID_SWEEP_NOTE})",
            "roll needs turns; an axis the drive does not constrain is listed as unchecked",
            "the translation of T_vehicle_lidar is not estimated",
        ),
    )


def _topic_records(ctx: PairContext, slot: str) -> list[CheckTopicRecord]:
    from calibrex.check.planner import slot_of

    return [record for record in ctx.topics if slot_of(record) == slot]


def run_imu_vehicle(ctx: PairContext) -> EstimatorRun:
    """``R_vehicle_imu`` from the body-frame velocity and rates of an INS in the IMU's frame."""

    from calibrex import __version__
    from calibrex.check import vehicle_inputs
    from calibrex.core.provenance import git_commit
    from calibrex.core.vehicle_frame_rotation import (
        VehicleFrameProvenance,
        load_vehicle_frame_rotation,
    )
    from calibrex.evaluation.vehicle_frame import (
        VehicleFrameRunOptions,
        body_motions,
        build_vehicle_frame_artifact,
        evaluate_vehicle_frame,
    )

    imu_frame = ctx.pair.frames[0]
    candidates = [
        record
        for record in _topic_records(ctx, "ins")
        if record.message_type == ODOMETRY_TYPE or record.message_type in TWIST_TYPES
    ]
    if not candidates:
        raise CheckSkipError(
            "missing_topic",
            "imu-vehicle needs the body-frame velocity and angular rate of an INS (a raw IMU "
            "has no velocity): an Odometry or TwistStamped topic of kind 'ins'; name it with "
            "oxts/ins/gnss or pass --topic-kind TOPIC=ins",
        )
    same_frame = [record for record in candidates if record.mapped_frame == imu_frame]
    if not same_frame:
        raise CheckSkipError(
            "unsupported_sensor",
            "the INS velocity topic(s) "
            + ", ".join(f"{r.topic} ({r.mapped_frame})" for r in candidates)
            + f" are not expressed in the IMU frame '{imu_frame}'",
        )
    # Odometry first (its twist is in the child frame by definition), then the busiest.
    same_frame.sort(key=lambda r: (r.message_type != ODOMETRY_TYPE, -(r.message_count or 0)))
    source = same_frame[0]
    options = VehicleFrameRunOptions()
    candidate = invert_transform(ctx.candidate)  # T_vehicle_imu

    def compute() -> Any:
        track = vehicle_inputs.read_twist_track(
            ctx.bag,
            source.topic,
            source.message_type,
            max_duration_s=ctx.controls.max_duration_s,
        )
        if not np.any(np.abs(track.linear_mps) > 1e-9):
            raise CheckSkipError(
                "unsupported_sensor", f"{source.topic} reports zero velocity in every message"
            )
        motions = []
        for position, piece in enumerate(vehicle_inputs.split_at_gaps(track.times_s)):
            motions += body_motions(
                list(track.times_s[piece]),
                list(track.linear_mps[piece]),
                list(track.angular_rps[piece]),
                block_duration_s=options.block_duration_s,
                block_offset=position * SEGMENT_BLOCK_STRIDE,
            )
        evaluation = evaluate_vehicle_frame(motions, options)
        return build_vehicle_frame_artifact(
            evaluation,
            modality="ins",
            motions=len(motions),
            references=[],
            options=options,
            include_step=False,
            limitations=[
                _PLANAR_NOTE,
                f"The velocity and rates are the INS's own navigation solution on {source.topic}, "
                "taken as expressed in the body frame of the IMU.",
                "The translation is not estimated.",
            ],
            provenance=VehicleFrameProvenance(
                generator=VEHICLE_PROVENANCE_GENERATOR,
                generator_version=__version__,
                git_commit=git_commit(),
                command=["calibrex", "check", str(ctx.bag), "imu-vehicle", source.topic],
                dataset_family=DATASET_FAMILY,
                sequence_ids=[ctx.bag.name],
                sensor=source.topic,
                input_sha256=_digest_or_topic(ctx, source.topic),
                dataset_license=DATASET_LICENSE,
            ),
        )

    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "imu_vehicle",
        {
            "topic": source.topic,
            "max_duration_s": ctx.controls.max_duration_s,
            "options": options,
        },
        loader=load_vehicle_frame_rotation,
        compute=compute,
        rebase=lambda hit: hit,
    )
    return _vehicle_run(
        artifact,
        estimator="vehicle_frame_rotation (ins)",
        cached=cached,
        ctx=ctx,
        estimates=_rotation_estimates_of(
            artifact.dofs, artifact.rotation_quat_xyzw, candidate[:3, :3]
        ),
        compared=candidate,
        solved=artifact.rotation_quat_xyzw is not None,
        notes=(
            f"velocity and angular rate from {source.topic} ({source.message_type}), "
            f"taken as expressed in '{imu_frame}'; the IMU topic itself carries no velocity",
            "the translation of T_vehicle_imu is not estimated",
        ),
    )


def run_ins_lidar(ctx: PairContext) -> EstimatorRun:
    """``T_ins_lidar`` by trajectory hand-eye calibration of LiDAR odometry against an INS."""

    from calibrex import __version__
    from calibrex.check import vehicle_inputs
    from calibrex.core.ins_lidar_hand_eye import (
        InsLidarHandEyeProvenance,
        load_ins_lidar_hand_eye,
    )
    from calibrex.core.provenance import git_commit
    from calibrex.evaluation.ins_lidar_hand_eye import (
        InsLidarRunOptions,
        _odometry_summary,
        build_ins_lidar_artifact,
        build_sensor_motions,
        evaluate_ins_lidar_hand_eye,
    )
    from calibrex.solvers.trajectory_hand_eye_solver import ReferenceTrajectory

    lidar_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("lidar", ()))
    ins_topics = [
        topic
        for topic in ctx.sensor_topics.get("ins", ())
        if ctx.topic_types.get(topic) == ODOMETRY_TYPE
    ]
    if not ins_topics:
        raise CheckSkipError(
            "unsupported_sensor",
            "ins-lidar needs an INS pose: a nav_msgs/Odometry topic of kind 'ins' "
            f"(found {', '.join(ctx.sensor_topics.get('ins', ())) or 'none'})",
        )
    ins_topic = ins_topics[0]
    options = InsLidarRunOptions()
    candidate = ctx.candidate  # T_ins_lidar

    def compute() -> Any:
        ins = vehicle_inputs.read_ins_track(
            ctx.bag, ins_topic, max_duration_s=ctx.controls.max_duration_s
        )
        track = _lidar_odometry(ctx, lidar_topic)
        times = list(track.times_s)
        motions = []
        for position, (piece_times, piece_poses) in enumerate(track.segment_slices()):
            motions += build_sensor_motions(
                piece_times,
                piece_poses,
                options,
                block_offset=position * SEGMENT_BLOCK_STRIDE,
                id_prefix=f"segment{position}/",
            )
        reference = ReferenceTrajectory(ins.times_s, ins.poses)
        evaluation = evaluate_ins_lidar_hand_eye(reference, motions, options)
        return build_ins_lidar_artifact(
            evaluation,
            options,
            odometry=_odometry_summary(list(track.registrations), len(times), []),
            provenance=InsLidarHandEyeProvenance(
                generator=VEHICLE_PROVENANCE_GENERATOR,
                generator_version=__version__,
                git_commit=git_commit(),
                command=["calibrex", "check", str(ctx.bag), "ins-lidar", ins_topic, lidar_topic],
                dataset_family=DATASET_FAMILY,
                drive_ids=[ctx.bag.name],
                input_sha256=_digest_or_topic(ctx, ins_topic, lidar_topic),
                dataset_license=DATASET_LICENSE,
            ),
            reference_calibration=None,
            limitations=[
                _RIGID_SWEEP_NOTE + ".",
                "A reference scale and a clock offset are estimated as nuisance parameters "
                "because INS tracks are not metrically exact and their stamps are logging times.",
                "Vehicle motion is close to planar, so the translation along the vertical "
                "rotation axis cannot be observed without a prior.",
                *_segment_note(track),
            ],
        )

    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "ins_lidar",
        {
            "ins_topic": ins_topic,
            "lidar_topic": lidar_topic,
            "max_duration_s": ctx.controls.max_duration_s,
            "options": options,
        },
        loader=load_ins_lidar_hand_eye,
        compute=compute,
        rebase=lambda hit: hit,
    )
    estimates: list[AxisEstimate]
    transform = artifact.transform
    solved = transform is not None and artifact.solver_status == "converged"
    if transform is not None:
        estimates = _rotation_estimates_of(
            artifact.dofs, transform.rotation_quat_xyzw, candidate[:3, :3]
        )
        estimates += translation_axis_estimates(
            [r for r in artifact.dofs if r.name in {"x", "y", "z"}],
            candidate[:3, 3],
            transform.translation_m,
        )
    else:
        estimates = [
            _unestimated(item)
            for item in rotation_axis_estimates(artifact.dofs, candidate[:3, :3], candidate[:3, :3])
        ]
    return _vehicle_run(
        artifact,
        estimator="ins_lidar_hand_eye",
        cached=cached,
        ctx=ctx,
        estimates=estimates,
        compared=candidate,
        solved=solved,
        time_offset=time_offset_of(artifact.dofs, artifact.time_offset_s or 0.0),
        notes=(
            f"INS pose from {ins_topic}, LiDAR odometry on {lidar_topic} ({_RIGID_SWEEP_NOTE})",
            "an axis constrained only by a prior or by nothing is listed as unchecked",
        ),
    )


def run_lidar_wheel(ctx: PairContext) -> EstimatorRun:
    """``R_wheel_lidar`` from LiDAR odometry against wheel speed and yaw rate."""

    from calibrex import __version__
    from calibrex.check import vehicle_inputs
    from calibrex.core.lidar_wheel_odometry import (
        LidarWheelProvenance,
        load_lidar_wheel_odometry,
    )
    from calibrex.core.provenance import git_commit
    from calibrex.evaluation.lidar_wheel import (
        LIMITATIONS,
        LidarWheelRunOptions,
        build_artifact,
        evaluate_lidar_wheel,
    )
    from calibrex.solvers.lidar_wheel_solver import WheelOdometry

    lidar_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("lidar", ()))
    wheel_topics = [
        topic
        for topic in ctx.sensor_topics.get("wheel", ())
        if ctx.topic_types.get(topic) == ODOMETRY_TYPE or ctx.topic_types.get(topic) in TWIST_TYPES
    ]
    if not wheel_topics:
        raise CheckSkipError(
            "unsupported_sensor",
            "lidar-wheel_odometry needs wheel speed and yaw rate from an Odometry or "
            f"TwistStamped topic (found {', '.join(ctx.sensor_topics.get('wheel', ())) or 'none'})",
        )
    wheel_topic = wheel_topics[0]
    wheel_type = ctx.topic_types[wheel_topic]
    options = LidarWheelRunOptions()
    candidate = invert_transform(ctx.candidate)  # T_wheel_lidar

    def compute() -> Any:
        track = _lidar_odometry(ctx, lidar_topic)
        motions = _lidar_motions(
            track, step=options.step_frames, block_duration_s=options.block_duration_s
        )
        twist = vehicle_inputs.read_twist_track(
            ctx.bag, wheel_topic, wheel_type, max_duration_s=ctx.controls.max_duration_s
        )
        wheel = WheelOdometry(twist.times_s, twist.linear_mps[:, 0], twist.angular_rps[:, 2])
        evaluation = evaluate_lidar_wheel(motions, wheel, options)
        return build_artifact(
            evaluation,
            options,
            motions=len(motions),
            references=[],
            provenance=LidarWheelProvenance(
                generator=VEHICLE_PROVENANCE_GENERATOR,
                generator_version=__version__,
                git_commit=git_commit(),
                command=["calibrex", "check", str(ctx.bag), "lidar-wheel", wheel_topic],
                dataset_family=DATASET_FAMILY,
                sequence_ids=[ctx.bag.name],
                wheel_source=f"{wheel_topic} ({wheel_type}): linear.x speed, angular.z yaw rate",
                input_sha256=_digest_or_topic(ctx, wheel_topic, lidar_topic),
                dataset_license=DATASET_LICENSE,
            ),
            limitations=[*LIMITATIONS, _RIGID_SWEEP_NOTE + ".", *_segment_note(track)],
        )

    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "lidar_wheel",
        {
            "wheel_topic": wheel_topic,
            "lidar_topic": lidar_topic,
            "max_duration_s": ctx.controls.max_duration_s,
            "options": options,
        },
        loader=load_lidar_wheel_odometry,
        compute=compute,
        rebase=lambda hit: hit,
    )
    return _vehicle_run(
        artifact,
        estimator="lidar_wheel_odometry",
        cached=cached,
        ctx=ctx,
        estimates=_rotation_estimates_of(
            artifact.parameters, artifact.rotation_quat_xyzw, candidate[:3, :3]
        ),
        compared=candidate,
        solved=artifact.rotation_quat_xyzw is not None,
        time_offset=time_offset_of(artifact.parameters, artifact.time_offset_s or 0.0),
        notes=(
            f"wheel speed (linear.x) and yaw rate (angular.z) from {wheel_topic} ({wheel_type}); "
            f"LiDAR odometry on {lidar_topic} ({_RIGID_SWEEP_NOTE})",
            "only the rotation is judged: the lever, speed scale and clock offset are reported "
            "in the evidence artifact, not compared with the candidate",
        ),
    )


# ------------------------------------------------------------ gnss-lidar / gnss-imu

GNSS_ROTATION_REASON = (
    "the GNSS antenna's orientation is not defined: the lever arm is a point offset, so "
    "the candidate's rotation about the antenna is not judged"
)
PAIR_RUN_KEY = "pair_run"
"""Memo key prefix under which the runner stores each pair's :class:`EstimatorRun`."""


def _unjudged_rotation_axes(reason: str = GNSS_ROTATION_REASON) -> list[AxisEstimate]:
    names: tuple[CheckAxisName, ...] = ("roll", "pitch", "yaw")
    return [AxisEstimate(name, "deg", 0.0, 0.0, False, reason, "unobservable") for name in names]


def _iter_scans_plain(bag: Path, profile: Any, max_seconds: float | None) -> Any:
    from calibrex.data.livox_ros2 import iter_livox_points

    return iter_livox_points(bag, profile, max_seconds=max_seconds)


GNSS_LIDAR_RIGID_TRANSLATION_FLOOR_M = 0.11
"""Translation floor of gnss-lidar (and gnss-imu) on rigid scans.

Largest measured rigid-minus-deskewed bias 41.5 mm (RTK-SLAM stadtgarten_seq2, y)
plus three times the largest rigid std measured on an estimated axis (22 mm)
= 0.108 m, rounded up to 0.11 m (the observability threshold below, 3 x 20 mm, gives
0.1015 m, the same after rounding).
"""
GNSS_LIDAR_RIGID_OBSERVABLE_TRANSLATION_STD_M = 0.02
"""Largest reported std of a rigid-scan lever-arm axis still called estimated.

Twice the deskewed 0.01 m: the rigid stds measured 11 to 22 mm against 5 to 41 mm.
"""
GNSS_LIDAR_RIGID_CONTROL_M = GNSS_LIDAR_RIGID_TRANSLATION_FLOOR_M
"""The estimator's own known-bad shift must be detectable at the size of the floor."""


def gnss_lidar_run_options(deskew: Literal["constant_velocity", "none"]) -> Any:
    """Estimator options of the gnss-lidar adapter for a deskew mode."""

    from dataclasses import replace as dc_replace

    from calibrex.evaluation.gnss_lidar_lever_arm import GnssLidarRunOptions

    options = GnssLidarRunOptions()
    if deskew == "constant_velocity":
        return options
    return dc_replace(
        options,
        deskew="none",
        lever_arm_control_m=GNSS_LIDAR_RIGID_CONTROL_M,
        solver=dc_replace(
            options.solver,
            observable_translation_std_m=GNSS_LIDAR_RIGID_OBSERVABLE_TRANSLATION_STD_M,
        ),
    )


def run_gnss_lidar(ctx: PairContext) -> EstimatorRun:
    """Antenna lever arm in the LiDAR frame from GNSS positions against LiDAR odometry.

    The planner's candidate is ``T_gnss_lidar``; the estimate is the antenna
    position in the LiDAR frame, the translation of ``T_lidar_gnss``. The fit
    does not use the candidate (it is compared afterwards), so it is cached.
    """

    from calibrex import __version__
    from calibrex.core.gnss_lidar_lever_arm import (
        GnssLidarLeverArmArtifact,
        GnssLidarLeverArmProvenance,
        GnssLidarSegmentSummary,
        load_gnss_lidar_lever_arm,
    )
    from calibrex.core.provenance import git_commit
    from calibrex.data.livox_ros2 import LivoxStreamProfile
    from calibrex.data.navsatfix_track import NavSatFixTrackOptions, read_navsatfix_track
    from calibrex.data.rosbag2 import NAVSATFIX_TYPE
    from calibrex.evaluation.gnss_lidar_lever_arm import (
        RIGID_SCAN_LIMITATIONS,
        RTK_SLAM_LIMITATIONS,
        build_gnss_lidar_artifact,
        collect_windows,
        evaluate_gnss_lidar_lever_arm,
        rebase_gnss_lidar_reference,
    )
    from calibrex.solvers.gnss_lever_arm_solver import GnssTrackModel

    gnss_topic = next(
        (t for t in ctx.sensor_topics.get("gnss", ()) if ctx.topic_types.get(t) == NAVSATFIX_TYPE),
        None,
    )
    if gnss_topic is None:
        raise CheckSkipError(
            "unsupported_sensor",
            f"GNSS topic(s) {', '.join(ctx.sensor_topics.get('gnss', ()))} are not "
            "sensor_msgs/NavSatFix",
        )
    lidar_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("lidar", ()))
    deskew, point_time = select_odometry_deskew(
        ctx.bag, lidar_topic, ctx.controls.gnss_lidar_deskew, "--gnss-lidar-deskew"
    )
    rigid = deskew == "none"
    field_name: str | None = None
    encoding: PointTimeEncoding = "offset_s"
    if point_time is not None:
        field_name, encoding = point_time
    profile = LivoxStreamProfile(
        name=f"ros2-pointcloud2:{lidar_topic}",
        point_topic=lidar_topic,
        imu_topic="",
        point_time_field=field_name,
        point_time_encoding=encoding,
        acceleration_unit=ctx.controls.acceleration_unit,
    )
    lever_candidate = invert_transform(ctx.candidate)[:3, 3]  # antenna in the LiDAR frame
    track_options = NavSatFixTrackOptions()
    run_options = gnss_lidar_run_options(deskew)
    max_seconds = (
        ctx.controls.gnss_max_duration_s
        if ctx.controls.gnss_max_duration_s is not None
        else ctx.controls.max_duration_s
    )
    ctx.controls.progress(f"gnss-lidar: {gnss_topic} against LiDAR odometry on {lidar_topic}")

    def compute() -> GnssLidarLeverArmArtifact:
        try:
            track, summary = read_navsatfix_track(ctx.bag, gnss_topic, track_options)
        except ValueError as exc:
            raise CheckSkipError(
                "unsupported_sensor",
                f"{exc}; a lever arm needs RTK-grade fixes (NavSatFix status >= 0 with a "
                f"3-D std of at most {track_options.max_sigma_m:g} m)",
            ) from exc
        model = GnssTrackModel(track.times_s, track.enu_m, track.sigma_m)
        scans = (
            ctx.controls.scan_store.scans(ctx.bag, profile, max_seconds=max_seconds)
            if ctx.controls.scan_store is not None
            else _iter_scans_plain(ctx.bag, profile, max_seconds)
        )
        emit_stage("gnss-lidar: LiDAR odometry windows against the GNSS track")
        segmenter = collect_windows(model, scans, run_options, prefix=f"{ctx.bag.name}/")
        emit_stage("gnss-lidar: lever-arm estimation")
        evaluation = evaluate_gnss_lidar_lever_arm(
            model, segmenter.windows, run_options, reference_lever_arm=lever_candidate
        )
        return build_gnss_lidar_artifact(
            evaluation,
            run_options,
            segments=GnssLidarSegmentSummary(
                scans_read=segmenter.scans_read,
                scans_in_gnss_coverage=segmenter.scans_covered,
                odometry_segments=segmenter.segments,
                unreliable_registrations=segmenter.unreliable,
                windows=len(segmenter.windows),
                gnss_epochs_used=len(track.times_s),
                gnss_epochs_rejected=track.rejected_epochs,
            ),
            provenance=GnssLidarLeverArmProvenance(
                generator=__name__,
                generator_version=__version__,
                git_commit=git_commit(),
                command=["calibrex", "check", str(ctx.bag), "gnss-lidar", gnss_topic, lidar_topic],
                dataset_family=DATASET_FAMILY,
                sequence_ids=[ctx.bag.name],
                input_sha256=_digest_or_topic(ctx, gnss_topic, lidar_topic),
                input_digest_scope=(
                    "bag digest (metadata.yaml in full; storage files by name, size and first "
                    f"64 MiB); GNSS from NavSatFix {gnss_topic}: {summary.kept} of "
                    f"{summary.messages} fixes usable (no fix {summary.rejected_no_fix}, "
                    f"std over {track_options.max_sigma_m:g} m {summary.rejected_sigma}, "
                    f"unknown covariance {summary.rejected_unknown_quality})"
                ),
                dataset_license=DATASET_LICENSE,
            ),
            reference="deployed candidate calibration (antenna position in the LiDAR frame)",
            limitations=(
                *RTK_SLAM_LIMITATIONS[1:],
                *(RIGID_SCAN_LIMITATIONS if rigid else ()),
                "GNSS fixes come from the bag's NavSatFix topic; usable fixes are those with "
                "status >= 0 and a 3-D std (sqrt of the covariance trace) of at most "
                f"{track_options.max_sigma_m:g} m.",
                "The reference is the deployed candidate transform, not metrology.",
            ),
        )

    artifact, from_cache = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "gnss_lidar_lever_arm",
        {
            "gnss_topic": gnss_topic,
            "profile": profile,
            "max_seconds": max_seconds,
            "track": track_options,
            "options": run_options,
        },
        loader=load_gnss_lidar_lever_arm,
        compute=compute,
        rebase=lambda hit: rebase_gnss_lidar_reference(hit, lever_candidate),
    )
    solved = artifact.lever_arm_m is not None and artifact.solver_status == "converged"
    estimates: list[AxisEstimate] = []
    if artifact.lever_arm_m is not None:
        estimates = _unjudged_rotation_axes() + translation_axis_estimates(
            artifact.dofs, lever_candidate, artifact.lever_arm_m
        )
        if artifact.policy_status == "fail":
            reason = "the lever-arm estimate failed its held-out check: " + "; ".join(
                artifact.policy_reasons
            )
            estimates = [
                item if item.unit == "deg" else _unestimated(item, reason) for item in estimates
            ]
        elif rigid:
            estimates = [
                item
                if item.unit == "deg"
                else replace(item, floor=GNSS_LIDAR_RIGID_TRANSLATION_FLOOR_M)
                for item in estimates
            ]
    segments = artifact.segments
    return EstimatorRun(
        estimator=artifact.method,
        artifacts=(
            EvidenceArtifact(
                "lever_arm",
                artifact,
                artifact.policy_status,
                from_cache if ctx.controls.cache is not None else None,
            ),
        ),
        solved=solved,
        policy_status=artifact.policy_status,
        policy_reasons=tuple(artifact.policy_reasons),
        estimates=tuple(estimates),
        compared=invert_transform(ctx.candidate),
        time_offset=time_offset_of(artifact.dofs, artifact.time_offset_s),
        deskew=deskew,
        notes=(
            f"GNSS from NavSatFix {gnss_topic}; "
            + (
                _rigid_note(ctx.controls.gnss_lidar_deskew, "--gnss-lidar-deskew", lidar_topic)
                + "; the time offset includes the driver's scan-stamp convention"
                if rigid
                else f"per-point time: field '{field_name}' read as {encoding}"
            ),
            f"{segments.windows} odometry windows in RTK-grade GNSS coverage "
            f"({segments.scans_in_gnss_coverage} of {segments.scans_read} scans, "
            f"{segments.gnss_epochs_used} usable fixes)",
            "lever arm = antenna position in the LiDAR frame (translation of T_lidar_gnss); "
            "rotation axes are not judged",
        ),
    )


def _find_runs(ctx: PairContext, pair: str, frame: str) -> dict[str, EstimatorRun]:
    """Runs of ``pair`` this check made with ``frame`` first, keyed by the second frame."""

    found: dict[str, EstimatorRun] = {}
    for key, value in ctx.controls.memo.items():
        if key[0] == PAIR_RUN_KEY and key[1] == pair:
            frames = key[2]
            if isinstance(frames, tuple) and frames[0] == frame:
                found[str(frames[1])] = value
    return found


def run_gnss_imu(ctx: PairContext) -> EstimatorRun:
    """Antenna lever arm in the IMU frame, composed from this check's GNSS-LiDAR and IMU-LiDAR.

    The planner's candidate is ``T_gnss_imu``; the estimate is the antenna
    position in the IMU frame, the translation of ``T_imu_gnss``.
    """

    from calibrex.evaluation.gnss_imu_lever_arm import compose_gnss_imu_lever_arm

    gnss_frame, imu_frame = ctx.pair.frames
    gnss_runs = _find_runs(ctx, "gnss-lidar", gnss_frame)
    imu_runs = _find_runs(ctx, "imu-lidar", imu_frame)
    shared = [lidar for lidar in gnss_runs if lidar in imu_runs]
    if not shared:
        missing = [
            name for name, runs in (("gnss-lidar", gnss_runs), ("imu-lidar", imu_runs)) if not runs
        ]
        raise CheckSkipError(
            "missing_dependency",
            "gnss-imu is composed from this check's own gnss-lidar and imu-lidar results for "
            "the same LiDAR; "
            + (
                f"{' and '.join(missing)} gave no result (not selected with --pairs, or skipped)"
                if missing
                else "no LiDAR has both"
            ),
        )
    lidar = shared[0]
    gnss_run, imu_run = gnss_runs[lidar], imu_runs[lidar]
    gnss_artifact = next((i.artifact for i in gnss_run.artifacts if i.role == "lever_arm"), None)
    rotation = next((i.artifact for i in imu_run.artifacts if i.role == "rotation"), None)
    translation = next((i.artifact for i in imu_run.artifacts if i.role == "translation"), None)
    problems: list[str] = []
    if (
        gnss_artifact is None
        or getattr(gnss_artifact, "lever_arm_m", None) is None
        or gnss_run.policy_status == "fail"
    ):
        problems.append("gnss-lidar has no usable lever arm")
    if (
        rotation is None
        or getattr(rotation, "rotation_quat_xyzw", None) is None
        or imu_run.policy_status == "fail"
    ):
        problems.append("imu-lidar has no usable rotation")
    if (
        translation is None
        or getattr(translation, "translation_m", None) is None
        or getattr(translation, "policy_status", None) == "fail"
    ):
        problems.append("imu-lidar has no usable lever arm (translation)")
    if problems or gnss_artifact is None or rotation is None or translation is None:
        raise CheckSkipError(
            "missing_dependency",
            "the composition needs a solved gnss-lidar lever arm and an imu-lidar rotation and "
            "lever arm that passed their held-out checks: " + "; ".join(problems),
        )
    lever_candidate = invert_transform(ctx.candidate)[:3, 3]  # antenna in the IMU frame
    ctx.controls.progress(f"gnss-imu: composing gnss-lidar and imu-lidar for LiDAR {lidar}")
    with tempfile.TemporaryDirectory() as scratch:
        paths: dict[str, Path] = {}
        for name, artifact in (
            ("gnss_lidar_lever_arm", gnss_artifact),
            ("imu_lidar_rotation", rotation),
            ("imu_lidar_translation", translation),
        ):
            paths[name] = Path(scratch) / f"{name}.yaml"
            artifact.save(paths[name])
        composed = compose_gnss_imu_lever_arm(
            paths["gnss_lidar_lever_arm"],
            paths["imu_lidar_rotation"],
            paths["imu_lidar_translation"],
            reference_m=[float(v) for v in lever_candidate],
            reference="deployed candidate calibration (antenna position in the IMU frame)",
            command=["calibrex", "check", str(ctx.bag), "gnss-imu"],
        )
    by_axis = {item.name: item for item in composed.axes}
    rigid = "none" in (gnss_run.deskew, imu_run.deskew)
    estimates = _unjudged_rotation_axes()
    xyz: tuple[Literal["x", "y", "z"], ...] = ("x", "y", "z")
    for index, name in enumerate(xyz):
        record = by_axis[name]
        observable = record.status == "estimated"
        estimates.append(
            AxisEstimate(
                name,
                "m",
                float(lever_candidate[index]) - record.value_m,
                record.std_m,
                observable,
                None
                if observable
                else (
                    f"the composed std is {record.std_m:.4g} m (over the bound) or an input axis "
                    "it depends on is not constrained"
                ),
                None if observable else "unobservable",
                GNSS_LIDAR_RIGID_TRANSLATION_FLOOR_M if rigid else None,
            )
        )
    return EstimatorRun(
        estimator=composed.method,
        artifacts=(EvidenceArtifact("composed_lever_arm", composed, composed.policy_status, None),),
        solved=True,
        policy_status=composed.policy_status,
        policy_reasons=tuple(composed.policy_reasons),
        estimates=tuple(estimates),
        compared=invert_transform(ctx.candidate),
        deskew="none" if rigid else None,
        notes=(
            f"composed from this check's gnss-lidar and imu-lidar evidence (LiDAR {lidar}); "
            "its std assumes independent inputs",
            "lever arm = antenna position in the IMU frame (translation of T_imu_gnss); "
            "rotation axes are not judged",
        ),
    )


# --------------------------------------------------------------- camera-lidar
#
# camera-lidar aligns image edges with LiDAR depth discontinuities (Levinson and Thrun's
# objective, pooled over frames; see calibrex.evaluation.camera_lidar_edge). The search
# starts at the candidate, so the estimate depends on it and the cache key includes it.
# Only the rotation is estimated: parallax resolves translation too weakly on every
# recording tried, and the artifact says so.

CAMERA_LIDAR_TRANSLATION_REASON = (
    "translation is not estimated: the edge objective sees it only through parallax, and on "
    "the recordings evaluated its estimate differed from the reference by up to 11 cm"
)


# An INS twist is the best velocity source, wheel odometry the least (it slips and may
# report the rear axle's motion only); an IMU gives rotation alone and comes last.
_MOTION_KIND_RANK = {"ins": 0, "unknown": 1, "wheel": 2}


def camera_lidar_motion_source(
    ctx: PairContext, lidar_frame: str
) -> tuple[str, str, FloatArray] | None:
    """The velocity topic that moves LiDAR points to the image time, if the bag has one.

    Returns ``(topic, message type, T_body_lidar)`` for the best odometry or twist topic
    (rotation and translation), else an IMU (rotation only), whose frame is connected to
    the LiDAR in the candidate tree; ``None`` when there is none or motion is turned off.
    """

    from calibrex.evaluation.camera_lidar_motion import IMU_TYPE, ODOMETRY_TYPE, TWIST_TYPES

    tree = ctx.controls.frame_tree
    if tree is None or ctx.controls.camera_lidar_motion == "none":
        return None

    def connected(record: CheckTopicRecord, types: frozenset[str]) -> bool:
        return (
            ctx.topic_types.get(record.topic) in types
            and record.mapped_frame is not None
            and tree.connected(record.mapped_frame, lidar_frame)
        )

    velocity_types = frozenset({ODOMETRY_TYPE, *TWIST_TYPES})
    velocity = sorted(
        (record for record in ctx.topics if connected(record, velocity_types)),
        key=lambda record: (
            _MOTION_KIND_RANK.get(record.odometry_kind or "unknown", 1),
            record.role != "odometry",
            record.topic,
        ),
    )
    gyro = sorted(
        (record for record in ctx.topics if connected(record, frozenset({IMU_TYPE}))),
        key=lambda record: record.topic,
    )
    for record in (*velocity, *gyro):
        assert record.mapped_frame is not None
        relation = tree.lookup(record.mapped_frame, lidar_frame)
        if relation is not None:
            return (
                record.topic,
                ctx.topic_types[record.topic],
                transform_matrix(relation.translation_m, relation.rotation_quat_xyzw),
            )
    return None


def camera_lidar_axis_estimates(
    artifact: CameraLidarEdgeArtifact, candidate: FloatArray
) -> list[AxisEstimate]:
    """The per-axis estimates a camera-lidar edge-alignment artifact gives the verdict."""

    solved = artifact.transform is not None
    estimate = (
        transform_matrix(artifact.transform.translation_m, artifact.transform.rotation_quat_xyzw)
        if artifact.transform is not None
        else candidate
    )
    errors = rotation_errors_deg(candidate[:3, :3], estimate[:3, :3])
    by_name = {record.name: record for record in artifact.dofs}
    names: tuple[CheckAxisName, ...] = ("roll", "pitch", "yaw", "x", "y", "z")
    estimates: list[AxisEstimate] = []
    for index, name in enumerate(names):
        record = by_name[name]
        rotation = index < 3
        unit: Literal["deg", "m"] = "deg" if rotation else "m"
        error = (
            errors[index] if rotation else float(candidate[index - 3, 3] - estimate[index - 3, 3])
        )
        control = record.known_bad_control
        if record.status == "estimated":
            estimates.append(AxisEstimate(name, unit, error, record.std_reported, True))
            continue
        code: Literal["unobservable", "control_not_detected", "no_estimate"] = "unobservable"
        if not solved:
            reason, code = "the estimator produced no estimate", "no_estimate"
        elif not rotation:
            reason = CAMERA_LIDAR_TRANSLATION_REASON
        elif control is not None and not control.detected:
            reason = (
                f"the estimator's known-bad control on this axis ({control.amount:g} deg) was "
                f"not detected on held-out frames (t = {control.min_t_statistic or 0.0:.1f}), "
                "so the estimate is not trusted as a yardstick"
            )
            code = "control_not_detected"
        else:
            bound = ", at the search bound" if record.at_search_bound else ""
            reason = (
                f"not constrained by the data (reported std {record.std_reported:.4g} deg{bound})"
            )
        estimates.append(AxisEstimate(name, unit, error, record.std_reported, False, reason, code))
    return estimates


def run_camera_lidar(ctx: PairContext) -> EstimatorRun:
    """``T_camera_lidar`` rotation by aligning LiDAR depth edges with image edges."""

    from calibrex.core.camera_lidar_edge import load_camera_lidar_edge
    from calibrex.evaluation.camera_lidar_edge import (
        CameraLidarRunInputs,
        EdgeAlignmentOptions,
        EdgeCamera,
        run_ros2_camera_lidar_edge,
    )
    from calibrex.evaluation.camera_lidar_motion import read_lidar_motion

    camera_topics = ctx.sensor_topics.get("camera", ())
    image_topic = next((t for t in camera_topics if ctx.topic_types.get(t) == IMAGE_TYPE), None)
    if image_topic is None:
        raise CheckSkipError(
            "unsupported_sensor",
            f"camera topic(s) {', '.join(camera_topics)} are not sensor_msgs/Image "
            "(compressed images are not supported by the check yet)",
        )
    lidar_topic = _require_pointcloud2(ctx, ctx.sensor_topics.get("lidar", ()))
    camera_frame, lidar_frame = ctx.pair.frames[0], ctx.pair.frames[1]
    model, intrinsics_source, _key = resolve_camera_intrinsics(ctx, image_topic, camera_frame)
    camera = EdgeCamera(
        model.fx, model.fy, model.cx, model.cy, model.distortion_model, tuple(model.distortion)
    )
    point_time = detect_point_time(ctx.bag, lidar_topic)
    candidate = ctx.candidate  # T_camera_lidar
    options = EdgeAlignmentOptions(
        max_seconds=ctx.controls.max_duration_s,
        start_scan_min_control_t=ctx.controls.camera_lidar_start_scan_min_control_t,
    )
    motion_source = camera_lidar_motion_source(ctx, lidar_frame)
    ctx.controls.progress(
        f"camera-lidar: {image_topic} + {lidar_topic} (intrinsics: {intrinsics_source})"
    )
    command = ("calibrex", "check", str(ctx.bag), "camera-lidar", image_topic, lidar_topic)
    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "camera_lidar_edge",
        {
            "image_topic": image_topic,
            "lidar_topic": lidar_topic,
            "camera": camera,
            "point_time": point_time,
            "candidate": [round(float(v), 9) for v in candidate.ravel()],
            "options": options,
            "motion": None
            if motion_source is None
            else {
                "topic": motion_source[0],
                "type": motion_source[1],
                "t_body_lidar": [round(float(v), 9) for v in motion_source[2].ravel()],
            },
        },
        loader=load_camera_lidar_edge,
        compute=lambda: run_ros2_camera_lidar_edge(
            ctx.bag,
            CameraLidarRunInputs(
                image_topic=image_topic,
                lidar_topic=lidar_topic,
                camera=camera,
                candidate=candidate,
                point_time=point_time,
                dataset_family=DATASET_FAMILY,
                dataset_license=DATASET_LICENSE,
                parent_frame=camera_frame,
                child_frame=lidar_frame,
                reference="deployed candidate calibration (T_camera_lidar)",
                command=command,
                input_sha256=_digest_or_topic(ctx, image_topic, lidar_topic),
                input_digest_scope=(
                    "bag digest (metadata.yaml in full; storage files by name, size and first "
                    "64 MiB)"
                ),
                motion=read_lidar_motion(ctx.bag, *motion_source)
                if motion_source is not None
                else None,
            ),
            options,
        ),
        rebase=lambda hit: hit,
    )
    solved = artifact.transform is not None
    estimates = camera_lidar_axis_estimates(artifact, candidate)
    samples = artifact.samples
    notes = [
        f"intrinsics: {intrinsics_source}",
        f"{samples.frames} frame(s), {samples.edge_points} LiDAR edge points; the camera frame "
        "must be the optical frame (z forward)",
    ]
    motion = artifact.motion_compensation
    if motion is not None:
        moved = (
            f"median {motion.median_rotation_deg:.2f} deg"
            + (
                f", {motion.median_translation_m:.3f} m"
                if motion.median_translation_m is not None
                else ""
            )
            if motion.median_rotation_deg is not None
            else "no frame had a motion sample nearby"
        )
        notes.append(
            f"LiDAR points moved to the image time with {motion.source_topic} "
            f"({'rotation and translation' if motion.kind == 'twist' else 'gyro, rotation only'}"
            f"; {moved})"
        )
    elif ctx.controls.camera_lidar_motion == "none":
        notes.append("scans used as recorded (--camera-lidar-motion none)")
    else:
        notes.append(
            "scans used as recorded: no odometry, twist or IMU topic is connected to the LiDAR "
            "in the frame tree"
        )
    if artifact.objective is not None:
        notes.append(
            f"edge-alignment objective {artifact.objective.start:.4f} at the candidate, "
            f"{artifact.objective.estimate:.4f} at the estimate"
        )
    return EstimatorRun(
        estimator=f"{artifact.method} (camera-lidar)",
        artifacts=(
            EvidenceArtifact(
                "edge_alignment",
                artifact,
                artifact.policy_status,
                cached if ctx.controls.cache is not None else None,
            ),
        ),
        solved=solved,
        policy_status=artifact.policy_status,
        policy_reasons=tuple(artifact.policy_reasons),
        estimates=tuple(estimates),
        compared=candidate,
        notes=tuple(notes),
    )


def run_camera_camera(ctx: PairContext) -> EstimatorRun:
    """A stereo pair's ``T_first_second`` rotation from its rectified row offsets."""

    from calibrex.core.provenance import git_commit
    from calibrex.core.stereo_rotation import StereoRotationProvenance, load_stereo_rotation
    from calibrex.evaluation.camera_lidar_motion import IMU_TYPE, read_lidar_motion
    from calibrex.evaluation.stereo_rotation import (
        StereoRotationOptions,
        StereoTiming,
        estimate_stereo_rotation,
        read_stereo_frames,
        shared_view_fraction,
        stereo_camera,
    )

    topics: list[str] = []
    for key in ("first", "second"):
        candidates = ctx.sensor_topics.get(key, ())
        image = next((t for t in candidates if ctx.topic_types.get(t) == IMAGE_TYPE), None)
        if image is None:
            raise CheckSkipError(
                "unsupported_sensor",
                f"camera topic(s) {', '.join(candidates)} are not sensor_msgs/Image "
                "(compressed images are not supported by the check yet)",
            )
        topics.append(image)
    left_topic, right_topic = topics
    left_frame, right_frame = ctx.pair.frames[0], ctx.pair.frames[1]
    left_model, left_source, _ = resolve_camera_intrinsics(ctx, left_topic, left_frame)
    right_model, right_source, _ = resolve_camera_intrinsics(ctx, right_topic, right_frame)
    ctx.controls.progress(f"camera-camera: {left_topic} + {right_topic}")
    candidate = ctx.candidate  # T_first_second
    options = StereoRotationOptions(max_seconds=ctx.controls.max_duration_s)
    command = ["calibrex", "check", str(ctx.bag), "camera-camera", left_topic, right_topic]
    gyro = _camera_gyro(ctx, left_frame)

    def compute() -> Any:
        motion = (
            read_lidar_motion(ctx.bag, gyro[0], IMU_TYPE, gyro[1]) if gyro is not None else None
        )
        tolerance = options.max_unsynchronised_gap_s if motion is not None else options.max_gap_s
        frames = read_stereo_frames(ctx.bag, left_topic, right_topic, options, max_gap_s=tolerance)
        pairs = frames.pairs
        timing = None
        if motion is not None and gyro is not None and pairs:
            sampled = [motion.velocity_at(float(time)) for time in frames.left_times_s]
            covered = [index for index, rate in enumerate(sampled) if rate is not None]
            rates = [rate[0] for rate in sampled if rate is not None]
            pairs = [pairs[index] for index in covered]
            timing = StereoTiming(
                gaps_s=(frames.right_times_s - frames.left_times_s)[covered],
                rates_rad_s=np.asarray(rates, dtype=np.float64).reshape(-1, 3),
                source=f"gyro {gyro[0]}",
            )
        if not pairs:
            raise CheckSkipError(
                "missing_topic",
                f"no image of {left_topic} and {right_topic} within "
                f"{tolerance * 1e3:g} ms of each other"
                + (" with the gyro's rate at the time" if motion is not None else ""),
            )
        left_height, left_width = pairs[0][0].shape
        right_height, right_width = pairs[0][1].shape
        left_camera = stereo_camera(left_model, left_width, left_height)
        right_camera = stereo_camera(right_model, right_width, right_height)
        shared = shared_view_fraction(left_camera, right_camera, candidate)
        if shared < options.min_shared_view:
            raise CheckSkipError(
                "no_shared_view",
                f"under the candidate the cameras share {shared:.0%} of the first one's view "
                f"(at 10 m), less than {options.min_shared_view:.0%}: not a stereo pair",
            )
        return estimate_stereo_rotation(
            pairs,
            left_camera,
            right_camera,
            candidate,
            provenance=StereoRotationProvenance(
                generator="calibrex.check.estimators.run_camera_camera",
                generator_version="0.1",
                git_commit=git_commit(),
                command=command,
                left_topic=left_topic,
                right_topic=right_topic,
                left_frame=left_frame,
                right_frame=right_frame,
                intrinsics_source=f"{left_source}; {right_source}",
                bag_sha256=ctx.controls.bag_sha256 or None,
            ),
            options=options,
            timing=timing,
        )

    artifact, cached = cached_artifact(
        ctx.controls.cache,
        ctx.controls.bag_sha256,
        "stereo_rotation",
        {
            "left_topic": left_topic,
            "right_topic": right_topic,
            "left_camera": left_model,
            "right_camera": right_model,
            "candidate": [round(float(v), 9) for v in candidate.ravel()],
            "gyro": None
            if gyro is None
            else {"topic": gyro[0], "t_imu_camera": [round(float(v), 9) for v in gyro[1].ravel()]},
            "options": options,
        },
        loader=load_stereo_rotation,
        compute=compute,
        rebase=lambda hit: hit,
    )
    by_name = {axis.name: axis for axis in artifact.axes}
    estimates: list[AxisEstimate] = []
    for name in ("roll", "pitch", "yaw"):
        axis = by_name.get(name)
        if axis is None:
            estimates.append(
                AxisEstimate(
                    name, "deg", 0.0, 0.0, False, "; ".join(artifact.policy_reasons), "no_estimate"
                )
            )
            continue
        response = axis.control_response
        control_missed = response is None or not (
            options.min_response <= response <= options.max_response
        )
        estimates.append(
            AxisEstimate(
                name,
                "deg",
                axis.candidate_error_deg,
                axis.std_deg,
                axis.observable,
                None if axis.observable else axis.reason,
                None
                if axis.observable
                else "control_not_detected"
                if control_missed
                else "unobservable",
            )
        )
    return EstimatorRun(
        estimator=artifact.method,
        artifacts=(
            EvidenceArtifact(
                "rotation",
                artifact,
                artifact.policy_status,
                cached if ctx.controls.cache is not None else None,
            ),
        ),
        solved=artifact.policy_status != "inconclusive",
        policy_status=artifact.policy_status,
        policy_reasons=tuple(artifact.policy_reasons),
        estimates=tuple(estimates),
        compared=candidate,
        time_offset=CheckTimeOffset(
            estimate_s=artifact.time_offset_s,
            std_s=artifact.time_offset_std_s,
            status=artifact.time_offset_status or "unobservable",
        )
        if artifact.time_offset_s is not None
        else None,
        notes=(
            f"intrinsics: {left_source}; {right_source}",
            "rotation only: the translation between the cameras is not judged",
            f"timing: {artifact.timing_source}, the cameras' clock offset fitted"
            if artifact.timing_source
            else "timing: no gyro connected to the cameras, so the pairs must be synchronised",
        ),
    )


def _camera_gyro(ctx: PairContext, camera_frame: str) -> tuple[str, FloatArray] | None:
    """An IMU topic connected to the camera in the candidate tree, and ``T_imu_camera``."""

    tree = ctx.controls.frame_tree
    if tree is None:
        return None
    for record in sorted(ctx.topics, key=lambda item: item.topic):
        if record.role != "imu" or record.mapped_frame is None:
            continue
        if ctx.topic_types.get(record.topic) != "sensor_msgs/msg/Imu":
            continue
        relation = tree.lookup(record.mapped_frame, camera_frame)
        if relation is not None:
            return record.topic, transform_matrix(
                relation.translation_m, relation.rotation_quat_xyzw
            )
    return None


ESTIMATORS: dict[str, Adapter] = {
    "imu-lidar": run_imu_lidar,
    "gnss-lidar": run_gnss_lidar,
    "gnss-imu": run_gnss_imu,
    "lidar-lidar": run_lidar_lidar,
    "camera-imu": run_camera_imu,
    "camera-focal": run_camera_focal,
    "camera-lidar": run_camera_lidar,
    "camera-camera": run_camera_camera,
    "lidar-vehicle": run_lidar_vehicle,
    "imu-vehicle": run_imu_vehicle,
    "ins-lidar": run_ins_lidar,
    "lidar-wheel_odometry": run_lidar_wheel,
}
WIRED_CHECK_PAIRS: frozenset[str] = frozenset(ESTIMATORS)
