Source code for suboptimumg.loganalysis.intake

"""PERDA log -> aligned/resampled pandas DataFrame, plus fast GPS-only loader."""

from __future__ import annotations

import os
import re
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path
from typing import TYPE_CHECKING

import numpy as np
import pandas as pd
from scipy.interpolate import interp1d

from ._utils import detect_time_divisor

if TYPE_CHECKING:
    from perda.analyzer import Analyzer


[docs] def perda_to_dataframe( analyzer: Analyzer, var_names: list[str], resample_method: str | dict[str, str] = "linear", resample_freq_hz: float | None = None, reference_var: str | None = None, deduplicate_vars: list[str] | None = None, ) -> pd.DataFrame: """Convert selected PERDA variables into a single aligned DataFrame. Only loads the explicitly requested variables — never touches the full set of 200-300+ variables in a log. Parameters ---------- analyzer : Analyzer A loaded PERDA Analyzer instance. var_names : list[str] CAN names of the variables to extract. resample_method : str or dict[str, str] Interpolation method for aligning to a common time grid. ``"linear"`` (default), ``"zoh"`` (zero-order hold / persist until update), ``"nearest"``, or ``"cubic"``. Pass a dict mapping variable names to methods to mix strategies. resample_freq_hz : float or None If given, resample to a uniform grid at this frequency (Hz). Otherwise the union of all source timestamps is used. reference_var : str or None Left-join all variables to this variable's timestamp grid. If ``None`` and *resample_freq_hz* is also ``None``, the variable with the most samples is chosen automatically. deduplicate_vars : list[str] or None Variable names whose consecutive duplicate (ZOH) values should be collapsed before resampling. Useful for signals like GPS position that are logged at a high CAN rate but only update at a lower real rate (e.g. 100 Hz CAN / 4 Hz real). Deduplication keeps the first sample of each constant run so that resampling interpolates smoothly between real updates. Returns ------- pd.DataFrame Index: ``time_s`` (float64 seconds, zero-based from log start). One column per variable. """ print( """ DEPRECATION WARNING: perda_to_dataframe is deprecated and will be removed in a future release. Please reinstall PERDA. The latest version features significant improvements to log loading speed. For maximum performance, install perda[notebook] instead of perda[full]. """ ) data = analyzer.data instances = {name: data[name] for name in var_names} # Auto-detect timestamp unit all_ts_global = np.array([data.data_start_time, data.data_end_time], dtype=np.int64) divisor = detect_time_divisor(all_ts_global) # Auto-select reference variable: pick whichever has the most samples if reference_var is None and resample_freq_hz is None: reference_var = max(instances, key=lambda n: len(instances[n].timestamp_np)) print( f"[intake] Auto-selected reference variable: {reference_var} " f"({len(instances[reference_var].timestamp_np)} pts)" ) # Build the target time grid if reference_var is not None: target_ts = instances[reference_var].timestamp_np.copy() elif resample_freq_hz is not None: all_starts = [ di.timestamp_np[0] for di in instances.values() if len(di.timestamp_np) ] all_ends = [ di.timestamp_np[-1] for di in instances.values() if len(di.timestamp_np) ] t_start = min(all_starts) t_end = max(all_ends) dt_units = divisor / resample_freq_hz target_ts = np.arange(t_start, t_end, dt_units, dtype=np.float64).astype( np.int64 ) else: all_ts = np.concatenate([di.timestamp_np for di in instances.values()]) target_ts = np.unique(all_ts) if resample_freq_hz is not None and reference_var is not None: dt_units = divisor / resample_freq_hz target_ts = np.arange( target_ts[0], target_ts[-1], dt_units, dtype=np.float64 ).astype(np.int64) target_f = target_ts.astype(np.float64) columns: dict[str, np.ndarray] = {} for name, di in instances.items(): if isinstance(resample_method, dict): method = resample_method.get(name, "linear") else: method = resample_method src_t = di.timestamp_np.astype(np.float64) src_v = di.value_np if deduplicate_vars and name in deduplicate_vars: keep = np.concatenate(([True], src_v[1:] != src_v[:-1])) n_before = len(src_v) src_t = src_t[keep] src_v = src_v[keep] print(f" [dedup] {name}: {n_before} → " f"{len(src_v)} unique samples") if method == "linear": columns[name] = np.interp(target_f, src_t, src_v) elif method == "zoh": f = interp1d( src_t, src_v, kind="previous", bounds_error=False, fill_value=(src_v[0], src_v[-1]), ) columns[name] = f(target_f) elif method == "nearest": f = interp1d( src_t, src_v, kind="nearest", bounds_error=False, fill_value=(src_v[0], src_v[-1]), ) columns[name] = f(target_f) elif method == "cubic": if len(src_t) < 4: columns[name] = np.interp(target_f, src_t, src_v) else: f = interp1d( src_t, src_v, kind="cubic", bounds_error=False, fill_value=(src_v[0], src_v[-1]), ) columns[name] = f(target_f) else: raise ValueError( f"Unknown resample method '{method}'. Use linear/zoh/nearest/cubic." ) # Convert timestamps to seconds (zero-based from log start) target_s = target_ts.astype(np.float64) / divisor df = pd.DataFrame(columns, index=pd.Index(target_s, name="time_s")) duration = df.index[-1] - df.index[0] added = list(df.columns) print( f"[intake] Created DataFrame: {len(df)} rows, {len(added)} columns, {duration:.2f} s" ) for col in added: print(f" + {col} ({len(df[col])} pts)") return df
# --------------------------------------------------------------------------- # Fast GPS-only loader (bypasses PERDA) # --------------------------------------------------------------------------- _HEADER_RE = re.compile(r"^Value\s+.+\((.+)\):\s*(\d+)\s*$") _GPS_FIX_KEY = "__gps_fix__"
[docs] def quick_load_gps( path: str | Path, lat_pattern: str = "posLla.latitude", lon_pattern: str = "posLla.longitude", gps_fix_pattern: str = "gpsFix", min_gps_fix: float = 1.0, extra_patterns: list[str] | None = None, _verbose: bool = True, ) -> pd.DataFrame | None: """Fast GPS extractor that parses a PER CSV without PERDA. Reads only the header to discover variable IDs, then streams the data section keeping only lat/lon rows. Typically 5-10x faster than a full PERDA load for GPS preview purposes. Parameters ---------- path : str or Path Path to a PER-format ``.csv`` log file. lat_pattern, lon_pattern : str Substrings matched (case-sensitive) against the ``cpp_name`` in the header to identify latitude and longitude variable IDs. extra_patterns : list[str] or None Additional variable patterns to extract alongside lat/lon (e.g. ``["gpsFix", "posLla.altitude"]``). Returns ------- pd.DataFrame or None DataFrame indexed by ``time_s`` with ``latitude``, ``longitude`` columns (plus any extras). Returns ``None`` if the file has no GPS lock (all lat/lon values are zero). """ print( """ DEPRECATION WARNING: quick_load_gps is deprecated and will be removed in a future release. Please reinstall PERDA. The latest version features significant improvements to log loading speed. For maximum performance, install perda[notebook] instead of perda[full]. """ ) path = Path(path) all_patterns = {"latitude": lat_pattern, "longitude": lon_pattern} if gps_fix_pattern: all_patterns[_GPS_FIX_KEY] = gps_fix_pattern if extra_patterns: for ep in extra_patterns: all_patterns[ep] = ep def _log(msg: str) -> None: if _verbose: print(msg) # --- Phase 1: parse header to find variable IDs --- id_to_col: dict[int, str] = {} header_lines = 0 with open(path, "r") as f: for line in f: m = _HEADER_RE.match(line) if m: cpp_name = m.group(1) var_id = int(m.group(2)) for col_name, pattern in all_patterns.items(): if pattern in cpp_name: id_to_col[var_id] = col_name break header_lines += 1 elif header_lines > 0: break else: header_lines += 1 if "latitude" not in id_to_col.values() or "longitude" not in id_to_col.values(): _log(f"[quick_gps] {path.name}: lat/lon variables not found in header") return None target_ids = set(id_to_col.keys()) # --- Phase 2: stream data, keep only matching IDs --- raw: dict[str, tuple[list[float], list[float]]] = { col: ([], []) for col in id_to_col.values() } with open(path, "r") as f: for _ in range(header_lines): next(f) for line in f: parts = line.split(",", 2) if len(parts) < 3: continue try: var_id = int(parts[1]) except ValueError: continue if var_id not in target_ids: continue col_name = id_to_col[var_id] ts = float(parts[0]) val = float(parts[2]) raw[col_name][0].append(ts) raw[col_name][1].append(val) lat_ts = np.asarray(raw["latitude"][0], dtype=np.float64) lat_vals = np.asarray(raw["latitude"][1], dtype=np.float64) if len(lat_ts) == 0: _log(f"[quick_gps] {path.name}: no GPS data rows found") return None lon_ts_arr = np.asarray(raw["longitude"][0], dtype=np.float64) lon_raw = np.asarray(raw["longitude"][1], dtype=np.float64) lon_interp = np.interp(lat_ts, lon_ts_arr, lon_raw) # Filter by GPS fix status before anything else. # gpsFix is a discrete flag (0=no fix, 3=3D fix) so we use # nearest-neighbor interpolation to map it onto lat timestamps. if _GPS_FIX_KEY in raw and len(raw[_GPS_FIX_KEY][0]) > 0: fix_ts = np.asarray(raw[_GPS_FIX_KEY][0], dtype=np.float64) fix_vals = np.asarray(raw[_GPS_FIX_KEY][1], dtype=np.float64) f_fix = interp1d( fix_ts, fix_vals, kind="nearest", bounds_error=False, fill_value=0.0, ) fix_at_lat = f_fix(lat_ts) no_fix = fix_at_lat < min_gps_fix n_no_fix = int(no_fix.sum()) if n_no_fix > 0: pct = 100.0 * n_no_fix / len(lat_ts) _log( f"[quick_gps] {path.name}: filtering " f"{n_no_fix}/{len(lat_ts)} ({pct:.1f}%) " f"pts with gpsFix < {min_gps_fix}" ) if n_no_fix == len(lat_ts): _log(f"[quick_gps] {path.name}: no points with valid GPS fix") return None keep_fix = ~no_fix lat_ts = lat_ts[keep_fix] lat_vals = lat_vals[keep_fix] lon_interp = lon_interp[keep_fix] n_total = len(lat_ts) # Filter: either value is zero, or out of physical range bad = ( (lat_vals == 0.0) | (lon_interp == 0.0) | (np.abs(lat_vals) > 90.0) | (np.abs(lon_interp) > 180.0) ) n_bad = int(bad.sum()) pct_bad = 100.0 * n_bad / n_total if n_bad == n_total: _log(f"[quick_gps] {path.name}: no valid GPS " f"({n_total} pts, 100% bad)") return None if n_bad > 0: _log( f"[quick_gps] {path.name}: WARNING " f"{n_bad}/{n_total} ({pct_bad:.1f}%) " f"invalid positions (zero or out-of-range) " f"-- trimming" ) valid = ~bad lat_ts = lat_ts[valid] lat_vals = lat_vals[valid] lon_interp = lon_interp[valid] # Convert timestamps to seconds divisor = detect_time_divisor(np.asarray([lat_ts[0], lat_ts[-1]], dtype=np.float64)) time_s = lat_ts / divisor cols: dict[str, np.ndarray] = { "latitude": lat_vals, "longitude": lon_interp, } for col_name in raw: if col_name in ("latitude", "longitude", _GPS_FIX_KEY): continue extra_ts = np.asarray(raw[col_name][0], dtype=np.float64) extra_vals = np.asarray(raw[col_name][1], dtype=np.float64) if len(extra_ts) > 0: cols[col_name] = np.interp(lat_ts, extra_ts, extra_vals) df = pd.DataFrame(cols, index=pd.Index(time_s, name="time_s")) duration = time_s[-1] - time_s[0] n_kept = len(df) suffix = f" (trimmed from {n_total})" if n_bad > 0 else "" _log(f"[quick_gps] {path.name}: " f"{n_kept} pts, {duration:.1f} s{suffix}") return df
def _worker_load_gps( path: str, lat_pattern: str, lon_pattern: str, extra_patterns: list[str] | None, ) -> tuple[str, pd.DataFrame | None]: """Subprocess entry point for parallel GPS loading.""" name = Path(path).name df = quick_load_gps( path, lat_pattern=lat_pattern, lon_pattern=lon_pattern, extra_patterns=extra_patterns, _verbose=False, ) return name, df
[docs] def quick_load_folder( folder: str | Path, lat_pattern: str = "posLla.latitude", lon_pattern: str = "posLla.longitude", extra_patterns: list[str] | None = None, parallel: bool = False, max_workers: int | None = None, ) -> dict[str, pd.DataFrame]: """Batch-load GPS previews for all PER CSV logs in a folder. Parameters ---------- folder : str or Path Directory containing ``.csv`` log files. lat_pattern, lon_pattern : str Passed to :func:`quick_load_gps`. extra_patterns : list[str] or None Passed to :func:`quick_load_gps`. parallel : bool If ``True``, load files in parallel using ``ProcessPoolExecutor``. max_workers : int or None Max worker processes when *parallel=True*. Defaults to ``os.cpu_count() - 2`` (minimum 1). Returns ------- dict[str, pd.DataFrame] Keyed by filename (not full path). Logs with no GPS lock are omitted. """ print( """ DEPRECATION WARNING: quick_load_gps is deprecated and will be removed in a future release. Please reinstall PERDA. The latest version features significant improvements to log loading speed. For maximum performance, install perda[notebook] instead of perda[full]. """ ) folder = Path(folder) csv_files = sorted(folder.glob("*.csv")) if not csv_files: print(f"[quick_gps] No .csv files found in {folder}") return {} if parallel and max_workers is None: max_workers = max(1, (os.cpu_count() or 4) - 2) mode = f"parallel, {max_workers} workers" if parallel else "sequential" print(f"[quick_gps] Scanning {len(csv_files)} file(s) " f"in {folder} ({mode})\n") results: dict[str, pd.DataFrame] = {} no_gps: list[str] = [] if parallel: future_to_name: dict = {} with ProcessPoolExecutor(max_workers=max_workers) as pool: for csv_path in csv_files: name = csv_path.name print(f"[quick_gps] Spawned process: {name}") fut = pool.submit( _worker_load_gps, str(csv_path), lat_pattern, lon_pattern, extra_patterns, ) future_to_name[fut] = name for fut in as_completed(future_to_name): name, df = fut.result() if df is not None: dur = df.index[-1] - df.index[0] print(f"[quick_gps] Loaded {name}: " f"{len(df)} pts, {dur:.1f} s") results[name] = df else: print(f"[quick_gps] Done {name}: " f"no usable GPS data") no_gps.append(name) else: for csv_path in csv_files: df = quick_load_gps( csv_path, lat_pattern=lat_pattern, lon_pattern=lon_pattern, extra_patterns=extra_patterns, ) if df is not None: results[csv_path.name] = df else: no_gps.append(csv_path.name) print(f"\n[quick_gps] Summary: {len(results)}/{len(csv_files)} logs with GPS data") if no_gps: print(f" Skipped (no GPS): {', '.join(no_gps)}") if results: print(f" {'File':<35s} {'Points':>8s} {'Duration':>10s}") print(f" {'-'*35} {'-'*8} {'-'*10}") for name, df in results.items(): dur = df.index[-1] - df.index[0] print(f" {name:<35s} {len(df):>8d} {dur:>9.1f}s") return results
# --------------------------------------------------------------------------- # PERDA-based GPS loader (uses Analyzer instead of custom CSV parsing) # ---------------------------------------------------------------------------
[docs] def perda_load_gps( path: str | Path, lat_var: str = "pcm.vnav.posLla.latitude", lon_var: str = "pcm.vnav.posLla.longitude", gps_fix_var: str = "pcm.vnav.gpsFix", min_gps_fix: float = 1.0, extra_vars: list[str] | None = None, deduplicate: bool = True, ) -> pd.DataFrame | None: """Load GPS data from a PER CSV log using PERDA's Analyzer. Parameters ---------- path : str or Path Path to a PER-format ``.csv`` log file. lat_var, lon_var : str Full CAN variable names for latitude and longitude. gps_fix_var : str CAN variable name for GPS fix status. min_gps_fix : float Minimum GPS fix value to keep (0 = no fix, 3 = 3D fix). extra_vars : list[str] or None Additional CAN variables to extract alongside GPS. deduplicate : bool Collapse consecutive duplicate GPS positions to remove ZOH-inflated samples (GPS updates at ~4 Hz but is logged at CAN rate ~100 Hz). Returns ------- pd.DataFrame or None DataFrame indexed by ``time_s`` with ``latitude``, ``longitude`` columns (plus extras). Returns ``None`` if the file has no usable GPS data. """ print( """ DEPRECATION WARNING: perda_load_gps is deprecated and will be removed in a future release. Similar functionality is available via the suboptimumg.log_analysis.preprocess_gps.preprocess_gps_data_instances, which has the same functionality and avoids the need to convert to pandas DataFrames. Please also reference PERDA documentation for additional DataInstance preprocessing utilities for filtering and interpolation. """ ) from perda.analyzer import Analyzer path = Path(path) try: aly = Analyzer(str(path)) except Exception as e: print(f"[perda_gps] {path.name}: failed to parse — {e}") return None data = aly.data if lat_var not in data or lon_var not in data: print(f"[perda_gps] {path.name}: GPS variables not found") return None lat_di = data[lat_var] lon_di = data[lon_var] lat_ts = lat_di.timestamp_np.astype(np.float64) lat_vals = lat_di.value_np.copy() if len(lat_ts) == 0: print(f"[perda_gps] {path.name}: no GPS data points") return None # Interpolate longitude onto latitude timestamps lon_interp = np.interp( lat_ts, lon_di.timestamp_np.astype(np.float64), lon_di.value_np, ) # Deduplicate consecutive identical GPS positions if deduplicate: keep = np.concatenate( ( [True], (lat_vals[1:] != lat_vals[:-1]) | (lon_interp[1:] != lon_interp[:-1]), ) ) n_before = len(lat_ts) lat_ts = lat_ts[keep] lat_vals = lat_vals[keep] lon_interp = lon_interp[keep] n_after = len(lat_ts) if n_before > n_after: print( f"[perda_gps] {path.name}: dedup {n_before}{n_after} " f"({100 * (n_before - n_after) / n_before:.1f}% removed)" ) # Filter by GPS fix status (nearest-neighbor since it's a discrete flag) if gps_fix_var and gps_fix_var in data: fix_di = data[gps_fix_var] fix_ts = fix_di.timestamp_np.astype(np.float64) fix_vals = fix_di.value_np if len(fix_ts) > 0: f_fix = interp1d( fix_ts, fix_vals, kind="nearest", bounds_error=False, fill_value=0.0, ) fix_at_lat = f_fix(lat_ts) no_fix = fix_at_lat < min_gps_fix n_no_fix = int(no_fix.sum()) if n_no_fix > 0: pct = 100.0 * n_no_fix / len(lat_ts) print( f"[perda_gps] {path.name}: filtering " f"{n_no_fix}/{len(lat_ts)} ({pct:.1f}%) " f"pts with gpsFix < {min_gps_fix}" ) if n_no_fix == len(lat_ts): print(f"[perda_gps] {path.name}: no points with valid GPS fix") return None keep_fix = ~no_fix lat_ts = lat_ts[keep_fix] lat_vals = lat_vals[keep_fix] lon_interp = lon_interp[keep_fix] n_total = len(lat_ts) # Filter zero or out-of-range positions bad = ( (lat_vals == 0.0) | (lon_interp == 0.0) | (np.abs(lat_vals) > 90.0) | (np.abs(lon_interp) > 180.0) ) n_bad = int(bad.sum()) if n_bad == n_total: print(f"[perda_gps] {path.name}: no valid GPS ({n_total} pts, 100% bad)") return None if n_bad > 0: pct_bad = 100.0 * n_bad / n_total print( f"[perda_gps] {path.name}: WARNING " f"{n_bad}/{n_total} ({pct_bad:.1f}%) " f"invalid positions — trimming" ) valid = ~bad lat_ts = lat_ts[valid] lat_vals = lat_vals[valid] lon_interp = lon_interp[valid] # Convert timestamps to seconds divisor = detect_time_divisor(np.asarray([lat_ts[0], lat_ts[-1]], dtype=np.float64)) time_s = lat_ts / divisor cols: dict[str, np.ndarray] = { "latitude": lat_vals, "longitude": lon_interp, } # Interpolate extra variables onto GPS timestamps if extra_vars: for var_name in extra_vars: if var_name not in data: continue di = data[var_name] if len(di.timestamp_np) == 0: continue cols[var_name] = np.interp( lat_ts, di.timestamp_np.astype(np.float64), di.value_np, ) df = pd.DataFrame(cols, index=pd.Index(time_s, name="time_s")) duration = time_s[-1] - time_s[0] n_kept = len(df) suffix = f" (trimmed from {n_total})" if n_bad > 0 else "" print(f"[perda_gps] {path.name}: {n_kept} pts, {duration:.1f} s{suffix}") return df
[docs] def perda_load_folder( folder: str | Path, extra_vars: list[str] | None = None, deduplicate: bool = True, lat_var: str = "pcm.vnav.posLla.latitude", lon_var: str = "pcm.vnav.posLla.longitude", gps_fix_var: str = "pcm.vnav.gpsFix", min_gps_fix: float = 1.0, ) -> dict[str, pd.DataFrame]: """Batch-load GPS data from all PER CSV logs using PERDA. Uses PERDA's ``Analyzer`` for CSV parsing instead of the custom lightweight parser in :func:`quick_load_gps`. Parameters ---------- folder : str or Path Directory containing ``.csv`` log files. extra_vars : list[str] or None Full CAN variable names to extract alongside GPS. deduplicate : bool Collapse consecutive duplicate GPS positions. lat_var, lon_var : str Full CAN variable names for latitude and longitude. gps_fix_var : str CAN variable name for GPS fix status. min_gps_fix : float Minimum GPS fix value to keep. Returns ------- dict[str, pd.DataFrame] Keyed by filename (not full path). Logs with no GPS lock are omitted. """ print( """ DEPRECATION WARNING: perda_load_folder is deprecated and will be removed in a future release. Similar functionality is available via the suboptimumg.log_analysis.preprocess_gps.preprocess_gps_data_instances, which has the same functionality and avoids the need to convert to pandas DataFrames. Please also reference PERDA documentation for additional DataInstance preprocessing utilities for filtering and interpolation. """ ) folder = Path(folder) csv_files = sorted(folder.glob("*.csv")) if not csv_files: print(f"[perda_gps] No .csv files found in {folder}") return {} print(f"[perda_gps] Loading {len(csv_files)} file(s) in {folder}\n") results: dict[str, pd.DataFrame] = {} no_gps: list[str] = [] for csv_path in csv_files: df = perda_load_gps( csv_path, lat_var=lat_var, lon_var=lon_var, gps_fix_var=gps_fix_var, min_gps_fix=min_gps_fix, extra_vars=extra_vars, deduplicate=deduplicate, ) if df is not None: results[csv_path.name] = df else: no_gps.append(csv_path.name) print(f"\n[perda_gps] Summary: {len(results)}/{len(csv_files)} logs with GPS data") if no_gps: print(f" Skipped (no GPS): {', '.join(no_gps)}") if results: print(f" {'File':<35s} {'Points':>8s} {'Duration':>10s}") print(f" {'-'*35} {'-'*8} {'-'*10}") for name, df in results.items(): dur = df.index[-1] - df.index[0] print(f" {name:<35s} {len(df):>8d} {dur:>9.1f}s") return results