ClaudeandClaude Opus 5.5
02421ae379
Default (no --horizon) keeps the original cap-11 scoring. --horizon N requires complete cycles 1..N in every case, passes the horizon to common_window and labels the JSON. The unqualified phase-locked HF force component is reported per case. Protocol section 6b declares the D40/60D 31-cycle duration-only diagnostic. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
5.3 KiB
Python
111 lines
5.3 KiB
Python
"""CPU tests of the Dütsch protocol arithmetic (docs/validation/duetsch_re100_kc5_protocol.md §6a)."""
|
|
import importlib.util
|
|
import math
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"duetsch_protocol", Path(__file__).resolve().parents[2] / "research/duetsch_re100_kc5/protocol.py")
|
|
p = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(p)
|
|
|
|
T = 2520
|
|
|
|
|
|
def test_phase_and_cycle_bookkeeping():
|
|
assert p.phase(0, T) == 270.0 and p.cycle(0, T) == 0 # +A, at rest, phi = -90
|
|
assert p.phase(T // 4, T) == 0.0 and p.cycle(T // 4, T) == 1 # mean position, u = -U_max
|
|
assert p.position(0, T, 1.0) == 1.0
|
|
assert p.cycle(T // 4 - 1, T) == 0 and p.cycle(T // 4 + T - 1, T) == 1
|
|
assert p.cycle(T // 4 + T, T) == 2
|
|
assert p.phase(T // 6, T) == pytest.approx(330.0) and p.cycle(T // 6, T) == 0 # startup quarter
|
|
assert p.phase(7 * T // 6, T) == pytest.approx(330.0) and p.cycle(7 * T // 6, T) == 1
|
|
for ph, frac in ((180.0, .75), (210.0, 5 / 6)):
|
|
s = round(frac * T) + T
|
|
assert p.phase(s, T) == pytest.approx(ph) and p.bin_phase(s, T) == ph
|
|
assert p.bin_phase(round(.75 * T) + T - 1, T) is None
|
|
assert sum(p.bin_phase(s, T) == 180.0 for s in range(T, 2 * T)) == T // 360
|
|
|
|
|
|
def test_settling_and_common_window():
|
|
passes = {k: True for k in range(3, 12)}
|
|
assert p.settled_start(passes) == 7 # pairs 3->4, 4->5, 5->6
|
|
late = {**passes, 5: False}
|
|
assert p.settled_start(late) == 9 # pairs 6->7, 7->8, 8->9
|
|
assert p.common_window([7, 9], [passes, passes]) == (9, 10, 11)
|
|
assert p.common_window([7, 10], [passes, passes]) is None # K + 2 = 12 beyond the cap
|
|
assert p.common_window([7, None], [passes, passes]) is None
|
|
assert p.common_window([7, 7], [passes, {**passes, 8: False}]) is None # must stay settled
|
|
# a case settled early must keep passing while waiting for the common K
|
|
assert p.common_window([7, 9], [{**passes, 8: False}, passes]) is None
|
|
long = {k: True for k in range(3, 32)}
|
|
assert p.common_window([10], [long]) is None # default cap 11
|
|
assert p.common_window([29], [long], last_scored=31) == (29, 30, 31)
|
|
assert p.common_window([30], [long], last_scored=31) is None
|
|
|
|
|
|
def test_solid_aware_bilinear():
|
|
field = np.arange(16.0).reshape(4, 4)
|
|
fluid = np.ones((4, 4), bool)
|
|
value, used, total = p.bilinear_fluid(field, fluid, 1.25, 1.5)
|
|
assert total == pytest.approx(1.0) and value == pytest.approx(0.75 * 0.5 * (5 + 9) + 0.25 * 0.5 * (6 + 10))
|
|
fluid[1, 1] = False # one solid corner
|
|
value, used, total = p.bilinear_fluid(field, fluid, 1.25, 1.5)
|
|
assert total == pytest.approx(1.0 - 0.75 * 0.5)
|
|
assert all(not (i == 1 and j == 1) for i, j, _ in used)
|
|
assert sum(w for *_, w in used) == pytest.approx(1.0)
|
|
fluid[:] = False
|
|
assert p.bilinear_fluid(field, fluid, 1.25, 1.5) == (None, (), 0.0)
|
|
|
|
|
|
def test_interval_bases_and_morison_fit_recover_known_coefficients():
|
|
b = np.array([p.interval_bases(n, T) for n in range(T // 4, T // 4 + 3 * T)])
|
|
phi = 2 * np.pi * (np.arange(T // 4, T // 4 + 3 * T) + .5) / T - np.pi / 2
|
|
assert np.allclose(b[:, 1], np.sin(phi), atol=1e-5) # midpoint ~ interval mean
|
|
f = 2.09 * b[:, 0] - (math.pi ** 2 * 1.45 / p.KC) * b[:, 1]
|
|
assert p.morison_fit(f, b[:, 0], b[:, 1]) == pytest.approx((2.09, 1.45), abs=1e-9)
|
|
|
|
|
|
def test_geometry_repeats_exactly_across_cycles():
|
|
A = p.amplitude(20)
|
|
for m in (0, 1, T // 4, T // 2 - 1, 7 * T // 12):
|
|
for c in (1, 5, 11):
|
|
assert p.position(m + c * T, T, A) == p.position(m, T, A)
|
|
assert p.phase(m + c * T, T) == p.phase(m, T)
|
|
v0 = p.position(m + 1, T, A) - p.position(m, T, A)
|
|
assert p.position(m + 1 + c * T, T, A) - p.position(m + c * T, T, A) == v0
|
|
assert p.cycle(T // 4 + 5 * T, T) == 6 # unwrapped index stays separate
|
|
assert p.interval_bases(123 + 7 * T, T) == p.interval_bases(123, T)
|
|
|
|
|
|
def test_exact_point_inside_body_is_missing_even_with_fluid_corners():
|
|
field = np.zeros((6, 6))
|
|
fluid = np.ones((6, 6), bool)
|
|
fluid[3, 3] = False # only one corner solid
|
|
centre, radius = (3.0, 3.0), 0.3
|
|
value, used, total = p.sample(field, fluid, 3.2, 3.1, centre, radius) # inside the circle
|
|
assert value is None and used == () and total == 0.0
|
|
value, used, total = p.sample(field, fluid, 3.35, 3.0, centre, radius) # just outside
|
|
assert value is not None and total > 0.0
|
|
|
|
|
|
def test_generated_config_passes_the_real_loader(tmp_path):
|
|
import json
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "research/duetsch_re100_kc5"))
|
|
try:
|
|
import run
|
|
finally:
|
|
sys.path.pop(0)
|
|
from CelerisLab.config import load_lbm_config
|
|
cfg_dict, u_max, n = run.config(20, 10.0, 2520)
|
|
path = tmp_path / "config_lbm.json"
|
|
path.write_text(json.dumps(cfg_dict))
|
|
cfg = load_lbm_config(str(path))
|
|
assert (cfg.nx, cfg.ny) == (n, n) == (200, 200)
|
|
assert cfg.collision == "MRT" and cfg.streaming == "double_buffer" and not cfg.ddf_shifting
|
|
assert cfg.velocity == 0.0 and cfg.viscosity == pytest.approx(u_max * 20 / 100)
|
|
assert cfg.y_wall_bc == "bounce_back"
|