Validate isolated author LIF semantics without changing frozen models
This commit is contained in:
Digital Drosophila Archive
@@ -0,0 +1,149 @@
|
||||
"""Compile-only audit of the pinned author's sensory Poisson interface.
|
||||
|
||||
This is separate from the common single-seed diagnostic. Zero simulation time
|
||||
is advanced, so this audit does not draw or validate any Poisson realization.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import types
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from context import CFG, EXP, digest, save
|
||||
from oracle_runner import AUTHOR_COMMIT, AUTHOR_SHA256, _code_records, _environment
|
||||
|
||||
|
||||
def audit_author_poi() -> dict:
|
||||
"""Create the original model and poi helper, compile at t=0, and save evidence."""
|
||||
author_path = EXP / "external-resources" / "model.py"
|
||||
author_bytes = author_path.read_bytes()
|
||||
source_hash = hashlib.sha256(author_bytes).hexdigest()
|
||||
if source_hash != AUTHOR_SHA256:
|
||||
raise ValueError("The Poisson audit requires the exact pinned author source")
|
||||
output = EXP / "oracle" / "poi-interface"
|
||||
output.mkdir(parents=True, exist_ok=False)
|
||||
environment = _environment()
|
||||
save(output / "environment.json", environment)
|
||||
if environment["packages"]["brian2"]["actual"] != "2.5.1":
|
||||
raise RuntimeError("The Poisson interface audit requires Brian2 2.5.1")
|
||||
|
||||
import brian2 as b2
|
||||
|
||||
preferences = {name: b2.prefs[name] for name in (
|
||||
"codegen.target", "core.default_float_dtype", "legacy.refractory_timing")}
|
||||
old_dt = b2.defaultclock.dt
|
||||
try:
|
||||
b2.start_scope()
|
||||
b2.prefs["codegen.target"] = "numpy"
|
||||
b2.prefs["core.default_float_dtype"] = np.float64
|
||||
b2.prefs["legacy.refractory_timing"] = False
|
||||
b2.defaultclock.dt = float(CFG["dt_ms"]) * b2.ms
|
||||
author = types.ModuleType("phase4ad_poi_author_" + source_hash[:12])
|
||||
author.__file__ = str(author_path)
|
||||
# Execute exactly the reviewed bytes; create no bytecode beside the source.
|
||||
exec(compile(author_bytes, str(author_path), "exec"), author.__dict__)
|
||||
parameters = dict(author.default_params)
|
||||
|
||||
nodes = np.array([-2, -1], dtype=np.int32)
|
||||
completeness = pd.DataFrame(
|
||||
{"local_index": np.array([0, 1], dtype=np.int32)},
|
||||
index=pd.Index(nodes, name="synthetic_global_neuron_idx"))
|
||||
connections = pd.DataFrame({
|
||||
"Presynaptic_Index": np.array([0], dtype=np.int32),
|
||||
"Postsynaptic_Index": np.array([1], dtype=np.int32),
|
||||
"Excitatory x Connectivity": np.array([1], dtype=np.int64),
|
||||
"raw_count": np.array([1], dtype=np.int64),
|
||||
"edge_id": np.array([-1], dtype=np.int64),
|
||||
"source_neuron_idx": nodes[:1], "target_neuron_idx": nodes[1:],
|
||||
})
|
||||
comp_path = output / "completeness.csv"
|
||||
con_path = output / "connectivity.parquet"
|
||||
completeness.to_csv(comp_path)
|
||||
connections.to_parquet(con_path, index=False, engine="pyarrow")
|
||||
neu, syn, spikes = author.create_model(comp_path, con_path, parameters)
|
||||
rfc_before = np.asarray(neu.rfc[:] / b2.ms, dtype=np.float64).copy()
|
||||
# One excited target plus one untouched cell makes the helper's scope
|
||||
# observable. The optional second-rate class is not exercised here.
|
||||
inputs, same_neu = author.poi(neu, [0], [], parameters)
|
||||
if same_neu is not neu:
|
||||
raise AssertionError("The author poi helper returned a different neuron group")
|
||||
rfc_after = np.asarray(neu.rfc[:] / b2.ms, dtype=np.float64).copy()
|
||||
before_v = np.asarray(neu.v_[:], dtype=np.float64).copy()
|
||||
before_g = np.asarray(neu.g_[:], dtype=np.float64).copy()
|
||||
before_w = np.asarray(syn.w_[:], dtype=np.float64).copy()
|
||||
network = b2.Network(neu, syn, spikes, *inputs)
|
||||
save(output / "schedule.json", {
|
||||
"schedule": list(network.schedule),
|
||||
"summary": str(network.scheduling_summary()),
|
||||
})
|
||||
network.run(0 * b2.ms, namespace={})
|
||||
save(output / "generated-code.json", _code_records(network))
|
||||
|
||||
input_records = []
|
||||
for item in inputs:
|
||||
# PoissonInput 2.5.1 retains its on-the-fly subgroup as _group;
|
||||
# N/rate/weight/target_var are its public read-only properties.
|
||||
target = item._group
|
||||
input_records.append({
|
||||
"name": item.name, "type": type(item).__name__,
|
||||
"target_group": target.name,
|
||||
"target_local_start": int(target.start),
|
||||
"target_local_stop": int(target.stop),
|
||||
"target_global_nodes": nodes[int(target.start):int(target.stop)].tolist(),
|
||||
"target_var": item.target_var, "N": int(item.N),
|
||||
"rate_hz": float(item.rate / b2.Hz),
|
||||
"weight_mv": float(item.weight / b2.mV),
|
||||
"weight_volt": float(item.weight / b2.volt),
|
||||
"when": item.when, "order": int(item.order),
|
||||
"dt_ms": float(item.clock.dt / b2.ms),
|
||||
"abstract_code": str(item.abstract_code),
|
||||
"generated_code": [str(code.code) for code in item.code_objects],
|
||||
})
|
||||
expected_rfc = float(parameters["t_rfc"] / b2.ms)
|
||||
expected_weight = float(parameters["w_syn"] * parameters["f_poi"] / b2.mV)
|
||||
expected_rate = float(parameters["r_poi"] / b2.Hz)
|
||||
checks = {
|
||||
"exact_author_source": digest(author_path) == source_hash,
|
||||
"one_poisson_input": len(input_records) == 1,
|
||||
"initial_refractory_is_original": bool(np.array_equal(rfc_before, [expected_rfc, expected_rfc])),
|
||||
"only_excited_target_loses_refractory": bool(np.array_equal(rfc_after, [0., expected_rfc])),
|
||||
"actual_voltage_interface": all(r["target_var"] == "v" for r in input_records),
|
||||
"actual_N_one": all(r["N"] == 1 for r in input_records),
|
||||
"actual_rate_150_hz": bool(expected_rate == 150. and all(r["rate_hz"] == expected_rate for r in input_records)),
|
||||
"actual_weight_68_75_mv": bool(np.isclose(expected_weight, 68.75, rtol=0, atol=1e-12)
|
||||
and all(np.isclose(r["weight_mv"], expected_weight, rtol=0, atol=1e-12) for r in input_records)),
|
||||
"actual_synapses_slot_order_zero": all(r["when"] == "synapses" and r["order"] == 0 for r in input_records),
|
||||
"zero_elapsed_time": float(network.t / b2.ms) == 0.,
|
||||
"no_recorded_spikes": len(spikes.i[:]) == 0,
|
||||
"compile_preserves_voltage": bool(np.array_equal(before_v, np.asarray(neu.v_[:]))),
|
||||
"compile_preserves_g": bool(np.array_equal(before_g, np.asarray(neu.g_[:]))),
|
||||
"compile_preserves_weights": bool(np.array_equal(before_w, np.asarray(syn.w_[:]))),
|
||||
"compile_preserves_refractory_configuration": bool(np.array_equal(rfc_after, np.asarray(neu.rfc[:] / b2.ms))),
|
||||
}
|
||||
result = {
|
||||
"gate": bool(all(checks.values())), "checks": checks,
|
||||
"author_commit": AUTHOR_COMMIT, "author_sha256": source_hash,
|
||||
"fixture": "two synthetic nodes, one raw_count=1 connection",
|
||||
"nodes": nodes.tolist(), "exc_local_indices": [0], "exc2_local_indices": [],
|
||||
"rfc_before_ms": rfc_before.tolist(), "rfc_after_ms": rfc_after.tolist(),
|
||||
"poisson_inputs": input_records,
|
||||
"synaptic_weight_mv": np.asarray(syn.w[:] / b2.mV, dtype=np.float64).tolist(),
|
||||
"input_hashes": {"completeness.csv": digest(comp_path), "connectivity.parquet": digest(con_path)},
|
||||
"simulation_duration_ms": 0., "stochastic_time_steps": 0,
|
||||
"random_realization_tested": False, "common_seed_experiment": False,
|
||||
"interpretation": "Original poi adds 68.75 mV to voltage at 150 Hz with N=1 in the synapses slot and sets its target rfc to zero. This interface is not the common initial seed experiment.",
|
||||
"scope": "Compile-time interface audit only; no stochastic trial, learning, or whole-brain simulation. The optional exc2 helper branch is not exercised.",
|
||||
}
|
||||
save(output / "audit.json", result)
|
||||
if not result["gate"]:
|
||||
raise AssertionError("Author Poisson interface gate failed: " + str(checks))
|
||||
return result
|
||||
except Exception as error:
|
||||
save(output / "error.json", {"type": type(error).__name__, "message": str(error)})
|
||||
raise
|
||||
finally:
|
||||
for name, value in preferences.items():
|
||||
b2.prefs[name] = value
|
||||
b2.defaultclock.dt = old_dt
|
||||
Reference in New Issue
Block a user