"""GPS visualization, trimming, coordinate transforms, and distance computation."""
from __future__ import annotations
import ipywidgets as widgets
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from ._utils import detect_time_divisor, downsample_indices, get_time_array
R_EARTH = 6_371_000 # mean Earth radius in meters
# ---------------------------------------------------------------------------
# DataFrame helpers
# ---------------------------------------------------------------------------
def _ensure_list(dfs: pd.DataFrame | list[pd.DataFrame]) -> list[pd.DataFrame]:
if isinstance(dfs, pd.DataFrame):
return [dfs]
return list(dfs)
[docs]
def resolve_gps_columns(
dfs: pd.DataFrame | list[pd.DataFrame],
ins_lat_col: str = "pcm.vnav.posLla.latitude",
ins_lon_col: str = "pcm.vnav.posLla.longitude",
fix_col: str | None = "pcm.vnav.gpsFix",
out_lat_col: str = "latitude",
out_lon_col: str = "longitude",
) -> pd.DataFrame | list[pd.DataFrame]:
"""Build unified latitude/longitude columns with INS-lock fallback.
The VN-300 INS solution (``posLla``) reads zero when the INS hasn't
converged. This function creates clean ``latitude``/``longitude``
columns that mask out invalid (zero) samples and, when the INS lock
is eventually acquired, use the INS-filtered position.
When ``fix_col`` is provided and present in the DataFrame, rows
where ``gpsFix == 0`` are also treated as invalid.
Invalid rows are set to ``NaN`` so downstream functions can handle
them (e.g. drop, interpolate, or skip).
Parameters
----------
dfs : DataFrame or list[DataFrame]
ins_lat_col, ins_lon_col : str
INS solution position columns (the only position variables in
these CAN logs).
fix_col : str or None
GPS fix status column. ``None`` skips the fix check.
out_lat_col, out_lon_col : str
Names for the output columns.
Returns
-------
Same type as input, with *out_lat_col* and *out_lon_col* added.
"""
print(
"""
DEPRECATION WARNING: resolve_gps_columns is deprecated and will be removed in a future release.
Please use suboptimumg.log_analysis.preprocess_gps.preprocess_gps_data_instances instead, which
has the same functionality and avoids the need to convert to pandas DataFrames.
"""
)
df_list = _ensure_list(dfs)
single = isinstance(dfs, pd.DataFrame)
for df in df_list:
lat = df[ins_lat_col].values.copy().astype(np.float64)
lon = df[ins_lon_col].values.copy().astype(np.float64)
invalid = (lat == 0.0) & (lon == 0.0)
if fix_col and fix_col in df.columns:
no_fix = df[fix_col].values == 0.0
invalid = invalid | no_fix
n_invalid = int(invalid.sum())
n_total = len(lat)
lat[invalid] = np.nan
lon[invalid] = np.nan
df[out_lat_col] = lat
df[out_lon_col] = lon
n_valid = n_total - n_invalid
print(
f"[gps] resolve_gps_columns: {n_valid}/{n_total} valid "
f"({100 * n_valid / n_total:.1f}%), {n_invalid} masked to NaN"
)
print(f"[gps] Added columns: {out_lat_col}, {out_lon_col}")
return df_list[0] if single else df_list
[docs]
def gps_to_local_cartesian(
dfs: pd.DataFrame | list[pd.DataFrame],
lat_col: str = "pcm.vnav.posLla.latitude",
lon_col: str = "pcm.vnav.posLla.longitude",
origin: tuple[float, float] | None = None,
) -> pd.DataFrame | list[pd.DataFrame]:
"""Equirectangular projection from lat/lon to local North/East (meters).
Adds ``posN`` and ``posE`` columns. If *origin* is ``None``, uses
the first data point of the first DataFrame as the reference.
Parameters
----------
dfs : DataFrame or list[DataFrame]
lat_col, lon_col : str
origin : (lat_deg, lon_deg) or None
Returns
-------
Same type as input, with ``posN`` and ``posE`` added.
"""
print(
"""
DEPRECATION WARNING: gps_to_local_cartesian is deprecated and will be removed in a future release.
Please use suboptimumg.track.gps.gps_to_local_xy instead, which has the same functionality
but avoids the need to convert to pandas DataFrames.
"""
)
df_list = _ensure_list(dfs)
single = isinstance(dfs, pd.DataFrame)
if origin is None:
first = df_list[0]
origin = (first[lat_col].iloc[0], first[lon_col].iloc[0])
lat_ref = np.radians(origin[0])
lon_ref = np.radians(origin[1])
added_cols: list[str] = []
for df in df_list:
df["posN"] = (np.radians(df[lat_col]) - lat_ref) * R_EARTH
df["posE"] = (np.radians(df[lon_col]) - lon_ref) * R_EARTH * np.cos(lat_ref)
added_cols = ["posN", "posE"]
print(
f"[gps] Added/updated columns: {added_cols} ({len(df_list)} df(s), {len(df_list[0])} rows each)"
)
return df_list[0] if single else df_list
[docs]
def compute_elapsed_distance(
dfs: pd.DataFrame | list[pd.DataFrame],
pos_n_col: str = "posN",
pos_e_col: str = "posE",
out_col: str = "distance",
) -> pd.DataFrame | list[pd.DataFrame]:
"""Cumulative arc-length distance from local cartesian position columns.
Parameters
----------
dfs : DataFrame or list[DataFrame]
pos_n_col, pos_e_col : str
Local cartesian position column names (must already exist).
out_col : str
Name of the new distance column.
Returns
-------
Same type as input, with *out_col* added.
"""
df_list = _ensure_list(dfs)
single = isinstance(dfs, pd.DataFrame)
for df in df_list:
dn = np.diff(df[pos_n_col].values)
de = np.diff(df[pos_e_col].values)
ds = np.sqrt(dn**2 + de**2)
df[out_col] = np.concatenate(([0.0], np.cumsum(ds)))
print(
f"[gps] Added column: {out_col} ({len(df_list)} df(s), {len(df_list[0])} rows each)"
)
return df_list[0] if single else df_list
[docs]
def prepare_gps_data(
df: pd.DataFrame,
lat_col: str = "latitude",
lon_col: str = "longitude",
max_radius_m: float = 5_000.0,
max_jump_m: float = 30.0,
) -> pd.DataFrame:
"""Full GPS cleaning pipeline: filter invalid positions, project to
local cartesian, and remove spatial outliers.
Steps applied in order:
1. Drop rows where lat/lon are zero or outside physical range
(``|lat| > 90`` or ``|lon| > 180``).
2. Equirectangular projection to local East/North meters, with the
median lat/lon as origin.
3. Drop points farther than *max_radius_m* from the origin
(warns if any are removed).
4. Drop points that jump more than *max_jump_m* from their
predecessor (warns if any are removed).
ZOH deduplication of stale GPS samples should be handled
upstream via ``perda_to_dataframe(deduplicate_vars=...)``.
Adds ``posE`` (East, meters) and ``posN`` (North, meters) columns
to the returned DataFrame.
Parameters
----------
df : pd.DataFrame
Must contain *lat_col* and *lon_col* columns.
lat_col, lon_col : str
Column names for latitude and longitude in degrees.
max_radius_m : float
Maximum distance from the median position to keep.
max_jump_m : float
Maximum point-to-point distance to keep.
Returns
-------
pd.DataFrame
Cleaned copy with ``posE`` and ``posN`` columns added and
invalid rows removed.
"""
print(
"""
DEPRECATION WARNING: prepare_gps_data is deprecated and will be removed in a future release.
Please use suboptimumg.track.gps_to_local_xy and suboptimumg.track.clean_gps_xy instead,
which have the same functionality but avoid the need to convert to pandas DataFrames.
"""
)
n_start = len(df)
lat = df[lat_col].values.astype(np.float64)
lon = df[lon_col].values.astype(np.float64)
# Step 1 — zero / out-of-range filter
valid = (
(lat != 0.0)
& (lon != 0.0)
& (np.abs(lat) <= 90.0)
& (np.abs(lon) <= 180.0)
& np.isfinite(lat)
& np.isfinite(lon)
)
n_invalid = int((~valid).sum())
if n_invalid == n_start:
print(
f"[gps] prepare_gps_data: all {n_start} rows invalid (zero/NaN/out-of-range)"
)
return df.iloc[0:0].copy()
if n_invalid > 0:
print(
f"[gps] prepare_gps_data: dropped {n_invalid}/{n_start} "
f"invalid positions (zero/NaN/out-of-range)"
)
out = df[valid].copy()
lat = out[lat_col].values.astype(np.float64)
lon = out[lon_col].values.astype(np.float64)
# Step 2 — project to local XY with median as origin
lat_ref = np.radians(np.median(lat))
lon_ref = np.radians(np.median(lon))
pos_e = (np.radians(lon) - lon_ref) * R_EARTH * np.cos(lat_ref)
pos_n = (np.radians(lat) - lat_ref) * R_EARTH
out["posE"] = pos_e
out["posN"] = pos_n
# Step 3 — radius filter from origin (median)
# dist = np.sqrt(pos_e**2 + pos_n**2)
# keep = dist <= max_radius_m
# n_far = int((~keep).sum())
# if n_far > 0:
# print(
# f"[gps] prepare_gps_data: WARNING — dropped {n_far}/{len(out)} "
# f"points beyond {max_radius_m:.0f} m from median"
# )
# out = out[keep].copy()
# if out.empty:
# print("[gps] prepare_gps_data: no points remain " "after radius filter")
# return out
# pos_e = out["posE"].values
# pos_n = out["posN"].values
# Step 4 — jump filter (large point-to-point gaps)
dx = np.diff(pos_e, prepend=pos_e[0])
dy = np.diff(pos_n, prepend=pos_n[0])
step = np.sqrt(dx**2 + dy**2)
keep = step <= max_jump_m
n_jump = int((~keep).sum())
if n_jump > 0:
print(
f"[gps] prepare_gps_data: dropped {n_jump}/{len(out)} "
f"points with jumps > {max_jump_m:.0f} m"
)
out = out[keep].copy()
n_final = len(out)
print(
f"[gps] prepare_gps_data: {n_final}/{n_start} points kept, "
f"added posE/posN columns"
)
return out
[docs]
def trim_dataframe(
df: pd.DataFrame,
time_col: str,
start_time: float,
end_time: float,
) -> pd.DataFrame:
"""Return rows where *time_col* is between *start_time* and *end_time* (inclusive)."""
print(
"""
DEPRECATION WARNING: trim_dataframe is deprecated and will be removed in a future release.
Please reinstall PERDA and check the documentation for DataInstance.trim() which natively
provides equivalent functionality.
"""
)
t = get_time_array(df, time_col)
mask = (t >= start_time) & (t <= end_time)
trimmed = df.loc[mask].copy()
print(
f"[gps] Trimmed: {len(df)} -> {len(trimmed)} rows (t=[{start_time:.2f}, {end_time:.2f}])"
)
return trimmed
[docs]
def plot_gps_trajectory(
df: pd.DataFrame,
lat_col: str = "pcm.vnav.posLla.latitude",
lon_col: str = "pcm.vnav.posLla.longitude",
time_col: str = "time_s",
color_by: str = "time",
max_display_hz: float = 100.0,
width: int = 700,
height: int = 700,
) -> tuple[go.FigureWidget, widgets.VBox]:
"""Interactive 2D GPS trajectory plot with time-based trim sliders.
Parameters
----------
df : pd.DataFrame
lat_col, lon_col : str
time_col : str
Column used for slider values and colour mapping.
color_by : str
``"time"`` colours by *time_col*.
max_display_hz : float
Target sample rate for display. Data is naively thinned (every Nth
point) to approximately this rate. The source DataFrame is not modified.
width, height : int
Returns
-------
(fig, widget_box)
``fig`` is the FigureWidget, ``widget_box`` is the VBox with
sliders that can be ``display()``-ed.
"""
lat_full = df[lat_col].to_numpy()
lon_full = df[lon_col].to_numpy()
time_full = get_time_array(df, time_col)
divisor = detect_time_divisor(
np.asarray([time_full[0], time_full[-1]], dtype=np.float64)
)
duration_s = (time_full[-1] - time_full[0]) / divisor
idx = downsample_indices(len(lat_full), duration_s, max_display_hz)
lat = lat_full[idx]
lon = lon_full[idx]
time = time_full[idx]
n = len(lat)
t_min, t_max = float(time[0]), float(time[-1])
def _fmt(t: float) -> str:
return f"{t / divisor:.2f}"
fig = go.FigureWidget()
fig.add_trace(
go.Scattergl(
x=lon,
y=lat,
mode="markers",
marker=dict(
size=4,
color=time,
colorscale="Viridis",
colorbar=dict(title="Time"),
showscale=True,
),
name="Trajectory",
)
)
fig.update_layout(
title="GPS Trajectory",
xaxis_title="Longitude",
yaxis_title="Latitude",
width=width,
height=height,
)
fig.update_yaxes(scaleanchor="x", scaleratio=1)
step = (t_max - t_min) / max(n - 1, 1)
slider_start = widgets.FloatSlider(
value=t_min,
min=t_min,
max=t_max,
step=step,
description="Start:",
continuous_update=True,
readout=False,
layout=widgets.Layout(width="100%"),
)
slider_end = widgets.FloatSlider(
value=t_max,
min=t_min,
max=t_max,
step=step,
description="End:",
continuous_update=True,
readout=False,
layout=widgets.Layout(width="100%"),
)
label = widgets.Label(value=f" {_fmt(t_min)} -> {_fmt(t_max)} ({n} samples)")
def _update(change):
s_val = slider_start.value
e_val = slider_end.value
if s_val >= e_val:
return
mask = (time >= s_val) & (time <= e_val)
with fig.batch_update():
fig.data[0].x = lon[mask]
fig.data[0].y = lat[mask]
fig.data[0].marker.color = time[mask]
count = int(mask.sum())
label.value = (
f" {_fmt(s_val)} -> {_fmt(e_val)} "
f"({count} samples, {(e_val - s_val) / divisor:.1f} s)"
)
slider_start.observe(_update, names="value")
slider_end.observe(_update, names="value")
box = widgets.VBox(
[fig, slider_start, slider_end, label],
layout=widgets.Layout(width=f"{width}px"),
)
return fig, box