#!/usr/bin/env python3 """Build a clean source revision and run a serial, fully recorded numerical campaign.""" import argparse import csv import datetime import hashlib import json import math import os from pathlib import Path import platform import shutil import subprocess import time def digest(path): result = hashlib.sha256() with path.open('rb') as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b''): result.update(chunk) return result.hexdigest() def utc(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def json_safe(value): if isinstance(value, float) and not math.isfinite(value): return None if isinstance(value, dict): return {key: json_safe(item) for key, item in value.items()} if isinstance(value, list): return [json_safe(item) for item in value] return value def write_json(path, value): path.write_text(json.dumps(json_safe(value), indent=2, allow_nan=False) + '\n') def read_rows(path): if not path.exists(): return [] with path.open() as stream: return list(csv.DictReader(stream)) def summarize(folder): result = {'newton': read_rows(folder / 'newton.csv'), 'frozen_linear': read_rows(folder / 'frozen_linear.csv'), 'trials': read_rows(folder / 'trials.csv')} meta = folder / 'metadata.txt' if meta.exists(): result['metadata'] = dict(line.split('=', 1) for line in meta.read_text().splitlines() if '=' in line) for stage in ['initial', 'final']: rows = read_rows(folder / f'{stage}_physical.csv') result[stage + '_physical'] = { q: {row['metric']: float(row['value']) for row in rows if row['quadrature'] == q} for q in sorted({row['quadrature'] for row in rows})} return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--source', required=True, type=Path) parser.add_argument('--build', required=True, type=Path) parser.add_argument('--mesh', required=True, type=Path) parser.add_argument('--cases', required=True, type=Path) parser.add_argument('--output', required=True, type=Path) parser.add_argument('--cmake', default='cmake') parser.add_argument('--jobs', type=int, default=2) parser.add_argument('--timeout', type=float, default=1800) args = parser.parse_args() source, build, mesh, output = [p.resolve() for p in [args.source, args.build, args.mesh, args.output]] if output.exists(): parser.error('Output directory already exists; use a new directory for every run.') def git(*words): return subprocess.check_output(['git', *words], cwd=source, text=True).strip() if git('status', '--porcelain', '--untracked-files=no'): parser.error('Tracked source must be committed before running a campaign.') cache = dict(line.split('=', 1) for line in (build / 'CMakeCache.txt').read_text().splitlines() if '=' in line and not line.startswith(('#', '//'))) if Path(cache['CMAKE_HOME_DIRECTORY:INTERNAL']).resolve() != source: parser.error('Build directory belongs to a different source checkout.') cases = json.loads(args.cases.read_text()) seen = set() for case in cases: name = case['id'] if not name or any(c not in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-' for c in name): parser.error('Case IDs may contain only letters, digits, underscore and hyphen.') if name in seen: parser.error('Duplicate case ID.') seen.add(name) if not isinstance(case['arguments'], list) or not all(isinstance(a, str) for a in case['arguments']): parser.error('Case arguments must be a list of strings.') if '--mesh' in case['arguments'] or '--output' in case['arguments']: parser.error('Mesh and output are controlled by the campaign runner.') output.mkdir(parents=True) build_command = [args.cmake, '--build', str(build), '--target', 'plateau_diagnostic', '-j', str(args.jobs)] provenance = {'started_utc': utc(), 'source': str(source), 'commit': git('rev-parse', 'HEAD'), 'branch': git('branch', '--show-current'), 'platform': platform.platform(), 'build_command': build_command, 'build_settings': cache, 'mesh': str(mesh), 'mesh_sha256': digest(mesh), 'cases': cases, 'runtime_environment': {'OMPI_MCA_btl': 'self,sm'}} write_json(output / 'provenance.json', provenance) shutil.copy2(args.cases, output / 'cases.json') print('BUILD', provenance['commit'], flush=True) with (output / 'build.log').open('w') as stream: subprocess.run(build_command, cwd=source, stdout=stream, stderr=subprocess.STDOUT, check=True) if git('rev-parse', 'HEAD') != provenance['commit'] or git('status', '--porcelain', '--untracked-files=no'): raise RuntimeError('Source changed during build.') # A private immutable executable prevents later branch/build operations from changing queued cases. executable = output / 'plateau_diagnostic' shutil.copy2(build / 'plateau_diagnostic', executable) provenance['binary_sha256'] = digest(executable) provenance['build_finished_utc'] = utc() write_json(output / 'provenance.json', provenance) summaries = [] for case in cases: name = case['id'] folder = output / name command = [str(executable), *case['arguments'], '--mesh', str(mesh), '--output', str(folder)] state_files = {} for i, word in enumerate(case['arguments']): if word in ('--probe-state', '--probe-direction', '--replay-candidate'): path = Path(case['arguments'][i + 1]) if not path.is_absolute(): path = source / path path = path.resolve() state_files[str(path)] = digest(path) print('START', name, flush=True) start = time.monotonic() record = {'id': name, 'command': command, 'started_utc': utc(), 'input_state_hashes': state_files} write_json(output / (name + '_run.json'), record) with (output / (name + '.log')).open('w') as stream: try: run = subprocess.run(command, cwd=source, stdout=stream, stderr=subprocess.STDOUT, env={**os.environ, 'OMPI_MCA_btl': 'self,sm'}, timeout=args.timeout) record['exit_code'] = run.returncode except subprocess.TimeoutExpired: record['timed_out'] = True record['exit_code'] = None record.update(seconds=time.monotonic() - start, finished_utc=utc()) record['output_hashes'] = { str(path.relative_to(folder)): digest(path) for path in sorted(folder.rglob('*')) if path.is_file() and path.suffix in ('.txt', '.csv', '.json') } write_json(output / (name + '_run.json'), record) summaries.append({**record, **summarize(folder)}) write_json(output / 'summary.json', summaries) print('FINISH', name, record['exit_code'], round(record['seconds'], 2), flush=True) if record['exit_code'] not in (0, 3): raise RuntimeError('Case failed to execute; see its retained log and run record.') # Exit 0 denotes completed observations; exit 3 from a model remains explicitly recorded as nonconvergence. return 0 if __name__ == '__main__': raise SystemExit(main())