File size: 6,019 Bytes
0d2f59e | 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 | """Independent conservation identity check and completed-artifact audit."""
import csv
import json
from pathlib import Path
import sys
import numpy as np
import torch
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from model import PINNShallowWater
from common import arguments, device_for, errors, load_config, output_path, runtime_info, save_json, sha256
from train import check_formal_config
def conservation_identity(config, device):
"""Compare expanded R=A*r to derivatives of the conserved fluxes.
This checks a separate mathematical representation, including mixed
momentum fluxes; it catches dropped mass-residual terms and axis mistakes.
"""
torch.manual_seed(config['seed'])
model = PINNShallowWater(**config['model']).to(device=device, dtype=torch.float64)
points = torch.rand(19, 3, dtype=torch.float64, device=device).requires_grad_(True)
points = points * torch.tensor([20, 20, 0.8], device=device) - torch.tensor([10, 10, 0], device=device)
state, expanded = model.residuals(points, config['physics']['gravity'], True)
h, u, v = state.unbind(1)
g = config['physics']['gravity']
conserved = torch.stack((h, h * u, h * v), dim=1)
flux_x = torch.stack((h * u, h * u.square() + 0.5 * g * h.square(), h * u * v), dim=1)
flux_y = torch.stack((h * v, h * u * v, h * v.square() + 0.5 * g * h.square()), dim=1)
expected = torch.zeros_like(expanded)
for component in range(3):
for value, axis in [(conserved, 2), (flux_x, 0), (flux_y, 1)]:
expected[:, component] += torch.autograd.grad(value[:, component].sum(), points, retain_graph=True)[0][:, axis]
maximum_error = (expanded - expected).abs().max().item()
torch.testing.assert_close(expanded, expected, rtol=1e-9, atol=1e-10)
(expanded.square().mean() + model.slope_recovery()).backward()
assert all(p.grad is not None and torch.isfinite(p.grad).all() for p in model.parameters())
assert model.slopes.numel() == 1500
return {'flux_vs_expanded_max_abs': maximum_error, 'backward_all_parameters_finite': True,
'neuron_slopes': model.slopes.numel(), 'double_precision_identity_passed': True}
def main():
parser = arguments(__doc__)
parser.add_argument('--preflight', action='store_true')
args = parser.parse_args()
config, root = load_config(args.config)
check_formal_config(config)
device = device_for(config)
identity = conservation_identity(config, device)
if args.preflight:
report = {'status': 'passed', 'execution_readiness': 'ready', 'preflight_passed': True,
'evidence': {'preflight': {'status': 'passed'}},
'runtime': runtime_info(device), 'conservation_identity': identity,
'entry_config': 'formal dCDB9 checked', 'training_performed': False}
save_json(root / '.onescience/preflight_compute.json', report)
print('PREFLIGHT_COMPUTE_PASSED', report, flush=True)
return
results = output_path(root, config['paths']['results'])
summary = json.loads((results / 'training_summary.json').read_text())
metrics = json.loads((results / 'metrics.json').read_text())
with (results / 'training_history.csv').open() as handle:
history = list(csv.DictReader(handle))
assert len(history) == 5000
assert [int(row['iteration']) for row in history] == list(range(1, 5001))
assert all(float(row['learning_rate']) == 0.001 for row in history)
assert all(np.isfinite(float(row[key])) for row in history for key in row)
best_path = output_path(root, config['paths']['best_checkpoint'])
best = torch.load(best_path, map_location='cpu', weights_only=False)
last = torch.load(output_path(root, config['paths']['last_checkpoint']), map_location='cpu', weights_only=False)
assert last['iteration'] == best['completed_iterations'] == summary['completed_iterations'] == 5000
assert best['iteration'] == summary['best_iteration']
assert best['formal_run_complete'] and not summary['stopped_early']
model = PINNShallowWater(**config['model']).to(device)
model.load_state_dict(best['model_state_dict'], strict=True)
assert all(torch.isfinite(p).all() for p in model.parameters())
assert sha256(best_path) == summary['best_checkpoint_sha256'] == metrics['checkpoint_sha256']
assert sha256(output_path(root, config['paths']['samples'])) == best['sampling_sha256']
with (np.load(output_path(root, config['paths']['predictions'])) as prediction,
np.load(output_path(root, config['paths']['hll_reference'])) as reference):
fields, refs = prediction['fields'], reference['fields']
assert fields.shape == refs.shape == (3, 500, 500, 3)
assert np.isfinite(fields).all() and np.isfinite(refs).all()
for k in range(3):
recomputed = errors(fields[k].reshape(-1, 3), refs[k].reshape(-1, 3))
for key in ['MAE', 'RMSE']:
np.testing.assert_allclose(recomputed[key], metrics['errors'][k][key], rtol=1e-10, atol=1e-12)
figures = ['dCDB9_fields_and_errors.png', 'dCDB9_fields_and_errors.pdf',
'dCDB9_centerline_comparison.png', 'dCDB9_water_depth_surfaces.png', 'training_loss.png']
for name in figures:
assert (results / name).stat().st_size > 1000
final = {'status': 'passed', 'formal_iterations': 5000, 'history_rows': len(history),
'learning_rate_constant': True, 'checkpoint_strict_load': True,
'checkpoint_finite': True, 'checkpoint_sha256': sha256(best_path),
'conservative_identity': identity, 'predictions_shape': list(fields.shape),
'metrics_independently_recomputed': True, 'figures_nonempty': figures,
'runtime': runtime_info(device), 'accuracy_claim': 'actual metrics reported; no claim of paper accuracy parity'}
save_json(results / 'verification.json', final)
print('FINAL_VERIFICATION_PASSED', final, flush=True)
if __name__ == '__main__':
main()
|