#!/usr/bin/env python3 """Scenario checks for the Python-driven OpenFOAM observability stepper.""" from __future__ import annotations import os import shutil import subprocess import sys from pathlib import Path def _drop_ambient_pythonpath() -> None: pythonpath = os.environ.pop("PYTHONPATH", "") for entry in pythonpath.split(os.pathsep): if not entry: continue while entry in sys.path: sys.path.remove(entry) _drop_ambient_pythonpath() import numpy as np from openfoam_env import apply_openfoam_env, openfoam_env ROOT = Path(__file__).resolve().parents[1] TUTORIAL = ROOT / "OpenFOAM-14/tutorials/incompressibleFluid/venturiTube" WORK = ROOT / "tmp/python_stepper_verify" LOAD_CASE = WORK / "load_case" TRANSFORM_CASE = WORK / "transform_case" PY_EQ_CASE = WORK / "python_equivalence_case" FOAM_EQ_CASE = WORK / "foam_equivalence_case" def run(cmd: list[str]) -> None: subprocess.run(cmd, cwd=ROOT, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT, env=openfoam_env()) def patch_control_dict(case: Path, *, latest: bool = False) -> None: path = case / "system/controlDict" text = path.read_text() replacements = { "startFrom startTime;": "startFrom latestTime;" if latest else "startFrom startTime;", "endTime 1000;": "endTime 1;", "writeInterval 50;": "writeInterval 1;", } for old, new in replacements.items(): text = text.replace(old, new) path.write_text(text) def prepare_case(dst: Path) -> None: if dst.exists(): shutil.rmtree(dst) shutil.copytree(TUTORIAL, dst, ignore=shutil.ignore_patterns("processor*", "postProcessing", "*.log")) for orig in (dst / "0").glob("*.orig"): shutil.copyfile(orig, orig.with_suffix("")) patch_control_dict(dst) run(["blockMesh", "-case", str(dst)]) run(["createZones", "-case", str(dst)]) def import_foam(): apply_openfoam_env() import foam_stepper as foam return foam def load_scenario() -> None: foam = import_foam() case = foam.Case(LOAD_CASE) stepper = case.make_stepper() mesh = stepper.mesh() fields = stepper.fields() assert mesh.n_cells == 57600 assert mesh.n_internal_faces == fields.phi.internal.shape[0] assert fields.U.internal.shape == (mesh.n_cells, 3) assert fields.p.internal.shape == (mesh.n_cells,) assert fields.phi.internal.shape == (mesh.n_internal_faces,) assert {patch.name for patch in mesh.boundary} == {"inlet", "outlet", "walls"} try: stepper.assemble_pressure_matrix() except foam.OpenFoamError as exc: assert "Momentum matrix is not assembled" in str(exc) else: raise AssertionError("assemble_pressure_matrix succeeded before momentum assembly") def transformation_scenario() -> None: foam = import_foam() stepper = foam.Case(TRANSFORM_CASE).make_stepper() stepper.pre_solve() stepper.advance_time() begin = stepper.begin_pimple_iteration() assert begin.outputs["active"] is True terms = stepper.assemble_momentum_terms() names = {term["name"] for term in terms.outputs["terms"]} assert { "assemble_momentum_ddt", "assemble_momentum_div_phi_U", "assemble_momentum_MRF_DDt", "assemble_momentum_divDevSigma", "assemble_momentum_sources", "assemble_momentum_pressure_rhs_grad_p", }.issubset(names) UEqn = stepper.assemble_momentum_matrix().outputs["UEqn"] assert UEqn.diag.shape == (57600,) assert UEqn.psi.internal.shape == (57600, 3) assert UEqn.source.shape == (57600, 3) assert UEqn.H is not None stepper.relax_matrix() stepper.constrain_matrix() momentum = stepper.solve_momentum() assert momentum.outputs["performance"].field_name == "U" pressure_inputs = stepper.compute_pressure_inputs() assert pressure_inputs.outputs["rAU"].internal.shape == (57600,) assert pressure_inputs.outputs["HbyA"].internal.shape == (57600, 3) assert pressure_inputs.outputs["phiHbyA"].internal.shape == (170624,) pEqn = stepper.assemble_pressure_matrix().outputs["pEqn"] assert pEqn.diag.shape == (57600,) assert pEqn.psi.internal.shape == (57600,) assert pEqn.flux is not None pressure = stepper.solve_pressure() assert pressure.name == "solve_pEqn" assert pressure.outputs["performance"].field_name == "p" assert pressure.outputs["p"].internal.shape == (57600,) assert pressure.outputs["phi"].internal.shape == (170624,) correction = stepper.correct_velocity_pressure_flux() assert correction.outputs["U"].internal.shape == (57600, 3) assert correction.outputs["phi"].internal.shape == (170624,) def equivalence_scenario() -> None: foam = import_foam() run(["foamRun", "-case", str(FOAM_EQ_CASE), "-solver", "incompressibleFluid", "-noFunctionObjects"]) patch_control_dict(FOAM_EQ_CASE, latest=True) baseline = foam.Case(FOAM_EQ_CASE).make_stepper().fields() result = foam.Case(PY_EQ_CASE).make_stepper().run_one_pimple_iteration() actual = result.outputs["fields"] comparisons = { "U": (actual["U"].internal, baseline.U.internal), "p": (actual["p"].internal, baseline.p.internal), "phi": (actual["phi"].internal, baseline.phi.internal), } for name, (lhs, rhs) in comparisons.items(): if lhs.shape != rhs.shape: raise AssertionError(f"{name} shape mismatch: {lhs.shape} != {rhs.shape}") max_abs = float(np.max(np.abs(lhs - rhs))) if not np.allclose(lhs, rhs, rtol=1e-7, atol=1e-7): raise AssertionError(f"{name} mismatch: max_abs={max_abs}") print(f"equivalence {name}: max_abs={max_abs:.3e}") graph = result.outputs["graph"] graph_names = {entry.name for entry in graph} assert {"pre_solve", "advance_time", "assemble_UEqn", "solve_UEqn", "assemble_pEqn", "solve_pEqn"}.issubset(graph_names) def main() -> None: if len(sys.argv) == 2: scenarios = { "load": load_scenario, "transformation": transformation_scenario, "equivalence": equivalence_scenario, } scenarios[sys.argv[1]]() return if WORK.exists(): shutil.rmtree(WORK) WORK.mkdir(parents=True) for case_dir in (LOAD_CASE, TRANSFORM_CASE, PY_EQ_CASE, FOAM_EQ_CASE): prepare_case(case_dir) for scenario in ("load", "transformation", "equivalence"): subprocess.run([sys.executable, __file__, scenario], cwd=ROOT, check=True, env=openfoam_env()) print("python stepper verification passed") if __name__ == "__main__": main()