"""Tiny CPU algebra/event tests; no full trajectory or GPU execution.""" from dataclasses import replace import json from pathlib import Path import tempfile import unittest import numpy as np import spacetime_reference as st class SpaceTime(unittest.TestCase): @classmethod def setUpClass(cls): cls.plan = st.Plan(12, (3, 3, 9, 8)) cls.spec = st.Spec(n_coarse=12, patch_lo=1, halo=2, tau_coarse=.5576, tau_fine=.6152, collision="srt") def state(self, tagged=False): arrays = [] for n, mask in ((12,self.plan.maps.coarse_active),(24,self.plan.maps.fine_active)): q, y, x = np.indices((9,n,n)) a = np.broadcast_to(st.cpu.W[:,None,None], (9,n,n)).copy() if tagged: a *= 1+1e-3*np.random.default_rng(n).normal(size=a.shape) a[:,~mask] = np.nan arrays.append(a) return st.cpu.State(*arrays) def test_events_and_corner_leftovers(self): groups = [g for g in self.plan.groups if None in g.events] self.assertEqual(len(groups), 8) self.assertTrue(all(g.events.count(None)==3 for g in groups)) for m in (0,1): events = list(map(tuple,self.plan.outgoing[m])) self.assertEqual(len(events),len(set(events))) st.Plan(16,(4,5,11,12)) def test_interpolation(self): for g in self.plan.groups: samples, a, e, coeff = st.interpolation(g.kind,g.points) np.testing.assert_allclose(g.weights@np.ones(len(samples)),1,rtol=0,atol=2e-14) np.testing.assert_allclose(g.weights.mean(axis=0), np.eye(len(samples))[1 if g.kind=='face' else 0],rtol=0,atol=2e-14) if g.kind=='face': beta=np.array([1.,.03,.02,0.]) else: cx,cy=st.cpu.CX[g.q],st.cpu.CY[g.q] beta=np.array([1.,.03,.02,-cx*.03-cy*.02]) steady=a@np.array([1.,.03,.02,0.]) self.assertLess(abs((coeff@steady)[3]),2e-14) np.testing.assert_allclose(g.weights@(a@beta),e@beta,rtol=0,atol=2e-14) def test_rest(self): initial=self.state() final,d=st.advance(self.spec,self.plan,initial) for a,b in zip((initial.coarse_f,initial.fine_f),(final.coarse_f,final.fine_f)): np.testing.assert_allclose(a,b,rtol=0,atol=2e-14,equal_nan=True) def test_frozen_recovers_ivp(self): for collision in ('identity','srt'): spec=replace(self.spec,collision=collision) a=b=self.state(True) for _ in range(2): a,_=st.advance(spec,self.plan,a,'frozen') b,_=st.cpu.advance_coarse_once(spec,self.plan.maps,b) for aa,bb in zip((a.coarse_f,a.fine_f),(b.coarse_f,b.fine_f)): np.testing.assert_allclose(aa,bb,rtol=0,atol=2e-14,equal_nan=True) def test_second_substep_uses_characteristic_origin(self): # Review counterexample in (x,y): ghost (7,10), m=1 traces back # to (6,9), parent (3,4), not its current parent (3,5). plan=st.Plan(16,(4,4,12,12)) owners=[g for g in plan.groups if g.q==5 and (1,10*32+7) in g.events] self.assertEqual(len(owners),1) self.assertEqual(owners[0].source,4*16+3) spec=st.Spec(n_coarse=16,patch_lo=2,halo=4,collision='identity') arrays=[] for n,mask in ((16,plan.maps.coarse_active),(32,plan.maps.fine_active)): a=np.broadcast_to(st.cpu.W[:,None,None],(9,n,n)).copy() a[:,~mask]=np.nan arrays.append(a) arrays[0][5,4,3]=1.; arrays[0][5,5,3]=2. final,_=st.advance(spec,plan,st.cpu.State(*arrays),'frozen') self.assertEqual(final.fine_f[5,11,8],1.) self.assertNotEqual(final.fine_f[5,11,8],arrays[0][5,5,3]) def test_corner_affine_tradeoff_is_explicit(self): group=next(g for g in self.plan.groups if g.kind=='corner') _,a,e,coeff=st.interpolation(group.kind,group.points) linear_x=np.array([0.,1.,0.,0.]) # The two paper constraints are consistent. Arbitrary affine exactness # is the incompatible additional demand. self.assertEqual(abs(float((e@linear_x).mean())),.125) fitted=coeff@(a@linear_x) self.assertLess(abs(fitted[3]),2e-14) self.assertLess(abs(float((e@fitted).mean())),2e-14) self.assertGreater(np.max(np.abs(e@fitted-e@linear_x)),.01) def test_directional_budgets(self): for collision in ('identity','srt'): spec=replace(self.spec,collision=collision) a=self.state(True) for _ in range(2): a,d=st.advance(spec,self.plan,a) np.testing.assert_allclose(d['directional_residual'],0,rtol=0,atol=2e-12) self.assertLess(d['group_error'],2e-14) def test_missing_packets_fail(self): for mutation in ('omit_leftover','omit_outgoing_0','omit_outgoing_1'): _,d=st.advance(self.spec,self.plan,self.state(True),mutation=mutation) self.assertGreater(np.max(np.abs(d['directional_residual'])),1e-5) def test_negative_interpolation_is_reported_not_clipped(self): initial=self.state() group=self.plan.groups[0] initial.coarse_f.reshape(9,-1)[group.q,group.source]=-.01 spec=replace(self.spec,collision='identity') _,d=st.advance(spec,self.plan,initial,'frozen') self.assertGreaterEqual(d['negative_boundary_count'],4) self.assertEqual(d['min_boundary'],-.01) self.assertEqual(d['nonfinite_boundary_count'],0) class Receipts(unittest.TestCase): def test_completed_and_failed_screen_receipts(self): for passed in (True, False): with self.subTest(passed=passed), tempfile.TemporaryDirectory() as directory: # Same NumPy comparison type produced by the final metric screen. checks = {'velocity_l2': np.float64(.5 if passed else 1.) <= .8, 'mass_momentum': True} self.assertIsInstance(checks['velocity_l2'], np.bool_) result = dict(completed=True, checks=checks, experimental_pass=all(checks.values()), runs=[dict(rows=[dict(step=np.int64(1024), mass_relative=np.float32(0.))])]) path = Path(directory)/'receipt.json' st.write_receipt(path, result) saved = json.loads(path.read_text()) self.assertIs(saved['completed'], True) self.assertIs(saved['experimental_pass'], passed) self.assertIs(saved['checks']['velocity_l2'], passed) self.assertEqual(saved['runs'][0]['rows'][0], dict(step=1024, mass_relative=0.)) def test_receipts_still_reject_nonfinite_and_unknown_values(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory)/'receipt.json' with self.assertRaises(ValueError): st.write_receipt(path, {'invalid': np.float32('nan')}) with self.assertRaises(TypeError): st.write_receipt(path, {'invalid': object()}) self.assertFalse(path.exists()) if __name__=='__main__': unittest.main(verbosity=2)