Skills › Scientific Agent Skills › fluidsim › simulation_dry_run.py
simulation_dry_run.py
290 lines · skills/fluidsim/scripts/simulation_dry_run.py at 5f2bc03
1#!/usr/bin/env python3
2"""Generate a reviewable, opt-in FluidSim launch script without running it."""
4from __future__ import annotations
6import argparse
7import hashlib
8import json
9from pathlib import Path
10from typing import Any
12try:
13 from ._common import (
14 ToolError,
15 atomic_write,
16 checked_input,
17 checked_output,
18 emit_json,
19 fail_json,
20 load_json,
21 )
22 from ._schema import SOLVER_IMPORTS, flatten_parameters, normalized_copy
23except ImportError: # Direct script execution.
24 from _common import (
25 ToolError,
26 atomic_write,
27 checked_input,
28 checked_output,
29 emit_json,
30 fail_json,
31 load_json,
32 )
33 from _schema import SOLVER_IMPORTS, flatten_parameters, normalized_copy
36TOOL = "fluidsim-simulation-dry-run"
39def _literal(value: Any) -> str:
40 """Render validated JSON scalars as fixed Python literals."""
42 if value is None or isinstance(value, (bool, int, float, str)):
43 return repr(value)
44 raise ToolError("only JSON scalar parameter values can be rendered")
47def render_script(config: dict[str, Any]) -> str:
48 """Render a script with a dry-run default and explicit execution gate."""
50 config = normalized_copy(config)
51 init_type = config["parameters"].get("init_fields", {}).get("type")
52 if init_type in {"in_script", "from_simul"}:
53 raise ToolError("in_script/from_simul initialization requires a separately implemented scientific script")
54 forcing = config["parameters"].get("forcing", {})
55 if forcing.get("enable") and forcing.get("type") in {"in_script", "in_script_coarse", "pseudo_spectral"}:
56 raise ToolError("this forcing requires an explicitly implemented forcing callback")
57 restart = config["provenance"].get("restart")
58 if init_type == "from_file":
59 if not restart or restart["path"] != config["parameters"]["init_fields"]["from_file"]["path"]:
60 raise ToolError("from_file requires matching restart path and SHA-256 provenance")
61 solver = config["solver"]
62 module_name = SOLVER_IMPORTS[solver]
63 assignments = []
64 for path, value in flatten_parameters(config["parameters"]):
65 dotted = ".".join(("params", *path))
66 assignments.append(f" {dotted} = {_literal(value)}")
68 resources = config["resources"]
69 execution = config["execution"]
70 provenance = config["provenance"]
71 plan = {
72 "config_id": provenance["config_id"],
73 "cpu_cores": resources["cpu_cores"],
74 "disk_gib": resources["disk_gib"],
75 "fluidfft": provenance["fluidfft"],
76 "fluidsim": provenance["fluidsim"],
77 "max_output_files": resources["max_output_files"],
78 "mode": execution["mode"],
79 "mpi_ranks": resources["mpi_ranks"],
80 "output_root": execution["output_root"],
81 "ram_gib": resources["ram_gib"],
82 "solver": solver,
83 "threads_per_rank": resources["threads_per_rank"],
84 "wall_time_minutes": resources["wall_time_minutes"],
85 }
86 plan_json = json.dumps(plan, allow_nan=False, sort_keys=True)
87 assignment_block = "\n".join(assignments)
88 expected_ranks = int(resources["mpi_ranks"])
89 return f'''#!/usr/bin/env python3
90"""Generated FluidSim {provenance["fluidsim"]} plan. Dry-run unless explicitly approved."""
92from __future__ import annotations
94import argparse
95import hashlib
96import json
97import os
98from importlib.metadata import version
99from pathlib import Path
102PLAN = json.loads({plan_json!r})
103CONFIG_ID = {provenance["config_id"]!r}
104OUTPUT_ROOT = {execution["output_root"]!r}
105EXPECTED_RANKS = {expected_ranks}
106MODE = {execution["mode"]!r}
107RESTART = {restart!r}
110def _detected_mpi_size() -> int:
111 for name in ("OMPI_COMM_WORLD_SIZE", "PMI_SIZE", "PMIX_SIZE", "MPI_LOCALNRANKS"):
112 value = os.environ.get(name)
113 if value and value.isdecimal():
114 return int(value)
115 return 1
118def _prepare_output_root() -> Path:
119 base = Path(__file__).resolve().parent
120 destination = base / OUTPUT_ROOT
121 if destination.exists() and destination.is_symlink():
122 raise RuntimeError("refusing a symlink output root")
123 destination.mkdir(mode=0o700, exist_ok=True)
124 resolved = destination.resolve(strict=True)
125 if resolved.parent != base:
126 raise RuntimeError("output root escaped the script directory")
127 os.environ["FLUIDSIM_PATH"] = str(resolved)
128 return resolved
131def run() -> None:
132 size = _detected_mpi_size()
133 if MODE == "serial" and size != 1:
134 raise RuntimeError("serial plan detected an MPI launcher")
135 if MODE == "mpi-preview" and size != EXPECTED_RANKS:
136 raise RuntimeError("launch manually with exactly the reviewed MPI rank count")
137 for package in ("fluidsim", "fluidfft"):
138 if version(package) != PLAN[package]:
139 raise RuntimeError("installed package version differs from the reviewed plan: " + package)
140 for name in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS", "NUMEXPR_NUM_THREADS"):
141 os.environ[name] = {str(resources["threads_per_rank"])!r}
142 restart_path = None
143 if RESTART:
144 base = Path(__file__).resolve().parent
145 restart_path = base / RESTART["path"]
146 for part in (restart_path, *restart_path.parents):
147 if part.is_symlink():
148 raise RuntimeError("restart path contains a symlink")
149 restart_path = restart_path.resolve(strict=True)
150 if not restart_path.is_relative_to(base):
151 raise RuntimeError("restart path escaped the script directory")
152 if not restart_path.is_file() or restart_path.stat().st_nlink != 1:
153 raise RuntimeError("restart must be a singly linked regular file")
154 if restart_path.stat().st_size > 8 * 1024**3:
155 raise RuntimeError("restart exceeds the reviewed eight-GiB hash limit")
156 digest = hashlib.sha256()
157 with restart_path.open("rb") as handle:
158 for chunk in iter(lambda: handle.read(1024**2), b""):
159 digest.update(chunk)
160 if digest.hexdigest() != RESTART["sha256"]:
161 raise RuntimeError("restart SHA-256 differs from the reviewed checkpoint")
162 _prepare_output_root()
164 import numpy as np
165 from {module_name} import Simul
166 from fluiddyn.util import mpi
168 if mpi.nb_proc != EXPECTED_RANKS:
169 raise RuntimeError("actual MPI rank count differs from the reviewed plan")
171 np.random.seed({int(execution["random_seed"])})
172 params = Simul.create_default_params()
173{assignment_block}
174 if restart_path is not None:
175 params.init_fields.from_file.path = str(restart_path)
176 sim = Simul(params)
177 sim.time_stepping.start()
180def main() -> int:
181 parser = argparse.ArgumentParser(description=__doc__)
182 parser.add_argument("--execute", action="store_true")
183 parser.add_argument("--acknowledge-config-id")
184 args = parser.parse_args()
185 if not args.execute:
186 print(json.dumps({{"dry_run": True, "plan": PLAN}}, indent=2, sort_keys=True))
187 return 0
188 if args.acknowledge_config_id != CONFIG_ID:
189 parser.error("--execute requires the exact reviewed --acknowledge-config-id")
190 run()
191 return 0
194if __name__ == "__main__":
195 raise SystemExit(main())
196'''
199def build_plan(config: dict[str, Any], script: str) -> dict[str, Any]:
200 execution = config["execution"]
201 resources = config["resources"]
202 execute_argv = [
203 "<python>",
204 execution["script_name"],
205 "--execute",
206 "--acknowledge-config-id",
207 config["provenance"]["config_id"],
208 ]
209 mpi_preview = None
210 if execution["mode"] == "mpi-preview":
211 mpi_preview = [
212 "<site-mpi-launcher>",
213 "-n",
214 str(resources["mpi_ranks"]),
215 *execute_argv,
216 ]
217 return {
218 "approval_gate": [
219 "review the validated scientific assumptions and acceptance criteria",
220 "review the resource estimate against CPU, RAM, disk, wall-time, and file limits",
221 "verify FFT/MPI backend on the target host with a tiny pilot",
222 "record the config, script, lock, environment, and restart hashes",
223 ],
224 "commands_executed": False,
225 "dry_run": True,
226 "execute_argv_after_approval": execute_argv,
227 "job_submitted": False,
228 "mpi_launch_preview_only": mpi_preview,
229 "mpi_launched": False,
230 "network_used": False,
231 "physical_validity_established": False,
232 "script": script,
233 "script_sha256": hashlib.sha256(script.encode("utf-8")).hexdigest(),
234 "tool": TOOL,
235 }
238def build_parser() -> argparse.ArgumentParser:
239 parser = argparse.ArgumentParser(
240 description=(
241 "Generate an opt-in FluidSim script from validated local JSON. The "
242 "generator never imports FluidSim, invokes MPI, submits a job, or runs code."
243 )
244 )
245 parser.add_argument("--config", required=True)
246 parser.add_argument("--root", default=".")
247 parser.add_argument(
248 "--output",
249 help="Optional .py file within --root; parent must exist.",
250 )
251 parser.add_argument(
252 "--force",
253 action="store_true",
254 help="Allow replacement of an existing regular output file.",
255 )
256 return parser
259def main(argv: list[str] | None = None) -> int:
260 args = build_parser().parse_args(argv)
261 try:
262 config_path = checked_input(
263 args.config,
264 root=args.root,
265 kind="file",
266 suffixes={".json"},
267 )
268 config = normalized_copy(load_json(config_path))
269 script = render_script(config)
270 report = build_plan(config, script)
271 if args.output:
272 output = checked_output(
273 args.output,
274 root=args.root,
275 suffixes={".py"},
276 force=args.force,
277 )
278 atomic_write(output, script.encode("utf-8"), force=args.force)
279 report["script"] = None
280 report["script_written"] = output.name
281 else:
282 report["script_written"] = None
283 emit_json(report)
284 return 0
285 except (OSError, ToolError, UnicodeError) as exc:
286 return fail_json(TOOL, exc)
289if __name__ == "__main__":
290 raise SystemExit(main())