"""Validate versioned editing evidence and render a local inspection gallery."""
import argparse
import html
import json
from pathlib import Path
from PIL import Image
from scripts.editing_protocol import (PROTOCOL, DEFAULT_SEEDS, validate_seeds, image_diagnostics,
load_input, edit_dimensions)
from scripts.integrity import sha256
from scripts.provenance import validate_roles
from scripts.report import load_records, compare_pixels
def validate_run(root, inputs):
root, inputs = Path(root), Path(inputs)
env = json.loads((root/'environment.json').read_text())
cases = json.loads((root/'cases.json').read_text(encoding='utf-8'))
complete = json.loads((root/'COMPLETE.json').read_text())
if env['protocol'] != PROTOCOL or env['before_load_memory']['allocated_bytes'] != 0:
raise ValueError('Wrong protocol or non-fresh process')
if sha256(root/'cases.json') != env['cases_sha256']:
raise ValueError('Case definitions changed')
validate_seeds(cases, env['seeds'])
records = load_records(root)
expected = {(c['id'], s) for c in cases for s in env['seeds']}
if len({c['id'] for c in cases}) != len(cases) or records.keys() != expected:
raise ValueError('Incomplete or mismatched case/seed matrix')
if (complete['records'] != len(records)
or complete['records_sha256'] != sha256(root/'records.jsonl')):
raise ValueError('Completion record differs from evidence')
for case in cases:
if Path(case['input_file']).name != case['input_file']:
raise ValueError('Invalid input path')
path = inputs/case['input_file']
if sha256(path) != case['input_sha256']:
raise ValueError('Input hash mismatch')
size = edit_dimensions(load_input(path).size, env['resolution'])
for seed in env['seeds']:
row = records[case['id'], seed]
for key, value in dict(protocol=PROTOCOL, prompt=case['prompt'],
source_seed=case.get('source_seed'), input_file=case['input_file'],
input_sha256=case['input_sha256'], width=size[0], height=size[1],
output_resolution=env['resolution'], steps=env['steps'], cfg=1.,
kv_cache=env['kv_cache'], vae_tiling=env['vae_tiling'],
offload=env['offload']).items():
if row[key] != value:
raise ValueError('Record setting mismatch: '+key)
if Path(row['image']).name != row['image']:
raise ValueError('Invalid output path')
with Image.open(root/row['image']) as image:
if image.size != size:
raise ValueError('PNG dimensions differ from requested dimensions')
for key, value in image_diagnostics(image).items():
if row[key] != value:
raise ValueError('Image diagnostic mismatch: '+key)
if 'decode_latents' in row:
if (Path(row['decode_latents']).name != row['decode_latents']
or sha256(root/row['decode_latents']) != row['decode_latents_sha256']):
raise ValueError('Decoder evidence changed')
return env, records, cases
def validate_suite(root, source_root=None):
root = Path(root)
runs = {p: validate_run(root/p, root/'inputs') for p in ('bf16', 'int8')}
a, b = (runs[p][0] for p in ('bf16', 'int8'))
validate_roles(a['model_identity'], b['model_identity'])
for key in ('protocol', 'seeds', 'resolution', 'steps', 'kv_cache', 'vae_tiling',
'offload', 'warmup', 'generator_device', 'cases_sha256', 'gpu', 'cuda',
'packages', 'source_sha256', 'upstream_source_sha256', 'scheduler_config',
'vae_dtype', 'vae_tiles'):
if a[key] != b[key]:
raise ValueError('Paired runtime mismatch: '+key)
if runs['bf16'][1].keys() != runs['int8'][1].keys():
raise ValueError('Paired cases/seeds differ')
if source_root:
required_sources = {'benchmark_edits_v2.py', 'editing_protocol.py', 'benchmark.py', 'runtime.py'}
if set(a['source_sha256']) != required_sources:
raise ValueError('Incomplete measured source inventory')
canonical = json.loads((Path(source_root).parent/'benchmarks/editing-v2.json').read_text(encoding='utf-8'))
if len(canonical) != 5 or runs['bf16'][2] != canonical:
raise ValueError('Release requires all five declared editing cases')
for key, value in dict(seeds=list(DEFAULT_SEEDS), resolution=1024, steps=40,
warmup=True, kv_cache=True, vae_tiling=False,
offload='model', generator_device='cpu', offload_aux_fix=True).items():
if any(run[0].get(key) != value for run in runs.values()):
raise ValueError('Unexpected editing release setting: '+key)
for name, digest in a['source_sha256'].items():
if sha256(Path(source_root)/name) != digest:
raise ValueError('Measured source differs from shipped source: '+name)
isolation = [json.loads(line) for line in (root/'process-isolation.jsonl').read_text().splitlines()]
if len(isolation) != 2 or {r['precision'] for r in isolation} != {'bf16', 'int8'}:
raise ValueError('Missing independent process evidence')
for index, row in enumerate(isolation):
if (row['exit_code'] != 0
or row.get('inference_pid', row['pid']) != runs[row['precision']][0]['pid']):
raise ValueError('Process did not complete successfully')
if row['ended'] < row['started'] or (index and row['started'] < isolation[index-1]['ended']):
raise ValueError('Evaluation processes overlapped')
for key, row in runs['bf16'][1].items():
if row['sigmas'] != runs['int8'][1][key]['sigmas']:
raise ValueError('Sampling schedule mismatch')
return runs
def render(root):
root = Path(root)
runs = validate_suite(root)
rows = []
parts = ['Editing v2 inspection',
'',
'Editing v2: independent editing seeds
Input / BF16 / INT8. '
'Open PNGs for 100% inspection. White, black and checker composites are display aids; '
'raw RGBA outputs are unchanged. Pixel similarity and alpha fractions are not quality scores.
']
for case in runs['bf16'][2]:
for seed in runs['bf16'][0]['seeds']:
key = case['id'], seed
a, b = (runs[p][1][key] for p in ('bf16', 'int8'))
metrics = compare_pixels(root/'bf16'/a['image'], root/'int8'/b['image'])
rows.append(dict(case_id=case['id'], seed=seed, **metrics))
parts.append(f'{html.escape(case["id"])} / {seed}
'
f'{html.escape(case["prompt"])}
')
for src in ['inputs/'+case['input_file'], 'bf16/'+a['image'], 'int8/'+b['image']]:
parts.append(f'})
')
for display in ('alpha', 'white', 'black', 'checker'):
parts.append(f'{display}
')
for p, row in [('bf16', a), ('int8', b)]:
src = f'{p}/diagnostics/{Path(row["image"]).stem}-{display}.png'
if not (root/src).is_file():
raise ValueError('Missing diagnostic preview: '+src)
parts.append(f'
')
parts.append('