127 lines
7.1 KiB
Python
127 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate projected scalar controls from completed frozen coupled-trial cases.
|
|
|
|
This performs saved-vector algebra only. Replaying x+t*p through the existing
|
|
full operator distinguishes scalar step selection from transverse SOC benefit.
|
|
"""
|
|
import argparse
|
|
import csv
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
PREFIX = 'coupled_trial_'
|
|
SHARED = ('base_normalized.txt', 'base_physical.txt', 'base_residual.txt', 'direction.txt',
|
|
'state_factors.txt', 'residual_factors.txt', 'layout.csv')
|
|
|
|
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def vector(path):
|
|
words = path.read_text().split()
|
|
values = [float(word) for word in words[1:]]
|
|
if not words or int(words[0]) != len(values) or not values or not all(map(math.isfinite, values)):
|
|
raise ValueError(f'Invalid finite vector: {path}')
|
|
return values
|
|
|
|
|
|
def rows(path):
|
|
with path.open() as stream:
|
|
return list(csv.DictReader(stream))
|
|
|
|
|
|
def norm(values):
|
|
return math.sqrt(math.fsum(value * value for value in values))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('screen', type=Path)
|
|
parser.add_argument('output', type=Path)
|
|
args = parser.parse_args()
|
|
screen, output = args.screen.resolve(), args.output.resolve()
|
|
if output.exists() or screen in output.parents:
|
|
parser.error('Use a fresh output directory outside the immutable source screen.')
|
|
source_provenance = screen / 'provenance.json'
|
|
campaign = json.loads(source_provenance.read_text())
|
|
input_hashes = {str(source_provenance): digest(source_provenance)}
|
|
expected_shared = None
|
|
generated, cases, observations = [], [], []
|
|
for source_case, label in (('late_raw_coupled', 'raw'), ('late_curvature_coupled', 'gravity')):
|
|
folder, record_path = screen / source_case, screen / (source_case + '_run.json')
|
|
record = json.loads(record_path.read_text())
|
|
if record.get('exit_code') != 0 or not record.get('finished_utc') or record.get('timed_out', False):
|
|
raise ValueError(f'Source case has not completed successfully: {source_case}')
|
|
input_hashes[str(record_path)] = digest(record_path)
|
|
names = [PREFIX + name for name in SHARED] + ['metadata.txt'] + [
|
|
PREFIX + name for name in ('summary.csv', 'trials.csv', 'selected_state_a2.txt')]
|
|
for name in names:
|
|
path = folder / name
|
|
value = digest(path)
|
|
if record.get('output_hashes', {}).get(name) != value:
|
|
raise ValueError(f'Source artifact does not match its completed-run hash: {path}')
|
|
input_hashes[str(path)] = value
|
|
shared = {name: input_hashes[str(folder / (PREFIX + name))] for name in SHARED}
|
|
if expected_shared is not None and shared != expected_shared:
|
|
raise ValueError('Source cases do not share exactly the same base, direction, normalization and layout.')
|
|
expected_shared = shared
|
|
meta = dict(line.split('=', 1) for line in (folder / 'metadata.txt').read_text().splitlines() if '=' in line)
|
|
if float(meta['n']) != 1.5 or float(meta['J']) != 0. or meta['central_phase'] != '0':
|
|
raise ValueError('These controls require the saved n=1.5, J=0 model without a central-density constraint.')
|
|
summary = {row['metric']: float(row['value']) for row in rows(folder / (PREFIX + 'summary.csv'))}
|
|
if summary['probe_exception'] != 0 or summary['outer_verified_gate'] != 1:
|
|
raise ValueError('Source probe did not complete with a verified outer Newton direction.')
|
|
matches = [row for row in rows(folder / (PREFIX + 'trials.csv')) if row['sample'] == '2']
|
|
if len(matches) != 1 or float(matches[0]['alpha']) != 1 / 256 or matches[0]['selected_accepted'] != '1' or \
|
|
int(matches[0]['soc_status']) != int(summary['soc_status_kept']):
|
|
raise ValueError('Expected a kept, Armijo-accepted coupled correction at sample 2, alpha=1/256.')
|
|
x, p, selected = [vector(folder / (PREFIX + name)) for name in
|
|
('base_normalized.txt', 'direction.txt', 'selected_state_a2.txt')]
|
|
if not len(x) == len(p) == len(selected):
|
|
raise ValueError('Source vectors have incompatible dimensions.')
|
|
displacement = [y - b for y, b in zip(selected, x)]
|
|
pp = math.fsum(value * value for value in p)
|
|
if not math.isfinite(pp) or pp <= 0.:
|
|
raise ValueError('Projection requires a finite nonzero direction norm.')
|
|
t = math.fsum(step * direction for step, direction in zip(displacement, p)) / pp
|
|
candidate = [math.fsum((b, t * direction)) for b, direction in zip(x, p)]
|
|
perpendicular = [math.fsum((step, -t * direction)) for step, direction in zip(displacement, p)]
|
|
if not math.isfinite(t) or t <= 0. or not all(map(math.isfinite, candidate)):
|
|
raise ValueError('Projected control is nonfinite or does not advance along the supplied direction.')
|
|
candidate_path = output / (label + '_projected_normalized.txt')
|
|
generated.append((candidate_path, candidate))
|
|
cases.append({'id': 'scalar_match_' + label + '_soc', 'arguments': [
|
|
'--index', '1.5', '--angular-momentum', '0', '--no-central', '--max-newton', '0',
|
|
'--strategy', 'baseline', '--preconditioner-variant', 'independent',
|
|
'--probe-state', str(folder / (PREFIX + 'base_physical.txt')),
|
|
'--replay-candidate', str(candidate_path)]})
|
|
observations.append({'source_case': source_case, 'sample': 2, 'nominal_alpha': 1 / 256, 't': t,
|
|
'selected_displacement_norm': norm(displacement), 'scalar_displacement_norm': t * math.sqrt(pp),
|
|
'perpendicular_norm': norm(perpendicular), 'perpendicular_dot_direction':
|
|
math.fsum(value * direction for value, direction in zip(perpendicular, p)),
|
|
'source_selected_residual_norm': float(matches[0]['selected_norm']), 'candidate': str(candidate_path)})
|
|
|
|
# Validate everything before creating output; never replace campaign inputs.
|
|
output.mkdir(parents=True)
|
|
for path, values in generated:
|
|
path.write_text(str(len(values)) + '\n' + ''.join(format(value, '.17g') + '\n' for value in values))
|
|
(output / 'cases.json').write_text(json.dumps(cases, indent=2, allow_nan=False) + '\n')
|
|
generator = Path(__file__).resolve()
|
|
(output / 'generator.py').write_bytes(generator.read_bytes())
|
|
provenance = {'created_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
'generator': str(generator), 'generator_sha256': digest(generator), 'python': sys.version,
|
|
'source_screen': str(screen), 'source_commit': campaign['commit'], 'input_hashes': input_hashes,
|
|
'output_hashes': {path.name: digest(path) for path in output.iterdir()}, 'controls': observations,
|
|
'scope': 'Projection controls only; no FEM evaluations, acceptance conclusions, or linear solves performed.'}
|
|
(output / 'generator_provenance.json').write_text(json.dumps(provenance, indent=2, allow_nan=False) + '\n')
|
|
print(output / 'cases.json')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|