Source code for suboptimumg.log_analysis.yaw_segmentation_models

import numpy as np
from numpy.typing import NDArray
from pydantic import BaseModel, Field




class StationarityCell(BaseModel):
    """One (window length, tolerance) cell of the stationarity scan."""

    window_s: float = Field(description="Sliding-window length tested (s)")
    tol_pct: float = Field(
        description="Allowed peak-to-peak speed variation, as a percent of the window mean"
    )
    n_windows: int = Field(description="Number of sliding windows evaluated")
    n_passing: int = Field(description="Number of windows whose speed held within tol_pct")
    fraction_passing: float = Field(
        description="n_passing / n_windows; NaN when the window is longer than the log"
    )





class StationarityReport(BaseModel):
    """Grid of ``StationarityCell`` from ``scan_speed_stationarity``.

    ``fraction_passing`` near zero at every long window means the car never
    holds a steady speed, so a nonparametric Welch FRF is not viable and the
    parametric LPV output-error fit is the right path.
    """

    speed_col: str = Field(description="Speed channel the scan was run on")
    overlap: float = Field(description="Sliding-window overlap fraction (0 = none, 0.5 = 50%)")
    cells: list[StationarityCell] = Field(
        description="One cell per (window length, tolerance) pair"
    )



    def fraction_grid(
        self, win_lengths_s: tuple[float, ...], tol_pcts: tuple[float, ...]
    ) -> NDArray[np.float64]:
        """``(len(win_lengths_s), len(tol_pcts))`` array of ``fraction_passing``."""
        lookup = {(c.window_s, c.tol_pct): c.fraction_passing for c in self.cells}
        return np.array([[lookup.get((w, t), np.nan) for t in tol_pcts] for w in win_lengths_s])


    def __str__(self) -> str:
        best = max(self.cells, key=lambda c: c.fraction_passing, default=None)
        head = f"Stationarity scan on {self.speed_col} (overlap={self.overlap:.0%})"
        if best is None:
            return head + "\n  (no windows)"
        return (
            f"{head}\n  best cell: {best.fraction_passing:.1%} of windows steady "
            f"at len={best.window_s:g}s, tol=+/-{best.tol_pct:g}%"
        )





class MaskReport(BaseModel):
    """Summary of how ``auto_segment`` reduced one manual window."""

    manual_window: tuple[float, float] = Field(
        description="The operator-picked coarse window (t_start, t_end) in seconds"
    )
    n_input_pts: int = Field(description="Samples falling inside the manual window")
    n_kept_pts: int = Field(description="Samples surviving the speed and slip masks")
    n_low_speed_pts: int = Field(description="Samples rejected for being below the speed floor")
    n_high_slip_pts: int = Field(description="Samples rejected for exceeding the slip ceiling")
    sub_windows: list[tuple[float, float]] = Field(
        description="Accepted sub-window (t_start, t_end) times (s)"
    )
    rejected_short: list[tuple[float, float]] = Field(
        description="Surviving runs dropped for being shorter than min_window_s (s)"
    )

    def __str__(self) -> str:
        w0, w1 = self.manual_window
        return (
            f"window ({w0:.1f}, {w1:.1f})s: {len(self.sub_windows)} kept, "
            f"{len(self.rejected_short)} short-dropped, "
            f"low-speed={self.n_low_speed_pts}, high-slip={self.n_high_slip_pts}"
        )





class SubwindowSummary(BaseModel):
    """Per-sub-window quality metrics, used to triage windows before fitting."""

    sub_index: int = Field(description="Position of the sub-window in its parent list")
    n_pts: int = Field(description="Number of samples in the sub-window")
    duration_s: float = Field(description="Sub-window duration (s)")
    fs_hz: float = Field(description="Estimated sample rate (Hz)")
    speed_mean: float = Field(default=float("nan"), description="Mean speed (m/s)")
    speed_pp_pct: float = Field(
        default=float("nan"),
        description="Peak-to-peak speed variation as a percent of the mean",
    )
    ax_rms: float = Field(default=float("nan"), description="RMS longitudinal acceleration (m/s^2)")
    input_pp: float = Field(
        default=float("nan"), description="Peak-to-peak steering excitation (deg)"
    )
    slip_rms: float = Field(default=float("nan"), description="RMS body slip angle (deg)")

    def __str__(self) -> str:
        return (
            f"sub {self.sub_index:>2d}: {self.duration_s:5.1f}s @ {self.fs_hz:4.0f}Hz  "
            f"V={self.speed_mean:5.1f} (+/-{self.speed_pp_pct:4.0f}%)  "
            f"|ax|rms={self.ax_rms:4.2f}  steer_pp={self.input_pp:5.1f}  "
            f"slip_rms={self.slip_rms:4.2f}"
        )