File size: 4,999 Bytes
f211247 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | #!/usr/bin/env python3
"""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))
# Encode JPEG
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()
# Topla
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()
# Summary
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()
|