Digital Drosophila Archive
541 lines
31 KiB
Python
541 lines
31 KiB
Python
"""Phase 4AF: a streaming float64 oracle for three fixed short full-graph trials.
|
|
|
|
Importing this module does not import Brian. queue_summary is shared with the
|
|
GPU comparison code. All scientific inputs and the sealed 4AD/4AE code remain
|
|
read-only. Every run_case call belongs in a fresh process.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import hashlib
|
|
import importlib.metadata
|
|
import json
|
|
import os
|
|
import platform
|
|
import struct
|
|
import sys
|
|
import types
|
|
from contextlib import ExitStack
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
|
|
N_NEURONS = 166700
|
|
N_EDGES = 25582938
|
|
BLOCK_BYTES = 8 << 20
|
|
QUEUE_HEADER = b"phase4af-queue-v1\0"
|
|
AUTHOR_PATH = Path("/data/digital-drosophila/experiments/phase4ad-author-semantics/external-resources/model.py")
|
|
AUTHOR_SHA256 = "fc45837d7122c6ce2a7f3f2f23c515992e4b232aadb919efabb72337fac88e4e"
|
|
AE_ORACLE_PATH = Path("/data/digital-drosophila/models/male-cns-lif/author-runtime-lab-v1/oracle_stress.py")
|
|
AE_ORACLE_SHA256 = "32eb1c42383068ce52701d2139a376ce155ad7072cde019949f8ca2d55492409"
|
|
EVENT_DTYPE = np.dtype([("step", "<i4"), ("neuron_idx", "<i4")])
|
|
_PROCESS_ALREADY_USED = False
|
|
|
|
|
|
def queue_summary(tick, edges, accepted, chunk_bytes=BLOCK_BYTES):
|
|
"""Hash original queue order without sorting, filtering, or changing inputs.
|
|
|
|
joint_sha256 is SHA256(header + pack('<qq', tick, count) + all edge bytes
|
|
in little-endian int64 order + all uint8 acceptance bytes). The two other
|
|
hashes omit the header and contain only their respective byte sequences.
|
|
``accepted`` may be a (count,) array or a pure callable(start, stop), which
|
|
lets the CPU read the post-threshold guard in bounded chunks. Empty ticks
|
|
retain their tick/count header in the joint hash.
|
|
"""
|
|
edge_view = np.asarray(edges)
|
|
if edge_view.ndim != 1 or edge_view.dtype.kind not in "iu":
|
|
raise ValueError("Queue edges must be a one-dimensional integer array")
|
|
tick_int = int(tick)
|
|
if tick_int != tick or tick_int < 0:
|
|
raise ValueError("Queue tick must be a nonnegative integer")
|
|
count = len(edge_view)
|
|
width = int(chunk_bytes) // np.dtype("<i8").itemsize
|
|
if width <= 0:
|
|
raise ValueError("chunk_bytes must hold at least one int64 edge")
|
|
mask_view = None if callable(accepted) else np.asarray(accepted)
|
|
if mask_view is not None and mask_view.shape != (count,):
|
|
raise ValueError("Queue acceptance mask must match the edge count")
|
|
edge_hash, mask_hash = hashlib.sha256(), hashlib.sha256()
|
|
joint = hashlib.sha256(QUEUE_HEADER + struct.pack("<qq", tick_int, count))
|
|
for begin in range(0, count, width):
|
|
original = edge_view[begin:begin + width]
|
|
if original.dtype.kind == "u" and np.any(original > np.iinfo(np.int64).max):
|
|
raise ValueError("Queue edge ID cannot be encoded as signed int64")
|
|
chunk = np.ascontiguousarray(original, dtype="<i8")
|
|
data = memoryview(chunk).cast("B")
|
|
edge_hash.update(data)
|
|
joint.update(data)
|
|
accepted_count = 0
|
|
for begin in range(0, count, width):
|
|
stop = min(count, begin + width)
|
|
original = np.asarray(accepted(begin, stop) if callable(accepted) else mask_view[begin:stop])
|
|
if original.shape != (stop - begin,) or original.dtype.kind not in "biu":
|
|
raise ValueError("Acceptance chunks must be one-dimensional boolean/integer masks")
|
|
if np.any((original != 0) & (original != 1)):
|
|
raise ValueError("Acceptance values must be zero or one")
|
|
chunk = np.ascontiguousarray(original, dtype=np.uint8)
|
|
accepted_count += int(chunk.sum(dtype=np.int64))
|
|
data = memoryview(chunk).cast("B")
|
|
mask_hash.update(data)
|
|
joint.update(data)
|
|
return {"tick": tick_int, "count": count, "accepted_count": accepted_count,
|
|
"ordered_edge_sha256": edge_hash.hexdigest(),
|
|
"accepted_mask_sha256": mask_hash.hexdigest(),
|
|
"joint_sha256": joint.hexdigest()}
|
|
|
|
|
|
def _save(path, value):
|
|
with Path(path).open("x", encoding="utf-8") as handle:
|
|
json.dump(value, handle, indent=2, sort_keys=True, allow_nan=False)
|
|
handle.write("\n")
|
|
|
|
|
|
def _digest(path):
|
|
result = hashlib.sha256()
|
|
with Path(path).open("rb") as handle:
|
|
for block in iter(lambda: handle.read(BLOCK_BYTES), b""):
|
|
result.update(block)
|
|
return result.hexdigest()
|
|
|
|
|
|
def _check(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _tree_bytes(root):
|
|
total = 0
|
|
for directory, _, filenames in os.walk(root, followlinks=False):
|
|
for name in filenames:
|
|
total += (Path(directory) / name).stat(follow_symlinks=False).st_size
|
|
return total
|
|
|
|
|
|
def _budget(root, limit, reserve=0):
|
|
actual = _tree_bytes(root)
|
|
if actual + int(reserve) > limit:
|
|
raise RuntimeError(f"New-output budget would be exceeded: {actual} + {reserve} > {limit} bytes; inputs were not adjusted")
|
|
return actual
|
|
|
|
|
|
def _new_memmap(path, dtype, shape):
|
|
if Path(path).exists():
|
|
raise FileExistsError(path)
|
|
return np.lib.format.open_memmap(path, mode="w+", dtype=np.dtype(dtype), shape=shape)
|
|
|
|
|
|
def _load_checked_module(path, expected_hash, name):
|
|
data = Path(path).read_bytes()
|
|
actual = hashlib.sha256(data).hexdigest()
|
|
_check(actual == expected_hash, f"Pinned source hash mismatch: {path}")
|
|
module = types.ModuleType(name)
|
|
module.__file__ = str(path)
|
|
# Execute checked bytes without producing __pycache__ beside a sealed file.
|
|
exec(compile(data, str(path), "exec"), module.__dict__)
|
|
return module
|
|
|
|
|
|
def _graph_inputs(graph_dir):
|
|
definitions = {
|
|
"nodes": (np.int32, (N_NEURONS,)), "edge_id": (np.int64, (N_EDGES,)),
|
|
"src": (np.int32, (N_EDGES,)), "dst": (np.int32, (N_EDGES,)),
|
|
"signed_raw": (np.int64, (N_EDGES,)), "ptr": (np.int64, (N_NEURONS + 1,)),
|
|
}
|
|
arrays = {}
|
|
for name, (dtype, shape) in definitions.items():
|
|
value = np.load(graph_dir / (name + ".npy"), mmap_mode="r", allow_pickle=False)
|
|
_check(value.dtype == np.dtype(dtype) and value.shape == shape, f"Incorrect canonical {name} shape/dtype")
|
|
arrays[name] = value
|
|
_check(np.array_equal(arrays["nodes"], np.arange(N_NEURONS, dtype=np.int32)),
|
|
"This full-graph protocol requires canonical global node indices 0..N-1")
|
|
ptr = arrays["ptr"]
|
|
_check(ptr[0] == 0 and ptr[-1] == N_EDGES and np.all(np.diff(ptr) >= 0), "Invalid canonical CSR pointer")
|
|
source_counts = np.zeros(N_NEURONS, dtype=np.int64)
|
|
previous_source = -1
|
|
width = BLOCK_BYTES // 8
|
|
for begin in range(0, N_EDGES, width):
|
|
stop = min(N_EDGES, begin + width)
|
|
_check(np.array_equal(arrays["edge_id"][begin:stop], np.arange(begin, stop, dtype=np.int64)),
|
|
"Full-graph edge identity/order is not the original canonical CSR order")
|
|
source, target = arrays["src"][begin:stop], arrays["dst"][begin:stop]
|
|
_check(source[0] >= previous_source and np.all(np.diff(source) >= 0), "Sources are not canonically sorted")
|
|
_check(source[0] >= 0 and source[-1] < N_NEURONS and np.all((target >= 0) & (target < N_NEURONS)),
|
|
"Canonical endpoint outside the full graph")
|
|
previous_source = int(source[-1])
|
|
source_counts += np.bincount(source, minlength=N_NEURONS)
|
|
_check(np.array_equal(source_counts, np.diff(ptr)), "src and CSR pointer disagree")
|
|
for name in ("completeness.csv", "connectivity.parquet"):
|
|
_check((graph_dir / name).is_file(), f"Missing author input file {name}")
|
|
return arrays
|
|
|
|
|
|
class _StreamingQueueProxy:
|
|
"""The 4AE forwarding contract with bounded hashing instead of retained lists."""
|
|
|
|
def __init__(self, original, get_tick, get_available, arrays, due_counts,
|
|
spike_ring, delay_steps, queue_file, records):
|
|
self._original_queue = original
|
|
self._get_tick = get_tick
|
|
self._get_available = get_available
|
|
self._arrays = arrays
|
|
self._due_counts = due_counts
|
|
self._spike_ring = spike_ring
|
|
self._delay_steps = delay_steps
|
|
self._queue_file = queue_file
|
|
self._records = records
|
|
|
|
def peek(self, *args, **kwargs):
|
|
original = self._original_queue.peek(*args, **kwargs)
|
|
edge_view = np.asarray(original)
|
|
tick = self._get_tick()
|
|
_check(tick == len(self._records["rows"]), "Queue peek ticks must be unique and consecutive")
|
|
_check(len(edge_view) == int(self._due_counts[tick % len(self._due_counts)]),
|
|
"Queue arrival count differs from delayed outgoing degree of real spikes")
|
|
available = np.asarray(self._get_available(), dtype=bool)
|
|
fired_then = np.zeros(N_NEURONS, dtype=bool)
|
|
if tick >= self._delay_steps:
|
|
fired_then[self._spike_ring[(tick - self._delay_steps) % len(self._spike_ring)]] = True
|
|
|
|
def accepted_chunk(begin, stop):
|
|
local_edges = edge_view[begin:stop]
|
|
_check(np.all((local_edges >= 0) & (local_edges < N_EDGES)), "Queue contains an invalid local edge index")
|
|
_check(np.all(fired_then[self._arrays["src"][local_edges]]),
|
|
"Queue arrival lacks a presynaptic spike at the fixed 18-step delay")
|
|
return available[self._arrays["dst"][local_edges]]
|
|
|
|
# edge_id[k] == k was independently checked for every canonical edge.
|
|
# Thus the untouched local queue array is also the original edge-ID stream.
|
|
summary = queue_summary(tick, edge_view, accepted_chunk)
|
|
self._queue_file.write(json.dumps(summary, sort_keys=True) + "\n")
|
|
self._records["rows"].append(summary)
|
|
self._records["peek_calls"] += 1
|
|
return original
|
|
|
|
def advance(self, *args, **kwargs):
|
|
self._records["advance_calls"] += 1
|
|
return self._original_queue.advance(*args, **kwargs)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._original_queue, name)
|
|
|
|
|
|
def run_case(case_dir: Path, out_dir: Path) -> dict:
|
|
"""Run exactly one preregistered float64 case in this process.
|
|
|
|
Outputs: v/g/ref history .npy memmaps; actual weights.npy; sparse state.npz;
|
|
queue.jsonl; events.bin with two little-endian int32 fields; request records;
|
|
small per-tick statistics and audit JSON. A failed trial retains partial
|
|
outputs and error.json and never writes summary.json with complete=True.
|
|
"""
|
|
global _PROCESS_ALREADY_USED
|
|
_check(not _PROCESS_ALREADY_USED, "Each full-graph oracle case requires a new process")
|
|
_PROCESS_ALREADY_USED = True
|
|
case_dir = Path(case_dir).resolve(strict=True)
|
|
out_dir = Path(out_dir).resolve()
|
|
spec = json.loads((case_dir / "spec.json").read_text(encoding="utf-8"))
|
|
graph_dir = Path(spec["graph_dir"]).resolve(strict=True)
|
|
steps = int(spec["steps"])
|
|
seeds = tuple(int(node) for node in spec["seed_nodes"])
|
|
fixed = {"no_input": (100, ()), "single": (500, (1381,)),
|
|
"five": (500, (1197, 1381, 1511, 82312, 133666))}
|
|
_check(spec["name"] in fixed and (steps, seeds) == fixed[spec["name"]], "Case is not one of the three fixed trials")
|
|
_check(float(spec["dt_ms"]) == .1 and np.isclose(float(spec["duration_ms"]), steps * .1, rtol=0, atol=1e-12),
|
|
"Case clock/duration differs from the fixed protocol")
|
|
_check(spec.get("precision", "float64") == "float64", "Full-graph oracle is float64 only")
|
|
_check(spec.get("all_requests_at_tick", 0) == 0 and not spec.get("apl_overlay", False) and not spec.get("training", False),
|
|
"This oracle does not permit altered input timing, APL, or training")
|
|
budget_root = Path(spec.get("budget_root", os.path.commonpath([str(case_dir), str(graph_dir), str(out_dir)]))).resolve()
|
|
budget_limit = int(spec.get("output_budget_bytes", 15 * 1024 ** 3))
|
|
_check(budget_limit > 0, "Output budget must be positive")
|
|
_check(out_dir.is_relative_to(budget_root), "Oracle output must be inside its budget root")
|
|
out_dir.mkdir(parents=True, exist_ok=False)
|
|
memmaps = []
|
|
last_complete_tick = -1
|
|
saved_preferences = None
|
|
old_dt = None
|
|
try:
|
|
_budget(budget_root, budget_limit, reserve=steps * N_NEURONS * 20 + N_EDGES * 8 + (64 << 20))
|
|
versions = {name: importlib.metadata.version(name)
|
|
for name in ("brian2", "numpy", "pandas", "pyarrow", "sympy", "scipy", "cython")}
|
|
_save(out_dir / "environment.json", {"python": sys.version, "platform": platform.platform(),
|
|
"packages": versions, "precision": "float64", "codegen": "numpy"})
|
|
for name, required in {"brian2": "2.5.1", "numpy": "1.22.3", "sympy": "1.12"}.items():
|
|
_check(versions[name] == required, f"{name} must be pinned to {required}")
|
|
import brian2 as b2
|
|
from brian2.core.base import BrianObject
|
|
|
|
ae = _load_checked_module(AE_ORACLE_PATH, AE_ORACLE_SHA256, "phase4af_readonly_ae_helpers")
|
|
author = _load_checked_module(AUTHOR_PATH, AUTHOR_SHA256, "phase4af_pinned_author")
|
|
arrays = _graph_inputs(graph_dir)
|
|
graph_hashes = {name: _digest(graph_dir / name) for name in (
|
|
"nodes.npy", "edge_id.npy", "src.npy", "dst.npy", "signed_raw.npy", "ptr.npy",
|
|
"completeness.csv", "connectivity.parquet")}
|
|
_save(out_dir / "input-audit.json", {"spec": spec, "spec_sha256": _digest(case_dir / "spec.json"),
|
|
"graph_hashes": graph_hashes, "author_sha256": AUTHOR_SHA256,
|
|
"sealed_ae_oracle_sha256": AE_ORACLE_SHA256,
|
|
"budget_root": str(budget_root), "budget_limit_bytes": budget_limit,
|
|
"source_order": "canonical CSR; local edge index equals original edge_id",
|
|
"common_interface_preserved": True, "poi_used": False, "apl_overlay": False, "training": False})
|
|
saved_preferences = {key: b2.prefs[key] for key in (
|
|
"codegen.target", "core.default_float_dtype", "legacy.refractory_timing")}
|
|
old_dt = b2.defaultclock.dt
|
|
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 = .1 * b2.ms
|
|
parameters = dict(author.default_params)
|
|
_save(out_dir / "parameter-checks.json", ae._parameters_match(parameters, spec["model"], b2))
|
|
neu, syn, spikes = author.create_model(graph_dir / "completeness.csv",
|
|
graph_dir / "connectivity.parquet", parameters)
|
|
_check(len(neu) == N_NEURONS and len(syn) == N_EDGES, "Author model did not load the whole graph")
|
|
_check(all(np.dtype(neu.variables[key].dtype) == np.float64 for key in ("v", "g")) and
|
|
np.dtype(syn.variables["w"].dtype) == np.float64, "Author state/weight dtype is not float64")
|
|
weights = _new_memmap(out_dir / "weights.npy", "<f8", (N_EDGES,))
|
|
memmaps.append(weights)
|
|
width = BLOCK_BYTES // 8
|
|
w_unit = float(parameters["w_syn"] / b2.volt)
|
|
for begin in range(0, N_EDGES, width):
|
|
stop = min(N_EDGES, begin + width)
|
|
np.testing.assert_array_equal(np.asarray(syn.i[begin:stop]), arrays["src"][begin:stop])
|
|
np.testing.assert_array_equal(np.asarray(syn.j[begin:stop]), arrays["dst"][begin:stop])
|
|
actual_w = np.asarray(syn.w_[begin:stop], dtype=np.float64)
|
|
expected_w = arrays["signed_raw"][begin:stop] * w_unit
|
|
_check(np.array_equal(actual_w.view(np.uint64), expected_w.view(np.uint64)),
|
|
"Author weights differ from unchanged signed raw times the original SI w_syn")
|
|
weights[begin:stop] = actual_w
|
|
weights.flush()
|
|
v_history = _new_memmap(out_dir / "v-history.float64.npy", "<f8", (steps, N_NEURONS))
|
|
g_history = _new_memmap(out_dir / "g-history.float64.npy", "<f8", (steps, N_NEURONS))
|
|
ref_history = _new_memmap(out_dir / "ref-history.int32.npy", "<i4", (steps, N_NEURONS))
|
|
memmaps.extend([v_history, g_history, ref_history])
|
|
_save(out_dir / "output-schema.json", {
|
|
"history_shape": [steps, N_NEURONS], "v_file": "v-history.float64.npy",
|
|
"g_file": "g-history.float64.npy", "ref_file": "ref-history.int32.npy",
|
|
"voltage_unit": "SI volt", "v_g_dtype": "little-endian float64", "ref_dtype": "little-endian int32",
|
|
"event_file": "events.bin", "event_record": [["step", "<i4"], ["neuron_idx", "<i4"]],
|
|
"event_order": "tick then canonical neuron_idx", "queue_file": "queue.jsonl",
|
|
"queue_joint_bytes": "b'phase4af-queue-v1\\0' + struct.pack('<qq',tick,count) + all ordered <i8 edge IDs + all uint8 acceptance bits",
|
|
"queue_separate_hashes": "SHA256 of each pure byte sequence without header",
|
|
"complete_marker": "summary.json with complete=true; partial memmaps alone are not a completed trial"})
|
|
_budget(budget_root, budget_limit, reserve=64 << 20)
|
|
delay_steps = int(round(float(parameters["t_dly"] / b2.ms) / .1))
|
|
refractory_steps = int(round(float(parameters["t_rfc"] / b2.ms) / .1))
|
|
_check(delay_steps == 18 and refractory_steps == 22, "Unexpected fixed delay or refractory duration")
|
|
queue_len = delay_steps + 1
|
|
due_counts = np.zeros(queue_len, dtype=np.int64)
|
|
spike_ring = [np.empty(0, dtype=np.int32) for _ in range(queue_len)]
|
|
outdegree = np.diff(arrays["ptr"])
|
|
request_count = np.zeros(steps, dtype=np.int32)
|
|
applicable_count = np.zeros(steps, dtype=np.int32)
|
|
rejected_count = np.zeros(steps, dtype=np.int32)
|
|
spike_count = np.zeros(steps, dtype=np.int32)
|
|
total_counts = np.zeros(N_NEURONS, dtype=np.int32)
|
|
request_seen = np.zeros(steps, dtype=np.uint8)
|
|
seed_local = np.asarray(seeds, dtype=np.int32)
|
|
permitted_seed = np.empty(0, dtype=np.int32)
|
|
request_rows = []
|
|
records = {"rows": [], "peek_calls": 0, "advance_calls": 0, "installations": []}
|
|
total_events = 0
|
|
v_raw = neu.variables["v"].get_value()
|
|
g_raw = neu.variables["g"].get_value()
|
|
last_raw = neu.variables["lastspike"].get_value()
|
|
available_raw = neu.variables["not_refractory"].get_value()
|
|
last_ticks = np.zeros(N_NEURONS, dtype=np.int64)
|
|
|
|
def tick_now():
|
|
return int(round(float(neu.clock.t / b2.second) / .0001))
|
|
|
|
with ExitStack() as stack:
|
|
queue_file = stack.enter_context((out_dir / "queue.jsonl").open("x", encoding="utf-8"))
|
|
event_file = stack.enter_context((out_dir / "events.bin").open("xb"))
|
|
tick_file = stack.enter_context((out_dir / "ticks.jsonl").open("x", encoding="utf-8"))
|
|
requests_file = stack.enter_context((out_dir / "requests.csv").open("x", encoding="utf-8", newline=""))
|
|
request_writer = csv.DictWriter(requests_file,
|
|
fieldnames=["tick", "neuron_idx", "attempted", "accepted", "rejected"])
|
|
request_writer.writeheader()
|
|
|
|
def request_operation():
|
|
nonlocal permitted_seed
|
|
tick = tick_now()
|
|
_check(0 <= tick < steps and not request_seen[tick], "Invalid/duplicate common request tick")
|
|
request_seen[tick] = 1
|
|
if tick == 0 and len(seed_local):
|
|
available = available_raw[seed_local].copy()
|
|
permitted_seed = seed_local[available]
|
|
request_count[tick] = len(seed_local)
|
|
applicable_count[tick] = len(permitted_seed)
|
|
rejected_count[tick] = len(seed_local) - len(permitted_seed)
|
|
for node, ok in zip(seed_local, available):
|
|
request_rows.append({"tick": 0, "neuron_idx": int(node), "attempted": 1,
|
|
"accepted": int(ok), "rejected": int(not ok)})
|
|
neu.v[permitted_seed] = parameters["v_th"] + 1 * b2.mV
|
|
|
|
def record_end():
|
|
nonlocal last_complete_tick, total_events
|
|
tick = tick_now()
|
|
_check(tick == last_complete_tick + 1 and tick < steps, "End-state ticks are not consecutive")
|
|
_check(request_seen[tick] == 1 and len(records["rows"]) == tick + 1,
|
|
"End state lacks its request/queue observation")
|
|
_check(np.isfinite(v_raw).all() and np.isfinite(g_raw).all(), "Author state became nonfinite")
|
|
seen = np.isfinite(last_raw) & (last_raw >= 0)
|
|
last_ticks.fill(0)
|
|
last_ticks[seen] = np.rint(last_raw[seen] / .0001).astype(np.int64)
|
|
ref = np.where(seen, np.maximum(refractory_steps - (tick - last_ticks), 0), 0).astype(np.int32)
|
|
_check(np.array_equal(ref == 0, available_raw), "Integer refractory history disagrees with Brian")
|
|
v_history[tick] = v_raw
|
|
g_history[tick] = g_raw
|
|
ref_history[tick] = ref
|
|
active = np.asarray(neu.spikes, dtype=np.int32)
|
|
_check(not len(active) or np.all(np.diff(active) > 0), "Spike order is not canonical and unique")
|
|
spike_count[tick] = len(active)
|
|
total_counts[active] += 1
|
|
if len(active):
|
|
rows = np.empty(len(active), dtype=EVENT_DTYPE)
|
|
rows["step"] = tick
|
|
rows["neuron_idx"] = arrays["nodes"][active]
|
|
event_file.write(memoryview(rows).cast("B"))
|
|
total_events += len(active)
|
|
spike_ring[tick % queue_len] = active.copy()
|
|
due_counts[(tick + delay_steps) % queue_len] = int(outdegree[active].sum(dtype=np.int64))
|
|
if tick == 0:
|
|
_check(np.isin(permitted_seed, active).all(), "An applicable initial request failed to create a real spike")
|
|
for row in request_rows:
|
|
request_writer.writerow(row)
|
|
queue_row = records["rows"][tick]
|
|
tick_record = {"tick": tick, "spikes": len(active), "requests": int(request_count[tick]),
|
|
"applicable_requests": int(applicable_count[tick]), "ref_rejected_requests": int(rejected_count[tick]),
|
|
"due_arrivals": queue_row["count"], "accepted_arrivals": queue_row["accepted_count"],
|
|
"event_sha256": hashlib.sha256(np.ascontiguousarray(arrays["nodes"][active], dtype="<i4").tobytes()).hexdigest()}
|
|
tick_file.write(json.dumps(tick_record, sort_keys=True) + "\n")
|
|
# Mapped files already reserve their full logical size. This
|
|
# checks the complete new experiment tree, not only this case.
|
|
_budget(budget_root, budget_limit, reserve=1 << 20)
|
|
last_complete_tick = tick
|
|
if (tick + 1) % 50 == 0 or tick + 1 == steps:
|
|
for mapped in (v_history, g_history, ref_history):
|
|
mapped.flush()
|
|
for handle in (queue_file, event_file, tick_file, requests_file):
|
|
handle.flush()
|
|
if (tick + 1) % 100 == 0 or tick + 1 == steps:
|
|
print(json.dumps({"oracle": spec["name"], "completed_ticks": tick + 1,
|
|
"spikes": total_events}), flush=True)
|
|
|
|
class QueueInstaller(BrianObject):
|
|
def __init__(self):
|
|
super().__init__(clock=neu.clock, when="end", order=1_000_000,
|
|
name="zz_fullgraph_queue_installer")
|
|
|
|
def before_run(self, run_namespace):
|
|
installed = []
|
|
for code in syn.pre.code_objects:
|
|
if "_queue" not in code.namespace or "_queue.peek()" not in str(code.code):
|
|
continue
|
|
original = code.namespace["_queue"]
|
|
if isinstance(original, _StreamingQueueProxy):
|
|
original = original._original_queue
|
|
code.namespace["_queue"] = _StreamingQueueProxy(
|
|
original, tick_now, lambda: available_raw, arrays, due_counts,
|
|
spike_ring, delay_steps, queue_file, records)
|
|
installed.append(code.name)
|
|
_check(len(installed) == 1, "Expected one original synaptic queue consumer")
|
|
records["installations"].append(installed[0])
|
|
|
|
request_op = b2.NetworkOperation(request_operation, clock=neu.clock,
|
|
when="before_thresholds", order=0, name="oracle_common_requests")
|
|
end_op = b2.NetworkOperation(record_end, clock=neu.clock,
|
|
when="end", order=0, name="oracle_streaming_end")
|
|
network = b2.Network(neu, syn, spikes, request_op, end_op, QueueInstaller())
|
|
_save(out_dir / "schedule.json", {"schedule": list(network.schedule),
|
|
"summary": str(network.scheduling_summary())})
|
|
network.run(0 * b2.ms, namespace={})
|
|
_save(out_dir / "compiled-code.json", ae._code_records(network))
|
|
_check(last_complete_tick == -1 and not records["rows"] and not request_seen.any(),
|
|
"Zero-duration compilation unexpectedly advanced a trial")
|
|
# Exactly one positive-duration call. Flushing files never resets a
|
|
# clock, reruns before_run, or changes the in-flight delay queue.
|
|
network.run(steps * .1 * b2.ms, namespace={})
|
|
_save(out_dir / "generated-code.json", ae._code_records(network))
|
|
|
|
_check(last_complete_tick == steps - 1 and np.all(request_seen == 1), "Trial stopped before all ticks completed")
|
|
_check(records["peek_calls"] == records["advance_calls"] == steps, "Queue observation was incomplete")
|
|
_check(total_events == int(spikes.num_spikes) and np.array_equal(total_counts, np.asarray(spikes.count[:])),
|
|
"Sparse streamed events disagree with original SpikeMonitor counts")
|
|
_check(np.all(np.asarray(neu.rfc[:] / b2.ms) == float(parameters["t_rfc"] / b2.ms)),
|
|
"Original refractory configuration changed")
|
|
for begin in range(0, N_EDGES, width):
|
|
stop = min(N_EDGES, begin + width)
|
|
_check(np.array_equal(np.asarray(syn.w_[begin:stop]).view(np.uint64),
|
|
np.asarray(weights[begin:stop]).view(np.uint64)),
|
|
"Original synaptic weights changed during execution")
|
|
events = (np.memmap(out_dir / "events.bin", dtype=EVENT_DTYPE, mode="r", shape=(total_events,))
|
|
if total_events else np.empty(0, dtype=EVENT_DTYPE))
|
|
_check((out_dir / "events.bin").stat().st_size == total_events * EVENT_DTYPE.itemsize,
|
|
"Sparse event file length is inconsistent")
|
|
event_block = BLOCK_BYTES // 8
|
|
for begin in range(0, total_events, event_block):
|
|
stop = min(total_events, begin + event_block)
|
|
np.testing.assert_array_equal(events["neuron_idx"][begin:stop],
|
|
np.asarray(spikes.i[begin:stop], dtype=np.int32))
|
|
native_ticks = np.rint(np.asarray(spikes.t[begin:stop] / b2.second) / .0001).astype(np.int32)
|
|
np.testing.assert_array_equal(events["step"][begin:stop], native_ticks)
|
|
_budget(budget_root, budget_limit, reserve=total_events * 8 + N_NEURONS * 24 + (8 << 20))
|
|
with (out_dir / "state.npz").open("xb") as handle:
|
|
np.savez_compressed(handle, step=events["step"], neuron_idx=events["neuron_idx"], counts=total_counts,
|
|
v=np.asarray(v_history[-1]), g=np.asarray(g_history[-1]), ref=np.asarray(ref_history[-1]))
|
|
with (out_dir / "tick-summary.npz").open("xb") as handle:
|
|
np.savez_compressed(handle, request_count=request_count, applicable_count=applicable_count,
|
|
rejected_count=rejected_count, spike_count=spike_count,
|
|
arrival_count=np.asarray([row["count"] for row in records["rows"]], dtype=np.int64),
|
|
accepted_arrival_count=np.asarray([row["accepted_count"] for row in records["rows"]], dtype=np.int64))
|
|
coefficients = ae._coefficients(neu, parameters, b2, .0001)
|
|
_save(out_dir / "coefficients.json", coefficients)
|
|
_save(out_dir / "queue-audit.json", {"peek_calls": records["peek_calls"],
|
|
"advance_calls": records["advance_calls"], "installations": records["installations"],
|
|
"due_count_matches_presynaptic_spikes_and_original_outdegrees": True,
|
|
"acceptance_observed_after_thresholds_before_synaptic_writes": True,
|
|
"same_peek_object_returned": True, "all_other_queue_methods_forwarded": True,
|
|
"original_order_not_sorted_or_filtered": True, "hash_header_hex": QUEUE_HEADER.hex(),
|
|
"edge_id_identity_verified_for_all_edges": True,
|
|
"fullgraph_unwrapped_repeat_performed": False,
|
|
"prior_evidence": "Phase4AE exact wrapped/unwrapped checks; current bounded-hash adapter preserves the same forwarding contract",
|
|
"limit": "Current full-graph streaming proxy is audited by forwarding logic and per-tick counts; no separate unwrapped full-graph run is claimed."})
|
|
_check(_digest(AUTHOR_PATH) == AUTHOR_SHA256 and _digest(AE_ORACLE_PATH) == AE_ORACLE_SHA256,
|
|
"A sealed author/dependency source changed during execution")
|
|
for name, expected in graph_hashes.items():
|
|
_check(_digest(graph_dir / name) == expected, f"Prepared graph input changed during execution: {name}")
|
|
for mapped in memmaps:
|
|
mapped.flush()
|
|
current_bytes = _budget(budget_root, budget_limit, reserve=1 << 20)
|
|
summary = {"name": spec["name"], "complete": True, "backend": "Brian2 2.5.1 numpy",
|
|
"precision": "float64", "nodes": N_NEURONS, "edges": N_EDGES,
|
|
"steps": steps, "completed_ticks": last_complete_tick + 1, "spikes": total_events,
|
|
"steps_expected": steps, "steps_completed": last_complete_tick + 1, "neurons": N_NEURONS,
|
|
"pass_all": True, "pass_all_scope": "CPU input, streaming, and internal self-checks only; GPU equivalence is not assessed here",
|
|
"requests": int(request_count.sum()), "applicable_requests": int(applicable_count.sum()),
|
|
"ref_rejected_requests": int(rejected_count.sum()),
|
|
"due_arrivals": sum(row["count"] for row in records["rows"]),
|
|
"accepted_arrivals": sum(row["accepted_count"] for row in records["rows"]),
|
|
"single_continuous_run": True, "source_and_graph_inputs_unchanged": True,
|
|
"output_budget_root": str(budget_root), "output_bytes_before_summary": current_bytes,
|
|
"output_budget_bytes": budget_limit, "weights_sha256": _digest(out_dir / "weights.npy"),
|
|
"queue_sha256": _digest(out_dir / "queue.jsonl"), "events_sha256": _digest(out_dir / "events.bin"),
|
|
"no_training": True, "no_apl_overlay": True, "no_poi": True,
|
|
"scope": "Only this fixed short numerical trajectory; not olfactory CAL, long-run equivalence, physiological validation, or learning.",
|
|
"pending_arrivals_limit": "Queue arrivals beyond the final recorded tick are outside this comparison window."}
|
|
_save(out_dir / "summary.json", summary)
|
|
return summary
|
|
except Exception as error:
|
|
for mapped in memmaps:
|
|
mapped.flush()
|
|
_save(out_dir / "error.json", {"complete": False, "last_complete_tick": last_complete_tick,
|
|
"type": type(error).__name__, "message": str(error),
|
|
"partial_files_preserved": True, "input_or_duration_adjusted": False})
|
|
raise
|
|
finally:
|
|
if saved_preferences is not None:
|
|
for key, value in saved_preferences.items():
|
|
b2.prefs[key] = value
|
|
b2.defaultclock.dt = old_dt
|