Temporal State Workflow

neureptrace.temporal_state_workflow

TemporalStateTask dataclass

One NOD task used in the calibration-aware temporal-state workflow.

Source code in src/neureptrace/temporal_state_workflow.py

@dataclass(frozen=True)
class TemporalStateTask:
    """One NOD task used in the calibration-aware temporal-state workflow."""

    task_id: str
    label: str
    manifest_name: str

TemporalStateTaskOutput dataclass

Compact paths produced for one temporal-state task.

Source code in src/neureptrace/temporal_state_workflow.py

@dataclass(frozen=True)
class TemporalStateTaskOutput:
    """Compact paths produced for one temporal-state task."""

    task: TemporalStateTask
    task_dir: Path
    manifest_csv: Path
    validation_csv: Path
    temporal_summary_csv: Path
    state_trace_csv: Path
    emission_compare_csv: Path
    emission_compare_report: Path
    semantic_time_csv: Path
    semantic_stages_csv: Path
    semantic_stages_report: Path

TemporalStateWorkflowRun dataclass

Top-level calibration-aware temporal-state workflow outputs.

Source code in src/neureptrace/temporal_state_workflow.py

@dataclass(frozen=True)
class TemporalStateWorkflowRun:
    """Top-level calibration-aware temporal-state workflow outputs."""

    out_dir: Path
    task_outputs: list[TemporalStateTaskOutput]
    temporal_state_summary_csv: Path
    temporal_state_figure: Path
    evidence_report: Path
    command_log: Path
    exported_artifacts: list[Path]

build_evidence_report(temporal_state_summary, *, out_dir, temporal_state_summary_csv, figure_path, temporal_all_csv, emission_all_csv, stage_time_all_csv, stages_all_csv)

Build a compact temporal-state evidence note from generated artifacts.

Source code in src/neureptrace/temporal_state_workflow.py

def build_evidence_report(
    temporal_state_summary: pd.DataFrame,
    *,
    out_dir: Path,
    temporal_state_summary_csv: Path,
    figure_path: Path,
    temporal_all_csv: Path,
    emission_all_csv: Path,
    stage_time_all_csv: Path,
    stages_all_csv: Path,
) -> str:
    """Build a compact temporal-state evidence note from generated artifacts."""
    columns = [
        "task_label",
        "decoder",
        "preferred_emission_mode",
        "delta_control_margin",
        "delta_effect_minus_baseline_gain",
        "calibrated_peak_posterior_true_class",
        "uncalibrated_peak_posterior_true_class",
        "calibrated_peak_n_subjects",
        "uncalibrated_peak_n_subjects",
    ]
    lines = [
        "# Evidence: Calibration-Aware Temporal State Inference",
        "",
        "Central claim under test: calibrated decoder probabilities can change downstream temporal state inference, not only reported uncertainty.",
        "",
        "## Central Table",
        "",
        *_markdown_table(temporal_state_summary, [column for column in columns if column in temporal_state_summary.columns]),
        "",
        "## Compact Artifacts",
        "",
        f"- Central table: `{_display_path(temporal_state_summary_csv, out_dir)}`",
        f"- Summary figure: `{_display_path(figure_path, out_dir)}`",
        f"- Temporal model rows: `{_display_path(temporal_all_csv, out_dir)}`",
        f"- Emission comparison rows: `{_display_path(emission_all_csv, out_dir)}`",
        f"- Semantic-stage time rows: `{_display_path(stage_time_all_csv, out_dir)}`",
        f"- Semantic-stage intervals: `{_display_path(stages_all_csv, out_dir)}`",
        "",
        "## Reading Rule",
        "",
        "The primary temporal-state metric is `delta_control_margin`: calibrated observed effect-window persistence gain minus the strongest calibrated control, compared with the same uncalibrated margin. Positive values favor calibrated emissions. Semantic-stage rows are supporting evidence and should be interpreted only together with the shuffled-time, shuffled-label, and baseline-window controls.",
        "",
    ]
    return "\n".join(lines)

build_temporal_state_summary(emission_compare, stages, stage_time)

Build the central compact table for the temporal-state evidence note.

Source code in src/neureptrace/temporal_state_workflow.py

def build_temporal_state_summary(emission_compare: pd.DataFrame, stages: pd.DataFrame, stage_time: pd.DataFrame) -> pd.DataFrame:
    """Build the central compact table for the temporal-state evidence note."""
    if emission_compare.empty:
        return pd.DataFrame()

    rows = []
    for row in emission_compare.itertuples(index=False):
        task = str(row.task)
        decoder = str(row.decoder)
        rows.append(
            {
                "task": task,
                "task_label": str(row.task_label),
                "decoder": decoder,
                "preferred_emission_mode": str(row.preferred_emission_mode),
                "delta_control_margin": float(row.delta_control_margin),
                "calibrated_control_margin": float(row.calibrated_control_margin),
                "uncalibrated_control_margin": float(row.uncalibrated_control_margin),
                "delta_effect_minus_baseline_gain": float(row.delta_effect_minus_baseline_gain),
                "calibrated_best_stay_probability": float(row.calibrated_best_stay_probability),
                "uncalibrated_best_stay_probability": float(row.uncalibrated_best_stay_probability),
                **_stage_stats(stages, stage_time, task, decoder, "calibrated"),
                **_stage_stats(stages, stage_time, task, decoder, "uncalibrated"),
            }
        )
    return pd.DataFrame(rows).sort_values(["task", "decoder"]).reset_index(drop=True)

export_temporal_state_artifacts(source_dir, destination_dir, *, max_mb=50.0, dry_run=False)

Copy compact temporal-state artifacts to the compact export directory.

Source code in src/neureptrace/temporal_state_workflow.py

def export_temporal_state_artifacts(
    source_dir: Path,
    destination_dir: Path,
    *,
    max_mb: float = 50.0,
    dry_run: bool = False,
) -> list[Path]:
    """Copy compact temporal-state artifacts to the compact export directory."""
    source_dir = source_dir.resolve()
    destination_dir = destination_dir.resolve()
    artifacts = _collect_compact_artifacts(source_dir)
    if not artifacts:
        raise FileNotFoundError(f"No compact temporal-state artifacts found in {source_dir}.")

    size_mb = sum(path.stat().st_size for path in artifacts) / (1024 * 1024)
    if size_mb > max_mb:
        raise ValueError(f"Compact temporal-state artifacts are {size_mb:.2f} MB, above limit {max_mb:.2f} MB.")
    if dry_run:
        return artifacts

    copied: list[Path] = []
    for artifact in artifacts:
        relative = artifact.relative_to(source_dir)
        target = destination_dir / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(artifact, target)
        copied.append(target)
    return copied

plot_temporal_state_reliability(temporal_state_summary, stage_time, out_path, *, plot_decoder=None)

Plot the calibration-aware temporal-state control-margin and semantic-stage reliability summary.

Source code in src/neureptrace/temporal_state_workflow.py

def plot_temporal_state_reliability(
    temporal_state_summary: pd.DataFrame,
    stage_time: pd.DataFrame,
    out_path: Path,
    *,
    plot_decoder: str | None = None,
) -> Path:
    """Plot the calibration-aware temporal-state control-margin and semantic-stage reliability summary."""
    fig, axes = plt.subplots(1, 2, figsize=(11.0, 4.2))

    ax = axes[0]
    if temporal_state_summary.empty:
        ax.text(0.5, 0.5, "No emission comparison rows", ha="center", va="center")
        ax.axis("off")
    else:
        summary = temporal_state_summary.copy()
        task_labels = list(dict.fromkeys(summary["task_label"].astype(str)))
        decoders = list(dict.fromkeys(summary["decoder"].astype(str)))
        width = 0.8 / max(len(decoders), 1)
        x = range(len(task_labels))
        for decoder_index, decoder in enumerate(decoders):
            values = []
            for task_label in task_labels:
                match = summary[(summary["task_label"] == task_label) & (summary["decoder"] == decoder)]
                values.append(float(match["delta_control_margin"].iloc[0]) if not match.empty else float("nan"))
            offsets = [position - 0.4 + width / 2 + decoder_index * width for position in x]
            ax.bar(offsets, values, width=width, label=decoder)
        ax.axhline(0.0, color="0.35", linewidth=1.0)
        ax.set_xticks(list(x))
        ax.set_xticklabels(task_labels, rotation=25, ha="right")
        ax.set_ylabel("Delta control margin")
        ax.set_title("Calibrated vs uncalibrated emissions")
        ax.legend(loc="best")
        ax.grid(axis="y", color="0.9", linewidth=0.8)

    ax = axes[1]
    if stage_time.empty or "posterior_true_class_mean" not in stage_time.columns:
        ax.text(0.5, 0.5, "No semantic-stage time rows", ha="center", va="center")
        ax.axis("off")
    else:
        plot_frame = stage_time.copy()
        available_decoders = list(dict.fromkeys(plot_frame["decoder"].astype(str))) if "decoder" in plot_frame else []
        if plot_decoder is None:
            plot_decoder = "linear_svm" if "linear_svm" in available_decoders else (available_decoders[0] if available_decoders else None)
        if plot_decoder is not None and "decoder" in plot_frame:
            plot_frame = plot_frame.loc[plot_frame["decoder"].astype(str) == plot_decoder]
        grouped = (
            plot_frame.groupby(["emission_mode", "time"], as_index=False)["posterior_true_class_mean"]
            .mean()
            .sort_values(["emission_mode", "time"])
        )
        for emission_mode, group in grouped.groupby("emission_mode", sort=True):
            ax.plot(group["time"], group["posterior_true_class_mean"], label=str(emission_mode))
        ax.axvline(0.0, color="0.6", linestyle=":", linewidth=1.0)
        ax.set_xlabel("Time (s)")
        ax.set_ylabel("Posterior on true class")
        ax.set_title(f"Semantic-stage reliability ({plot_decoder or 'decoder'})")
        ax.legend(loc="best")
        ax.grid(True, color="0.9", linewidth=0.8)

    fig.tight_layout()
    out_path.parent.mkdir(parents=True, exist_ok=True)
    fig.savefig(out_path, dpi=160)
    plt.close(fig)
    return out_path

prepare_temporal_state_manifest(source_manifest, out_manifest, *, data_root, decoders, max_subjects=None, expected_subjects=19)

Prepare a task manifest with runner-local data paths and decoder rows.

Source code in src/neureptrace/temporal_state_workflow.py

def prepare_temporal_state_manifest(
    source_manifest: Path,
    out_manifest: Path,
    *,
    data_root: Path | None,
    decoders: tuple[str, ...],
    max_subjects: int | None = None,
    expected_subjects: int | None = 19,
) -> pd.DataFrame:
    """Prepare a task manifest with runner-local data paths and decoder rows."""
    manifest = pd.read_csv(source_manifest)
    if "subject" not in manifest.columns:
        raise ValueError(f"{source_manifest} is missing a subject column.")

    if max_subjects is not None:
        if max_subjects < 1:
            raise ValueError("max_subjects must be positive.")
        keep_subjects = list(dict.fromkeys(manifest["subject"].astype(str)))[:max_subjects]
        manifest = manifest.loc[manifest["subject"].astype(str).isin(keep_subjects)].copy()

    n_subjects = manifest["subject"].astype(str).nunique()
    if max_subjects is None and expected_subjects is not None and n_subjects != expected_subjects:
        raise ValueError(f"{source_manifest} has {n_subjects} unique subject(s), expected {expected_subjects}.")

    if data_root is not None:
        data_root = data_root.expanduser().resolve()
        for column in ("epochs", "events_csv"):
            if column in manifest.columns:
                manifest[column] = manifest[column].map(lambda value: str(data_root / Path(str(value)).name))

    decoder_rows = []
    for decoder in decoders:
        decoder_frame = manifest.copy()
        decoder_frame["decoder"] = decoder
        decoder_rows.append(decoder_frame)
    prepared = pd.concat(decoder_rows, ignore_index=True)
    out_manifest.parent.mkdir(parents=True, exist_ok=True)
    prepared.to_csv(out_manifest, index=False)
    return prepared

run_temporal_state_workflow(*, out_dir, data_root=None, compact_export_dir=None, task_ids=None, decoders=DEFAULT_DECODERS, n_permutations=100, random_seed=13, stay_grid_size=200, posterior_threshold=0.6, match_threshold=0.6, min_duration=0.04, max_subjects=None, expected_subjects=19, resume=True, max_export_mb=50.0, command_line='python -m neureptrace.temporal_state_workflow')

Run the reproducible calibration-aware NOD temporal-state workflow.

Source code in src/neureptrace/temporal_state_workflow.py

def run_temporal_state_workflow(
    *,
    out_dir: Path,
    data_root: Path | None = None,
    compact_export_dir: Path | None = None,
    task_ids: tuple[str, ...] | None = None,
    decoders: tuple[str, ...] = DEFAULT_DECODERS,
    n_permutations: int = 100,
    random_seed: int = 13,
    stay_grid_size: int = 200,
    posterior_threshold: float = 0.6,
    match_threshold: float = 0.6,
    min_duration: float = 0.04,
    max_subjects: int | None = None,
    expected_subjects: int | None = 19,
    resume: bool = True,
    max_export_mb: float = 50.0,
    command_line: str = "python -m neureptrace.temporal_state_workflow",
) -> TemporalStateWorkflowRun:
    """Run the reproducible calibration-aware NOD temporal-state workflow."""
    repo_root = _repo_root()
    out_dir = out_dir.resolve()
    tasks = _selected_tasks(task_ids)
    decoders = _normal_decoders(decoders)
    task_outputs: list[TemporalStateTaskOutput] = []

    for task in tasks:
        task_dir = out_dir / task.task_id
        manifest_csv = task_dir / "manifest.csv"
        validation_csv = task_dir / "validation.csv"
        source_manifest = repo_root / "benchmarks" / task.manifest_name
        prepare_temporal_state_manifest(
            source_manifest,
            manifest_csv,
            data_root=data_root,
            decoders=decoders,
            max_subjects=max_subjects,
            expected_subjects=expected_subjects,
        )
        _validate_prepared_manifest(manifest_csv, validation_csv)
        run_benchmark_manifest(
            manifest_csv,
            out_dir=task_dir,
            aggregate_out=task_dir / "summary.csv",
            plot_out=task_dir / "summary.png",
            chance=0.5,
            default_emission_mode="both",
            calibration_dir=task_dir / "calibration",
            observation_dir=task_dir / "observations",
            resume=resume,
        )

        observation_paths = _task_observation_paths(task_dir)
        if not observation_paths:
            raise FileNotFoundError(f"No observation CSVs were produced for {task.task_id}.")

        temporal_summary_csv = task_dir / "temporal_model.csv"
        state_trace_csv = task_dir / "state_trace.csv"
        fit_temporal_models(
            observation_paths,
            n_permutations=n_permutations,
            random_seed=random_seed,
            stay_grid_size=stay_grid_size,
            out_summary=temporal_summary_csv,
            out_states=state_trace_csv,
        )
        emission_compare_csv = task_dir / "emission_compare.csv"
        emission_compare_report = task_dir / "emission_compare.md"
        compare_temporal_summary(
            temporal_summary_csv,
            out_csv=emission_compare_csv,
            out_report=emission_compare_report,
        )
        semantic_time_csv = task_dir / "semantic_stage_time.csv"
        semantic_stages_csv = task_dir / "semantic_stages.csv"
        semantic_stages_report = task_dir / "semantic_stages.md"
        analyze_semantic_stages(
            [state_trace_csv],
            posterior_threshold=posterior_threshold,
            match_threshold=match_threshold,
            min_duration=min_duration,
            out_time=semantic_time_csv,
            out_stages=semantic_stages_csv,
            out_report=semantic_stages_report,
        )

        try:
            plot_time_decode_results(task_dir / "summary.csv", task_dir / "summary.png", chance=0.5, title=task.label)
        except ValueError:
            pass

        task_outputs.append(
            TemporalStateTaskOutput(
                task=task,
                task_dir=task_dir,
                manifest_csv=manifest_csv,
                validation_csv=validation_csv,
                temporal_summary_csv=temporal_summary_csv,
                state_trace_csv=state_trace_csv,
                emission_compare_csv=emission_compare_csv,
                emission_compare_report=emission_compare_report,
                semantic_time_csv=semantic_time_csv,
                semantic_stages_csv=semantic_stages_csv,
                semantic_stages_report=semantic_stages_report,
            )
        )

    temporal_all_csv = out_dir / "temporal_model_all.csv"
    emission_all_csv = out_dir / "emission_compare_all.csv"
    stage_time_all_csv = out_dir / "semantic_stage_time_all.csv"
    stages_all_csv = out_dir / "semantic_stages_all.csv"
    _write_combined_csv(task_outputs, "temporal_summary_csv", temporal_all_csv)
    emission_compare = _write_combined_csv(task_outputs, "emission_compare_csv", emission_all_csv)
    stage_time = _write_combined_csv(task_outputs, "semantic_time_csv", stage_time_all_csv)
    stages = _write_combined_csv(task_outputs, "semantic_stages_csv", stages_all_csv)

    temporal_state_summary = build_temporal_state_summary(emission_compare, stages, stage_time)
    temporal_state_summary_csv = out_dir / "temporal_state_summary.csv"
    temporal_state_summary.to_csv(temporal_state_summary_csv, index=False)
    temporal_state_figure = out_dir / "temporal_state_reliability.png"
    plot_temporal_state_reliability(temporal_state_summary, stage_time, temporal_state_figure)
    evidence_report = out_dir / "temporal_state_evidence.md"
    evidence_report.write_text(
        build_evidence_report(
            temporal_state_summary,
            out_dir=out_dir,
            temporal_state_summary_csv=temporal_state_summary_csv,
            figure_path=temporal_state_figure,
            temporal_all_csv=temporal_all_csv,
            emission_all_csv=emission_all_csv,
            stage_time_all_csv=stage_time_all_csv,
            stages_all_csv=stages_all_csv,
        ),
        encoding="utf-8",
    )
    command_log = out_dir / "temporal_state_commands.md"
    _write_command_log(
        command_log,
        command_line=command_line,
        task_outputs=task_outputs,
        decoders=decoders,
        n_permutations=n_permutations,
        stay_grid_size=stay_grid_size,
    )

    exported_artifacts: list[Path] = []
    if compact_export_dir is not None:
        exported_artifacts = export_temporal_state_artifacts(out_dir, compact_export_dir, max_mb=max_export_mb)

    return TemporalStateWorkflowRun(
        out_dir=out_dir,
        task_outputs=task_outputs,
        temporal_state_summary_csv=temporal_state_summary_csv,
        temporal_state_figure=temporal_state_figure,
        evidence_report=evidence_report,
        command_log=command_log,
        exported_artifacts=exported_artifacts,
    )