802 lines
27 KiB
Python
802 lines
27 KiB
Python
# CelerisLab/tests/run_kan99b_rotating_cylinder.py
|
|
"""Kan99b rotating-cylinder validation driver.
|
|
|
|
This script executes the rotating-cylinder campaign in
|
|
``tests/Rotating_cylinder_validation_plan.md`` against Kan99b anchors.
|
|
|
|
Core lattice mapping (fixed by campaign contract):
|
|
- D = 30, R = 15
|
|
- U_inf = 0.03
|
|
- nu = U_inf * D / Re = 0.9 / Re
|
|
- omega_body = 2 * alpha * U_inf / D = 0.002 * alpha
|
|
- inlet.profile = uniform
|
|
- y_wall_bc = free_slip
|
|
- outlet.mode = neq_extrap
|
|
- streaming = double_buffer
|
|
|
|
Phases:
|
|
- A: domain independence at Re=100, alpha=1.0 (MRT, domains S/M/L)
|
|
- B: anchor collision sweep at Re=100, alpha=1.0 (SRT/TRT/MRT)
|
|
- C: Re=100 alpha scan
|
|
- D: Re=60 and Re=160 threshold scan
|
|
|
|
Usage examples::
|
|
conda run -n pycuda_3_10 python tests/run_kan99b_rotating_cylinder.py --phase a
|
|
conda run -n pycuda_3_10 python tests/run_kan99b_rotating_cylinder.py --phase b --domain M
|
|
conda run -n pycuda_3_10 python tests/run_kan99b_rotating_cylinder.py --phase c --minimal
|
|
conda run -n pycuda_3_10 python tests/run_kan99b_rotating_cylinder.py --phase all --minimal
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
|
|
|
|
import numpy as np
|
|
import pycuda.driver as cuda
|
|
|
|
_REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
_DEFAULT_LBM = os.path.join(_REPO, "src", "CelerisLab", "configs", "config_lbm.json")
|
|
|
|
U_INF = 0.03
|
|
D_LATTICE = 30.0
|
|
R_LATTICE = 15.0
|
|
|
|
# Kan99b Table I anchor (Re=100, alpha=1.0).
|
|
KAN99B_ANCHOR = {
|
|
"St": 0.1655,
|
|
"mean_cl": -2.4881,
|
|
"mean_cd": 1.1040,
|
|
"amp_cl": 0.3631,
|
|
"amp_cd": 0.0993,
|
|
}
|
|
|
|
# Preferred agreement bands from the validation plan (fractional errors).
|
|
ANCHOR_BANDS = {
|
|
"St": 0.03,
|
|
"mean_cl": 0.04,
|
|
"mean_cd": 0.05,
|
|
"amp_cl": 0.08,
|
|
"amp_cd": 0.10,
|
|
}
|
|
|
|
# Domain sensitivity thresholds vs domain L (fractional errors).
|
|
DOMAIN_THRESH = {
|
|
"St": 0.01,
|
|
"mean_cl": 0.02,
|
|
"mean_cd": 0.02,
|
|
"amp_cl": 0.03,
|
|
"amp_cd": 0.03,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DomainSpec:
|
|
"""Rectangular domain defined in lattice units."""
|
|
|
|
key: str
|
|
nx: int
|
|
ny: int
|
|
center: Tuple[float, float]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RunSpec:
|
|
"""One executable run specification."""
|
|
|
|
phase: str
|
|
collision: str
|
|
domain: str
|
|
re: float
|
|
alpha: float
|
|
steps: int
|
|
burn: int
|
|
|
|
|
|
def _load_json(path: str) -> dict:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _write_json(path: str, payload: dict) -> None:
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f, indent=2)
|
|
|
|
|
|
def _domain_specs() -> Dict[str, DomainSpec]:
|
|
return {
|
|
"S": DomainSpec("S", 1081, 481, (360.0, 240.0)),
|
|
"M": DomainSpec("M", 1351, 601, (450.0, 300.0)),
|
|
"L": DomainSpec("L", 1801, 721, (600.0, 360.0)),
|
|
}
|
|
|
|
|
|
def _nu_from_re(re: float) -> float:
|
|
return U_INF * D_LATTICE / float(re)
|
|
|
|
|
|
def _omega_body(alpha: float) -> float:
|
|
return 2.0 * float(alpha) * U_INF / D_LATTICE
|
|
|
|
|
|
def _run_id(spec: RunSpec) -> str:
|
|
a = f"{spec.alpha:.3f}".replace(".", "p")
|
|
return f"phase{spec.phase}_dom{spec.domain}_re{int(spec.re)}_a{a}_{spec.collision.lower()}"
|
|
|
|
|
|
def _build_cfg(base_cfg: dict, *, nx: int, ny: int, collision: str, re: float) -> dict:
|
|
cfg = json.loads(json.dumps(base_cfg))
|
|
cfg["grid"]["nx"] = int(nx)
|
|
cfg["grid"]["ny"] = int(ny)
|
|
cfg["grid"]["nz"] = 1
|
|
cfg["physics"]["velocity"] = float(U_INF)
|
|
cfg["physics"]["viscosity"] = float(_nu_from_re(re))
|
|
cfg["physics"]["rho"] = 1.0
|
|
cfg["method"]["collision"] = str(collision).upper()
|
|
cfg["method"]["streaming"] = "double_buffer"
|
|
cfg["method"]["store_precision"] = "FP32"
|
|
cfg["method"]["ddf_shifting"] = False
|
|
cfg["method"]["les"]["enabled"] = False
|
|
cfg["method"]["inlet"]["profile"] = "uniform"
|
|
cfg["method"]["outlet"]["mode"] = "neq_extrap"
|
|
cfg["method"]["y_wall_bc"] = "free_slip"
|
|
return cfg
|
|
|
|
|
|
def _body_doc(center: Tuple[float, float], *, alpha: float) -> dict:
|
|
return {
|
|
"objects": [
|
|
{
|
|
"type": "cylinder",
|
|
"center": [float(center[0]), float(center[1])],
|
|
"radius": float(R_LATTICE),
|
|
"omega": float(_omega_body(alpha)),
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def _rfft_spectrum(x: np.ndarray, sample_dt: float) -> Tuple[np.ndarray, np.ndarray]:
|
|
v = np.asarray(x, dtype=np.float64)
|
|
if v.size < 64:
|
|
return np.zeros(0, dtype=np.float64), np.zeros(0, dtype=np.float64)
|
|
v = v - np.mean(v)
|
|
win = np.hanning(v.size)
|
|
spec = np.abs(np.fft.rfft(v * win)) ** 2
|
|
freqs = np.fft.rfftfreq(v.size, d=float(sample_dt))
|
|
return freqs.astype(np.float64), spec.astype(np.float64)
|
|
|
|
|
|
def _peak_freq_parabolic(freqs: np.ndarray, spec: np.ndarray, idx: int) -> float:
|
|
i = int(np.clip(idx, 0, spec.size - 1))
|
|
if i <= 0 or i + 1 >= spec.size:
|
|
return float(freqs[i])
|
|
y0 = np.log(spec[i - 1] + 1e-30)
|
|
y1 = np.log(spec[i] + 1e-30)
|
|
y2 = np.log(spec[i + 1] + 1e-30)
|
|
den = y0 - 2.0 * y1 + y2
|
|
if abs(den) < 1e-20:
|
|
return float(freqs[i])
|
|
delta = 0.5 * (y0 - y2) / den
|
|
delta = float(np.clip(delta, -1.0, 1.0))
|
|
df = float(freqs[i + 1] - freqs[i])
|
|
return float(freqs[i]) + delta * df
|
|
|
|
|
|
def _st_from_lift(lift: np.ndarray, sample_dt: float) -> float:
|
|
freqs, spec = _rfft_spectrum(lift, sample_dt=sample_dt)
|
|
if freqs.size <= 1:
|
|
return float("nan")
|
|
# Ignore DC bin.
|
|
idx = int(np.argmax(spec[1:])) + 1
|
|
f_peak = _peak_freq_parabolic(freqs, spec, idx)
|
|
return float(f_peak * D_LATTICE / U_INF)
|
|
|
|
|
|
def _cycle_half_p2p(y: np.ndarray) -> float:
|
|
"""Mean half peak-to-peak amplitude over cycles of demeaned signal."""
|
|
s = np.asarray(y, dtype=np.float64)
|
|
if s.size < 8:
|
|
return float("nan")
|
|
d = s - np.mean(s)
|
|
crossing = np.where((d[:-1] <= 0.0) & (d[1:] > 0.0))[0]
|
|
if crossing.size >= 2:
|
|
amps: List[float] = []
|
|
for i in range(crossing.size - 1):
|
|
seg = s[crossing[i] + 1 : crossing[i + 1] + 1]
|
|
if seg.size < 3:
|
|
continue
|
|
amps.append(0.5 * (float(np.max(seg)) - float(np.min(seg))))
|
|
if amps:
|
|
return float(np.mean(amps))
|
|
return 0.5 * (float(np.max(s)) - float(np.min(s)))
|
|
|
|
|
|
def _vorticity_z(ux: np.ndarray, uy: np.ndarray) -> np.ndarray:
|
|
ux = np.asarray(ux, dtype=np.float64)
|
|
uy = np.asarray(uy, dtype=np.float64)
|
|
return np.gradient(uy, axis=1) - np.gradient(ux, axis=0)
|
|
|
|
|
|
def _save_vorticity_png(path: str, ux: np.ndarray, uy: np.ndarray, title: str) -> None:
|
|
try:
|
|
import matplotlib
|
|
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
except ImportError:
|
|
return
|
|
|
|
omega = _vorticity_z(ux, uy)
|
|
abs_o = np.abs(omega[np.isfinite(omega)])
|
|
vmax = float(np.percentile(abs_o, 99.5)) if abs_o.size else 1.0
|
|
if vmax <= 0.0:
|
|
vmax = 1.0
|
|
ny, nx = omega.shape
|
|
fig, ax = plt.subplots(figsize=(min(18.0, max(8.0, nx / 100.0)), min(12.0, max(3.0, ny / 40.0))))
|
|
im = ax.imshow(
|
|
omega,
|
|
origin="lower",
|
|
aspect="equal",
|
|
cmap="RdBu_r",
|
|
vmin=-vmax,
|
|
vmax=vmax,
|
|
extent=(0, nx - 1, 0, ny - 1),
|
|
)
|
|
ax.set_xlabel("x (lattice)")
|
|
ax.set_ylabel("y (lattice)")
|
|
ax.set_title(title)
|
|
fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04, label="omega_z")
|
|
fig.tight_layout()
|
|
fig.savefig(path, dpi=150, bbox_inches="tight")
|
|
plt.close(fig)
|
|
|
|
|
|
def _run_one(
|
|
spec: RunSpec,
|
|
*,
|
|
domain: DomainSpec,
|
|
base_cfg: dict,
|
|
out_dir: str,
|
|
record_every: int,
|
|
field_every: int,
|
|
save_vorticity: bool,
|
|
) -> Dict[str, Any]:
|
|
cfg = _build_cfg(base_cfg, nx=domain.nx, ny=domain.ny, collision=spec.collision, re=spec.re)
|
|
bdoc = _body_doc(domain.center, alpha=spec.alpha)
|
|
|
|
tmpd = tempfile.mkdtemp(prefix="celeris_kan99b_")
|
|
lbm_tmp = os.path.join(tmpd, "config_lbm.json")
|
|
body_tmp = os.path.join(tmpd, "config_body.json")
|
|
_write_json(lbm_tmp, cfg)
|
|
_write_json(body_tmp, bdoc)
|
|
|
|
from CelerisLab import Simulation # noqa: WPS433
|
|
|
|
sim = Simulation(lbm_config_path=lbm_tmp, body_config_path=body_tmp)
|
|
# Contract: body omega is host-side runtime state from alpha conversion.
|
|
if sim.bodies.count < 1:
|
|
sim.close()
|
|
raise RuntimeError("Expected one cylinder in body config.")
|
|
sim.bodies.get(0).state.omega = np.float32(_omega_body(spec.alpha))
|
|
sim.initialize()
|
|
|
|
stream = cuda.Stream()
|
|
rec = max(1, int(record_every))
|
|
total = int(spec.burn) + int(spec.steps)
|
|
if total < 1:
|
|
sim.close()
|
|
raise ValueError("burn + steps must be >= 1")
|
|
|
|
steps: List[int] = []
|
|
fx_hist: List[float] = []
|
|
fy_hist: List[float] = []
|
|
field_snapshots: List[str] = []
|
|
run_id = _run_id(spec)
|
|
snap_dir = os.path.join(out_dir, "fields", run_id)
|
|
if field_every > 0:
|
|
os.makedirs(snap_dir, exist_ok=True)
|
|
|
|
for step in range(1, total + 1):
|
|
sim.bodies.zero_force_segment_async(stream)
|
|
sim.stepper.step(
|
|
1,
|
|
action_gpu=sim.bodies.action_gpu,
|
|
obs_gpu=sim.bodies.obs_gpu,
|
|
stream=stream,
|
|
)
|
|
if step % rec == 0 or step == total:
|
|
stream.synchronize()
|
|
sim.bodies.download_obs_full_async(stream)
|
|
stream.synchronize()
|
|
fvec = sim.bodies.read_force(0)
|
|
fx = float(fvec[0])
|
|
fy = float(fvec[1])
|
|
steps.append(step)
|
|
fx_hist.append(fx)
|
|
fy_hist.append(fy)
|
|
if not np.isfinite(fx) or not np.isfinite(fy):
|
|
sim.close()
|
|
raise RuntimeError(f"NaN/Inf force at step {step}")
|
|
if field_every > 0 and (step % int(field_every) == 0 or step == total):
|
|
stream.synchronize()
|
|
macro = sim.get_macroscopic()
|
|
save_p = os.path.join(snap_dir, f"macro_step{step:08d}.npz")
|
|
np.savez_compressed(
|
|
save_p,
|
|
rho=np.asarray(macro["rho"], dtype=np.float32),
|
|
ux=np.asarray(macro["ux"], dtype=np.float32),
|
|
uy=np.asarray(macro["uy"], dtype=np.float32),
|
|
)
|
|
field_snapshots.append(save_p)
|
|
|
|
stream.synchronize()
|
|
macro_last = sim.get_macroscopic()
|
|
ux_last = np.asarray(macro_last["ux"], dtype=np.float64).reshape(domain.ny, domain.nx)
|
|
uy_last = np.asarray(macro_last["uy"], dtype=np.float64).reshape(domain.ny, domain.nx)
|
|
rho_last = np.asarray(macro_last["rho"], dtype=np.float64).reshape(domain.ny, domain.nx)
|
|
sim.close()
|
|
|
|
step_arr = np.asarray(steps, dtype=np.int64)
|
|
fx_arr = np.asarray(fx_hist, dtype=np.float64)
|
|
fy_arr = np.asarray(fy_hist, dtype=np.float64)
|
|
burn_mask = step_arr >= int(spec.burn)
|
|
if not np.any(burn_mask):
|
|
burn_mask = np.ones_like(step_arr, dtype=bool)
|
|
|
|
cl = 2.0 * fy_arr / (1.0 * (U_INF ** 2) * D_LATTICE)
|
|
cd = 2.0 * fx_arr / (1.0 * (U_INF ** 2) * D_LATTICE)
|
|
cl_tail = cl[burn_mask]
|
|
cd_tail = cd[burn_mask]
|
|
st = _st_from_lift(cl_tail, sample_dt=float(rec))
|
|
amp_cl = _cycle_half_p2p(cl_tail)
|
|
amp_cd = _cycle_half_p2p(cd_tail)
|
|
|
|
csv_dir = os.path.join(out_dir, "force_csv")
|
|
os.makedirs(csv_dir, exist_ok=True)
|
|
csv_path = os.path.join(csv_dir, f"{run_id}.csv")
|
|
with open(csv_path, "w", newline="", encoding="utf-8") as f:
|
|
w = csv.writer(f)
|
|
w.writerow(["step", "fx", "fy", "cd", "cl"])
|
|
for i, s in enumerate(step_arr.tolist()):
|
|
w.writerow([s, fx_arr[i], fy_arr[i], cd[i], cl[i]])
|
|
|
|
if save_vorticity:
|
|
vdir = os.path.join(out_dir, "vorticity")
|
|
os.makedirs(vdir, exist_ok=True)
|
|
_save_vorticity_png(
|
|
os.path.join(vdir, f"{run_id}.png"),
|
|
ux_last,
|
|
uy_last,
|
|
title=(
|
|
f"Kan99b {spec.phase.upper()} {spec.collision} dom={spec.domain} "
|
|
f"Re={spec.re:.0f} alpha={spec.alpha:.3f}"
|
|
),
|
|
)
|
|
|
|
return {
|
|
"run_id": run_id,
|
|
"phase": spec.phase,
|
|
"collision": spec.collision,
|
|
"domain": spec.domain,
|
|
"re": float(spec.re),
|
|
"alpha": float(spec.alpha),
|
|
"omega_body": float(_omega_body(spec.alpha)),
|
|
"nu": float(_nu_from_re(spec.re)),
|
|
"steps": int(spec.steps),
|
|
"burn": int(spec.burn),
|
|
"total_steps": int(total),
|
|
"record_every": int(rec),
|
|
"n_samples": int(step_arr.size),
|
|
"mean_cd": float(np.mean(cd_tail)),
|
|
"mean_cl": float(np.mean(cl_tail)),
|
|
"amp_cd": float(amp_cd),
|
|
"amp_cl": float(amp_cl),
|
|
"st": float(st),
|
|
"rho_min_final": float(np.min(rho_last)),
|
|
"rho_max_final": float(np.max(rho_last)),
|
|
"force_csv": csv_path,
|
|
"field_snapshots": field_snapshots,
|
|
}
|
|
|
|
|
|
def _alpha_list_from_str(text: str) -> List[float]:
|
|
vals: List[float] = []
|
|
for t in text.split(","):
|
|
t = t.strip()
|
|
if t:
|
|
vals.append(float(t))
|
|
return vals
|
|
|
|
|
|
def _phase_runs(
|
|
phase: str,
|
|
*,
|
|
minimal: bool,
|
|
domain_key: str,
|
|
collisions: Sequence[str],
|
|
alpha_override: Optional[List[float]],
|
|
steps: int,
|
|
burn: int,
|
|
) -> List[RunSpec]:
|
|
runs: List[RunSpec] = []
|
|
|
|
def add_many(
|
|
p: str,
|
|
ds: Iterable[str],
|
|
cs: Iterable[str],
|
|
res: Iterable[float],
|
|
alphas: Iterable[float],
|
|
*,
|
|
phase_steps: Optional[int] = None,
|
|
phase_burn: Optional[int] = None,
|
|
) -> None:
|
|
for d in ds:
|
|
for c in cs:
|
|
for re in res:
|
|
for a in alphas:
|
|
runs.append(
|
|
RunSpec(
|
|
phase=p,
|
|
collision=str(c).upper(),
|
|
domain=d,
|
|
re=float(re),
|
|
alpha=float(a),
|
|
steps=int(phase_steps if phase_steps is not None else steps),
|
|
burn=int(phase_burn if phase_burn is not None else burn),
|
|
)
|
|
)
|
|
|
|
# Plan-driven defaults.
|
|
alpha_c = [0.0, 0.5, 1.0, 1.5, 1.7, 1.8, 1.9, 2.0]
|
|
alpha_c_min = [0.0, 1.0, 1.5, 1.8, 2.0]
|
|
alpha_d_60 = [0.0, 0.5, 1.0, 1.2, 1.4, 1.6]
|
|
alpha_d_160 = [0.0, 0.5, 1.0, 1.5, 1.8, 1.9, 2.0]
|
|
alpha_d_min = {60.0: [1.4], 160.0: [1.9]}
|
|
anchor_steps = 200_000
|
|
anchor_burn = 80_000
|
|
near_steps = 240_000
|
|
near_burn = 120_000
|
|
periodic_steps = 160_000
|
|
periodic_burn = 64_000
|
|
|
|
if phase in ("a", "all"):
|
|
add_many("a", ["S", "M", "L"], ["MRT"], [100.0], [1.0], phase_steps=anchor_steps, phase_burn=anchor_burn)
|
|
if phase in ("anchor", "b", "all"):
|
|
add_many("b", [domain_key], collisions, [100.0], [1.0], phase_steps=anchor_steps, phase_burn=anchor_burn)
|
|
if phase in ("c", "all"):
|
|
alphas = alpha_override if alpha_override is not None else (alpha_c_min if minimal else alpha_c)
|
|
# Near-critical values need longer windows.
|
|
for a in alphas:
|
|
ps = near_steps if abs(a - 1.8) < 0.11 else periodic_steps
|
|
pb = near_burn if abs(a - 1.8) < 0.11 else periodic_burn
|
|
add_many("c", [domain_key], collisions, [100.0], [a], phase_steps=ps, phase_burn=pb)
|
|
if phase in ("d", "all"):
|
|
if minimal:
|
|
for re, alphas in alpha_d_min.items():
|
|
add_many("d", [domain_key], collisions, [re], alphas, phase_steps=near_steps, phase_burn=near_burn)
|
|
else:
|
|
add_many("d", [domain_key], collisions, [60.0], alpha_d_60, phase_steps=periodic_steps, phase_burn=periodic_burn)
|
|
add_many("d", [domain_key], collisions, [160.0], alpha_d_160, phase_steps=periodic_steps, phase_burn=periodic_burn)
|
|
|
|
# CLI override for quick tests.
|
|
if steps > 0:
|
|
for i in range(len(runs)):
|
|
runs[i] = RunSpec(
|
|
phase=runs[i].phase,
|
|
collision=runs[i].collision,
|
|
domain=runs[i].domain,
|
|
re=runs[i].re,
|
|
alpha=runs[i].alpha,
|
|
steps=steps,
|
|
burn=burn,
|
|
)
|
|
return runs
|
|
|
|
|
|
def _rel_err(meas: float, ref: float) -> Optional[float]:
|
|
if not np.isfinite(meas) or ref == 0:
|
|
return None
|
|
return abs(float(meas) - float(ref)) / abs(float(ref))
|
|
|
|
|
|
def _phase_a_gate(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
dom = {r["domain"]: r for r in rows}
|
|
out: Dict[str, Any] = {"phase": "a", "pass": False}
|
|
if not all(k in dom for k in ("S", "M", "L")):
|
|
out["error"] = "Phase A needs S, M, L rows."
|
|
return out
|
|
l = dom["L"]
|
|
compare: Dict[str, Any] = {}
|
|
for k in ("S", "M"):
|
|
r = dom[k]
|
|
compare[k] = {
|
|
"St": _rel_err(r["st"], l["st"]),
|
|
"mean_cl": _rel_err(r["mean_cl"], l["mean_cl"]),
|
|
"mean_cd": _rel_err(r["mean_cd"], l["mean_cd"]),
|
|
"amp_cl": _rel_err(r["amp_cl"], l["amp_cl"]),
|
|
"amp_cd": _rel_err(r["amp_cd"], l["amp_cd"]),
|
|
}
|
|
choose = "L"
|
|
m_ok = all(
|
|
(compare["M"][metric] is not None and compare["M"][metric] <= DOMAIN_THRESH[metric])
|
|
for metric in DOMAIN_THRESH
|
|
)
|
|
if m_ok:
|
|
choose = "M"
|
|
else:
|
|
s_ok = all(
|
|
(compare["S"][metric] is not None and compare["S"][metric] <= DOMAIN_THRESH[metric])
|
|
for metric in DOMAIN_THRESH
|
|
)
|
|
if s_ok:
|
|
choose = "S"
|
|
out["compare_vs_L"] = compare
|
|
out["recommended_domain"] = choose
|
|
out["pass"] = True
|
|
return out
|
|
|
|
|
|
def _phase_b_anchor_eval(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
by_coll = {r["collision"]: r for r in rows}
|
|
out: Dict[str, Any] = {"phase": "b", "rows": {}}
|
|
for coll in ("SRT", "TRT", "MRT"):
|
|
row = by_coll.get(coll)
|
|
if row is None:
|
|
continue
|
|
metrics = {
|
|
"St": _rel_err(row["st"], KAN99B_ANCHOR["St"]),
|
|
"mean_cl": _rel_err(row["mean_cl"], KAN99B_ANCHOR["mean_cl"]),
|
|
"mean_cd": _rel_err(row["mean_cd"], KAN99B_ANCHOR["mean_cd"]),
|
|
"amp_cl": _rel_err(row["amp_cl"], KAN99B_ANCHOR["amp_cl"]),
|
|
"amp_cd": _rel_err(row["amp_cd"], KAN99B_ANCHOR["amp_cd"]),
|
|
}
|
|
out["rows"][coll] = {
|
|
"rel_err": metrics,
|
|
"pass_bands": {
|
|
m: (metrics[m] is not None and metrics[m] <= ANCHOR_BANDS[m]) for m in ANCHOR_BANDS
|
|
},
|
|
}
|
|
return out
|
|
|
|
|
|
def _save_summary_plots(rows: List[Dict[str, Any]], out_dir: str) -> None:
|
|
try:
|
|
import matplotlib
|
|
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
except ImportError:
|
|
return
|
|
|
|
summary_dir = os.path.join(out_dir, "summary_plots")
|
|
os.makedirs(summary_dir, exist_ok=True)
|
|
|
|
def plot_metric(metric: str, ylabel: str, filename: str) -> None:
|
|
fig, ax = plt.subplots(figsize=(8, 5))
|
|
for coll in ("SRT", "TRT", "MRT"):
|
|
coll_rows = [r for r in rows if r["collision"] == coll]
|
|
if not coll_rows:
|
|
continue
|
|
# Use Re=100 sweep first if present, else all points sorted by alpha.
|
|
target = [r for r in coll_rows if abs(r["re"] - 100.0) < 1e-9]
|
|
data = target if target else coll_rows
|
|
data = sorted(data, key=lambda r: (r["re"], r["alpha"]))
|
|
ax.plot(
|
|
[r["alpha"] for r in data],
|
|
[r[metric] for r in data],
|
|
marker="o",
|
|
linewidth=1.4,
|
|
label=coll,
|
|
)
|
|
ax.set_xlabel("alpha")
|
|
ax.set_ylabel(ylabel)
|
|
ax.set_title(f"{ylabel} vs alpha")
|
|
ax.grid(True, alpha=0.3)
|
|
ax.legend(loc="best")
|
|
fig.tight_layout()
|
|
fig.savefig(os.path.join(summary_dir, filename), dpi=150, bbox_inches="tight")
|
|
plt.close(fig)
|
|
|
|
plot_metric("mean_cl", "mean C_L", "mean_cl_vs_alpha.png")
|
|
plot_metric("mean_cd", "mean C_D", "mean_cd_vs_alpha.png")
|
|
plot_metric("amp_cl", "C'_L (half peak-to-peak)", "amp_cl_vs_alpha.png")
|
|
plot_metric("st", "St", "st_vs_alpha.png")
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description="Kan99b rotating-cylinder validation driver")
|
|
ap.add_argument("--phase", default="all", choices=("anchor", "a", "b", "c", "d", "all"))
|
|
ap.add_argument("--minimal", action="store_true", help="Run reduced minimum set from the plan.")
|
|
ap.add_argument("--domain", default="M", choices=("S", "M", "L"), help="Default domain for phases B/C/D.")
|
|
ap.add_argument("--collision", default="all", choices=("SRT", "TRT", "MRT", "all"))
|
|
ap.add_argument("--alpha", type=float, default=None, help="Single alpha override (for c/d phases).")
|
|
ap.add_argument("--alpha-list", type=str, default="", help="Comma-separated alpha list override.")
|
|
ap.add_argument("--steps", type=int, default=0, help="Override run steps for all selected runs.")
|
|
ap.add_argument("--burn", type=int, default=0, help="Override burn steps for all selected runs.")
|
|
ap.add_argument("--record-every", type=int, default=100)
|
|
ap.add_argument("--field-every", type=int, default=0, help="Dump macro field .npz every N steps (0 disables).")
|
|
ap.add_argument("--out-dir", type=str, default=os.path.join(_REPO, "tests", "output", "kan99b_validation"))
|
|
ap.add_argument("--smoke", action="store_true", help="Very short run for wiring checks.")
|
|
ap.add_argument("--save-vorticity", action="store_true", help="Save final vorticity PNG per run.")
|
|
ap.add_argument("--json-out", type=str, default="", help="Optional explicit summary JSON output path.")
|
|
args = ap.parse_args()
|
|
|
|
if not os.path.isfile(_DEFAULT_LBM):
|
|
print(f"Missing base config: {_DEFAULT_LBM}", file=sys.stderr)
|
|
return 2
|
|
base_cfg = _load_json(_DEFAULT_LBM)
|
|
|
|
out_dir = os.path.abspath(args.out_dir)
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
|
|
collisions = ["SRT", "TRT", "MRT"] if args.collision == "all" else [str(args.collision).upper()]
|
|
alpha_override: Optional[List[float]] = None
|
|
if args.alpha is not None:
|
|
alpha_override = [float(args.alpha)]
|
|
elif args.alpha_list.strip():
|
|
alpha_override = _alpha_list_from_str(args.alpha_list)
|
|
|
|
if args.smoke:
|
|
o_steps = 2000
|
|
o_burn = 800
|
|
else:
|
|
o_steps = max(0, int(args.steps))
|
|
o_burn = max(0, int(args.burn))
|
|
|
|
runs = _phase_runs(
|
|
args.phase,
|
|
minimal=bool(args.minimal),
|
|
domain_key=args.domain,
|
|
collisions=collisions,
|
|
alpha_override=alpha_override,
|
|
steps=o_steps,
|
|
burn=o_burn,
|
|
)
|
|
if not runs:
|
|
print("No runs selected.", file=sys.stderr)
|
|
return 2
|
|
|
|
domains = _domain_specs()
|
|
rows: List[Dict[str, Any]] = []
|
|
|
|
contract = {
|
|
"U_inf": U_INF,
|
|
"D_lattice": D_LATTICE,
|
|
"R_lattice": R_LATTICE,
|
|
"nu_formula": "nu = U_inf * D / Re = 0.9 / Re",
|
|
"omega_formula": "omega_body = 2 * alpha * U_inf / D = 0.002 * alpha",
|
|
"method_contract": {
|
|
"inlet_profile": "uniform",
|
|
"y_wall_bc": "free_slip",
|
|
"outlet_mode": "neq_extrap",
|
|
"streaming": "double_buffer",
|
|
"store_precision": "FP32",
|
|
"les_enabled": False,
|
|
},
|
|
}
|
|
|
|
for spec in runs:
|
|
dspec = domains[spec.domain]
|
|
print(
|
|
f"--- {spec.phase.upper()} {spec.collision} dom={spec.domain} Re={spec.re:.0f} "
|
|
f"alpha={spec.alpha:.3f} burn={spec.burn} steps={spec.steps} ---",
|
|
flush=True,
|
|
)
|
|
try:
|
|
row = _run_one(
|
|
spec,
|
|
domain=dspec,
|
|
base_cfg=base_cfg,
|
|
out_dir=out_dir,
|
|
record_every=max(1, int(args.record_every)),
|
|
field_every=max(0, int(args.field_every)),
|
|
save_vorticity=bool(args.save_vorticity),
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
rows.append(
|
|
{
|
|
"run_id": _run_id(spec),
|
|
"phase": spec.phase,
|
|
"collision": spec.collision,
|
|
"domain": spec.domain,
|
|
"re": float(spec.re),
|
|
"alpha": float(spec.alpha),
|
|
"error": str(e),
|
|
}
|
|
)
|
|
print(f"FAILED: {e}", flush=True)
|
|
continue
|
|
rows.append(row)
|
|
print(
|
|
" "
|
|
f"St={row['st']:.5f} mean_CL={row['mean_cl']:.4f} mean_CD={row['mean_cd']:.4f} "
|
|
f"C'L={row['amp_cl']:.4f} C'D={row['amp_cd']:.4f}",
|
|
flush=True,
|
|
)
|
|
|
|
# Summary table outputs
|
|
summary_csv = os.path.join(out_dir, "summary_runs.csv")
|
|
csv_keys = [
|
|
"run_id",
|
|
"phase",
|
|
"collision",
|
|
"domain",
|
|
"re",
|
|
"alpha",
|
|
"omega_body",
|
|
"nu",
|
|
"burn",
|
|
"steps",
|
|
"total_steps",
|
|
"record_every",
|
|
"n_samples",
|
|
"st",
|
|
"mean_cl",
|
|
"mean_cd",
|
|
"amp_cl",
|
|
"amp_cd",
|
|
"rho_min_final",
|
|
"rho_max_final",
|
|
"force_csv",
|
|
"error",
|
|
]
|
|
with open(summary_csv, "w", newline="", encoding="utf-8") as f:
|
|
w = csv.DictWriter(f, fieldnames=csv_keys)
|
|
w.writeheader()
|
|
for r in rows:
|
|
w.writerow({k: r.get(k, "") for k in csv_keys})
|
|
|
|
phase_reports: Dict[str, Any] = {}
|
|
phase_a_rows = [r for r in rows if r.get("phase") == "a" and "error" not in r]
|
|
if phase_a_rows:
|
|
phase_reports["a"] = _phase_a_gate(phase_a_rows)
|
|
phase_b_rows = [r for r in rows if r.get("phase") == "b" and "error" not in r]
|
|
if phase_b_rows:
|
|
phase_reports["b"] = _phase_b_anchor_eval(phase_b_rows)
|
|
|
|
_save_summary_plots([r for r in rows if "error" not in r], out_dir)
|
|
|
|
summary = {
|
|
"contract": contract,
|
|
"requested": {
|
|
"phase": args.phase,
|
|
"minimal": bool(args.minimal),
|
|
"domain": args.domain,
|
|
"collision": args.collision,
|
|
"steps_override": int(o_steps),
|
|
"burn_override": int(o_burn),
|
|
"record_every": int(args.record_every),
|
|
"field_every": int(args.field_every),
|
|
"save_vorticity": bool(args.save_vorticity),
|
|
},
|
|
"counts": {
|
|
"requested_runs": len(runs),
|
|
"completed_runs": sum(1 for r in rows if "error" not in r),
|
|
"failed_runs": sum(1 for r in rows if "error" in r),
|
|
},
|
|
"phase_reports": phase_reports,
|
|
"rows": rows,
|
|
}
|
|
json_out = (
|
|
os.path.abspath(args.json_out)
|
|
if args.json_out.strip()
|
|
else os.path.join(out_dir, "summary_runs.json")
|
|
)
|
|
_write_json(json_out, summary)
|
|
|
|
print(f"Wrote: {summary_csv}", flush=True)
|
|
print(f"Wrote: {json_out}", flush=True)
|
|
print(f"Wrote: {os.path.join(out_dir, 'summary_plots')}", flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|