"""CPU remap tests: full-population conservation, never a runtime AMR claim.""" from dataclasses import replace import unittest import numpy as np from CelerisLab.config import LBMConfig from CelerisLab.lbm.blocks import BlockLayout2D from CelerisLab.lbm.descriptors import D2Q9_EX, D2Q9_EY from CelerisLab.lbm.refinement import FixedPatch2D, FixedPatchSet2D from CelerisLab.lbm.synchronized_remap import RemapTopology, remap_synchronized CFG = LBMConfig(nx=48, ny=32, viscosity=.1) PARENTS = [(4, 4, 20, 28), (24, 4, 44, 28)] A, B = [(16, 16, 32, 40)], [(20, 16, 36, 40)] def topology(rectangles, parent=None, cfg=CFG): if parent is None: parent = FixedPatchSet2D.from_config(cfg, PARENTS) return RemapTopology.from_config(cfg, parent, rectangles) def fields(mesh, dtype=np.float64): rng = np.random.default_rng(8473) result = [] for leaf in mesh.leaf_masks: # Direction-dependent and non-equilibrium, with independently varying # children: q0 and momentum cannot be recovered from equilibrium rho/u. g = rng.uniform(.03, .15, (9, *leaf.shape)).astype(dtype) g += np.arange(9, dtype=dtype)[:, None, None] * .01 g[:, ~leaf] = np.nan result.append(g) return result def remap(old, new, g, **kwargs): options = dict(base_count=7, level_steps=(7, 14, 28), phase='postcollision_g', origin=(-3., 2.), spacing=1.25) options.update(kwargs) return remap_synchronized(old, new, g, **options) def integrals(mesh, g, spacing=1.25): populations = sum(a[:, mask].sum(axis=1, dtype=np.float64) * spacing**2 / 4**level for level, (a, mask) in enumerate(zip(g, mesh.leaf_masks))) return np.r_[populations, populations.sum(), populations @ D2Q9_EX, populations @ D2Q9_EY] class SynchronizedRemapTests(unittest.TestCase): def setUp(self): self.old, self.new = topology(A), topology(B) self.g = fields(self.old) def test_identity_bitwise_and_no_mutation_or_count_advance(self): copies = [a.copy() for a in self.g] flags = [a.flags.writeable for a in self.g] out = remap(self.old, self.old, self.g) self.assertEqual((out.base_count, out.level_steps, out.phase, out.origin, out.spacing), (7, (7, 14, 28), 'postcollision_g', (-3., 2.), 1.25)) for a, b, saved, leaf, flag in zip(self.g, out.g, copies, self.old.leaf_masks, flags): self.assertEqual(a[:, leaf].tobytes(), b[:, leaf].tobytes()) np.testing.assert_array_equal(a, saved) self.assertEqual(a.flags.writeable, flag) self.assertFalse(np.shares_memory(a, b)) self.assertTrue(np.isnan(b[:, ~leaf]).all()) def test_shift_preserves_physical_overlap_not_local_patch_index(self): out = remap(self.old, self.new, self.g) for level in range(3): overlap = self.old.leaf_masks[level] & self.new.leaf_masks[level] self.assertEqual(self.g[level][:, overlap].tobytes(), out.g[level][:, overlap].tobytes()) y, x = np.nonzero(self.new.leaf_masks[2] & ~self.old.leaf_masks[2]) np.testing.assert_array_equal(out.g[2][:, y, x], self.g[1][:, y//2, x//2]) y, x = np.nonzero(self.new.leaf_masks[1] & ~self.old.leaf_masks[1]) expected = sum(self.g[2][:, 2*y+dy, 2*x+dx] * .25 for dy in (0, 1) for dx in (0, 1)) np.testing.assert_array_equal(out.g[1][:, y, x], expected) self.assertGreater(np.max(np.abs(expected[0] - self.g[2][0, 2*y, 2*x])), .001) def test_fp64_and_fp32_area_population_mass_momentum_conservation(self): for dtype, tolerance in ((np.float64, 2e-13), (np.float32, 2e-7)): g = fields(self.old, dtype) out = remap(self.old, self.new, g) before, after = integrals(self.old, g), integrals(self.new, out.g) # FP64 accumulation; absolute scale handles near-cancelled momentum. scale = np.sum(np.abs(before[:9])) np.testing.assert_allclose(after, before, rtol=0, atol=tolerance * scale) for array, leaf in zip(out.g, self.new.leaf_masks): self.assertEqual(array.dtype, dtype) self.assertTrue(np.isfinite(array[:, leaf]).all()) self.assertTrue(np.isnan(array[:, ~leaf]).all()) def test_layout_and_rectangle_permutation_independence(self): old = topology(A + [(56, 16, 80, 40)]) new = topology(B + [(56, 16, 80, 40)]) reordered = topology(list(reversed(B + [(56, 16, 80, 40)])), FixedPatchSet2D.from_config(CFG, list(reversed(PARENTS)))) g = fields(old) expected = remap(old, new, g) for block in ((4, 4), (2, 2)): layouts = (BlockLayout2D.build(CFG.ny, CFG.nx, *block), old.parent.fine_layout(block), old.child.fine_layout(block)) blocked = [layout.to_blocked(a) for layout, a in zip(layouts, g)] # Corrupt all block halos; only owner-core populations are admissible. for layout, a in zip(layouts, blocked): h = layout.halo a[:, :, :h, :] = np.inf a[:, :, -h:, :] = np.inf a[:, :, :, :h] = np.inf a[:, :, :, -h:] = np.inf saved = [a.copy() for a in blocked] actual = remap(old, reordered, blocked, layouts=layouts) for a, b in zip(expected.g, actual.g): np.testing.assert_array_equal(a, b) for a, b in zip(saved, blocked): np.testing.assert_array_equal(a, b) def test_shifted_parent_physical_frame(self): # L1 window origin=(2,2) in root units; child coordinates remain parent-local. cfg = LBMConfig(nx=32, ny=32, viscosity=.1) parent = FixedPatch2D.from_config(cfg, (4, 4, 28, 28)) old, new = topology(A, parent, cfg), topology(B, parent, cfg) g = fields(old) out = remap(old, new, g, origin=(10., -7.)) self.assertEqual(out.topology.parent.fine_origin, (2., 2.)) overlap = old.leaf_masks[2] & new.leaf_masks[2] np.testing.assert_array_equal(out.g[2][:, overlap], g[2][:, overlap]) np.testing.assert_allclose(integrals(old, g), integrals(new, out.g), rtol=2e-13, atol=2e-12) def test_outside_leaf_nan_inf_ignored_required_leaf_rejected(self): for level in range(3): dirty = [a.copy() for a in self.g] dirty[level][:, ~self.old.leaf_masks[level]] = np.inf actual = remap(self.old, self.new, dirty) expected = remap(self.old, self.new, self.g) for a, b in zip(expected.g, actual.g): np.testing.assert_array_equal(a, b) # Specifically test restriction children, prolongation donors, and overlap. required = [(0, self.old.leaf_masks[0]), (1, self.old.leaf_masks[1] & ~self.new.leaf_masks[1]), (2, self.old.leaf_masks[2] & ~self.new.leaf_masks[2])] for level, mask in required: y, x = np.argwhere(mask)[0] for bad in (np.nan, np.inf, -np.inf): dirty = [a.copy() for a in self.g] dirty[level][0, y, x] = bad with self.assertRaisesRegex(ValueError, 'required leaf donors'): remap(self.old, self.new, dirty) dirty = [a.copy() for a in self.g] dirty[0][:, 0, 0] = -1 with self.assertRaisesRegex(ValueError, 'positive density'): remap(self.old, self.new, dirty) def test_absent_required_block_rejected(self): layouts = (None, None, BlockLayout2D.build(*self.old.leaf_masks[2].shape, 4, 4, periodic=(False, False), origins=[(0, 0)])) arrays = [*self.g[:2], layouts[2].to_blocked(self.g[2])] with self.assertRaisesRegex(ValueError, 'own every required leaf'): remap(self.old, self.new, arrays, layouts=layouts) def test_coarse_fine_coarse_identity_not_general_inverse(self): # Grow then shrink: newly refined cells copy one parent value, so means # recover it exactly. Shrink then grow loses arbitrary child variation. wide = topology([(16, 16, 36, 40)]) refined = remap(self.old, wide, self.g) restored = remap(wide, self.old, refined.g) for a, b in zip(self.g, restored.g): np.testing.assert_array_equal(a, b) shifted = remap(self.old, self.new, self.g) returned = remap(self.new, self.old, shifted.g) lost = self.old.leaf_masks[2] & ~self.new.leaf_masks[2] self.assertFalse(np.array_equal(self.g[2][:, lost], returned.g[2][:, lost])) def test_invalid_geometry_counts_representation_frame(self): for rectangles in ([], [(2, 2, 10, 10)], [(16, 16, 33.5, 40)]): with self.assertRaises(ValueError): topology(rectangles) parent = self.old.parent for changes in (dict(lattice_model='D3Q19'), dict(collision='TRT'), dict(les_enabled=True), dict(nz=2)): with self.assertRaises(ValueError): RemapTopology.from_config(replace(CFG, **changes), parent, A) for kwargs in (dict(coarse_boundary='slot_reservoir'), dict(objects=(object(),))): with self.assertRaisesRegex(ValueError, 'periodic 2D'): RemapTopology.from_config(CFG, parent, A, **kwargs) different = topology(A, cfg=replace(CFG, viscosity=.12)) with self.assertRaisesRegex(ValueError, 'unchanged'): remap(self.old, different, self.g) for kwargs in (dict(base_count=True), dict(base_count=-1), dict(level_steps=(7, 14, 27)), dict(level_steps=(7, 14)), dict(level_steps=None), dict(phase='arrival_f'), dict(origin=(np.nan, 0)), dict(spacing=0), dict(spacing='bad')): with self.assertRaises(ValueError): remap(self.old, self.new, self.g, **kwargs) for g in (self.g[:2], [self.g[0].astype(np.float32), *self.g[1:]], [self.g[0][0], *self.g[1:]]): with self.assertRaises(ValueError): remap(self.old, self.new, g) if __name__ == '__main__': unittest.main()