#!/usr/bin/env python3 """Prepare a v14-compatible AirfRANS raw OpenFOAM case for the Python stepper. The source raw data lives in the sibling ../airfrans repo and must not be modified. This helper copies only the files needed for one local v14 OpenFOAM/stepper iteration: mesh, initial fields, fvSchemes, and fvSolution. """ from __future__ import annotations import argparse import json import math import re import shutil from dataclasses import asdict, dataclass from pathlib import Path ROOT = Path(__file__).resolve().parents[1] RAW_ROOT = ROOT.parent / "airfrans/data/raw/OF_dataset" DEFAULT_SIMULATION = "airFoil2D_SST_93.213_3.79_0.418_0.0_9.665" DEFAULT_SOURCE = RAW_ROOT / DEFAULT_SIMULATION DEFAULT_DEST = ROOT / "tmp/airfrans_stepper_case" / f"{DEFAULT_SIMULATION}_v14" FLOAT_RE = re.compile(r"[-+]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][-+]?\d+)?") @dataclass(frozen=True) class AirfransCaseMetadata: simulation: str source: str u_inf: float velocity: tuple[float, float, float] nu: float rho_inf: float reynolds: float mach: float alpha_deg: float drag_dir: tuple[float, float, float] lift_dir: tuple[float, float, float] source_end_time: int migrated_end_time: int def read_text(path: Path) -> str: return path.read_text(errors="replace") def assignment(text: str, key: str) -> str: match = re.search(rf"^\s*{re.escape(key)}\s+([^;]+);", text, flags=re.MULTILINE) if not match: raise ValueError(f"missing assignment {key!r}") return match.group(1).strip() def vector_assignment(text: str, key: str) -> tuple[float, float, float]: values = [float(value) for value in FLOAT_RE.findall(assignment(text, key))] if len(values) != 3: raise ValueError(f"expected 3-vector for {key!r}, got {values!r}") return (values[0], values[1], values[2]) def copy_required_file(src_root: Path, dst_root: Path, relative: str) -> None: src = src_root / relative dst = dst_root / relative if not src.exists(): raise FileNotFoundError(src) dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) def copy_required_tree(src_root: Path, dst_root: Path, relative: str) -> None: src = src_root / relative dst = dst_root / relative if not src.exists(): raise FileNotFoundError(src) if dst.exists(): shutil.rmtree(dst) dst.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(src, dst) def header(class_name: str, location: str, object_name: str) -> str: location_line = f' location "{location}";\n' if location else "" return f"""/*--------------------------------*- C++ -*----------------------------------*\\ ========= | \\ / F ield | OpenFOAM: The Open Source CFD Toolbox \\ / O peration | Website: https://openfoam.org \\ / A nd | Version: 14 \\/ M anipulation | \\*---------------------------------------------------------------------------*/ FoamFile {{ format ascii; class {class_name}; {location_line} object {object_name}; }} // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * // """ def write_control_dict(dst: Path, *, u_inf: float, end_time: int) -> None: content = header("dictionary", "system", "controlDict") + f"""Uinf {u_inf:.17g}; solver incompressibleFluid; startFrom startTime; startTime 0; stopAt endTime; endTime {end_time}; deltaT 1; writeControl timeStep; writeInterval {end_time}; purgeWrite 0; writeFormat ascii; writePrecision 17; writeCompression off; timeFormat general; timePrecision 6; runTimeModifiable true; // Function objects are intentionally disabled for first-step parity; run // foamRun/foam_stepper with -noFunctionObjects and parse archived raw force // outputs separately in the notebook. // ************************************************************************* // """ (dst / "system/controlDict").write_text(content) def write_physical_properties(dst: Path, *, nu: float, rho_inf: float) -> None: content = header("dictionary", "constant", "physicalProperties") + f"""viscosityModel constant; rho {rho_inf:.17g}; nu {nu:.17g}; // ************************************************************************* // """ (dst / "constant/physicalProperties").write_text(content) def write_momentum_transport(dst: Path) -> None: content = header("dictionary", "constant", "momentumTransport") + """simulationType RAS; RAS { model kOmegaSST; turbulence on; } // ************************************************************************* // """ (dst / "constant/momentumTransport").write_text(content) def metadata_from_source(src: Path, migrated_end_time: int) -> AirfransCaseMetadata: control = read_text(src / "system/controlDict") transport = read_text(src / "constant/transportProperties") u_field = read_text(src / "0/U") u_inf = float(assignment(control, "Uinf")) velocity = vector_assignment(u_field, "field") nu = float(assignment(transport, "nu")) rho_inf = float(assignment(control, "rhoInf")) drag_dir = vector_assignment(control, "dragDir") lift_dir = vector_assignment(control, "liftDir") source_end_time = int(float(assignment(control, "endTime"))) alpha_deg = math.degrees(math.atan2(drag_dir[1], drag_dir[0])) return AirfransCaseMetadata( simulation=src.name, source=str(src), u_inf=u_inf, velocity=velocity, nu=nu, rho_inf=rho_inf, reynolds=u_inf / nu, mach=u_inf / 346.1, alpha_deg=alpha_deg, drag_dir=drag_dir, lift_dir=lift_dir, source_end_time=source_end_time, migrated_end_time=migrated_end_time, ) def prepare_case(src: Path, dst: Path, *, end_time: int = 1) -> AirfransCaseMetadata: src = src.resolve() dst = dst.resolve() if not src.exists(): raise FileNotFoundError(src) required = [ "system/fvSchemes", "system/fvSolution", "system/blockMeshDict", "0/U", "0/p", "0/nut", "0/k", "0/omega", "constant/transportProperties", "constant/turbulenceProperties", "constant/polyMesh/boundary", "constant/polyMesh/points.gz", "constant/polyMesh/faces.gz", "constant/polyMesh/owner.gz", "constant/polyMesh/neighbour.gz", ] missing = [relative for relative in required if not (src / relative).exists()] if missing: raise FileNotFoundError(f"missing required AirfRANS files: {missing}") if dst.exists(): shutil.rmtree(dst) dst.mkdir(parents=True) copy_required_tree(src, dst, "constant/polyMesh") for relative in [ "system/fvSchemes", "system/fvSolution", "system/blockMeshDict", "0/U", "0/p", "0/nut", "0/k", "0/omega", "constant/transportProperties", "constant/turbulenceProperties", ]: copy_required_file(src, dst, relative) meta = metadata_from_source(src, end_time) write_control_dict(dst, u_inf=meta.u_inf, end_time=end_time) write_physical_properties(dst, nu=meta.nu, rho_inf=meta.rho_inf) write_momentum_transport(dst) # Keep the original dictionaries for audit without letting v14 select them. shutil.move(dst / "constant/transportProperties", dst / "constant/transportProperties.v2112") shutil.move(dst / "constant/turbulenceProperties", dst / "constant/turbulenceProperties.v2112") (dst / "airfrans_case_metadata.json").write_text(json.dumps(asdict(meta), indent=2, sort_keys=True) + "\n") return meta def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE) parser.add_argument("--dest", type=Path, default=DEFAULT_DEST) parser.add_argument("--end-time", type=int, default=1) args = parser.parse_args() meta = prepare_case(args.source, args.dest, end_time=args.end_time) print(json.dumps(asdict(meta), indent=2, sort_keys=True)) print(f"prepared={args.dest}") if __name__ == "__main__": main()