| |
| """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) |
| |
| |
| |
| |
| if view_hint in ('obverse', 'reverse'): |
| return 0 if aspect >= 0.7 else 90 |
| 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'))) |
| |
| 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 |
| |
| |
| 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() |
|
|
