#!/usr/bin/env python3 """SOTA preprocessing pipeline — config-driven. Her image için config'e göre apply eder: - Letterbox 224 (DeepScribe stil, margin_ratio) - CLAHE (conditional: dark/low-contrast ise) - Gamma correction - MSII proxy (multi-scale Laplacian of Gaussian) - Sauvola binarization (auxiliary) - Dataset-specific normalization Deterministik (pipeline_hash ile reproducibility). """ import hashlib, json from pathlib import Path from typing import Tuple import numpy as np from PIL import Image import cv2 try: import yaml except ImportError: yaml = None ROOT = Path("/arf/scratch/stakan/hitit-proje") def load_config(path=None): path = path or ROOT / "hitit_ocr" / "configs" / "preprocessing.yaml" if yaml: return yaml.safe_load(open(path)) # Fallback: basit YAML parse (dışa bağımlı değil) import re text = open(path).read() # Minimal YAML; tam cover etmez ama config'i okumak yeter import subprocess result = subprocess.run(['python3', '-c', f"import yaml; print(__import__('json').dumps(yaml.safe_load(open(r'{path}'))))"], capture_output=True, text=True) if result.returncode == 0: return json.loads(result.stdout) return {} def compute_pipeline_hash(config: dict) -> str: """Config + kod version'u hash'le — reproducibility.""" h = hashlib.sha256() h.update(json.dumps(config, sort_keys=True).encode()) # Kod version: bu dosyanın hash'ı try: src_hash = hashlib.sha256(open(__file__, 'rb').read()).hexdigest()[:16] except Exception: src_hash = "unknown" h.update(src_hash.encode()) return h.hexdigest()[:16] def _median_border_color(img: np.ndarray, border_ratio: float = 0.05) -> Tuple[int, int, int]: """Image border'ın median RGB'si — letterbox fill için.""" h, w = img.shape[:2] bw = max(1, int(min(h, w) * border_ratio)) borders = np.concatenate([ img[:bw, :].reshape(-1, img.shape[2]), img[-bw:, :].reshape(-1, img.shape[2]), img[:, :bw].reshape(-1, img.shape[2]), img[:, -bw:].reshape(-1, img.shape[2]), ]) med = np.median(borders, axis=0).astype(np.uint8) return tuple(int(x) for x in med) def letterbox(img: np.ndarray, target: int, margin_ratio: float = 0.1, fill: str = "median_border"): """Aspect-preserving letterbox with margin + median-fill padding.""" h, w = img.shape[:2] inner = int(target * (1 - margin_ratio)) scale = min(inner / max(h, 1), inner / max(w, 1)) new_h, new_w = int(h * scale), int(w * scale) if new_h > 0 and new_w > 0: resized = cv2.resize(img, (new_w, new_h), interpolation=cv2.INTER_LANCZOS4) else: resized = img.copy() if fill == "median_border" and img.shape[2] == 3: pad_color = _median_border_color(img) else: pad_color = (0, 0, 0) canvas = np.full((target, target, img.shape[2]), pad_color, dtype=np.uint8) y = (target - new_h) // 2 x = (target - new_w) // 2 canvas[y:y+new_h, x:x+new_w] = resized return canvas def apply_clahe(img: np.ndarray, clip_limit: float = 2.5, tile_grid: Tuple[int,int] = (8,8)) -> np.ndarray: """CLAHE on L-channel of LAB.""" lab = cv2.cvtColor(img, cv2.COLOR_RGB2LAB) l, a, b = cv2.split(lab) clahe = cv2.createCLAHE(clipLimit=clip_limit, tileGridSize=tile_grid) l = clahe.apply(l) lab = cv2.merge([l, a, b]) return cv2.cvtColor(lab, cv2.COLOR_LAB2RGB) def apply_gamma(img: np.ndarray, gamma: float = 1.2) -> np.ndarray: table = np.array([((i / 255.0) ** (1.0/gamma)) * 255 for i in range(256)]).astype(np.uint8) return cv2.LUT(img, table) def is_low_quality(img: np.ndarray) -> bool: """CLAHE'nin conditional uygulanıp uygulanmayacağı.""" gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY) blur = cv2.Laplacian(gray, cv2.CV_64F).var() mean = gray.mean() std = gray.std() return blur < 100 or mean < 80 or mean > 200 or std < 30 def msii_proxy(img: np.ndarray, sigmas=(1.0, 2.0, 4.0, 8.0)) -> np.ndarray: """Multi-scale Laplacian of Gaussian — curvature proxy for cuneiform wedges. Reference: Mara/Bogacz MSII pipeline, HeiCuBeDa (JOAD 2025). Output: single-channel float [-1..1], wedge incisions bright. """ gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY).astype(np.float32) / 255.0 channels = [] for s in sigmas: blur = cv2.GaussianBlur(gray, (0, 0), sigmaX=s) lap = cv2.Laplacian(blur, cv2.CV_32F, ksize=3) # Normalize [-1, 1] mx = max(abs(lap.min()), abs(lap.max()), 1e-6) channels.append(lap / mx) # Ortalama (farklı ölçeklerin aggregate'i) stacked = np.mean(channels, axis=0) # Return [0, 255] uint8 return ((stacked * 0.5 + 0.5) * 255).astype(np.uint8) def apply_sauvola(img: np.ndarray, window_size: int = 25, k: float = 0.2) -> np.ndarray: """Sauvola binarization (classic) — wedge marks vs tablet surface.""" gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY) # OpenCV'nin kendi sauvola'sı yok; scikit-image ile yerine orta tahmin # Basit implementasyon: local mean + std mean = cv2.boxFilter(gray.astype(np.float32), -1, (window_size, window_size)) sq = cv2.boxFilter((gray.astype(np.float32))**2, -1, (window_size, window_size)) std = np.sqrt(np.maximum(sq - mean**2, 0)) R = 128.0 # dinamik aralık threshold = mean * (1 + k * (std / R - 1)) binary = (gray < threshold).astype(np.uint8) * 255 return binary def normalize(img: np.ndarray, mean, std) -> np.ndarray: """RGB [0,255] uint8 → normalized float32.""" arr = img.astype(np.float32) / 255.0 arr = (arr - np.array(mean)) / np.array(std) return arr def preprocess(img_path: str, config: dict, norm_stats: dict = None) -> dict: """Config'e göre pipeline çalıştır, çıktıyı dict olarak döndür.""" img = np.array(Image.open(img_path).convert('RGB')) enh = config.get('enhancement', {}) # CLAHE (conditional) if enh.get('clahe', {}).get('enabled'): if not enh['clahe'].get('conditional') or is_low_quality(img): img = apply_clahe(img, clip_limit=enh['clahe'].get('clip_limit', 2.5), tile_grid=tuple(enh['clahe'].get('tile_grid', [8,8]))) # Gamma if enh.get('gamma_correction', {}).get('enabled'): if not enh['gamma_correction'].get('conditional') or is_low_quality(img): img = apply_gamma(img, gamma=enh['gamma_correction'].get('gamma', 1.2)) # Letterbox geom = config.get('geometric', {}) lb = geom.get('letterbox', {}) if lb: img = letterbox(img, target=lb.get('target_size', 224), margin_ratio=lb.get('margin_ratio', 0.1), fill=lb.get('fill_mode', 'median_border')) outputs = {"rgb": img} # Auxiliary channels cs = config.get('cuneiform_specific', {}) if cs.get('msii_proxy', {}).get('enabled'): msii = msii_proxy(img, sigmas=tuple(cs['msii_proxy'].get('sigmas', [1.0,2.0,4.0,8.0]))) outputs['msii'] = msii bn = config.get('binarization', {}) if bn.get('enabled'): sv = bn.get('sauvola', {}) binary = apply_sauvola(img, window_size=sv.get('window_size', 25), k=sv.get('k', 0.2)) outputs['binary'] = binary # Normalization nm = config.get('normalization', {}) if nm.get('strategy') == 'dataset_specific' and norm_stats: mean = norm_stats['global_rgb']['mean'] std = norm_stats['global_rgb']['std'] outputs['normalized'] = normalize(img, mean, std) elif nm.get('strategy') == 'imagenet': outputs['normalized'] = normalize(img, [0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) return outputs if __name__ == '__main__': import sys cfg = load_config() h = compute_pipeline_hash(cfg) print(f"pipeline_hash: {h}") print(f"steps enabled:") for sect in ['enhancement','geometric','cuneiform_specific','binarization']: print(f" {sect}:") for k, v in cfg.get(sect, {}).items(): if isinstance(v, dict) and v.get('enabled'): print(f" - {k}: ON")