File size: 6,252 Bytes
eae86f2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | #!/usr/bin/env python3
"""Scientific field, error and convergence plots from actual run artifacts."""
import argparse
import csv
import json
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import matplotlib.tri as mtri
from scripts.common import load_config, atomic_json
def save(fig, directory, stem):
for ext in ('png', 'pdf'):
fig.savefig(directory / f'{stem}.{ext}', dpi=300, bbox_inches='tight')
plt.close(fig)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--config', default='config/config.yaml')
args = parser.parse_args()
cfg, root = load_config(args.config)
results = root / cfg['paths']['results']
metrics = json.loads((results / 'metrics.json').read_text())
manifest = []
for label in metrics['checkpoints']:
data = np.load(results / f'predictions_{label}.npz')
xy, pred, ref = data['xy'], data['prediction_uvp'], data['reference_uvp']
tri = mtri.Triangulation(xy[:, 0], xy[:, 1])
fig, axes = plt.subplots(3, 1, figsize=(15, 5.5), constrained_layout=True)
lower, upper = min(ref[:, 0].min(), pred[:, 0].min()), max(ref[:, 0].max(), pred[:, 0].max())
values = [ref[:, 0], pred[:, 0], abs(pred[:, 0] - ref[:, 0])]
titles = ['SU2 reference u', f'Adaptive pseudo-time PINN u ({label}, step {int(data["checkpoint_step"])})', '|u_PINN - u_SU2|']
for i, (ax, val, title) in enumerate(zip(axes, values, titles)):
kwargs = dict(cmap='viridis', shading='gouraud')
if i < 2:
kwargs.update(vmin=lower, vmax=upper)
c = ax.tripcolor(tri, val, **kwargs)
fig.colorbar(c, ax=ax, pad=0.01)
ax.set(xlim=(0, 15), ylim=(-.5, .5), xlabel='x / H', ylabel='y / H', title=title)
ax.set_aspect('equal')
fig.suptitle(f'Re = 800; relative L2(u) = {metrics["checkpoints"][label]["relative_l2_u"]:.6g}')
save(fig, results, f'velocity_u_{label}')
manifest += [f'velocity_u_{label}.png', f'velocity_u_{label}.pdf']
fig, axes = plt.subplots(2, 1, figsize=(15, 4), constrained_layout=True)
for ax, val, title, cmap in [(axes[0], np.linalg.norm(pred[:, :2], axis=1), 'Predicted speed', 'viridis'),
(axes[1], pred[:, 2], 'Predicted pressure', 'coolwarm')]:
c = ax.tripcolor(tri, val, cmap=cmap, shading='gouraud')
fig.colorbar(c, ax=ax, pad=.01)
ax.set(xlabel='x / H', ylabel='y / H', title=title, xlim=(0, 15), ylim=(-.5, .5))
ax.set_aspect('equal')
save(fig, results, f'speed_pressure_{label}')
manifest += [f'speed_pressure_{label}.png', f'speed_pressure_{label}.pdf']
rows = list(csv.DictReader((results / 'training_history.csv').open()))
def series(key):
valid = [r for r in rows if r.get(key, '')]
return np.array([float(r['step']) for r in valid]), np.array([float(r[key]) for r in valid])
fig, axes = plt.subplots(2, 2, figsize=(11, 7), constrained_layout=True)
x, y = series('loss')
axes[0, 0].semilogy(x, y, label='Weighted relaxed loss')
for key in ('raw_u_res', 'raw_v_res', 'raw_p_res'):
x, y = series(key)
axes[0, 0].semilogy(x, y, label=key)
axes[0, 0].legend(fontsize=8)
x, y = series('relative_l2_u')
axes[0, 1].semilogy(x, y, label='Actual SU2 relative L2(u)')
axes[0, 1].axhline(cfg['evaluation']['paper_relative_l2_u'], color='gray', linestyle='--', label='Table 1: 2.32e-3')
axes[0, 1].legend(fontsize=8)
for key in ('tau_u', 'tau_v', 'tau_p'):
x, y = series(key)
axes[1, 0].semilogy(x, y, label=key)
axes[1, 0].legend()
x, y = series('learning_rate')
axes[1, 1].plot(x, y)
axes[1, 1].set_ylabel('Learning rate')
for ax in axes.flat:
ax.set_xlabel('Actual optimizer iterations')
ax.grid(alpha=.2)
save(fig, results, 'training_curves')
manifest += ['training_curves.png', 'training_curves.pdf']
summary = json.loads((results / 'training_summary.json').read_text())
report = ['# BFS Re=800 formal reproduction', '',
f'Actual completed steps: {summary["completed_steps"]} / {summary["requested_steps"]}.',
f'Slurm job: {summary["slurm_job_id"]}; host: {summary["host"]}.',
f'Elapsed training seconds: {summary["elapsed_seconds"]:.1f}.', '',
'| Selection | Checkpoint step | Relative L2(u) |', '|---|---:|---:|']
for key, value in metrics['checkpoints'].items():
report.append(f'| {key} | {value["checkpoint_step"]} | {value["relative_l2_u"]:.8g} |')
report += ['', 'Paper Table 1 adaptive value: 0.00232. This is a comparison target, not a measured result.',
'Training used only analytic boundary conditions and resampled PDE collocation points.',
'The SU2 reference was used only for reporting. Checkpoint selection used an independent unlabeled physics validation set.',
'Full parameter provenance and paper/repository conflicts are recorded in the reproduction specification.',
'', 'Figures:', ''] + [f'- {f}' for f in manifest if f.endswith('.png')]
error_history = results / 'error_history.jsonl'
if error_history.exists():
report += ['', 'Execution errors and repairs:', '']
for line in error_history.read_text().splitlines():
event = json.loads(line)
report.append(f'- Job {event["job_id"]}, actual formal steps {event["completed_steps"]}: '
f'{event["error"]}. Repair: {event["repair"]}.')
(results / 'report.md').write_text('\n'.join(report) + '\n')
atomic_json(results / 'visualization_manifest.json', {'status': 'completed', 'figures': manifest,
'reference': metrics['reference'], 'dpi': 300, 'interpolation': 'triangulation only for display; metrics use original nodes'})
print(json.dumps({'event': 'visualization_completed', 'figures': manifest, 'report': str(results / 'report.md')}), flush=True)
if __name__ == '__main__':
main()
|