"""Log introspection utilities: variable enumeration, frequency analysis, config loading."""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
import numpy as np
import pandas as pd
import yaml
from ._utils import detect_time_divisor
if TYPE_CHECKING:
from perda.analyzer import Analyzer
[docs]
def dump_variables(analyzer: Analyzer) -> pd.DataFrame:
"""Enumerate every variable in a loaded PERDA log.
Variable IDs are intentionally excluded because they are
log-internal and must not be used for access.
Parameters
----------
analyzer : Analyzer
A loaded PERDA Analyzer instance.
Returns
-------
pd.DataFrame
Columns: cpp_name, num_points, start_time, end_time,
duration_s, min_value, max_value, median_freq_hz.
"""
print(
"""
DEPRECATION WARNING: dump_variables is deprecated and will be removed in a future release.
Please check the latest PERDA documentation. Use analyzer.search() to find relevant variables.
Easily retrieve variables by providing either index or CPP name. Please also check GitHub releases
to see documentation about the variables available in the latest version of logs.
"""
)
rows: list[dict] = []
data = analyzer.data
# Auto-detect timestamp unit from the global log timespan
all_ts = np.array([data.data_start_time, data.data_end_time], dtype=np.int64)
divisor = detect_time_divisor(all_ts)
for var_id, di in data.id_to_instance.items():
ts = di.timestamp_np
vals = di.value_np
n = len(ts)
if n == 0:
continue
cpp_name = data.id_to_cpp_name.get(var_id, f"unknown_{var_id}")
start_t = int(ts[0])
end_t = int(ts[-1])
duration_s = (end_t - start_t) / divisor
med_freq = 0.0
if n > 1:
diffs = np.diff(ts).astype(np.float64)
pos_diffs = diffs[diffs > 0]
if len(pos_diffs) > 0:
med_freq = divisor / np.median(pos_diffs)
rows.append(
{
"cpp_name": cpp_name,
"num_points": n,
"start_time": start_t,
"end_time": end_t,
"duration_s": round(duration_s, 3),
"min_value": float(np.nanmin(vals)),
"max_value": float(np.nanmax(vals)),
"median_freq_hz": round(med_freq, 2),
}
)
result = pd.DataFrame(rows)
if not result.empty:
result = result.sort_values("cpp_name", ignore_index=True)
return result
[docs]
def logging_frequencies(analyzer: Analyzer, var_names: list[str]) -> pd.DataFrame:
"""Compute sample-rate statistics for a list of variables.
Parameters
----------
analyzer : Analyzer
A loaded PERDA Analyzer instance.
var_names : list[str]
CAN variable names (``cpp_name``) to analyse.
Returns
-------
pd.DataFrame
Columns: var_name, median_freq_hz, mean_freq_hz, min_freq_hz,
max_freq_hz, num_samples, num_gaps.
``num_gaps`` counts intervals exceeding 3x the median interval,
indicating dropped samples.
"""
print(
"""
DEPRECATION WARNING: logging_frequencies is deprecated and will be removed in a future release.
Use analyzer.analyze_frequency(var_name) for rich per-variable frequency diagnostics,
including a Plotly figure with instantaneous frequency over time and a histogram.
"""
)
rows: list[dict] = []
data = analyzer.data
# Auto-detect timestamp unit from the global log timespan
all_ts = np.array([data.data_start_time, data.data_end_time], dtype=np.int64)
divisor = detect_time_divisor(all_ts)
for name in var_names:
di = data[name]
ts = di.timestamp_np
n = len(ts)
if n < 2:
rows.append(
{
"var_name": name,
"median_freq_hz": 0.0,
"mean_freq_hz": 0.0,
"min_freq_hz": 0.0,
"max_freq_hz": 0.0,
"num_samples": n,
"num_gaps": 0,
}
)
continue
diffs = np.diff(ts).astype(np.float64)
pos_diffs = diffs[diffs > 0]
if len(pos_diffs) == 0:
rows.append(
{
"var_name": name,
"median_freq_hz": np.inf,
"mean_freq_hz": np.inf,
"min_freq_hz": np.inf,
"max_freq_hz": np.inf,
"num_samples": n,
"num_gaps": 0,
}
)
continue
med = np.median(pos_diffs)
rows.append(
{
"var_name": name,
"median_freq_hz": round(divisor / med, 2),
"mean_freq_hz": round(divisor / np.mean(pos_diffs), 2),
"min_freq_hz": round(divisor / np.max(pos_diffs), 2),
"max_freq_hz": round(divisor / np.min(pos_diffs), 2),
"num_samples": n,
"num_gaps": int(np.sum(pos_diffs > 3.0 * med)),
}
)
return pd.DataFrame(rows)
[docs]
def sample_log(
df: pd.DataFrame,
patterns: list[str] | None = None,
percentiles: list[int] | None = None,
) -> None:
"""Print sampled values at key points through the log for debugging.
By default matches columns containing 'pos', 'lat', 'latitude',
'lon', 'long', or 'longitude' (case-insensitive).
Parameters
----------
df : pd.DataFrame
The log DataFrame (output of ``perda_to_dataframe``).
patterns : list[str] or None
Substrings to match against column names.
percentiles : list[int] or None
Which percentile indices to sample. Default: [0, 25, 50, 75, 100].
"""
print(
"""
DEPRECATION WARNING: sample_log is deprecated and will be removed in a future release.
PERDA's primary API surfaces DataInstance objects rather than flat DataFrames.
Use analyzer.plot(var_name) for interactive visualization, or access
DataInstance.value_np / DataInstance.timestamp_np directly for programmatic inspection.
"""
)
if patterns is None:
patterns = ["pos", "lat", "lon"]
if percentiles is None:
percentiles = [0, 25, 50, 75, 100]
matched = [col for col in df.columns if any(p in col.lower() for p in patterns)]
if not matched:
print(f"No columns matched patterns: {patterns}")
print(f"Available columns ({len(df.columns)}): {list(df.columns)[:20]}...")
return
n = len(df)
idx_name = df.index.name or "index"
print(f"Sampling {len(matched)} column(s) at {percentiles}% ({n} total rows):\n")
for col in matched:
vals = df[col].values
nonzero = int(
np.count_nonzero(
vals[~np.isnan(vals)]
if np.issubdtype(vals.dtype, np.floating)
else vals
)
)
print(f" {col} (non-zero: {nonzero}/{n}, {100*nonzero/n:.1f}%)")
for p in percentiles:
idx = min(int(n * p / 100), n - 1)
t_val = df.index[idx]
print(f" {p:3d}% {idx_name}={t_val:>10.3f} val={vals[idx]}")
print()
def _process_curves(obj):
"""Recursively convert ``{data, poly_order}`` dicts to FittedCurve."""
print(
"""
DEPRECATION WARNING: _process_curves is deprecated and will be removed in a future release.
Use suboptimumg.vehicle.yaml.load_car_from_yaml instead, which can load both a Car and an IrlCar,
and will automatically convert any curve entries in the vehicle setup section to FittedCurve objects.
"""
)
from .fitted_curve import FittedCurve
if isinstance(obj, dict):
if "data" in obj and "poly_order" in obj:
return FittedCurve(obj["data"], obj["poly_order"])
return {k: _process_curves(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_process_curves(item) for item in obj]
return obj
[docs]
def load_vehicle_setup(path: str | Path = "parameters/vehicle_setup.yaml") -> dict:
"""Load a vehicle_setup.yaml configuration file.
Any YAML mapping containing both ``data`` (list of ``[x, y]``
pairs) and ``poly_order`` (int) is automatically converted to a
:class:`~suboptimumg.loganalysis.fitted_curve.FittedCurve`.
Everything else passes through as plain Python types.
Parameters
----------
path : str or Path
Path to the YAML file.
Returns
-------
dict
Parsed YAML contents with curve entries as FittedCurve objects.
Typical top-level keys: ``alignment``, ``steering``,
``suspension``.
"""
print(
"""
DEPRECATION WARNING: load_vehicle_setup is deprecated and will be removed in a future release.
Use suboptimumg.vehicle.yaml.load_car_from_yaml instead, which can load both a Car and an IrlCar,
depending on the presence of vehicle setup parameters.
"""
)
path = Path(path)
if not path.exists():
raise FileNotFoundError(f"Vehicle setup file not found: {path}")
with open(path, "r") as f:
raw = yaml.safe_load(f)
return _process_curves(raw)