Skip to content

Emission Compare

neureptrace.emission_compare

build_emission_comparison_report(comparison, *, summary_csv)

Build a compact Markdown report for calibrated-vs-uncalibrated emissions.

Source code in src/neureptrace/emission_compare.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def build_emission_comparison_report(comparison: pd.DataFrame, *, summary_csv: Path) -> str:
    """Build a compact Markdown report for calibrated-vs-uncalibrated emissions."""
    lines = [
        "# NeuRepTrace Emission Calibration Comparison",
        "",
        f"- Temporal-model summary: `{summary_csv}`",
        "",
        "Question: do calibrated probabilities produce cleaner state inference than",
        "uncalibrated score-derived emissions?",
        "",
        "The main comparison is the control margin: observed effect-window",
        "persistence gain minus the strongest baseline, shuffled-time, or",
        "shuffled-label control gain. Positive deltas favor calibrated emissions.",
        "",
    ]
    if comparison.empty:
        lines.append("No decoder had both calibrated and uncalibrated emission-mode rows.")
        lines.append("")
        return "\n".join(lines)

    lines.extend(
        [
            "| Decoder | Preferred | Delta control margin | Calibrated margin | Uncalibrated margin | Delta effect-baseline | Calibrated p(time) | Uncalibrated p(time) |",
            "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: |",
        ]
    )
    for row in comparison.itertuples(index=False):
        lines.append(
            f"| {row.decoder} | {row.preferred_emission_mode} | {_format_float(row.delta_control_margin)} | "
            f"{_format_float(row.calibrated_control_margin)} | {_format_float(row.uncalibrated_control_margin)} | "
            f"{_format_float(row.delta_effect_minus_baseline_gain)} | {_format_float(row.calibrated_shuffled_time_p)} | "
            f"{_format_float(row.uncalibrated_shuffled_time_p)} |"
        )
    lines.append("")
    return "\n".join(lines)

compare_emission_modes(summary)

Compare calibrated and uncalibrated temporal-model evidence by decoder.

Source code in src/neureptrace/emission_compare.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def compare_emission_modes(summary: pd.DataFrame) -> pd.DataFrame:
    """Compare calibrated and uncalibrated temporal-model evidence by decoder."""
    required = {"decoder", "emission_mode", "condition", "persistence_gain_per_observation"}
    missing = sorted(required.difference(summary.columns))
    if missing:
        raise ValueError(f"Temporal-model summary is missing required columns: {missing}")

    rows = []
    for decoder, decoder_frame in summary.groupby("decoder", sort=True):
        modes = {mode: frame for mode, frame in decoder_frame.groupby("emission_mode", sort=True)}
        if "calibrated" not in modes or "uncalibrated" not in modes:
            continue
        calibrated = summarize_emission_mode(modes["calibrated"])
        uncalibrated = summarize_emission_mode(modes["uncalibrated"])
        delta_control_margin = calibrated["control_margin"] - uncalibrated["control_margin"]
        delta_effect_minus_baseline = calibrated["effect_minus_baseline_gain"] - uncalibrated["effect_minus_baseline_gain"]
        delta_observed_gain = calibrated["observed_gain"] - uncalibrated["observed_gain"]
        preferred = "calibrated" if delta_control_margin >= 0 else "uncalibrated"
        rows.append(
            {
                "decoder": decoder,
                "calibrated_observed_gain": calibrated["observed_gain"],
                "uncalibrated_observed_gain": uncalibrated["observed_gain"],
                "delta_observed_gain": delta_observed_gain,
                "calibrated_control_margin": calibrated["control_margin"],
                "uncalibrated_control_margin": uncalibrated["control_margin"],
                "delta_control_margin": delta_control_margin,
                "calibrated_effect_minus_baseline_gain": calibrated["effect_minus_baseline_gain"],
                "uncalibrated_effect_minus_baseline_gain": uncalibrated["effect_minus_baseline_gain"],
                "delta_effect_minus_baseline_gain": delta_effect_minus_baseline,
                "calibrated_shuffled_time_p": calibrated["shuffled_time_p"],
                "uncalibrated_shuffled_time_p": uncalibrated["shuffled_time_p"],
                "calibrated_shuffled_label_p": calibrated["shuffled_label_p"],
                "uncalibrated_shuffled_label_p": uncalibrated["shuffled_label_p"],
                "calibrated_best_stay_probability": calibrated["best_stay_probability"],
                "uncalibrated_best_stay_probability": uncalibrated["best_stay_probability"],
                "preferred_emission_mode": preferred,
            }
        )
    return pd.DataFrame(rows).sort_values("delta_control_margin", ascending=False).reset_index(drop=True)

compare_temporal_summary(summary_csv, *, out_csv=None, out_report=None)

Compare calibrated and uncalibrated emission rows from a temporal-model summary CSV.

Source code in src/neureptrace/emission_compare.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def compare_temporal_summary(
    summary_csv: Path,
    *,
    out_csv: Path | None = None,
    out_report: Path | None = None,
) -> tuple[pd.DataFrame, str | None]:
    """Compare calibrated and uncalibrated emission rows from a temporal-model summary CSV."""
    comparison = compare_emission_modes(pd.read_csv(summary_csv))
    if out_csv is not None:
        out_csv.parent.mkdir(parents=True, exist_ok=True)
        comparison.to_csv(out_csv, index=False)
    report = None
    if out_report is not None:
        report = build_emission_comparison_report(comparison, summary_csv=summary_csv)
        out_report.parent.mkdir(parents=True, exist_ok=True)
        out_report.write_text(report, encoding="utf-8")
    return comparison, report

summarize_emission_mode(frame)

Summarize temporal-model evidence for one emission mode.

Source code in src/neureptrace/emission_compare.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def summarize_emission_mode(frame: pd.DataFrame) -> dict[str, float]:
    """Summarize temporal-model evidence for one emission mode."""
    observed_gain = _condition_value(frame, "observed_effect", "persistence_gain_per_observation")
    baseline_gain = _condition_value(frame, "baseline_window", "persistence_gain_per_observation")
    shuffled_time_gain = _condition_value(frame, "shuffled_time", "persistence_gain_per_observation")
    shuffled_label_gain = _condition_value(frame, "shuffled_label", "persistence_gain_per_observation")
    return {
        "observed_gain": observed_gain,
        "baseline_gain": baseline_gain,
        "effect_minus_baseline_gain": observed_gain - baseline_gain,
        "shuffled_time_gain": shuffled_time_gain,
        "shuffled_label_gain": shuffled_label_gain,
        "effect_minus_shuffled_time_gain": observed_gain - shuffled_time_gain,
        "effect_minus_shuffled_label_gain": observed_gain - shuffled_label_gain,
        "control_margin": _control_margin(frame),
        "shuffled_time_p": _condition_value(frame, "shuffled_time", "empirical_p_value"),
        "shuffled_label_p": _condition_value(frame, "shuffled_label", "empirical_p_value"),
        "best_stay_probability": _condition_value(frame, "observed_effect", "best_stay_probability"),
    }