04c2bc75ea
- Replace split zero_force_segment / zero_sensor_segment with unified zero_obs_async() — a single memset covers all three obs segments (force, torque, sensor), resetting the step accumulator. - Add _obs_accum_steps counter so read_*(normalize=True) returns the physically meaningful per-step average for all telemetry fields. - Sensor now always applies area-normalisation internally; the normalize parameter only controls the additional time-normalisation step. - run() gains zero_obs=True parameter (default) to control reset-on-step. - 7 new integration tests covering accumulation, zeroing, and normalise. - Fix bug in test_sensor_accuracy.py (undefined loop variable i). - Bump version to 0.4.0 for the API change. Co-authored-by: Cursor <cursoragent@cursor.com>
124 lines
4.0 KiB
Python
124 lines
4.0 KiB
Python
# CelerisLab/tests/validation/test_sensor_accuracy.py
|
|
"""Sensor accuracy validation: compare sensor readings to direct flow field averages.
|
|
|
|
This script validates that the GPU sensor kernel accumulation matches a
|
|
CPU-side manual average of the macroscopic field over the same cell footprint.
|
|
|
|
Usage::
|
|
|
|
conda run -n pycuda_3_10 python tests/validation/test_sensor_accuracy.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
_REPO = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(_REPO / "src"))
|
|
|
|
from CelerisLab import Simulation
|
|
|
|
|
|
def test_sensor_accuracy() -> dict:
|
|
"""Run sensor accuracy validation with multiple sensor positions."""
|
|
cfg = json.loads(
|
|
(Path(_REPO) / "src" / "CelerisLab" / "configs" / "config_lbm.json").read_text()
|
|
)
|
|
cfg["grid"]["nx"] = 256
|
|
cfg["grid"]["ny"] = 128
|
|
cfg["grid"]["nz"] = 1
|
|
cfg["physics"]["viscosity"] = 0.009
|
|
cfg["physics"]["velocity"] = 0.03
|
|
cfg["method"]["collision"] = "MRT"
|
|
cfg["method"]["inlet"]["scheme"] = "regularized"
|
|
cfg["method"]["inlet"]["profile"] = "uniform"
|
|
cfg["method"]["y_wall_bc"] = "free_slip"
|
|
|
|
tmpd = tempfile.mkdtemp(prefix="sensor_test_")
|
|
lbm_path = os.path.join(tmpd, "config_lbm.json")
|
|
with open(lbm_path, "w") as f:
|
|
json.dump(cfg, f)
|
|
|
|
sim = Simulation(lbm_config_path=lbm_path)
|
|
sim.add_body("circle", center=(80, 64), radius=15)
|
|
|
|
positions = [(120, 50), (120, 64), (120, 78), (150, 64)]
|
|
sensor_ids = []
|
|
for cx, cy in positions:
|
|
sid = sim.add_body("sensor", center=(cx, cy), radius=10)
|
|
sensor_ids.append(sid)
|
|
|
|
sim.initialize()
|
|
print(f"Initialized: nx={cfg['grid']['nx']} ny={cfg['grid']['ny']} "
|
|
f"n_curved={sim.field.n_curved} n_sensor={sim.field.n_sensor}")
|
|
|
|
# Step to develop wake
|
|
for _ in range(50):
|
|
sim.run(20)
|
|
|
|
# Get macroscopic field after one more step (with sensor accumulation)
|
|
import pycuda.driver as cuda
|
|
stream = cuda.Stream()
|
|
sim.run(1, zero_obs=True, upload_act=False, sync_obs=True, stream=stream)
|
|
# stream.synchronize() is called inside run()
|
|
|
|
macro = sim.get_macroscopic()
|
|
ux = macro["ux"]
|
|
uy = macro["uy"]
|
|
|
|
results = {}
|
|
all_pass = True
|
|
for idx, sid in enumerate(sensor_ids):
|
|
pos = positions[idx]
|
|
cells_arr, _ = sim.bodies.get(sid).get_sensor_list(
|
|
sim.lbm_cfg.nx, sim.lbm_cfg.ny
|
|
)
|
|
cell_idx = np.asarray(cells_arr, dtype=np.int64)
|
|
ux_rav = ux.ravel().astype(np.float64)
|
|
uy_rav = uy.ravel().astype(np.float64)
|
|
|
|
sensor_ux_mean = float(np.mean(ux_rav[cell_idx]))
|
|
sensor_uy_mean = float(np.mean(uy_rav[cell_idx]))
|
|
|
|
sensor_reading = sim.read_sensor(sid)
|
|
sensor_reading_x = float(sensor_reading[0])
|
|
sensor_reading_y = float(sensor_reading[1])
|
|
|
|
diff_ux = abs(sensor_reading_x - sensor_ux_mean)
|
|
diff_uy = abs(sensor_reading_y - sensor_uy_mean)
|
|
passed = diff_ux < 1e-4 and diff_uy < 1e-4
|
|
if not passed:
|
|
all_pass = False
|
|
|
|
results[f"sensor_{sid}_pos{pos}"] = {
|
|
"sensor_reading": [sensor_reading_x, sensor_reading_y],
|
|
"manual_average": [sensor_ux_mean, sensor_uy_mean],
|
|
"diff": [float(diff_ux), float(diff_uy)],
|
|
"n_cells": int(len(cells_arr)),
|
|
"pass": bool(passed),
|
|
}
|
|
status = "PASS" if passed else "FAIL"
|
|
print(
|
|
f" Sensor {sid} @ {pos}: "
|
|
f"reading=({sensor_reading_x:.8f},{sensor_reading_y:.8f}) "
|
|
f"manual=({sensor_ux_mean:.8f},{sensor_uy_mean:.8f}) "
|
|
f"diff=({diff_ux:.2e},{diff_uy:.2e}) "
|
|
f"cells={len(cells_arr)} [{status}]"
|
|
)
|
|
|
|
sim.close()
|
|
summary = {"all_pass": bool(all_pass), "results": results}
|
|
print(f"\nSensor accuracy: {'ALL PASS' if all_pass else 'SOME FAILED'}")
|
|
return summary
|
|
|
|
|
|
if __name__ == "__main__":
|
|
result = test_sensor_accuracy()
|
|
sys.exit(0 if result["all_pass"] else 1)
|