"""Public, stable API for the underframe deflection algorithm. This module is the contract integrators should depend on. The internal algorithm modules (`beams`, `fitting`, `reference`, ...) are implementation details and may change between releases without notice; this module's function signatures and `DeflectionResult` schema are the stable surface. Typical usage:: from underframe_deflection import compute_deflection, load_point_cloud points = load_point_cloud("scan.ply") result = compute_deflection(points) print(result.left.selected_signed_mm, result.right.selected_signed_mm) """ from __future__ import annotations from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Callable, Sequence import numpy as np from . import io as _io from ._version import ALGORITHM_VERSION, SCHEMA_VERSION from .beams import candidate_rows, extract_beam_slices, track_side_candidates from .config import cloud_geometry, load_config, resolve_numeric_parameters from .exceptions import InsufficientPointsError, InvalidPointCloudError from .fitting import ( extract_representative_points, fit_representative_line, representatives_to_rows, evaluate_line, ) from .frame import ( build_bottom_roi_mask, frame_hypotheses_to_dict, infer_vehicle_frame, transform_points, transform_vectors, ) from .preprocess import estimate_spacing, preprocess_point_cloud from .reference import detect_reference_line, evaluate_reference_line from .structures import ( FittedLine, FrameHypothesis, ReferenceLine, RepresentativePoint, SliceResult, TrackSolution, ) ProgressCallback = Callable[[str, float], None] """Signature for progress callbacks: ``callback(stage_name, fraction_in_[0,1])``.""" # --------------------------------------------------------------------------- # Public result types # --------------------------------------------------------------------------- @dataclass(frozen=True) class SideDeflection: """Per-side deflection summary. All distances are millimetres. ``signed`` values are positive when the beam sits above the reference line and negative when it sags below. Fields whose value is ``None`` were not measurable on this side; check ``available`` and the parent result's ``quality_flags`` for the reason. """ side: str available: bool reference_available: bool reference_confidence: float max_downward_mm: float | None max_downward_x_mm: float | None max_upward_mm: float | None max_upward_x_mm: float | None selected_x_mm: float | None selected_signed_mm: float | None selected_absolute_mm: float | None beam_fit_residual_rms_z_mm: float | None @dataclass(frozen=True) class DeflectionCurves: """Sampled deflection along the longitudinal axis. Arrays are aligned by index to ``x_mm``. Deflection arrays may contain ``NaN`` where the curve is undefined (no overlap of beam fit and reference line, or low confidence). """ x_mm: np.ndarray left_deflection_mm: np.ndarray right_deflection_mm: np.ndarray left_confidence: np.ndarray right_confidence: np.ndarray @dataclass class _ComputeArtifacts: """Internal payload consumed by ``artifacts`` writers. Not part of the public API. Fields may change between releases and should only be accessed via the artifact writer functions in ``underframe_deflection.artifacts``. """ raw_points: np.ndarray filtered_vehicle: np.ndarray work_vehicle: np.ndarray work_roi: np.ndarray filtered_roi: np.ndarray frame_hypotheses: list[FrameHypothesis] best_frame: FrameHypothesis slices: list[SliceResult] left_tracks: list[TrackSolution] right_tracks: list[TrackSolution] best_left: TrackSolution | None best_right: TrackSolution | None left_reps: list[RepresentativePoint] right_reps: list[RepresentativePoint] left_line: FittedLine | None right_line: FittedLine | None left_reference: ReferenceLine right_reference: ReferenceLine x_grid: np.ndarray left_payload: dict[str, np.ndarray] right_payload: dict[str, np.ndarray] deflection_rows: list[dict] left_summary: dict right_summary: dict resolved_config: dict[str, Any] @dataclass class DeflectionResult: """The result of :func:`compute_deflection`. Public, stable fields: - :attr:`schema_version`, :attr:`algorithm_version` — version stamps - :attr:`left`, :attr:`right` — :class:`SideDeflection` headline results - :attr:`curves` — :class:`DeflectionCurves` sampled curves - :attr:`quality_flags` — list of non-fatal data-quality warnings - :attr:`report` — full nested dict (same content as ``report.json``) The private attribute :attr:`_artifacts` carries internal data needed by :mod:`underframe_deflection.artifacts` writers and should not be accessed directly by integrators. """ schema_version: str algorithm_version: str left: SideDeflection right: SideDeflection curves: DeflectionCurves quality_flags: list[str] report: dict _artifacts: _ComputeArtifacts | None = field(default=None, repr=False, compare=False) def to_dict(self) -> dict: """Return a fully-serializable dict (same content as ``report.json``).""" return self.report # --------------------------------------------------------------------------- # Public entry points # --------------------------------------------------------------------------- def load_point_cloud(path: str | Path) -> np.ndarray: """Load a point cloud from disk into an ``(N, 3)`` ``float64`` array. Supported formats: ``.ply`` (ASCII), ``.xyz``, ``.txt``, ``.csv``. Raises: UnsupportedFormatError: extension is not recognised. InvalidPointCloudError: file is malformed. """ try: return _io.read_point_cloud(path) except ValueError as exc: # io.read_point_cloud raises ValueError for both unsupported format # and malformed PLY headers; surface as a structured exception. from .exceptions import UnsupportedFormatError if "Unsupported point cloud format" in str(exc): raise UnsupportedFormatError(str(exc)) from exc raise InvalidPointCloudError(str(exc)) from exc def compute_deflection( points: np.ndarray | Sequence[Sequence[float]], config: dict | None = None, profile: str = "auto", *, progress: ProgressCallback | None = None, roi_aabb: dict | None = None, ) -> DeflectionResult: """Run the full deflection pipeline on an in-memory point cloud. This function performs no file I/O. To persist the result use the writers in :mod:`underframe_deflection.artifacts`. Args: points: ``(N, 3)`` array-like of XYZ coordinates in millimetres. The X axis is assumed longitudinal, Y lateral, Z vertical (positive up). If your scan uses a different convention, transform it before calling this function. config: Optional config override dict. Use the same shape as :data:`underframe_deflection.config.DEFAULT_CONFIG`. Values you omit fall back to the profile defaults. profile: Tuning profile name. ``"auto"`` works for most scans. See :data:`underframe_deflection.config.PROFILE_OVERRIDES` for the available profiles. progress: Optional callback invoked at each major stage with the stage name and a ``fraction in [0, 1]``. Returns: :class:`DeflectionResult` with the headline numbers, sampled curves, full report dict, and quality flags. Raises: InvalidPointCloudError: ``points`` is not a 2D ``(N, 3)`` array. InsufficientPointsError: too few points to run the pipeline. """ raw_points = _validate_points(points) def _emit(stage: str, fraction: float) -> None: if progress is not None: progress(stage, float(fraction)) _emit("preprocess", 0.0) base_config = load_config(None, profile) if config: from .config import deep_merge base_config = deep_merge(base_config, config) if roi_aabb is not None: from .roi import normalize_aabb normalized = normalize_aabb(roi_aabb) base_config["roi"]["aabb"] = normalized passthrough = base_config["preprocess"].setdefault( "passthrough", {"enabled": False, "aabb": None} ) if passthrough.get("aabb") is None: passthrough["enabled"] = True passthrough["aabb"] = normalized initial_geometry = cloud_geometry(raw_points) initial_spacing = estimate_spacing( raw_points, sample_size=base_config["preprocess"]["spacing_sample_size"], ) preprocess_config = resolve_numeric_parameters(base_config, initial_geometry, initial_spacing) preprocess_result = preprocess_point_cloud(raw_points, preprocess_config) _emit("preprocess", 1.0) _emit("frame", 0.0) if base_config["frame"].get("use_input_axes", False): best_frame = FrameHypothesis( name="input_axes_identity", axes=np.eye(3, dtype=np.float64), origin=np.zeros(3, dtype=np.float64), score=1.0, bottom_score=1.0, symmetry_score=1.0, flatness_score=1.0, hole_score=1.0, continuity_score=1.0, ) frame_hypotheses = [best_frame] else: frame_hypotheses, best_frame = infer_vehicle_frame( preprocess_result.work_points, preprocess_result.work_normals, preprocess_result.work_curvature, preprocess_config, ) _emit("frame", 1.0) filtered_vehicle = transform_points( preprocess_result.filtered_points, best_frame.axes, best_frame.origin ) work_vehicle = transform_points( preprocess_result.work_points, best_frame.axes, best_frame.origin ) work_normals_vehicle = transform_vectors(preprocess_result.work_normals, best_frame.axes) vehicle_geometry = cloud_geometry(work_vehicle) resolved_config = resolve_numeric_parameters( base_config, vehicle_geometry, preprocess_result.spacing ) roi_mask_work, roi_payload = build_bottom_roi_mask(work_vehicle, resolved_config) roi_mask_filtered = _apply_roi_mask(filtered_vehicle, roi_payload) work_roi = work_vehicle[roi_mask_work] work_roi_normals = work_normals_vehicle[roi_mask_work] filtered_roi = filtered_vehicle[roi_mask_filtered] _emit("beam_tracking", 0.0) slices = extract_beam_slices(work_roi, work_roi_normals, resolved_config) left_tracks = track_side_candidates(slices, "left", resolved_config) right_tracks = track_side_candidates(slices, "right", resolved_config) best_left = left_tracks[0] if left_tracks else None best_right = right_tracks[0] if right_tracks else None _emit("beam_tracking", 1.0) _emit("line_fitting", 0.0) left_reps = ( extract_representative_points(filtered_roi, best_left, resolved_config) if best_left else [] ) right_reps = ( extract_representative_points(filtered_roi, best_right, resolved_config) if best_right else [] ) left_line = fit_representative_line(left_reps, resolved_config, "left") right_line = fit_representative_line(right_reps, resolved_config, "right") _emit("line_fitting", 1.0) _emit("reference", 0.0) left_reference = ( detect_reference_line(filtered_roi, left_line, resolved_config, "left") if left_line is not None else ReferenceLine("left", False, None, None, None, 0.0, None) ) right_reference = ( detect_reference_line(filtered_roi, right_line, resolved_config, "right") if right_line is not None else ReferenceLine("right", False, None, None, None, 0.0, None) ) _emit("reference", 1.0) _emit("deflection", 0.0) line_candidates = [line for line in (left_line, right_line) if line is not None] if line_candidates: grid_step = ( resolved_config["representative"]["grid_step_mm"] * resolved_config["deflection"]["evaluation_grid_factor"] ) x_min = min(float(line.x_support.min()) for line in line_candidates) x_max = max(float(line.x_support.max()) for line in line_candidates) x_grid = np.arange(x_min, x_max + grid_step, grid_step) else: x_grid = np.zeros(0, dtype=np.float64) left_summary, left_payload = _evaluate_side_deflection("left", left_line, left_reference, x_grid) right_summary, right_payload = _evaluate_side_deflection( "right", right_line, right_reference, x_grid ) deflection_rows = ( _build_deflection_rows(x_grid, left_payload, right_payload) if len(x_grid) else [] ) _emit("deflection", 1.0) symmetry = _side_symmetry_metrics(best_left, best_right) quality_flags = _collect_quality_flags( best_left, best_right, left_line, right_line, left_reference, right_reference, symmetry, resolved_config, ) report = { "schema_version": SCHEMA_VERSION, "algorithm_version": ALGORITHM_VERSION, "axis_convention": {"x": "longitudinal", "y": "lateral", "z": "height"}, "profile": profile, "resolved_config": resolved_config, "preprocess": preprocess_result.stats, "frame_hypotheses": frame_hypotheses_to_dict(frame_hypotheses), "selected_frame": frame_hypotheses_to_dict([best_frame])[0], "roi": roi_payload, "symmetry_check": symmetry, "left_track": _track_to_dict(best_left), "right_track": _track_to_dict(best_right), "representative_lines": { "left": _fitted_line_to_dict(left_line), "right": _fitted_line_to_dict(right_line), }, "reference_lines": { "left": _reference_to_dict(left_reference), "right": _reference_to_dict(right_reference), }, "alternate_solutions": { "left_tracks": [_track_to_dict(track) for track in left_tracks[1:]], "right_tracks": [_track_to_dict(track) for track in right_tracks[1:]], "left_reference": [dict(item) for item in left_reference.alternates], "right_reference": [dict(item) for item in right_reference.alternates], }, "deflection": {"left": left_summary, "right": right_summary}, "quality_flags": quality_flags, "representative_points": { "left": representatives_to_rows(left_reps), "right": representatives_to_rows(right_reps), }, } artifacts = _ComputeArtifacts( raw_points=raw_points, filtered_vehicle=filtered_vehicle, work_vehicle=work_vehicle, work_roi=work_roi, filtered_roi=filtered_roi, frame_hypotheses=frame_hypotheses, best_frame=best_frame, slices=slices, left_tracks=left_tracks, right_tracks=right_tracks, best_left=best_left, best_right=best_right, left_reps=left_reps, right_reps=right_reps, left_line=left_line, right_line=right_line, left_reference=left_reference, right_reference=right_reference, x_grid=x_grid, left_payload=left_payload, right_payload=right_payload, deflection_rows=deflection_rows, left_summary=left_summary, right_summary=right_summary, resolved_config=resolved_config, ) return DeflectionResult( schema_version=SCHEMA_VERSION, algorithm_version=ALGORITHM_VERSION, left=_side_summary_to_public("left", left_summary, left_line), right=_side_summary_to_public("right", right_summary, right_line), curves=DeflectionCurves( x_mm=x_grid, left_deflection_mm=left_payload["deflection"], right_deflection_mm=right_payload["deflection"], left_confidence=np.minimum(left_payload["beam_conf"], left_payload["ref_conf"]), right_confidence=np.minimum(right_payload["beam_conf"], right_payload["ref_conf"]), ), quality_flags=quality_flags, report=report, _artifacts=artifacts, ) # --------------------------------------------------------------------------- # Helpers (private) # --------------------------------------------------------------------------- def _validate_points(points: np.ndarray | Sequence[Sequence[float]]) -> np.ndarray: array = np.asarray(points, dtype=np.float64) if array.ndim != 2 or array.shape[1] < 3: raise InvalidPointCloudError( f"Point cloud must be a 2D (N, 3+) array, got shape {array.shape}." ) if array.shape[1] > 3: array = array[:, :3] if len(array) < 256: raise InsufficientPointsError( f"Need at least 256 points to run the pipeline, got {len(array)}." ) return np.ascontiguousarray(array) def _apply_roi_mask(points_vehicle: np.ndarray, roi_payload: dict) -> np.ndarray: mask = ( (points_vehicle[:, 0] >= roi_payload["x_min"]) & (points_vehicle[:, 0] <= roi_payload["x_max"]) & (points_vehicle[:, 2] <= roi_payload["z_ceiling"]) ) if roi_payload.get("source") == "external_aabb": mask &= ( (points_vehicle[:, 1] >= roi_payload["y_min"]) & (points_vehicle[:, 1] <= roi_payload["y_max"]) & (points_vehicle[:, 2] >= roi_payload["z_min"]) ) return mask def _track_to_dict(track: TrackSolution | None) -> dict | None: if track is None: return None return { "side": track.side, "rank": track.rank, "score": track.score, "coverage_ratio": track.coverage_ratio, "coverage_count": track.coverage_count, "score_margin": track.score_margin, "confidence": track.confidence, "candidate_ids": track.candidate_ids, "candidates": [ { "candidate_id": candidate.candidate_id, "slice_index": candidate.slice_index, "x_center": candidate.x_center, "y_center": candidate.y_center, "z_low": candidate.z_low, "width": candidate.width, "score": candidate.score, } for candidate in track.candidates ], } def _fitted_line_to_dict(line: FittedLine | None) -> dict | None: if line is None: return None return { "side": line.side, "y_residual_rms": line.y_residual_rms, "z_residual_rms": line.z_residual_rms, "support_x_min": float(line.x_support.min()), "support_x_max": float(line.x_support.max()), "support_count": int(len(line.x_support)), "low_confidence_segments": [list(segment) for segment in line.low_confidence_segments], "gaps": [list(segment) for segment in line.gaps], } def _reference_to_dict(reference: ReferenceLine) -> dict: return { "side": reference.side, "available": reference.available, "confidence": reference.confidence, "residual": reference.residual, "point_front": None if reference.point_front is None else reference.point_front.tolist(), "point_back": None if reference.point_back is None else reference.point_back.tolist(), "direction": None if reference.direction is None else reference.direction.tolist(), "candidates_front": [ asdict(candidate) | {"center": candidate.center.tolist()} for candidate in reference.candidates_front ], "candidates_back": [ asdict(candidate) | {"center": candidate.center.tolist()} for candidate in reference.candidates_back ], "alternates": [dict(item) for item in reference.alternates], } def _side_symmetry_metrics( left_track: TrackSolution | None, right_track: TrackSolution | None ) -> dict: if not left_track or not right_track or not left_track.candidates or not right_track.candidates: return {"available": False} y_mid_samples = [ candidate.y_mid for candidate in left_track.candidates + right_track.candidates ] y_mid = float(np.median(y_mid_samples)) left_offset = float( np.median([abs(candidate.y_center - y_mid) for candidate in left_track.candidates]) ) right_offset = float( np.median([abs(candidate.y_center - y_mid) for candidate in right_track.candidates]) ) symmetry_error = abs(left_offset - right_offset) return { "available": True, "midline_y": y_mid, "left_offset": left_offset, "right_offset": right_offset, "symmetry_error": symmetry_error, "relative_symmetry_error": symmetry_error / max((left_offset + right_offset) * 0.5, 1.0), } def _evaluate_side_deflection( side: str, line: FittedLine | None, reference: ReferenceLine, x_grid: np.ndarray, ) -> tuple[dict, dict[str, np.ndarray]]: if line is None: empty = np.full_like(x_grid, np.nan, dtype=np.float64) return ( { "side": side, "available": False, "reason": "representative_line_unavailable", "selected_measurement": { "available": False, "reason": "representative_line_unavailable", }, }, { "beam_y": empty, "beam_z": empty, "beam_conf": np.zeros_like(x_grid), "ref_y": empty, "ref_z": empty, "ref_conf": np.zeros_like(x_grid), "deflection": empty, }, ) beam_y, beam_z, beam_conf = evaluate_line(line, x_grid) ref_eval, ref_conf = evaluate_reference_line(reference, x_grid) if np.ndim(ref_eval) != 2: ref_y = np.full_like(x_grid, np.nan, dtype=np.float64) ref_z = np.full_like(x_grid, np.nan, dtype=np.float64) else: ref_y = ref_eval[:, 0] ref_z = ref_eval[:, 1] deflection = beam_z - ref_z invalid = np.isnan(ref_z) | (beam_conf <= 0.0) | (ref_conf <= 0.0) deflection[invalid] = np.nan valid = ~np.isnan(deflection) summary = { "side": side, "available": bool(np.any(valid)), "reference_available": reference.available, "beam_fit_residual_rms_y": line.y_residual_rms, "beam_fit_residual_rms_z": line.z_residual_rms, "reference_confidence": reference.confidence, "reference_residual": reference.residual, } if np.any(valid): valid_x = x_grid[valid] valid_deflection = deflection[valid] valid_indices = np.flatnonzero(valid) down_index = int(np.argmin(valid_deflection)) up_index = int(np.argmax(valid_deflection)) abs_index = int(np.argmax(np.abs(valid_deflection))) selected_x = float(valid_x[abs_index]) global_index = int(valid_indices[abs_index]) beam_point = np.array( [selected_x, float(beam_y[global_index]), float(beam_z[global_index])], dtype=np.float64, ) reference_point = np.array( [selected_x, float(ref_y[global_index]), float(ref_z[global_index])], dtype=np.float64, ) vertical_anchor = np.array( [selected_x, float(beam_y[global_index]), float(ref_z[global_index])], dtype=np.float64, ) selected_value = float(valid_deflection[abs_index]) summary.update( { "max_downward_deflection_mm": float(valid_deflection[down_index]), "max_downward_x_mm": float(valid_x[down_index]), "max_upward_deflection_mm": float(valid_deflection[up_index]), "max_upward_x_mm": float(valid_x[up_index]), "selected_measurement": { "available": True, "selection_rule": "maximum_absolute_vertical_deflection", "selected_type": "downward" if selected_value < 0.0 else "upward", "x_mm": selected_x, "signed_deflection_mm": selected_value, "absolute_deflection_mm": float(abs(selected_value)), "beam_point": beam_point.tolist(), "reference_point": reference_point.tolist(), "vertical_anchor_point": vertical_anchor.tolist(), "vertical_distance_mm": float(abs(selected_value)), }, } ) else: summary["reason"] = "reference_unavailable_or_no_valid_overlap" summary["selected_measurement"] = { "available": False, "reason": "reference_unavailable_or_no_valid_overlap", } return ( summary, { "beam_y": beam_y, "beam_z": beam_z, "beam_conf": beam_conf, "ref_y": ref_y, "ref_z": ref_z, "ref_conf": ref_conf, "deflection": deflection, }, ) def _build_deflection_rows( x_grid: np.ndarray, left_payload: dict[str, np.ndarray], right_payload: dict[str, np.ndarray], ) -> list[dict]: rows: list[dict] = [] for index, x_value in enumerate(x_grid): rows.append( { "x": float(x_value), "left_beam_y": _nan_safe(left_payload["beam_y"][index]), "left_beam_z": _nan_safe(left_payload["beam_z"][index]), "left_ref_y": _nan_safe(left_payload["ref_y"][index]), "left_ref_z": _nan_safe(left_payload["ref_z"][index]), "left_deflection": _nan_safe(left_payload["deflection"][index]), "left_confidence": float( min(left_payload["beam_conf"][index], left_payload["ref_conf"][index]) ), "right_beam_y": _nan_safe(right_payload["beam_y"][index]), "right_beam_z": _nan_safe(right_payload["beam_z"][index]), "right_ref_y": _nan_safe(right_payload["ref_y"][index]), "right_ref_z": _nan_safe(right_payload["ref_z"][index]), "right_deflection": _nan_safe(right_payload["deflection"][index]), "right_confidence": float( min(right_payload["beam_conf"][index], right_payload["ref_conf"][index]) ), } ) return rows def _nan_safe(value: float) -> float: return float(value) if not np.isnan(value) else float("nan") def _collect_quality_flags( best_left: TrackSolution | None, best_right: TrackSolution | None, left_line: FittedLine | None, right_line: FittedLine | None, left_reference: ReferenceLine, right_reference: ReferenceLine, symmetry: dict, resolved_config: dict, ) -> list[str]: flags: list[str] = [] low_conf = resolved_config["reference"]["low_confidence_threshold"] if best_left is None: flags.append("left_track_missing") if best_right is None: flags.append("right_track_missing") if left_line is None: flags.append("left_representative_line_missing") if right_line is None: flags.append("right_representative_line_missing") if not left_reference.available: flags.append("left_reference_unavailable") if not right_reference.available: flags.append("right_reference_unavailable") if left_reference.available and left_reference.confidence < low_conf: flags.append("left_reference_low_confidence") if right_reference.available and right_reference.confidence < low_conf: flags.append("right_reference_low_confidence") if symmetry.get("available") and symmetry["relative_symmetry_error"] > 0.18: flags.append("left_right_symmetry_warning") if best_left and best_left.coverage_ratio < 0.45: flags.append("left_track_low_coverage") if best_right and best_right.coverage_ratio < 0.45: flags.append("right_track_low_coverage") return flags def _side_summary_to_public( side: str, summary: dict, line: FittedLine | None ) -> SideDeflection: selected = summary.get("selected_measurement", {}) or {} return SideDeflection( side=side, available=bool(summary.get("available", False)), reference_available=bool(summary.get("reference_available", False)), reference_confidence=float(summary.get("reference_confidence", 0.0) or 0.0), max_downward_mm=summary.get("max_downward_deflection_mm"), max_downward_x_mm=summary.get("max_downward_x_mm"), max_upward_mm=summary.get("max_upward_deflection_mm"), max_upward_x_mm=summary.get("max_upward_x_mm"), selected_x_mm=selected.get("x_mm"), selected_signed_mm=selected.get("signed_deflection_mm"), selected_absolute_mm=selected.get("absolute_deflection_mm"), beam_fit_residual_rms_z_mm=line.z_residual_rms if line is not None else None, ) __all__ = [ "DeflectionResult", "SideDeflection", "DeflectionCurves", "ProgressCallback", "compute_deflection", "load_point_cloud", ]