scripts/editing_report.py

"""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 = ['<!doctype html><meta charset="utf-8"><title>Editing v2 inspection</title>',
             '<style>body{font:16px sans-serif;margin:24px}img{max-width:32%;height:auto} '
             'details img{max-width:48%}article{border-top:1px solid #aaa;margin-top:32px}</style>',
             '<h1>Editing v2: independent editing seeds</h1><p>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.</p>']
    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'<article><h2>{html.escape(case["id"])} / {seed}</h2>'
                         f'<p>{html.escape(case["prompt"])}</p>')
            for src in ['inputs/'+case['input_file'], 'bf16/'+a['image'], 'int8/'+b['image']]:
                parts.append(f'<a href="{html.escape(src)}"><img src="{html.escape(src)}"></a>')
            for display in ('alpha', 'white', 'black', 'checker'):
                parts.append(f'<details><summary>{display}</summary>')
                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'<a href="{src}"><img loading="lazy" src="{src}"></a>')
                parts.append('</details>')
            parts.append('</article>')
    (root/'comparison.html').write_text('\n'.join(parts), encoding='utf-8')
    summary = dict(protocol=PROTOCOL, pairs=len(rows), cases=len(runs['bf16'][2]),
                   seeds=runs['bf16'][0]['seeds'], resolution=runs['bf16'][0]['resolution'],
                   observations='See AUDIT.md; similarity is not an edit-success score.', comparisons=rows)
    (root/'summary.json').write_text(json.dumps(summary, indent=2), encoding='utf-8')
    return summary


if __name__ == '__main__':
    ap = argparse.ArgumentParser(__doc__)
    ap.add_argument('--root', required=True)
    args = ap.parse_args()
    print(json.dumps(render(args.root), indent=2))