Files
MeanField/experiments/plateau_analysis/analyze_pressure.py

476 lines
33 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Read completed/partial pressure diagnostics; never launch a solver or modify inputs."""
import argparse
import csv
import json
import math
from pathlib import Path
ROOT = Path(__file__).resolve().parent
ITERATIONS = (2, 3, 4, 7, 11)
def rows(path):
if not path.exists():
return []
with path.open() as stream:
return [r for r in csv.DictReader(stream) if None not in r and all(v is not None for v in r.values())]
def metrics(path):
return {r['metric']: float(r['value']) for r in rows(path)}
def metadata(path):
if not path.exists():
return {}
return dict(line.split('=', 1) for line in path.read_text().splitlines() if '=' in line)
def numeric(row):
result = {}
for key, value in row.items():
try:
result[key] = float(value)
except ValueError:
result[key] = value
return result
def negative_h_rejection(row):
# Enums at source9e1b65d: thermodynamic_domain=2, pressure_force=6,
# outside_domain=5. The prepared pressure operator emits this for h<0.
return all(float(row.get(k, -999)) == v for k, v in
[('admissible', 0), ('reason', 2), ('stage', 6), ('eos_code', 5)])
def sanitize(value):
if isinstance(value, float) and not math.isfinite(value):
return None
if isinstance(value, dict):
return {k: sanitize(v) for k, v in value.items()}
if isinstance(value, list):
return [sanitize(v) for v in value]
return value
def case(iteration):
folder = ROOT / 'focused' / f'n1_J0p10_iteration_{iteration}'
capture = metadata(ROOT / 'captures/n1_J0p10_capture' / f'capture_{iteration}/capture.txt')
bound = metrics(folder / 'frozen_enthalpy_bound.csv')
inputs = metrics(folder / 'frozen_input_checks.csv')
replay = rows(folder / 'replayed_candidate.csv')
trial_rows = rows(folder / 'pressure_obstruction_trials.csv')
limits = [numeric(r) for r in rows(folder / 'pressure_obstruction_limits.csv')]
summary = metrics(folder / 'pressure_obstruction_summary.csv')
live = next((r for r in rows(ROOT / 'captures/n1_J0p10_capture/newton.csv')
if int(r['iteration']) == iteration), None)
result = {'iteration': iteration, 'folder': str(folder), 'capture': capture,
'bounds': bound, 'input_checks': inputs, 'limits': limits,
'exact_candidate': numeric(replay[0]) if replay else None,
'exact_candidate_confirms_pressure_negative_h': negative_h_rejection(replay[0]) if replay else None,
'restoration_completed': 'final_restoration_residual_difference' in inputs,
'pressure_trials_completed': len(trial_rows),
'pressure_trials_expected': summary.get('candidate_trial_count'),
'trial_checks': [], 'live_newton_step': numeric(live) if live else None}
result['pressure_grid_complete'] = (summary.get('candidate_trial_count') is not None and
len(trial_rows) == summary['candidate_trial_count'])
if capture:
result['direction_roundoff_relative_estimate'] = (
float(capture['direction_roundoff_scale_estimate']) / float(capture['direction_norm']))
if bound:
result['closure_to_pressure_bound_ratio'] = bound['closure_alpha_h'] / bound['pressure_alpha_h']
result['exact_first_trial_over_pressure_bound'] = float(capture['first_trial_alpha']) / bound['pressure_alpha_h']
if live:
result['accepted_alpha_over_union_bound'] = float(live['alpha']) / bound['alpha_thermodynamic']
result['tightest_sampled_family'] = ('closure' if bound['closure_alpha_h'] < bound['pressure_alpha_h'] else
'pressure' if bound['pressure_alpha_h'] < bound['closure_alpha_h'] else 'tie')
midpoint = .5 * (bound['closure_alpha_h'] + bound['pressure_alpha_h'])
result['between_bounds_midpoint'] = {
'alpha': midpoint,
'closure_limiting_h_prediction': bound['limiting_h'] + midpoint * bound['limiting_dh'],
'pressure_limiting_h_prediction': bound['pressure_limiting_h'] + midpoint * bound['pressure_limiting_dh'],
}
for row in trial_rows:
r = numeric(row)
errors = [abs(r[f'{family}_limiting_h_evaluated'] - r[f'{family}_limiting_h_prediction'])
for family in ('pressure', 'closure')]
sign_checks = [((r[f'{family}_limiting_h_evaluated'] < 0) ==
(r[f'{family}_limiting_h_prediction'] < 0)) for family in ('pressure', 'closure')]
ratio = r['alpha_over_union_bound']
r['maximum_limiting_h_evaluation_prediction_absolute_difference'] = max(errors)
r['both_limiting_point_signs_match_prediction'] = all(sign_checks)
r['confirms_pressure_negative_h'] = negative_h_rejection(row)
# Descriptive expectation check for these fixed cases, not a general
# assertion that h>=0 guarantees preparation or merit acceptance.
expected = (r['direction'] == 'shape_only' or ratio < 1)
r['admissibility_matches_enthalpy_only_prediction'] = bool(r['admissible']) == expected
result['trial_checks'].append(r)
if trial_rows:
result['all_limiting_sign_predictions_match'] = all(r['both_limiting_point_signs_match_prediction']
for r in result['trial_checks'])
result['all_admissibility_matches_enthalpy_only_prediction'] = all(
r['admissibility_matches_enthalpy_only_prediction'] for r in result['trial_checks'])
result['mapping_rejections'] = sum(r['admissible'] == 0 and r['reason'] in (0, 1)
for r in result['trial_checks'])
result['maximum_limiting_h_absolute_disagreement'] = max(
r['maximum_limiting_h_evaluation_prediction_absolute_difference'] for r in result['trial_checks'])
shape_rows = [r for r in result['trial_checks'] if r['direction'] == 'shape_only']
h_crossing_rows = [r for r in result['trial_checks']
if r['direction'] == 'enthalpy_only' and r['alpha_over_union_bound'] > 1]
result['shape_only_all_prepared'] = all(r['admissible'] == 1 for r in shape_rows) if len(shape_rows) == 4 else None
result['enthalpy_only_crossing_failures_all_pressure_negative_h'] = (
all(r['confirms_pressure_negative_h'] for r in h_crossing_rows) if len(h_crossing_rows) == 2 else None)
return result
def fmt(value):
return 'pending' if value is None else f'{value:.9g}'
def vector_entries(path, indices):
"""Read only selected zero-based entries; reject an incompatible saved vector."""
if not path.exists():
return None
selected = set(indices)
found = {}
with path.open() as stream:
size = int(next(stream).strip())
if not selected or min(selected) < 0 or max(selected) >= size:
raise ValueError(f'Carrier indices incompatible with {path}')
count = 0
for index, line in enumerate(stream):
value = float(line)
count += 1
if index in selected:
found[index] = value
if count != size or len(found) != len(selected):
raise ValueError(f'Incomplete vector {path}')
return [found[index] for index in indices]
def extrema(values):
return {'count': len(values), 'minimum': min(values), 'maximum': max(values),
'maximum_absolute': max(abs(x) for x in values),
'l2_norm': math.sqrt(math.fsum(x*x for x in values)),
'negative_count': sum(x < 0 for x in values),
'nonzero_count': sum(x != 0 for x in values)}
def supplemental(cases):
"""Find the two supplemental case IDs without imposing an output-parent name."""
result = {'cases': [], 'offline_surface_checks': []}
case_ids = [('n1_J0p10_onset_between_bounds', 2), ('n1_J0p10_late_surface_trace', 11)]
reference_rows = None
for name, iteration in case_ids:
candidates = [path.parent for path in ROOT.glob(f'**/{name}/surface_trace_summary.csv')]
if len(candidates) > 1:
raise ValueError(f'Ambiguous duplicate supplemental case: {name}')
if not candidates:
result['cases'].append({'id': name, 'iteration': iteration, 'available': False})
continue
folder = candidates[0]
carrier_rows = [numeric(r) for r in rows(folder / 'surface_trace_rows.csv')]
summary = metrics(folder / 'surface_trace_summary.csv')
if 'pressure_alpha_h' not in summary:
result['cases'].append({'id': name, 'iteration': iteration, 'available': False})
continue
contribution_rows = [numeric(r) for r in rows(folder / 'surface_trace_limits.csv')]
node_rows = [numeric(r) for r in rows(folder / 'surface_trace_nodes.csv')]
replay_rows = rows(folder / 'replayed_candidate.csv')
source_case = next(c for c in cases if c['iteration'] == iteration)
details = {'id': name, 'iteration': iteration, 'available': True, 'folder': str(folder),
'summary': summary, 'contributions': contribution_rows,
'replayed_candidate': numeric(replay_rows[0]) if replay_rows else None,
'carrier_rows_complete': len(carrier_rows) == summary.get('surface_row_count'),
'nodal_checks': [],
'restoration': metrics(folder / 'frozen_input_checks.csv')}
provenance = folder.parent / 'provenance.json'
if provenance.exists():
p = json.loads(provenance.read_text())
details['provenance'] = {key: p.get(key) for key in ('commit', 'branch', 'mesh_sha256')}
if carrier_rows and details['carrier_rows_complete']:
indices = [int(r['root_index']) for r in carrier_rows]
if len(set(indices)) != len(indices):
raise ValueError(f'Duplicate surface indices in {folder}')
if reference_rows is not None and indices != reference_rows:
raise ValueError('Supplemental surface carrier maps differ')
reference_rows = indices
for family in ('closure', 'pressure'):
nodal = [r for r in node_rows if r['family'] == family]
if not nodal:
continue
if len(nodal) != nodal[0]['node_count']:
continue # A live CSV may not yet have all nodal rows.
union = nodal[0]['alpha_union']
factors = [('base', 0), ('near_below_union', .99*union),
('near_above_union', 1.01*union), ('unit_step', 1)]
if source_case['capture']:
factors.append(('captured_first_alpha', float(source_case['capture']['first_trial_alpha'])))
if source_case['live_newton_step']:
factors.append(('accepted_alpha', source_case['live_newton_step']['alpha']))
scale = max(abs(r[k]) for r in nodal for k in ('h', 'dh'))
tolerance = 64 * math.ulp(1.0) * scale
limit = next((r for r in contribution_rows if r['family'] == family), None)
if limit is None:
continue
for label, alpha in factors:
values = [r['h'] + alpha*r['dh'] for r in nodal]
stats = extrema(values)
stats.update({'family': family, 'element': nodal[0]['element'], 'fe_order': nodal[0]['fe_order'],
'label': label, 'alpha': alpha, 'roundoff_scale_estimate': tolerance,
'negative_beyond_roundoff_count': sum(v < -tolerance for v in values),
'limiting_point_h': limit['full_h'] + alpha*limit['full_dh']})
details['nodal_checks'].append(stats)
if name.endswith('between_bounds') and details['replayed_candidate']:
r = details['replayed_candidate']
details['replay_confirms_closure_outside_domain'] = all(r.get(k) == v for k, v in
[('admissible', 0), ('reason', 2), ('stage', 3), ('eos_code', 5)])
details['candidate_derivation'] = json.loads((ROOT / 'supplemental_inputs/between_bounds_derivation.json').read_text())
result['cases'].append(details)
if reference_rows is not None:
result['surface_root_indices'] = reference_rows
for c in cases:
folder = ROOT / 'captures/n1_J0p10_capture' / f"capture_{c['iteration']}"
base = vector_entries(folder / 'physical_base.txt', reference_rows)
candidate = vector_entries(folder / 'physical_candidate.txt', reference_rows)
if base is None or candidate is None:
continue
alpha = float(c['capture']['first_trial_alpha'])
dh = [(v-b)/alpha for b, v in zip(base, candidate)]
result['offline_surface_checks'].append({'iteration': c['iteration'], 'first_trial_alpha': alpha,
'base_h': extrema(base), 'exact_candidate_h': extrema(candidate),
'reconstructed_direction_dh': extrema(dh),
'base_and_exact_candidate_entire_surface_identically_zero': all(v == 0 for v in base + candidate)})
return result
def main():
global ROOT
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('campaign_root', nargs='?', type=Path, default=ROOT,
help='Campaign artifact directory (default: directory containing this script).')
ROOT = parser.parse_args().campaign_root.expanduser().resolve()
if not ROOT.is_dir():
parser.error(f'Campaign directory does not exist: {ROOT}')
cases = [case(i) for i in ITERATIONS]
transitions = []
for first, second in zip(cases, cases[1:]):
if second['iteration'] != first['iteration'] + 1 or not first['bounds'] or not second['bounds']:
continue
alpha = first['live_newton_step']['alpha']
for family, prefix in [('closure', ''), ('pressure', 'pressure_')]:
b, c = first['bounds'], second['bounds']
same = all(b[prefix + key] == c[prefix + key] for key in ['limiting_element', 'limiting_point'])
predicted = b[prefix + 'limiting_h'] + alpha * b[prefix + 'limiting_dh']
observed = c[prefix + 'limiting_h']
transitions.append({'from_iteration': first['iteration'], 'to_iteration': second['iteration'],
'family': family, 'same_limiting_point': same, 'accepted_alpha': alpha,
'predicted_next_h': predicted, 'observed_next_h': observed,
'absolute_disagreement': abs(predicted - observed) if same else None})
supplements = supplemental(cases)
provenance_file = ROOT / 'focused/provenance.json'
source_revision = json.loads(provenance_file.read_text()).get('commit', '9e1b65d') if provenance_file.exists() else '9e1b65d'
report = {'schema_version': 2, 'source_revision': source_revision,
'interpretation': 'Exact candidate replay is authoritative; direction probes include reconstruction roundoff.',
'cases': cases, 'consecutive_captured_step_checks': transitions, 'supplemental': supplements}
polynomial_path = ROOT / 'pressure_polynomial_checks.json'
polynomial = json.loads(polynomial_path.read_text()) if polynomial_path.exists() else None
if polynomial:
report['offline_polynomial_artifacts'] = ['pressure_polynomial_checks.json', 'pressure_polynomial_analysis.md']
(ROOT / 'pressure_checks.json').write_text(json.dumps(sanitize(report), indent=2, allow_nan=False) + '\n')
lines = ['# Pressure obstruction analysis', '',
'Read-only analysis of the frozen n=1, J=.1 direction captures. This script launches no solver and does not modify case data. '
f'The original focused source is {source_revision}; iteration numbers are zero-based. '
'Supplemental source revisions are recorded per case in pressure_checks.json. Partial cases are explicitly pending.', '',
'| Iteration | Exact first alpha | Pressure bound | Closure bound | Limiting pressure h | dh | Element / point | Exact replay |',
'|---:|---:|---:|---:|---:|---:|---|---|']
for c in cases:
b = c['bounds']
loc = f"{int(b['pressure_limiting_element'])} / {int(b['pressure_limiting_point'])}" if b else 'pending'
replay = ('pressure negative h' if c['exact_candidate_confirms_pressure_negative_h'] else
('other; inspect JSON' if c['exact_candidate'] else 'pending'))
alpha = float(c['capture']['first_trial_alpha']) if c['capture'] else None
lines.append(f"| {c['iteration']} | {fmt(alpha)} | {fmt(b.get('pressure_alpha_h'))} | "
f"{fmt(b.get('closure_alpha_h'))} | {fmt(b.get('pressure_limiting_h'))} | "
f"{fmt(b.get('pressure_limiting_dh'))} | {loc} | {replay} |")
if all(c['exact_candidate_confirms_pressure_negative_h'] for c in cases):
lines += ['', '**All five exact saved first candidates reproduce finite negative enthalpy in pressure preparation.** '
'These failures are thermodynamic-domain rejections, not geometry rejections. The saved candidates are '
'replayed directly, so this conclusion does not depend on reconstructing the Newton direction.']
populated = [c for c in cases if c['bounds']]
if populated:
lines += ['', 'The family attaining the smaller sampled bound is ' + ', '.join(
f"iteration {c['iteration']}: {c['tightest_sampled_family']}" for c in populated) + '.', '',
'| Iteration | Live accepted alpha | Accepted alpha / union bound |', '|---:|---:|---:|']
for c in populated:
lines.append(f"| {c['iteration']} | {fmt(c['live_newton_step']['alpha'])} | {fmt(c['accepted_alpha_over_union_bound'])} |")
if cases[-1]['bounds'] and cases[0]['bounds']:
first, last = cases[0]['bounds'], cases[-1]['bounds']
lines += ['', f"From iteration 2 to 11 the pressure bound shrinks by "
f"{first['pressure_alpha_h']/last['pressure_alpha_h']:.6g} times. The limiting pressure sample remains "
f"element {int(last['pressure_limiting_element'])}, point {int(last['pressure_limiting_point'])}; "
'its h approaches zero while dh remains negative. '
f"At iteration 11 the closure bound is {last['closure_alpha_h']/last['pressure_alpha_h']:.6g} times larger. "
'The mechanism of the step-length collapse is established along these directions; why the Newton '
'directions continue to decrease this h remains unresolved.']
if transitions:
maximum = max(r['absolute_disagreement'] for r in transitions if r['absolute_disagreement'] is not None)
lines += ['', f"For consecutive captured states at the same limiting point, h_next = h + alpha_accepted*dh "
f"agrees with the independently scanned next base to at most {maximum:.3g} absolute. "
'The pointwise approach to the enthalpy boundary is therefore directly tied to the accepted steps.']
lines += ['', '## Crossing controls', '',
'| Iteration | Candidate rows / expected | Shape-only all prepared | h-only crossing rejects negative h | Geometry rejections | Max h prediction/evaluation difference | Restored |',
'|---:|---:|---|---|---:|---:|---|']
for c in cases:
if c['iteration'] not in (2, 11):
continue
lines.append(f"| {c['iteration']} | {c['pressure_trials_completed']} / {fmt(c['pressure_trials_expected'])} | "
f"{c.get('shape_only_all_prepared') if c.get('shape_only_all_prepared') is not None else 'pending'} | "
f"{c.get('enthalpy_only_crossing_failures_all_pressure_negative_h') if c.get('enthalpy_only_crossing_failures_all_pressure_negative_h') is not None else 'pending'} | "
f"{c.get('mapping_rejections', 'pending')} | {fmt(c.get('maximum_limiting_h_absolute_disagreement'))} | "
f"{c['restoration_completed']} |")
onset = cases[0]
if onset['bounds']:
b = onset['bounds']
midpoint = onset['between_bounds_midpoint']
midpoint_case = next(c for c in supplements['cases'] if c['iteration'] == 2)
measured_midpoint = midpoint_case.get('replay_confirms_closure_outside_domain')
midpoint_outcome = ('The supplemental candidate reproduces thermodynamic-domain rejection at the closure stage '
'(reason=2, stage=3, EOS code=5), confirming pressure preparation passes at this alpha before closure fails.'
if measured_midpoint is True else
'The supplemental replay has a different outcome; inspect pressure_checks.json.'
if measured_midpoint is False else
'Its production outcome is pending; the supplemental replay can separate closure from pressure.')
lines += ['', '## Onset interpretation', '',
f"The onset has a tighter **{onset['tightest_sampled_family']}** bound: closure {b['closure_alpha_h']:.12g}, "
f"pressure {b['pressure_alpha_h']:.12g}. The original alpha=1 trial crosses both. "
'Pressure is prepared before closure, so the recorded pressure-stage rejection identifies the first failing stage at that alpha; '
'it does not establish the pressure quadrature as the tightest restriction.', '',
f"A midpoint alpha={midpoint['alpha']:.12g} predicts closure h="
f"{midpoint['closure_limiting_h_prediction']:.9g} and pressure h="
f"{midpoint['pressure_limiting_h_prediction']:.9g} at the respective limiting points. "
'This midpoint is outside the original probe grid. ' + midpoint_outcome, '']
populated_limits = [c for c in cases if c['limits']]
if populated_limits:
lines += ['', '## Limiting-point locations', '',
'| Iteration | Family | Surface-adjacent element | Reference radius | Physical base (x, y, z) |',
'|---:|---|---|---:|---|']
for c in populated_limits:
for r in c['limits']:
xyz = ', '.join(fmt(r[f'physical_{d}']) for d in 'xyz')
lines.append(f"| {c['iteration']} | {r['family']} | {int(r['element'])}; "
f"{bool(r['element_touches_stellar_surface'])} | {fmt(r['reference_radius'])} | ({xyz}) |")
lines += ['', '## Surface trace is an enforced equation', '',
'The runtime isobaric condition is not absent merely because the field registry has no essential constraint. '
'The root constructs an enthalpy boundary map on StellarSurface (prepared_stellar_equilibrium.cpp:407), '
'replaces surface hydrostatic residual rows (lines 809–810), and replaces their Jacobian rows (904–908). '
'The compiled constraint is h−h(P_surface), with identity h Jacobian for fixed pressure '
'(surface/compiled.cppm:47–59). At P_surface=0 the polytropic EOS gives h=0. '
'The radial seed explicitly sets these boundary coefficients to zero '
'(seed/stellar_equilibrium_projection.cppm:203–208).', '',
'This is algebraic enforcement, not a per-trial coefficient projection. Source inspection alone does not '
'establish exact zero in every captured iterate under an inexact linear solve. Even a zero polynomial '
'boundary trace does not guarantee nonnegative enthalpy within the adjacent high-order element. '
'The observed near-surface wall therefore does not establish a missing boundary condition or interpolation '
'as the cause of the Newton direction.']
trace_cases = [c for c in supplements['cases'] if c['available']]
if trace_cases:
lines += ['', '## Supplemental surface carrier and nodal checks', '',
'The actual surface carrier indices come from GetPressureSurfaceRows(), then select entries in each '
'saved physical state. Boundary-only fields retain those coefficients and zero every other enthalpy '
'coefficient; evaluating this field at a limiting interior point gives the boundary contribution to '
'that polynomial value, not a separate trace sample on the stellar surface.', '',
'| Iteration | Carrier rows | Base max abs h | Direction max abs dh | Unit-step max abs h |',
'|---:|---:|---:|---:|---:|']
for c in trace_cases:
m = c['summary']
lines.append(f"| {c['iteration']} | {fmt(m.get('surface_row_count'))} | {fmt(m.get('base_h_maximum_absolute'))} | "
f"{fmt(m.get('direction_dh_maximum_absolute'))} | {fmt(m.get('h_plus_dh_maximum_absolute'))} |")
offline = supplements['offline_surface_checks']
if len(offline) == len(ITERATIONS) and all(c['base_and_exact_candidate_entire_surface_identically_zero'] for c in offline):
lines += ['', '**All actual surface enthalpy coefficients are exactly zero in all five saved bases and all '
'five exact first candidates.** Their reconstructed boundary directions are therefore also zero. '
'Boundary-row leakage is ruled out for these captured directions, independently of the inexact '
'linear-solve possibility from source inspection.']
else:
lines += ['', 'Offline surface checks (including nonzero and negative counts) are recorded per capture in '
'pressure_checks.json; inspect them before inferring exact trace enforcement in captured states.']
lines += ['', '| Iteration | Family | Full h | Boundary contribution h | Full dh | Boundary contribution dh |',
'|---:|---|---:|---:|---:|---:|']
for c in trace_cases:
for r in c['contributions']:
lines.append(f"| {c['iteration']} | {r['family']} | {fmt(r['full_h'])} | {fmt(r['boundary_h'])} | "
f"{fmt(r['full_dh'])} | {fmt(r['boundary_dh'])} |")
lines += ['', 'Full physical fields are scattered through the production field adapter, then evaluated at FE '
'GetNodes() coordinates. These values do not assume that raw signed local DOF indices are nodal values. '
'The same element is intentionally exported separately for each limiting rule family.', '',
'| Iteration | Family | Step | Nodal minimum h | Negative nodes | Limiting interior-point h |',
'|---:|---|---|---:|---:|---:|']
for c in trace_cases:
for r in c['nodal_checks']:
if r['label'] in ('base', 'near_above_union', 'unit_step'):
lines.append(f"| {c['iteration']} | {r['family']} | {r['label']} | {fmt(r['minimum'])} | "
f"{r['negative_count']} | {fmt(r['limiting_point_h'])} |")
undershoots = [(c['iteration'], r['family']) for c in trace_cases for r in c['nodal_checks']
if r['label'] == 'near_above_union' and r['minimum'] >= 0 and r['limiting_point_h'] < 0]
if undershoots:
lines += ['', 'At 1.01 times the union bound, nonnegative values at every FE node coexist with negative '
'enthalpy at the limiting interior point in ' + ', '.join(f'iteration {i} {family}' for i, family in undershoots) +
'. This demonstrates an interior polynomial undershoot in these trial directions. It does not '
'establish an interpolation implementation error or explain why Newton selects this direction.']
lines += ['', 'Nodal nonnegativity is not a certificate of positivity between nodes. The JSON includes strict '
'negative counts and a separate 64-epsilon times nodal-value-scale estimate, which is descriptive '
'rather than a certified rounding bound. Equality exactly at the union bound is deliberately not '
'used to decide a crossing. The production quadrature union bound also does not certify positivity '
'at unsampled points; these results do not establish the minimum enthalpy over an entire element.']
if polynomial:
lines += ['', '## Offline interior polynomial check', '',
'A separate independently validated tensor-Q3 reconstruction samples 495,264 strictly interior '
'reference points in each exported limiting element, including logarithmic distances to its zero '
'face. It reconstructs all 64 nodes exactly and both production limit points to roundoff.']
for c in polynomial['cases']:
minimum = c['sampled_extrema']['base']['minimum']
if minimum < -c['roundoff_scale_estimate']:
point = ', '.join(fmt(x) for x in c['sampled_extrema']['base']['minimum_reference_point'])
lines += ['', f"**The saved iteration {c['iteration']} base already has negative enthalpy between "
f"production quadrature points:** sampled h={minimum:.9g} at reference ({point}). "
'This base-state finding is independent of Newton-direction reconstruction. The surface '
'coefficients and all 64 FE node values remain nonnegative; production quadrature positivity '
'therefore misses a real polynomial undershoot in this accepted state.']
else:
lines += ['', f"Iteration {c['iteration']} has no negative base samples on this grid; its sampled "
f"directional bound {c['directional_bound']['sampled_admissible_step_bound']:.12g} is tighter "
f"than the production bound {c['union_production_bound']:.12g}. Absence of negative samples "
'does not certify positivity everywhere.']
lines += ['', 'See pressure_polynomial_analysis.md and pressure_polynomial_checks.json for interpolation '
'validation, minima, locations and limitations. No solver or preparation rule was changed.']
lines += ['', '## Scope and interpretation', '',
'A replay with reason=2, stage=6, EOS code=5 is the production pressure operator’s finite h<0 branch. '
'It resolves the old CSV ambiguity with inverted geometry. Missing determinant/F data on rejection are not measured nonfinite residuals.', '',
'The bound scans the exact production pressure/closure quadrature points using pointwise polynomial contractions. '
'The independently re-evaluated h values in the crossing table concern the two frozen limiting points, not a full search for each trial’s most negative point. '
'Optimized tensor contractions can differ at roundoff; .99/1.01 factors deliberately avoid equality at the crossing.', '',
'Element-to-surface adjacency, when available in pressure_obstruction_limits.csv, is an exact topology flag. '
'It is not distance to the surface. Physical/reference Cartesian coordinates and element-local integration coordinates are distinct.', '',
'At iteration 11 the supplied direction is reconstructed from a first-trial difference at alpha=1.9073486328125e-6; '
'its relative subtraction-roundoff scale estimate is approximately 4.88e-7, not a certified error bound. '
'The exact saved candidate is replayed without that reconstruction and is authoritative for the observed rejection.', '',
'Preparation success in a frozen probe does not establish line-search merit acceptance, physical equilibrium, or eventual Newton convergence. '
'No requested stopping test, enthalpy admissibility rule, or solver was changed. '
'The evolving bound is directional: a falling h/dh bound does not by itself identify why the nonlinear correction keeps pointing outward from the admissible set.', '',
'Machine-readable evidence and all individual trial checks: pressure_checks.json. '
'Reproduce this analysis with `python3 analyze_pressure.py /path/to/campaign_artifacts`; '
'the optional campaign path defaults to the script directory.']
(ROOT / 'pressure_analysis.md').write_text('\n'.join(lines) + '\n')
print(json.dumps({'cases_with_exact_replay': sum(c['exact_candidate'] is not None for c in cases),
'cases_fully_restored': sum(c['restoration_completed'] for c in cases),
'supplemental_available': sum(c['available'] for c in supplements['cases']),
'supplemental_fully_restored': sum('final_restoration_residual_difference' in c.get('restoration', {})
for c in supplements['cases']),
'candidate_rows': {c['iteration']: c['pressure_trials_completed'] for c in cases}}, sort_keys=True))
if __name__ == '__main__':
main()