"""Local, synthetic tests only: no CUDA, scientific trajectories, or heldout data.""" import copy import unittest import numpy as np from calibration import (ORIGINAL_MODES, ORIGINAL_SEEDS, ORIGINAL_CAL_IDS, APL_FILES, STATE_DTYPES, arrays_bitwise_equal, evaluate_gates, exact_repeat, pair_metrics, sample_cv, summarize_mode, validate_inventory, validate_state) GATES = {"sparse_min_fraction": .03, "sparse_max_fraction": .09, "minimum_separable_odor_pairs": 3, "max_fraction_strict": .5, "min_fraction_strict": 0., "same_jaccard_gt_between": True, "between_euclidean_gt_within": True, "both_modes_required": True, "finite_required": True, "repeat_exact_required": True} def synthetic_runs(silent=False): result = {} for mode in ORIGINAL_MODES: for odor in range(6): for seed in ORIGINAL_SEEDS: counts = np.zeros(1000, dtype=np.int32) if not silent: counts[odor * 50:(odor + 1) * 50] = 1 result[(mode, str(odor), seed)] = {"kc_counts": counts, "expected_external_events": 10., "external_events": 0 if silent else 10} return result class CalibrationTests(unittest.TestCase): def test_complete_gate_and_original_pair_counts(self): summary = summarize_mode(ORIGINAL_MODES[0], list(map(str, range(6))), ORIGINAL_SEEDS, synthetic_runs(), GATES) self.assertTrue(summary["gate"]) self.assertEqual(summary["within_distance_count"], 60) self.assertEqual(summary["between_distance_count"], 75) self.assertEqual(summary["pair_count"], 15) self.assertEqual(summary["pairwise_separation_pass_count"], 15) self.assertTrue(all(row["within_distance_count"] == 20 and row["between_distance_count"] == 5 for row in summary["pairs"])) def test_silent_is_not_pass(self): summary = summarize_mode(ORIGINAL_MODES[0], list(map(str, range(6))), ORIGINAL_SEEDS, synthetic_runs(True), GATES) self.assertFalse(summary["gate"]) self.assertFalse(summary["conditions"]["minimum_strictly_positive"]) self.assertFalse(summary["conditions"]["same_jaccard_gt_different"]) self.assertFalse(summary["conditions"]["different_euclidean_gt_same"]) self.assertEqual(summary["pairwise_separation_pass_count"], 0) self.assertIsNone(summary["actual_event_count_cv"]) def test_every_original_conjunct_is_necessary(self): baseline = summarize_mode(ORIGINAL_MODES[0], list(map(str, range(6))), ORIGINAL_SEEDS, synthetic_runs(), GATES) changes = (("KC_active_fraction_mean", .029), ("KC_active_fraction_mean", .091), ("KC_active_fraction_max", .5), ("KC_active_fraction_min", 0.), ("same_odor_jaccard_mean", baseline["different_odor_jaccard_mean"]), ("different_odor_euclidean_mean", baseline["same_odor_euclidean_mean"]), ("pairwise_separation_pass_count", 2)) for key, value in changes: with self.subTest(key=key, value=value): modified = dict(baseline, **{key: value}) self.assertFalse(all(evaluate_gates(modified, GATES).values())) for value in (.03, .09): self.assertTrue(evaluate_gates(dict(baseline, KC_active_fraction_mean=value), GATES)["mean_sparsity_in_closed_interval"]) def test_pair_specific_not_global_baseline(self): runs = synthetic_runs() mode = ORIGINAL_MODES[0] for odor in range(6): for index, seed in enumerate(ORIGINAL_SEEDS): counts = np.array([index + odor * 100] if odor < 2 else [10000 * odor + index * 1000], dtype=np.int32) runs[(mode, str(odor), seed)]["kc_counts"] = counts summary = summarize_mode(mode, list(map(str, range(6))), ORIGINAL_SEEDS, runs, GATES) first_pair = summary["pairs"][0] self.assertEqual(first_pair["within_this_pair_mean"], 2.) self.assertEqual(first_pair["between_same_seed_mean"], 100.) self.assertTrue(first_pair["gate"]) self.assertGreater(summary["same_odor_euclidean_mean"], 100.) def test_original_metric_zero_and_count_semantics(self): zero = np.zeros(3, dtype=np.int32) self.assertEqual(pair_metrics(zero, zero)["active_jaccard"], 1.) self.assertEqual(pair_metrics(zero, zero)["correlation"], 1.) result = pair_metrics(np.array([2, 0, 0]), np.array([1, 0, 0])) self.assertEqual(result["euclidean_distance"], 1.) self.assertEqual(result["active_jaccard"], 1.) self.assertAlmostEqual(sample_cv([1., 2., 3.]), .5) def test_repeat_comparison_is_bits_and_dtype(self): self.assertFalse(arrays_bitwise_equal(np.array([0.], np.float64), np.array([-0.], np.float64))) self.assertFalse(arrays_bitwise_equal(np.array([1], np.int32), np.array([1], np.int64))) self.assertTrue(arrays_bitwise_equal(np.array([1.], np.float32), np.array([1.], np.float32))) def test_exact_repeat_covers_all_recorded_semantics(self): first = {"state": {name: np.zeros(1, dtype) for name, dtype in STATE_DTYPES.items()}, "apl": {name: np.zeros((1, width), np.float32) for name, width in APL_FILES.items()}, "hashes": [{"tick": 0}], "queues": [{"tick": 0}], "requests": [], "component": {"environment": {"tf32": False}}} self.assertTrue(exact_repeat(first, copy.deepcopy(first))["pass_all"]) for category in ("state", "apl"): for name in first[category]: bad = copy.deepcopy(first) bad[category][name].flat[0] = 1 with self.subTest(category=category, name=name): self.assertFalse(exact_repeat(first, bad)["pass_all"]) for key in ("hashes", "queues", "requests", "component"): bad = copy.deepcopy(first) bad[key] = {} if isinstance(first[key], list) else [] with self.subTest(key=key): self.assertFalse(exact_repeat(first, bad)["pass_all"]) def test_raw_state_rejects_nonfinite_duplicate_and_wrong_counts(self): state = {"step": np.array([0], np.int32), "neuron_idx": np.array([1], np.int32), "counts": np.array([0, 1], np.int32), "v": np.zeros(2, np.float64), "g": np.zeros(2, np.float64), "ref": np.zeros(2, np.int32)} validate_state(state, 2, 10) for mutation in ("nan", "counts", "duplicate", "wrong_dtype"): bad = copy.deepcopy(state) if mutation == "nan": bad["g"][0] = np.nan if mutation == "counts": bad["counts"][0] = 1 if mutation == "duplicate": bad["step"] = np.array([0, 0], np.int32) bad["neuron_idx"] = np.array([1, 1], np.int32) bad["counts"][1] = 2 if mutation == "wrong_dtype": bad["v"] = bad["v"].astype(np.float32) with self.subTest(mutation=mutation), self.assertRaises(ValueError): validate_state(bad, 2, 10) def test_inventory_rejects_missing_duplicate_or_heldout_identity(self): odors = sorted(ORIGINAL_CAL_IDS) records = [{"name": f"cal-{mode}-{index:02d}-s{seed}", "mode": mode, "odor_index": index, "odor_id": odors[index], "seed": seed} for mode in ORIGINAL_MODES for index in range(6) for seed in ORIGINAL_SEEDS] cfg = {"calibration": {"modes": list(ORIGINAL_MODES), "seeds": list(ORIGINAL_SEEDS), "duration_ms": 100., "odor_count": 6, "split": "calibration", "repeat_first_seed_each_odor": True}, "gates": GATES} self.assertEqual(validate_inventory(records, cfg), odors) for malformed in (records[:-1], records[:-1] + [records[0]], [dict(row, odor_id="heldout") if row["odor_index"] == 0 else row for row in records]): with self.assertRaises(ValueError): validate_inventory(malformed, cfg) if __name__ == "__main__": unittest.main()