Skills › Scientific Agent Skills › neurokit2 › inspect_signal.py
inspect_signal.py
362 lines · skills/neurokit2/scripts/inspect_signal.py at 5f2bc03
1#!/usr/bin/env python3
2"""Inspect bounded local biosignal CSV structure without exposing row values."""
4from __future__ import annotations
6import argparse
7import csv
8import math
9import statistics
10from pathlib import Path
11from typing import Any
13from _common import (
14 MAX_CELL_CHARS,
15 MAX_CHANNELS,
16 MAX_CSV_BYTES,
17 MAX_ROWS,
18 MISSING_TOKENS,
19 CliError,
20 checked_input_file,
21 emit_json,
22 finite_float,
23 parse_name_list,
24 require_deidentified,
25 run_cli,
26)
29class OnlineStats:
30 def __init__(self) -> None:
31 self.numeric = 0
32 self.missing = 0
33 self.nonfinite = 0
34 self.nonnumeric = 0
35 self.mean = 0.0
36 self.m2 = 0.0
37 self.minimum = math.inf
38 self.maximum = -math.inf
39 self.previous: float | None = None
40 self.transitions = 0
41 self.flat_transitions = 0
42 self.flat_run = 0
43 self.max_flat_run = 0
44 self.gap_run = 0
45 self.max_gap_run = 0
47 def add_cell(self, cell: str) -> float | None:
48 normalized = cell.strip()
49 if normalized.lower() in MISSING_TOKENS:
50 self.missing += 1
51 self.gap_run += 1
52 self.max_gap_run = max(self.max_gap_run, self.gap_run)
53 self.previous = None
54 self.flat_run = 0
55 return None
56 try:
57 value = float(normalized)
58 except ValueError:
59 self.nonnumeric += 1
60 self.gap_run += 1
61 self.max_gap_run = max(self.max_gap_run, self.gap_run)
62 self.previous = None
63 self.flat_run = 0
64 return None
65 if not math.isfinite(value):
66 self.nonfinite += 1
67 self.gap_run += 1
68 self.max_gap_run = max(self.max_gap_run, self.gap_run)
69 self.previous = None
70 self.flat_run = 0
71 return None
73 self.gap_run = 0
74 self.numeric += 1
75 delta = value - self.mean
76 self.mean += delta / self.numeric
77 self.m2 += delta * (value - self.mean)
78 self.minimum = min(self.minimum, value)
79 self.maximum = max(self.maximum, value)
80 if self.previous is not None:
81 self.transitions += 1
82 if value == self.previous:
83 self.flat_transitions += 1
84 self.flat_run += 1
85 else:
86 self.flat_run = 0
87 self.max_flat_run = max(self.max_flat_run, self.flat_run)
88 self.previous = value
89 return value
91 def report(self, unit: str | None) -> dict[str, Any]:
92 variance = self.m2 / (self.numeric - 1) if self.numeric > 1 else None
93 return {
94 "flat_transition_fraction": (
95 self.flat_transitions / self.transitions if self.transitions else None
96 ),
97 "maximum": self.maximum if self.numeric else None,
98 "max_flat_run_samples": self.max_flat_run + 1 if self.max_flat_run else 0,
99 "max_gap_run_samples": self.max_gap_run,
100 "mean": self.mean if self.numeric else None,
101 "minimum": self.minimum if self.numeric else None,
102 "missing_count": self.missing,
103 "nonfinite_count": self.nonfinite,
104 "nonnumeric_count": self.nonnumeric,
105 "numeric_count": self.numeric,
106 "sample_sd": math.sqrt(variance) if variance is not None else None,
107 "unit": unit or "unspecified",
108 }
111def _parse_units(value: str | None, selected: list[str]) -> dict[str, str]:
112 if value is None:
113 return {}
114 result: dict[str, str] = {}
115 for item in value.split(","):
116 if "=" not in item:
117 raise CliError("--units entries must use COLUMN=UNIT")
118 column, unit = (part.strip() for part in item.split("=", 1))
119 if not column or not unit:
120 raise CliError("--units entries must have nonempty column and unit")
121 if column in result:
122 raise CliError(f"duplicate unit declaration for {column!r}")
123 if column not in selected:
124 raise CliError(f"unit declared for unselected column {column!r}")
125 if len(unit) > 32 or any(ord(character) < 32 for character in unit):
126 raise CliError("units must be short printable strings")
127 result[column] = unit
128 return result
131def inspect_csv(
132 path: Path,
133 *,
134 selected_names: list[str] | None,
135 time_column: str | None,
136 declared_sampling_rate: float | None,
137 units_value: str | None,
138 max_rows: int,
139 max_channels: int,
140) -> dict[str, Any]:
141 if not 1 <= max_rows <= MAX_ROWS:
142 raise CliError(f"--max-rows must be between 1 and {MAX_ROWS}")
143 if not 1 <= max_channels <= MAX_CHANNELS:
144 raise CliError(f"--max-channels must be between 1 and {MAX_CHANNELS}")
146 try:
147 handle = path.open("r", encoding="utf-8-sig", newline="")
148 except (OSError, UnicodeError) as exc:
149 raise CliError(f"cannot open CSV: {exc}") from exc
150 with handle:
151 reader = csv.reader(handle)
152 try:
153 header = [name.strip() for name in next(reader)]
154 except StopIteration as exc:
155 raise CliError("CSV is empty") from exc
156 if not header or any(not name for name in header):
157 raise CliError("CSV header names must be nonempty")
158 if len(header) != len(set(header)):
159 raise CliError("CSV header names must be unique")
160 if len(header) > max_channels:
161 raise CliError(
162 f"CSV has {len(header)} columns; --max-channels is {max_channels}"
163 )
164 if time_column is not None and time_column not in header:
165 raise CliError(f"time column {time_column!r} is absent")
166 selected = selected_names or []
167 if not selected:
168 raise CliError("no signal columns selected")
169 if len(selected) > max_channels:
170 raise CliError(
171 f"selected {len(selected)} columns; --max-channels is {max_channels}"
172 )
173 missing = [name for name in selected if name not in header]
174 if missing:
175 raise CliError(f"CSV is missing selected columns: {', '.join(missing)}")
176 if time_column in selected:
177 raise CliError("--time-column must not also be a signal column")
178 units = _parse_units(units_value, selected)
179 positions = {name: header.index(name) for name in selected}
180 time_position = header.index(time_column) if time_column else None
181 stats = {name: OnlineStats() for name in selected}
182 time_stats = OnlineStats() if time_column else None
183 positive_deltas: list[float] = []
184 previous_time: float | None = None
185 duplicate_times = 0
186 backward_times = 0
187 row_count = 0
189 try:
190 for row_count, row in enumerate(reader, start=1):
191 if row_count > max_rows:
192 raise CliError(f"CSV exceeds --max-rows={max_rows}")
193 if len(row) != len(header):
194 raise CliError(
195 f"CSV row {row_count + 1} has {len(row)} cells; "
196 f"expected {len(header)}"
197 )
198 for cell in row:
199 if len(cell) > MAX_CELL_CHARS:
200 raise CliError(
201 f"CSV row {row_count + 1} contains a cell longer than "
202 f"{MAX_CELL_CHARS} characters"
203 )
204 for name, position in positions.items():
205 stats[name].add_cell(row[position])
206 if time_position is not None and time_stats is not None:
207 time_value = time_stats.add_cell(row[time_position])
208 if time_value is not None and previous_time is not None:
209 delta = time_value - previous_time
210 if delta > 0:
211 positive_deltas.append(delta)
212 elif delta == 0:
213 duplicate_times += 1
214 else:
215 backward_times += 1
216 if time_value is not None:
217 previous_time = time_value
218 except csv.Error as exc:
219 raise CliError(f"cannot parse CSV: {exc}") from exc
221 if row_count == 0:
222 raise CliError("CSV has a header but no data rows")
224 observed_sampling_rate: float | None = None
225 median_delta: float | None = None
226 maximum_relative_jitter: float | None = None
227 if positive_deltas:
228 median_delta = statistics.median(positive_deltas)
229 if median_delta > 0:
230 observed_sampling_rate = 1.0 / median_delta
231 maximum_relative_jitter = max(
232 abs(delta - median_delta) / median_delta for delta in positive_deltas
233 )
235 duration_s: float | None = None
236 effective_rate = observed_sampling_rate or declared_sampling_rate
237 if effective_rate and row_count > 1:
238 duration_s = (row_count - 1) / effective_rate
240 channel_reports = {name: stats[name].report(units.get(name)) for name in selected}
241 warnings: list[str] = []
242 for name, report in channel_reports.items():
243 invalid = (
244 report["missing_count"]
245 + report["nonfinite_count"]
246 + report["nonnumeric_count"]
247 )
248 if invalid:
249 warnings.append(
250 f"{name}: {invalid} missing, non-finite, or non-numeric samples"
251 )
252 flat_fraction = report["flat_transition_fraction"]
253 if isinstance(flat_fraction, float) and flat_fraction > 0.20:
254 warnings.append(
255 f"{name}: more than 20% of adjacent finite samples are identical"
256 )
257 if duplicate_times:
258 warnings.append("time column contains duplicate timestamps")
259 if backward_times:
260 warnings.append("time column contains backward timestamps")
261 if (
262 declared_sampling_rate
263 and observed_sampling_rate
264 and abs(observed_sampling_rate - declared_sampling_rate)
265 / declared_sampling_rate
266 > 0.01
267 ):
268 warnings.append(
269 "observed timestamp rate differs from declared sampling rate by more than 1%"
270 )
271 if maximum_relative_jitter is not None and maximum_relative_jitter > 0.01:
272 warnings.append("timestamp intervals vary by more than 1% from the median")
274 return {
275 "channel_count_in_file": len(header),
276 "channels": channel_reports,
277 "declared_sampling_rate_hz": declared_sampling_rate,
278 "duration_s_estimate": duration_s,
279 "observed_sampling_rate_hz": observed_sampling_rate,
280 "path_redacted": True,
281 "research_use_only": True,
282 "row_count": row_count,
283 "selected_columns": selected,
284 "time": {
285 "backward_step_count": backward_times,
286 "duplicate_count": duplicate_times,
287 "maximum_relative_interval_jitter": maximum_relative_jitter,
288 "median_interval_s": median_delta,
289 "name": time_column,
290 "stats": time_stats.report("s") if time_stats else None,
291 },
292 "warnings": warnings,
293 }
296def build_parser() -> argparse.ArgumentParser:
297 parser = argparse.ArgumentParser(
298 description=(
299 "Inspect a bounded local CSV for numeric coverage, gaps, flat runs, "
300 "timestamp order, and sampling-rate consistency. Row values and paths "
301 "are never emitted."
302 )
303 )
304 parser.add_argument("--input", required=True, help="local .csv input")
305 parser.add_argument("--root", default=".", help="existing local I/O boundary")
306 parser.add_argument(
307 "--columns",
308 required=True,
309 help="explicit comma-separated signal columns",
310 )
311 parser.add_argument("--time-column", help="numeric time column in seconds")
312 parser.add_argument(
313 "--sampling-rate", type=float, help="declared sampling rate in Hz"
314 )
315 parser.add_argument(
316 "--units", help="comma-separated COLUMN=UNIT declarations, e.g. ECG=mV,EDA=uS"
317 )
318 parser.add_argument("--max-rows", type=int, default=MAX_ROWS)
319 parser.add_argument("--max-channels", type=int, default=32)
320 parser.add_argument(
321 "--deidentified",
322 action="store_true",
323 help="confirm direct and reviewed quasi-identifiers were removed",
324 )
325 parser.add_argument("--output", help="optional local .json report")
326 parser.add_argument("--force", action="store_true")
327 return parser
330def main() -> None:
331 args = build_parser().parse_args()
332 require_deidentified(args.deidentified)
333 declared_rate = (
334 finite_float(
335 args.sampling_rate,
336 name="--sampling-rate",
337 minimum=0.01,
338 maximum=1_000_000.0,
339 )
340 if args.sampling_rate is not None
341 else None
342 )
343 path = checked_input_file(
344 args.input,
345 root=args.root,
346 suffixes={".csv"},
347 max_bytes=MAX_CSV_BYTES,
348 )
349 report = inspect_csv(
350 path,
351 selected_names=parse_name_list(args.columns, name="--columns"),
352 time_column=args.time_column,
353 declared_sampling_rate=declared_rate,
354 units_value=args.units,
355 max_rows=args.max_rows,
356 max_channels=args.max_channels,
357 )
358 emit_json(report, output=args.output, root=args.root, force=args.force)
361if __name__ == "__main__":
362 raise SystemExit(run_cli(main))