Digital Drosophila Archive
260 lines
13 KiB
Python
260 lines
13 KiB
Python
"""Synthetic eligibility tests only; never load recorded brain trajectories."""
|
|
import ast
|
|
import importlib.util
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import tempfile
|
|
import unittest
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
import eligibility_bridge as bridge
|
|
|
|
|
|
def source_path():
|
|
explicit = os.environ.get("PHASE4J_SOURCE")
|
|
if explicit:
|
|
return Path(explicit)
|
|
parent = Path(__file__).resolve().parents[1]
|
|
paths = (parent / "models/male-cns-lif/phase4j.py", parent / "mb-physiology-v1/phase4j.py")
|
|
for path in paths:
|
|
if path.is_file():
|
|
return path
|
|
raise FileNotFoundError("Set PHASE4J_SOURCE to the SHA-pinned original phase4j.py")
|
|
|
|
|
|
SOURCE = source_path()
|
|
KC = np.array([1, 3, 7], np.int32)
|
|
|
|
|
|
def run_trace(rows=(), steps=8, commands=(0, 3, 7), backend="numpy"):
|
|
events = np.asarray(rows, np.int32).reshape(-1, 2)
|
|
return bridge.trace_from_events(events[:, 0], events[:, 1], KC, steps, commands,
|
|
phase4j_source=SOURCE, backend=backend, n_neurons=10)
|
|
|
|
|
|
def update_args():
|
|
return dict(trace=np.array([1, .25, 0], np.float32), kc=KC.copy(),
|
|
candidate_sources=np.array([1, 1, 3, 7, 3], np.int32),
|
|
candidate_compartments=np.array(["y5B'2a", "a1", "y5B'2a", "y5B'2a", "y2a'1"]),
|
|
compartment="y5B'2a", delta=np.zeros(5, np.float32),
|
|
base_candidate=np.array([.275, .55, .0001, .275, 2.75], np.float32),
|
|
d=1., phase4j_source=SOURCE, n_neurons=10)
|
|
|
|
|
|
class FrozenTraceTests(unittest.TestCase):
|
|
def test_no_input_positive_zero_and_no_command(self):
|
|
result = run_trace(commands=())
|
|
self.assertEqual(result.snapshots.shape, (0, 3))
|
|
self.assertTrue(bridge.bit_equal(result.final, np.zeros(3, np.float32)))
|
|
self.assertTrue(bridge.bit_equal(result.last_command_trace, result.final))
|
|
self.assertFalse(result.audit["training_executed"])
|
|
self.assertFalse(result.audit["torch_numpy_bitwise_comparison_performed"])
|
|
json.dumps(result.audit, allow_nan=False)
|
|
|
|
def test_simultaneous_spike_is_seen_and_cap_is_one(self):
|
|
result = run_trace([(0, 1), (1, 1), (3, 3), (3, 7)])
|
|
self.assertEqual(result.snapshots[0, 0], np.float32(1))
|
|
self.assertEqual(result.snapshots[1, 1:].tolist(), [1., 1.])
|
|
self.assertLessEqual(float(result.final.max()), 1.)
|
|
|
|
def test_interval_decay_is_each_tick_not_power_jump(self):
|
|
result = run_trace([(0, 1)], steps=1001, commands=(0, 1, 1000))
|
|
expected = np.float32(1)
|
|
decay = np.float32(math.exp(-.1 / 100.))
|
|
for _ in range(1000):
|
|
expected = np.float32(expected * decay)
|
|
self.assertEqual(result.snapshots[-1, 0].view(np.uint32), expected.view(np.uint32))
|
|
self.assertEqual(result.snapshots[1, 0].view(np.uint32), decay.view(np.uint32))
|
|
|
|
def test_us_before_future_cs_has_zero_delta(self):
|
|
result = run_trace([(2, 1)], commands=(1,))
|
|
args = update_args()
|
|
args["trace"] = result.last_command_trace
|
|
delta, _ = bridge.apply_eligibility(**args)
|
|
self.assertTrue(bridge.bit_equal(delta, args["delta"]))
|
|
|
|
def test_plus_1000_ms_ratio_strict_below_point01(self):
|
|
result = run_trace([(0, 1)], steps=10001, commands=(0, 10000))
|
|
args = update_args()
|
|
args["base_candidate"][:] = 100 # no saturation in this temporal check
|
|
args["trace"] = result.snapshots[0]
|
|
zero, _ = bridge.apply_eligibility(**args)
|
|
args["trace"] = result.snapshots[1]
|
|
late, _ = bridge.apply_eligibility(**args)
|
|
self.assertLess(float(np.abs(late).sum() / np.abs(zero).sum()), .01)
|
|
self.assertGreater(float(result.snapshots[-1, 0]), 0.)
|
|
|
|
def test_non_kc_events_do_not_enter_trace(self):
|
|
a = run_trace([(0, 1), (3, 3)])
|
|
b = run_trace([(0, 0), (0, 1), (2, 9), (3, 3)])
|
|
self.assertTrue(bridge.bit_equal(a.snapshots, b.snapshots))
|
|
self.assertEqual(b.audit["actual_kc_events"], 2)
|
|
|
|
def test_inputs_and_outputs_are_not_mutated(self):
|
|
step, node, kc = np.array([0, 7], np.int32), np.array([1, 3], np.int32), KC.copy()
|
|
before = [a.copy() for a in (step, node, kc)]
|
|
result = bridge.trace_from_events(step, node, kc, 8, [7], phase4j_source=SOURCE, n_neurons=10)
|
|
for a, b in zip((step, node, kc), before):
|
|
self.assertTrue(bridge.bit_equal(a, b))
|
|
for a in (result.final, result.snapshots, result.command_steps, result.last_command_trace):
|
|
self.assertFalse(a.flags.writeable)
|
|
|
|
def test_bad_event_and_command_schema_is_rejected(self):
|
|
for rows in ([(0, 1), (0, 1)], [(1, 1), (0, 1)], [(-1, 1)], [(8, 1)], [(0, 10)]):
|
|
with self.subTest(rows=rows), self.assertRaises(ValueError):
|
|
run_trace(rows)
|
|
for commands in ((1, 1), (3, 1), (-1,), (8,), (.5,), (True,), np.empty((0, 2), np.int32)):
|
|
with self.subTest(commands=commands), self.assertRaises(ValueError):
|
|
run_trace(commands=commands)
|
|
with self.assertRaises(ValueError):
|
|
bridge.trace_from_events(np.array([], np.int64), np.array([], np.int32), KC, 8, [], phase4j_source=SOURCE)
|
|
|
|
def test_bad_kc_and_backend_are_rejected(self):
|
|
for kc in (np.array([3, 1], np.int32), np.array([1, 1], np.int32), np.array([], np.int32)):
|
|
with self.assertRaises(ValueError):
|
|
bridge.trace_from_events(np.array([], np.int32), np.array([], np.int32), kc, 1, [], phase4j_source=SOURCE)
|
|
with self.assertRaises(ValueError):
|
|
run_trace(backend="auto")
|
|
|
|
@unittest.skipUnless(importlib.util.find_spec("torch") is not None, "Torch absent: CPU bitwise compatibility NOT established locally")
|
|
def test_torch_original_expression_bitwise_against_numpy(self):
|
|
rng = np.random.default_rng(719)
|
|
mask = rng.random((1001, 10)) < .07
|
|
tick, node = np.nonzero(mask)
|
|
rows = np.stack((tick, node), axis=1).tolist()
|
|
numpy_result = run_trace(rows, steps=1001, commands=(0, 1, 7, 999, 1000))
|
|
torch_result = run_trace(rows, steps=1001, commands=(0, 1, 7, 999, 1000), backend="torch")
|
|
self.assertTrue(bridge.bit_equal(numpy_result.snapshots, torch_result.snapshots))
|
|
self.assertTrue(bridge.bit_equal(numpy_result.final, torch_result.final))
|
|
empty_np = run_trace([(0, 1)], steps=10001, commands=(0, 10000))
|
|
empty_torch = run_trace([(0, 1)], steps=10001, commands=(0, 10000), backend="torch")
|
|
self.assertTrue(bridge.bit_equal(empty_np.snapshots, empty_torch.snapshots))
|
|
|
|
|
|
class FrozenUpdateTests(unittest.TestCase):
|
|
def test_actual_source_sha_and_ast_only_extraction(self):
|
|
_, report = bridge.apply_eligibility(**update_args())
|
|
self.assertEqual(report["source_sha256"], bridge.PHASE4J_SHA256)
|
|
self.assertFalse(report["actual_dan_activity_used_to_gate"])
|
|
self.assertFalse(report["learning_pass_claimed"])
|
|
json.dumps(report, allow_nan=False)
|
|
|
|
def test_source_drift_is_rejected_before_execution(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "phase4j.py"
|
|
path.write_bytes(SOURCE.read_bytes() + b"\nraise RuntimeError('must not execute')\n")
|
|
args = update_args()
|
|
args["phase4j_source"] = path
|
|
with self.assertRaisesRegex(ValueError, "SHA256 mismatch"):
|
|
bridge.apply_eligibility(**args)
|
|
|
|
def test_bitwise_against_independently_extracted_original_apply_v2(self):
|
|
tree = ast.parse(SOURCE.read_text(encoding="utf-8"))
|
|
function = next(x for x in tree.body if isinstance(x, ast.FunctionDef) and x.name == "apply_v2")
|
|
namespace = {"np": np, "Validator": object, "Any": Any}
|
|
exec(compile(ast.Module(body=[function], type_ignores=[]), "independent-old-apply-v2", "exec"), namespace)
|
|
|
|
class Column:
|
|
def __init__(self, values):
|
|
self.values = values
|
|
|
|
def to_numpy(self, dtype=None):
|
|
return np.array(self.values, dtype=dtype)
|
|
|
|
def astype(self, dtype):
|
|
return Column(np.asarray(self.values).astype(dtype))
|
|
|
|
args = update_args()
|
|
validator = SimpleNamespace(n=10, kc=KC, base_candidate=args["base_candidate"],
|
|
cand={"source_neuron_idx": Column(args["candidate_sources"]),
|
|
"target_compartment_verbatim": Column(args["candidate_compartments"])})
|
|
cfg = {"eligibility_v2": {"eta_effective_mv": .00275, "max_fractional_depression": .25}}
|
|
for _ in range(50):
|
|
actual, report = bridge.apply_eligibility(**args)
|
|
expected, old_report = namespace["apply_v2"](validator, args["trace"], args["compartment"], 1., args["delta"], cfg)
|
|
self.assertTrue(bridge.bit_equal(actual, expected))
|
|
self.assertEqual(report["original_report"], old_report)
|
|
args["delta"] = actual
|
|
|
|
def test_zero_d_and_frozen_preserve_existing_overlay(self):
|
|
args = update_args()
|
|
args["delta"][:] = -.25 * args["base_candidate"]
|
|
for d, frozen in ((0., False), (1., True), (0., True)):
|
|
args.update(d=d, frozen=frozen)
|
|
result, report = bridge.apply_eligibility(**args)
|
|
self.assertTrue(bridge.bit_equal(result, args["delta"]))
|
|
self.assertEqual(report["actual_changed_edges"], 0)
|
|
|
|
def test_off_compartment_and_inactive_source_exact_unchanged(self):
|
|
args = update_args()
|
|
args["delta"][1] = -.01
|
|
result, _ = bridge.apply_eligibility(**args)
|
|
self.assertTrue(bridge.bit_equal(result[[1, 3, 4]], args["delta"][[1, 3, 4]]))
|
|
self.assertEqual(result[0], np.float32(-.00275))
|
|
|
|
def test_original_floor_clamps_and_does_not_reverse_sign(self):
|
|
args = update_args()
|
|
result, report = bridge.apply_eligibility(**args)
|
|
self.assertEqual(result[2], (-.25 * args["base_candidate"])[2])
|
|
args["delta"] = result
|
|
again, _ = bridge.apply_eligibility(**args)
|
|
self.assertEqual(again[2].view(np.uint32), result[2].view(np.uint32))
|
|
self.assertTrue(np.all(again <= 0))
|
|
# Original updated_edges counts attempted updates, including saturation.
|
|
self.assertEqual(report["original_report"]["updated_edges"], 2)
|
|
|
|
def test_all_original_compartments_are_literal_not_inferred(self):
|
|
args = update_args()
|
|
args["candidate_sources"] = np.full(4, 1, np.int32)
|
|
args["candidate_compartments"] = np.array(bridge.COMPARTMENTS)
|
|
args["delta"] = np.zeros(4, np.float32)
|
|
args["base_candidate"] = np.ones(4, np.float32)
|
|
for comp in bridge.COMPARTMENTS:
|
|
args["compartment"] = comp
|
|
result, _ = bridge.apply_eligibility(**args)
|
|
self.assertEqual(int(np.count_nonzero(result)), 1)
|
|
self.assertNotEqual(result[list(bridge.COMPARTMENTS).index(comp)], 0)
|
|
|
|
def test_update_does_not_mutate_inputs_and_returns_readonly(self):
|
|
args = update_args()
|
|
before = {k: v.copy() for k, v in args.items() if isinstance(v, np.ndarray)}
|
|
result, _ = bridge.apply_eligibility(**args)
|
|
for k, value in before.items():
|
|
self.assertTrue(bridge.bit_equal(args[k], value))
|
|
self.assertFalse(result.flags.writeable)
|
|
|
|
def test_invalid_units_dtypes_values_and_compartment_fail_closed(self):
|
|
bad = [("trace", np.array([1, .25, 0], np.float64)),
|
|
("trace", np.array([1, np.nan, 0], np.float32)),
|
|
("trace", np.array([1.1, 0, 0], np.float32)),
|
|
("candidate_sources", np.array([1, 1, 3, 8, 3], np.int32)),
|
|
("compartment", "PAM15"), ("d", -.1), ("d", .5), ("d", float("nan")),
|
|
("d", True), ("frozen", 1),
|
|
("base_candidate", np.zeros(5, np.float32)),
|
|
("delta", np.ones(5, np.float32)),
|
|
("delta", np.full(5, -.5, np.float32)),
|
|
("delta", np.full(5, -0., np.float32))]
|
|
for key, value in bad:
|
|
with self.subTest(key=key, value=value):
|
|
args = update_args()
|
|
args[key] = value
|
|
with self.assertRaises(ValueError):
|
|
bridge.apply_eligibility(**args)
|
|
|
|
def test_empty_candidate_subset_is_safe_but_not_claimed_full_mask(self):
|
|
args = update_args()
|
|
for key in ("candidate_sources", "candidate_compartments", "delta", "base_candidate"):
|
|
args[key] = args[key][:0]
|
|
result, report = bridge.apply_eligibility(**args)
|
|
self.assertEqual(result.shape, (0,))
|
|
self.assertFalse(report["complete_mask_identity_verified_by_this_function"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|