core/music_parser.py

"""
core/music_parser.py

OMR 결과를 NoteEvent 리스트로 변환하는 파서.

지원 형식:
  - "mock_events": MockOMRAdapter 출력 (dict 리스트)
  - "musicxml":    MusicXML 문자열 (score-partwise 형식, stdlib xml.etree 파싱)

MusicXML 파싱 처리 범위:
  - 단음/화음(<chord>) 처리
  - 쉼표(<rest>) → pitch=0 NoteEvent로 변환 (offset 흐름 유지)
  - <backup>/<forward> 처리 (다성부 offset 유지)
  - <divisions> 변경 추적
  - <key><fifths> 조표 파싱 → 음표 반음 보정
  - <sound tempo=""> 템포 파싱
  - voice/staff 정보 보존
  - namespace 자동 제거
  - 타이(<tie>) 처리: 같은 피치 음표 duration 합산

미지원:
  - score-timewise 형식
  - grace note (skip)
  - 슬러(articulation)
  - 반복기호 펼치기 (D.S. / D.C. / Coda / Segno)
"""

from __future__ import annotations

import xml.etree.ElementTree as ET
from dataclasses import replace as dc_replace
from typing import List

from .models import NoteEvent


class ParseError(Exception):
    """파싱 오류. 어떤 파일/단계에서 실패했는지 포함."""
    pass


# ---------------------------------------------------------------------------
# 공개 인터페이스
# ---------------------------------------------------------------------------

def parse_omr_result(omr_result: dict) -> tuple[List[NoteEvent], dict]:
    """
    OMR 결과 dict를 (NoteEvent 리스트, 메타데이터 dict)로 변환.

    메타데이터 키:
      - "tempo": int  (BPM, 0이면 미발견)

    Raises:
        ParseError: 알 수 없는 형식이거나 파싱 실패 시
    """
    fmt = omr_result.get("format")

    if fmt == "mock_events":
        notes = _parse_mock_events(omr_result["data"])
        return notes, {"tempo": 0}
    elif fmt == "musicxml":
        return _parse_musicxml(omr_result["data"])
    else:
        raise ParseError(f"지원하지 않는 OMR 결과 형식: '{fmt}'")


def parse_musicxml_file(xml_path: str) -> tuple[List[NoteEvent], dict]:
    """
    MusicXML 파일 경로에서 직접 파싱. 테스트 및 직접 사용에 유용.

    Returns:
        (notes, metadata) — parse_omr_result와 동일한 형식

    Raises:
        ParseError: 파일 읽기 실패 또는 파싱 오류
    """
    try:
        with open(xml_path, "r", encoding="utf-8", errors="replace") as f:
            xml_string = f.read()
    except OSError as e:
        raise ParseError(f"MusicXML 파일 읽기 실패 ({xml_path}): {e}")
    return _parse_musicxml(xml_string, source_hint=xml_path)


# ---------------------------------------------------------------------------
# 내부 구현
# ---------------------------------------------------------------------------

def _parse_mock_events(raw_notes: list) -> List[NoteEvent]:
    events = []
    for raw in raw_notes:
        try:
            event = NoteEvent(
                pitch=int(raw["pitch"]),
                start=float(raw["start"]),
                duration=float(raw["duration"]),
                staff=int(raw.get("staff", 1)),
                voice=int(raw.get("voice", 1)),
                part_hint=raw.get("part_hint"),
            )
            events.append(event)
        except (KeyError, ValueError) as e:
            raise ParseError(f"음표 데이터 파싱 오류: {raw!r} — {e}")

    events.sort(key=lambda n: (n.start, n.staff, n.voice))
    return events


# MIDI 음계 반음 수: C=0, D=2, E=4, F=5, G=7, A=9, B=11
_STEP_SEMITONE = {"C": 0, "D": 2, "E": 4, "F": 5, "G": 7, "A": 9, "B": 11}

# 조표 샤프/플랫 순서
_KEY_SHARPS = ["F", "C", "G", "D", "A", "E", "B"]  # 1♯=F#, 2♯=F#C#, ...
_KEY_FLATS  = ["B", "E", "A", "D", "G", "C", "F"]  # 1♭=B♭, 2♭=B♭E♭, ...


def _get_key_alters(fifths: int) -> dict[str, int]:
    """
    조표 fifths 값 → {음이름: alter} 딕셔너리.

    예: fifths=2 (D장조) → {"F": 1, "C": 1}
        fifths=-1 (F장조) → {"B": -1}
    """
    alters: dict[str, int] = {}
    if fifths > 0:
        for i in range(min(fifths, 7)):
            alters[_KEY_SHARPS[i]] = 1
    elif fifths < 0:
        for i in range(min(-fifths, 7)):
            alters[_KEY_FLATS[i]] = -1
    return alters


def _parse_tempo(root: ET.Element) -> int:
    """
    루트 요소에서 첫 번째 <sound tempo="N"/> 값을 반환.
    없으면 0 반환.
    """
    for elem in root.iter("sound"):
        tempo_str = elem.get("tempo")
        if tempo_str:
            try:
                return int(float(tempo_str))
            except (ValueError, TypeError):
                pass
    return 0


def _parse_musicxml(xml_string: str, source_hint: str = "") -> tuple[List[NoteEvent], dict]:
    """
    MusicXML 문자열을 (NoteEvent 리스트, 메타데이터)로 변환.

    Args:
        xml_string:  MusicXML XML 문자열
        source_hint: 오류 메시지에 포함할 파일명/경로 (선택)

    Returns:
        (events, metadata) — metadata에 "tempo" 포함
    """
    src = f" ({source_hint})" if source_hint else ""

    try:
        root = ET.fromstring(xml_string)
    except ET.ParseError as e:
        raise ParseError(f"MusicXML XML 구문 오류{src}: {e}")

    # namespace 제거 (xmlns가 있어도 동일하게 처리)
    for elem in root.iter():
        if "}" in elem.tag:
            elem.tag = elem.tag.split("}")[1]

    root_tag = root.tag
    if root_tag != "score-partwise":
        raise ParseError(
            f"지원하지 않는 MusicXML 루트 요소{src}: '{root_tag}'\n"
            f"  score-partwise 형식만 지원합니다. "
            f"score-timewise는 MuseScore/Audiveris에서 변환 가능합니다."
        )

    # 템포 추출
    tempo = _parse_tempo(root)

    events: List[NoteEvent] = []
    # 타이 추적: key=(part_id, pitch, voice, staff) → events 리스트 인덱스
    tie_pending: dict[tuple, int] = {}

    for part_idx, part_elem in enumerate(root.findall("part")):
        part_id = part_elem.get("id", f"P{part_idx + 1}")
        divisions = 1            # <divisions>: 4분음표당 XML duration 단위
        measure_start = 0.0      # 현재 마디의 시작 beat
        key_alters: dict[str, int] = {}  # 조표 반음 보정 (음이름 → alter)

        for measure_elem in part_elem.findall("measure"):
            current_beat = 0.0       # 마디 내 현재 위치
            prev_note_beat = 0.0     # 직전 비-chord 음표의 시작 위치 (chord 처리용)
            max_beat = 0.0           # 마디 내 도달한 최대 위치 (backup 후에도 유지)

            for child in measure_elem:
                tag = child.tag

                # attributes: divisions, key 업데이트
                if tag == "attributes":
                    div_elem = child.find("divisions")
                    if div_elem is not None and div_elem.text:
                        try:
                            divisions = int(div_elem.text)
                        except ValueError:
                            pass

                    key_elem = child.find("key")
                    if key_elem is not None:
                        fifths_elem = key_elem.find("fifths")
                        if fifths_elem is not None and fifths_elem.text:
                            try:
                                key_alters = _get_key_alters(int(fifths_elem.text))
                            except ValueError:
                                pass

                elif tag == "note":
                    note_event = _parse_note(
                        child, divisions, part_idx,
                        measure_start, current_beat, prev_note_beat,
                        part_id, source_hint, key_alters,
                    )
                    if note_event is not None:
                        # 타이 처리 (쉼표 제외)
                        if note_event.pitch != 0:
                            tie_stop = any(
                                t.get("type") == "stop"
                                for t in child.findall("tie")
                            )
                            tie_start = any(
                                t.get("type") == "start"
                                for t in child.findall("tie")
                            )
                            tie_key = (
                                part_id,
                                note_event.pitch,
                                note_event.voice,
                                note_event.staff,
                            )
                            if tie_stop and tie_key in tie_pending:
                                # 이전 타이 음표에 duration 합산
                                idx = tie_pending.pop(tie_key)
                                old = events[idx]
                                events[idx] = dc_replace(
                                    old, duration=old.duration + note_event.duration
                                )
                                if tie_start:
                                    tie_pending[tie_key] = idx
                                # 새 이벤트는 추가하지 않음
                            else:
                                events.append(note_event)
                                if tie_start:
                                    tie_pending[tie_key] = len(events) - 1
                        else:
                            events.append(note_event)

                    # chord가 아닌 경우에만 위치 전진
                    is_chord = child.find("chord") is not None
                    dur_beats = _get_duration_beats(child, divisions)
                    if not is_chord:
                        prev_note_beat = current_beat
                        current_beat += dur_beats
                        max_beat = max(max_beat, current_beat)

                elif tag == "backup":
                    dur_beats = _get_duration_beats(child, divisions)
                    current_beat = max(0.0, current_beat - dur_beats)

                elif tag == "forward":
                    dur_beats = _get_duration_beats(child, divisions)
                    current_beat += dur_beats
                    max_beat = max(max_beat, current_beat)

            # backup이 있어도 마디 길이는 최대 도달 위치 기준
            measure_start += max_beat

    if not events:
        return [], {"tempo": tempo}

    events.sort(key=lambda n: (n.start, n.staff, n.voice))
    return events, {"tempo": tempo}


def _parse_note(
    note_elem: ET.Element,
    divisions: int,
    part_idx: int,
    measure_start: float,
    current_beat: float,
    prev_note_beat: float,
    part_id: str,
    source_hint: str,
    key_alters: dict[str, int],
) -> NoteEvent | None:
    """
    단일 <note> 요소를 NoteEvent로 변환.

    grace note처럼 duration이 없는 경우는 None 반환 (skip).
    key_alters: 조표에서 파생된 {음이름: alter} 딕셔너리.
                <alter> 태그가 없는 음표의 반음 보정에 사용.
    """
    is_chord = note_elem.find("chord") is not None
    is_rest = note_elem.find("rest") is not None

    dur_beats = _get_duration_beats(note_elem, divisions)
    if dur_beats == 0.0:
        # grace note 또는 duration 0 — skip
        return None

    note_beat = prev_note_beat if is_chord else current_beat
    abs_start = measure_start + note_beat

    voice_elem = note_elem.find("voice")
    voice = int(voice_elem.text) if voice_elem is not None and voice_elem.text else 1

    staff_elem = note_elem.find("staff")
    if staff_elem is not None and staff_elem.text:
        staff_raw = int(staff_elem.text)
    else:
        # <staff> 없으면 voice를 staff 대리자로 사용 (Audiveris 등)
        # voice별로 파트가 분리되도록 함
        staff_raw = voice
    # part 간 staff 번호가 겹치지 않도록 전역 고유값으로 변환
    # ex) P1/staff1=1, P1/staff2=2, P2/staff1=11, P2/staff2=12
    staff = part_idx * 10 + staff_raw

    if is_rest:
        return NoteEvent(
            pitch=0,
            start=abs_start,
            duration=dur_beats,
            staff=staff,
            voice=voice,
        )

    pitch_elem = note_elem.find("pitch")
    if pitch_elem is None:
        return None

    step_elem = pitch_elem.find("step")
    octave_elem = pitch_elem.find("octave")
    alter_elem = pitch_elem.find("alter")

    step = step_elem.text.strip().upper() if step_elem is not None and step_elem.text else "C"
    octave = int(octave_elem.text) if octave_elem is not None and octave_elem.text else 4

    # <alter> 명시 시 우선 사용, 없으면 조표 기본값 적용
    if alter_elem is not None and alter_elem.text:
        alter = int(float(alter_elem.text))
    else:
        alter = key_alters.get(step, 0)

    semitone = _STEP_SEMITONE.get(step, 0)
    pitch = (octave + 1) * 12 + semitone + alter
    pitch = max(0, min(127, pitch))

    return NoteEvent(
        pitch=pitch,
        start=abs_start,
        duration=dur_beats,
        staff=staff,
        voice=voice,
    )


def _get_duration_beats(elem: ET.Element, divisions: int) -> float:
    """<duration> 요소를 4분음표 기준 beats로 변환."""
    dur_elem = elem.find("duration")
    if dur_elem is None or not dur_elem.text:
        return 0.0
    try:
        return int(dur_elem.text) / max(1, divisions)
    except (ValueError, ZeroDivisionError):
        return 0.0