Source code for suboptimumg.loganalysis._utils

"""Shared internal utilities for the loganalysis package."""

from __future__ import annotations

import numpy as np
import pandas as pd


def get_time_array(df: pd.DataFrame, time_col: str = "time_s") -> np.ndarray:
    """Extract the time array from a DataFrame, checking columns then index."""
    if time_col in df.columns:
        return df[time_col].values.astype(np.float64)
    if df.index.name == time_col:
        return df.index.values.astype(np.float64)
    raise KeyError(
        f"Time column '{time_col}' not found in columns or index. "
        f"Columns: {list(df.columns)}, Index name: {df.index.name}"
    )


def detect_time_divisor(timestamps: np.ndarray) -> float:
    """Return the divisor to convert timestamp *differences* to seconds.

    Heuristic based on the total span of the timestamp array:

    * span > 1e8  →  microseconds  (divisor = 1 000 000)
    * span > 1e4  →  milliseconds  (divisor = 1 000)
    * else         →  seconds       (divisor = 1)

    This works reliably for any log session longer than ~100 seconds
    (the typical minimum for CAN logs).  For unusually short sessions
    the caller can override via an explicit parameter.
    """
    print(
        """
DEPRECATION WARNING: detect_time_divisor is deprecated and will be removed in a future release.

Please check the latest PERDA documentation. PERDA now deterministically parses and stores
the timestamp units of the log (e.g. seconds, milliseconds, microseconds), and supports converting
between units.
"""
    )

    if len(timestamps) < 2:
        return 1.0
    span = float(timestamps[-1] - timestamps[0])
    if span > 1e8:
        return 1e6  # microseconds
    elif span > 1e4:
        return 1e3  # milliseconds
    return 1.0  # seconds


def time_label(divisor: float) -> str:
    """Human-readable label for the timestamp unit."""
    if divisor >= 1e6:
        return "μs"
    elif divisor >= 1e3:
        return "ms"
    return "s"




def unwrap_degrees(
    df: pd.DataFrame,
    columns: str | list[str],
    suffix: str = "",
    discontinuity: float = 180.0,
) -> pd.DataFrame:
    """Unwrap angle columns (degrees) to remove +-180 discontinuities.

    The VN-300 yaw output reads 0 at north, +90 east, +180 south,
    then jumps to -180 and climbs to 0.  This creates step
    discontinuities that break derivatives and integrations.
    ``np.unwrap`` removes them by adding/subtracting 360 at each jump.

    Parameters
    ----------
    df : pd.DataFrame
    columns : str or list[str]
        Column name(s) to unwrap.
    suffix : str
        If non-empty, the unwrapped result is written to
        ``<col><suffix>`` and the original column is kept.
        If empty (default), the column is overwritten in place.
    discontinuity : float
        Maximum allowed step between consecutive samples (degrees).
        Jumps larger than this are treated as wraps.

    Returns
    -------
    pd.DataFrame
        Same DataFrame with unwrapped column(s).
    """
    if isinstance(columns, str):
        columns = [columns]

    for col in columns:
        raw = df[col].values.astype(np.float64)
        unwrapped = np.degrees(
            np.unwrap(np.radians(raw), discont=np.radians(discontinuity))
        )
        out_col = col + suffix if suffix else col
        df[out_col] = unwrapped
        print(f"[utils] unwrap_degrees: {col} -> {out_col}  ({len(df)} pts)")

    return df



def downsample_indices(
    n_points: int,
    duration_s: float,
    max_display_hz: float | None,
) -> np.ndarray:
    """Return integer indices that thin *n_points* to ~*max_display_hz*.

    If *max_display_hz* is ``None`` or ``0``, returns all indices (no
    downsampling).  If the actual sample rate is already at or below
    *max_display_hz*, returns all indices and prints a warning.

    Parameters
    ----------
    n_points : int
        Total number of data points.
    duration_s : float
        Time span of the data in seconds.
    max_display_hz : float or None
        Target display sample rate.  ``None`` or ``0`` disables.

    Returns
    -------
    np.ndarray[int]
        Indices into the original array.
    """
    print(
        """
DEPRECATION WARNING: downsample_indices is deprecated and will be removed in a future release.

Please check the latest PERDA documentation. PERDA now includes built-in support for downsampling data for
display, either manually via `perda.plotting.subplots.stride_downsample` or automatically by passing in `max_display_resolution`
configuration objects when creating plots.
      """
    )

    if not max_display_hz or duration_s <= 0 or n_points <= 1:
        return np.arange(n_points)

    actual_hz = n_points / duration_s
    if actual_hz <= max_display_hz:
        import warnings

        warnings.warn(
            f"[plot] Requested {max_display_hz:.0f} Hz display rate but data "
            f"is only {actual_hz:.1f} Hz — not upsampling."
        )
        return np.arange(n_points)

    step = max(1, int(round(actual_hz / max_display_hz)))
    idx = np.arange(0, n_points, step)
    print(
        f"[plot] Downsampled {n_points} → {len(idx)} pts for display "
        f"(~{max_display_hz:.0f} Hz). Pass max_display_hz=0 to disable."
    )
    return idx