"""Bounded, cache-bypassing execution of official Flyvis moving-edge machinery. Run from the isolated Linux run directory, not the source repository. No training, motor decoder, sensory-to-steering mapping, or original Shiu model is invoked. """ import csv import hashlib import importlib.metadata import json import os from pathlib import Path import platform import resource import time ROOT = Path(__file__).resolve().parent os.environ.setdefault("FLYVIS_ROOT_DIR", str(ROOT / "data")) os.environ.setdefault("MPLBACKEND", "Agg") os.environ.setdefault("CUBLAS_WORKSPACE_CONFIG", ":4096:8") import numpy as np import torch import flyvis from flyvis.datasets.moving_bar import MovingEdge from flyvis.analysis.stimulus_responses import compute_responses def sha(p): return hashlib.sha256(Path(p).read_bytes()).hexdigest() def parameter_sha(network): h = hashlib.sha256() for k, x in sorted(network.state_dict().items()): h.update(k.encode()) h.update(x.detach().cpu().numpy().tobytes()) return h.hexdigest() class UniformGreyEdge(MovingEdge): """Same labels/duration as official movies, with every pixel held at 0.5.""" def __getitem__(self, index): return torch.full_like(super().__getitem__(index), 0.5) def main(): plan = json.loads((ROOT / "plan.json").read_text()) source_manifest = json.loads((ROOT / "source-manifest.json").read_text()) assert sha(ROOT / "plan.json") == source_manifest["plan_sha256"] for name, h in source_manifest["files"].items(): assert sha(ROOT / "source" / name) == h, name out = ROOT / "results-v01" out.mkdir(exist_ok=False) torch.set_num_threads(1) torch.set_num_interop_threads(1) torch.manual_seed(plan["seed"]) np.random.seed(plan["seed"]) torch.use_deterministic_algorithms(True) assert torch.cuda.is_available(), "GPU setup must be verified before executing" torch.cuda.reset_peak_memory_stats() t0 = time.monotonic() model_dir = flyvis.results_dir / "flow/0000/000" view = flyvis.NetworkView(model_dir) checkpointed = view.network(checkpoint="best", lazy=True) network = checkpointed.recover() initial_param_sha = parameter_sha(network) central = np.asarray(network.connectome.central_cells_index[:]) types = network.connectome.nodes.type[:].astype(str)[central] config = dict(plan["stimulus"]) config.pop("dataset") config["device"] = flyvis.device dataset = MovingEdge(**config) t_stim = float(dataset.t_stim_max) times = np.arange(dataset[0].shape[0]) * config["dt"] - config["t_pre"] dataset.arg_df.to_csv(out / "stimulus-labels.csv", index=False) conditions = {} params = {} for name, cls in [("moving", MovingEdge), ("repeat", MovingEdge), ("grey", UniformGreyEdge)]: torch.manual_seed(plan["seed"]) np.random.seed(plan["seed"]) ts = time.monotonic() # Direct call deliberately bypasses both NetworkView and joblib response caches. ds = compute_responses(checkpointed, cls, dict(config), 1, 1.0, 0.0, "central") response = ds.responses.values[0] stimulus = ds.stimulus.values assert np.isfinite(response).all() and np.isfinite(stimulus).all() np.savez_compressed(out / (name + ".npz"), responses=response, stimulus=stimulus, time=times, cell_types=types) conditions[name] = {"seconds": time.monotonic() - ts, "responses_shape": list(response.shape), "stimulus_shape": list(stimulus.shape)} params[name] = parameter_sha(checkpointed.network) print(name, conditions[name], flush=True) del response, stimulus, ds moving = np.load(out / "moving.npz") repeat = np.load(out / "repeat.npz") grey = np.load(out / "grey.npz") r, rr, rg = moving["responses"], repeat["responses"], grey["responses"] mask = (times >= 0) & (times < t_stim) baseline_index = np.flatnonzero(times < 0)[-1] labels = dataset.arg_df rows, curves = [], [] for cell in plan["readouts"]: ni = np.flatnonzero(types == cell) assert len(ni) == 1, (cell, ni) ni = int(ni[0]) intensity = 1 if cell.startswith("T4") else 0 indices = np.flatnonzero(labels.intensity.values == intensity) # This is a prespecified baseline-subtracted peak, not the paper's full # population-level validation score or a measured firing-rate estimate. peaks = np.max(r[indices][:, mask, ni] - r[indices, baseline_index, ni, None], axis=1) angles = labels.angle.values[indices].astype(int) known = plan["known_directions_degrees"][cell] opposite = (known + 180) % 360 preferred_peak = float(peaks[np.flatnonzero(angles == known)[0]]) opposite_peak = float(peaks[np.flatnonzero(angles == opposite)[0]]) observed = int(angles[int(np.argmax(peaks))]) error = min((observed-known) % 360, (known-observed) % 360) rows.append(dict(cell_type=cell, intensity=intensity, known_direction=known, observed_peak_direction=observed, angle_error=error, preferred_peak=preferred_peak, opposite_peak=opposite_peak, difference=preferred_peak-opposite_peak, passed=bool(preferred_peak > opposite_peak and error <= 30))) for angle, peak in zip(angles, peaks): curves.append(dict(cell_type=cell, intensity=intensity, angle=int(angle), peak=float(peak))) for filename, records in [("metrics.csv", rows), ("direction-curves.csv", curves)]: with (out / filename).open("w") as h: writer = csv.DictWriter(h, fieldnames=list(records[0])); writer.writeheader(); writer.writerows(records) repeat_diff = float(np.max(np.abs(r-rr))) grey_label_diff = float(np.max(np.abs(rg-rg[0:1]))) grey_variation = float(np.max(np.ptp(rg[:, mask, :], axis=1))) assert np.array_equal(moving["stimulus"], repeat["stimulus"]) assert np.all(grey["stimulus"] == 0.5) for name, h in source_manifest["files"].items(): assert sha(ROOT / "source" / name) == h, name summary = { "run_id": plan["run_id"], "source_commit": source_manifest["commit"], "plan_sha256": sha(ROOT / "plan.json"), "seed": plan["seed"], "checkpoint": str(checkpointed.checkpoint.relative_to(ROOT)), "checkpoint_sha256": sha(checkpointed.checkpoint), "parameter_sha_before": initial_param_sha, "parameter_sha_after_conditions": params, "weights_unchanged": all(x == initial_param_sha for x in params.values()), "cache_bypassed": True, "training_performed": False, "integration_dt_seconds": config["dt"], "motion_duration_seconds": t_stim, "metric_window": {"time_ge":0, "time_lt":t_stim, "baseline_frame":int(baseline_index)}, "central_cell_types": types.tolist(), "network_neurons":int(len(network.connectome.nodes.type)), "network_edges":int(len(network.connectome.edges.source_index)), "conditions": conditions, "direction_passed":sum(x["passed"] for x in rows), "direction_total":len(rows), "repeat_max_absolute_difference":repeat_diff, "repeat_passed": repeat_diff <= 1e-6, "grey_max_difference_across_labels":grey_label_diff, "grey_label_independence_passed":grey_label_diff <= 1e-6, "grey_max_temporal_range_during_motion_window":grey_variation, "elapsed_seconds":time.monotonic()-t0, "maximum_rss_kib":resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, "peak_cuda_allocated_bytes":torch.cuda.max_memory_allocated(), "environment":{"python":platform.python_version(), "platform":platform.platform(), "gpu":torch.cuda.get_device_name(), "torch_cuda":torch.version.cuda, "torch_deterministic_algorithms":torch.are_deterministic_algorithms_enabled(), "packages":{x:importlib.metadata.version(x) for x in ["flyvis","torch","torchvision","numpy","datamate","xarray","pandas","numba"]}}, "scope":plan["scope"]} summary["passed"] = bool(summary["weights_unchanged"] and summary["direction_passed"] == 8 and summary["repeat_passed"] and summary["grey_label_independence_passed"]) (out / "summary.json").write_text(json.dumps(summary, indent=2)+"\n") manifest = {p.name:{"bytes":p.stat().st_size,"sha256":sha(p)} for p in sorted(out.iterdir()) if p.is_file()} (out / "manifest.json").write_text(json.dumps(manifest, indent=2)+"\n") print(json.dumps({k:summary[k] for k in ["passed","direction_passed","direction_total","elapsed_seconds"]}), flush=True) if __name__ == "__main__": main()