192 lines
11 KiB
Python
192 lines
11 KiB
Python
"""Host-only ownership/geometry tests, never a GPU or accuracy admission."""
|
|
import copy
|
|
import json
|
|
import sys
|
|
import unittest
|
|
|
|
import numpy as np
|
|
from CelerisLab.config import LBMConfig
|
|
from CelerisLab.lbm.blocks import BlockLayout2D
|
|
from CelerisLab.lbm.refinement import FixedPatch2D, FixedPatchSet2D
|
|
from spacetime_union_plan import UnionPlan, UnresolvedTopology, owned_slots, validate_ownership
|
|
import spacetime_reference as ref
|
|
import spacetime_gpu as frozen # pack_plan only: no CUDA import/owner construction
|
|
from spacetime_table_guard import exact_weights
|
|
|
|
MATRIX=[]
|
|
CASES={
|
|
'rectangle': ([(4,4,12,12)],True),
|
|
'two_separated': ([(3,3,6,6),(9,9,12,12)],True),
|
|
'one_cell_face_gap': ([(3,3,6,10),(7,3,10,10)],True),
|
|
'one_cell_diagonal_gap': ([(3,3,6,6),(7,7,10,10)],True),
|
|
'staggered_gap': ([(3,3,6,8),(7,5,11,10)],True),
|
|
'thin_rectangle': ([(4,4,5,11)],True),
|
|
'concave_L': ([(4,4,5,7),(4,5,7,7)],False),
|
|
'corner_contact': ([(3,3,6,6),(6,6,9,9)],False),
|
|
'cross': ([(3,6,13,9),(6,3,9,13)],False),
|
|
'narrow_notch': ([(3,3,6,11),(7,3,10,11),(3,9,10,11)],False),
|
|
'hole': ([(3,3,13,5),(3,11,13,13),(3,5,5,11),(11,5,13,11)],False),
|
|
}
|
|
|
|
|
|
def pair(rects,n=16,ny=None):
|
|
cfg=LBMConfig(nx=n,ny=n if ny is None else ny,viscosity=.0192,velocity=0.)
|
|
return FixedPatchSet2D.from_config(cfg,rects)
|
|
|
|
|
|
def assert_same(test,a,b):
|
|
# Includes event-child order, zero padding, integer dtype/shape and FP64 bits.
|
|
aa,bb=frozen.pack_plan(a),frozen.pack_plan(b)
|
|
test.assertEqual(aa.keys(),bb.keys())
|
|
for key in aa:
|
|
test.assertEqual((aa[key].dtype,aa[key].shape,aa[key].tobytes()),
|
|
(bb[key].dtype,bb[key].shape,bb[key].tobytes()),key)
|
|
for x,y in zip(a.groups,b.groups):
|
|
test.assertEqual((x.q,x.source,x.receiver,x.kind,x.points,x.events),
|
|
(y.q,y.source,y.receiver,y.kind,y.points,y.events))
|
|
|
|
|
|
class UnionContract(unittest.TestCase):
|
|
def test_single_rectangle_exact_707afe6(self):
|
|
frozen.check_sources()
|
|
for n,b in ((12,(3,3,9,8)),(16,(4,5,11,12)),(16,(4,4,5,7)),(64,(8,8,56,48))):
|
|
with self.subTest(n=n,bounds=b):
|
|
assert_same(self,UnionPlan(pair([b],n)),ref.Plan(n,b))
|
|
self.assertNotIn('pycuda.driver',sys.modules)
|
|
|
|
def test_union_decomposition_and_permutation_invariance(self):
|
|
baseline=UnionPlan(pair([(4,4,12,12)]))
|
|
decompositions=[[(4,4,8,12),(8,4,12,12)],[(4,4,10,12),(6,4,12,12)],
|
|
[(4,4,12,8),(4,8,12,12)],[(4,4,12,12),(4,4,12,12)]]
|
|
for rectangles in decompositions:
|
|
for order in (rectangles,list(reversed(rectangles))):
|
|
assert_same(self,baseline,UnionPlan(pair(order)))
|
|
two=CASES['two_separated'][0]
|
|
assert_same(self,UnionPlan(pair(two)),UnionPlan(pair(list(reversed(two)))))
|
|
|
|
def test_supported_unresolved_matrix(self):
|
|
for name,(rectangles,supported) in CASES.items():
|
|
geometry=pair(rectangles) # Existing production accepts EVERY fixture.
|
|
row=dict(case=name,rectangles=rectangles,production_geometry_accepted=True,
|
|
runtime_supported=False,interpolation_accuracy='unqualified_for_new_unions')
|
|
if supported:
|
|
plan=UnionPlan(geometry); row.update(plan.proof)
|
|
self.assertEqual(plan.proof['receiver_quarter_packets'],4*len(plan.receivers))
|
|
else:
|
|
with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(geometry)
|
|
self.assertEqual(caught.exception.code,'fine_exit_reentry')
|
|
row.update(topology='unresolved',code=caught.exception.code,witness=caught.exception.detail)
|
|
MATRIX.append(row)
|
|
|
|
def test_L_witness_and_reflections(self):
|
|
rectangles=CASES['concave_L'][0]
|
|
with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(pair(rectangles))
|
|
self.assertEqual(caught.exception.detail,dict(q=5,substep=0,
|
|
fine_path_yx=[[8,9],[9,10],[10,11]],coarse_target_yx=[5,5]))
|
|
directions=set()
|
|
for reflect in (False,True):
|
|
transformed=[(16-x1,y0,16-x0,y1) if reflect else (x0,y0,x1,y1) for x0,y0,x1,y1 in rectangles]
|
|
for _ in range(4):
|
|
with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(pair(transformed))
|
|
detail=caught.exception.detail; directions.add(detail['q'])
|
|
p=np.array(detail['fine_path_yx']); geometry=pair(transformed)
|
|
self.assertEqual([bool(geometry.fine_kind[tuple(z)]==0) for z in p],[True,False,True])
|
|
transformed=[(16-y1,x0,16-y0,x1) for x0,y0,x1,y1 in transformed]
|
|
self.assertGreater(len(directions),1)
|
|
|
|
def test_ordered_provenance_and_packet_mutations_fail(self):
|
|
baseline=UnionPlan(pair(CASES['two_separated'][0]))
|
|
def reject(change):
|
|
bad=copy.deepcopy(baseline); change(bad)
|
|
with self.assertRaises((ValueError,KeyError)): validate_ownership(bad)
|
|
reject(lambda p:setattr(p.groups[0],'source',p.groups[0].source+1))
|
|
reject(lambda p:setattr(p.groups[0],'samples',p.groups[0].samples+1))
|
|
reject(lambda p:setattr(p.groups[0],'events',tuple(reversed(p.groups[0].events))))
|
|
reject(lambda p:setattr(p.groups[0],'receiver',p.groups[0].receiver+1))
|
|
reject(lambda p:p.outgoing.__setitem__(0,p.outgoing[0][1:]))
|
|
reject(lambda p:p.outgoing.__setitem__(1,np.vstack((p.outgoing[1],p.outgoing[1][0]))))
|
|
reject(lambda p:setattr(p,'receivers',p.receivers[1:]))
|
|
def balanced(p):
|
|
p.groups[0].weights[:2,:2]+=1e-5*np.array([[1,-1],[-1,1]])
|
|
reject(balanced) # Preserves row/column sums, corrupts the actual operator.
|
|
|
|
def test_coefficients_and_directional_transfer_budget(self):
|
|
rng=np.random.default_rng(31)
|
|
for name in ('two_separated','one_cell_face_gap','thin_rectangle'):
|
|
p=UnionPlan(pair(CASES[name][0])); total=np.zeros(9)
|
|
receiver={tuple(row):[] for row in p.receivers}; first_leftover=None
|
|
for g in p.groups:
|
|
samples,a,e,coeff=ref.interpolation(g.kind,g.points)
|
|
np.testing.assert_allclose(g.weights,exact_weights(g.kind,g.points),rtol=0,atol=2e-14)
|
|
np.testing.assert_allclose(g.weights@np.ones(len(samples)),1,rtol=0,atol=2e-14)
|
|
if g.kind=='face': beta=np.array([1.,.03,.02,0.])
|
|
else:
|
|
beta=np.array([1.,.03,.02,-ref.cpu.CX[g.q]*.03-ref.cpu.CY[g.q]*.02])
|
|
self.assertLess(abs((coeff@(a@np.array([1.,.03,.02,0.])))[3]),2e-14)
|
|
np.testing.assert_allclose(g.weights@(a@beta),e@beta,rtol=0,atol=2e-14)
|
|
values=rng.normal(size=len(samples)); h=g.weights@values
|
|
origin=values[0 if g.kind=='corner' else 1]
|
|
# Fine incoming plus coarse leftovers must exactly split one
|
|
# coarse directional source budget, regardless of sample values.
|
|
incoming=sum(h[k] for k,event in enumerate(g.events) if event is not None)
|
|
leftover=sum(h[k] for k,event in enumerate(g.events) if event is None)
|
|
self.assertLess(abs(.25*(incoming+leftover)-origin),2e-14)
|
|
total[g.q]+=.25*incoming-origin
|
|
for k,event in enumerate(g.events):
|
|
if event is None:
|
|
receiver[(g.q,g.receiver)].append(h[k])
|
|
if first_leftover is None: first_leftover=(g.q,h[k])
|
|
# Tagged collision-free boundary-flux ledger, not a solver trajectory:
|
|
# every outgoing fine debit is credited to its actual receiver.
|
|
for rows in p.outgoing:
|
|
for q,slot,target in rows:
|
|
value=float(rng.normal()); total[q]-=.25*value
|
|
receiver[(q,target)].append(value)
|
|
for (q,target),values in receiver.items():
|
|
self.assertEqual(len(values),4); total[q]+=.25*sum(values)
|
|
np.testing.assert_allclose(total,0,rtol=0,atol=2e-14)
|
|
damaged=total.copy(); q,value=first_leftover; damaged[q]-=.25*value
|
|
self.assertGreater(float(abs(damaged).max()),1e-5)
|
|
self.assertEqual(validate_ownership(p)['topology'],'supported')
|
|
|
|
def test_existing_block_addressing_and_sparse_holes(self):
|
|
geometry=pair(CASES['two_separated'][0]); p=UnionPlan(geometry)
|
|
layouts=(BlockLayout2D.build(16,16,4,4),geometry.fine_layout((4,4)))
|
|
for fine,layout in enumerate(layouts):
|
|
cells=layout.shape[0]*layout.shape[1]
|
|
dense=np.arange(9*cells,dtype=np.float64).reshape(9,*layout.shape)
|
|
blocked=layout.to_blocked(dense).ravel()
|
|
refs=[]
|
|
if not fine:
|
|
refs=[(g.q,int(k)) for g in p.groups for k in [g.source,g.receiver,*g.samples]]
|
|
else:
|
|
refs=[(g.q,slot) for g in p.groups for event in g.events if event is not None for _,slot in [event]]
|
|
refs += [(int(q),int(slot)) for rows in p.outgoing for q,slot,_ in rows]
|
|
q,k=np.array(refs).T
|
|
np.testing.assert_array_equal(blocked[owned_slots(layout,q,k)],dense.reshape(9,-1)[q,k])
|
|
holes=np.flatnonzero(~layouts[1].owned)
|
|
self.assertGreater(len(holes),0)
|
|
with self.assertRaises(UnresolvedTopology): owned_slots(layouts[1],np.array([1]),holes[:1])
|
|
|
|
def test_deferred_layouts_and_phase_are_explicit(self):
|
|
with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(pair([(3,3,8,8)],16,12))
|
|
self.assertEqual(caught.exception.code,'requires_level_wide_square_pair')
|
|
cfg=LBMConfig(nx=16,ny=16,viscosity=.0192,velocity=0.)
|
|
scalar=FixedPatch2D.from_config(cfg,(4,4,8,8))
|
|
with self.assertRaises(UnresolvedTopology): UnionPlan(scalar)
|
|
self.assertEqual(UnionPlan.phase,'arrival_f'); self.assertFalse(UnionPlan.runtime_supported)
|
|
# Restriction and collision do not commute, even at omega=1. This is
|
|
# why postcollision G storage cannot be relabeled/inverted into this F contract.
|
|
w=ref.cpu.W; cx,cy=ref.cpu.CX,ref.cpu.CY
|
|
def equilibrium(u):
|
|
cu=cx*u;return w*(1+3*cu+4.5*cu*cu-1.5*u*u)
|
|
children=np.array([equilibrium(u) for u in (.03,-.03,.03,-.03)])
|
|
self.assertAlmostEqual(float(abs(w-children.mean(axis=0)).max()),.0006,places=14)
|
|
|
|
|
|
if __name__=='__main__':
|
|
result=unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(UnionContract))
|
|
print(json.dumps(dict(passed=result.wasSuccessful(),tests=result.testsRun,
|
|
scope='host-only topology/provenance/coefficient checks; no runtime or accuracy admission',matrix=MATRIX),indent=2))
|
|
sys.exit(not result.wasSuccessful())
|