Onset Validation

neureptrace.onset_validation runs the onset detector inside named time chunks. Use it as a compact negative/positive control before interpreting onset latencies.

Example:

python -m neureptrace.onset_validation \
  results/nod_sub-01_animate_observations.csv \
  --threshold-window -0.10 0.00 \
  --threshold-quantile 0.95 \
  --threshold-method max_run \
  --min-consecutive 2 \
  --chunk pre:-0.30:-0.05:null \
  --chunk early:0.05:0.20:early \
  --chunk late:0.20:0.60:positive \
  --out-events results/nod_sub-01_animate_onset_chunk_events.csv \
  --out-summary results/nod_sub-01_animate_onset_chunk_summary.csv

Pre-stimulus chunks should show low detection rates. Post-stimulus chunks should show higher detection rates only when the decoded probability traces contain stable task-related evidence.

neureptrace.onset_validation

OnsetChunk dataclass

One time chunk used to validate onset behavior.

Source code in src/neureptrace/onset_validation.py

@dataclass(frozen=True)
class OnsetChunk:
    """One time chunk used to validate onset behavior."""

    name: str
    start: float
    stop: float
    expected_response: str = "unknown"

parse_chunk_spec(spec)

Parse a chunk specification of the form name:start:stop[:expected].

Source code in src/neureptrace/onset_validation.py

def parse_chunk_spec(spec: str) -> OnsetChunk:
    """Parse a chunk specification of the form ``name:start:stop[:expected]``."""

    parts = spec.split(":")
    if len(parts) not in (3, 4):
        raise ValueError("Chunk specs must have the form name:start:stop[:expected].")
    name, raw_start, raw_stop = parts[:3]
    if not name:
        raise ValueError("Chunk name must not be empty.")
    start = float(raw_start)
    stop = float(raw_stop)
    if start > stop:
        raise ValueError("Chunk start must be less than or equal to chunk stop.")
    expected = parts[3] if len(parts) == 4 else "unknown"
    return OnsetChunk(name=name, start=start, stop=stop, expected_response=expected)

run_onset_chunk_validation(observation_csvs, *, chunks=DEFAULT_CHUNKS, threshold_window=DEFAULT_THRESHOLD_WINDOW, threshold_quantile=DEFAULT_THRESHOLD_QUANTILE, threshold_method='max_run', score_column='confidence', min_consecutive=2, min_duration=None, require_stable_prediction=True, out_events=None, out_summary=None)

Read observation CSVs and write optional chunk-validation summaries.

Source code in src/neureptrace/onset_validation.py

def run_onset_chunk_validation(
    observation_csvs: list[Path],
    *,
    chunks: list[OnsetChunk] | tuple[OnsetChunk, ...] = DEFAULT_CHUNKS,
    threshold_window: tuple[float, float] = DEFAULT_THRESHOLD_WINDOW,
    threshold_quantile: float = DEFAULT_THRESHOLD_QUANTILE,
    threshold_method: str = "max_run",
    score_column: str = "confidence",
    min_consecutive: int = 2,
    min_duration: float | None = None,
    require_stable_prediction: bool = True,
    out_events: Path | None = None,
    out_summary: Path | None = None,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Read observation CSVs and write optional chunk-validation summaries."""

    observations = read_probability_observations(observation_csvs)
    events, summary = summarize_onset_chunks(
        observations,
        chunks,
        threshold_window=threshold_window,
        threshold_quantile=threshold_quantile,
        threshold_method=threshold_method,
        score_column=score_column,
        min_consecutive=min_consecutive,
        min_duration=min_duration,
        require_stable_prediction=require_stable_prediction,
    )
    if out_events is not None:
        out_events.parent.mkdir(parents=True, exist_ok=True)
        events.to_csv(out_events, index=False)
    if out_summary is not None:
        out_summary.parent.mkdir(parents=True, exist_ok=True)
        summary.to_csv(out_summary, index=False)
    return events, summary

summarize_onset_chunks(observations, chunks=DEFAULT_CHUNKS, *, threshold_window=DEFAULT_THRESHOLD_WINDOW, threshold_quantile=DEFAULT_THRESHOLD_QUANTILE, threshold_method='max_run', score_column='confidence', min_consecutive=2, min_duration=None, require_stable_prediction=True)

Run onset detection separately inside named time chunks.

The threshold is still estimated from threshold_window over the full observation table. Each chunk only limits the candidate event window. This makes pre-stimulus chunks useful as negative controls and post-stimulus chunks useful as coarse positive checks.

Source code in src/neureptrace/onset_validation.py

def summarize_onset_chunks(
    observations: pd.DataFrame,
    chunks: list[OnsetChunk] | tuple[OnsetChunk, ...] = DEFAULT_CHUNKS,
    *,
    threshold_window: tuple[float, float] = DEFAULT_THRESHOLD_WINDOW,
    threshold_quantile: float = DEFAULT_THRESHOLD_QUANTILE,
    threshold_method: str = "max_run",
    score_column: str = "confidence",
    min_consecutive: int = 2,
    min_duration: float | None = None,
    require_stable_prediction: bool = True,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Run onset detection separately inside named time chunks.

    The threshold is still estimated from ``threshold_window`` over the full
    observation table. Each chunk only limits the candidate event window. This
    makes pre-stimulus chunks useful as negative controls and post-stimulus
    chunks useful as coarse positive checks.
    """

    if not chunks:
        raise ValueError("At least one onset-validation chunk is required.")
    thresholded = annotate_threshold_crossings(
        observations,
        threshold_window=threshold_window,
        threshold_quantile=threshold_quantile,
        threshold_method=threshold_method,
        score_column=score_column,
        min_consecutive=min_consecutive,
        min_duration=min_duration,
        require_stable_prediction=require_stable_prediction,
    )
    event_frames = []
    summary_frames = []
    for chunk in chunks:
        scan_mask = (
            thresholded["time"].between(threshold_window[0], threshold_window[1])
            | thresholded["time"].between(chunk.start, chunk.stop)
        )
        events = detect_onsets(
            thresholded.loc[scan_mask],
            threshold_window=threshold_window,
            threshold_quantile=threshold_quantile,
            threshold_method=threshold_method,
            score_column=score_column,
            detection_window=(chunk.start, chunk.stop),
            min_consecutive=min_consecutive,
            min_duration=min_duration,
            require_stable_prediction=require_stable_prediction,
        )
        summary = summarize_onset_events(events)
        event_frames.append(_tag_chunk(events, chunk))
        summary_frames.append(_tag_chunk(summary, chunk))
    return pd.concat(event_frames, ignore_index=True), pd.concat(summary_frames, ignore_index=True)