File size: 3,687 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 | #!/usr/bin/env python3
"""View-aware rotation canonicalization (heuristic).
view_ile_hint tablet_view alanı zaten var. Şimdi: aspect ratio + view hint ile
doğru orientation'a rotate et (offline pass).
"""
import json, os
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor
import numpy as np
from PIL import Image
ROOT = Path("/arf/scratch/stakan/hitit-proje")
SOURCES = ROOT / "datasets" / "sources"
def infer_canonical_rotation(img_path, view_hint):
"""
View hint + aspect ratio ile kanonik rotation tespit et.
Return: rotation_degrees (0, 90, 180, 270) ve landscape_score
"""
try:
with Image.open(img_path) as img:
w, h = img.size
aspect = w / max(h, 1)
# Tablet'ler genelde landscape veya kare. Çok aspect lehinde olan bir view'a göre:
# - obverse/reverse: landscape (aspect>1) tipik
# - edges: portrait (aspect<1) veya ince
# Bu basit heuristic; asıl model-based classifier ayrı.
if view_hint in ('obverse', 'reverse'):
return 0 if aspect >= 0.7 else 90 # portrait'i landscape'e çevir
elif view_hint in ('left_edge', 'right_edge'):
return 90 if aspect > 1 else 0
return 0
except Exception:
return 0
def worker(item):
rid, path, view = item
return (rid, infer_canonical_rotation(path, view))
def main():
items = []
for d in sorted(SOURCES.iterdir()):
if not d.is_dir(): continue
mp = d / "manifest.jsonl"
if not mp.exists(): continue
with open(mp) as f:
for line in f:
r = json.loads(line)
p = r.get('path')
if p and r.get('storage') == 'fs' and r.get('integrity_ok') is True:
items.append((r['id'], p, r.get('view', 'unknown')))
# Unique by path (daha hızlı)
seen_paths = set()
uniq = []
for rid, path, view in items:
if path in seen_paths: continue
seen_paths.add(path)
uniq.append((rid, path, view))
print(f"Tarama: {len(uniq):,} unique image")
results = {}
with ProcessPoolExecutor(max_workers=200) as ex:
for rid, rot in ex.map(worker, uniq, chunksize=500):
results[rid] = rot
# Manifest güncelle
from collections import Counter
rot_counts = Counter()
for d in sorted(SOURCES.iterdir()):
if not d.is_dir(): continue
for mf in ['manifest.jsonl', 'manifest_classification.jsonl', 'manifest_detection.jsonl']:
mp = d / mf
if not mp.exists(): continue
records = []
with open(mp) as f:
for line in f:
r = json.loads(line)
rot = results.get(r.get('id'))
if rot is not None:
r['canonical_rotation_deg'] = int(rot)
if mf == 'manifest.jsonl':
rot_counts[rot] += 1
records.append(r)
with open(mp, 'w') as f:
for r in records:
f.write(json.dumps(r, ensure_ascii=False) + '\n')
print(f"Rotation counts: {dict(rot_counts)}")
with open(ROOT / "datasets" / "processed" / "view_canonicalization_report.json", 'w') as f:
json.dump({
"method": "heuristic (aspect + view_hint)",
"n_images_processed": len(results),
"rotation_distribution": dict(rot_counts),
"note": "Heuristic v1. Model-based classifier (MobileNetV3) için ayrı training gerekir."
}, f, indent=2, ensure_ascii=False)
if __name__ == '__main__':
main()
|