166 lines
9.2 KiB
Python
166 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Tiny recorded-trial examples distinguish artifact validity from success.
|
|
|
|
These fixtures exercise live scalar joins and counter accounting, without
|
|
pretending to contain FEM states or independently computed Jacobian actions.
|
|
"""
|
|
import csv
|
|
import math
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
import check_damped_coupled_newton as audit
|
|
|
|
|
|
def write_rows(path, records):
|
|
with path.open('w', newline='') as stream:
|
|
writer = csv.DictWriter(stream, fieldnames=list(records[0]))
|
|
writer.writeheader()
|
|
writer.writerows(records)
|
|
|
|
|
|
class LiveCoupledCheckerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.folder = Path(self.temporary.name)
|
|
|
|
def fixture(self, accepted=True, false_gate=False):
|
|
"""One bounded iteration with two beta observations and one solve."""
|
|
final = .9 if accepted else 1.
|
|
trial_f = .9 if accepted else 1.05
|
|
corrected_merit = .5*trial_f**2
|
|
metadata = {
|
|
'n': '1.5', 'K': '1', 'M': '1', 'J': '0', 'max_newton': '6', 'linear_tol': '.03',
|
|
'coupled_attempts_per_iteration': '1', 'coupled_beta_steps': '2', 'coupled_beta_reduction': '.5',
|
|
'coupled_iterations': '160', 'coupled_tolerance': '.03', 'coupled_excess_ratio': '1',
|
|
'strategy': 'curvature-backtracked-coupled', 'accepted_steps': str(int(accepted)),
|
|
'converged': '0', 'failure': 'Explicit synthetic bounded failure.'}
|
|
(self.folder/'metadata.txt').write_text('\n'.join(f'{k}={v}' for k, v in metadata.items()))
|
|
self.write('newton', [dict(iteration=0, accepted=int(accepted), F=final, alpha=1 if accepted else 0,
|
|
trials=1, linear_status=0, linear_iterations=2, linear_true=.02, seconds=2, linear_seconds=.5,
|
|
prepare_seconds=.8, refresh_seconds=.1, linear_rhs_norm=1)])
|
|
self.write('layout', [dict(kind='residual', block='balance', offset=0, size=1, scale=1)])
|
|
self.write('blocks', [dict(iteration=0, stage=stage, block='balance', normalized_norm=value)
|
|
for stage, value in [('before', 1), ('after', final)]])
|
|
self.write('trials', [dict(iteration=0, trial=0, alpha=1, disposition=0 if accepted else 4,
|
|
source='' if accepted else 'residual metric', F=trial_f)])
|
|
totals = dict(attempts=1, kept=1, fallbacks=0, iterations=2, preparations=2, restorations=0,
|
|
solve_seconds=.2, verification_seconds=.01, preparation_seconds=.5, refresh_seconds=.05)
|
|
self.write('coupled_trials', [dict(iteration=0, trial=0, alpha=1, diagnostics_available=1, status=3,
|
|
**{key+'_total': value for key, value in totals.items()}, original_merit=.6,
|
|
corrected_merit=corrected_merit, has_solve=1, linear_status=0, linear_iterations=2, rhs_norm=.5,
|
|
reported_true_norm=.01, verified_true_norm=.01, threshold=.015, verified_gate=0 if false_gate else 1,
|
|
correction_norm=.8, maximum_correction_norm=1)])
|
|
self.write('coupled_search', [dict(iteration=0, trial=0, alpha=1, status=3, deferred_total=0,
|
|
selected_beta=.5, candidate_count=2, base_merit=.5, predicted_merit=.25,
|
|
predicted_decrease=.25, excess_ratio=.4)])
|
|
self.write('coupled_candidates', [dict(iteration=0, trial=0, candidate=i, beta=beta, status=0,
|
|
merit=merit, accepted=int(passed), preparation_seconds=.2)
|
|
for i, beta, merit, passed in [(0, 1, .8, False), (1, .5, corrected_merit, accepted)]])
|
|
self.write('trial_state_consistency', [dict(iteration=0, trial=0,
|
|
normalized_physical_roundtrip_error=0, relative_error=0)])
|
|
self.write('numerical_strategies', [dict(metric='coupled_'+key, value=value)
|
|
for key, value in {**totals, 'deferred_attempts': 0}.items()])
|
|
for stage, residual in [('initial', 1.), ('final', final)]:
|
|
measurements = []
|
|
for quadrature in (14, 18):
|
|
metrics = dict(polytropic_index=1.5, polytropic_constant=1, complete_volume_measurement=1,
|
|
mass_absolute_error=0, mass=1, angular_momentum_z=0, angular_momentum_absolute_error=0,
|
|
binding_energy=-3, force_binding_energy=-3, kinetic_energy=0, pressure_integral=math.nan,
|
|
enthalpy_pressure_integral=1, pressure_from_density_defined_everywhere=0,
|
|
pressure_from_enthalpy_defined_everywhere=1, negative_density_samples=1,
|
|
negative_enthalpy_samples=0, normalized_bordered_residual=residual)
|
|
for pressure in ('', 'enthalpy_'):
|
|
for force in ('', 'force_'):
|
|
metrics[pressure+force+'virial_signed'] = 0 if pressure else math.nan
|
|
metrics[pressure+force+'virial_error'] = 0 if pressure else math.nan
|
|
measurements.extend(dict(quadrature=quadrature, metric=key, value=value)
|
|
for key, value in metrics.items())
|
|
self.write(stage+'_physical', measurements)
|
|
|
|
def write(self, name, records):
|
|
write_rows(self.folder/(name+'.csv'), records)
|
|
|
|
def inspect(self):
|
|
checks = []
|
|
result = audit.inspect_case(self.folder, {'exit_code': 3},
|
|
lambda name, value: checks.append((name, bool(value))))
|
|
return result, [name for name, passed in checks if not passed]
|
|
|
|
def test_two_beta_observations_still_describe_one_auxiliary_solve(self):
|
|
self.fixture()
|
|
result, failed = self.inspect()
|
|
self.assertFalse(failed, failed)
|
|
self.assertEqual(result['accepted_steps'], 1)
|
|
self.assertEqual(result['auxiliary_totals']['attempts'], 1)
|
|
self.assertEqual(result['auxiliary_totals']['iterations'], 2)
|
|
self.assertEqual(result['auxiliary_totals']['preparations'], 2)
|
|
self.assertEqual(result['steps'][0]['cumulative_iteration_seconds'], 2)
|
|
self.assertFalse(result['converged'])
|
|
|
|
def test_two_iterations_difference_cumulative_counters_instead_of_summing_them(self):
|
|
self.fixture()
|
|
for name in ('newton', 'trials', 'coupled_trials', 'coupled_search', 'coupled_candidates', 'trial_state_consistency', 'blocks'):
|
|
records = audit.rows(self.folder/(name+'.csv'))
|
|
second = [{**row, 'iteration': 1} for row in records]
|
|
for row in second:
|
|
if name in ('newton', 'trials'):
|
|
row['F'] = .81
|
|
if name == 'newton':
|
|
row['linear_rhs_norm'] = .9
|
|
if name == 'blocks':
|
|
row['normalized_norm'] = .9 if row['stage'] == 'before' else .81
|
|
if name == 'coupled_trials':
|
|
for counter in audit.COUNTERS:
|
|
row[counter+'_total'] = 2*float(row[counter+'_total'])
|
|
row['original_merit'] = .6*.81
|
|
row['corrected_merit'] = .5*.81**2
|
|
if name == 'coupled_search':
|
|
for metric in ('base_merit', 'predicted_merit', 'predicted_decrease'):
|
|
row[metric] = .81*float(row[metric])
|
|
if name == 'coupled_candidates':
|
|
row['merit'] = float(row['merit'])*.81
|
|
self.write(name, records+second)
|
|
summary = audit.rows(self.folder/'numerical_strategies.csv')
|
|
for row in summary:
|
|
row['value'] = 2*float(row['value'])
|
|
self.write('numerical_strategies', summary)
|
|
final = audit.rows(self.folder/'final_physical.csv')
|
|
for row in final:
|
|
if row['metric'] == 'normalized_bordered_residual':
|
|
row['value'] = .81
|
|
self.write('final_physical', final)
|
|
metadata = self.folder/'metadata.txt'
|
|
metadata.write_text(metadata.read_text().replace('accepted_steps=1', 'accepted_steps=2'))
|
|
result, failed = self.inspect()
|
|
self.assertFalse(failed, failed)
|
|
self.assertEqual(result['auxiliary_totals']['attempts'], 2)
|
|
self.assertEqual(result['auxiliary_totals']['iterations'], 4)
|
|
self.assertEqual(result['steps'][1]['auxiliary_changes']['attempts'], 1)
|
|
self.assertEqual(result['steps'][1]['cumulative_iteration_seconds'], 4)
|
|
|
|
def test_retained_but_outer_rejected_correction_is_honest_failure_evidence(self):
|
|
self.fixture(accepted=False)
|
|
result, failed = self.inspect()
|
|
self.assertFalse(failed, failed)
|
|
self.assertEqual(result['accepted_steps'], 0)
|
|
self.assertEqual(result['auxiliary_totals']['kept'], 1)
|
|
self.assertFalse(result['converged'])
|
|
physical = result['endpoint_physical']['18']
|
|
self.assertFalse(physical['pressure_from_density_defined'])
|
|
self.assertIsNone(physical['virial_milestones']['virial_within_1e5'])
|
|
self.assertTrue(physical['virial_milestones']['enthalpy_virial_within_1e5'])
|
|
self.assertEqual(physical['global_enthalpy_certificate'], 'not_measured')
|
|
|
|
def test_false_retained_auxiliary_gate_is_rejected(self):
|
|
self.fixture(false_gate=True)
|
|
_, failed = self.inspect()
|
|
self.assertIn('0:0:retained_auxiliary_certificate', failed)
|
|
self.assertIn('0:0:auxiliary_true_gate', failed)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|