From afa8f6aca10177d5901a9aeae650d3419212c6d0 Mon Sep 17 00:00:00 2001 From: TRON Date: Sun, 18 Oct 2026 23:48:18 -0700 Subject: [PATCH] test: add bounded square shear orientation diagnostic --- docs/validation/taylor_green.md | 40 ++++ tests/validation/taylor_green/run_case.py | 60 ++++-- .../taylor_green/shear_axis_square.py | 136 ++++++++++++++ .../taylor_green/test_axis_square.py | 172 ++++++++++++++++++ 4 files changed, 394 insertions(+), 14 deletions(-) create mode 100644 tests/validation/taylor_green/shear_axis_square.py create mode 100644 tests/validation/taylor_green/test_axis_square.py diff --git a/docs/validation/taylor_green.md b/docs/validation/taylor_green.md index 61bd23e..e9dabaf 100644 --- a/docs/validation/taylor_green.md +++ b/docs/validation/taylor_green.md @@ -508,3 +508,43 @@ receipts as a startup-state counterfactual. The host first-interval pressure RMS reduction is not a prediction of long-time improvement, and individual probe errors can increase. All equilibrium paths and historical verdicts stay unchanged. + +### Short square orientation diagnostic (`axis_square_shear_n128`) + +Four fresh cases only: U128, U256, constant R128 and projected R128 depth 2. +Use `--flow shear_wave --diagnostic axis_square_shear_n128 --probe-trace`. +The supported union is `[[(32,32,96,96)]]`; a full-height periodic fine strip +is not supported by the unchanged production geometry. All four cases use +POST G = feq - (tau - 1) A, mode (2,0), polarization (0,1), phase pi/4, +U=.03 and Re100. Root/fine nu are .0384/.0768 and tau are .6152/.7304. +This first-gradient low-Mach initialization is not an exact discrete mode. + +Stop at root step 8: two global rows at 0/8, nine local rows at 0..8; +U256 advances two native steps per root step. Refined leaves are [12288,16384], +hops [8,16], work 360,448 per refined case. Four-case work is 1,900,544 ACTIVE +updates, six build sidecars and 720 native cell-time samples. Existing runtime, +build/source guards and receipt/trace caps apply. No snapshots or late fit. +`screening_pass=null`, `qualification_eligible=false`: exit zero means valid +finite completion, with no physical pass threshold or legacy assessor. + +Eight root-unit-cell groups (i,j): inner L/R/B/T = (32,64)/(95,63)/(64,32)/(63,95), +outer L/R/B/T = (31,64)/(96,63)/(64,31)/(63,96). Refined inner groups contain four +fine cells, outer groups one coarse cell; U128 uses one and U256 four per group. +Retain raw rho/ux/uy, native indices, physical centers/areas and parity. Group +residuals are signed means of rho-1 and rho*u-u_exact, evaluated at each actual +center; normalization uses that group's mean analytic d(uy)/dx. Only inner +L/B and R/T pairs have exactly opposite equal-magnitude gradients. Controls +have their own native quadrature. Probe density is ungauged; global pressure +RMS uses one domain gauge. + +Actual-table local corridors match tangentially extended straight faces beyond +22.5 root cells. The conservative paired-map radius is 2.5 root cells per root +step, so these selected groups through step 8 are locally straight-face +equivalent. This is not a claim about global corner errors. For projected +explosion with actual missing-normal-gradient fallback, the stationary +linearized affine reference after the first complete root interval predicts +zero conserved-moment error on vertical faces. At the +bottom/top inner fine quartets its mean delta jx / d(uy)/dx is +-/+ 1/(576*tau_f), with opposite adjacent-coarse redistribution. +Keep raw values, step-zero rounding and fresh controls: sinusoidal curvature +and finite-Mach effects prevent treating this coefficient as an exact gate. diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py index 9d580c1..1ef60fa 100644 --- a/tests/validation/taylor_green/run_case.py +++ b/tests/validation/taylor_green/run_case.py @@ -27,6 +27,8 @@ TWIN_SHEAR = "twin_shear_n128" # depth-2 twin: nested L0->L1 boxes without L2 SHEAR_TWIN_SHA256 = "1cbf37f18e89aea577581631c4565f51e68d22a3739b698816ef64acbf2b2bb4" WIDE_X_SHEAR = "wide_x_shear_n128" # one projected x-gap-width contrast SHEAR_WIDE_X_SHA256 = "13d16ba355117e2d074edccd9dc5d694c2fc13c673a0abdfca6401f6ed917733" +AXIS_SQUARE = "axis_square_shear_n128" # short four-face orientation diagnostic +SHEAR_AXIS_SQUARE_SHA256 = "0baffa6eb99c268ce998d90300c993d45bb3b7b27c524b636874d75808bd0a8c" POSTG_SHEAR = "postg_shear_n128" # initialization-only three-case counterfactual SHEAR_POSTG_SHA256 = "9c909fa013c092490ee21532e3125b2a78a4c5102c7a1bdc0e76986d3f0ef903" SHEAR_PROBES_SHA256 = "e2e6490d281cc275f448dc6d6ea7b9b4407cc74a51e9256b86c08c8626b89b76" # signed interface-probe mechanism diagnostic @@ -125,6 +127,13 @@ def verify_source(repo, revision, flow="taylor_green", probes=False): or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_PROBES_SHA256): raise RuntimeError("Wrong shear_probes module checkout or content") result["shear_probes_sha256"] = SHEAR_PROBES_SHA256 + if flow == "shear_axis_square": + import shear_axis_square + module = harness / "shear_axis_square.py" + if (Path(shear_axis_square.__file__).resolve() != module + or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_AXIS_SQUARE_SHA256): + raise RuntimeError("Wrong shear_axis_square module checkout or content") + result["shear_axis_square_sha256"] = SHEAR_AXIS_SQUARE_SHA256 if flow == "shear_postg": import shear_postg module = harness / "shear_postg.py" @@ -153,7 +162,7 @@ def verify_source(repo, revision, flow="taylor_green", probes=False): or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_NESTED_SHA256): raise RuntimeError("Wrong shear_nested module checkout or content") result["shear_nested_sha256"] = SHEAR_NESTED_SHA256 - if flow in ("shear_wave", "shear_nested", "shear_twin", "shear_wide_x", "shear_postg"): + if flow in ("shear_wave", "shear_nested", "shear_twin", "shear_wide_x", "shear_postg", "shear_axis_square"): import shear_wave module = harness / "shear_wave.py" if (Path(shear_wave.__file__).resolve() != module @@ -228,16 +237,18 @@ def main(): p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant") p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC, SCALED_SHEAR, NESTED_SHEAR, TWIN_SHEAR, - WIDE_X_SHEAR, POSTG_SHEAR], + WIDE_X_SHEAR, POSTG_SHEAR, AXIS_SQUARE], help="explicit four-case N256/fine512 same-placement diagnostic " "(scaled_n256: Taylor-Green; scaled_shear_n256: shear wave) or the " "fixed three-level/two-region shear preset (nested_shear_n128) " "or its depth-2 twin (twin_shear_n128), or the single projected " "x-gap-width contrast (wide_x_shear_n128), or three-case POST-G " - "initialization counterfactual (postg_shear_n128)") + "initialization counterfactual (postg_shear_n128), or short square " + "orientation diagnostic (axis_square_shear_n128)") p.add_argument("--probe-trace", action="store_true", help="signed interface-probe trace (nested U128/U512/projected d3, " - "twin projected d2, wide-x projected d3, POST-G three-case set); " + "twin projected d2, wide-x projected d3, POST-G three-case set, " + "axis-square four-case set); " "mechanism diagnostic, no gates") p.add_argument("--final-snapshot", action="store_true", help="nested_shear_n128 projected depth-3 only: final macro snapshot") @@ -245,10 +256,11 @@ def main(): help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64") p.add_argument("--flow", choices=["taylor_green", "shear_wave"], default="taylor_green") args = p.parse_args() + axis_square = args.diagnostic == AXIS_SQUARE nested = args.diagnostic == NESTED_SHEAR wide_x = args.diagnostic == WIDE_X_SHEAR postg = args.diagnostic == POSTG_SHEAR - if (wide_x or postg) and not args.probe_trace: + if (wide_x or postg or axis_square) and not args.probe_trace: p.error(f"{args.diagnostic} requires --probe-trace") if args.flow == "shear_wave" and (args.re != 100 or (args.depth == 3 and not (nested or wide_x or postg)) or args.early_pressure): @@ -259,9 +271,13 @@ def main(): (WIDE_X_SHEAR, 128, 3, "projected_linear"), (POSTG_SHEAR, 128, 1, "constant"), (POSTG_SHEAR, 512, 1, "constant"), (POSTG_SHEAR, 128, 3, "projected_linear")} + if axis_square: + import shear_axis_square + probe_cases.update((AXIS_SQUARE, n, depth, scheme) + for _, n, depth, scheme in shear_axis_square.CASES) if args.probe_trace and (args.final_snapshot or ( args.diagnostic, args.n, args.depth, args.explosion) not in probe_cases): - p.error("--probe-trace requires a fixed nested, twin, wide-x or POST-G case, " + p.error("--probe-trace requires a fixed nested, twin, wide-x, POST-G or axis-square case, " "without --final-snapshot") if args.final_snapshot and not (nested and args.depth == 3 and args.explosion == "projected_linear"): @@ -272,7 +288,12 @@ def main(): if args.explosion != "constant" and args.depth != 2 and not (nested or wide_x or postg): p.error("projected_linear requires depth 2") twin = args.diagnostic == TWIN_SHEAR - if nested or twin or wide_x or postg: + if axis_square: + preset = shear_axis_square + if args.flow != "shear_wave" or (args.re, args.n, args.depth, args.explosion) not in preset.CASES: + p.error("axis_square_shear_n128 requires one of its four Re100 shear cases") + initial, metrics, limits, initialization = preset.initial, preset.metrics, {}, preset.INITIALIZATION + elif nested or twin or wide_x or postg: import shear_nested import shear_twin import shear_wide_x @@ -304,7 +325,7 @@ def main(): p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required") repo = args.repo.resolve() sys.path[:0] = [str(repo / "src"), str(repo)] - source_flow = ("shear_postg" if postg else "shear_wide_x" if wide_x else "shear_nested" if nested + source_flow = ("shear_axis_square" if axis_square else "shear_postg" if postg else "shear_wide_x" if wide_x else "shear_nested" if nested else "shear_twin" if twin else args.flow) source = verify_source(repo, args.expected_revision, source_flow, probes=args.probe_trace) runtime = runtime_source(repo) @@ -322,7 +343,7 @@ def main(): cfg = config(args.n, args.re) if args.depth == 1: refinements = [] - elif nested or twin or wide_x or postg: + elif nested or twin or wide_x or postg or axis_square: refinements = [list(level) for level in preset.REFINEMENTS] else: refinements = rectangles(args.n, args.depth) @@ -330,6 +351,8 @@ def main(): np.testing.assert_array_equal(leaf_partition(h), np.ones((args.n, args.n))) masks = [k == ACTIVE for k in h.kinds] main_steps, early_steps = sample_steps(args.n, args.early_pressure) + if axis_square: + main_steps = [0, preset.case_steps(args.n)[-1]] report = dict( schema=1, source=source, @@ -357,7 +380,7 @@ def main(): cuda_visible_devices=selected, device_id=0, limits=limits, - workload=workload(masks, round(TIMES[-1] * args.n / U)), + workload=workload(masks, main_steps[-1]), rows=[], early_pressure=dict(enabled=args.early_pressure, rows=[], sampled_max_pressure_l2=None), @@ -365,14 +388,19 @@ def main(): ) if args.flow != "taylor_green": import shear_wave - report.update(flow=args.flow, mode=list(shear_wave.MODE), - polarization=shear_wave.E.tolist(), angle_deg=shear_wave.ANGLE_DEG, - phase=shear_wave.PHI, eps=shear_wave.eps(args.n)) + wave = preset if axis_square else shear_wave + report.update(flow=args.flow, mode=list(wave.MODE), + polarization=wave.E.tolist(), angle_deg=wave.ANGLE_DEG, + phase=wave.PHI, eps=wave.eps(args.n)) if args.diagnostic: report["diagnostic"] = args.diagnostic if postg: report.update(initialization_id=preset.INITIALIZATION_ID, qualification_eligible=False, screening_scope="descriptive scalar screens only; no qualification assessor") + if axis_square: + report.update(initialization_id=preset.INITIALIZATION_ID, qualification_eligible=False, + reference_harness=None, historical_baseline_revision=None, + screening_scope="no numerical gates; short orientation diagnostic") args.output.parent.mkdir(parents=True, exist_ok=True) def save(): temporary = args.output.with_suffix(args.output.suffix + ".tmp") @@ -410,6 +438,8 @@ def main(): probe_steps = set() if args.probe_trace: import shear_probes + if axis_square: + shear_probes = preset probe_root = dict(zip(shear_probes.case_steps(args.n), shear_probes.STEPS)) probe_steps = set(probe_root) probes = shear_probes.select(masks, args.n) @@ -491,7 +521,7 @@ def main(): report.update( completed=True, seconds=time.monotonic() - started, - screening_pass=all( + screening_pass=None if axis_square else all( row[key] <= limit for row in report["rows"] for key, limit in limits.items() ), ) @@ -502,6 +532,8 @@ def main(): ), flush=True, ) + if axis_square: + return int("error" in report["probe_trace"]) return 0 if report["screening_pass"] else 1 diff --git a/tests/validation/taylor_green/shear_axis_square.py b/tests/validation/taylor_green/shear_axis_square.py new file mode 100644 index 0000000..90fe0a8 --- /dev/null +++ b/tests/validation/taylor_green/shear_axis_square.py @@ -0,0 +1,136 @@ +"""Short axis-shear orientation diagnostic; no physical acceptance gates.""" + +import json +import numpy as np + +import shear_wave as sw +from shear_probes import sample, validate_row, TRACE_MAX_BYTES +from oracle import CX, CY, W, U, centers + +ROOT_N = 128 +MODE, E, PHI, ANGLE_DEG = (2, 0), np.array([0., 1.]), np.pi / 4, 0. +REFINEMENTS = [[(32, 32, 96, 96)]] +CASES = ((100, 128, 1, "constant"), (100, 256, 1, "constant"), + (100, 128, 2, "constant"), (100, 128, 2, "projected_linear")) +STEPS = tuple(range(9)) +INITIALIZATION_ID = "postg_first_gradient_v1" +INITIALIZATION = "physical FP64 axis-shear POST G=feq-(tau-1)A; first-gradient low-Mach CE" +GROUPS = dict(inner_left=(32, 64), inner_right=(95, 63), + inner_bottom=(64, 32), inner_top=(63, 95), + outer_left=(31, 64), outer_right=(96, 63), + outer_bottom=(64, 31), outer_top=(63, 96)) + + +def eps(n): + return (4 * np.pi / n)**2 + + +def exact(x, t, length, viscosity): + """Transverse velocity and physical x derivative in the supplied length units.""" + k = 4 * np.pi / length + amplitude = U * np.exp(-viscosity * k * k * t) + return amplitude * np.sin(k * x + PHI), amplitude * k * np.cos(k * x + PHI) + + +def initial(shape, level, length, viscosity): + x, _ = centers(shape, level) + uy, gradient = exact(x, 0, length, viscosity) + tau = .5 + 3 * viscosity * 2**level + a = (3 * W * CX * CY)[:, None, None] * gradient * 2.**-level + return sw.equilibrium(np.ones_like(uy), np.zeros_like(uy), uy) - (tau - 1) * a + + +def metrics(arrays, masks, t, length, viscosity, initial_mass=None): + rho, ux, uy, x, _, w, _ = sw._leaves(arrays, masks) + if not np.isclose(w.sum(), length**2, rtol=0, atol=1e-10): + raise ValueError("leaf partition does not cover domain once") + ey, _ = exact(x, t, length, viscosity) + p = (rho - 1) / 3 + p -= w @ p / w.sum() # one domain gauge, never the probe mean + mass = float(w @ rho) + return dict(tstar=t * U / length, mass=mass, + mass_relative=0. if initial_mass is None else abs(mass-initial_mass)/initial_mass, + velocity_l2=float(np.sqrt(w @ (ux**2 + (uy-ey)**2) / (w @ ey**2))), + spurious_pressure_rms=float(np.sqrt(w @ p**2 / w.sum())), + rho_min=float(rho.min()), rho_max=float(rho.max()), + max_speed=float(np.hypot(ux, uy).max()), leaf_cells=len(rho)) + + +def case_steps(n): + return [r * (n // ROOT_N) for r in STEPS] + + +def select(masks, n=ROOT_N): + """Exact native cells filling eight unit-area groups; no interpolated samples.""" + if (n, len(masks)) not in ((128, 1), (256, 1), (128, 2)): + raise ValueError("unsupported square probe layout") + probes = [] + for name, (x, y) in GROUPS.items(): + level = int(len(masks) == 2 and name.startswith("inner")) + factor = (n // ROOT_N) * 2**level + for dy in range(factor): + for dx in range(factor): + i, j = factor*x + dx, factor*y + dy + if not masks[level][j, i]: + raise ValueError("square probe is not ACTIVE") + probes.append(dict(group=name, level=level, i=i, j=j, + center=[(i+.5)/factor, (j+.5)/factor], + width=1/factor, area=1/factor**2, parity=[i % 2, j % 2])) + if len({(p["level"], p["i"], p["j"]) for p in probes}) != len(probes): + raise ValueError("duplicate square probes") + return probes + + +def new_trace(probes, case_viscosity, case): + return dict(schema=1, kind="axis-square orientation diagnostic; no pass gates", case=case, + time_unit="root step (N=128)", sample_phase="postcollision G at root synchronization", + root_n=ROOT_N, viscosity_root=U*ROOT_N/100, case_viscosity=case_viscosity, + mode=list(MODE), phase=PHI, amplitude=U, fields=["rho", "ux", "uy"], + probes=probes, steps=[], values=[]) + + +def group_moments(trace): + """Signed unit-area means; normalize by each group's mean actual-center derivative. + + Momentum is rho*u. Raw density is ungauged. Per-cell parity stays in the trace. + U128 samples one center, whereas U256/refined-inner use four different centers. + """ + out = [] + for step, row in zip(trace["steps"], trace["values"]): + values, groups = np.asarray(row).reshape(-1, 3), {} + for name in GROUPS: + indices = [i for i, p in enumerate(trace["probes"]) if p["group"] == name] + probes = [trace["probes"][i] for i in indices] + weights = np.array([p["area"] for p in probes]) + if weights.sum() != 1: + raise ValueError("probe group does not have unit area") + rho, ux, uy = values[indices].T + ey, gradient = exact(np.array([p["center"][0] for p in probes]), step, + ROOT_N, trace["viscosity_root"]) + drho, djx, djy = (float(weights @ v) for v in (rho-1, rho*ux, rho*uy-ey)) + derivative = float(weights @ gradient) + groups[name] = dict(drho=drho, djx=djx, djy=djy, mean_duy_dx=derivative, + djx_over_duy_dx=djx/derivative, djy_over_duy_dx=djy/derivative) + out.append(dict(root_step=step, groups=groups)) + return out + + +def validate(trace): + if trace["steps"] != list(STEPS) or len(trace["values"]) != len(STEPS): + raise ValueError("square trace schedule mismatch") + _, n, depth, scheme = trace["case"] + if (100, n, depth, scheme) not in CASES or len(trace["probes"]) != ( + 20 if depth == 2 else 8*(n//ROOT_N)**2): + raise ValueError("square trace case/count mismatch") + for row in trace["values"]: + validate_row(row, len(trace["probes"])) + trace["group_moments"] = group_moments(trace) + size = 0 + while True: + final = len(json.dumps(dict(trace, bytes=size), allow_nan=False).encode()) + if final == size: + break + size = final + if size > TRACE_MAX_BYTES: + raise ValueError("square trace exceeds byte cap") + return size diff --git a/tests/validation/taylor_green/test_axis_square.py b/tests/validation/taylor_green/test_axis_square.py new file mode 100644 index 0000000..1666a66 --- /dev/null +++ b/tests/validation/taylor_green/test_axis_square.py @@ -0,0 +1,172 @@ +"""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)