logic.py

"""Small deterministic contracts shared by the Space workers."""

from __future__ import annotations

import re
from dataclasses import dataclass
from math import isfinite
from statistics import median


MAX_SECONDS = 35.0
INSTRUMENTS = {
    "Флейта": (73, "flute"),
    "Кларнет": (71, "clarinet"),
    "Фортепиано": (0, "acoustic piano"),
    "Виолончель": (42, "cello"),
}


@dataclass(frozen=True)
class Beat:
    time: float
    position: int
    meter: int


def drum_hits(beat_lab: str, melody_spans: list[tuple[float, float]],
              duration: float) -> list[tuple[float, int, int]]:
    """Place a small GM kit on SheetSage's source-time beat grid."""
    beats = []
    lines = beat_lab.splitlines()
    if (len(lines) > 1 and lines[0].split() == ["0.0", "1", "1", "8"]
            and lines[1].split()[1:2] == ["1"]):
        # SheetSage sometimes exports a 1/8 time-zero anchor before its real grid.
        lines = lines[1:]
    for line in lines:
        fields = line.split()
        if len(fields) != 4:
            raise ValueError("SheetSage2 вернул некорректную сетку долей")
        time, position, meter, denominator = map(float, fields)
        if (not all(map(isfinite, (time, position, meter, denominator)))
                or int(position) != position or int(meter) != meter
                or int(denominator) != denominator or time < 0
                or int(meter) not in (2, 3, 4, 6)
                or int(denominator) not in (4, 8)
                or not 1 <= position <= meter):
            raise ValueError("SheetSage2 вернул неподдерживаемый размер или долю")
        if beats and time <= beats[-1].time:
            raise ValueError("Доли SheetSage2 идут не по времени")
        beats.append(Beat(time, int(position), int(meter)))
    if len(beats) < 3 or not melody_spans:
        raise ValueError("Для барабанов нужно хотя бы три доли и слышимая мелодия")
    # SheetSage can emit a time-zero anchor immediately before its first real
    # downbeat. Two adjacent beat-1 labels cannot both start the same measure.
    if (beats[0].time == 0 and beats[0].position == beats[1].position == 1
            and beats[0].meter == beats[1].meter):
        beats = beats[1:]
    period = median(b.time - a.time for a, b in zip(beats, beats[1:]))
    if not 0.12 <= period <= 2.0:
        raise ValueError("Не удалось определить разумный темп барабанов")

    def active(time: float) -> bool:
        return any(start < time + period / 2 and end > time - period / 2
                   for start, end in melody_spans)

    hits = []
    for index, beat in enumerate(beats):
        if beat.time >= duration or not active(beat.time):
            continue
        if beat.position == 1 or (beat.meter == 4 and beat.position == 3):
            hits.append((beat.time, 36, 90))  # kick
        elif beat.position in ({2, 4} if beat.meter == 4 else
                              {2} if beat.meter == 2 else
                              {3} if beat.meter == 3 else {4}):
            hits.append((beat.time, 38, 76))  # snare
        hits.append((beat.time, 42, 46))  # closed hi-hat
        if index + 1 < len(beats):
            interval = beats[index + 1].time - beat.time
            halfway = beat.time + interval / 2
            if interval <= period * 1.5 and halfway < duration and active(halfway):
                hits.append((halfway, 42, 32))
    if not hits:
        raise ValueError("Сетка долей не пересекается со звучащим напевом")
    return hits


def instrumental_abc(native: str, keep_chords: bool = False) -> str:
    """Move the native Vocal melody into Ins, failing on unknown ABC layout."""
    if "V: Vocal" not in native and 'V: Ins clef=treble name="Instrumental Melody"' in native:
        return native
    lines = native.splitlines()
    header = []
    body = []
    in_body = False
    for line in lines:
        if line == "V: Vocal":
            in_body = True
        (body if in_body else header).append(line)
    if not body or not any(line.startswith("V: Vocal ") for line in header):
        raise ValueError("SheetSage2 ABC has no expected Vocal melody")
    header = [line for line in header if not line.startswith(("V: Vocal ", "V: Ins "))]
    key_index = next((i for i, line in enumerate(header) if line.startswith("K:")), None)
    if key_index is None:
        raise ValueError("ABC has no key signature")
    header.insert(key_index, 'V: Ins clef=treble name="Instrumental Melody" snm="Ins"')
    out = []
    index = 0
    moved = 0
    while index < len(body):
        line = body[index]
        if line.startswith("%") or not line.strip():
            out.append(line)
            index += 1
            continue
        if (index + 3 >= len(body) or body[index] != "V: Vocal"
                or body[index + 2] != "V: Ins"
                or not re.fullmatch(r"Z\d*\|", body[index + 3])):
            raise ValueError("Unexpected SheetSage2 ABC voice layout")
        melody = body[index + 1]
        if not keep_chords:
            melody = re.sub(r'"[^"\n]*"', "", melody)
        out.extend(("V: Ins", melody))
        moved += 1
        index += 4
    if not moved:
        raise ValueError("No melody bars in SheetSage2 ABC")
    result = "\n".join(header + out) + "\n"
    if "Vocal" in result or (not keep_chords and any('"' in line for line in out)):
        raise ValueError("Vocal label or unexpected chord survived conversion")
    return result


def notes_to_abc(notes: list[tuple[float, float, int]], duration: float) -> str:
    """Make a coarse instrumental ABC solely for YuE when SheetSage has no notes."""
    if not notes:
        raise ValueError("ABC requires notes")
    unit = .0625  # A 1/32 ABC note at 120 quarter-notes per minute.
    events = []
    previous = 0
    for start, end, pitch in sorted(notes):
        first = max(previous, round(start / unit))
        last = max(first + 1, round(end / unit))
        events.append((first, last, pitch))
        previous = last
    total = max(32, round(duration / unit), events[-1][1])
    names = ("C", "^C", "D", "^D", "E", "F", "^F", "G", "^G", "A", "^A", "B")

    def name(pitch: int) -> str:
        octave = pitch // 12 - 1
        base = names[pitch % 12]
        if octave >= 5:
            return base.lower() + "'" * (octave - 5)
        return base + "," * (4 - octave)

    body = []
    cursor = 0
    for first, last, pitch in events:
        for event_start, event_end, symbol in ((cursor, first, "z"),
                                               (first, last, name(pitch))):
            while event_start < event_end:
                chunk = min(event_end - event_start, 32 - event_start % 32)
                more = symbol != "z" and event_start + chunk < event_end
                body.append(f"{symbol}{chunk}{'-' if more else ''}")
                event_start += chunk
                if event_start % 32 == 0:
                    body.append("|")
        cursor = last
    while cursor < total:
        chunk = min(total - cursor, 32 - cursor % 32)
        body.append(f"z{chunk}")
        cursor += chunk
        if cursor % 32 == 0:
            body.append("|")
    if body[-1] != "|":
        body.append("|")
    return ('X:1\nT:Hum F0 fallback\nM:4/4\nL:1/32\nQ:1/4=120\n'
            'V: Ins clef=treble name="Instrumental Melody" snm="Ins"\n'
            'K:C\nV: Ins\n' + " ".join(body) + "\n")


def validate_duration(seconds: float) -> None:
    if not 0.4 <= seconds <= MAX_SECONDS:
        raise ValueError(f"Нужен напев от 0,4 до {MAX_SECONDS:g} секунд; сейчас {seconds:.1f} с.")