Skills › Scientific Agent Skills › fluidsim › budget_summary.py
budget_summary.py
415 lines · skills/fluidsim/scripts/budget_summary.py at 5f2bc03
1#!/usr/bin/env python3
2"""Summarize bounded FluidSim scalar and spectral diagnostics."""
4from __future__ import annotations
6import argparse
7import math
8import re
9from pathlib import Path
10from typing import Any
12try:
13 from ._common import (
14 MAX_FILES,
15 MAX_RECORDS,
16 MAX_TEXT_BYTES,
17 ToolError,
18 bounded_int,
19 checked_input,
20 checked_root,
21 emit_json,
22 fail_json,
23 iter_local_files,
24 relative_display,
25 strict_json_loads,
26 )
27except ImportError: # Direct script execution.
28 from _common import (
29 MAX_FILES,
30 MAX_RECORDS,
31 MAX_TEXT_BYTES,
32 ToolError,
33 bounded_int,
34 checked_input,
35 checked_root,
36 emit_json,
37 fail_json,
38 iter_local_files,
39 relative_display,
40 strict_json_loads,
41 )
44TOOL = "fluidsim-budget-summary"
45_ASSIGNMENT = re.compile(
46 r"^\s*([A-Za-z][A-Za-z0-9_.-]{0,127})\s*=\s*"
47 r"([-+]?(?:(?:\d+(?:\.\d*)?|\.\d+)(?:[Ee][-+]?\d+)?|nan|inf(?:inity)?))\s*$",
48 re.IGNORECASE,
49)
50_SPECTRAL_PREFIXES = (
51 "budget",
52 "diss",
53 "flux",
54 "forcing",
55 "spectr",
56 "transfer",
57)
58_HDF5_SUFFIXES = (".h5", ".hdf5", ".nc")
61class OnlineStats:
62 """Constant-memory finite scalar aggregation."""
64 def __init__(self) -> None:
65 self.count = 0
66 self.first: float | None = None
67 self.last: float | None = None
68 self.minimum = math.inf
69 self.maximum = -math.inf
70 self.mean = 0.0
71 self.nonfinite_count = 0
73 def add(self, value: Any) -> None:
74 if isinstance(value, bool) or not isinstance(value, (int, float)):
75 return
76 try:
77 number = float(value)
78 except OverflowError:
79 self.nonfinite_count += 1
80 return
81 if not math.isfinite(number):
82 self.nonfinite_count += 1
83 return
84 if self.first is None:
85 self.first = number
86 self.last = number
87 self.minimum = min(self.minimum, number)
88 self.maximum = max(self.maximum, number)
89 self.count += 1
90 # Weighted form avoids overflowing a same-sign finite running sum.
91 self.mean = self.mean * ((self.count - 1) / self.count) + number / self.count
93 def report(self) -> dict[str, Any]:
94 return {
95 "count": self.count,
96 "first": self.first,
97 "last": self.last,
98 "max": self.maximum if self.count else None,
99 "mean": self.mean if self.count and math.isfinite(self.mean) else None,
100 "min": self.minimum if self.count else None,
101 "nonfinite_count": self.nonfinite_count,
102 }
105def _iter_text(path: Path, *, max_records: int):
106 if path.stat().st_size > MAX_TEXT_BYTES:
107 raise ToolError("scalar diagnostic exceeds the text-size limit")
108 consumed = 0
109 with path.open("rb") as handle:
110 for line_number, raw in enumerate(handle, start=1):
111 if line_number > max_records:
112 raise ToolError("scalar diagnostic record limit exceeded")
113 consumed += len(raw)
114 if consumed > MAX_TEXT_BYTES or len(raw) > 1024**2:
115 raise ToolError("scalar diagnostic line/byte limit exceeded")
116 if b"\x00" in raw:
117 raise ToolError("NUL byte in scalar diagnostic")
118 try:
119 yield line_number, raw.decode("utf-8").rstrip("\r\n")
120 except UnicodeDecodeError as exc:
121 raise ToolError("scalar diagnostic is not UTF-8") from exc
124def summarize_scalar_file(path: Path, *, max_records: int) -> dict[str, Any]:
125 metrics: dict[str, OnlineStats] = {}
126 parsed_lines = 0
127 json_lines = path.name.casefold().endswith(".json")
128 for _, line in _iter_text(path, max_records=max_records):
129 if not line.strip():
130 continue
131 if json_lines:
132 record = strict_json_loads(line)
133 if not isinstance(record, dict):
134 raise ToolError("spatial_means JSON line must be an object")
135 parsed_lines += 1
136 for key, value in record.items():
137 if not isinstance(key, str) or not re.fullmatch(
138 r"[A-Za-z][A-Za-z0-9_.-]{0,127}", key
139 ):
140 continue
141 metrics.setdefault(key, OnlineStats()).add(value)
142 else:
143 matches = [_ASSIGNMENT.fullmatch(part) for part in line.split(";") if part.strip()]
144 if not matches or any(match is None for match in matches):
145 continue
146 parsed_lines += 1
147 for match in matches:
148 key, text = match.groups()
149 metrics.setdefault(key, OnlineStats()).add(float(text))
150 if len(metrics) > 1024:
151 raise ToolError("scalar metric-count limit exceeded")
152 return {
153 "format": "json-lines" if json_lines else "FluidSim key-value text",
154 "metrics": {
155 key: metrics[key].report()
156 for key in sorted(metrics)
157 if metrics[key].count or metrics[key].nonfinite_count
158 },
159 "parsed_lines": parsed_lines,
160 "raw_records_emitted": False,
161 "averaging": "finite saved-sample mean, not time-weighted or budget closure",
162 }
165def _bounded_slice(shape: tuple[int, ...], maximum: int) -> tuple[Any, ...]:
166 if not shape:
167 return ()
168 if len(shape) == 1:
169 return (slice(0, min(shape[0], maximum)),)
170 slices: list[Any] = [slice(max(0, shape[0] - 1), shape[0])]
171 remaining = maximum
172 for size in shape[1:]:
173 take = min(size, max(1, remaining))
174 slices.append(slice(0, take))
175 remaining = max(1, remaining // max(1, take))
176 return tuple(slices)
179def _summarize_values(values: Any, *, complete_record: bool) -> dict[str, Any]:
180 flat = values.reshape(-1)
181 if flat.dtype.kind == "c":
182 flat = abs(flat)
183 statistic = "magnitude"
184 else:
185 statistic = "value"
186 finite = flat[flat == flat]
187 finite = finite[abs(finite) != math.inf]
188 count = int(finite.size)
190 def converted(operation: str) -> float | None:
191 if not count:
192 return None
193 value = float(getattr(finite, operation)())
194 return value if math.isfinite(value) else None
196 return {
197 "complete_latest_record": complete_record,
198 "finite_values": count,
199 "max": converted("max"),
200 "mean": converted("mean"),
201 "min": converted("min"),
202 "sampled_values": int(flat.size),
203 "statistic": statistic,
204 "sum": converted("sum") if complete_record else None,
205 }
208def summarize_spectral_file(
209 path: Path, *, max_datasets: int, max_values: int
210) -> dict[str, Any]:
211 try:
212 import h5py # Lazy optional dependency.
213 except ImportError as exc:
214 raise ToolError(
215 "spectral summary requires optional h5py from the pinned environment"
216 ) from exc
218 results: dict[str, Any] = {}
219 time_range = None
220 external_links = 0
221 try:
222 with h5py.File(path, "r") as handle:
223 times_link = handle.get("times", getlink=True)
224 if isinstance(times_link, h5py.HardLink) and isinstance(handle["times"], h5py.Dataset):
225 times = handle["times"]
226 if times.is_virtual or times.external:
227 raise ToolError("external or virtual dataset storage is not read")
228 if times.ndim == 1 and times.shape[0] and times.dtype.kind in "iuf":
229 time_range = [
230 float(times[0]),
231 float(times[times.shape[0] - 1]),
232 ]
233 if not all(math.isfinite(value) for value in time_range):
234 raise ToolError("spectral timestamps must be finite")
235 stack = [(handle, "/")]
236 seen: set[int] = set()
237 while stack:
238 group, prefix = stack.pop()
239 try:
240 address = int(h5py.h5o.get_info(group.id).addr)
241 except (AttributeError, TypeError, ValueError):
242 address = hash(group.id)
243 if address in seen:
244 continue
245 seen.add(address)
246 for name in sorted(group.keys(), reverse=True):
247 link = group.get(name, getlink=True)
248 if isinstance(link, h5py.ExternalLink):
249 external_links += 1
250 continue
251 if isinstance(link, h5py.SoftLink):
252 continue
253 obj = group.get(name, getlink=False)
254 object_path = f"/{name}" if prefix == "/" else f"{prefix}/{name}"
255 if isinstance(obj, h5py.Group):
256 stack.append((obj, object_path))
257 continue
258 if not isinstance(obj, h5py.Dataset):
259 continue
260 leaf = name.casefold()
261 if (
262 obj.dtype.kind not in "iufc"
263 or not leaf.startswith(_SPECTRAL_PREFIXES)
264 ):
265 continue
266 if len(results) >= max_datasets:
267 raise ToolError("spectral dataset-count limit exceeded")
268 if obj.is_virtual or obj.external:
269 raise ToolError("external or virtual dataset storage is not read")
270 shape = tuple(int(size) for size in obj.shape)
271 if not shape or math.prod(shape) == 0:
272 continue
273 selection = _bounded_slice(shape, max_values)
274 values = obj[selection]
275 latest_record_size = (
276 math.prod(shape[1:]) if len(shape) > 1 else shape[0]
277 )
278 results[object_path[:300]] = {
279 "dtype": str(obj.dtype)[:100],
280 "shape": list(shape),
281 **_summarize_values(
282 values,
283 complete_record=latest_record_size <= max_values,
284 ),
285 }
286 except OSError as exc:
287 return {
288 "datasets": {},
289 "external_links_followed": False,
290 "external_links_not_followed": external_links,
291 "hdf5_readable": False,
292 "message": str(exc)[:300],
293 "time_range": None,
294 }
295 return {
296 "datasets": dict(sorted(results.items())),
297 "external_links_followed": False,
298 "external_links_not_followed": external_links,
299 "hdf5_readable": True,
300 "time_range": time_range,
301 }
304def summarize(
305 path: Path,
306 *,
307 root: Path,
308 max_files: int,
309 max_records: int,
310 max_datasets: int,
311 max_values: int,
312) -> dict[str, Any]:
313 files = iter_local_files(
314 path, suffixes=None, max_files=max_files, recursive=True
315 )
316 scalar: list[dict[str, Any]] = []
317 spectra: list[dict[str, Any]] = []
318 for file_path in files:
319 lowered = file_path.name.casefold()
320 if lowered in {"spatial_means.txt", "spatial_means.json"}:
321 scalar.append(
322 {
323 "path": relative_display(file_path, root),
324 "summary": summarize_scalar_file(
325 file_path, max_records=max_records
326 ),
327 }
328 )
329 elif lowered.endswith(_HDF5_SUFFIXES) and lowered.startswith(
330 ("spectra", "spect_energy", "spectra_multidim")
331 ):
332 spectra.append(
333 {
334 "path": relative_display(file_path, root),
335 "summary": summarize_spectral_file(
336 file_path,
337 max_datasets=max_datasets,
338 max_values=max_values,
339 ),
340 }
341 )
342 return {
343 "arrays_fully_loaded": False,
344 "interpretation": (
345 "Descriptive diagnostics only. Budget closure, conservation, spectral "
346 "resolution, stationarity, and convergence require solver-aware checks."
347 ),
348 "network_used": False,
349 "numerical_convergence_established": False,
350 "ok": bool(scalar or spectra),
351 "physical_validity_established": False,
352 "scalar_outputs": scalar,
353 "spectral_outputs": spectra,
354 "tool": TOOL,
355 }
358def build_parser() -> argparse.ArgumentParser:
359 parser = argparse.ArgumentParser(
360 description=(
361 "Summarize bounded local spatial means and latest spectral/budget "
362 "hyperslabs. Large arrays are never loaded in full."
363 )
364 )
365 parser.add_argument("--path", required=True)
366 parser.add_argument("--root", default=".")
367 parser.add_argument("--max-files", type=int, default=128)
368 parser.add_argument("--max-records", type=int, default=200_000)
369 parser.add_argument("--max-datasets", type=int, default=256)
370 parser.add_argument("--max-values-per-dataset", type=int, default=4_096)
371 return parser
374def main(argv: list[str] | None = None) -> int:
375 args = build_parser().parse_args(argv)
376 try:
377 max_files = bounded_int(
378 args.max_files, name="max_files", minimum=1, maximum=MAX_FILES
379 )
380 max_records = bounded_int(
381 args.max_records,
382 name="max_records",
383 minimum=1,
384 maximum=MAX_RECORDS,
385 )
386 max_datasets = bounded_int(
387 args.max_datasets,
388 name="max_datasets",
389 minimum=1,
390 maximum=10_000,
391 )
392 max_values = bounded_int(
393 args.max_values_per_dataset,
394 name="max_values_per_dataset",
395 minimum=1,
396 maximum=65_536,
397 )
398 root = checked_root(args.root)
399 path = checked_input(args.path, root=root, kind="any")
400 report = summarize(
401 path,
402 root=root,
403 max_files=max_files,
404 max_records=max_records,
405 max_datasets=max_datasets,
406 max_values=max_values,
407 )
408 emit_json(report)
409 return 0 if report["ok"] else 2
410 except (OSError, ToolError, UnicodeError) as exc:
411 return fail_json(TOOL, exc)
414if __name__ == "__main__":
415 raise SystemExit(main())