Skills › Scientific Agent Skills › neurokit2 › validate_multimodal.py
validate_multimodal.py
386 lines · skills/neurokit2/scripts/validate_multimodal.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate bounded multimodal stream schemas and temporal alignment."""
4from __future__ import annotations
6import argparse
7import math
8import re
9import statistics
10from itertools import pairwise
11from pathlib import Path
12from typing import Any
14from _common import (
15 MAX_CSV_BYTES,
16 MAX_ROWS,
17 CliError,
18 checked_input_file,
19 checked_root,
20 emit_json,
21 finite_float,
22 load_json_object,
23 read_numeric_columns,
24 require_deidentified,
25 run_cli,
26 validate_keys,
27)
29MAX_STREAMS = 16
30NAME_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{0,63}$")
31SYNC_METHODS = {
32 "hardware_trigger",
33 "shared_clock",
34 "timestamp",
35 "validated_manual",
36}
39def _stream_profile(
40 stream: dict[str, Any],
41 *,
42 root: Path,
43 max_rows: int,
44) -> tuple[dict[str, Any], list[str], list[str], list[float]]:
45 validate_keys(
46 stream,
47 allowed={
48 "name",
49 "path",
50 "value_column",
51 "time_column",
52 "sampling_rate_hz",
53 "unit",
54 "start_time_s",
55 },
56 required={"name", "path", "value_column", "sampling_rate_hz", "unit"},
57 context="stream",
58 )
59 name = stream["name"]
60 if not isinstance(name, str) or not NAME_PATTERN.fullmatch(name):
61 raise CliError("stream name must be a short identifier beginning with a letter")
62 for key in ("path", "value_column", "unit"):
63 if not isinstance(stream[key], str) or not stream[key].strip():
64 raise CliError(f"stream {name!r} {key} must be a nonempty string")
65 if len(stream["unit"]) > 32:
66 raise CliError(f"stream {name!r} unit is too long")
67 time_column = stream.get("time_column")
68 if time_column is not None and (
69 not isinstance(time_column, str) or not time_column.strip()
70 ):
71 raise CliError(f"stream {name!r} time_column must be a nonempty string")
72 if time_column is not None and "start_time_s" in stream:
73 raise CliError(
74 f"stream {name!r} must not combine time_column with start_time_s"
75 )
76 declared_rate = finite_float(
77 stream["sampling_rate_hz"],
78 name=f"{name}.sampling_rate_hz",
79 minimum=0.01,
80 maximum=1_000_000.0,
81 )
82 start_time = (
83 finite_float(stream["start_time_s"], name=f"{name}.start_time_s")
84 if "start_time_s" in stream
85 else 0.0
86 )
87 path = checked_input_file(
88 stream["path"],
89 root=root,
90 suffixes={".csv"},
91 max_bytes=MAX_CSV_BYTES,
92 )
93 columns = [stream["value_column"]]
94 if time_column is not None:
95 columns.append(time_column)
96 values, _, row_count = read_numeric_columns(
97 path,
98 columns,
99 max_rows=max_rows,
100 allow_missing=True,
101 )
102 signal = values[stream["value_column"]]
103 missing_count = sum(value is None for value in signal)
104 finite_pairs = [
105 (previous, current) for previous, current in pairwise(signal)
106 if previous is not None and current is not None
107 ]
108 flat_count = sum(current == previous for previous, current in finite_pairs)
109 flat_fraction = (
110 flat_count / len(finite_pairs) if finite_pairs else None
111 )
112 errors: list[str] = []
113 warnings: list[str] = []
114 observed_rate: float | None = None
115 interval_jitter: float | None = None
117 if time_column is not None:
118 raw_times = values[time_column]
119 if any(value is None for value in raw_times):
120 errors.append("time column contains missing values")
121 times = [float(value) for value in raw_times if value is not None]
122 else:
123 times = [float(value) for value in raw_times]
124 if len(times) >= 2:
125 deltas = [current - previous for previous, current in pairwise(times)]
126 if any(delta <= 0 for delta in deltas):
127 errors.append("timestamps are not strictly increasing")
128 positive = [delta for delta in deltas if delta > 0]
129 if positive:
130 median_delta = statistics.median(positive)
131 observed_rate = 1.0 / median_delta
132 interval_jitter = max(
133 abs(delta - median_delta) / median_delta for delta in positive
134 )
135 start_time = times[0]
136 end_time = times[-1]
137 else:
138 errors.append("fewer than two valid timestamps")
139 end_time = start_time
140 else:
141 end_time = start_time + (row_count - 1) / declared_rate
142 times = [start_time + index / declared_rate for index in range(row_count)]
144 grid_error_samples = (
145 max(abs((time - start_time) * declared_rate - index)
146 for index, time in enumerate(times))
147 if len(times) == row_count else None
148 )
150 if observed_rate is not None:
151 relative_rate_error = abs(observed_rate - declared_rate) / declared_rate
152 if relative_rate_error > 0.01:
153 errors.append(
154 "observed timestamp rate differs from declared rate by more than 1%"
155 )
156 if interval_jitter is not None and interval_jitter > 0.01:
157 warnings.append("timestamp interval jitter exceeds 1% of the median")
158 if grid_error_samples is not None and grid_error_samples > 0.5:
159 warnings.append("timestamps deviate more than half a sample from the declared grid")
160 if missing_count:
161 warnings.append(f"{missing_count} signal values are missing")
162 if flat_fraction is not None and flat_fraction > 0.20:
163 warnings.append("more than 20% of adjacent finite signal values are identical")
165 profile = {
166 "declared_sampling_rate_hz": declared_rate,
167 "duration_s": max(0.0, end_time - start_time),
168 "end_time_s": end_time,
169 "flat_transition_fraction": flat_fraction,
170 "missing_count": missing_count,
171 "name": name,
172 "observed_sampling_rate_hz": observed_rate,
173 "path_redacted": True,
174 "row_count": row_count,
175 "start_time_s": start_time,
176 "timestamp_maximum_relative_jitter": interval_jitter,
177 "timestamp_maximum_grid_error_samples": grid_error_samples,
178 "time_basis": "observed" if time_column is not None else "declared_uniform_grid",
179 "unit": stream["unit"],
180 }
181 return profile, errors, warnings, times
184def validate_manifest(
185 document: dict[str, Any],
186 *,
187 root: Path,
188 max_rows: int,
189) -> dict[str, Any]:
190 validate_keys(
191 document,
192 allowed={"schema_version", "streams", "alignment"},
193 required={"schema_version", "streams", "alignment"},
194 context="manifest",
195 )
196 if document["schema_version"] != "1.0":
197 raise CliError("schema_version must be '1.0'")
198 streams = document["streams"]
199 if not isinstance(streams, list) or not 1 <= len(streams) <= MAX_STREAMS:
200 raise CliError(f"streams must contain 1 to {MAX_STREAMS} objects")
201 if any(not isinstance(stream, dict) for stream in streams):
202 raise CliError("every stream must be an object")
203 alignment = document["alignment"]
204 if not isinstance(alignment, dict):
205 raise CliError("alignment must be an object")
206 validate_keys(
207 alignment,
208 allowed={
209 "reference_stream",
210 "synchronization",
211 "max_start_offset_ms",
212 "minimum_overlap_s",
213 },
214 required={
215 "reference_stream",
216 "synchronization",
217 "max_start_offset_ms",
218 "minimum_overlap_s",
219 },
220 context="alignment",
221 )
222 if alignment["synchronization"] not in SYNC_METHODS:
223 raise CliError(
224 "alignment.synchronization must be one of: "
225 + ", ".join(sorted(SYNC_METHODS))
226 )
227 max_offset_ms = finite_float(
228 alignment["max_start_offset_ms"],
229 name="alignment.max_start_offset_ms",
230 minimum=0.0,
231 maximum=60_000.0,
232 )
233 minimum_overlap_s = finite_float(
234 alignment["minimum_overlap_s"],
235 name="alignment.minimum_overlap_s",
236 minimum=0.0,
237 maximum=31_536_000.0,
238 )
240 profiles: list[dict[str, Any]] = []
241 timestamps: dict[str, list[float]] = {}
242 errors: list[str] = []
243 warnings: list[str] = []
244 for index, stream in enumerate(streams, start=1):
245 try:
246 profile, stream_errors, stream_warnings, stream_times = _stream_profile(
247 stream, root=root, max_rows=max_rows
248 )
249 except CliError as exc:
250 raise CliError(f"stream {index}: {exc}") from exc
251 profiles.append(profile)
252 timestamps[profile["name"]] = stream_times
253 errors.extend(f"{profile['name']}: {message}" for message in stream_errors)
254 warnings.extend(f"{profile['name']}: {message}" for message in stream_warnings)
256 names = [profile["name"] for profile in profiles]
257 if len(names) != len(set(names)):
258 raise CliError("stream names must be unique")
259 reference_name = alignment["reference_stream"]
260 if reference_name not in names:
261 raise CliError("alignment.reference_stream is not a stream name")
262 reference = next(
263 profile for profile in profiles if profile["name"] == reference_name
264 )
266 start_offsets_ms: dict[str, float] = {}
267 for profile in profiles:
268 offset_ms = (profile["start_time_s"] - reference["start_time_s"]) * 1000.0
269 start_offsets_ms[profile["name"]] = offset_ms
270 if abs(offset_ms) > max_offset_ms:
271 errors.append(
272 f"{profile['name']}: start offset {offset_ms:.6g} ms exceeds "
273 f"{max_offset_ms:.6g} ms"
274 )
275 overlap_start = max(profile["start_time_s"] for profile in profiles)
276 overlap_end = min(profile["end_time_s"] for profile in profiles)
277 overlap_s = max(0.0, overlap_end - overlap_start)
278 if overlap_s < minimum_overlap_s:
279 errors.append(
280 f"common overlap {overlap_s:.6g} s is below {minimum_overlap_s:.6g} s"
281 )
283 rates = [profile["declared_sampling_rate_hz"] for profile in profiles]
284 counts = [profile["row_count"] for profile in profiles]
285 same_rate = all(
286 math.isclose(rate, rates[0], rel_tol=1e-9, abs_tol=1e-12) for rate in rates[1:]
287 )
288 same_count = all(count == counts[0] for count in counts[1:])
289 no_missing = all(profile["missing_count"] == 0 for profile in profiles)
290 starts_within_sample = all(
291 abs(start_offsets_ms[profile["name"]])
292 <= (500.0 / profile["declared_sampling_rate_hz"])
293 for profile in profiles
294 )
295 maximum_sample_offsets_ms: dict[str, float | None] = {}
296 reference_times = timestamps[reference_name]
297 for profile in profiles:
298 times = timestamps[profile["name"]]
299 maximum_sample_offsets_ms[profile["name"]] = (
300 max(abs(time - reference_time) * 1000.0
301 for time, reference_time in zip(times, reference_times))
302 if same_rate and same_count and len(times) == len(reference_times)
303 and len(times) == profile["row_count"] else None
304 )
305 all_samples_aligned = all(
306 offset is not None and offset <= 500.0 / rates[0]
307 for offset in maximum_sample_offsets_ms.values()
308 )
309 uniform_grids = all(
310 profile["timestamp_maximum_grid_error_samples"] is not None
311 and profile["timestamp_maximum_grid_error_samples"] <= 0.5
312 for profile in profiles
313 )
314 bio_process_compatible = (
315 same_rate and same_count and no_missing and starts_within_sample and not errors
316 and all_samples_aligned and uniform_grids
317 )
318 if not same_rate:
319 warnings.append(
320 "native sampling rates differ; process at native rates, align clocks, "
321 "then resample continuous channels to an explicit common grid"
322 )
323 if not bio_process_compatible:
324 warnings.append(
325 "do not pass these streams directly to bio_process(); NeuroKit2 "
326 "0.2.13 does not synchronize or automatically resample inputs"
327 )
329 return {
330 "alignment": {
331 "common_overlap_s": overlap_s,
332 "max_start_offset_ms": max_offset_ms,
333 "minimum_overlap_s": minimum_overlap_s,
334 "reference_stream": reference_name,
335 "start_offsets_ms": start_offsets_ms,
336 "maximum_sample_offsets_ms": maximum_sample_offsets_ms,
337 "grid_tolerance_samples": 0.5,
338 "synchronization": alignment["synchronization"],
339 },
340 "bio_process_direct_input_compatible": bio_process_compatible,
341 "errors": errors,
342 "path_redacted": True,
343 "schema_version": "1.0",
344 "streams": profiles,
345 "valid": not errors,
346 "warnings": warnings,
347 }
350def build_parser() -> argparse.ArgumentParser:
351 parser = argparse.ArgumentParser(
352 description=(
353 "Validate a strict JSON manifest of bounded local biosignal CSV "
354 "streams, units, clocks, rates, starts, overlap, and missingness."
355 )
356 )
357 parser.add_argument("--manifest", required=True, help="local .json manifest")
358 parser.add_argument("--root", default=".", help="existing local I/O boundary")
359 parser.add_argument("--max-rows", type=int, default=MAX_ROWS)
360 parser.add_argument(
361 "--deidentified",
362 action="store_true",
363 help="confirm direct and reviewed quasi-identifiers were removed",
364 )
365 parser.add_argument("--output", help="optional local .json report")
366 parser.add_argument("--force", action="store_true")
367 return parser
370def main() -> None:
371 args = build_parser().parse_args()
372 require_deidentified(args.deidentified)
373 if not 1 <= args.max_rows <= MAX_ROWS:
374 raise CliError(f"--max-rows must be between 1 and {MAX_ROWS}")
375 root = checked_root(args.root)
376 document = load_json_object(args.manifest, root=root)
377 report = validate_manifest(document, root=root, max_rows=args.max_rows)
378 emit_json(report, output=args.output, root=root, force=args.force)
379 if not report["valid"]:
380 raise CliError(
381 f"multimodal validation failed with {len(report['errors'])} error(s)"
382 )
385if __name__ == "__main__":
386 raise SystemExit(run_cli(main))