Digital Drosophila Archive
150 lines
8.2 KiB
Python
150 lines
8.2 KiB
Python
"""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
|