249 lines
20 KiB
Python
249 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Small synthetic searches exercise certificates, selection and provenance."""
|
|
import csv
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
import check_damped_coupled_trial as c
|
|
from test_coupled_trial_checker import csv_file, vec_file
|
|
|
|
|
|
class DampedCheckerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name).resolve(); self.folder = self.root / 'synthetic'; self.folder.mkdir()
|
|
self.make_case()
|
|
|
|
def make_case(self, kind='damped'):
|
|
for path in self.folder.iterdir(): path.unlink()
|
|
pfx, folder, nan = c.P, self.folder, math.nan
|
|
mode = 'near' if kind in ('near', 'raw_failure') else ('none' if kind == 'none' else ('full' if kind in ('full', 'budget', 'linear_failure', 'cap') else 'damped'))
|
|
x, p, f = [1., 2., 3.], [-1., -1., -1.], [1., 1., 1.]
|
|
sf, rf = [2.] * 3, [3.] * 3
|
|
save = lambda name, value: vec_file(folder / (pfx + name + '.txt'), value)
|
|
for name, values in {'base': x, 'base_physical': [v / 2 for v in x], 'base_residual': f, 'direction': p,
|
|
'base_action': p, 'state_factors': sf, 'residual_factors': rf, 'restored_residual': f, 'restored_action': p}.items(): save(name, values)
|
|
layout = [{'kind': typ, 'block': name, 'offset': i, 'size': 1} for typ, names in (
|
|
('value', ('density', 'shape', 'enthalpy')), ('residual', ('closure', 'shape', 'hydrostatic'))) for i, name in enumerate(names)]
|
|
csv_file(folder / (pfx + 'layout.csv'), layout)
|
|
max_steps = 1 if mode in ('none', 'full') else 4
|
|
cap_ratio = .01 if kind == 'cap' else 1.
|
|
records, candidates, evaluations = [], [], []
|
|
attempts = iterations = deferred = 0
|
|
plans = ['attempt']
|
|
if kind == 'none': plans = ['accepted']
|
|
elif kind in ('full', 'linear_failure', 'cap', 'best'): plans += ['accepted']
|
|
elif kind == 'budget': plans += ['budget', 'accepted']
|
|
elif kind == 'near': plans = ['deferred', 'attempt']
|
|
elif kind == 'raw_failure': plans = ['raw_failure', 'accepted']
|
|
for i, plan in enumerate(plans):
|
|
alpha, suffix = 2. ** -i, '_a' + str(i)
|
|
raw = [v + alpha * w for v, w in zip(x, p)]; prediction = [1. - alpha] * 3
|
|
save('raw' + suffix, raw); save('prediction' + suffix, prediction)
|
|
usable = plan != 'raw_failure'; attempted = plan == 'attempt'; before_accepted = plan == 'accepted'
|
|
before_f = [.5] * 3 if before_accepted else ([3., 0., 0.] if plan == 'deferred' else [2., 0., 0.])
|
|
if usable:
|
|
save('raw_residual' + suffix, before_f); save('before' + suffix, raw); save('before_residual' + suffix, before_f)
|
|
status = {'accepted': 'not_attempted' if mode == 'none' else 'original_accepted', 'budget': 'budget_exhausted',
|
|
'deferred': 'prediction_deferred', 'raw_failure': 'not_attempted'}.get(plan, 'kept')
|
|
selected, selected_f, selected_beta = raw, before_f, nan
|
|
core, count, preps, restores = -1, 0, 0, 0
|
|
correction = [-.5, 0., 0.]; rhs = [a - b for a, b in zip(prediction, before_f)]
|
|
action = [0., 0., 0.] if kind == 'linear_failure' else rhs
|
|
defect = c.minus(action, rhs); gate = c.norm(defect) <= .03 * c.norm(rhs)
|
|
if attempted:
|
|
attempts += 1; iterations += 2
|
|
core = 1 if kind == 'linear_failure' else (2 if kind == 'cap' else 0)
|
|
for name, values in {'rhs': rhs, 'correction': correction, 'action': action, 'defect': defect}.items(): save(name + suffix, values)
|
|
if core:
|
|
status = 'linear_failure' if core == 1 else 'norm_limit'
|
|
else:
|
|
proposal_norms = [2.1] if mode == 'full' else ([1.8, 1.9, 2., 2.1] if kind == 'best' else [2.1, 1.])
|
|
observed = []
|
|
for k, value in enumerate(proposal_norms):
|
|
beta = .5 ** k; state = [v + beta * w for v, w in zip(raw, correction)]; residual = [value, 0., 0.]
|
|
failed = kind == 'candidate_failure' and k == 0
|
|
cm = .5 * value * value; accepted = cm <= (1 - 1.e-4 * alpha) * 1.5
|
|
candidates.append({'trial': i, 'candidate': k, 'beta': beta, 'status': 1 if failed else 0,
|
|
'merit': nan if failed else cm, 'accepted': int(accepted and not failed), 'preparation_seconds': .05})
|
|
count += 1; preps += 1
|
|
if not failed: observed.append((beta, state, residual))
|
|
if kind == 'best':
|
|
selected_beta = 1.; selected = [v + w for v, w in zip(raw, correction)]; selected_f = [1.8, 0., 0.]
|
|
observed.append((1., selected, selected_f)); restores = 1; preps += 1
|
|
elif mode == 'full':
|
|
status = 'not_improved'; observed.append((0., raw, before_f)); restores = 1; preps += 1
|
|
else:
|
|
selected_beta, selected, selected_f = observed[-1]
|
|
for e, (beta, state, residual) in enumerate(observed):
|
|
tag = suffix + '_e' + str(e)
|
|
save('evaluated_state' + tag, state); save('evaluated_residual' + tag, residual)
|
|
cm = c.merit(residual)
|
|
evaluations.append({'trial': i, 'evaluation': e, 'beta': beta, 'residual_norm': c.norm(residual),
|
|
'merit': cm, 'accepted': int(cm <= (1 - 1.e-4 * alpha) * 1.5)})
|
|
deferred += plan == 'deferred'
|
|
accepted_now = usable and c.merit(selected_f) <= (1 - 1.e-4 * alpha) * 1.5
|
|
if usable: save('selected' + suffix, selected); save('selected_residual' + suffix, selected_f)
|
|
if accepted_now: vec_file(folder / ('coupled_trial_selected_physical' + suffix + '.txt'), [v / 2 for v in selected])
|
|
gate_used = mode == 'near' and usable and not before_accepted and plan != 'budget'
|
|
_, decrease, excess = c.eligibility(1.5, c.merit(prediction), c.merit(before_f), 1.)
|
|
solve, verification = (.1, .01) if attempted else (0., 0.)
|
|
refresh, prep_seconds = (.02 * (1 + restores) if attempted else 0.), .05 * preps
|
|
correction_seconds = solve + verification + refresh + prep_seconds + (.01 if attempted else 0.)
|
|
records.append({'trial': i, 'alpha': alpha, 'raw_prepared': int(usable), 'raw_reason': -1 if usable else 2,
|
|
'raw_stage': -1 if usable else 6, 'gravity_gate': 1, 'usable': int(usable),
|
|
'before_norm': c.norm(before_f) if usable else nan, 'before_merit': c.merit(before_f) if usable else nan,
|
|
'before_accepted': int(before_accepted), 'status': c.STATUS.index(status), 'attempted': int(attempted),
|
|
'attempts_total': attempts, 'aux_iterations': 2 if attempted else 0, 'aux_gate': int(attempted and gate),
|
|
'aux_threshold': .03 * c.norm(rhs) if attempted else nan, 'aux_defect': c.norm(defect) if attempted else nan,
|
|
'correction_norm': .5 if attempted else nan, 'correction_cap': cap_ratio * alpha * math.sqrt(3) if attempted else nan,
|
|
'selected_beta': selected_beta, 'candidate_count': count, 'predicted_decrease': decrease if gate_used else nan,
|
|
'excess_ratio': excess if gate_used else nan, 'selected_prepared': int(usable),
|
|
'selected_norm': c.norm(selected_f) if usable else nan, 'selected_merit': c.merit(selected_f) if usable else nan,
|
|
'selected_accepted': int(accepted_now), 'selected_original': int(usable and selected == raw),
|
|
'raw_seconds': .1, 'gravity_seconds': 0., 'correction_seconds': correction_seconds, 'preparations': preps,
|
|
'restorations': restores, 'solve_seconds': solve, 'refresh_seconds': refresh, 'verification_seconds': verification,
|
|
'preparation_seconds': prep_seconds, 'aux_core_status': core, 'aux_status': 0 if attempted else -1,
|
|
'aux_reported_gate': int(attempted), 'aux_rhs_norm': c.norm(rhs) if attempted else nan,
|
|
'aux_reported_true_norm': 0. if attempted else nan, 'aux_verification_applied': int(attempted),
|
|
'rhs_saved': int(attempted), 'correction_saved': int(attempted), 'verified_vectors_saved': int(attempted),
|
|
'gravity_target_defect': nan, 'gravity_target_threshold': nan, 'gravity_affine_disagreement': nan})
|
|
csv_file(folder / (pfx + 'trials.csv'), records)
|
|
csv_file(folder / (pfx + 'candidates.csv'), candidates, ['trial', 'candidate', 'beta', 'status', 'merit', 'accepted', 'preparation_seconds'])
|
|
csv_file(folder / (pfx + 'evaluations.csv'), evaluations, ['trial', 'evaluation', 'beta', 'residual_norm', 'merit', 'accepted'])
|
|
last = records[-1]
|
|
summary = {'schema_version': 1, 'mode': mode, 'gravity_path': 0, 'attempt_budget': 1, 'maximum_trials': 13,
|
|
'maximum_iterations': 160, 'restart': 160, 'maximum_beta_steps': max_steps, 'beta_reduction': .5,
|
|
'maximum_excess_ratio': 1., 'sufficient_decrease': 1.e-4, 'relative_tolerance': .03, 'absolute_tolerance': 1.e-14,
|
|
'maximum_correction_ratio': cap_ratio, 'accepted_commits': 0, 'shared_iteration_budget': 1,
|
|
'base_norm': math.sqrt(3), 'base_merit': 1.5, 'outer_relative_defect': 0., 'outer_linear_tolerance': .03,
|
|
'outer_verified_gate': 1, 'capture_seconds': .1, 'search_wall_seconds': 10.,
|
|
'search_operation_seconds': sum(row['raw_seconds'] + row['correction_seconds'] for row in records),
|
|
'accepted_trial': len(records) - 1, 'accepted_alpha': last['alpha'], 'accepted_norm': last['selected_norm'],
|
|
'accepted_beta': last['selected_beta'], 'attempts': attempts, 'aux_iterations': iterations, 'deferred_attempts': deferred,
|
|
'probe_exception': 0, 'accepted_buffers_bitwise_identical': 1, 'direction_buffers_bitwise_identical': 1,
|
|
'normalization_factors_bitwise_identical': 1, 'restored_residual_error': 0., 'restored_action_error': 0.,
|
|
'fresh_operator_and_inverse_current': 1}
|
|
summary.update({'status_' + name: c.STATUS.index(name) for name in ('original_accepted', 'budget_exhausted', 'kept', 'prediction_deferred')})
|
|
summary.update({'candidate_status_' + name: i for i, name in enumerate(('evaluated', 'inadmissible', 'non_finite'))})
|
|
csv_file(folder / (pfx + 'summary.csv'), [{'metric': key, 'value': value} for key, value in summary.items()])
|
|
metadata = {'n': 1.5, 'K': 1., 'M': 1., 'J': 0., 'central_phase': 0, 'damped_coupled_probe': 1,
|
|
'coupled_trial_alphas_are_independent': 0, 'coupled_trial_admissibility': 'sampled_enthalpy_without_repair',
|
|
'coupled_search': mode, 'coupled_beta_steps': 4, 'coupled_beta_reduction': .5, 'coupled_excess_ratio': 1.,
|
|
'linear_tol': .03, 'coupled_tolerance': .03, 'coupled_iterations': 160, 'coupled_restart': 160,
|
|
'coupled_maximum_ratio': cap_ratio, 'coupled_attempts_per_iteration': 1, 'strategy': 'coupled'}
|
|
(folder / 'metadata.txt').write_text(''.join(f'{key}={value}\n' for key, value in metadata.items()))
|
|
self.physical(last)
|
|
self.make_campaign(mode)
|
|
|
|
def physical(self, selected):
|
|
records = []
|
|
def add(sample, source, q, key, value): records.append({'sample': sample, 'source': source, 'quadrature': q, 'metric': key, 'value': value})
|
|
for key in ('accepted_buffers_unchanged_before_restoration', 'accepted_buffers_unchanged_after_restoration',
|
|
'normalization_factors_unchanged', 'direction_buffers_unchanged', 'accepted_operator_and_inverse_current'):
|
|
add(-1, 'restoration', 0, key, 1)
|
|
add(-1, 'replay', 0, 'selected_snapshots_measured', 1)
|
|
i = selected['trial']; source = f'coupled_trial_selected_physical_a{i}.txt'
|
|
add(i, source, 0, 'fresh_preparation_passed', 1)
|
|
for q in (14, 18):
|
|
values = {'quadrature_order': q, 'polytropic_index': 1.5, 'polytropic_constant': 1., 'target_mass': 1.,
|
|
'stellar_samples': 10, 'valid_stellar_samples': 10, 'invalid_stellar_samples': 0, 'negative_density_samples': 0,
|
|
'negative_enthalpy_samples': 0, 'invalid_density_eos_samples': 0, 'invalid_enthalpy_eos_samples': 0,
|
|
'volume': 1., 'complete_volume_measurement': 1, 'pressure_from_enthalpy_defined_everywhere': 1,
|
|
'pressure_from_density_defined_everywhere': 1, 'saved_prepared_l2_difference': 0., 'saved_prepared_maximum_difference': 0.,
|
|
'saved_prepared_maximum_scaled_difference': 0., 'normalized_bordered_residual': selected['selected_norm'],
|
|
'normalized_unbordered_residual': selected['selected_norm'], 'normalized_border_action': 0., 'central_border': 0.,
|
|
'mass': 1., 'mass_absolute_error': 0., 'minimum_density': 0., 'minimum_enthalpy': 0.,
|
|
'binding_energy': -3., 'force_binding_energy': -3., 'kinetic_energy': 0., 'pressure_integral': 1., 'enthalpy_pressure_integral': 1.}
|
|
values.update({prefix + force + 'virial_' + name: 0. for prefix in ('', 'enthalpy_') for force in ('', 'force_') for name in ('signed', 'error')})
|
|
for key, value in values.items(): add(i, source, q, key, value)
|
|
csv_file(self.folder / 'coupled_trial_physical.csv', records)
|
|
|
|
def make_campaign(self, mode):
|
|
root = self.root
|
|
state, direction, mesh = root / 'state.txt', root / 'direction.txt', root / 'mesh'
|
|
vec_file(state, [.5, 1., 1.5]); vec_file(direction, [-1.] * 3); mesh.write_text('synthetic mesh')
|
|
(root / 'plateau_diagnostic').write_text('synthetic executable')
|
|
arguments = ['--damped-coupled-probe', '--coupled-search', mode, '--strategy', 'coupled', '--no-central',
|
|
'--probe-state', str(state), '--probe-direction', str(direction)]
|
|
cases = [{'id': 'synthetic', 'arguments': arguments}]
|
|
(root / 'cases.json').write_text(json.dumps(cases))
|
|
(root / 'provenance.json').write_text(json.dumps({'cases': cases, 'commit': 'synthetic', 'source': str(root),
|
|
'mesh': str(mesh), 'mesh_sha256': c.digest(mesh), 'binary_sha256': c.digest(root / 'plateau_diagnostic')}))
|
|
(root / 'synthetic_run.json').write_text(json.dumps({'id': 'synthetic', 'finished_utc': 'synthetic', 'exit_code': 0,
|
|
'command': [str(root / 'plateau_diagnostic'), *arguments, '--mesh', str(mesh), '--output', str(self.folder)],
|
|
'input_state_hashes': {str(path): c.digest(path) for path in (state, direction)},
|
|
'output_hashes': {path.name: c.digest(path) for path in self.folder.iterdir()}}))
|
|
|
|
def audit(self): return c.inspect_case(self.folder)
|
|
def assert_passes(self):
|
|
result = self.audit(); self.assertTrue(result['passed'], [row for row in result['checks'] if not row['passed']])
|
|
def mutate_row(self, file, key, value, row=0):
|
|
path = self.folder / (c.P + file + '.csv'); records = c.rows(path); records[row][key] = value; csv_file(path, records)
|
|
|
|
def add_gravity_path(self):
|
|
folder = self.folder; delta = [0., .01, .02]
|
|
for name in ('before_a0', 'selected_a0', 'evaluated_state_a0_e0', 'evaluated_state_a0_e1'):
|
|
path = folder / (c.P + name + '.txt'); values = c.vector(path)
|
|
vec_file(path, [a + b for a, b in zip(values, delta)])
|
|
vec_file(folder / (c.P + 'gravity_correction_a0.txt'), delta)
|
|
vec_file(folder / (c.P + 'raw_residual_a0.txt'), [2., .2, -.1])
|
|
selected = c.vector(folder / (c.P + 'selected_a0.txt'))
|
|
vec_file(folder / 'coupled_trial_selected_physical_a0.txt', [value / 2 for value in selected])
|
|
layout = [{'kind': kind, 'block': name, 'offset': i, 'size': 1} for kind, names in (
|
|
('value', ('density', 'gravity_gradient', 'gravity_potential')),
|
|
('residual', ('closure', 'gravity_gradient_relation', 'poisson_balance'))) for i, name in enumerate(names)]
|
|
csv_file(folder / (c.P + 'layout.csv'), layout)
|
|
for key, value in (('gravity_target_defect', 0.), ('gravity_target_threshold', 1.e-12), ('gravity_affine_disagreement', 0.)):
|
|
self.mutate_row('trials', key, value)
|
|
summary = c.rows(folder / (c.P + 'summary.csv'))
|
|
next(row for row in summary if row['metric'] == 'gravity_path')['value'] = 1
|
|
csv_file(folder / (c.P + 'summary.csv'), summary)
|
|
path = folder / 'metadata.txt'; path.write_text(path.read_text().replace('strategy=coupled', 'strategy=curvature-coupled'))
|
|
|
|
def test_damped_half_step(self): self.assert_passes()
|
|
def test_near_gate_preserves_budget(self): self.make_case('near'); self.assert_passes()
|
|
def test_full_fallback(self): self.make_case('full'); self.assert_passes()
|
|
def test_shared_budget_exhaustion(self): self.make_case('budget'); self.assert_passes()
|
|
def test_best_beta_replay(self): self.make_case('best'); self.assert_passes()
|
|
def test_unprepared_beta_skipped(self): self.make_case('candidate_failure'); self.assert_passes()
|
|
def test_no_correction_control(self): self.make_case('none'); self.assert_passes()
|
|
def test_pressure_failure_preserves_budget(self): self.make_case('raw_failure'); self.assert_passes()
|
|
def test_failed_full_true_gate(self): self.make_case('linear_failure'); self.assert_passes()
|
|
def test_full_cap_rejects_before_damping(self): self.make_case('cap'); self.assert_passes()
|
|
def test_gravity_path_uses_full_coupled_target(self): self.add_gravity_path(); self.assert_passes()
|
|
def test_corrupt_gravity_target_fails(self):
|
|
self.add_gravity_path(); self.mutate_row('trials', 'gravity_target_defect', 1.); self.assertFalse(self.audit()['passed'])
|
|
def test_provenance_passes(self): self.assertTrue(c.inspect_campaign(self.root)['passed'])
|
|
def test_missing_true_action_fails(self): (self.folder / (c.P + 'action_a0.txt')).unlink(); self.assertFalse(self.audit()['passed'])
|
|
def test_scaled_cannot_replace_verified_c(self):
|
|
vec_file(self.folder / (c.P + 'correction_a0.txt'), [-.25, 0., 0.]); self.assertFalse(self.audit()['passed'])
|
|
def test_corrupt_rhs_fails(self):
|
|
vec_file(self.folder / (c.P + 'rhs_a0.txt'), [-1., 0., 0.]); self.assertFalse(self.audit()['passed'])
|
|
def test_reset_budget_fails(self):
|
|
self.make_case('budget'); self.mutate_row('trials', 'attempts_total', 0, 1); self.assertFalse(self.audit()['passed'])
|
|
def test_deferred_gate_corruption_fails(self):
|
|
self.make_case('near'); self.mutate_row('trials', 'excess_ratio', .5); self.assertFalse(self.audit()['passed'])
|
|
def test_wrong_beta_fails(self):
|
|
self.mutate_row('candidates', 'beta', .25, 1); self.assertFalse(self.audit()['passed'])
|
|
def test_later_beta_after_acceptance_fails(self):
|
|
self.mutate_row('candidates', 'accepted', 1, 0); self.assertFalse(self.audit()['passed'])
|
|
def test_selected_stale_residual_fails(self):
|
|
vec_file(self.folder / (c.P + 'selected_residual_a0.txt'), [2.1, 0., 0.]); self.assertFalse(self.audit()['passed'])
|
|
def test_missing_physical_replay_fails(self):
|
|
(self.folder / 'coupled_trial_physical.csv').unlink(); self.assertFalse(self.audit()['passed'])
|
|
def test_campaign_hash_corruption_suppresses_observations(self):
|
|
self.mutate_row('candidates', 'merit', 99.); result = c.inspect_campaign(self.root)
|
|
self.assertFalse(result['passed']); self.assertFalse(result['cases']['synthetic']['observations_valid'])
|
|
def test_positive_predicted_decrease_required(self): self.assertFalse(c.eligibility(1., 1., 1., 1.)[0])
|
|
def test_beta_remainder_identity(self):
|
|
result = c.correction_mechanism([1.], [2.], [1.7], [-.9], [-1.], .5)
|
|
self.assertLess(result['decomposition_error_norm'], 1.e-15)
|
|
self.assertAlmostEqual(result['own_taylor_remainder_norm'], .15)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|