ClaudeandClaude Opus 5.5
48ee0982c9
Tests load research drivers via importlib under unique module names instead of a bare 'import run' (shared-process collision between the Dütsch and VIV drivers). lifecycle.md qualifies immutable geometry and the host-only get_flags for translating circles; supported_settings.md points to the dense-only moving-circle feature; leontini_viv_predeclare.md is renamed viv_re150_predeclare.md (Re 150 Kolahdouz/Qin case) with references updated. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
"""CPU tests of the predeclared VIV logic (research/viv_re150/viv.py; stdlib only)."""
|
|
import importlib.util
|
|
import math
|
|
from pathlib import Path
|
|
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"viv_logic", Path(__file__).resolve().parents[2] / "research/viv_re150/viv.py")
|
|
viv = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(viv)
|
|
|
|
def _load(rel, name):
|
|
"""Load a research driver module under a unique name (no bare ``import run``)."""
|
|
import sys
|
|
path = Path(__file__).resolve().parents[2] / rel
|
|
sys.path.insert(0, str(path.parent)) # for the driver's own sibling imports
|
|
try:
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
finally:
|
|
sys.path.remove(str(path.parent))
|
|
return mod
|
|
|
|
|
|
D, U = 40, 0.04
|
|
|
|
|
|
def _feed(cyc, amps, period=100):
|
|
"""Feed sine cycles of the given amplitudes (in D); return the first non-None result."""
|
|
n = 0
|
|
for a in amps:
|
|
for i in range(period):
|
|
out = cyc.update(a * D * math.sin(2 * math.pi * (i + 1) / period))
|
|
n += 1
|
|
if out:
|
|
return out, n
|
|
return None, n
|
|
|
|
|
|
def test_structure_exact_update_conserves_free_oscillation_and_equilibrium():
|
|
s = viv.Structure(D, U)
|
|
assert math.isclose(1 / s.f_n, 5000.0) and math.isclose(s.m, 2 * D * D)
|
|
s.y, s.v = 0.3 * D, 0.0
|
|
e0 = 0.5 * s.k * s.y ** 2
|
|
for _ in range(12345):
|
|
s.y, s.v = s.candidate(0.0)
|
|
assert math.isclose(0.5 * s.k * s.y ** 2 + 0.5 * s.m * s.v ** 2, e0, rel_tol=1e-9)
|
|
s.y, s.v = 2.0 / s.k, 0.0
|
|
assert all(math.isclose(a, b, abs_tol=1e-12) for a, b in zip(s.candidate(2.0), (2.0 / s.k, 0.0)))
|
|
assert math.isclose(s.seed_velocity(D) / s.w, viv.ELIGIBLE * D)
|
|
|
|
|
|
def test_constant_amplitude_settles_then_scores_twenty_cycles():
|
|
cyc = viv.Cycles(D)
|
|
out, _ = _feed(cyc, [0.5] * 60)
|
|
assert out == "scored" and len(cyc.window) == 20 and cyc.settled_at == 1
|
|
r = viv.score(cyc.window)
|
|
assert math.isclose(r["A_max"], 0.5, rel_tol=1e-3) and abs(r["offset"]) < 1e-3
|
|
assert r["T_resp"] == 100 and r["within_band"] # |.500 - .520| <= .052
|
|
|
|
|
|
def test_small_cycles_are_not_counted_and_y0_is_not_a_crossing():
|
|
cyc = viv.Cycles(D)
|
|
assert cyc.update(0.0) is None and cyc.log == []
|
|
out, _ = _feed(cyc, [0.005] * 50)
|
|
assert out is None and cyc.counted == 0 and cyc.log and not any(c["eligible"] for c in cyc.log)
|
|
|
|
|
|
def test_scoring_failure_restarts_from_the_failing_block():
|
|
cyc = viv.Cycles(D)
|
|
amps = [0.5] * 20 + [0.55] * 10 + [0.55] * 40 # settle at block 1, block 2 jumps 10 %
|
|
out, _ = _feed(cyc, amps)
|
|
assert out == "scored"
|
|
assert all(math.isclose(c["amp"], 0.55, rel_tol=1e-3) for c in cyc.window)
|
|
|
|
|
|
def test_seed_reset_discards_incomplete_cycle_and_proof():
|
|
cyc = viv.Cycles(D)
|
|
_feed(cyc, [0.5] * 15)
|
|
cyc.update(0.2 * D) # inside an incomplete cycle
|
|
counted = cyc.counted
|
|
cyc.discard_current()
|
|
cyc.reset_proof()
|
|
assert cyc.amps == [] and cyc.blocks == [] and cyc.counted == counted
|
|
|
|
|
|
def test_viv_config_passes_the_real_loader(tmp_path):
|
|
import json
|
|
run = _load("research/viv_re150/run.py", "viv_run")
|
|
from CelerisLab.config import load_lbm_config
|
|
cfg_dict, nx, ny = run.config()
|
|
path = tmp_path / "config_lbm.json"
|
|
path.write_text(json.dumps(cfg_dict))
|
|
cfg = load_lbm_config(str(path))
|
|
assert (cfg.nx, cfg.ny) == (nx, ny) == (3000, 2400)
|
|
assert cfg.collision == "MRT" and cfg.y_wall_bc == "free_slip" and cfg.velocity == .04
|
|
assert math.isclose(cfg.viscosity, .04 * 40 / 150)
|
|
|
|
|
|
def test_finite_rejects_any_nonfinite_recorded_output():
|
|
row = (3, .1, .2, .3, .4, .5, .6, .7, 0.0)
|
|
assert viv.finite(row)
|
|
for i in range(len(row)):
|
|
for bad in (float("nan"), float("inf"), -float("inf")):
|
|
assert not viv.finite(row[:i] + (bad,) + row[i + 1:])
|