score_system_split.py

"""
세로로 긴 악보 페이지를 시스템(큰보표 단위) 경계에서 잘라 staff 검출을 안정화한다.
가로 투영(오선 강조) 후 잉크가 오래 비는 구간을 찾아 분할한다.
"""

from __future__ import annotations

import os
from typing import List, Tuple

import cv2
import numpy as np

_DISABLE = os.environ.get("STELLA_SYSTEM_SPLIT_DISABLE", "").lower() in ("1", "true", "yes")


def horizontal_row_density(image_bgr: np.ndarray) -> np.ndarray:
    gray = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY)
    blur = cv2.GaussianBlur(gray, (3, 3), 0)
    binary = cv2.adaptiveThreshold(
        blur,
        255,
        cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
        cv2.THRESH_BINARY_INV,
        31,
        15,
    )
    h, w = binary.shape
    kernel_width = max(25, w // 12)
    horizontal_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (kernel_width, 1))
    horizontal = cv2.morphologyEx(binary, cv2.MORPH_OPEN, horizontal_kernel, iterations=1)
    return np.sum(horizontal > 0, axis=1).astype(np.float64)


def _moving_average(arr: np.ndarray, window: int) -> np.ndarray:
    window = max(3, window | 1)
    pad = window // 2
    padded = np.pad(arr, (pad, pad), mode="edge")
    cumsum = np.cumsum(np.insert(padded, 0, 0))
    out = (cumsum[window:] - cumsum[:-window]) / float(window)
    return out[: len(arr)]


def _propose_cut_rows(row_density: np.ndarray, h: int) -> List[int]:
    """각 시스템 사이의 컷 y(밴드 경계)."""
    if h < 200:
        return []
    win = max(21, min(71, h // 55))
    sm = _moving_average(row_density, win)
    mx = float(np.max(sm)) or 1.0
    low_thresh = max(12.0, 0.11 * mx)
    min_run = max(12, h // 100)
    below = sm < low_thresh

    runs: List[Tuple[int, int]] = []
    i = 0
    while i < len(below):
        if not below[i]:
            i += 1
            continue
        j = i
        while j < len(below) and below[j]:
            j += 1
        if j - i >= min_run:
            runs.append((i, j))
        i = j

    line_guess = max(10, h // 140)
    min_band = max(8 * line_guess, int(h * 0.11))
    cuts = [(a + b) // 2 for a, b in runs if (b - a) >= min_run * 0.35]

    header = int(h * 0.06)
    cuts = [c for c in cuts if c > header + min_band // 2]

    merged: List[int] = []
    for c in sorted(cuts):
        if not merged or c - merged[-1] >= min_band:
            merged.append(c)
        elif (merged[-1] + c) // 2 != merged[-1]:
            merged[-1] = (merged[-1] + c) // 2

    filtered: List[int] = []
    prev = 0
    for c in merged:
        if c - prev >= min_band and h - c >= min_band:
            filtered.append(c)
            prev = c
    return filtered


def split_work_bgr_into_bands(work_bgr: np.ndarray) -> List[Tuple[np.ndarray, int, int]]:
    """
    work 좌표계에서 (밴드 이미지, y0, x0) 목록. x0는 항상 0.
    비활성·컷 없음·너무 작은 페이지면 [(전체, 0, 0)].
    """
    if _DISABLE:
        return [(work_bgr, 0, 0)]
    h, w = work_bgr.shape[:2]
    rd = horizontal_row_density(work_bgr)
    cuts = _propose_cut_rows(rd, h)
    if not cuts:
        return [(work_bgr, 0, 0)]

    bounds = [0] + cuts + [h]
    bands: List[Tuple[np.ndarray, int, int]] = []
    for i in range(len(bounds) - 1):
        y0, y1 = bounds[i], bounds[i + 1]
        if y1 - y0 < 50:
            continue
        bands.append((work_bgr[y0:y1, 0:w], y0, 0))
    return bands if bands else [(work_bgr, 0, 0)]