# legacy_test/test_karman_cloak_re100.py """Karman Cloak Re100 — flagship legacy test. Builds the Karman cloak environment with LegacyCelerisLab, loads the d1a3o12_re100 PPO model, runs deterministic inference for 200 steps, and compares output against SR_analysis reference data. Usage:: conda run -n pycuda_3_10 python test_karman_cloak_re100.py --device 0 Expected: near-perfect match (same CFD, same model). """ from __future__ import annotations import argparse import json import os import sys import time from collections import deque import numpy as np _REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")) _SRC = os.path.join(_REPO, "src") _DRL = os.path.join(_SRC, "drl_pinball") for p in [_REPO, _SRC, _DRL]: if p not in sys.path: sys.path.insert(0, p) from LegacyCelerisLab import FlowField # noqa: E402 from legacy_test.core.legacy_env_builder import ( # noqa: E402 FIFO_LEN, CONV_LEN, U0, DATA_TYPE, ) from legacy_test.core.model_loader import load_model # noqa: E402 from legacy_test.core.comparator import compare_scene # noqa: E402 from legacy_test.core.io_helpers import ( # noqa: E402 save_signals, save_target, save_norm, ) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- L0 = 20.0 SAMPLE_INTERVAL = 800 S_DIM, A_DIM = 12, 3 ACTION_SCALE = 8.0 ACTION_BIAS = np.array([0.0, -4.0, 4.0], dtype=np.float32) NUM_STEPS = 200 # matches SR_analysis reference REF_DIR = os.path.join(_SRC, "SR_analysis", "data", "karman", "karman_re100") OUT_DIR = os.path.join(os.path.dirname(__file__), "output", "karman_cloak_re100") def log(msg: str) -> None: print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(description="Karman Cloak Re100 legacy test") ap.add_argument("--device", type=int, default=0, help="GPU device ID") ap.add_argument("--out", type=str, default=OUT_DIR, help="Output directory") ap.add_argument("--model", type=str, default="d1a3o12_re100", help="Model name") args = ap.parse_args() os.makedirs(args.out, exist_ok=True) log("=== Karman Cloak Re100: Legacy Test ===") log(f"Model: {args.model}, Device: {args.device}") log(f"Reference: {REF_DIR}") log(f"Output: {args.out}") # ---- Phase 1: Build environment ---- log("Building Karman cloak environment...") from legacy_test.core.legacy_env_builder import build_karman_cloak data = build_karman_cloak(device_id=args.device, re_code=100.0) ff: FlowField = data["flow_field"] target_states = data["target_states"] norm = data["norm"] n_obj_total = norm.get("n_obj_total", 7) log(f" force_norm_fact = {norm['force_norm_fact']:.6f}") log(f" sens_deviation = {norm['sens_deviation']}") # Save target and norm as reference save_target(args.out, target_states) save_norm(args.out, norm) # ---- Phase 2: Load model ---- log(f"Loading model: {args.model}") model = load_model(args.model) log(" Model loaded on CPU") # ---- Phase 3: Inference ---- log(f"Running {NUM_STEPS} steps of deterministic inference...") force_norm_fact = float(norm["force_norm_fact"]) sens_deviation = np.array(norm["sens_deviation"], dtype=np.float32) sens_norm_fact = np.array(norm["sens_norm_fact"], dtype=np.float32) # Restore DDF to steady pinball state (pre-bias) ff.restore_ddf() ff.apply_ddf() # Bias-action FIFO init (FlowField.run() has BUILT-IN EMA) fifo = deque(maxlen=FIFO_LEN) bias_arr = np.zeros(n_obj_total, dtype=DATA_TYPE) bias_arr[n_obj_total - 3] = float(ACTION_BIAS[0] * U0) # front bias_arr[n_obj_total - 2] = float(ACTION_BIAS[1] * U0) # top bias_arr[n_obj_total - 1] = float(ACTION_BIAS[2] * U0) # bottom for _ in range(FIFO_LEN): ff.run(SAMPLE_INTERVAL, bias_arr) fifo.append(ff.obs.copy()[2:14]) # DRL inference loop sig_s = np.zeros((NUM_STEPS, 6), dtype=np.float32) sig_f = np.zeros((NUM_STEPS, 6), dtype=np.float32) sig_a = np.zeros((NUM_STEPS, 3), dtype=np.float32) sig_r = np.zeros(NUM_STEPS, dtype=np.float32) obs = np.zeros(S_DIM, dtype=np.float32) for step in range(NUM_STEPS): # PPO action action, _states = model.predict(obs, deterministic=True) action = action.astype(np.float32).flatten() sig_a[step] = action.copy() # Convert to legacy action array action_arr = np.zeros(n_obj_total, dtype=DATA_TYPE) omega = (action * ACTION_SCALE + ACTION_BIAS) * U0 action_arr[n_obj_total - 3:] = omega # Run CFD (FlowField.run has internal EMA smoothing) ff.context.push() try: ff.run(SAMPLE_INTERVAL, action_arr) finally: ff.context.pop() # Read telemetry obs_slice = ff.obs.copy()[2:14] fifo.append(obs_slice) sig_s[step] = obs_slice[0:6].copy() sig_f[step] = obs_slice[6:12].copy() # Build normalised observation forces_norm = obs_slice[6:12] / force_norm_fact sens_norm = (obs_slice[0:6] - sens_deviation) / sens_norm_fact obs = np.clip(np.hstack([forces_norm, sens_norm]), -1.0, 1.0).astype(np.float32) # Compute reward (exact legacy formula) if step >= CONV_LEN: states_arr = np.array(fifo, dtype=np.float32) forces = states_arr[-1, 6:12] / force_norm_fact cd = float((forces[0] + forces[2] + forces[4]) / 3.0) cl = float((forces[1] + forces[3] + forces[5]) / 3.0) # DTW similarity (legacy calc_lag + calc_dtw_sim) from legacy_test.core.dtw_metrics import calc_lag, calc_dtw_sim mid_idx = 1 # sensor1_uy t_seq = target_states[CONV_LEN:2 * CONV_LEN, mid_idx] s_seq = states_arr[-CONV_LEN:, mid_idx] lag = calc_lag(t_seq, s_seq) sim_sum = 0.0 for i in range(6): t_seq2 = np.roll(target_states[:, i], -lag)[CONV_LEN:2 * CONV_LEN] s_seq2 = states_arr[-CONV_LEN:, i] sim_sum += calc_dtw_sim(t_seq2, s_seq2) sim_val = float(sim_sum / 6.0) r_cd = float(np.exp(-abs(cd * 20.0))) r_cl = float(np.exp(-abs(cl * 80.0))) r_sim = float(np.exp(-10.0 * abs(sim_val - 1.0))) sig_r[step] = float(min(0.3 * r_cd + 0.4 * r_cl + 0.3 * r_sim, 1.0)) # Save signals save_signals(args.out, sig_s, sig_f, sig_a, name="controlled") save_signals(args.out, sig_s, sig_f, sig_a, name="uncontrolled") # Also save to match SR_analysis format (with rewards) np.savez_compressed( os.path.join(args.out, "controlled.npz"), sensors=sig_s, forces=sig_f, actions=sig_a, rewards=sig_r, ) # Save config with open(os.path.join(args.out, "config.json"), "w") as f: json.dump({ "device_id": args.device, "re_code": 100.0, "viscosity": 0.004, "u0": float(U0), "sample_interval": SAMPLE_INTERVAL, "num_steps": NUM_STEPS, "action_scale": ACTION_SCALE, "action_bias": ACTION_BIAS.tolist(), "model": args.model, }, f, indent=2) # ---- Phase 4: Compare against reference ---- log("\n=== Comparison against SR_analysis reference ===") result = compare_scene( REF_DIR, sig_s, sig_f, sig_a, conv_len=CONV_LEN, label="karman_re100", ) with open(os.path.join(args.out, "result.json"), "w") as f: json.dump(result, f, indent=2) log(f"\nFinal reward: mean={sig_r.mean():.4f}, last_50={sig_r[-50:].mean():.4f}") log(f"DTW similarity: {result['dtw_sim']:.4f}") log(f"Action corr: {result['action_corr']}") if result["passed"]: log("PASS — All metrics within thresholds.") else: log("FAIL — One or more metrics below threshold.") # Cleanup del ff log("Done.") return 0 if result["passed"] else 1 if __name__ == "__main__": raise SystemExit(main())