"""CPU-only mechanics checks for the continuous Harnett-signature BCI.""" import os import sys import torch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sdil.bci import (BCIConfig, BCISDIL, causal_roles, generate_trajectories, run_day) from sdil.bci_metrics import (annotate_day_events, grouped_classification_accuracy, grouped_regression_correlation, signature_metrics) def check_roles_and_pairing(): cfg = BCIConfig(days=2, episodes_per_day=8, steps_per_episode=4) role = causal_roles(cfg) assert torch.allclose(role[:5], torch.full((5,), 0.2)) assert torch.allclose(role[5:10], torch.full((5,), -0.2)) assert torch.equal(role[10:], torch.zeros(30)) first = generate_trajectories(cfg, 11) second = generate_trajectories(cfg, 11) third = generate_trajectories(cfg, 12) assert torch.equal(first.context, second.context) assert torch.equal(first.process_noise, second.process_noise) assert torch.equal(first.perturbations, second.perturbations) assert not torch.equal(first.context, third.context) assert torch.equal(first.context[..., 0], torch.ones_like(first.context[..., 0])) print("BCI causal roles and task-seed pairing: exact") def check_causal_estimator(): cfg = BCIConfig(days=2, episodes_per_day=4096, steps_per_episode=2) model = BCISDIL(cfg, model_seed=3) generator = torch.Generator().manual_seed(4) soma = torch.randn(4096, cfg.n_neurons, generator=generator) * 0.2 xi = torch.empty_like(soma).bernoulli_(0.5, generator=generator).mul_(2).sub_(1) estimate = model.causal_targets(soma, cfg.target, xi) exact = model.exact_causal_direction(soma, cfg.target) mean_estimate = estimate.mean(0) mean_exact = exact.mean(0) cosine = torch.nn.functional.cosine_similarity( mean_estimate, mean_exact, dim=0).item() assert cosine > 0.99 assert estimate[:, 10:].abs().mean() > 0 # simultaneous cross-cell variance assert mean_estimate[10:].abs().max() < 0.03 print(f"BCI simultaneous causal estimator mean cosine: {cosine:.6f}") def check_predictor_identification(): cfg = BCIConfig(days=2, episodes_per_day=256, steps_per_episode=2, predictor_eta=0.2) model = BCISDIL(cfg, model_seed=5) generator = torch.Generator().manual_seed(6) active = torch.ones(256, dtype=torch.bool) for _ in range(100): soma = torch.randn(256, cfg.n_neurons, generator=generator).tanh() ordinary = model.coupling * soma model.update_predictor(soma, ordinary, active) assert (model.P - model.coupling).abs().max() < 1e-4 assert model.P_bias.abs().max() < 1e-5 print("BCI neutral per-cell predictor identifies ordinary coupling: exact") def check_phase_masks_and_velocity_reset(): cfg = BCIConfig(days=2, episodes_per_day=8, steps_per_episode=4, kappa=0.3, feedback="error_velocity") trajectories = generate_trajectories(cfg, 7) initial = BCISDIL(cfg, model_seed=8) intact = initial.clone() online_lesion = initial.clone() plasticity_lesion = initial.clone() active = torch.ones(8, dtype=torch.bool) args = ( trajectories.context[0, :, 0], trajectories.process_noise[0, :, 0], trajectories.perturbations[0, :, 0], torch.zeros(8, cfg.n_neurons), None, active, 1) out_intact = intact.step(*args, control_gain=1.0, plasticity_gain=1.0, learn_vectorizer=False) out_online = online_lesion.step(*args, control_gain=0.0, plasticity_gain=1.0, learn_vectorizer=False) out_plastic = plasticity_lesion.step( *args, control_gain=1.0, plasticity_gain=0.0, learn_vectorizer=False) assert not torch.equal(out_intact["soma"], out_online["soma"]) assert torch.equal(intact.W, online_lesion.W) assert torch.equal(out_intact["soma"], out_plastic["soma"]) assert torch.equal(plasticity_lesion.W, initial.W) assert torch.equal(out_intact["feedback"][:, 1], torch.zeros(8)) day = run_day(initial.clone(), trajectories, 0, collect=True) assert len(day["events"]) == cfg.steps_per_episode assert day["global_step"] == cfg.steps_per_episode print("BCI online/plasticity phase masks and episode velocity reset: exact") def check_temporal_difference_role_vectorizer(): """Role perturbations times performance change give the Harnett sign.""" cfg = BCIConfig( days=2, episodes_per_day=4096, steps_per_episode=2, feedback="performance_velocity", vectorizer_eta=0.5) model = BCISDIL(cfg, model_seed=12) generator = torch.Generator().manual_seed(13) soma = 0.2 * torch.randn( cfg.episodes_per_day, cfg.n_neurons, generator=generator) xi = torch.empty_like(soma).bernoulli_( 0.5, generator=generator).mul_(2).sub_(1) cursor_plus = (soma + cfg.perturb_sigma * xi) @ model.role cursor_minus = (soma - cfg.perturb_sigma * xi) @ model.role target = model.causal_role_targets(cursor_plus, cursor_minus, xi) saved_role = model.role.clone() model.role.zero_() assert torch.equal( target, model.causal_role_targets(cursor_plus, cursor_minus, xi)) model.role.copy_(saved_role) mean_target = target.mean(0) cosine = torch.nn.functional.cosine_similarity( mean_target, model.role, dim=0).item() assert cosine > 0.99 active = torch.ones(cfg.episodes_per_day, dtype=torch.bool) model.A.zero_() model.update_vectorizer( torch.ones(cfg.episodes_per_day, 1), torch.zeros(cfg.episodes_per_day, cfg.n_neurons), target, active) learned_cosine = torch.nn.functional.cosine_similarity( model.A[:, 0], model.role, dim=0).item() assert learned_cosine > 0.99 # With the neutral predictor exact, r_i = role_i * delta_error. Hence the # P+ minus P- residual is positive in improving epochs and negative in # worsening epochs, independent of instantaneous error magnitude. model.A[:, 0].copy_(model.role) model.P.copy_(model.coupling) model.P_bias.zero_() previous = torch.tensor([0.8, 0.4, 0.9, 0.3]) current = torch.tensor([0.6, 0.5, 0.5, 0.6]) delta = model.feedback_features(current, previous) _, innovation, _, _ = model.apical_components(soma[:4], delta) difference = (innovation[:, :cfg.n_plus].mean(1) - innovation[:, cfg.n_plus:cfg.n_plus + cfg.n_minus].mean(1)) improving = delta[:, 0] > 0 worsening = delta[:, 0] < 0 sign_index = 0.5 * ( difference[improving].mean() - difference[worsening].mean()) assert sign_index > 0 print( "BCI temporal-difference role vectorizer: " f"cos={cosine:.6f}, sign_index={sign_index.item():.6f}") def check_temporal_difference_plasticity_boundary(): cfg = BCIConfig( days=2, episodes_per_day=32, steps_per_episode=6, feedback="performance_velocity", forward_eta=0.03, kappa=0.0) trajectories = generate_trajectories(cfg, 14) initial = BCISDIL(cfg, model_seed=15) initial.P.copy_(initial.coupling) initial.A[:, 0].copy_(initial.role) intact = initial.clone() lesion = initial.clone() report_intact = run_day( intact, trajectories, 0, plasticity_gain=1.0, control_gain=0.0, learn_vectorizer=False, learn_predictor=False, collect=True) report_control_sham = run_day( initial.clone(), trajectories, 0, plasticity_gain=1.0, control_gain=1.0, learn_vectorizer=False, learn_predictor=False, collect=True) run_day( lesion, trajectories, 0, plasticity_gain=0.0, control_gain=0.0, learn_vectorizer=False, learn_predictor=False) assert not torch.equal(intact.W, initial.W) assert torch.equal(lesion.W, initial.W) for left, right in zip( report_intact["events"], report_control_sham["events"]): assert torch.equal(left["soma"], right["soma"]) print("BCI temporal-difference plasticity/online boundary: exact") def check_grouped_decoders_and_metrics(): generator = torch.Generator().manual_seed(9) groups = torch.arange(200) x = torch.randn(200, 3, generator=generator) labels = x[:, 0] + 0.2 * x[:, 1] > 0 accuracy, _ = grouped_classification_accuracy(x, labels, groups) correlation = grouped_regression_correlation(x, 2 * x[:, 0] - x[:, 2], groups) assert accuracy > 0.9 assert correlation > 0.99 cfg = BCIConfig(days=4, episodes_per_day=24, steps_per_episode=6, kappa=0.1) trajectories = generate_trajectories(cfg, 10) model = BCISDIL(cfg, model_seed=11) training_events = [] global_step = 0 for day in range(cfg.days): report = run_day( model, trajectories, day, collect=True, global_step=global_step) global_step = report["global_step"] annotate_day_events( report["events"], day, report["success"], episode_offset=day * cfg.episodes_per_day) training_events.extend(report["events"]) evaluation = run_day( model, trajectories, cfg.days - 1, collect=True, global_step=global_step, plasticity_gain=0.0, learn_vectorizer=False, learn_predictor=False) annotate_day_events(evaluation["events"], 0, evaluation["success"], 10000) metrics = signature_metrics( training_events, evaluation["events"], cfg, model.role) assert metrics["active_training_events"] > 0 assert metrics["evaluation_episodes"] == cfg.episodes_per_day assert all(torch.isfinite(torch.tensor(value)) for value in metrics.values()) print(f"BCI grouped decoder mechanics: acc={accuracy:.3f}, corr={correlation:.3f}") print("BCI preregistered signature metrics: finite") if __name__ == "__main__": check_roles_and_pairing() check_causal_estimator() check_predictor_identification() check_phase_masks_and_velocity_reset() check_temporal_difference_role_vectorizer() check_temporal_difference_plasticity_boundary() check_grouped_decoders_and_metrics() print("ALL BCI MECHANICS CHECKS PASSED")