File indexing completed on 2026-11-09 09:32:56
0001 """CPU-only checks for the optical serial/MT timeline integration validator."""
0002
0003 import csv
0004 import json
0005 from pathlib import Path
0006 import subprocess
0007 import sys
0008 from tempfile import TemporaryDirectory
0009 import unittest
0010
0011
0012 CHECKER = Path(__file__).with_name("check_simg4ox_timing.py")
0013
0014
0015 def event_row(event_id, start, submit, gpu_start, gpu_end, release, end, threads=2):
0016 row = {
0017 "event_id": event_id, "dispatch_mode": "synchronous" if threads > 1 else "blocking",
0018 "start_offset_s": start, "cpu_pre_start_time_offset_s": start,
0019 "cpu_pre_end_time_offset_s": submit, "gpu_submit_time_offset_s": submit,
0020 "gpu_wait_start_time_offset_s": submit, "gpu_start_time_offset_s": gpu_start,
0021 "gpu_end_time_offset_s": gpu_end, "gpu_wait_end_time_offset_s": gpu_end,
0022 "cpu_post_start_time_offset_s": gpu_end, "cpu_post_end_time_offset_s": end,
0023 "end_offset_s": end, "gpu_release_time_offset_s": release,
0024 "runtime_s": end - start, "cpu_pre_runtime_s": submit - start,
0025 "cpu_post_runtime_s": end - gpu_end,
0026 "cpu_runtime_s": submit - start + end - gpu_end,
0027 "gpu_queue_delay_s": gpu_start - submit, "gpu_runtime_s": gpu_end - gpu_start,
0028 "gpu_wait_runtime_s": gpu_end - submit, "process_cpu_s": 0.1, "thread_cpu_s": 0.1,
0029 "cpu_post_locked_runtime_s": release - gpu_end,
0030 "cpu_post_unlocked_runtime_s": end - release, "gpu_gate_runtime_s": release - gpu_start,
0031 "primary_particle": "opticalphoton", "num_photons": 100,
0032 "worker_id": event_id % threads if threads > 1 else -1, "num_threads": threads,
0033 }
0034 return row
0035
0036
0037 class TimingCheckTest(unittest.TestCase):
0038 def invoke(self, rows, *, threads=2, recorder_enabled=False):
0039 with TemporaryDirectory() as directory:
0040 path = Path(directory) / "events.csv"
0041 with path.open("w", newline="") as stream:
0042 writer = csv.DictWriter(stream, fieldnames=list(rows[0]))
0043 writer.writeheader()
0044 writer.writerows(rows)
0045 manifest = {
0046 "schema_version": 2, "dispatch_mode": "synchronous" if threads > 1 else "blocking",
0047 "event_count": len(rows), "num_threads": threads,
0048 "primary_particle": "opticalphoton", "primary_multiplicity": 100,
0049 "cpu_recorder_enabled": recorder_enabled, "torch_seed": 0,
0050 }
0051 path.with_suffix(".manifest.json").write_text(json.dumps(manifest))
0052 return subprocess.run(
0053 [sys.executable, str(CHECKER), str(path), "--events", str(len(rows)),
0054 "--threads", str(threads), "--expected-photons", "100", "--expect-no-cpu-recorder"],
0055 capture_output=True, text=True, check=False,
0056 )
0057
0058 def test_overlapping_cpu_intervals_and_serialized_gpu_gates_are_valid(self):
0059 result = self.invoke([
0060 event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8),
0061 event_row(1, 0.1, 0.5, 0.8, 1.0, 1.1, 1.2),
0062 ])
0063 self.assertEqual(result.returncode, 0, result.stderr)
0064
0065 def test_gate_overlap_is_rejected_even_when_gpu_service_does_not_overlap(self):
0066 result = self.invoke([
0067 event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8),
0068 event_row(1, 0.1, 0.5, 0.65, 1.0, 1.1, 1.2),
0069 ])
0070 self.assertNotEqual(result.returncode, 0)
0071 self.assertIn("GPU dispatch gate overlaps", result.stderr)
0072
0073 def test_incorrect_primary_photon_count_is_rejected(self):
0074 row = event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8)
0075 row["num_photons"] = 0
0076 result = self.invoke([row])
0077 self.assertNotEqual(result.returncode, 0)
0078 self.assertIn("expected 100 photons", result.stderr)
0079
0080 def test_matched_serial_control_accepts_the_worker_free_schema(self):
0081 result = self.invoke([event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8, threads=1)], threads=1)
0082 self.assertEqual(result.returncode, 0, result.stderr)
0083
0084 def test_mt_requires_every_gpu_gate_measurement(self):
0085 for missing in ('gpu_release_time_offset_s', 'cpu_post_locked_runtime_s',
0086 'cpu_post_unlocked_runtime_s', 'gpu_gate_runtime_s'):
0087 row = event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8)
0088 del row[missing]
0089 with self.subTest(missing=missing):
0090 result = self.invoke([row])
0091 self.assertNotEqual(result.returncode, 0)
0092 self.assertIn('Missing MT gate measurements', result.stderr)
0093
0094 def test_legacy_serial_schema_keeps_optional_gate_columns(self):
0095 row = event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8, threads=1)
0096 for name in ('gpu_release_time_offset_s', 'cpu_post_locked_runtime_s',
0097 'cpu_post_unlocked_runtime_s', 'gpu_gate_runtime_s'):
0098 del row[name]
0099 result = self.invoke([row], threads=1)
0100 self.assertEqual(result.returncode, 0, result.stderr)
0101
0102 def test_core_durations_must_match_their_recorded_boundaries(self):
0103 for column in ('runtime_s', 'cpu_pre_runtime_s', 'cpu_post_runtime_s',
0104 'cpu_runtime_s', 'gpu_runtime_s', 'gpu_queue_delay_s'):
0105 row = event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8)
0106 row[column] = 1000.0
0107 with self.subTest(column=column):
0108 self.assertNotEqual(self.invoke([row]).returncode, 0)
0109
0110 def test_cpu_recorder_mismatch_is_rejected(self):
0111 result = self.invoke([event_row(0, 0, 0.3, 0.4, 0.6, 0.7, 0.8)], recorder_enabled=True)
0112 self.assertNotEqual(result.returncode, 0)
0113 self.assertIn("CPU recorder", result.stderr)
0114
0115
0116 if __name__ == "__main__":
0117 unittest.main()
