"""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_finest((PINS+MOVED+ADDED)[::-1], 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_finest(PINS+FLUID, 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.boundaries[2].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_finest(rects, 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()