#!/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 hashlib 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" METADATA_FILENAME = "airfrans_case_metadata.json" MANIFEST_FILENAME = "airfrans_case_manifest.json" REQUIRED_COMPARISON_FIELDS = ("U", "p", "phi", "nut", "k", "omega") REQUIRED_SOURCE_FILES = ( "system/controlDict", "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", ) SOURCE_PARAMETER_FILES = ( "system/controlDict", ) COPIED_SOURCE_FILES = ( "system/fvSchemes", "system/fvSolution", "system/blockMeshDict", "0/U", "0/p", "0/nut", "0/k", "0/omega", "constant/transportProperties", "constant/turbulenceProperties", ) TOPOLOGY_EVIDENCE_KEYS = ( "n_points", "n_faces", "n_internal_faces", "n_cells", "patches", "topology_sha256", "geometry_sha256", ) 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 sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def file_record(root: Path, relative: str) -> dict[str, object]: path = root / relative stat = path.stat() return { "path": relative, "size": stat.st_size, "sha256": sha256_file(path), } def case_file_inventory(root: Path, *, exclude: tuple[str, ...] = ()) -> list[dict[str, object]]: excluded = set(exclude) records = [] for path in sorted(root.rglob("*")): if not path.is_file(): continue relative = path.relative_to(root).as_posix() if relative in excluded: continue records.append(file_record(root, relative)) return records def source_input_inventory(src: Path) -> list[dict[str, object]]: seen: set[str] = set() records: list[dict[str, object]] = [] def add(relative: str) -> None: if relative in seen: return seen.add(relative) records.append(file_record(src, relative)) for relative in SOURCE_PARAMETER_FILES + COPIED_SOURCE_FILES: add(relative) for path in sorted((src / "constant/polyMesh").rglob("*")): if path.is_file(): add(path.relative_to(src).as_posix()) return records def inventory_digest(records: list[dict[str, object]]) -> str: digest = hashlib.sha256() for record in records: payload = json.dumps(record, sort_keys=True, separators=(",", ":")).encode("utf-8") digest.update(len(payload).to_bytes(8, "little")) digest.update(payload) return digest.hexdigest() def build_case_manifest(src: Path, dst: Path, meta: AirfransCaseMetadata) -> dict[str, object]: source_records = source_input_inventory(src) prepared_records = case_file_inventory(dst, exclude=(MANIFEST_FILENAME,)) return { "schema_version": 1, "simulation": meta.simulation, "source_case": { "path": str(src), "read_only": True, "required_files": source_records, "required_files_sha256": inventory_digest(source_records), }, "prepared_case": { "path": str(dst), "openfoam_version": 14, "migrated_end_time": meta.migrated_end_time, "files": prepared_records, "files_sha256": inventory_digest(prepared_records), }, "comparison_contract": { "fields": list(REQUIRED_COMPARISON_FIELDS), "oracle_output": { "producer": "foamRun -solver incompressibleFluid -noFunctionObjects", "time": str(meta.migrated_end_time), }, "repository_outputs": { "run_one": "foam_stepper run_one_pimple_iteration", "split": "foam_stepper split solver stages", "time": str(meta.migrated_end_time), }, "mesh_identity_evidence": list(TOPOLOGY_EVIDENCE_KEYS), }, "metadata": asdict(meta), } def write_case_manifest(src: Path, dst: Path, meta: AirfransCaseMetadata) -> dict[str, object]: manifest = build_case_manifest(src, dst, meta) (dst / MANIFEST_FILENAME).write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") return manifest def load_case_manifest(case: Path) -> dict[str, object]: return json.loads((case / MANIFEST_FILENAME).read_text()) 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) missing = [relative for relative in REQUIRED_SOURCE_FILES 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 COPIED_SOURCE_FILES: 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 / METADATA_FILENAME).write_text(json.dumps(asdict(meta), indent=2, sort_keys=True) + "\n") write_case_manifest(src, dst, meta) 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}") print(f"manifest={args.dest / MANIFEST_FILENAME}") if __name__ == "__main__": main()