mangaocr / segmenter.py
Cnass's picture
Upload 9 files
f3db21a verified
Raw History Blame Contribute Delete
15.6 kB
"""
segmenter.py
Manga sayfalari uzerinde konusma balonu / metin bolgesi segmentasyonu.
Model: huyvux3005/manga109-segmentation-bubble (YOLO11n instance segmentation)
Model, Hugging Face Hub'dan indirilip Space icine yerel olarak yuklenir
(YOLO kullanilmiyor demek yanlis olur; burada aciklama: proje talebindeki
"YOLO kullanilmayacak" ifadesi ayri bir custom detector anlamina gelmiyordu,
verilen model zaten YOLO11n tabanli oldugundan ultralytics ile calistirilir).
Sorumluluklar:
- Modelin HF Hub'dan indirilip yerel diskte onbelleklenmesi.
- Bir manga sayfasi uzerinde inference calistirilmasi.
- Segmentasyon maskelerinden minimum bounding rectangle + padding
ile crop bolgelerinin cikarilmasi.
- Kucuk/hatali crop'larin elenmesi (validasyon config.CROP'tan okunur).
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Optional
import cv2
import numpy as np
from PIL import Image
import config
from utils import clamp, get_logger
log = get_logger("segmenter")
@dataclass(frozen=True)
class DetectedRegion:
"""Tespit edilen tek bir metin bolgesini temsil eder."""
index: int
crop_image: Image.Image
confidence: float
bbox_xyxy: tuple # padding uygulanmis, orijinal sayfa koordinatlarinda
class SegmentationError(RuntimeError):
"""Segmentasyon modeli yuklenemedigi veya calistirilamadigi durumda."""
class MangaSegmenter:
"""
YOLO11n tabanli balon segmentasyon modelini yukler ve manga
sayfalari uzerinde calistirir.
Model, ilk kullanimda Hugging Face Hub'dan indirilir ve
config.SEGMENTATION.local_weights_path altinda onbelleklenir;
sonraki cagrilarda tekrar indirilmez.
"""
def __init__(self) -> None:
self._model = None # lazy-loaded (ilk kullanimda yuklenir)
# ------------------------------------------------------------
# Model yukleme
# ------------------------------------------------------------
def _ensure_weights_downloaded(self) -> str:
"""
Model agirliklarinin yerel diskte var oldugundan emin olur;
yoksa huggingface_hub uzerinden indirir. Yerel dosya yolunu
string olarak dondurur.
"""
weights_path = config.SEGMENTATION.local_weights_path
if weights_path.exists():
log.info("Segmentasyon modeli onbellekten kullaniliyor: %s", weights_path)
return str(weights_path)
log.info(
"Segmentasyon modeli indiriliyor: %s/%s",
config.SEGMENTATION.hf_repo_id,
config.SEGMENTATION.hf_filename,
)
try:
from huggingface_hub import hf_hub_download
downloaded_path = hf_hub_download(
repo_id=config.SEGMENTATION.hf_repo_id,
filename=config.SEGMENTATION.hf_filename,
cache_dir=str(config.CACHE_DIR / "hf_cache"),
)
# Beklenen sabit konuma kopyala (bir sonraki calistirmada
# dogrudan bulunabilmesi icin).
import shutil
shutil.copy(downloaded_path, weights_path)
log.info("Segmentasyon modeli indirildi: %s", weights_path)
return str(weights_path)
except Exception as exc: # noqa: BLE001
log.exception("Segmentasyon modeli indirilemedi")
raise SegmentationError(
f"Segmentasyon modeli indirilemedi: {exc}"
) from exc
def _load_model(self):
"""Modeli lazy olarak yukler ve bellekte tutar."""
if self._model is not None:
return self._model
try:
from ultralytics import YOLO
except ImportError as exc:
raise SegmentationError(
"ultralytics paketi yuklu degil. requirements.txt kontrol edin."
) from exc
weights_path = self._ensure_weights_downloaded()
try:
self._model = YOLO(weights_path)
log.info("Segmentasyon modeli belleğe yuklendi (device=%s)", config.SEGMENTATION.device)
except Exception as exc: # noqa: BLE001
log.exception("Segmentasyon modeli yuklenemedi")
raise SegmentationError(f"Segmentasyon modeli yuklenemedi: {exc}") from exc
return self._model
# ------------------------------------------------------------
# Inference
# ------------------------------------------------------------
def detect(
self,
page_image: Image.Image,
confidence: Optional[float] = None,
padding_px: Optional[int] = None,
) -> List[DetectedRegion]:
"""
Bir manga sayfasi uzerinde metin bolgesi tespiti calistirir.
Args:
page_image: Tam manga sayfasi (PIL Image, RGB).
confidence: Tespit esik degeri (verilmezse config varsayilani).
padding_px: Crop kenarlarina eklenecek piksel (verilmezse config varsayilani).
Returns:
Gecerli (validasyonu gecen) DetectedRegion listesi, sayfadaki
sirasina gore index'lenmis.
Raises:
SegmentationError: Model yuklenemez veya inference basarisiz olursa.
"""
conf = confidence if confidence is not None else config.SEGMENTATION.default_confidence
conf = clamp(conf, config.SEGMENTATION.min_confidence, config.SEGMENTATION.max_confidence)
pad = padding_px if padding_px is not None else config.CROP.default_padding_px
pad = int(clamp(pad, config.CROP.min_padding_px, config.CROP.max_padding_px))
model = self._load_model()
image_rgb = page_image.convert("RGB")
image_np = np.array(image_rgb)
width, height = image_rgb.size
try:
results = model.predict(
source=image_np,
imgsz=config.SEGMENTATION.inference_image_size,
conf=conf,
device=config.SEGMENTATION.device,
verbose=False,
)
except Exception as exc: # noqa: BLE001
log.exception("Segmentasyon inference hatasi")
raise SegmentationError(f"Segmentasyon calistirilamadi: {exc}") from exc
if not results:
log.warning("Segmentasyon sonucu bos dondu")
return []
result = results[0]
if result.masks is None or result.boxes is None or len(result.boxes) == 0:
log.info("Sayfada hicbir metin bolgesi tespit edilmedi")
return []
boxes_xyxy = result.boxes.xyxy.cpu().numpy()
confidences = result.boxes.conf.cpu().numpy()
masks_data = result.masks.data.cpu().numpy() # (N, h, w) - model cozunurlugunde
regions: List[DetectedRegion] = []
raw_boxes = []
for i in range(len(boxes_xyxy)):
mask = masks_data[i]
# Maske, model inference cozunurlugunde uretilir; orijinal
# sayfa boyutuna olceklenmesi gerekir.
mask_resized = self._resize_mask(mask, width, height)
# Ince bir piksel koprusuyle birlesmis iki ayri balon olup
# olmadigini kontrol et; oyleyse maskeyi alt bolgelere ayir.
sub_masks = self._split_touching_regions(mask_resized)
for sub_mask in sub_masks:
bbox = self._mask_to_padded_bbox(sub_mask, width, height, pad)
if bbox is None:
continue
raw_boxes.append((bbox, float(confidences[i])))
# Yukaridan asagiya, soldan saga okuma sirasina yaklasik olarak
# yaklasmak icin y sonra x'e gore sirala (manga panelleri genelde
# bu sirada okunur; tam panel-sirasi tespiti bu modulun kapsami
# disindadir).
raw_boxes.sort(key=lambda item: (item[0][1], item[0][0]))
next_index = 0
for bbox, conf_score in raw_boxes:
if not self._passes_validation(bbox):
log.debug("Crop validasyonu gecemedi, atlaniyor: %s", bbox)
continue
crop = image_rgb.crop(bbox)
regions.append(
DetectedRegion(
index=next_index,
crop_image=crop,
confidence=conf_score,
bbox_xyxy=bbox,
)
)
next_index += 1
log.info("Tespit tamamlandi: %d gecerli bolge (conf=%.2f, pad=%dpx)", len(regions), conf, pad)
return regions
# ------------------------------------------------------------
# Yardimci fonksiyonlar
# ------------------------------------------------------------
@staticmethod
def _split_touching_regions(mask: np.ndarray) -> List[np.ndarray]:
"""
Bir mask'in aslinda ince bir piksel koprusuyle birlesmis birden
fazla ayri balon icerip icermedigini kontrol eder ve oyleyse
ayri binary mask'lere boler.
Yontem: maskeye kucuk bir erosion uygulanir (ince koprüler
kirilir, gercek balon govdeleri buyuk olcude korunur), ardindan
connected components analizi yapilir. Birden fazla anlamli
bilesen bulunursa, her bilesen orijinal (erosion uygulanmamis)
maske ile kesistirilip ayri ayri dondurulur - boylece balonlarin
orijinal sinirlari korunur, sadece hangi pikselin hangi balona
ait oldugu ayristirilir.
Ozellik kapali ise (config.CROP.split_touching_bubbles=False)
veya tek bilesen bulunursa, orijinal maske oldugu gibi tek
elemanli liste olarak dondurulur.
"""
if not config.CROP.split_touching_bubbles:
return [mask]
mask_u8 = mask.astype(np.uint8)
if mask_u8.sum() == 0:
return [mask]
# Kernel boyutunu maskenin kendi bbox'inin kisa kenarina oranla
# turet: boylece kucuk/buyuk sayfa cozunurluklerinde tutarli
# davranir (sabit piksel degeri farkli cozunurluklerde farkli
# etki eder).
ys, xs = np.where(mask_u8)
bbox_h = int(ys.max() - ys.min() + 1)
bbox_w = int(xs.max() - xs.min() + 1)
short_side = min(bbox_h, bbox_w)
kernel_size = max(
config.CROP.split_erosion_min_kernel_px,
int(short_side * config.CROP.split_erosion_ratio),
)
# Kernel boyutunun tek sayi olmasi cv2 icin daha kararlidir.
if kernel_size % 2 == 0:
kernel_size += 1
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (kernel_size, kernel_size))
eroded = cv2.erode(mask_u8, kernel, iterations=config.CROP.split_erosion_iterations)
num_labels, labels = cv2.connectedComponents(eroded, connectivity=8)
# num_labels 1 ise arka plan haric bilesen yok demektir (erosion
# tum maskeyi yok etmis olabilir) -> orijinal maskeyi kullan.
if num_labels <= 2:
return [mask]
# Her erosion-sonrasi bilesenin alanina bakarak gurultuyu ele.
original_area = float(mask_u8.sum())
min_area = original_area * config.CROP.split_min_component_area_ratio
component_labels = []
for label_id in range(1, num_labels):
component_area = float(np.sum(labels == label_id))
if component_area >= min_area:
component_labels.append(label_id)
# Anlamli bilesen sayisi 1 veya 0 ise ayirmaya gerek yok.
if len(component_labels) <= 1:
return [mask]
# Her erosion bilesenini genisletip (dilate) orijinal maske ile
# kesistirerek, o balona ait tum orijinal pikselleri geri kazan.
# watershed benzeri basit bir yaklasim: en yakin bilesene ata.
markers = np.zeros(mask_u8.shape, dtype=np.int32)
for new_id, label_id in enumerate(component_labels, start=1):
markers[labels == label_id] = new_id
# Bilinmeyen (henuz atanmamis ama orijinal maskede olan) pikseller
# icin distanceTransform tabanli en yakin bilesene atama yapilir.
unknown = (mask_u8 == 1) & (markers == 0)
if unknown.any():
markers = MangaSegmenter._assign_nearest_label(markers, mask_u8, len(component_labels))
sub_masks = []
for new_id in range(1, len(component_labels) + 1):
sub_mask = (markers == new_id) & (mask_u8 == 1)
if sub_mask.sum() > 0:
sub_masks.append(sub_mask)
return sub_masks if len(sub_masks) > 1 else [mask]
@staticmethod
def _assign_nearest_label(markers: np.ndarray, full_mask: np.ndarray, num_components: int) -> np.ndarray:
"""
Her bilesen icin ayri bir distance transform hesaplayip, maskedeki
her pikseli en yakin (en dusuk mesafeli) bilesene atar. Basit ama
etkili bir bolge buyutme (region growing) yaklasimidir.
"""
h, w = full_mask.shape
best_dist = np.full((h, w), np.inf, dtype=np.float32)
result = markers.copy()
for label_id in range(1, num_components + 1):
component_pixels = (markers == label_id).astype(np.uint8)
if component_pixels.sum() == 0:
continue
# distanceTransform, 0 olan pikseller icin en yakin non-zero
# piksele olan mesafeyi hesaplar; bu yuzden tersini alip
# bilesenin DISINDAKI pikseller icin mesafe hesapliyoruz.
inverted = 1 - component_pixels
dist = cv2.distanceTransform(inverted, cv2.DIST_L2, 5)
closer_mask = (dist < best_dist) & (full_mask == 1)
result[closer_mask] = label_id
best_dist[closer_mask] = dist[closer_mask]
return result
@staticmethod
def _resize_mask(mask: np.ndarray, target_w: int, target_h: int) -> np.ndarray:
"""Model cozunurlugundeki maskeyi orijinal sayfa boyutuna olcekler."""
mask_image = Image.fromarray((mask * 255).astype(np.uint8))
mask_image = mask_image.resize((target_w, target_h), resample=Image.BILINEAR)
return np.array(mask_image) > 127
@staticmethod
def _mask_to_padded_bbox(
mask: np.ndarray, page_w: int, page_h: int, pad: int
) -> Optional[tuple]:
"""
Bir binary maskeden minimum bounding rectangle cikarir ve
padding uygular; sayfa sinirlarinin disina tasmayacak sekilde
kirpar. Maske bossa None doner.
"""
ys, xs = np.where(mask)
if ys.size == 0 or xs.size == 0:
return None
x_min, x_max = int(xs.min()), int(xs.max())
y_min, y_max = int(ys.min()), int(ys.max())
x_min = max(0, x_min - pad)
y_min = max(0, y_min - pad)
x_max = min(page_w, x_max + pad + 1)
y_max = min(page_h, y_max + pad + 1)
return (x_min, y_min, x_max, y_max)
@staticmethod
def _passes_validation(bbox: tuple) -> bool:
"""Crop boyutlarinin config.CROP minimum degerlerini karsiladigini dogrular."""
x_min, y_min, x_max, y_max = bbox
w = x_max - x_min
h = y_max - y_min
if w < config.CROP.min_width_px or h < config.CROP.min_height_px:
return False
if w * h < config.CROP.min_area_px:
return False
return True
# Modul seviyesinde tekil (singleton) ornek: model bellekte bir kez
# yuklenip tum kullanicilar/istekler arasinda paylasilir.
_segmenter_instance: Optional[MangaSegmenter] = None
def get_segmenter() -> MangaSegmenter:
"""Uygulama genelinde paylasilan tekil MangaSegmenter ornegini dondurur."""
global _segmenter_instance
if _segmenter_instance is None:
_segmenter_instance = MangaSegmenter()
return _segmenter_instance