150 lines
7.1 KiB
Python
150 lines
7.1 KiB
Python
"""Multiple body pins and variable fluid sets; mocked runtime is host-only."""
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
from CelerisLab.config import LBMConfig
|
|
from CelerisLab.body.jets import Jet, Tube, Parallel
|
|
from CelerisLab.body.objects import SurfaceActuation
|
|
from CelerisLab.lbm.refinement import FixedPatch2D
|
|
from CelerisLab.lbm.synchronized_remap import RemapTopology
|
|
from tests.unit.test_pinned_body_remap import body, mesh
|
|
from tests.unit.test_refinement_three_level_multipatch import mock_backend
|
|
from tests.unit.test_synchronized_remap import fields, integrals, remap
|
|
|
|
PINS = ((16, 16, 32, 24), (56, 16, 72, 24))
|
|
FLUID = ((16, 32, 32, 40), (56, 32, 72, 40))
|
|
ADDED = ((16, 44, 32, 48),)
|
|
MOVED = ((20, 32, 36, 40), (60, 32, 76, 40))
|
|
|
|
|
|
def objects():
|
|
return (body(center=(12., 10.)), body(center=(32., 10.)))
|
|
|
|
|
|
def ready():
|
|
bodies = objects()
|
|
jets = tuple(Jet(Tube((x-1.5, 10.), 0., 3., .3), Parallel(0.),
|
|
'parabolic', .003) for x in (12., 32.))
|
|
with patch('tests.unit.test_refinement_three_level_multipatch.bodies_and_jets',
|
|
return_value=(bodies, jets)):
|
|
run, module, allocations = mock_backend(children=PINS+FLUID, bodies=True)
|
|
storage = {}
|
|
module.cuda.memcpy_htod.side_effect = lambda p, a: storage.__setitem__(id(p), np.array(a, copy=True))
|
|
module.cuda.memcpy_dtoh.side_effect = lambda a, p: np.copyto(a, storage[id(p)])
|
|
run.initialize(*fields(mesh(PINS+FLUID, bodies, jets), np.float32))
|
|
run._step_count = 7
|
|
return run, module, allocations, bodies
|
|
|
|
|
|
class MultiplePinnedRemap(unittest.TestCase):
|
|
def check_transfer(self, old, new, arrays, tolerance):
|
|
out = remap(old, new, arrays)
|
|
before = integrals(old, arrays)
|
|
np.testing.assert_allclose(integrals(new, out.g), before, rtol=0,
|
|
atol=tolerance*np.abs(before[:9]).sum())
|
|
for a, b, om, nm in zip(arrays, out.g, old.leaf_masks, new.leaf_masks):
|
|
self.assertEqual(a[:, om & nm].tobytes(), b[:, om & nm].tobytes())
|
|
self.assertTrue(np.isnan(b[:, ~nm]).all())
|
|
return out.g
|
|
|
|
def test_add_move_remove_and_body_only_all_populations(self):
|
|
for dtype, tolerance in ((np.float64, 2e-13), (np.float32, 2e-7)):
|
|
old = mesh(PINS+FLUID, objects())
|
|
arrays = fields(old, dtype)
|
|
for rects in (PINS+FLUID+ADDED, PINS+MOVED+ADDED, PINS, PINS+FLUID):
|
|
new = mesh(rects, objects())
|
|
self.assertEqual(new.pinned_rectangles, PINS)
|
|
arrays = self.check_transfer(old, new, arrays, tolerance)
|
|
old = new
|
|
|
|
def test_shared_pin_order_invariance_and_body_free(self):
|
|
bodies = (body(center=(11., 14.)), body(center=(14., 14.)))
|
|
rects = ((16, 16, 32, 40), (56, 16, 72, 40))
|
|
old, new = mesh(rects, bodies), mesh(rects[::-1], bodies)
|
|
self.assertEqual(old.pinned_rectangles, (rects[0],))
|
|
self.assertEqual(new.pinned_rectangles, old.pinned_rectangles)
|
|
self.check_transfer(old, new, fields(old), 2e-13)
|
|
only = mesh((rects[0],), bodies)
|
|
self.check_transfer(old, only, fields(old), 2e-13)
|
|
self.assertEqual(mesh(rects, ()).pinned_rectangles, ())
|
|
|
|
def test_shifted_scalar_parent_coordinates(self):
|
|
cfg = LBMConfig(nx=48, ny=32, viscosity=.1)
|
|
parent = FixedPatch2D.from_config(cfg, (4, 4, 44, 28))
|
|
self.assertEqual(parent.fine_origin, (2., 2.))
|
|
bodies = (body(center=(14., 12.)), body(center=(34., 12.)))
|
|
def topology(rects):
|
|
return RemapTopology.from_config(cfg, parent, rects, child_objects=bodies)
|
|
old, new = topology(PINS+FLUID), topology(PINS+MOVED)
|
|
self.assertEqual(old.pinned_rectangles, PINS)
|
|
self.check_transfer(old, new, fields(old), 2e-13)
|
|
|
|
def test_direct_helper_rejects_changed_body_order(self):
|
|
old = mesh(PINS+FLUID, objects())
|
|
candidates = [mesh(PINS+FLUID, objects()[::-1])]
|
|
for new in candidates:
|
|
# Invalid donors ensure boundary validation precedes array reads.
|
|
with self.assertRaisesRegex(ValueError, 'pinned'):
|
|
remap(old, new, None)
|
|
|
|
def test_live_independent_jets_and_source_ownership(self):
|
|
run, module, allocations, bodies = ready()
|
|
run.set_jet_amplitude(0, .017, level=2)
|
|
run.set_jet_amplitude(1, -.011, level=2)
|
|
before = run.download()
|
|
clone = run.remap_child((PINS+MOVED+ADDED)[::-1], child_objects=bodies)
|
|
try:
|
|
self.assertEqual(tuple(j.amplitude for j in clone._boundaries[2].jets), (.017, -.011))
|
|
np.testing.assert_array_equal(clone._boundaries[2].tables['surface_velocity'],
|
|
run._boundaries[2].tables['surface_velocity'])
|
|
clone.set_jet_amplitude(1, .005, level=2)
|
|
self.assertEqual(run._boundaries[2].jets[1].amplitude, -.011)
|
|
clone.step()
|
|
self.assertEqual((run.step_count, clone.step_count), (7, 8))
|
|
for got, want in zip(run.download(), before):
|
|
np.testing.assert_array_equal(got, want)
|
|
finally:
|
|
clone.close()
|
|
run.close()
|
|
|
|
def test_sub_ulp_cut_table_geometry_change_rejected(self):
|
|
run, module, allocations, bodies = ready()
|
|
changed = objects()
|
|
changed[1].geometry._radius = np.nextafter(.8, np.inf)
|
|
start = len(allocations)
|
|
module.cuda.memcpy_dtoh.reset_mock()
|
|
with self.assertRaisesRegex(ValueError, 'boundary'):
|
|
run.remap_child(PINS+FLUID, child_objects=changed)
|
|
self.assertEqual(len(allocations), start)
|
|
module.cuda.memcpy_dtoh.assert_not_called()
|
|
with self.assertRaisesRegex(ValueError, 'pinned'):
|
|
remap(mesh(PINS+FLUID, bodies), mesh(PINS+FLUID, changed), None)
|
|
run.close()
|
|
|
|
def test_each_pin_and_body_identity_rejected_before_device_access(self):
|
|
run, module, allocations, bodies = ready()
|
|
cases = [(PINS+FLUID, bodies[:1]), (PINS+FLUID, bodies[::-1]),
|
|
(PINS+FLUID, (bodies[0], body(center=(32.1, 10.)))),
|
|
(PINS+FLUID, (bodies[0], body(SurfaceActuation(0., .7, 0.), (32., 10.)))),
|
|
(PINS+FLUID, run.child_boundary.objects)]
|
|
for index in range(2):
|
|
changed = list(PINS+FLUID)
|
|
x0, y0, x1, y1 = changed[index]
|
|
changed[index] = (x0-4, y0, x1, y1)
|
|
changed[index] = (x0+8, y0, x1, y1) # removes actual body support
|
|
cases.append((changed, bodies))
|
|
cases.append((tuple(r for i, r in enumerate(PINS+FLUID) if i != index), bodies))
|
|
start = len(allocations)
|
|
module.cuda.memcpy_dtoh.reset_mock()
|
|
run._stream.synchronize.reset_mock()
|
|
for rects, supplied in cases:
|
|
with self.subTest(rects=rects), self.assertRaises(ValueError):
|
|
run.remap_child(rects, child_objects=supplied)
|
|
self.assertEqual(len(allocations), start)
|
|
module.cuda.memcpy_dtoh.assert_not_called()
|
|
run._stream.synchronize.assert_not_called()
|
|
self.assertEqual((run._status, run.step_count), ('ready', 7))
|
|
run.close()
|