File size: 8,265 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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | #!/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")
|