84 lines
4.5 KiB
Python
84 lines
4.5 KiB
Python
"""Pinned-body GPU lifecycle gates. NOT executed by CPU test collection.
|
|
|
|
No evolution accuracy tolerance is asserted here: an independent BFL replay
|
|
reference is required before a 3e-6 FP32-vs-FP64 gate can be claimed. These
|
|
checks require exact physical boundary/velocity, remap overlap and no-op next
|
|
step identity; they never compare blocked boundary addresses across layouts.
|
|
"""
|
|
from contextlib import ExitStack
|
|
import unittest
|
|
import numpy as np
|
|
from CelerisLab.cuda.context import CudaContext
|
|
from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
|
|
from CelerisLab.lbm.synchronized_remap import RemapTopology, remap_synchronized
|
|
from CelerisLab.body.geometry.circle import CircleGeometry
|
|
from CelerisLab.body.objects import SimObject
|
|
from CelerisLab.body.jets import Jet, Tube, Parallel
|
|
from test_runtime_child_remap_gpu import RuntimeChildRemapGPU, initial
|
|
from test_refinement_three_level_multipatch_gpu import CFG, PARENTS
|
|
from test_refinement_blocked_base import exact
|
|
|
|
A = ((16,16,32,40), (56,16,72,40))
|
|
B = ((16,16,32,40), (60,16,76,40))
|
|
|
|
|
|
def make(ctx):
|
|
body = SimObject(71, geometry=CircleGeometry(12.,14.,.8),center=(12.,14.),radius=.8)
|
|
jet = Jet(Tube((10.5,14.),0.,3.,.3),Parallel(0.),'parabolic',.003)
|
|
run = FixedPatchStepper(CFG,context=ctx,refinements=[PARENTS,A],level_blocks=[(4,4)]*3,
|
|
level_objects=[(),(),(body,)],level_jets=[(),(),(jet,)])
|
|
return run,(body,)
|
|
|
|
|
|
class PinnedBodyRemapGPU(unittest.TestCase):
|
|
check_public = RuntimeChildRemapGPU.check_public
|
|
check_reset = RuntimeChildRemapGPU.check_reset
|
|
|
|
def check_boundary(self, before, after):
|
|
for name in ('flags','boundary_cells','curved_hits'):
|
|
exact(self,getattr(after.boundaries[2],name),getattr(before.boundaries[2],name),name)
|
|
for a,b in zip(after.boundaries[2].curved_host,before.boundaries[2].curved_host):
|
|
exact(self,a,b,'physical cut links')
|
|
exact(self,after._level_boundaries[2].tables['surface_velocity'],
|
|
before._level_boundaries[2].tables['surface_velocity'],'current per-link velocities')
|
|
self.assertEqual(after._level_boundaries[2].jets[0].amplitude,.017)
|
|
|
|
def test_noop_next_step_bitwise(self):
|
|
with CudaContext(0) as ctx, ExitStack() as owned:
|
|
old,objects=make(ctx);owned.enter_context(old)
|
|
old.initialize(*initial(old,False));old.step(2)
|
|
old.set_jet_amplitude(0,.017,level=2)
|
|
before=old.download()
|
|
new=owned.enter_context(old.remap_finest(A,objects=objects))
|
|
self.check_boundary(old,new);self.check_reset(new,before)
|
|
old.step();new.step()
|
|
for a,b in zip(new.download(),old.download()): exact(self,a,b,'no-op next step bitwise')
|
|
self.assertEqual((old.step_count,new.step_count),(3,3))
|
|
self.check_public(new)
|
|
|
|
def test_moved_fluid_return_and_source_independence(self):
|
|
with CudaContext(0) as ctx, ExitStack() as owned:
|
|
current,objects=make(ctx);owned.enter_context(current)
|
|
current.initialize(*initial(current,False));current.step(2)
|
|
current.set_jet_amplitude(0,.017,level=2)
|
|
for rects in (B,A):
|
|
previous=current;before=previous.download();count=previous.step_count
|
|
jets=previous._level_boundaries[2].jets
|
|
old=RemapTopology.from_config(CFG,previous.hierarchy.pairs[0],previous.hierarchy.pairs[1].bounds,
|
|
child_objects=objects,child_jets=jets)
|
|
new=RemapTopology.from_config(CFG,previous.hierarchy.pairs[0],rects,
|
|
child_objects=objects,child_jets=jets)
|
|
expected=remap_synchronized(old,new,before,base_count=count,
|
|
level_steps=(count,2*count,4*count),phase='postcollision_g').g
|
|
current=owned.enter_context(previous.remap_finest(rects,objects=objects))
|
|
self.check_boundary(previous,current);self.check_reset(current,expected)
|
|
for a,b,om,nm in zip(current.download(),before,old.leaf_masks,new.leaf_masks):
|
|
exact(self,a[:,om & nm],b[:,om & nm],'physical-fluid overlap')
|
|
current.step(2)
|
|
self.assertEqual((previous.step_count,current.step_count),(count,count+2))
|
|
for a,b in zip(previous.download(),before): exact(self,a,b,'source independent')
|
|
self.check_public(current)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|