ClaudeandClaude Opus 5.5
6c53d2e74a
Simulation.get_ddf() is flat (9*ny*nx); view it as (Q, cells) before applying the fluid mask, with a flat-DDF regression. Label A/E as runner-like force-read workloads (values discarded, no aggregation or validation; no macro/probe/CSV cost; not benchmark throughput) and split status lines into CONSTRUCTED and WARM_READY. Test/doc only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
87 lines
4.0 KiB
Python
87 lines
4.0 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.assertFalse(t.VARIANTS[name]["force_history_valid"]) # reads discarded
|
|
self.assertTrue(t.VARIANTS[name]["force_reads"])
|
|
self.assertIn("runner-like", t.VARIANTS[name]["force_note"])
|
|
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]))
|
|
self.assertIn("level count", t.same_state([a], [a, a]))
|
|
inf = a.copy(); inf[1, 0] = np.inf
|
|
self.assertIn("infinite", t.same_state([inf], [inf.copy()])) # matching inf still fails
|
|
self.assertIn("infinite", t.same_state([a], [inf]))
|
|
|
|
def test_flat_uniform_ddf(self):
|
|
fluid = np.array([[True, False, True], [True, True, False]])
|
|
flat = np.ones(9 * fluid.size, np.float32) # Simulation.get_ddf() layout
|
|
self.assertTrue(t.uniform_populations_finite(flat, fluid))
|
|
flat[1 * fluid.size + 1] = np.nan # q=1, solid cell 1: ignored
|
|
self.assertTrue(t.uniform_populations_finite(flat, fluid))
|
|
flat[4 * fluid.size + 0] = np.nan # q=4, fluid cell 0: detected
|
|
self.assertFalse(t.uniform_populations_finite(flat, fluid))
|
|
self.assertTrue(t.uniform_populations_finite(np.ones((9,) + fluid.shape), fluid))
|
|
with self.assertRaises(RuntimeError):
|
|
t.uniform_populations_finite(np.ones(9 * fluid.size + 1), fluid)
|
|
|
|
def test_physical_health_limits(self):
|
|
ok = (np.array([1., 1.01]), np.array([.01, .02]), np.array([0., .01]))
|
|
health = t.physical_health([ok])
|
|
self.assertAlmostEqual(health["rho_max"], 1.01)
|
|
for bad in ((np.array([1., np.nan]),) + ok[1:],
|
|
(np.array([.4, 1.]),) + ok[1:],
|
|
(ok[0], np.array([.2, 0.]), ok[2]),
|
|
(np.array([]), np.array([]), np.array([]))):
|
|
with self.assertRaises(RuntimeError):
|
|
t.physical_health([bad])
|
|
|
|
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()
|