| |
| """Letterbox 224 preprocessed crop'ları WebDataset shard'larına yaz. |
| |
| Config-driven preprocessing.yaml'dan okur. |
| Pipeline: read → CLAHE (conditional) → gamma → letterbox 224 → JPEG |
| MSII proxy optional 2. shard olarak. |
| """ |
| import json, os, io, time, argparse |
| from pathlib import Path |
| from concurrent.futures import ProcessPoolExecutor |
| import sys |
|
|
| ROOT = Path("/arf/scratch/stakan/hitit-proje") |
| SOURCES = ROOT / "datasets" / "sources" |
| sys.path.insert(0, str(ROOT / "hitit_ocr" / "src")) |
|
|
| def process_one(item): |
| """Tek image: preprocess et, JPEG bytes döndür.""" |
| rid, path, label, source, fold = item |
| try: |
| import numpy as np |
| from PIL import Image |
| import cv2 |
| |
| from preprocessing.pipeline import load_config, apply_clahe, apply_gamma, letterbox, is_low_quality, msii_proxy |
| |
| cfg = load_config() |
| |
| img = np.array(Image.open(path).convert('RGB')) |
| enh = cfg.get('enhancement', {}) |
| if enh.get('clahe', {}).get('enabled') and (not enh['clahe'].get('conditional') or is_low_quality(img)): |
| img = apply_clahe(img, clip_limit=enh['clahe'].get('clip_limit', 2.5)) |
| if enh.get('gamma_correction', {}).get('enabled') and (not enh['gamma_correction'].get('conditional') or is_low_quality(img)): |
| img = apply_gamma(img, gamma=enh['gamma_correction'].get('gamma', 1.2)) |
| |
| lb = cfg.get('geometric', {}).get('letterbox', {}) |
| img = letterbox(img, target=lb.get('target_size', 224), margin_ratio=lb.get('margin_ratio', 0.1)) |
| |
| |
| buf = io.BytesIO() |
| Image.fromarray(img).save(buf, format='JPEG', quality=90) |
| jpeg_bytes = buf.getvalue() |
| |
| return (rid, jpeg_bytes, label, source, fold) |
| except Exception as e: |
| return (rid, None, None, None, None) |
|
|
| def main(): |
| import webdataset as wds |
| |
| ap = argparse.ArgumentParser() |
| ap.add_argument('--subset', default='curated', help='curated|all|hitit') |
| ap.add_argument('--workers', type=int, default=64) |
| ap.add_argument('--shard-size', type=int, default=2000) |
| ap.add_argument('--out', default=str(ROOT / "datasets" / "streaming" / "preprocessed_224")) |
| args = ap.parse_args() |
| |
| |
| items = [] |
| for d in sorted(SOURCES.iterdir()): |
| if not d.is_dir(): continue |
| mp = d / "manifest_classification.jsonl" |
| if not mp.exists(): continue |
| with open(mp) as f: |
| for line in f: |
| r = json.loads(line) |
| if r.get('storage') != 'fs' or not r.get('path') or not os.path.exists(r['path']): continue |
| if r.get('quality_gate_pass') is False: continue |
| if args.subset == 'curated' and not r.get('in_curated_pretrain'): continue |
| if args.subset == 'hitit' and r.get('source') != 'hitit': continue |
| items.append((r['id'], r['path'], r.get('unified_label',''), r.get('source',''), r.get('fold', 0))) |
| print(f"İşlenecek: {len(items):,}", flush=True) |
| |
| out_dir = Path(args.out) |
| out_dir.mkdir(parents=True, exist_ok=True) |
| pattern = str(out_dir / "preprocessed-%06d.tar") |
| |
| sink = wds.ShardWriter(pattern, maxcount=args.shard_size, encoder=False) |
| t0 = time.time() |
| done = 0 |
| |
| try: |
| with ProcessPoolExecutor(max_workers=args.workers) as ex: |
| for res in ex.map(process_one, items, chunksize=50): |
| rid, jpeg_bytes, label, source, fold = res |
| if jpeg_bytes: |
| sink.write({ |
| '__key__': rid, |
| 'jpeg': jpeg_bytes, |
| 'json': json.dumps({ |
| 'id': rid, 'unified_label': label, |
| 'source': source, 'fold': fold, |
| }, ensure_ascii=False).encode('utf-8'), |
| }) |
| done += 1 |
| if done % 5000 == 0 and done > 0: |
| rate = done / max(time.time() - t0, 1) |
| print(f" {done:,}/{len(items):,} ({rate:.0f} img/s)", flush=True) |
| finally: |
| sink.close() |
| |
| |
| shards = sorted(out_dir.glob("preprocessed-*.tar")) |
| total_size = sum(s.stat().st_size for s in shards) |
| print(f"\n{len(shards)} shard, {done:,} kayıt, {total_size/1024/1024:.1f} MB") |
| |
| meta = { |
| "format": "WebDataset (preprocessed letterbox 224, CLAHE+gamma conditional)", |
| "n_shards": len(shards), |
| "n_records": done, |
| "total_size_mb": round(total_size/1024/1024, 1), |
| "pattern": f"preprocessed-{{000000..{len(shards)-1:06d}}}.tar", |
| "config_version": "1.0", |
| "preprocessing_steps": ["CLAHE (conditional)", "Gamma (conditional)", "Letterbox 224 median-fill"], |
| } |
| with open(out_dir / "shards.json", 'w') as f: |
| json.dump(meta, f, indent=2, ensure_ascii=False) |
|
|
| if __name__ == '__main__': |
| main() |
|
|
