173 lines
9.8 KiB
Python
173 lines
9.8 KiB
Python
"""Fixed square geometry, native sampling, POST phase and runner admission; CPU only."""
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from contextlib import redirect_stdout
|
|
|
|
import numpy as np
|
|
import run_case as rc
|
|
import shear_axis_square as axis
|
|
import shear_wave as sw
|
|
import test_shear_wave as legacy
|
|
from oracle import CX, CY, W, U, centers
|
|
from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition
|
|
from CelerisLab.lbm.refinement import ACTIVE
|
|
|
|
REPO = Path(__file__).resolve().parents[3]
|
|
|
|
|
|
def masks(n, depth):
|
|
h = build_hierarchy(rc.config(n, 100), [] if depth == 1 else axis.REFINEMENTS)
|
|
np.testing.assert_array_equal(leaf_partition(h), np.ones((n, n)))
|
|
return [k == ACTIVE for k in h.kinds]
|
|
|
|
|
|
class AxisSquareTests(unittest.TestCase):
|
|
def test_geometry_native_groups_and_work(self):
|
|
work = 0
|
|
for _, n, depth, _ in axis.CASES:
|
|
m = masks(n, depth)
|
|
p = axis.select(m, n)
|
|
self.assertEqual(len(p), 20 if depth == 2 else 8*(n//128)**2)
|
|
self.assertEqual(len({(v['level'], v['i'], v['j']) for v in p}), len(p))
|
|
roots = [(32, 64), (95, 63), (64, 32), (63, 95),
|
|
(31, 64), (96, 63), (64, 31), (63, 96)]
|
|
expected = []
|
|
for group, (x, y) in enumerate(roots):
|
|
level = 1 if depth == 2 and group < 4 else 0
|
|
factor = (n//128) * 2**level
|
|
expected += [(level, factor*x+dx, factor*y+dy,
|
|
[x+(dx+.5)/factor, y+(dy+.5)/factor])
|
|
for dy in range(factor) for dx in range(factor)]
|
|
self.assertEqual([(v['level'], v['i'], v['j'], v['center']) for v in p], expected)
|
|
for name in axis.GROUPS:
|
|
group = [v for v in p if v['group'] == name]
|
|
self.assertEqual(sum(v['area'] for v in group), 1)
|
|
np.testing.assert_array_equal(np.mean([v['center'] for v in group], axis=0),
|
|
np.array(axis.GROUPS[name])+.5)
|
|
for v in group:
|
|
self.assertTrue(m[v['level']][v['j'], v['i']])
|
|
w = rc.workload(m, axis.case_steps(n)[-1])
|
|
work += w['active_collision_updates']
|
|
if depth == 2:
|
|
self.assertEqual(w['active_cells_by_level'], [12288, 16384])
|
|
self.assertEqual(w['level_hops'], [8, 16])
|
|
self.assertEqual(axis.case_steps(n), [r*(n//128) for r in range(9)])
|
|
self.assertEqual(work, 1900544)
|
|
|
|
def test_postg_moments_stress_and_matching_fine_control(self):
|
|
for level in (0, 1):
|
|
x, _ = centers((33, 41), level)
|
|
k, h, tau = 4*np.pi/128, 2.**-level, .5+3*.0384*2**level
|
|
uy = U*np.sin(k*x+np.pi/4)
|
|
eq = sw.equilibrium(np.ones_like(x), np.zeros_like(x), uy)
|
|
g = axis.initial(x.shape, level, 128, .0384)
|
|
a = 3*W[:, None, None]*CX[:, None, None]*CY[:, None, None]*h*k*U*np.cos(k*x+np.pi/4)
|
|
np.testing.assert_allclose(g, eq-(tau-1)*a, rtol=0, atol=6e-17)
|
|
for v in (np.ones(9), CX, CY):
|
|
np.testing.assert_allclose(np.einsum('q,qyx->yx', v, g-eq), 0, rtol=0, atol=2e-16)
|
|
np.testing.assert_allclose(np.einsum('q,qyx->yx', CX*CY, g-eq),
|
|
(1-tau)*h*k*U*np.cos(k*x+np.pi/4)/3, atol=6e-17)
|
|
self.assertGreater(g.min(), 0)
|
|
np.testing.assert_array_equal(axis.initial((256, 256), 0, 256, .0768),
|
|
axis.initial((256, 256), 1, 128, .0384))
|
|
|
|
def test_signed_momentum_and_actual_center_time(self):
|
|
m = masks(128, 2)
|
|
probes = axis.select(m)
|
|
trace = axis.new_trace(probes, .0384, [rc.AXIS_SQUARE, 128, 2, 'projected_linear'])
|
|
for step in axis.STEPS:
|
|
arrays = []
|
|
for level, mask in enumerate(m):
|
|
x, _ = centers(mask.shape, level)
|
|
uy, _ = axis.exact(x, step, 128, .0384)
|
|
arrays.append(sw.equilibrium(np.full_like(x, 1.001), np.zeros_like(x), uy))
|
|
trace['steps'].append(step)
|
|
trace['values'].append(axis.sample(arrays, probes))
|
|
trace['bytes'] = axis.validate(trace)
|
|
self.assertEqual(trace['bytes'], len(json.dumps(trace, allow_nan=False).encode()))
|
|
for row in trace['group_moments']:
|
|
for name, v in row['groups'].items():
|
|
self.assertAlmostEqual(v['drho'], .001, places=14)
|
|
x = np.array([p['center'][0] for p in probes if p['group'] == name])
|
|
uy, gradient = axis.exact(x, row['root_step'], 128, .0384)
|
|
self.assertAlmostEqual(v['djy'], .001*uy.mean(), places=14)
|
|
self.assertAlmostEqual(v['mean_duy_dx'], gradient.mean(), places=14)
|
|
self.assertAlmostEqual(row['groups']['inner_left']['mean_duy_dx'],
|
|
-row['groups']['inner_bottom']['mean_duy_dx'], places=14)
|
|
with patch.object(axis, 'TRACE_MAX_BYTES', trace['bytes']-1), self.assertRaises(ValueError):
|
|
axis.validate(trace)
|
|
for mutate in (lambda t: t['steps'].pop(),
|
|
lambda t: t['steps'].__setitem__(1, 0),
|
|
lambda t: t['values'][1].__setitem__(0, float('nan')),
|
|
lambda t: t['probes'].pop()):
|
|
bad = deepcopy(trace); mutate(bad)
|
|
with self.assertRaises(ValueError): axis.validate(bad)
|
|
|
|
def test_four_full_mock_schedules_and_old_rejection(self):
|
|
helper = legacy.ProbeTraceTests()
|
|
helper.setUp()
|
|
def arrays(m, n, t):
|
|
return [axis.initial(v.shape, level, n, U*n/100) for level, v in enumerate(m)]
|
|
for _, n, depth, scheme in axis.CASES:
|
|
with patch.object(helper, 'exact_arrays', side_effect=arrays):
|
|
r, counts, files = helper.run_mocked(n, depth, rc.AXIS_SQUARE, scheme,
|
|
masks(n, depth), initializer=axis.initial)
|
|
self.assertEqual(counts[-1], 8*(n//128))
|
|
self.assertEqual([v['step'] for v in r['rows']], [0, 8*(n//128)])
|
|
self.assertEqual(r['probe_trace']['steps'], list(range(9)))
|
|
self.assertEqual(r['probe_trace']['probes'], axis.select(masks(n, depth), n))
|
|
self.assertEqual(r['mode'], [2, 0])
|
|
self.assertEqual(r['phase'], np.pi/4)
|
|
self.assertEqual(r['initialization'], axis.INITIALIZATION)
|
|
self.assertEqual(r['limits'], {})
|
|
self.assertIsNone(r['screening_pass'])
|
|
self.assertFalse(r['qualification_eligible'])
|
|
self.assertEqual(files, ['probe.json'])
|
|
old = legacy.NestedPresetTests(); old.setUp()
|
|
records = old.nested_set(); records[-1]['diagnostic'] = rc.AXIS_SQUARE
|
|
self.assertEqual(old.mod.assess_nested(records, 'r')['status'], 'INVALID_SET')
|
|
records = [legacy.receipt(128, 1), legacy.receipt(256, 1), legacy.receipt(128, 2),
|
|
legacy.receipt(128, 2, 'projected_linear')]
|
|
records[-1].update(diagnostic=rc.AXIS_SQUARE, initialization=axis.INITIALIZATION)
|
|
self.assertEqual(old.mod.assess(records, 'r')['status'], 'INVALID_SET')
|
|
self.assertNotIn(rc.AXIS_SQUARE, old.mod.PRESETS)
|
|
with patch.object(axis, 'sample', return_value=[float('nan')]*24):
|
|
r, _, _ = helper.run_mocked(128, 1, rc.AXIS_SQUARE, 'constant', masks(128, 1),
|
|
initializer=axis.initial)
|
|
self.assertIn('error', r['probe_trace'])
|
|
self.assertTrue(r['completed'])
|
|
self.assertEqual(len(r['rows']), 2)
|
|
|
|
def test_admission_and_module_pin(self):
|
|
from tests.validation.multigrid import check_projected_pairs as gate
|
|
self.assertEqual(hashlib.sha256(Path(axis.__file__).read_bytes()).hexdigest(), rc.SHEAR_AXIS_SQUARE_SHA256)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
base = ['run_case.py', '--repo', str(REPO), '--expected-revision', 'f'*40,
|
|
'--output', str(Path(tmp)/'case.json'), '--dry-run', '--re', '100',
|
|
'--flow', 'shear_wave', '--diagnostic', rc.AXIS_SQUARE, '--probe-trace']
|
|
for n, depth, scheme, valid in [(n, d, s, True) for _, n, d, s in axis.CASES] + [
|
|
(512, 1, 'constant', False), (256, 2, 'projected_linear', False), (128, 3, 'constant', False)]:
|
|
for extra in ([], ['--final-snapshot'], ['--early-pressure'], ['--flow', 'taylor_green'], ['--re', '1000']):
|
|
argv = base + ['--n', str(n), '--depth', str(depth), '--explosion', scheme] + extra
|
|
with patch.object(sys, 'argv', argv), redirect_stdout(io.StringIO()), patch('sys.stderr', io.StringIO()), \
|
|
patch.object(rc, 'verify_source', return_value={}) as verify, patch.object(rc, 'runtime_source', return_value={}):
|
|
if valid and not extra:
|
|
self.assertEqual(rc.main(), 0)
|
|
verify.assert_called_once_with(REPO, 'f'*40, 'shear_axis_square', probes=True)
|
|
else:
|
|
with self.assertRaises(SystemExit): rc.main()
|
|
verify.assert_not_called()
|
|
with patch.object(gate, 'preflight', return_value=rc.NUMERICAL_REVISION), \
|
|
patch.object(rc, 'provenance', return_value=dict(source_tree_sha256=rc.SOURCE_SHA256)):
|
|
rc.verify_source(REPO, 'r', 'shear_axis_square', probes=True)
|
|
for module, key, value in ((axis, '__file__', '/wrong/helper.py'), (rc, 'SHEAR_AXIS_SQUARE_SHA256', 'wrong')):
|
|
with patch.object(module, key, value), self.assertRaisesRegex(RuntimeError, 'shear_axis_square'):
|
|
rc.verify_source(REPO, 'r', 'shear_axis_square', probes=True)
|