ClaudeandClaude Opus 5.5
77bab1c4ea
Explicit stream synchronization before/after each timed loop (refined FixedPatchStepper stream, uniform Simulation stream, in their contexts); required --max-wall-seconds outer timeout checked between variants and in per-step loops; untimed finite leaf/fluid health checks; force notes for B/C/E' (no host reads; final base interval only; whole-window average with different summation order). Parity stays A/B/C vs A and E/E' vs E only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
58 lines
2.4 KiB
Python
58 lines
2.4 KiB
Python
"""CPU contract of the measurement-only refined DFG timing harness (no CUDA)."""
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
|
|
import time_refined_dfg as t # noqa: E402
|
|
|
|
|
|
class TimingHarnessContract(unittest.TestCase):
|
|
def test_equal_simulated_duration(self):
|
|
s = t.schedule(2000, 200)
|
|
self.assertEqual(s["refined"], (200, 2000))
|
|
self.assertEqual(s["uniform"], (400, 4000)) # one refined base step = two D80 steps
|
|
self.assertAlmostEqual(s["tstar_window"], 2000 * .01 / 40)
|
|
for bad in ((0, 1), (1, -1)):
|
|
with self.assertRaises(ValueError):
|
|
t.schedule(*bad)
|
|
|
|
def test_timing_only_labels(self):
|
|
for name in ("B", "C", "E'"):
|
|
self.assertFalse(t.VARIANTS[name]["force_history_valid"])
|
|
self.assertTrue(t.VARIANTS[name]["timing_only"])
|
|
for name in ("A", "E"):
|
|
self.assertTrue(t.VARIANTS[name]["force_history_valid"])
|
|
self.assertEqual({v["case"] for v in t.VARIANTS.values()}, {"refined", "uniform"})
|
|
self.assertIn("final base interval", t.VARIANTS["C"]["force_note"])
|
|
self.assertIn("whole-window", t.VARIANTS["E'"]["force_note"])
|
|
|
|
def test_deadline(self):
|
|
import time
|
|
with self.assertRaises(TimeoutError):
|
|
t.check_deadline(time.monotonic() - 1)
|
|
t.check_deadline(time.monotonic() + 60)
|
|
|
|
def test_parity_detects_bits_and_nan_masks(self):
|
|
a = np.array([[1., np.nan], [2., 3.]], np.float32)
|
|
self.assertIsNone(t.same_state([a], [a.copy()]))
|
|
b = a.copy(); b[1, 1] = np.nextafter(np.float32(3), np.float32(4))
|
|
self.assertIn("bits", t.same_state([a], [b]))
|
|
c = a.copy(); c[0, 0] = np.nan
|
|
self.assertIn("NaN", t.same_state([a], [c]))
|
|
|
|
def test_summary_reports_differences_without_claims(self):
|
|
rows = [dict(variant=v, seconds=s) for v, s in
|
|
(("A", 3.), ("A", 5.), ("B", 2.), ("C", 1.), ("E", 2.), ("E'", 1.5))]
|
|
s = t.summarize(rows, .5)
|
|
self.assertEqual(s["A"]["median_seconds"], 4.)
|
|
self.assertEqual(s["A_minus_B_seconds"], 2.)
|
|
self.assertEqual(s["B_minus_C_seconds"], 1.)
|
|
self.assertEqual(s["A_over_E"], 2.)
|
|
self.assertIn("not exact causal", s["interpretation"])
|
|
|
|
|
|
if __name__ == "__main__": unittest.main()
|