From 7dcefd441d01085421186fa976d0ba4314026267 Mon Sep 17 00:00:00 2001 From: Jeffrey-Moon Date: Fri, 17 Jul 2026 17:48:34 -0400 Subject: [PATCH 1/6] Plotting Max Number of Points Downsampling --- perda/plotting/data_instance_plotter.py | 45 ++++++++++++++++++++----- 1 file changed, 36 insertions(+), 9 deletions(-) diff --git a/perda/plotting/data_instance_plotter.py b/perda/plotting/data_instance_plotter.py index a692677..99f708d 100644 --- a/perda/plotting/data_instance_plotter.py +++ b/perda/plotting/data_instance_plotter.py @@ -45,6 +45,7 @@ def plot_single_axis( timestamp_unit: Timescale = Timescale.MS, vlines: List[float] | None = None, vline_config: "VLineConfig" = DEFAULT_VLINE_CONFIG, + max_points: int | None = None, ) -> go.Figure: """ Plot one or more DataInstances on a single y-axis using Plotly. @@ -67,6 +68,8 @@ def plot_single_axis( X-axis positions (in seconds) where vertical lines are drawn. Default is None. vline_config : VLineConfig, optional Visual configuration for the vertical lines. + max_points : int | None, optional + Maximum number of data points to plot per trace (enables zero-copy step downsampling). Default is None (full fidelity). Returns ------- @@ -88,13 +91,20 @@ def plot_single_axis( print(f"Warning: No data points in DataInstance for {di.label}") continue - # Convert timestamps from the log unit to seconds for plotting. - timestamps_s = _to_seconds(di.timestamp_np.astype(float), timestamp_unit) + # Convert timestamps from the log unit to seconds for plotting with optional downsampling. + n_points = len(di) + if max_points is not None and n_points > max_points and max_points > 0: + step = n_points // max_points + timestamps_s = _to_seconds(di.timestamp_np[::step], timestamp_unit) + values_np = di.value_np[::step] + else: + timestamps_s = _to_seconds(di.timestamp_np, timestamp_unit) + values_np = di.value_np fig.add_trace( go.Scattergl( x=timestamps_s, - y=di.value_np, + y=values_np, mode="lines", name=di.label, ) @@ -139,6 +149,7 @@ def plot_dual_axis( timestamp_unit: Timescale = Timescale.MS, vlines: List[float] | None = None, vline_config: VLineConfig = DEFAULT_VLINE_CONFIG, + max_points: int | None = None, ) -> go.Figure: """ Plot DataInstances on dual y-axes using Plotly. @@ -167,6 +178,8 @@ def plot_dual_axis( X-axis positions (in seconds) where vertical lines are drawn. Default is None. vline_config : VLineConfig, optional Visual configuration for the vertical lines. + max_points : int | None, optional + Maximum number of data points to plot per trace (enables zero-copy step downsampling). Default is None (full fidelity). Returns ------- @@ -190,13 +203,20 @@ def plot_dual_axis( print(f"Warning: No data points in DataInstance for {di.label}") continue - # Convert timestamps from the log unit to seconds for plotting. - timestamps_s = _to_seconds(di.timestamp_np.astype(float), timestamp_unit) + # Convert timestamps from the log unit to seconds for plotting with optional downsampling. + n_points = len(di) + if max_points is not None and n_points > max_points and max_points > 0: + step = n_points // max_points + timestamps_s = _to_seconds(di.timestamp_np[::step], timestamp_unit) + values_np = di.value_np[::step] + else: + timestamps_s = _to_seconds(di.timestamp_np, timestamp_unit) + values_np = di.value_np fig.add_trace( go.Scattergl( x=timestamps_s, - y=di.value_np, + y=values_np, mode="lines", name=di.label, ), @@ -209,13 +229,20 @@ def plot_dual_axis( print(f"Warning: No data points in DataInstance for {di.label}") continue - # Convert timestamps from the log unit to seconds for plotting. - timestamps_s = _to_seconds(di.timestamp_np.astype(float), timestamp_unit) + # Convert timestamps from the log unit to seconds for plotting with optional downsampling. + n_points = len(di) + if max_points is not None and n_points > max_points and max_points > 0: + step = n_points // max_points + timestamps_s = _to_seconds(di.timestamp_np[::step], timestamp_unit) + values_np = di.value_np[::step] + else: + timestamps_s = _to_seconds(di.timestamp_np, timestamp_unit) + values_np = di.value_np fig.add_trace( go.Scattergl( x=timestamps_s, - y=di.value_np, + y=values_np, mode="lines", name=di.label, line=dict(dash="dash"), From 915fc7490b5b3538d54eb83c54c2a47c62800249 Mon Sep 17 00:00:00 2001 From: Jeffrey-Moon Date: Fri, 17 Jul 2026 17:49:50 -0400 Subject: [PATCH 2/6] Optimized Hybrid Search Using all-MiniLM-L6-V2 --- perda/utils/search.py | 133 +++++++++++++++++++++++------------------- 1 file changed, 72 insertions(+), 61 deletions(-) diff --git a/perda/utils/search.py b/perda/utils/search.py index e0faef3..1825457 100644 --- a/perda/utils/search.py +++ b/perda/utils/search.py @@ -8,17 +8,12 @@ from ..constants import DELIMITER, title_block from ..core_data_structures.single_run_data import SingleRunData -try: - from sentence_transformers.cross_encoder import CrossEncoder +import numpy as np - _SEMANTIC_AVAILABLE: bool = True -except ImportError: - _SEMANTIC_AVAILABLE = False +_MODEL_DIR = Path(__file__).resolve().parents[1] / "models" / "all-MiniLM-L6-v2" +_HF_MODEL_ID = "sentence-transformers/all-MiniLM-L6-v2" -_MODEL_DIR = Path(__file__).resolve().parents[1] / "models" / "stsb-cross-encoder" -_HF_MODEL_ID = "cross-encoder/stsb-distilroberta-base" - -# CrossEncoder instance when loaded, else None +# SentenceTransformer instance when loaded, else None _model: Any = None ABBREVIATIONS: dict[str, str] = { @@ -83,44 +78,42 @@ def __str__(self) -> str: def install_encoder() -> bool: - """Download and save the cross-encoder model for semantic search. + """Download and save the SentenceTransformer model for semantic search. Returns ------- bool True if the model loaded successfully, False otherwise. - - Notes - ----- - Returns False immediately if ``sentence-transformers`` is not installed - (i.e. ``perda[semantic]`` extra was not requested). - Any download or filesystem error is caught and printed; the function - returns False so callers fall back to keyword-only search. """ global _model - if not _SEMANTIC_AVAILABLE: + try: + from sentence_transformers import SentenceTransformer + except ImportError: + print("Warning: sentence-transformers is not installed. Falling back to keyword-only search.") return False try: if not _MODEL_DIR.exists(): - print("Downloading cross-encoder model (one-time setup)...") - _model = CrossEncoder(_HF_MODEL_ID) + print("Downloading SentenceTransformer model (one-time setup)...") + _model = SentenceTransformer(_HF_MODEL_ID) _MODEL_DIR.parent.mkdir(parents=True, exist_ok=True) _model.save(str(_MODEL_DIR)) print(f"Model saved to: {_MODEL_DIR}") else: - _model = CrossEncoder(str(_MODEL_DIR)) + _model = SentenceTransformer(str(_MODEL_DIR)) return True except Exception as e: print( - f"Warning: cross-encoder model unavailable ({e}). Falling back to keyword-only search." + f"Warning: SentenceTransformer model unavailable ({e}). Falling back to keyword-only search." ) _model = None return False -def search(data: SingleRunData, query: str, top_n: int = 10) -> list[SearchResult]: +def search( + data: SingleRunData, query: str, top_n: int = 10, semantic: bool = False +) -> list[SearchResult]: """Search telemetry variables, print the top matches, and return them. Parameters @@ -131,6 +124,9 @@ def search(data: SingleRunData, query: str, top_n: int = 10) -> list[SearchResul Free-text search query (e.g. "bat wheel"). top_n : int Maximum number of results to return and display (default 10). + semantic : bool + If True, runs a hybrid search with sentence-transformer embeddings. + If False, uses fast keyword-only rapidfuzz search (default). Returns ------- @@ -138,27 +134,10 @@ def search(data: SingleRunData, query: str, top_n: int = 10) -> list[SearchResul Top matches in descending relevance order (at most ``top_n`` entries). Each entry exposes ``rank``, ``score``, ``var_id``, ``cpp_name``, and ``descript`` for programmatic access. - - Notes - ----- - When ``perda[semantic]`` is installed and the cross-encoder model loads - successfully, results are ranked by a weighted blend of semantic score and - rapidfuzz keyword score. Otherwise falls back to keyword-only scoring with - no error raised. - - Short queries lean on keyword matching; longer queries lean on semantic - ranking when the model is available. - - Examples - -------- - >>> results = aly.search("front wheel speed") - >>> names = [r.cpp_name for r in results] """ if top_n <= 0: raise ValueError("top_n must be a positive integer.") - semantic_ready = install_encoder() - query = query.strip() if not query: raise ValueError("Search query cannot be empty.") @@ -167,31 +146,63 @@ def search(data: SingleRunData, query: str, top_n: int = 10) -> list[SearchResul if not keyword_query: raise ValueError("Search query must contain letters or numbers.") - deck = build_search_deck(data) + # Lazy-build search deck + if getattr(data, "_search_deck", None) is None: + data._search_deck = build_search_deck(data) + deck = data._search_deck + + if not deck: + return [] + num_terms = len(keyword_query) + semantic_ready = install_encoder() if semantic else False if semantic_ready and _model is not None: - semantic_query = preprocess_query(query) - # rank() returns dicts with "corpus_id" (index into deck) and "score" - semantic_scores = { - int(r["corpus_id"]): float(r["score"]) - for r in _model.rank(semantic_query, [e.card for e in deck]) - } - ranked = sorted( - ( + try: + # Lazy-build card embeddings + if getattr(data, "_search_embeddings", None) is None: + cards = [e.card for e in deck] + card_embeddings = _model.encode(cards, convert_to_numpy=True) + norms = np.linalg.norm(card_embeddings, axis=1, keepdims=True) + norms[norms == 0.0] = 1.0 + data._search_embeddings = card_embeddings / norms + + embeddings = data._search_embeddings + + # Encode query and project + semantic_query = preprocess_query(query) + query_emb = _model.encode(semantic_query, convert_to_numpy=True) + q_norm = np.linalg.norm(query_emb) + if q_norm > 0: + query_emb = query_emb / q_norm + else: + query_emb = np.zeros_like(query_emb) + + # Dot-product similarity scores + semantic_scores_arr = np.dot(embeddings, query_emb) + semantic_scores = { + idx: float(score) for idx, score in enumerate(semantic_scores_arr) + } + + ranked = sorted( ( - combine_scores( - semantic_scores.get(idx, 0.0), - keyword_score(keyword_query, entry), - num_terms, - ), - idx, - ) - for idx, entry in enumerate(deck) - ), - reverse=True, - ) - else: + ( + combine_scores( + semantic_scores.get(idx, 0.0), + keyword_score(keyword_query, entry), + num_terms, + ), + idx, + ) + for idx, entry in enumerate(deck) + ), + reverse=True, + ) + except Exception as e: + print(f"Warning: Semantic search failed ({e}). Falling back to keyword search.") + semantic_ready = False + + if not semantic_ready or _model is None: ranked = sorted( ( (keyword_score(keyword_query, entry), idx) From 9bacda3c8d2304bfcffaad0b1ec5b11ac5c70bdc Mon Sep 17 00:00:00 2001 From: Jeffrey-Moon Date: Fri, 17 Jul 2026 17:59:58 -0400 Subject: [PATCH 3/6] Multilog Analysis and CSV Loading Optimization --- notebooks/Tutorial_[advanced].ipynb | 67 ++++++ notebooks/Tutorial_[simple].ipynb | 44 +++- perda/analyzer/__init__.py | 2 + perda/analyzer/analyzer.py | 16 +- perda/analyzer/comparison.py | 126 +++++++++++ perda/analyzer/csv.py | 68 +++++- perda/analyzer/run_collection.py | 210 ++++++++++++++++++ perda/core_data_structures/single_run_data.py | 21 +- 8 files changed, 535 insertions(+), 19 deletions(-) create mode 100644 perda/analyzer/comparison.py create mode 100644 perda/analyzer/run_collection.py diff --git a/notebooks/Tutorial_[advanced].ipynb b/notebooks/Tutorial_[advanced].ipynb index 10ecaf3..f8c3989 100644 --- a/notebooks/Tutorial_[advanced].ipynb +++ b/notebooks/Tutorial_[advanced].ipynb @@ -157,6 +157,73 @@ "fig.show()" ] }, + { + "cell_type": "markdown", + "id": "run-collection-header", + "metadata": {}, + "source": [ + "#### Multi-Log Comparative Analysis (RunCollection)\n", + "\n", + "PERDA provides a high-level `RunCollection` utility to manage, filter, and compare multiple log files from a directory chronologically." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "run-collection-code-1", + "metadata": {}, + "outputs": [], + "source": [ + "from perda.analyzer import RunCollection\n", + "from datetime import datetime\n", + "\n", + "# Index a directory recursively to scan logs in all subdirectories\n", + "col_recursive = RunCollection.from_directory(\"csv_files\", recursive=True)\n", + "print(f\"Indexed {len(col_recursive)} runs recursively.\")\n", + "\n", + "# Alternatively, scan a mixed list of specific files and folders using from_paths\n", + "col = RunCollection.from_paths([\n", + " \"csv_files/LogBrowseTest/16thApr10-52-00.csv\",\n", + " \"csv_files/LogBrowseTest\"\n", + "])\n", + "print(f\"Indexed {len(col)} runs via specific paths.\")\n", + "\n", + "# Filter logs chronologically by a date range\n", + "filtered_col = col.filter_by_date(start_date=datetime(2026, 4, 1), end_date=datetime(2026, 6, 1))\n", + "print(f\"Filtered to {len(filtered_col)} runs.\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "run-collection-code-2", + "metadata": {}, + "outputs": [], + "source": [ + "# Find which logs in the collection hit a watchdog error condition (evaluates lazily to avoid memory leaks)\n", + "error_logs = col.check_any(\n", + " \"pcm.watchdog.sensorTimeout\",\n", + " lambda values: (values == 1.0).any()\n", + ")\n", + "print(f\"Runs with watchdog timeout error: {error_logs}\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "run-collection-code-3", + "metadata": {}, + "outputs": [], + "source": [ + "# Generate statistical summaries of a variable across all runs\n", + "stats = col.compare_summary(\"pcm.wheelSpeeds.frontLeft\")\n", + "print(\"Stats Summary:\", stats)\n", + "\n", + "# Overlay plot of a variable across all runs, downsampled to 20,000 points per trace for fast rendering\n", + "fig = col.plot_comparison(\"pcm.wheelSpeeds.frontLeft\", max_points=20000)\n", + "fig.show()" + ] + }, { "cell_type": "markdown", "id": "9", diff --git a/notebooks/Tutorial_[simple].ipynb b/notebooks/Tutorial_[simple].ipynb index 23dddbd..f142591 100644 --- a/notebooks/Tutorial_[simple].ipynb +++ b/notebooks/Tutorial_[simple].ipynb @@ -96,6 +96,29 @@ "search_results = aly.search(\"vectornav\")" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### Semantic Search\n", + "\n", + "PERDA supports hybrid semantic search. By passing `semantic=True`, the library uses a local bi-encoder model to understand the meaning of your query, finding relevant variables even if they don't share exact keywords (e.g. searching 'battery current limit' can match specific BMS stack parameters)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Run a semantic search for a query\n", + "semantic_results = aly.search(\"battery current limit\", semantic=True)\n", + "\n", + "# Print top 5 matches with descriptions\n", + "for r in semantic_results[:5]:\n", + " print(f\"{r.score:.2f} | {r.cpp_name} - {r.descript}\")" + ] + }, { "cell_type": "markdown", "metadata": {}, @@ -167,6 +190,25 @@ "aly.plot(variables)" ] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "#### Parameterized Downsampling\n", + "\n", + "To speed up rendering or save network bandwidth, you can optionally restrict the number of points plotted using the `max_points` parameter. This applies a memory-efficient zero-copy step downsampling." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Graph with a hard number limit of 10,000 points per trace\n", + "aly.plot(variables, max_points=10000)" + ] + }, { "cell_type": "markdown", "metadata": {}, @@ -307,4 +349,4 @@ }, "nbformat": 4, "nbformat_minor": 4 -} +} \ No newline at end of file diff --git a/perda/analyzer/__init__.py b/perda/analyzer/__init__.py index 044ee88..438fa82 100644 --- a/perda/analyzer/__init__.py +++ b/perda/analyzer/__init__.py @@ -2,3 +2,5 @@ from .analyzer_factory import * from .concat_helpers import concat from .csv import * +from .comparison import plot_comparison, compare_summary +from .run_collection import RunCollection diff --git a/perda/analyzer/analyzer.py b/perda/analyzer/analyzer.py index c498b76..a5e44ec 100644 --- a/perda/analyzer/analyzer.py +++ b/perda/analyzer/analyzer.py @@ -101,6 +101,7 @@ def __init__( parsing_errors_limit=parsing_errors_limit, verbose=verbose, ) + self.file_path = filepath if preprocessing: self.data = apply_preprocessing(self.data, preprocessing) @@ -120,7 +121,9 @@ def __str__(self) -> str: return output - def search(self, query: str, top_n: int = 10) -> list[SearchResult]: + def search( + self, query: str, top_n: int = 10, semantic: bool = False + ) -> list[SearchResult]: """ Natural language search for available variables in the parsed data. @@ -132,6 +135,8 @@ def search(self, query: str, top_n: int = 10) -> list[SearchResult]: Free-text search query (e.g. "front wheel speed"). top_n : int Maximum number of results to return and display (default 10). + semantic : bool + Whether to use semantic search embeddings (default False). Returns ------- @@ -143,10 +148,10 @@ def search(self, query: str, top_n: int = 10) -> list[SearchResult]: Examples -------- >>> results = aly.search("front wheel speed") - >>> results = aly.search("front wheel speed", top_n=5) + >>> results = aly.search("front wheel speed", top_n=5, semantic=True) >>> names = [r.cpp_name for r in results] """ - return search(self.data, query, top_n) + return search(self.data, query, top_n, semantic) def plot( self, @@ -163,6 +168,7 @@ def plot( font_config: FontConfig = DEFAULT_FONT_CONFIG, layout_config: LayoutConfig = DEFAULT_LAYOUT_CONFIG, vline_config: VLineConfig = DEFAULT_VLINE_CONFIG, + max_points: int | None = None, ) -> go.Figure: """ Display variables from the parsed data on an interactive Plotly plot. @@ -192,6 +198,8 @@ def plot( Layout configuration for plot dimensions. vline_config : VLineConfig, optional Visual configuration for concat boundary lines. + max_points : int | None, optional + Maximum number of data points to plot per trace (enables zero-copy step downsampling). Default is None (full fidelity). Examples -------- @@ -237,6 +245,7 @@ def plot( timestamp_unit=self.data.timestamp_unit, vlines=vlines, vline_config=vline_config, + max_points=max_points, ) else: return plot_single_axis( @@ -249,6 +258,7 @@ def plot( timestamp_unit=self.data.timestamp_unit, vlines=vlines, vline_config=vline_config, + max_points=max_points, ) def subplots( diff --git a/perda/analyzer/comparison.py b/perda/analyzer/comparison.py new file mode 100644 index 0000000..a9d1ab7 --- /dev/null +++ b/perda/analyzer/comparison.py @@ -0,0 +1,126 @@ +import os +from typing import Any, Dict, List +import numpy as np +import plotly.graph_objects as go + +from ..plotting.plotting_constants import ( + DEFAULT_FONT_CONFIG, + DEFAULT_LAYOUT_CONFIG, +) +from ..units import _to_seconds + + +def plot_comparison( + analyzers: List[Any], cpp_name: str, max_points: int | None = None +) -> go.Figure: + """Compare and overlay a variable from multiple runs on a single Plotly figure. + + Aligns each run's timestamps to start at relative seconds (t = 0). + + Parameters + ---------- + analyzers : List[Analyzer] + List of Analyzer instances to compare. + cpp_name : str + C++ variable name to plot and overlay. + max_points : int | None, optional + Maximum number of data points to plot per run (enables zero-copy step downsampling). Default is None (full fidelity). + + Returns + ------- + go.Figure + Plotly Figure with overlaid traces from all matching runs. + """ + fig = go.Figure() + + for aly in analyzers: + if cpp_name not in aly.data: + continue + di = aly.data[cpp_name] + if len(di) == 0: + continue + + # Shift timestamps relative to start time and convert to seconds + raw_start = aly.data.data_start_time + rel_timestamps_raw = di.timestamp_np - raw_start + + # Downsample coordinates if data points exceed the limit + n_points = len(di) + if max_points is not None and n_points > max_points and max_points > 0: + step = n_points // max_points + timestamps_s = _to_seconds(rel_timestamps_raw[::step], aly.data.timestamp_unit) + values_np = di.value_np[::step] + else: + timestamps_s = _to_seconds(rel_timestamps_raw, aly.data.timestamp_unit) + values_np = di.value_np + + # Get file name for legend label + file_path = getattr(aly, "file_path", None) or getattr(aly, "filename", None) or "Run" + run_label = os.path.basename(str(file_path)) + + fig.add_trace( + go.Scattergl( + x=timestamps_s, + y=values_np, + mode="lines", + name=f"{run_label} ({di.label})", + ) + ) + + fig.update_layout( + title=dict( + text=f"Multi-Log Comparison: {cpp_name}", + font=dict(size=DEFAULT_FONT_CONFIG.large) + ), + xaxis_title=dict( + text="Time (seconds from start)", + font=dict(size=DEFAULT_FONT_CONFIG.medium) + ), + yaxis_title=dict( + text=cpp_name, + font=dict(size=DEFAULT_FONT_CONFIG.medium) + ), + template="plotly_dark", + width=DEFAULT_LAYOUT_CONFIG.width, + height=DEFAULT_LAYOUT_CONFIG.height, + margin=DEFAULT_LAYOUT_CONFIG.margin, + ) + return fig + + +def compare_summary(analyzers: List[Any], cpp_name: str) -> Dict[str, Dict[str, float]]: + """Compare basic statistics of a variable across multiple logs. + + Parameters + ---------- + analyzers : List[Analyzer] + List of Analyzer instances to compare. + cpp_name : str + C++ variable name to summarize. + + Returns + ------- + Dict[str, Dict[str, float]] + A dictionary mapping run filenames to statistical descriptors: + {"run_filename.csv": {"min": x, "max": y, "mean": z, "std": w, "len": n}} + """ + comparison = {} + for aly in analyzers: + if cpp_name not in aly.data: + continue + di = aly.data[cpp_name] + if len(di) == 0: + continue + + vals = di.value_np + file_path = getattr(aly, "file_path", None) or getattr(aly, "filename", None) or "Run" + run_label = os.path.basename(str(file_path)) + + comparison[run_label] = { + "min": float(np.min(vals)), + "max": float(np.max(vals)), + "mean": float(np.mean(vals)), + "std": float(np.std(vals)), + "len": int(len(vals)), + } + return comparison diff --git a/perda/analyzer/csv.py b/perda/analyzer/csv.py index b2c7dc7..6294457 100644 --- a/perda/analyzer/csv.py +++ b/perda/analyzer/csv.py @@ -1,3 +1,5 @@ +from datetime import datetime +import re import numpy as np import polars as pl from tqdm import tqdm @@ -7,6 +9,30 @@ from ..units import Timescale +def _find_contiguous_slices(var_ids: np.ndarray) -> dict[int, tuple[int, int]]: + """Find contiguous slice boundary index pairs (start, end) for a sorted 1D array. + + Avoids np.unique sorting overhead by performing a single-pass O(N) diff. + """ + if len(var_ids) == 0: + return {} + + diff_mask = var_ids[:-1] != var_ids[1:] + change_indices = np.flatnonzero(diff_mask) + 1 + + start_indices = np.empty(len(change_indices) + 1, dtype=np.int64) + start_indices[0] = 0 + start_indices[1:] = change_indices + + end_indices = np.append(start_indices[1:], len(var_ids)) + unique_ids = var_ids[start_indices] + + return { + int(uid): (start, end) + for uid, start, end in zip(unique_ids, start_indices, end_indices) + } + + def parse_csv( file_path: str, ts_offset: int = 0, @@ -42,9 +68,22 @@ def parse_csv( parse_unit = ( Timescale.US if header_line.rstrip().endswith("v2.0") else Timescale.MS ) + + creation_time = None + try: + match = re.search(r"PER Log:\s*(.*?)(?:\s+v\d+\.\d+)?$", header_line) + if match: + date_str = match.group(1).strip() + # Parse format: Thu Jun 11 17:06:37 2026 + creation_time = datetime.strptime(date_str, "%a %b %d %H:%M:%S %Y") + except Exception: + pass + if verbose >= 1: print(f"Header: {header_line.rstrip()}") print(f"Timestamp unit: {parse_unit.value}") + if creation_time: + print(f"Log recorded on: {creation_time}") # Block 1: Variable ID/Name pairs if verbose >= 2: @@ -130,15 +169,15 @@ def parse_csv( data_start_time = int(df["timestamp"].min()) data_end_time = int(df["timestamp"].max()) - # Build per-variable numpy arrays from grouped Polars data - var_arrays: dict[int, tuple] = {} - for (var_id,), group in df.group_by(["var_id"], maintain_order=True): - var_arrays[int(var_id)] = ( - group["timestamp"].to_numpy(), - group["value"].to_numpy(), - ) + # Extract sorted columns to main numpy arrays + var_ids = df["var_id"].to_numpy() + timestamps_all = df["timestamp"].to_numpy() + values_all = df["value"].to_numpy() - # Format data as DataInstances + # Fast O(N) boundary scan using helper function + slices = _find_contiguous_slices(var_ids) + + # Format data as DataInstances (zero-copy slicing views) id_to_instance: dict[int, DataInstance] = {} cpp_name_to_id: dict[str, int] = {} if verbose >= 2: @@ -147,9 +186,15 @@ def parse_csv( name = id_to_cpp_name[var_id] descript = id_to_descript[var_id] cpp_name_to_id[name] = var_id - timestamps_np, values_np = var_arrays.get( - var_id, (np.array([], dtype=np.int64), np.array([], dtype=np.float64)) - ) + + if var_id in slices: + start, end = slices[var_id] + timestamps_np = timestamps_all[start:end] + values_np = values_all[start:end] + else: + timestamps_np = np.array([], dtype=np.int64) + values_np = np.array([], dtype=np.float64) + id_to_instance[var_id] = DataInstance( timestamp_np=timestamps_np, value_np=values_np, @@ -170,6 +215,7 @@ def parse_csv( cpp_name_to_id=cpp_name_to_id, id_to_cpp_name=id_to_cpp_name, id_to_descript=id_to_descript, + creation_time=creation_time, total_data_points=total_data_points, data_start_time=data_start_time, data_end_time=data_end_time, diff --git a/perda/analyzer/run_collection.py b/perda/analyzer/run_collection.py new file mode 100644 index 0000000..0d4e1d4 --- /dev/null +++ b/perda/analyzer/run_collection.py @@ -0,0 +1,210 @@ +import os +import glob +import re +from datetime import datetime +from typing import List, Callable, Union, Dict, Any, Optional +import numpy as np +import plotly.graph_objects as go + +from .analyzer import Analyzer +from .comparison import plot_comparison, compare_summary + + +class RunCollection: + """A collection of telemetry run logs (represented by Analyzer instances). + + Scans directories quickly by parsing only the first line header for date metadata, + and lazy-loads full datasets only when accessed. + """ + + def __init__(self, items: List[Dict[str, Any]]): + # items is a list of dicts: {"file_path": str, "filename": str, "date": Optional[datetime], "analyzer": Optional[Analyzer]} + self._items = items + + @classmethod + def from_paths( + cls, paths: Union[str, List[str]], recursive: bool = False, pattern: str = "*.csv" + ) -> "RunCollection": + """Build a RunCollection from a single path or a list of paths (directories or files). + + If a path is a directory, it scans files matching the pattern inside it (recursively if recursive=True). + If it is a file, it adds it directly. + """ + # Normalize input to list of paths + if isinstance(paths, str): + path_list = [paths] + else: + path_list = list(paths) + + resolved_files = [] + for p in path_list: + if os.path.isdir(p): + if recursive: + # Recursive scanning using glob double asterisk ** + search_pattern = os.path.join(p, "**", pattern) + files = glob.glob(search_pattern, recursive=True) + else: + search_pattern = os.path.join(p, pattern) + files = glob.glob(search_pattern) + resolved_files.extend(files) + elif os.path.isfile(p): + resolved_files.append(p) + else: + # Support direct glob pattern strings + files = glob.glob(p, recursive=recursive) + if files: + resolved_files.extend(files) + + # De-duplicate resolved files + resolved_files = sorted(list(set(resolved_files))) + + items = [] + for filepath in resolved_files: + filename = os.path.basename(filepath) + date = None + try: + with open(filepath, "r") as f: + header_line = f.readline() + match = re.search(r"PER Log:\s*(.*?)(?:\s+v\d+\.\d+)?$", header_line) + if match: + date_str = match.group(1).strip() + date = datetime.strptime(date_str, "%a %b %d %H:%M:%S %Y") + except Exception: + try: + date = datetime.fromtimestamp(os.path.getmtime(filepath)) + except Exception: + pass + + items.append({ + "file_path": filepath, + "filename": filename, + "date": date, + "analyzer": None + }) + + # Sort runs chronologically by date + items.sort(key=lambda x: (x["date"] is None, x["date"], x["filename"])) + return cls(items) + + @classmethod + def from_directory( + cls, dir_path: str, pattern: str = "*.csv", recursive: bool = False + ) -> "RunCollection": + """Scan a directory of CSV logs and construct a collection index using header dates. + + Supports recursive scanning of subdirectories if recursive=True. + """ + if not os.path.isdir(dir_path): + raise ValueError(f"'{dir_path}' is not a valid directory.") + return cls.from_paths(dir_path, recursive=recursive, pattern=pattern) + + @classmethod + def from_files(cls, filepaths: List[str]) -> "RunCollection": + """Build a collection from a specific list of file paths.""" + return cls.from_paths(filepaths) + + @property + def runs(self) -> List[Analyzer]: + """Load and return all Analyzer instances in this collection.""" + for item in self._items: + if item["analyzer"] is None: + item["analyzer"] = Analyzer(item["file_path"], verbose=0) + return [item["analyzer"] for item in self._items] + + def get_run(self, index: int) -> Analyzer: + """Load and return a specific Analyzer by index.""" + if index < 0 or index >= len(self._items): + raise IndexError("Collection index out of range.") + item = self._items[index] + if item["analyzer"] is None: + item["analyzer"] = Analyzer(item["file_path"], verbose=0) + return item["analyzer"] + + def __len__(self) -> int: + return len(self._items) + + def __getitem__(self, index: int) -> Analyzer: + return self.get_run(index) + + def filter(self, predicate: Callable[[Dict[str, Any]], bool]) -> "RunCollection": + """Filter the collection by metadata items and return a new RunCollection.""" + filtered_items = [item for item in self._items if predicate(item)] + return RunCollection(filtered_items) + + def filter_by_date( + self, start_date: Union[str, datetime] = None, end_date: Union[str, datetime] = None + ) -> "RunCollection": + """Filter runs within a specific date range (inclusive).""" + if isinstance(start_date, str): + start_date = datetime.fromisoformat(start_date) + if isinstance(end_date, str): + end_date = datetime.fromisoformat(end_date) + + def predicate(item: Dict[str, Any]) -> bool: + d = item["date"] + if d is None: + return False + if start_date and d < start_date: + return False + if end_date and d > end_date: + return False + return True + + return self.filter(predicate) + + def check_any(self, cpp_name: str, condition_fn: Callable[[np.ndarray], bool]) -> List[str]: + """Find log filenames in the collection where a condition holds for a given variable. + + Evaluates runs lazily to keep memory usage bounded. + + Parameters + ---------- + cpp_name : str + C++ name of the variable to check. + condition_fn : Callable[[np.ndarray], bool] + Boolean condition function operating on the variable's numpy array values. + + Returns + ------- + List[str] + List of filenames where the condition evaluated to True. + """ + matching_filenames = [] + for i in range(len(self._items)): + item = self._items[i] + # Use cached analyzer if available, otherwise load temporarily without caching + # to keep memory footprint low and avoid OOM crashes + if item["analyzer"] is not None: + aly = item["analyzer"] + else: + try: + aly = Analyzer(item["file_path"], verbose=0) + except Exception: + continue # Skip unreadable file + + if cpp_name in aly.data: + di = aly.data[cpp_name] + if condition_fn(di.value_np): + matching_filenames.append(item["filename"]) + return matching_filenames + + def plot_comparison(self, cpp_name: str, max_points: int | None = None) -> go.Figure: + """Overlay traces of the variable across all runs in the collection on a single plot. + + Parameters + ---------- + cpp_name : str + C++ name of the variable to compare. + max_points : int | None, optional + Maximum data points to plot per run (enables zero-copy step downsampling). Default is None (full fidelity). + + Returns + ------- + go.Figure + Plotly Figure with overlaid traces from all runs. + """ + return plot_comparison(self.runs, cpp_name, max_points=max_points) + + def compare_summary(self, cpp_name: str) -> Dict[str, Dict[str, float]]: + """Compare variable statistics across all runs in the collection.""" + return compare_summary(self.runs, cpp_name) diff --git a/perda/core_data_structures/single_run_data.py b/perda/core_data_structures/single_run_data.py index 90c8005..de4bec4 100644 --- a/perda/core_data_structures/single_run_data.py +++ b/perda/core_data_structures/single_run_data.py @@ -1,6 +1,7 @@ -from typing import Dict, List, Union +from datetime import datetime +from typing import Dict, List, Optional, Union -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, PrivateAttr from ..units import Timescale from .data_instance import DataInstance @@ -11,6 +12,10 @@ class SingleRunData(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) + # Private caches (excluded from Pydantic serialization) + _search_deck: Optional[list] = PrivateAttr(default=None) + _search_embeddings: Optional[object] = PrivateAttr(default=None) + # Core data storage id_to_instance: Dict[int, DataInstance] = Field( description="Mapping from variable ID to DataInstance" @@ -26,6 +31,10 @@ class SingleRunData(BaseModel): ) # Metadata + creation_time: Optional[datetime] = Field( + default=None, + description="Timestamp when this run log was recorded", + ) total_data_points: int = Field( description="Total number of data points across all variables" ) @@ -130,6 +139,8 @@ def add(self, cpp_name: str, di: DataInstance) -> None: self.cpp_name_to_id[cpp_name] = synthetic_id self.id_to_cpp_name[synthetic_id] = cpp_name self.id_to_descript[synthetic_id] = di.label or "" + self._search_deck = None + self._search_embeddings = None def replace(self, cpp_name: str, di: DataInstance) -> None: """ @@ -138,9 +149,9 @@ def replace(self, cpp_name: str, di: DataInstance) -> None: Parameters ---------- cpp_name : str - C++ variable name of the variable to replace. + C++ variable name of the variable to replace. di : DataInstance - DataInstance whose ``value_np`` that replaces the stored one. + DataInstance whose ``value_np`` that replaces the stored one. """ if cpp_name not in self: raise KeyError( @@ -164,6 +175,8 @@ def replace(self, cpp_name: str, di: DataInstance) -> None: var_id=old.var_id, cpp_name=old.cpp_name, ) + self._search_deck = None + self._search_embeddings = None def __contains__(self, input_var_id_name: Union[str, int]) -> bool: """ From 6b885116ad94fe13a14e7fd68e59cf55fb4ee3e9 Mon Sep 17 00:00:00 2001 From: Jeffrey-Moon Date: Fri, 17 Jul 2026 18:31:33 -0400 Subject: [PATCH 4/6] Downsampling Fix --- perda/analyzer/comparison.py | 2 +- perda/plotting/data_instance_plotter.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/perda/analyzer/comparison.py b/perda/analyzer/comparison.py index a9d1ab7..487f513 100644 --- a/perda/analyzer/comparison.py +++ b/perda/analyzer/comparison.py @@ -47,7 +47,7 @@ def plot_comparison( # Downsample coordinates if data points exceed the limit n_points = len(di) if max_points is not None and n_points > max_points and max_points > 0: - step = n_points // max_points + step = (n_points + max_points - 1) // max_points timestamps_s = _to_seconds(rel_timestamps_raw[::step], aly.data.timestamp_unit) values_np = di.value_np[::step] else: diff --git a/perda/plotting/data_instance_plotter.py b/perda/plotting/data_instance_plotter.py index 99f708d..23a8572 100644 --- a/perda/plotting/data_instance_plotter.py +++ b/perda/plotting/data_instance_plotter.py @@ -94,7 +94,7 @@ def plot_single_axis( # Convert timestamps from the log unit to seconds for plotting with optional downsampling. n_points = len(di) if max_points is not None and n_points > max_points and max_points > 0: - step = n_points // max_points + step = (n_points + max_points - 1) // max_points timestamps_s = _to_seconds(di.timestamp_np[::step], timestamp_unit) values_np = di.value_np[::step] else: @@ -206,7 +206,7 @@ def plot_dual_axis( # Convert timestamps from the log unit to seconds for plotting with optional downsampling. n_points = len(di) if max_points is not None and n_points > max_points and max_points > 0: - step = n_points // max_points + step = (n_points + max_points - 1) // max_points timestamps_s = _to_seconds(di.timestamp_np[::step], timestamp_unit) values_np = di.value_np[::step] else: @@ -232,7 +232,7 @@ def plot_dual_axis( # Convert timestamps from the log unit to seconds for plotting with optional downsampling. n_points = len(di) if max_points is not None and n_points > max_points and max_points > 0: - step = n_points // max_points + step = (n_points + max_points - 1) // max_points timestamps_s = _to_seconds(di.timestamp_np[::step], timestamp_unit) values_np = di.value_np[::step] else: From 8e96c665999335377839247004cbd7cb0b3652ff Mon Sep 17 00:00:00 2001 From: Jeffrey-Moon Date: Fri, 17 Jul 2026 18:47:54 -0400 Subject: [PATCH 5/6] Added Tests for New Features --- tests/analyzer/test_run_collection.py | 106 ++++++++++++++++++++++++++ tests/plotting/test_downsampling.py | 33 ++++++++ tests/utils/test_search.py | 57 ++++++++++++++ 3 files changed, 196 insertions(+) create mode 100644 tests/analyzer/test_run_collection.py create mode 100644 tests/plotting/test_downsampling.py create mode 100644 tests/utils/test_search.py diff --git a/tests/analyzer/test_run_collection.py b/tests/analyzer/test_run_collection.py new file mode 100644 index 0000000..32dcde8 --- /dev/null +++ b/tests/analyzer/test_run_collection.py @@ -0,0 +1,106 @@ +import textwrap +from datetime import datetime +import pytest + +from perda.analyzer.csv import parse_csv +from perda.analyzer.run_collection import RunCollection + +def test_creation_time_parsing(tmp_path): + # Header format: "PER Log: Thu Jun 11 17:06:37 2026 v2.0" + content = textwrap.dedent( + """\ + PER Log: Thu Jun 11 17:06:37 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,12.5 + 1000,1,12.6 + """ + ) + p = tmp_path / "test_date.csv" + p.write_text(content) + + srd = parse_csv(str(p), verbose=0) + assert srd.creation_time == datetime(2026, 6, 11, 17, 6, 37) + + +def test_no_date_header_graceful_fallback(tmp_path): + content = textwrap.dedent( + """\ + Standard Generic Log File Header With No Date + Value voltage (ams.pack.voltage): 1 + 0,1,12.5 + """ + ) + p = tmp_path / "no_date.csv" + p.write_text(content) + + srd = parse_csv(str(p), verbose=0) + assert srd.creation_time is None + + +def test_run_collection_sorting_and_filtering(tmp_path): + # Older log + content1 = textwrap.dedent( + """\ + PER Log: Thu Jun 11 10:00:00 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,12.5 + """ + ) + p1 = tmp_path / "log_old.csv" + p1.write_text(content1) + + # Newer log + content2 = textwrap.dedent( + """\ + PER Log: Thu Jun 11 12:00:00 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,13.0 + """ + ) + p2 = tmp_path / "log_new.csv" + p2.write_text(content2) + + # Load with order p2, p1 to ensure sorting works + col = RunCollection.from_paths([str(p2), str(p1)]) + + # Check chronological ordering: old (10:00) should be first + assert len(col) == 2 + assert col.runs[0].data.creation_time == datetime(2026, 6, 11, 10, 0, 0) + assert col.runs[1].data.creation_time == datetime(2026, 6, 11, 12, 0, 0) + + # Check date filtering + filtered = col.filter_by_date( + datetime(2026, 6, 11, 9, 0, 0), + datetime(2026, 6, 11, 11, 0, 0) + ) + assert len(filtered) == 1 + assert filtered.runs[0].data.creation_time == datetime(2026, 6, 11, 10, 0, 0) + + +def test_run_collection_check_any(tmp_path): + content1 = textwrap.dedent( + """\ + PER Log: Thu Jun 11 10:00:00 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,5.0 + 1000,1,6.0 + """ + ) + p1 = tmp_path / "log1.csv" + p1.write_text(content1) + + content2 = textwrap.dedent( + """\ + PER Log: Thu Jun 11 12:00:00 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,15.0 + """ + ) + p2 = tmp_path / "log2.csv" + p2.write_text(content2) + + col = RunCollection.from_paths([str(p1), str(p2)]) + + # Query if voltage ever exceeded 10.0 + matching = col.check_any("ams.pack.voltage", lambda val: (val > 10.0).any()) + assert matching == ["log2.csv"] diff --git a/tests/plotting/test_downsampling.py b/tests/plotting/test_downsampling.py new file mode 100644 index 0000000..7b8d77e --- /dev/null +++ b/tests/plotting/test_downsampling.py @@ -0,0 +1,33 @@ +import textwrap +import pytest + +from perda.analyzer.csv import parse_csv +from perda.analyzer.analyzer import Analyzer +from perda.analyzer.comparison import plot_comparison + +def test_downsampling_strict_ceiling(tmp_path): + # Write a log with 105 data points + lines = ["PER Log: Thu Jun 11 10:00:00 2026 v2.0", "Value voltage (ams.pack.voltage): 1"] + for i in range(105): + lines.append(f"{i * 100},1,{i * 0.1}") + + p = tmp_path / "log_large.csv" + p.write_text("\n".join(lines)) + + # Initialize Analyzer directly with file path + aly = Analyzer(str(p), verbose=0) + di = aly.data["ams.pack.voltage"] + assert len(di) == 105 + + # Request max_points = 20 + # Expected ceiling division step: (105 + 20 - 1) // 20 = 124 // 20 = 6 + # Slicing length: ceil(105 / 6) = 18 points (which is <= 20) + fig = plot_comparison([aly], "ams.pack.voltage", max_points=20) + assert len(fig.data) == 1 + trace_points = len(fig.data[0].x) + assert trace_points <= 20 + assert trace_points == 18 + + # Verify that requesting max_points larger than length preserves all points + fig_full = plot_comparison([aly], "ams.pack.voltage", max_points=200) + assert len(fig_full.data[0].x) == 105 diff --git a/tests/utils/test_search.py b/tests/utils/test_search.py new file mode 100644 index 0000000..34f69bb --- /dev/null +++ b/tests/utils/test_search.py @@ -0,0 +1,57 @@ +import textwrap +import numpy as np +import pytest +from unittest.mock import patch + +from perda.analyzer.csv import parse_csv +from perda.utils.search import search +from perda.core_data_structures.data_instance import DataInstance + +def test_search_semantic_fallback(tmp_path): + content = textwrap.dedent( + """\ + PER Log: Thu Jun 11 10:00:00 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,12.5 + """ + ) + p = tmp_path / "test_search.csv" + p.write_text(content) + srd = parse_csv(str(p), verbose=0) + + # Mock SentenceTransformer constructor to fail + with patch("sentence_transformers.SentenceTransformer", side_effect=RuntimeError("Mocked transformer loading failure")): + # Semantic search should fail gracefully, issue warning, and fall back to keyword search + results = search(srd, "voltage", semantic=True) + assert len(results) > 0 + assert results[0].cpp_name == "ams.pack.voltage" + + +def test_search_cache_invalidation(tmp_path): + content = textwrap.dedent( + """\ + PER Log: Thu Jun 11 10:00:00 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,12.5 + """ + ) + p = tmp_path / "test_cache.csv" + p.write_text(content) + srd = parse_csv(str(p), verbose=0) + + # Initial search populates the search deck cache + search(srd, "voltage") + assert srd._search_deck is not None + + # Mutate SingleRunData by adding a new variable + new_di = DataInstance(timestamp_np=np.array([0]), value_np=np.array([1.0]), label="New custom variable") + srd.add("test.new_var", new_di) + + # Assert caches are cleared + assert srd._search_deck is None + assert srd._search_embeddings is None + + # Verify new variable is searchable + results = search(srd, "custom") + assert len(results) > 0 + assert results[0].cpp_name == "test.new_var" From ecedc94b464e69b3aabefa451188bcc3227965cd Mon Sep 17 00:00:00 2001 From: Jeffrey-Moon Date: Fri, 17 Jul 2026 19:29:53 -0400 Subject: [PATCH 6/6] Formatting Fix --- notebooks/Tutorial_[advanced].ipynb | 4 +- perda/analyzer/__init__.py | 5 +- perda/analyzer/csv.py | 33 ++----- perda/analyzer/csv_utils.py | 36 ++++++++ perda/analyzer/run_collection.py | 120 ++++++++++++++------------ perda/utils/__init__.py | 1 + perda/utils/csv_utils.py | 52 +++++++++++ tests/analyzer/test_run_collection.py | 20 ++++- 8 files changed, 182 insertions(+), 89 deletions(-) create mode 100644 perda/analyzer/csv_utils.py create mode 100644 perda/utils/csv_utils.py diff --git a/notebooks/Tutorial_[advanced].ipynb b/notebooks/Tutorial_[advanced].ipynb index f8c3989..dea4e6d 100644 --- a/notebooks/Tutorial_[advanced].ipynb +++ b/notebooks/Tutorial_[advanced].ipynb @@ -178,11 +178,11 @@ "from datetime import datetime\n", "\n", "# Index a directory recursively to scan logs in all subdirectories\n", - "col_recursive = RunCollection.from_directory(\"csv_files\", recursive=True)\n", + "col_recursive = RunCollection(\"csv_files\", recursive=True)\n", "print(f\"Indexed {len(col_recursive)} runs recursively.\")\n", "\n", "# Alternatively, scan a mixed list of specific files and folders using from_paths\n", - "col = RunCollection.from_paths([\n", + "col = RunCollection([\n", " \"csv_files/LogBrowseTest/16thApr10-52-00.csv\",\n", " \"csv_files/LogBrowseTest\"\n", "])\n", diff --git a/perda/analyzer/__init__.py b/perda/analyzer/__init__.py index 438fa82..eb45a87 100644 --- a/perda/analyzer/__init__.py +++ b/perda/analyzer/__init__.py @@ -1,6 +1,7 @@ +# Submodule re-exports from .analyzer import * from .analyzer_factory import * from .concat_helpers import concat -from .csv import * from .comparison import plot_comparison, compare_summary -from .run_collection import RunCollection +from .csv import * +from .run_collection import RunCollection, RunMetadata diff --git a/perda/analyzer/csv.py b/perda/analyzer/csv.py index 6294457..06bf397 100644 --- a/perda/analyzer/csv.py +++ b/perda/analyzer/csv.py @@ -8,29 +8,7 @@ from ..core_data_structures.single_run_data import SingleRunData from ..units import Timescale - -def _find_contiguous_slices(var_ids: np.ndarray) -> dict[int, tuple[int, int]]: - """Find contiguous slice boundary index pairs (start, end) for a sorted 1D array. - - Avoids np.unique sorting overhead by performing a single-pass O(N) diff. - """ - if len(var_ids) == 0: - return {} - - diff_mask = var_ids[:-1] != var_ids[1:] - change_indices = np.flatnonzero(diff_mask) + 1 - - start_indices = np.empty(len(change_indices) + 1, dtype=np.int64) - start_indices[0] = 0 - start_indices[1:] = change_indices - - end_indices = np.append(start_indices[1:], len(var_ids)) - unique_ids = var_ids[start_indices] - - return { - int(uid): (start, end) - for uid, start, end in zip(unique_ids, start_indices, end_indices) - } +from ..utils.csv_utils import _find_contiguous_slices def parse_csv( @@ -176,6 +154,7 @@ def parse_csv( # Fast O(N) boundary scan using helper function slices = _find_contiguous_slices(var_ids) + slice_map = slices.slices # Format data as DataInstances (zero-copy slicing views) id_to_instance: dict[int, DataInstance] = {} @@ -187,10 +166,10 @@ def parse_csv( descript = id_to_descript[var_id] cpp_name_to_id[name] = var_id - if var_id in slices: - start, end = slices[var_id] - timestamps_np = timestamps_all[start:end] - values_np = values_all[start:end] + if var_id in slice_map: + slice_bounds = slice_map[var_id] + timestamps_np = timestamps_all[slice_bounds.start:slice_bounds.end] + values_np = values_all[slice_bounds.start:slice_bounds.end] else: timestamps_np = np.array([], dtype=np.int64) values_np = np.array([], dtype=np.float64) diff --git a/perda/analyzer/csv_utils.py b/perda/analyzer/csv_utils.py new file mode 100644 index 0000000..f9d7d8c --- /dev/null +++ b/perda/analyzer/csv_utils.py @@ -0,0 +1,36 @@ +import numpy as np +from numpy.typing import NDArray + + +def _find_contiguous_slices(var_ids: NDArray[np.int64]) -> dict[int, tuple[int, int]]: + """Find contiguous slice boundary index pairs for a sorted 1D array. + + Avoids ``np.unique`` sorting overhead by performing a single-pass O(N) diff. + + Parameters + ---------- + var_ids : NDArray[np.int64] + Sorted 1D array of variable identifiers. + + Returns + ------- + dict[int, tuple[int, int]] + Mapping from each unique variable ID to ``(start, end)`` slice bounds. + """ + if len(var_ids) == 0: + return {} + + diff_mask = var_ids[:-1] != var_ids[1:] + change_indices = np.flatnonzero(diff_mask) + 1 + + start_indices = np.empty(len(change_indices) + 1, dtype=np.int64) + start_indices[0] = 0 + start_indices[1:] = change_indices + + end_indices = np.append(start_indices[1:], len(var_ids)) + unique_ids = var_ids[start_indices] + + return { + int(uid): (start, end) + for uid, start, end in zip(unique_ids, start_indices, end_indices) + } diff --git a/perda/analyzer/run_collection.py b/perda/analyzer/run_collection.py index 0d4e1d4..f1da190 100644 --- a/perda/analyzer/run_collection.py +++ b/perda/analyzer/run_collection.py @@ -5,11 +5,24 @@ from typing import List, Callable, Union, Dict, Any, Optional import numpy as np import plotly.graph_objects as go +from pydantic import BaseModel, Field from .analyzer import Analyzer from .comparison import plot_comparison, compare_summary +class RunMetadata(BaseModel): + """Internal model for telemetry run index metadata.""" + file_path: str = Field(description="Absolute path to the log file") + filename: str = Field(description="Display filename of the log") + date: Optional[datetime] = Field(default=None, description="Recording date parsed from header or file stats") + analyzer: Optional[Analyzer] = Field(default=None, description="Lazily-loaded Analyzer instance") + + model_config = { + "arbitrary_types_allowed": True + } + + class RunCollection: """A collection of telemetry run logs (represented by Analyzer instances). @@ -17,19 +30,34 @@ class RunCollection: and lazy-loads full datasets only when accessed. """ - def __init__(self, items: List[Dict[str, Any]]): - # items is a list of dicts: {"file_path": str, "filename": str, "date": Optional[datetime], "analyzer": Optional[Analyzer]} - self._items = items + def __init__( + self, + paths: Optional[Union[str, List[str]]] = None, + recursive: bool = False, + pattern: str = "*.csv", + items: Optional[List[RunMetadata]] = None, + ): + """Construct a RunCollection directly from paths (or a pre-built list of RunMetadata). - @classmethod - def from_paths( - cls, paths: Union[str, List[str]], recursive: bool = False, pattern: str = "*.csv" - ) -> "RunCollection": - """Build a RunCollection from a single path or a list of paths (directories or files). - - If a path is a directory, it scans files matching the pattern inside it (recursively if recursive=True). - If it is a file, it adds it directly. + Parameters + ---------- + paths : Union[str, List[str]], optional + Directory paths, list of files, or wildcards to scan. + recursive : bool, optional + Recursively search directories if True. Default is False. + pattern : str, optional + Filename glob pattern to match. Default is "*.csv". + items : List[RunMetadata], optional + Internal list of pre-built items. Used internally for filtering. """ + if items is not None: + self._items = items + return + + if paths is None: + self._items = [] + return + # Normalize input to list of paths if isinstance(paths, str): path_list = [paths] @@ -40,7 +68,6 @@ def from_paths( for p in path_list: if os.path.isdir(p): if recursive: - # Recursive scanning using glob double asterisk ** search_pattern = os.path.join(p, "**", pattern) files = glob.glob(search_pattern, recursive=True) else: @@ -50,15 +77,14 @@ def from_paths( elif os.path.isfile(p): resolved_files.append(p) else: - # Support direct glob pattern strings files = glob.glob(p, recursive=recursive) if files: resolved_files.extend(files) - # De-duplicate resolved files + # De-duplicate and sort resolved files resolved_files = sorted(list(set(resolved_files))) - items = [] + self._items = [] for filepath in resolved_files: filename = os.path.basename(filepath) date = None @@ -75,50 +101,34 @@ def from_paths( except Exception: pass - items.append({ - "file_path": filepath, - "filename": filename, - "date": date, - "analyzer": None - }) + self._items.append( + RunMetadata( + file_path=filepath, + filename=filename, + date=date, + analyzer=None + ) + ) # Sort runs chronologically by date - items.sort(key=lambda x: (x["date"] is None, x["date"], x["filename"])) - return cls(items) - - @classmethod - def from_directory( - cls, dir_path: str, pattern: str = "*.csv", recursive: bool = False - ) -> "RunCollection": - """Scan a directory of CSV logs and construct a collection index using header dates. - - Supports recursive scanning of subdirectories if recursive=True. - """ - if not os.path.isdir(dir_path): - raise ValueError(f"'{dir_path}' is not a valid directory.") - return cls.from_paths(dir_path, recursive=recursive, pattern=pattern) - - @classmethod - def from_files(cls, filepaths: List[str]) -> "RunCollection": - """Build a collection from a specific list of file paths.""" - return cls.from_paths(filepaths) + self._items.sort(key=lambda x: (x.date is None, x.date, x.filename)) @property def runs(self) -> List[Analyzer]: """Load and return all Analyzer instances in this collection.""" for item in self._items: - if item["analyzer"] is None: - item["analyzer"] = Analyzer(item["file_path"], verbose=0) - return [item["analyzer"] for item in self._items] + if item.analyzer is None: + item.analyzer = Analyzer(item.file_path, verbose=0) + return [item.analyzer for item in self._items] def get_run(self, index: int) -> Analyzer: """Load and return a specific Analyzer by index.""" if index < 0 or index >= len(self._items): raise IndexError("Collection index out of range.") item = self._items[index] - if item["analyzer"] is None: - item["analyzer"] = Analyzer(item["file_path"], verbose=0) - return item["analyzer"] + if item.analyzer is None: + item.analyzer = Analyzer(item.file_path, verbose=0) + return item.analyzer def __len__(self) -> int: return len(self._items) @@ -126,10 +136,10 @@ def __len__(self) -> int: def __getitem__(self, index: int) -> Analyzer: return self.get_run(index) - def filter(self, predicate: Callable[[Dict[str, Any]], bool]) -> "RunCollection": + def filter(self, predicate: Callable[[RunMetadata], bool]) -> "RunCollection": """Filter the collection by metadata items and return a new RunCollection.""" filtered_items = [item for item in self._items if predicate(item)] - return RunCollection(filtered_items) + return RunCollection(items=filtered_items) def filter_by_date( self, start_date: Union[str, datetime] = None, end_date: Union[str, datetime] = None @@ -140,8 +150,8 @@ def filter_by_date( if isinstance(end_date, str): end_date = datetime.fromisoformat(end_date) - def predicate(item: Dict[str, Any]) -> bool: - d = item["date"] + def predicate(item: RunMetadata) -> bool: + d = item.date if d is None: return False if start_date and d < start_date: @@ -172,20 +182,18 @@ def check_any(self, cpp_name: str, condition_fn: Callable[[np.ndarray], bool]) - matching_filenames = [] for i in range(len(self._items)): item = self._items[i] - # Use cached analyzer if available, otherwise load temporarily without caching - # to keep memory footprint low and avoid OOM crashes - if item["analyzer"] is not None: - aly = item["analyzer"] + if item.analyzer is not None: + aly = item.analyzer else: try: - aly = Analyzer(item["file_path"], verbose=0) + aly = Analyzer(item.file_path, verbose=0) except Exception: continue # Skip unreadable file if cpp_name in aly.data: di = aly.data[cpp_name] if condition_fn(di.value_np): - matching_filenames.append(item["filename"]) + matching_filenames.append(item.filename) return matching_filenames def plot_comparison(self, cpp_name: str, max_points: int | None = None) -> go.Figure: diff --git a/perda/utils/__init__.py b/perda/utils/__init__.py index 5311be8..743e078 100644 --- a/perda/utils/__init__.py +++ b/perda/utils/__init__.py @@ -1,4 +1,5 @@ from ..units import * +from .csv_utils import * from .data_summary import * from .filtering import * from .frequency_analysis import * diff --git a/perda/utils/csv_utils.py b/perda/utils/csv_utils.py new file mode 100644 index 0000000..a333b71 --- /dev/null +++ b/perda/utils/csv_utils.py @@ -0,0 +1,52 @@ +from pydantic import BaseModel +import numpy as np +from numpy.typing import NDArray + + +class ContiguousSlice(BaseModel): + """Represents a contiguous segment of variable IDs in a sorted array.""" + + var_id: int + start: int + end: int + + +class ContiguousSliceResult(BaseModel): + """Mapping of variable IDs to contiguous slice boundaries.""" + + slices: dict[int, ContiguousSlice] + + +def _find_contiguous_slices(var_ids: NDArray[np.int64]) -> ContiguousSliceResult: + """Find contiguous slice boundary index pairs for a sorted 1D array. + + Avoids ``np.unique`` sorting overhead by performing a single-pass O(N) diff. + + Parameters + ---------- + var_ids : NDArray[np.int64] + Sorted 1D array of variable identifiers. + + Returns + ------- + ContiguousSliceResult + Pydantic model containing each unique variable ID mapped to its ``(start, end)`` bounds. + """ + if len(var_ids) == 0: + return ContiguousSliceResult(slices={}) + + diff_mask = var_ids[:-1] != var_ids[1:] + change_indices = np.flatnonzero(diff_mask) + 1 + + start_indices = np.empty(len(change_indices) + 1, dtype=np.int64) + start_indices[0] = 0 + start_indices[1:] = change_indices + + end_indices = np.append(start_indices[1:], len(var_ids)) + unique_ids = var_ids[start_indices] + + slices = { + int(uid): ContiguousSlice(var_id=int(uid), start=int(start), end=int(end)) + for uid, start, end in zip(unique_ids, start_indices, end_indices) + } + return ContiguousSliceResult(slices=slices) diff --git a/tests/analyzer/test_run_collection.py b/tests/analyzer/test_run_collection.py index 32dcde8..6c03327 100644 --- a/tests/analyzer/test_run_collection.py +++ b/tests/analyzer/test_run_collection.py @@ -61,7 +61,7 @@ def test_run_collection_sorting_and_filtering(tmp_path): p2.write_text(content2) # Load with order p2, p1 to ensure sorting works - col = RunCollection.from_paths([str(p2), str(p1)]) + col = RunCollection([str(p2), str(p1)]) # Check chronological ordering: old (10:00) should be first assert len(col) == 2 @@ -99,8 +99,24 @@ def test_run_collection_check_any(tmp_path): p2 = tmp_path / "log2.csv" p2.write_text(content2) - col = RunCollection.from_paths([str(p1), str(p2)]) + col = RunCollection([str(p1), str(p2)]) # Query if voltage ever exceeded 10.0 matching = col.check_any("ams.pack.voltage", lambda val: (val > 10.0).any()) assert matching == ["log2.csv"] + + +def test_run_collection_direct_init(tmp_path): + content = textwrap.dedent( + """\ + PER Log: Thu Jun 11 10:00:00 2026 v2.0 + Value voltage (ams.pack.voltage): 1 + 0,1,12.5 + """ + ) + p = tmp_path / "log.csv" + p.write_text(content) + + col = RunCollection(str(p)) + assert len(col) == 1 + assert col.runs[0].data.creation_time == datetime(2026, 6, 11, 10, 0, 0)