| """Create a conservative experiment split without asserting identity separation.""" |
| from __future__ import annotations |
| import argparse |
| from collections import Counter |
| import copy |
| import csv |
| import hashlib |
| import json |
| from pathlib import Path |
| import sys |
| import cv2 |
| import numpy as np |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT)) |
| from tools.review_dataset import digest, safe_path |
|
|
|
|
| def quarantine_paths(rows, groups): |
| """Quarantine non-holdout endpoints, not identity labels or inferred groups.""" |
| excluded, count = set(), 0 |
| for row in rows: |
| a, b = row['left'], row['right'] |
| if not b: |
| continue |
| if a not in groups or b not in groups: |
| raise ValueError('Review references an unknown image') |
| if row['left_split'] != groups[a] or row['right_split'] != groups[b]: |
| raise ValueError('Review split metadata changed') |
| if groups[a] != groups[b]: |
| count += 1 |
| excluded.update(name for name in (a, b) if groups[name] != 'holdout') |
| return excluded, count |
|
|
|
|
| def technical_quality(image): |
| h, w = image.shape[:2] |
| gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) |
| scale = 256 / max(h, w) |
| normalized = cv2.resize(gray, (max(1, round(w*scale)), max(1, round(h*scale)))) |
| sharpness = float(cv2.Laplacian(normalized, cv2.CV_64F).var()) |
| dark, light = float(np.mean(gray <= 5)), float(np.mean(gray >= 250)) |
| flags = [] |
| if min(h, w) < 256: |
| flags.append('small_image') |
| if sharpness < 30: |
| flags.append('possible_global_blur') |
| if dark > .2 or light > .2: |
| flags.append('large_clipped_area') |
| return {'width': w, 'height': h, 'sharpness': round(sharpness, 3), |
| 'dark_fraction': round(dark, 4), 'light_fraction': round(light, 4), 'flags': flags} |
|
|
|
|
| def prepare(root, source, review, output): |
| if output.exists(): |
| raise FileExistsError('Choose a new experiment split directory') |
| split = json.loads((source / 'split.json').read_text(encoding='utf-8')) |
| evidence = json.loads((review / 'review.json').read_text(encoding='utf-8')) |
| if evidence['split_sha256'] != digest(source / 'split.json') or evidence['holdout_sha256'] != digest(source / 'holdout_paths.txt'): |
| raise ValueError('Review no longer matches the source split') |
| if evidence['candidates_sha256'] != digest(review / 'review_candidates.csv'): |
| raise ValueError('Candidate evidence changed') |
| entries = evidence['entries'] |
| groups = {entry['path']: entry['split'] for entry in entries} |
| if len(groups) != len(entries): |
| raise ValueError('Repeated paths in source evidence') |
| for group in ('train', 'validation'): |
| names = [row['path'] for row in split[group]] |
| if names != (source / f'{group}.txt').read_text(encoding='utf-8').splitlines(): |
| raise ValueError('Source manifest changed') |
| if names != [row['path'] for row in entries if row['split'] == group]: |
| raise ValueError('Source evidence changed') |
| with (review / 'review_candidates.csv').open(encoding='utf-8', newline='') as handle: |
| excluded, pair_count = quarantine_paths(list(csv.DictReader(handle)), groups) |
| reasons = {name: 'cross_split_candidate_quarantine' for name in excluded} |
| quality, pixels = [], {} |
| for index, entry in enumerate(entries): |
| name = entry['path'] |
| image = cv2.imread(str(safe_path(root, name))) |
| if image is None: |
| if groups[name] == 'holdout': |
| raise ValueError('Unreadable holdout; benchmark needs repair before proceeding') |
| excluded.add(name) |
| reasons[name] = 'unreadable' |
| quality.append({'path': name, 'split': groups[name], 'flags': ['unreadable']}) |
| continue |
| pixel_hash = hashlib.sha256(str(image.shape).encode()+image.tobytes()).hexdigest() |
| if pixel_hash != entry['pixels_sha256']: |
| raise ValueError(f'Image changed since review: {name}') |
| pixels.setdefault(pixel_hash, []).append(name) |
| stats = technical_quality(image) |
| quality.append(dict(stats, path=name, split=groups[name])) |
| if 'small_image' in stats['flags'] and groups[name] != 'holdout': |
| excluded.add(name) |
| reasons[name] = 'small_image' |
| if (index+1) % 500 == 0: |
| print(f'Technical checks: {index+1}/{len(entries)}', flush=True) |
| |
| priority = {'holdout': 0, 'validation': 1, 'train': 2} |
| for names in pixels.values(): |
| names = sorted((n for n in names if n not in excluded), key=lambda n: (priority[groups[n]], n)) |
| for name in names[1:]: |
| if groups[name] != 'holdout': |
| excluded.add(name) |
| reasons[name] = 'exact_pixel_duplicate' |
| result = copy.deepcopy(split) |
| result.pop('review_receipt', None) |
| for group in ('train', 'validation'): |
| result[group] = [row for row in split[group] if row['path'] not in excluded] |
| if len(result['train']) < 1000 or len(result['validation']) < 20: |
| raise ValueError('Conservative split below experiment minimum; replenish data') |
| result.update(review_status='experimental', lifecycle='frozen', identity_separation_verified=False, |
| mitigation={'policy': 'quarantine_all_non_holdout_cross_split_endpoints_v1', |
| 'source_split_sha256': digest(source/'split.json'), |
| 'candidate_sha256': digest(review/'review_candidates.csv'), |
| 'cross_split_rows': pair_count, 'excluded': reasons, |
| 'technical_checks_complete': True, |
| 'limitation': 'Unflagged identity overlap may remain; not approved for production promotion.'}) |
| kept = {row['path'] for group in ('train', 'validation') for row in result[group]} |
| kept.update(name for name, group in groups.items() if group == 'holdout') |
| sample = [] |
| for group, limit in [('train', 24), ('validation', 8), ('holdout', 8)]: |
| rows = [r for r in quality if r['split'] == group and r['path'] in kept] |
| rows.sort(key=lambda r: hashlib.sha256(('42:'+r['path']).encode()).hexdigest()) |
| sample.extend(rows[:limit]) |
| flagged = [r for r in quality if r['flags'] and r['path'] in kept] |
| |
| shortlist = flagged[:12] |
| report = {'status': 'ready_for_experiment', 'train_count': len(result['train']), |
| 'validation_count': len(result['validation']), 'holdout_path_count': sum(g=='holdout' for g in groups.values()), |
| 'quarantined_count': len(excluded), 'quarantine_reasons': dict(Counter(reasons.values())), |
| 'quality_flag_count': len(flagged), 'sample': sample, 'shortlist': shortlist, |
| 'quality': quality, 'manual_decisions_required': 0, |
| 'identity_separation_verified': False} |
| output.mkdir(parents=True) |
| (output/'thumbnails').mkdir() |
| for row in {r['path']: r for r in sample+shortlist}.values(): |
| image = cv2.imread(str(safe_path(root, row['path']))) |
| h,w = image.shape[:2] |
| thumb = cv2.resize(image, (max(1,round(w*192/max(h,w))),max(1,round(h*192/max(h,w))))) |
| name = hashlib.sha256(row['path'].encode()).hexdigest()+'.jpg' |
| if not cv2.imwrite(str(output/'thumbnails'/name), thumb): |
| raise OSError('Thumbnail write failed') |
| for group in ('train', 'validation'): |
| (output/f'{group}.txt').write_text(''.join(r['path']+'\n' for r in result[group]), encoding='utf-8') |
| (output/'holdout_paths.txt').write_bytes((source/'holdout_paths.txt').read_bytes()) |
| (output/'quality.json').write_text(json.dumps(report, indent=2), encoding='utf-8') |
| result['mitigation']['quality_sha256'] = digest(output/'quality.json') |
| (output/'split.json').write_text(json.dumps(result, indent=2), encoding='utf-8') |
| print(json.dumps({k:v for k,v in report.items() if k not in ('sample','shortlist','quality')}), flush=True) |
| return result, report |
|
|
|
|
| if __name__ == '__main__': |
| p = argparse.ArgumentParser(description=__doc__) |
| p.add_argument('--root', type=Path, default=ROOT/'models/CodeFormer/datasets/ffhq/ffhq_512') |
| p.add_argument('--source', type=Path, default=ROOT/'benchmarks/splits/real_portraits_v1') |
| p.add_argument('--review', type=Path, default=ROOT/'benchmarks/reports/dataset_review_v2') |
| p.add_argument('--output', type=Path, required=True) |
| args = p.parse_args() |
| prepare(args.root, args.source, args.review, args.output) |
|
|
