| |
| """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)) |
| |
| import re |
| text = open(path).read() |
| |
| 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()) |
| |
| 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) |
| |
| mx = max(abs(lap.min()), abs(lap.max()), 1e-6) |
| channels.append(lap / mx) |
| |
| stacked = np.mean(channels, axis=0) |
| |
| 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) |
| |
| |
| 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 |
| 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', {}) |
| |
| 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]))) |
| |
| 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)) |
| |
| |
| 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} |
| |
| |
| 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 |
| |
| |
| 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") |
|
|
