code/src/preprocessing/pipeline.py

#!/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")