1571 lines
54 KiB
Python
1571 lines
54 KiB
Python
"""Convert timed musical events into validated two-voice ABC notation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import os
|
|
import re
|
|
from collections import Counter
|
|
from dataclasses import dataclass, replace
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Iterable, Sequence
|
|
|
|
import numpy as np
|
|
import pretty_midi
|
|
|
|
from .io_sheetsage2 import atomic_write_text
|
|
|
|
|
|
SUBBEAT_DIVISION = 4
|
|
VOICE_IDS = ("Vocal", "Ins")
|
|
NO_CHORDS = frozenset({"N", "X", "?"})
|
|
|
|
|
|
class AbcRebuildError(ValueError):
|
|
"""Base class for deterministic reconstruction failures."""
|
|
|
|
|
|
class BeatGridError(AbcRebuildError):
|
|
pass
|
|
|
|
|
|
class ChordSymbolError(AbcRebuildError):
|
|
pass
|
|
|
|
|
|
class MelodyVoiceError(AbcRebuildError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BeatEvent:
|
|
time: float
|
|
beat_id: int
|
|
declared_numerator: int
|
|
denominator: int
|
|
line_no: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Measure:
|
|
index: int
|
|
start_beat: int
|
|
end_beat: int
|
|
numerator: int
|
|
denominator: int
|
|
pickup: bool = False
|
|
partial: bool = False
|
|
inferred: bool = False
|
|
notated_numerator: int | None = None
|
|
notated_denominator: int | None = None
|
|
pad_before: bool = False
|
|
|
|
@property
|
|
def beat_count(self) -> int:
|
|
return self.end_beat - self.start_beat
|
|
|
|
@property
|
|
def start_t(self) -> int:
|
|
return self.start_beat * SUBBEAT_DIVISION
|
|
|
|
@property
|
|
def end_t(self) -> int:
|
|
return self.end_beat * SUBBEAT_DIVISION
|
|
|
|
@property
|
|
def abc_numerator(self) -> int:
|
|
return self.notated_numerator or self.numerator
|
|
|
|
@property
|
|
def abc_denominator(self) -> int:
|
|
return self.notated_denominator or self.denominator
|
|
|
|
|
|
@dataclass
|
|
class RebuiltAbcScore:
|
|
beats: list[BeatEvent]
|
|
measures: list[Measure]
|
|
subbeat_times: np.ndarray
|
|
subbeat_quarters: np.ndarray
|
|
subbeat_denominators: np.ndarray
|
|
key_arr: np.ndarray
|
|
chord_arr: np.ndarray
|
|
structure_events: list[tuple[int, str]]
|
|
voice_arrs: dict[str, np.ndarray]
|
|
diagnostics: list[str]
|
|
subbeat_div: int = SUBBEAT_DIVISION
|
|
|
|
|
|
@dataclass
|
|
class MeasureGroup:
|
|
measures: list[Measure]
|
|
structure_labels: list[str]
|
|
meter_changed: bool
|
|
key_changed: bool
|
|
|
|
|
|
_QUALITY_TO_ABC = {
|
|
"maj": "",
|
|
"min": "m",
|
|
"dim": "dim",
|
|
"aug": "aug",
|
|
"7": "7",
|
|
"maj7": "maj7",
|
|
"min7": "m7",
|
|
"dim7": "dim7",
|
|
"hdim7": "m7b5",
|
|
"sus4": "sus4",
|
|
"sus2": "sus2",
|
|
"maj6": "6",
|
|
"min6": "m6",
|
|
"sus4(b7)": "7sus4",
|
|
# abc2midi and SymMusic both accept the parenthesized major seventh.
|
|
# Common aliases such as mmaj7/mM7 trigger abc2midi diagnostics.
|
|
"minmaj7": "m(maj7)",
|
|
}
|
|
|
|
_NATURAL_PITCH_CLASS = {
|
|
"C": 0,
|
|
"D": 2,
|
|
"E": 4,
|
|
"F": 5,
|
|
"G": 7,
|
|
"A": 9,
|
|
"B": 11,
|
|
}
|
|
_LETTERS = "CDEFGAB"
|
|
_SHARP_PITCH_NAMES = ("C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B")
|
|
_FLAT_PITCH_NAMES = ("C", "Db", "D", "Eb", "E", "F", "Gb", "G", "Ab", "A", "Bb", "B")
|
|
_ROOT_RE = re.compile(r"^(?P<letter>[A-G])(?P<accidental>#{0,2}|b{0,2})$")
|
|
_BASS_DEGREE_RE = re.compile(r"^(?P<accidental>#{0,2}|b{0,2})(?P<degree>[1-9]|1[0-3])$")
|
|
|
|
_KEY_SIGNATURE_ACCIDENTALS = {
|
|
"C": 0,
|
|
"G": 1,
|
|
"D": 2,
|
|
"A": 3,
|
|
"E": 4,
|
|
"B": 5,
|
|
"F#": 6,
|
|
"C#": 7,
|
|
"F": -1,
|
|
"Bb": -2,
|
|
"Eb": -3,
|
|
"Ab": -4,
|
|
"Db": -5,
|
|
"Gb": -6,
|
|
"Cb": -7,
|
|
"Am": 0,
|
|
"Em": 1,
|
|
"Bm": 2,
|
|
"F#m": 3,
|
|
"C#m": 4,
|
|
"G#m": 5,
|
|
"D#m": 6,
|
|
"A#m": 7,
|
|
"Dm": -1,
|
|
"Gm": -2,
|
|
"Cm": -3,
|
|
"Fm": -4,
|
|
"Bbm": -5,
|
|
"Ebm": -6,
|
|
"Abm": -7,
|
|
}
|
|
|
|
# Keep the standard key-relative chromatic spelling. MIDI carries only
|
|
# pitch, not note names, so a deterministic key-based table is preferable to
|
|
# rewriting every chromatic pitch with a single sharp/flat. In remote sharp
|
|
# and flat keys this deliberately permits musically useful double accidentals,
|
|
# for example MIDI G as F## in G# minor.
|
|
_KEY_RELATIVE_PITCH_NAMES = {
|
|
7: ("B#", "C#", "C##", "D#", "D##", "E#", "F#", "F##", "G#", "G##", "A#", "B"),
|
|
6: ("B#", "C#", "C##", "D#", "E", "E#", "F#", "F##", "G#", "G##", "A#", "B"),
|
|
5: ("B#", "C#", "C##", "D#", "E", "E#", "F#", "F##", "G#", "A", "A#", "B"),
|
|
4: ("B#", "C#", "D", "D#", "E", "E#", "F#", "F##", "G#", "A", "A#", "B"),
|
|
3: ("B#", "C#", "D", "D#", "E", "E#", "F#", "G", "G#", "A", "A#", "B"),
|
|
2: ("C", "C#", "D", "D#", "E", "E#", "F#", "G", "G#", "A", "A#", "B"),
|
|
1: ("C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"),
|
|
0: ("C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "Bb", "B"),
|
|
-1: ("C", "C#", "D", "Eb", "E", "F", "F#", "G", "G#", "A", "Bb", "B"),
|
|
-2: ("C", "C#", "D", "Eb", "E", "F", "F#", "G", "Ab", "A", "Bb", "B"),
|
|
-3: ("C", "Db", "D", "Eb", "E", "F", "F#", "G", "Ab", "A", "Bb", "B"),
|
|
-4: ("C", "Db", "D", "Eb", "E", "F", "Gb", "G", "Ab", "A", "Bb", "B"),
|
|
-5: ("C", "Db", "D", "Eb", "E", "F", "Gb", "G", "Ab", "A", "Bb", "Cb"),
|
|
-6: ("C", "Db", "D", "Eb", "Fb", "F", "Gb", "G", "Ab", "A", "Bb", "Cb"),
|
|
-7: ("C", "Db", "D", "Eb", "Fb", "F", "Gb", "G", "Ab", "Bbb", "Bb", "Cb"),
|
|
}
|
|
|
|
|
|
def _read_tsv(path: os.PathLike[str] | str, min_columns: int) -> list[tuple[int, list[str]]]:
|
|
rows = []
|
|
with open(path, "r", encoding="utf-8-sig") as handle:
|
|
for line_no, raw_line in enumerate(handle, 1):
|
|
line = raw_line.rstrip("\r\n")
|
|
if not line.strip():
|
|
continue
|
|
columns = line.split("\t")
|
|
if len(columns) < min_columns:
|
|
raise AbcRebuildError(
|
|
f"{path}:{line_no}: expected at least {min_columns} tab-separated columns"
|
|
)
|
|
rows.append((line_no, columns))
|
|
return rows
|
|
|
|
|
|
def read_beats(path: os.PathLike[str] | str) -> list[BeatEvent]:
|
|
return _parse_beats(_read_tsv(path, 3), path)
|
|
|
|
|
|
def _row_entries(rows, source):
|
|
entries = []
|
|
for line_no, row in enumerate(rows, 1):
|
|
if len(row) < 3:
|
|
raise AbcRebuildError(f"{source}:{line_no}: expected at least 3 columns")
|
|
entries.append((line_no, [str(value) for value in row]))
|
|
return entries
|
|
|
|
|
|
def _parse_beats(entries, path):
|
|
beats = []
|
|
for line_no, row in entries:
|
|
meter_text = row[2]
|
|
if len(row) >= 4:
|
|
numerator_text, denominator_text = meter_text, row[3]
|
|
elif "/" in meter_text:
|
|
numerator_text, denominator_text = meter_text.split("/", 1)
|
|
else:
|
|
numerator_text, denominator_text = meter_text, "4"
|
|
try:
|
|
beat = BeatEvent(
|
|
time=float(row[0]),
|
|
beat_id=int(row[1]),
|
|
declared_numerator=int(numerator_text),
|
|
denominator=int(denominator_text),
|
|
line_no=line_no,
|
|
)
|
|
except ValueError as exc:
|
|
raise BeatGridError(f"{path}:{line_no}: invalid beat row {row!r}") from exc
|
|
if beat.beat_id < 1:
|
|
raise BeatGridError(f"{path}:{line_no}: beat ID must be positive")
|
|
if beat.declared_numerator < 1:
|
|
raise BeatGridError(f"{path}:{line_no}: meter numerator must be positive")
|
|
if beat.denominator < 1 or beat.denominator & (beat.denominator - 1):
|
|
raise BeatGridError(
|
|
f"{path}:{line_no}: meter denominator must be a positive power of two"
|
|
)
|
|
if beats and beat.time <= beats[-1].time:
|
|
raise BeatGridError(f"{path}:{line_no}: beat times must be strictly increasing")
|
|
beats.append(beat)
|
|
if len(beats) < 2:
|
|
raise BeatGridError(f"{path}: at least two beat events are required")
|
|
return beats
|
|
|
|
|
|
def read_chords(path: os.PathLike[str] | str) -> list[tuple[float, float, str]]:
|
|
return _parse_chords(_read_tsv(path, 3), path)
|
|
|
|
|
|
def _parse_chords(entries, path):
|
|
rows = []
|
|
previous_end = None
|
|
for line_no, row in entries:
|
|
start, end, chord = float(row[0]), float(row[1]), row[2].strip()
|
|
if end <= start:
|
|
raise ChordSymbolError(f"{path}:{line_no}: chord end must be after start")
|
|
if previous_end is not None and start < previous_end - 1e-6:
|
|
raise ChordSymbolError(f"{path}:{line_no}: overlapping chord intervals")
|
|
chord_symbol_to_abc(chord)
|
|
rows.append((start, end, chord))
|
|
previous_end = end
|
|
return rows
|
|
|
|
|
|
def read_keys(path: os.PathLike[str] | str) -> list[tuple[float, float, str]]:
|
|
return _parse_keys(_read_tsv(path, 3), path)
|
|
|
|
|
|
def _parse_keys(entries, path):
|
|
rows = []
|
|
previous_end = None
|
|
for line_no, row in entries:
|
|
start, end, key = float(row[0]), float(row[1]), row[2].strip()
|
|
if end <= start:
|
|
raise AbcRebuildError(f"{path}:{line_no}: key end must be after start")
|
|
if previous_end is not None and start < previous_end - 1e-6:
|
|
raise AbcRebuildError(f"{path}:{line_no}: overlapping key intervals")
|
|
normalized = key_symbol_to_abc(key)
|
|
rows.append((start, end, normalized))
|
|
previous_end = end
|
|
if not rows:
|
|
raise AbcRebuildError(f"{path}: at least one key interval is required")
|
|
return rows
|
|
|
|
|
|
def read_structures(path: os.PathLike[str] | str) -> list[tuple[float, float, str]]:
|
|
return _parse_structures(_read_tsv(path, 3), path)
|
|
|
|
|
|
def _parse_structures(entries, path):
|
|
rows = []
|
|
previous_end = None
|
|
for line_no, row in entries:
|
|
start, end, label = float(row[0]), float(row[1]), row[2].strip()
|
|
if end <= start:
|
|
raise AbcRebuildError(f"{path}:{line_no}: structure end must be after start")
|
|
if previous_end is not None and start < previous_end - 1e-6:
|
|
raise AbcRebuildError(f"{path}:{line_no}: overlapping structure intervals")
|
|
rows.append((start, end, label))
|
|
previous_end = end
|
|
return rows
|
|
|
|
|
|
def _mode_with_first_tiebreak(values: Sequence[int]) -> int:
|
|
counts = Counter(values)
|
|
maximum = max(counts.values())
|
|
return next(value for value in values if counts[value] == maximum)
|
|
|
|
|
|
def infer_measures(
|
|
beats: Sequence[BeatEvent],
|
|
*,
|
|
meter_conflict: str = "infer",
|
|
) -> tuple[list[Measure], list[str]]:
|
|
"""Infer self-consistent measures from actual downbeat boundaries."""
|
|
|
|
if meter_conflict not in {"infer", "reject"}:
|
|
raise ValueError("meter_conflict must be 'infer' or 'reject'")
|
|
downbeat_indices = [index for index, beat in enumerate(beats) if beat.beat_id == 1]
|
|
if not downbeat_indices:
|
|
raise BeatGridError("No downbeat (beat ID 1) exists in the beat lab")
|
|
spans: list[tuple[int, int, bool, bool]] = []
|
|
if downbeat_indices[0] > 0:
|
|
spans.append((0, downbeat_indices[0], True, False))
|
|
spans.extend(
|
|
(start, end, False, False)
|
|
for start, end in zip(downbeat_indices, downbeat_indices[1:])
|
|
)
|
|
if downbeat_indices[-1] < len(beats) - 1:
|
|
# Exported beat labs use their last row as the score end boundary. If
|
|
# that row is not a downbeat, the final bar is intentionally truncated.
|
|
spans.append((downbeat_indices[-1], len(beats) - 1, False, True))
|
|
if not spans:
|
|
raise BeatGridError("No positive-length measure exists between downbeats")
|
|
|
|
diagnostics = []
|
|
measures = []
|
|
for measure_index, (start, end, pickup, partial) in enumerate(spans):
|
|
events = list(beats[start:end])
|
|
beat_count = len(events)
|
|
if beat_count < 1:
|
|
raise BeatGridError(f"Measure {measure_index}: empty downbeat span")
|
|
ids = [event.beat_id for event in events]
|
|
expected_ids = list(range(ids[0], ids[0] + beat_count))
|
|
if ids != expected_ids:
|
|
line_numbers = [event.line_no for event in events]
|
|
raise BeatGridError(
|
|
f"Measure {measure_index} (beat rows {line_numbers[0]}-{line_numbers[-1]}): "
|
|
f"non-consecutive beat IDs {ids!r}"
|
|
)
|
|
if not pickup and ids[0] != 1:
|
|
raise BeatGridError(f"Measure {measure_index}: full measure does not start at beat ID 1")
|
|
|
|
denominators = [event.denominator for event in events]
|
|
denominator = _mode_with_first_tiebreak(denominators)
|
|
declared_numerators = [event.declared_numerator for event in events]
|
|
declared_numerator = _mode_with_first_tiebreak(declared_numerators)
|
|
numerator_conflict = any(value != beat_count for value in declared_numerators)
|
|
denominator_conflict = any(value != denominator for value in denominators)
|
|
pad_final_partial = (
|
|
partial
|
|
and len(set(declared_numerators)) == 1
|
|
and not denominator_conflict
|
|
and declared_numerator >= beat_count
|
|
)
|
|
inferred = pickup or partial or numerator_conflict or denominator_conflict
|
|
unresolved_numerator_conflict = (
|
|
numerator_conflict
|
|
and not pad_final_partial
|
|
and not pickup
|
|
)
|
|
if (
|
|
unresolved_numerator_conflict or denominator_conflict
|
|
) and meter_conflict == "reject":
|
|
raise BeatGridError(
|
|
f"Measure {measure_index}: {beat_count} actual beats conflict with declarations "
|
|
f"{list(zip(declared_numerators, denominators))!r}"
|
|
)
|
|
if pad_final_partial and declared_numerator > beat_count:
|
|
diagnostics.append(
|
|
f"measure {measure_index}: padded final {beat_count}/{denominator} span "
|
|
f"to declared {declared_numerator}/{denominator} with trailing rest"
|
|
)
|
|
elif numerator_conflict:
|
|
diagnostics.append(
|
|
f"measure {measure_index}: inferred {beat_count}/{denominator} from downbeat span; "
|
|
f"declared numerators were {declared_numerators}"
|
|
)
|
|
if denominator_conflict:
|
|
diagnostics.append(
|
|
f"measure {measure_index}: placed denominator {denominator} at the measure boundary; "
|
|
f"row declarations were {denominators}"
|
|
)
|
|
measures.append(
|
|
Measure(
|
|
index=measure_index,
|
|
start_beat=start,
|
|
end_beat=end,
|
|
numerator=beat_count,
|
|
denominator=denominator,
|
|
pickup=pickup,
|
|
partial=partial,
|
|
inferred=inferred,
|
|
notated_numerator=(
|
|
declared_numerator if pad_final_partial else beat_count
|
|
),
|
|
)
|
|
)
|
|
if len(measures) >= 2:
|
|
first = measures[0]
|
|
following = measures[1]
|
|
first_duration = first.numerator / first.denominator
|
|
following_duration = (
|
|
following.abc_numerator / following.abc_denominator
|
|
)
|
|
if first_duration < following_duration:
|
|
measures[0] = replace(
|
|
first,
|
|
inferred=True,
|
|
notated_numerator=following.abc_numerator,
|
|
notated_denominator=following.abc_denominator,
|
|
pad_before=True,
|
|
)
|
|
diagnostics.append(
|
|
f"measure 0: padded leading {first.numerator}/{first.denominator} span "
|
|
f"to {following.abc_numerator}/{following.abc_denominator} "
|
|
f"with preceding rest"
|
|
)
|
|
return measures, diagnostics
|
|
|
|
|
|
def _build_grid(beats: Sequence[BeatEvent], measures: Sequence[Measure]):
|
|
interval_denominators = np.zeros(len(beats) - 1, dtype=np.int32)
|
|
for measure in measures:
|
|
interval_denominators[measure.start_beat:measure.end_beat] = measure.denominator
|
|
if np.any(interval_denominators == 0):
|
|
raise BeatGridError("Downbeat spans do not cover every beat interval")
|
|
|
|
subbeat_times = []
|
|
subbeat_denominators = []
|
|
quarter_positions = [0.0]
|
|
current_quarter = 0.0
|
|
for index in range(len(beats) - 1):
|
|
start = beats[index].time
|
|
end = beats[index + 1].time
|
|
denominator = int(interval_denominators[index])
|
|
times = np.linspace(start, end, SUBBEAT_DIVISION + 1)[:-1]
|
|
subbeat_times.extend(float(value) for value in times)
|
|
subbeat_denominators.extend([denominator] * SUBBEAT_DIVISION)
|
|
quarter_step = 4.0 / denominator / SUBBEAT_DIVISION
|
|
for _ in range(SUBBEAT_DIVISION):
|
|
current_quarter += quarter_step
|
|
quarter_positions.append(current_quarter)
|
|
subbeat_times.append(beats[-1].time)
|
|
subbeat_denominators.append(int(interval_denominators[-1]))
|
|
return (
|
|
np.asarray(subbeat_times, dtype=np.float64),
|
|
np.asarray(quarter_positions, dtype=np.float64),
|
|
np.asarray(subbeat_denominators, dtype=np.int32),
|
|
)
|
|
|
|
|
|
def _subbeat_boundaries(subbeat_times: np.ndarray) -> np.ndarray:
|
|
return (subbeat_times[:-1] + subbeat_times[1:]) / 2
|
|
|
|
|
|
def _quantize_time(time: float, subbeat_times: np.ndarray) -> int:
|
|
return int(np.searchsorted(_subbeat_boundaries(subbeat_times), float(time)))
|
|
|
|
|
|
def _fill_intervals(rows, subbeat_times, *, default, dtype):
|
|
result = np.full(len(subbeat_times), default, dtype=dtype)
|
|
for start, end, value in rows:
|
|
start_t = _quantize_time(start, subbeat_times)
|
|
end_t = _quantize_time(end, subbeat_times)
|
|
start_t = max(0, min(start_t, len(result) - 1))
|
|
end_t = max(0, min(end_t, len(result) - 1))
|
|
if end_t <= start_t:
|
|
raise AbcRebuildError(
|
|
f"Interval {start:.6f}-{end:.6f} ({value}) is shorter than the ABC subbeat grid"
|
|
)
|
|
result[start_t:end_t] = value
|
|
if len(result) > 1:
|
|
result[-1] = result[-2]
|
|
return result
|
|
|
|
|
|
def _structure_events(rows, subbeat_times):
|
|
events = []
|
|
for start, _, label in rows:
|
|
t = _quantize_time(start, subbeat_times)
|
|
t = max(0, min(t, len(subbeat_times) - 1))
|
|
events.append((t, label))
|
|
return events
|
|
|
|
|
|
def _classify_melody_tracks(midi: pretty_midi.PrettyMIDI):
|
|
classified = {"Vocal": [], "Ins": []}
|
|
unknown = []
|
|
for instrument in midi.instruments:
|
|
if instrument.is_drum:
|
|
continue
|
|
name = (instrument.name or "").strip().lower()
|
|
if "vocal" in name:
|
|
classified["Vocal"].append(instrument)
|
|
elif "ins" in name or "instrument" in name:
|
|
classified["Ins"].append(instrument)
|
|
elif instrument.notes:
|
|
unknown.append(instrument)
|
|
if unknown:
|
|
if not classified["Vocal"] and not classified["Ins"] and len(unknown) == 1:
|
|
classified["Ins"].extend(unknown)
|
|
else:
|
|
names = [instrument.name or "<unnamed>" for instrument in unknown]
|
|
raise MelodyVoiceError(
|
|
f"Cannot map non-empty melody track(s) {names!r} to fixed Vocal/Ins voices"
|
|
)
|
|
return classified
|
|
|
|
|
|
def _notes_to_arr(notes, subbeat_times, voice_id):
|
|
result = np.zeros(len(subbeat_times), dtype=np.int32)
|
|
boundaries = _subbeat_boundaries(subbeat_times)
|
|
for note in sorted(notes, key=lambda item: (item.start, item.end, item.pitch)):
|
|
start_t = int(np.searchsorted(boundaries, note.start))
|
|
end_t = int(np.searchsorted(boundaries, note.end))
|
|
start_t = max(0, min(start_t, len(result) - 1))
|
|
end_t = max(0, min(end_t, len(result) - 1))
|
|
if end_t <= start_t:
|
|
raise MelodyVoiceError(
|
|
f"{voice_id}: MIDI note pitch={note.pitch} at {note.start:.6f}-{note.end:.6f} "
|
|
"cannot be represented on the decoded subbeat grid"
|
|
)
|
|
if np.any(result[start_t:end_t] != 0):
|
|
raise MelodyVoiceError(
|
|
f"{voice_id}: overlapping quantized melody notes at subbeats {start_t}:{end_t}"
|
|
)
|
|
sustain = note.pitch * 2 + 2
|
|
result[start_t:end_t] = sustain
|
|
result[start_t] = sustain + 1
|
|
return result
|
|
|
|
|
|
def _pitch_class(root: str) -> tuple[int, str, str]:
|
|
match = _ROOT_RE.fullmatch(root)
|
|
if match is None:
|
|
raise ChordSymbolError(f"Invalid pitch spelling {root!r}")
|
|
letter = match.group("letter")
|
|
accidental = match.group("accidental")
|
|
offset = accidental.count("#") - accidental.count("b")
|
|
return (_NATURAL_PITCH_CLASS[letter] + offset) % 12, letter, accidental
|
|
|
|
|
|
def portable_pitch_name(root: str, *, preserve_double: bool = False) -> str:
|
|
pitch_class, _, accidental = _pitch_class(root)
|
|
if preserve_double or len(accidental) <= 1:
|
|
return root
|
|
names = _SHARP_PITCH_NAMES if accidental.startswith("#") else _FLAT_PITCH_NAMES
|
|
return names[pitch_class]
|
|
|
|
|
|
def _bass_degree_to_pitch(root: str, degree_text: str) -> str:
|
|
if _ROOT_RE.fullmatch(degree_text):
|
|
return portable_pitch_name(degree_text, preserve_double=True)
|
|
match = _BASS_DEGREE_RE.fullmatch(degree_text)
|
|
if match is None:
|
|
raise ChordSymbolError(f"Invalid chord bass degree {degree_text!r}")
|
|
root_pc, root_letter, root_accidental = _pitch_class(root)
|
|
degree = int(match.group("degree"))
|
|
degree_accidental = match.group("accidental")
|
|
scale_semitones = (0, 2, 4, 5, 7, 9, 11)
|
|
interval = scale_semitones[(degree - 1) % 7] + 12 * ((degree - 1) // 7)
|
|
interval += degree_accidental.count("#") - degree_accidental.count("b")
|
|
target_pc = (root_pc + interval) % 12
|
|
|
|
target_letter_index = (_LETTERS.index(root_letter) + degree - 1) % 7
|
|
target_letter = _LETTERS[target_letter_index]
|
|
natural_pc = _NATURAL_PITCH_CLASS[target_letter]
|
|
difference = (target_pc - natural_pc + 6) % 12 - 6
|
|
if difference in {-2, -1, 0, 1, 2}:
|
|
accidental = {-2: "bb", -1: "b", 0: "", 1: "#", 2: "##"}[difference]
|
|
return target_letter + accidental
|
|
names = _SHARP_PITCH_NAMES if "#" in (root_accidental + degree_accidental) else _FLAT_PITCH_NAMES
|
|
return names[target_pc]
|
|
|
|
|
|
def chord_symbol_to_abc(chord: str) -> str | None:
|
|
chord = chord.strip()
|
|
if chord in NO_CHORDS:
|
|
return None
|
|
if ":" not in chord:
|
|
raise ChordSymbolError(f"Chord {chord!r} is missing the ':' quality separator")
|
|
root, descriptor = chord.split(":", 1)
|
|
if "/" in descriptor:
|
|
quality, bass_degree = descriptor.split("/", 1)
|
|
else:
|
|
quality, bass_degree = descriptor, None
|
|
if quality not in _QUALITY_TO_ABC:
|
|
raise ChordSymbolError(
|
|
f"Unsupported chord quality {quality!r} in {chord!r}; refusing to rewrite it as major"
|
|
)
|
|
chord_root = portable_pitch_name(root, preserve_double=True)
|
|
text = chord_root + _QUALITY_TO_ABC[quality]
|
|
if bass_degree:
|
|
text += "/" + _bass_degree_to_pitch(root, bass_degree)
|
|
return text
|
|
|
|
|
|
def key_symbol_to_abc(key: str) -> str:
|
|
key = key.strip()
|
|
if ":" in key:
|
|
root, mode = key.split(":", 1)
|
|
if mode not in {"major", "minor"}:
|
|
raise AbcRebuildError(f"Unsupported key mode {mode!r} in {key!r}")
|
|
elif key.endswith("m"):
|
|
root, mode = key[:-1], "minor"
|
|
else:
|
|
root, mode = key, "major"
|
|
root_pc, _, accidental = _pitch_class(root)
|
|
candidate = portable_pitch_name(root) + ("m" if mode == "minor" else "")
|
|
if candidate in _KEY_SIGNATURE_ACCIDENTALS:
|
|
return candidate
|
|
names = _FLAT_PITCH_NAMES if "b" in accidental else _SHARP_PITCH_NAMES
|
|
candidate = names[root_pc] + ("m" if mode == "minor" else "")
|
|
if candidate not in _KEY_SIGNATURE_ACCIDENTALS:
|
|
fallback_names = _SHARP_PITCH_NAMES if names is _FLAT_PITCH_NAMES else _FLAT_PITCH_NAMES
|
|
candidate = fallback_names[root_pc] + ("m" if mode == "minor" else "")
|
|
if candidate not in _KEY_SIGNATURE_ACCIDENTALS:
|
|
raise AbcRebuildError(f"Cannot encode portable ABC key for {key!r}")
|
|
return candidate
|
|
|
|
|
|
def get_key_accidentals(key: str) -> list[int]:
|
|
try:
|
|
count = _KEY_SIGNATURE_ACCIDENTALS[key]
|
|
except KeyError as exc:
|
|
raise AbcRebuildError(f"Unsupported ABC key signature {key!r}") from exc
|
|
accidentals = [0] * 7
|
|
order = "FCGDAEB" if count > 0 else "BEADGCF"
|
|
for letter in order[:abs(count)]:
|
|
accidentals[_LETTERS.index(letter)] = 1 if count > 0 else -1
|
|
return accidentals
|
|
|
|
|
|
def note_to_abc(note: int, key_accidentals: Sequence[int], measure_accidentals: dict) -> str:
|
|
"""Use key-relative spelling and write only bar-state changes.
|
|
|
|
The two target parsers propagate an accidental to the same note letter in
|
|
every octave until the next barline. ``measure_accidentals`` is therefore
|
|
keyed by letter and reset by the caller for every bar (and after an inline
|
|
key change). This preserves pitches across parsers while still omitting
|
|
repeated accidental marks. The key-relative spelling can use double
|
|
accidentals in remote keys; MIDI G is F## in G# minor, for example.
|
|
"""
|
|
|
|
accidental_count = sum(key_accidentals)
|
|
try:
|
|
pitch_name = _KEY_RELATIVE_PITCH_NAMES[accidental_count][note % 12]
|
|
except KeyError as exc:
|
|
raise AbcRebuildError(
|
|
f"Unsupported key signature accidental count {accidental_count}"
|
|
) from exc
|
|
letter = pitch_name[0]
|
|
accidental = pitch_name[1:]
|
|
accidental_number = {"": 0, "#": 1, "##": 2, "b": -1, "bb": -2}[accidental]
|
|
octave = (note - 60) // 12
|
|
# Cb and B# cross the MIDI octave boundary even though their written note
|
|
# letter does not.
|
|
if note % 12 == 11 and accidental_number == -1:
|
|
octave += 1
|
|
elif note % 12 == 0 and accidental_number == 1:
|
|
octave -= 1
|
|
scale_index = _LETTERS.index(letter)
|
|
current_accidental = measure_accidentals.get(
|
|
scale_index,
|
|
key_accidentals[scale_index],
|
|
)
|
|
accidental_text = ""
|
|
if current_accidental != accidental_number:
|
|
measure_accidentals[scale_index] = accidental_number
|
|
accidental_text = {-2: "__", -1: "_", 0: "=", 1: "^", 2: "^^"}[
|
|
accidental_number
|
|
]
|
|
|
|
if octave > 0:
|
|
letter = letter.lower()
|
|
if octave > 1:
|
|
letter += "'" * (octave - 1)
|
|
elif octave < 0:
|
|
letter += "," * abs(octave)
|
|
return accidental_text + letter
|
|
|
|
|
|
def build_rebuilt_abc_score(
|
|
melody_midi_path,
|
|
beats_path,
|
|
chords_path,
|
|
keys_path,
|
|
structures_path,
|
|
*,
|
|
meter_conflict: str = "infer",
|
|
melody_only: bool = False,
|
|
) -> RebuiltAbcScore:
|
|
beats = read_beats(beats_path)
|
|
keys = read_keys(keys_path)
|
|
structures = read_structures(structures_path)
|
|
chords = [] if melody_only else read_chords(chords_path)
|
|
midi = pretty_midi.PrettyMIDI(str(melody_midi_path))
|
|
return _assemble_abc_score(midi, beats, keys, structures, chords,
|
|
meter_conflict=meter_conflict, melody_only=melody_only)
|
|
|
|
|
|
def build_rebuilt_abc_score_from_data(
|
|
melody_midi, beats, chords, keys, structures, *, meter_conflict="infer", melody_only=False,
|
|
) -> RebuiltAbcScore:
|
|
"""Build from MIDI bytes/BytesIO/PrettyMIDI and beat/interval rows.
|
|
|
|
BeatEvent lists are also accepted. The file and memory interfaces share
|
|
interval validation, score construction, serialization and ABC validation.
|
|
"""
|
|
beats = list(beats)
|
|
if beats and isinstance(beats[0], BeatEvent):
|
|
beat_entries = [(b.line_no, [str(b.time), str(b.beat_id), str(b.declared_numerator), str(b.denominator)]) for b in beats]
|
|
else:
|
|
beat_entries = _row_entries(beats, "beats")
|
|
beats = _parse_beats(beat_entries, "beats")
|
|
keys = _parse_keys(_row_entries(keys, "keys"), "keys")
|
|
structures = _parse_structures(_row_entries(structures, "structures"), "structures")
|
|
chords = [] if melody_only else _parse_chords(_row_entries(chords, "chords"), "chords")
|
|
if isinstance(melody_midi, (bytes, bytearray)):
|
|
melody_midi = BytesIO(melody_midi)
|
|
if not isinstance(melody_midi, (BytesIO, pretty_midi.PrettyMIDI)):
|
|
raise TypeError("melody_midi must be MIDI bytes, BytesIO, or PrettyMIDI")
|
|
midi = melody_midi if isinstance(melody_midi, pretty_midi.PrettyMIDI) else pretty_midi.PrettyMIDI(melody_midi)
|
|
return _assemble_abc_score(midi, beats, keys, structures, chords,
|
|
meter_conflict=meter_conflict, melody_only=melody_only)
|
|
|
|
|
|
def _assemble_abc_score(midi, beats, keys, structures, chords, *, meter_conflict, melody_only):
|
|
measures, diagnostics = infer_measures(beats, meter_conflict=meter_conflict)
|
|
subbeat_times, subbeat_quarters, subbeat_denominators = _build_grid(beats, measures)
|
|
classified = _classify_melody_tracks(midi)
|
|
voice_arrs = {}
|
|
for voice_id in VOICE_IDS:
|
|
notes = [
|
|
note
|
|
for instrument in classified[voice_id]
|
|
for note in instrument.notes
|
|
]
|
|
voice_arrs[voice_id] = _notes_to_arr(notes, subbeat_times, voice_id)
|
|
|
|
key_arr = _fill_intervals(keys, subbeat_times, default=keys[0][2], dtype="<U16")
|
|
if melody_only:
|
|
# Do not even read chord labels in melody-only mode. A constant no-chord
|
|
# timeline removes chord-only render boundaries, allowing held notes and
|
|
# rests to be serialized as their original semantic segments.
|
|
chord_arr = np.full(len(subbeat_times), "N", dtype="<U64")
|
|
else:
|
|
chord_arr = _fill_intervals(
|
|
chords,
|
|
subbeat_times,
|
|
default="N",
|
|
dtype="<U64",
|
|
)
|
|
return RebuiltAbcScore(
|
|
beats=list(beats),
|
|
measures=measures,
|
|
subbeat_times=subbeat_times,
|
|
subbeat_quarters=subbeat_quarters,
|
|
subbeat_denominators=subbeat_denominators,
|
|
key_arr=key_arr,
|
|
chord_arr=chord_arr,
|
|
structure_events=_structure_events(structures, subbeat_times),
|
|
voice_arrs=voice_arrs,
|
|
diagnostics=diagnostics,
|
|
)
|
|
|
|
|
|
def abc_unit_denominator(score: RebuiltAbcScore) -> int:
|
|
values = [
|
|
denominator * score.subbeat_div
|
|
for measure in score.measures
|
|
for denominator in (measure.denominator, measure.abc_denominator)
|
|
]
|
|
denominator = math.lcm(*values)
|
|
if denominator > 1024:
|
|
raise AbcRebuildError(f"Required ABC unit length 1/{denominator} is unreasonably small")
|
|
return denominator
|
|
|
|
|
|
def _measure_actual_units(measure: Measure, unit_denominator: int) -> int:
|
|
return measure.numerator * unit_denominator // measure.denominator
|
|
|
|
|
|
def _measure_abc_units(measure: Measure, unit_denominator: int) -> int:
|
|
return measure.abc_numerator * unit_denominator // measure.abc_denominator
|
|
|
|
|
|
def _measure_padding_units(measure: Measure, unit_denominator: int) -> int:
|
|
return (
|
|
_measure_abc_units(measure, unit_denominator)
|
|
- _measure_actual_units(measure, unit_denominator)
|
|
)
|
|
|
|
|
|
def _duration_units(score: RebuiltAbcScore, start_t: int, end_t: int, unit_denominator: int) -> int:
|
|
units = 0
|
|
for denominator in score.subbeat_denominators[start_t:end_t]:
|
|
divisor = int(denominator) * score.subbeat_div
|
|
if unit_denominator % divisor:
|
|
raise AbcRebuildError(
|
|
f"ABC L:1/{unit_denominator} cannot express a 1/{divisor} subbeat exactly"
|
|
)
|
|
units += unit_denominator // divisor
|
|
return units
|
|
|
|
|
|
def estimate_tempo(score: RebuiltAbcScore) -> float:
|
|
seconds = score.subbeat_times[-1] - score.subbeat_times[0]
|
|
quarter_notes = score.subbeat_quarters[-1] - score.subbeat_quarters[0]
|
|
if seconds <= 0 or quarter_notes <= 0:
|
|
raise AbcRebuildError("Cannot estimate tempo from a zero-duration score")
|
|
return float(quarter_notes / seconds * 60.0)
|
|
|
|
|
|
def _continues_pitch(value: int, next_value: int) -> bool:
|
|
if value <= 0:
|
|
return False
|
|
pitch = value // 2 - 1
|
|
return next_value == pitch * 2 + 2
|
|
|
|
|
|
def _same_note_segment(value: int, next_value: int) -> bool:
|
|
if value == 0:
|
|
return next_value == 0
|
|
pitch = value // 2 - 1
|
|
return next_value == pitch * 2 + 2
|
|
|
|
|
|
_SUPPORTED_DURATION_UNITS = frozenset(
|
|
{1, 2, 3, 4, 6, 8, 12, 16, 24, 32, 48}
|
|
)
|
|
|
|
|
|
def _split_duration_units(duration: int) -> list[int]:
|
|
"""Split a duration into values accepted by strict music parsers."""
|
|
|
|
if duration <= 0:
|
|
raise AbcRebuildError(f"Cannot serialize non-positive duration {duration}")
|
|
result = []
|
|
remaining = int(duration)
|
|
while remaining:
|
|
if remaining in _SUPPORTED_DURATION_UNITS:
|
|
result.append(remaining)
|
|
break
|
|
candidates = [
|
|
value
|
|
for value in _SUPPORTED_DURATION_UNITS
|
|
if value < remaining
|
|
]
|
|
if not candidates:
|
|
raise AbcRebuildError(
|
|
f"Duration {duration} cannot be split into representable ABC values"
|
|
)
|
|
chunk = max(candidates)
|
|
result.append(chunk)
|
|
remaining -= chunk
|
|
return result
|
|
|
|
|
|
def _duration_text(duration: int) -> str:
|
|
return "" if duration == 1 else str(duration)
|
|
|
|
|
|
def _render_duration_tokens(
|
|
prefix: str,
|
|
note_text: str,
|
|
duration: int,
|
|
*,
|
|
tie_out: bool,
|
|
) -> list[str]:
|
|
chunks = _split_duration_units(duration)
|
|
tokens = []
|
|
for index, chunk in enumerate(chunks):
|
|
continues = note_text != "z" and (
|
|
index + 1 < len(chunks) or tie_out
|
|
)
|
|
tokens.append(
|
|
(prefix if index == 0 else "")
|
|
+ note_text
|
|
+ _duration_text(chunk)
|
|
+ ("-" if continues else "")
|
|
)
|
|
return tokens
|
|
|
|
|
|
def _render_voice_measure(
|
|
score: RebuiltAbcScore,
|
|
voice_id: str,
|
|
measure: Measure,
|
|
unit_denominator: int,
|
|
) -> str:
|
|
voice = score.voice_arrs[voice_id]
|
|
show_chords = voice_id == "Vocal"
|
|
measure_accidentals = {}
|
|
current_key = str(score.key_arr[measure.start_t])
|
|
key_accidentals = get_key_accidentals(current_key)
|
|
parts = []
|
|
padding = _measure_padding_units(measure, unit_denominator)
|
|
if padding < 0:
|
|
raise AbcRebuildError(
|
|
f"Measure {measure.index}: notated meter is shorter than its decoded span"
|
|
)
|
|
leading_padding = padding if measure.pad_before else 0
|
|
trailing_padding = 0 if measure.pad_before else padding
|
|
t = measure.start_t
|
|
while t < measure.end_t:
|
|
change_points = [measure.end_t]
|
|
for probe in range(t + 1, measure.end_t):
|
|
if not _same_note_segment(int(voice[t]), int(voice[probe])):
|
|
change_points.append(probe)
|
|
break
|
|
for probe in range(t + 1, measure.end_t):
|
|
if score.key_arr[probe] != score.key_arr[probe - 1]:
|
|
change_points.append(probe)
|
|
break
|
|
if show_chords:
|
|
for probe in range(t + 1, measure.end_t):
|
|
if score.chord_arr[probe] != score.chord_arr[probe - 1]:
|
|
change_points.append(probe)
|
|
break
|
|
next_t = min(change_points)
|
|
|
|
prefix = ""
|
|
key = str(score.key_arr[t])
|
|
if t > measure.start_t and key != current_key:
|
|
current_key = key
|
|
key_accidentals = get_key_accidentals(current_key)
|
|
measure_accidentals = {}
|
|
prefix += f"[K:{current_key}]"
|
|
|
|
if show_chords and (t == measure.start_t or score.chord_arr[t] != score.chord_arr[t - 1]):
|
|
chord = str(score.chord_arr[t])
|
|
chord_text = chord_symbol_to_abc(chord)
|
|
if chord_text is not None:
|
|
prefix += f'"{chord_text}"'
|
|
|
|
value = int(voice[t])
|
|
if value == 0:
|
|
note_text = "z"
|
|
else:
|
|
note_text = note_to_abc(value // 2 - 1, key_accidentals, measure_accidentals)
|
|
duration = _duration_units(score, t, next_t, unit_denominator)
|
|
if t == measure.start_t and leading_padding:
|
|
if value == 0 and not prefix:
|
|
duration += leading_padding
|
|
else:
|
|
parts.extend(
|
|
_render_duration_tokens(
|
|
"",
|
|
"z",
|
|
leading_padding,
|
|
tie_out=False,
|
|
)
|
|
)
|
|
leading_padding = 0
|
|
if value == 0 and next_t == measure.end_t and trailing_padding:
|
|
duration += trailing_padding
|
|
trailing_padding = 0
|
|
if duration <= 0:
|
|
raise AbcRebuildError(f"Non-positive ABC duration at subbeats {t}:{next_t}")
|
|
tie_out = (
|
|
value > 0
|
|
and next_t < len(voice)
|
|
and _continues_pitch(value, int(voice[next_t]))
|
|
)
|
|
parts.extend(
|
|
_render_duration_tokens(
|
|
prefix,
|
|
note_text,
|
|
duration,
|
|
tie_out=tie_out,
|
|
)
|
|
)
|
|
t = next_t
|
|
if leading_padding:
|
|
raise AbcRebuildError(
|
|
f"Measure {measure.index}: leading rest padding was not serialized"
|
|
)
|
|
if trailing_padding:
|
|
parts.extend(
|
|
_render_duration_tokens(
|
|
"",
|
|
"z",
|
|
trailing_padding,
|
|
tie_out=False,
|
|
)
|
|
)
|
|
return "".join(parts)
|
|
|
|
|
|
def _is_compressible_full_rest(rendered_measure: str) -> bool:
|
|
"""Whether a rendered measure can be losslessly replaced by ABC ``Z``."""
|
|
|
|
cursor = 0
|
|
saw_note = False
|
|
for match in _MUSIC_ELEMENT_RE.finditer(rendered_measure):
|
|
if rendered_measure[cursor:match.start()]:
|
|
return False
|
|
cursor = match.end()
|
|
if match.group("quoted") is not None or match.group("key") is not None:
|
|
return False
|
|
saw_note = True
|
|
if match.group("note") != "z" or match.group("tie"):
|
|
return False
|
|
return saw_note and cursor == len(rendered_measure)
|
|
|
|
|
|
def _render_voice_group(
|
|
score: RebuiltAbcScore,
|
|
voice_id: str,
|
|
measures: list[Measure],
|
|
unit_denominator: int,
|
|
) -> str:
|
|
rendered = [
|
|
_render_voice_measure(
|
|
score,
|
|
voice_id,
|
|
measure,
|
|
unit_denominator,
|
|
)
|
|
for measure in measures
|
|
]
|
|
parts = []
|
|
index = 0
|
|
while index < len(rendered):
|
|
if not _is_compressible_full_rest(rendered[index]):
|
|
parts.append(rendered[index] + "|")
|
|
index += 1
|
|
continue
|
|
end = index + 1
|
|
while (
|
|
end < len(rendered)
|
|
and _is_compressible_full_rest(rendered[end])
|
|
):
|
|
end += 1
|
|
count = end - index
|
|
parts.append("Z" + (str(count) if count > 1 else "") + "|")
|
|
index = end
|
|
return "".join(parts)
|
|
|
|
|
|
def _sanitize_structure_label(value: str) -> str:
|
|
return " ".join(str(value).split())
|
|
|
|
|
|
def _measure_groups(score: RebuiltAbcScore) -> list[MeasureGroup]:
|
|
first_measure = score.measures[0]
|
|
active_meter = (
|
|
first_measure.abc_numerator,
|
|
first_measure.abc_denominator,
|
|
)
|
|
active_key = str(score.key_arr[first_measure.start_t])
|
|
active_structure = ""
|
|
groups: list[MeasureGroup] = []
|
|
|
|
for measure in score.measures:
|
|
meter = (measure.abc_numerator, measure.abc_denominator)
|
|
key = str(score.key_arr[measure.start_t])
|
|
meter_changed = meter != active_meter
|
|
key_changed = key != active_key
|
|
new_structure_labels = []
|
|
for t, label in score.structure_events:
|
|
if not measure.start_t <= t < measure.end_t:
|
|
continue
|
|
clean_label = _sanitize_structure_label(label)
|
|
if clean_label and clean_label != active_structure:
|
|
new_structure_labels.append(clean_label)
|
|
active_structure = clean_label
|
|
|
|
start_group = (
|
|
not groups
|
|
or len(groups[-1].measures) >= 4
|
|
or meter_changed
|
|
or key_changed
|
|
or bool(new_structure_labels)
|
|
)
|
|
if start_group:
|
|
groups.append(
|
|
MeasureGroup(
|
|
measures=[measure],
|
|
structure_labels=new_structure_labels,
|
|
meter_changed=meter_changed,
|
|
key_changed=key_changed,
|
|
)
|
|
)
|
|
else:
|
|
groups[-1].measures.append(measure)
|
|
|
|
active_meter = meter
|
|
active_key = str(score.key_arr[measure.end_t - 1])
|
|
return groups
|
|
|
|
|
|
def score_to_abc(score: RebuiltAbcScore) -> str:
|
|
unit_denominator = abc_unit_denominator(score)
|
|
first_measure = score.measures[0]
|
|
first_key = str(score.key_arr[first_measure.start_t])
|
|
lines = [
|
|
"X:1",
|
|
"T:",
|
|
f"M:{first_measure.abc_numerator}/{first_measure.abc_denominator}",
|
|
f"L:1/{unit_denominator}",
|
|
f"Q:1/4={int(round(estimate_tempo(score)))}",
|
|
'V: Vocal clef=treble name="Vocal Melody" snm="Vocal"',
|
|
'V: Ins clef=treble name="Ins Melody" snm="Inst."',
|
|
f"K:{first_key}",
|
|
]
|
|
for group in _measure_groups(score):
|
|
lines.extend(f"% {label}" for label in group.structure_labels)
|
|
first_group_measure = group.measures[0]
|
|
for voice_id in VOICE_IDS:
|
|
lines.append(f"V: {voice_id}")
|
|
if group.meter_changed:
|
|
lines.append(
|
|
f"M:{first_group_measure.abc_numerator}/"
|
|
f"{first_group_measure.abc_denominator}"
|
|
)
|
|
if group.key_changed:
|
|
lines.append(
|
|
f"K:{score.key_arr[first_group_measure.start_t]}"
|
|
)
|
|
lines.append(
|
|
_render_voice_group(
|
|
score,
|
|
voice_id,
|
|
group.measures,
|
|
unit_denominator,
|
|
)
|
|
)
|
|
text = "\n".join(lines) + "\n"
|
|
validate_serialized_abc(text, score)
|
|
return text
|
|
|
|
|
|
_MUSIC_ELEMENT_RE = re.compile(
|
|
r'"(?P<quoted>[^"]*)"'
|
|
r"|\[K:(?P<key>[^\]]+)\]"
|
|
r"|(?P<note>[_=^]*[A-Ga-gz][,']*)(?P<duration>\d*)(?P<tie>-?)"
|
|
)
|
|
|
|
|
|
def _parse_music_measure(line: str, expected: int, context: str):
|
|
body = line
|
|
if body == "Z":
|
|
return [], []
|
|
position = 0
|
|
cursor = 0
|
|
quoted_events = []
|
|
key_events = []
|
|
for match in _MUSIC_ELEMENT_RE.finditer(body):
|
|
gap = body[cursor:match.start()]
|
|
if gap.strip():
|
|
raise AbcRebuildError(f"{context}: unsupported serialized ABC tokens {gap!r}")
|
|
cursor = match.end()
|
|
if match.group("quoted") is not None:
|
|
quoted_events.append((position, match.group("quoted")))
|
|
continue
|
|
if match.group("key") is not None:
|
|
key_events.append((position, match.group("key")))
|
|
continue
|
|
note = match.group("note")
|
|
tie = match.group("tie")
|
|
if tie and note == "z":
|
|
raise AbcRebuildError(f"{context}: a rest cannot be tied")
|
|
duration_text = match.group("duration")
|
|
duration = int(duration_text) if duration_text else 1
|
|
if duration not in _SUPPORTED_DURATION_UNITS:
|
|
raise AbcRebuildError(
|
|
f"{context}: duration {duration} is not parser-representable"
|
|
)
|
|
position += duration
|
|
if body[cursor:].strip():
|
|
raise AbcRebuildError(
|
|
f"{context}: unsupported serialized ABC tokens {body[cursor:]!r}"
|
|
)
|
|
if position != expected:
|
|
raise AbcRebuildError(
|
|
f"{context}: duration {position} does not match meter duration {expected}"
|
|
)
|
|
if re.search(r"(^|[\s|])-[_=^A-Ga-g]", line):
|
|
raise AbcRebuildError(f"{context}: tie is written before its second note")
|
|
return quoted_events, key_events
|
|
|
|
|
|
def _expected_measure_chords(
|
|
score: RebuiltAbcScore,
|
|
measure: Measure,
|
|
unit_denominator: int,
|
|
) -> list[tuple[int, str]]:
|
|
expected = []
|
|
leading_padding = (
|
|
_measure_padding_units(measure, unit_denominator)
|
|
if measure.pad_before
|
|
else 0
|
|
)
|
|
for t in range(measure.start_t, measure.end_t):
|
|
if t != measure.start_t and score.chord_arr[t] == score.chord_arr[t - 1]:
|
|
continue
|
|
position = leading_padding + _duration_units(
|
|
score,
|
|
measure.start_t,
|
|
t,
|
|
unit_denominator,
|
|
)
|
|
chord = str(score.chord_arr[t])
|
|
text = chord_symbol_to_abc(chord)
|
|
if text is not None:
|
|
expected.append((position, text))
|
|
return expected
|
|
|
|
|
|
def _expected_measure_keys(
|
|
score: RebuiltAbcScore,
|
|
measure: Measure,
|
|
unit_denominator: int,
|
|
) -> list[tuple[int, str]]:
|
|
leading_padding = (
|
|
_measure_padding_units(measure, unit_denominator)
|
|
if measure.pad_before
|
|
else 0
|
|
)
|
|
return [
|
|
(
|
|
leading_padding
|
|
+ _duration_units(score, measure.start_t, t, unit_denominator),
|
|
str(score.key_arr[t]),
|
|
)
|
|
for t in range(measure.start_t + 1, measure.end_t)
|
|
if score.key_arr[t] != score.key_arr[t - 1]
|
|
]
|
|
|
|
|
|
def _parse_voice_group(lines, cursor, voice_id, group_index):
|
|
expected_voice_field = f"V: {voice_id}"
|
|
if cursor >= len(lines) or lines[cursor] != expected_voice_field:
|
|
observed = lines[cursor] if cursor < len(lines) else "<end>"
|
|
raise AbcRebuildError(
|
|
f"Group {group_index}: expected {expected_voice_field}, got {observed!r}"
|
|
)
|
|
cursor += 1
|
|
fields = {}
|
|
while cursor < len(lines) and (
|
|
lines[cursor].startswith("M:")
|
|
or lines[cursor].startswith("K:")
|
|
):
|
|
name, value = lines[cursor].split(":", 1)
|
|
if name in fields:
|
|
raise AbcRebuildError(
|
|
f"Group {group_index} {voice_id}: repeated {name}: field"
|
|
)
|
|
fields[name] = value
|
|
cursor += 1
|
|
if cursor >= len(lines):
|
|
raise AbcRebuildError(
|
|
f"Group {group_index} {voice_id}: missing music line"
|
|
)
|
|
music_line = lines[cursor]
|
|
if music_line.startswith(("V:", "M:", "K:", "%")):
|
|
raise AbcRebuildError(
|
|
f"Group {group_index} {voice_id}: invalid music line {music_line!r}"
|
|
)
|
|
cursor += 1
|
|
split_bars = music_line.split("|")
|
|
if not split_bars or split_bars[-1].strip():
|
|
raise AbcRebuildError(
|
|
f"Group {group_index} {voice_id}: music line must end with a barline"
|
|
)
|
|
serialized_bars = [bar.strip() for bar in split_bars[:-1]]
|
|
if any(not bar for bar in serialized_bars):
|
|
raise AbcRebuildError(
|
|
f"Group {group_index} {voice_id}: empty serialized measure"
|
|
)
|
|
bars = []
|
|
for bar in serialized_bars:
|
|
match = re.fullmatch(r"Z(?P<count>[1-4])?", bar)
|
|
if match is None:
|
|
bars.append(bar)
|
|
continue
|
|
if match.group("count") == "1":
|
|
raise AbcRebuildError(
|
|
f"Group {group_index} {voice_id}: Z1 must be written as Z"
|
|
)
|
|
bars.extend(["Z"] * int(match.group("count") or "1"))
|
|
if not 1 <= len(bars) <= 4:
|
|
raise AbcRebuildError(
|
|
f"Group {group_index} {voice_id}: expected 1-4 semantic measures"
|
|
)
|
|
return cursor, fields, bars
|
|
|
|
|
|
def validate_serialized_abc(text: str, score: RebuiltAbcScore) -> None:
|
|
"""Validate invariants that must hold before any ABC is written."""
|
|
|
|
lines = text.splitlines()
|
|
if not lines or lines[0] != "X:1":
|
|
raise AbcRebuildError("ABC must start with X:1")
|
|
if len(lines) < 2 or lines[1] != "T:":
|
|
raise AbcRebuildError("ABC title must be fixed as empty T:")
|
|
if any(line.startswith(("%abc-", "I:abc-creator")) for line in lines):
|
|
raise AbcRebuildError("ABC must not contain version or creator metadata")
|
|
if "%%MIDI gchordoff" in text:
|
|
raise AbcRebuildError("ABC must not contain %%MIDI gchordoff")
|
|
if "% ss2" in text:
|
|
raise AbcRebuildError("ABC must not contain % ss2 metadata")
|
|
header_voice_ids = [
|
|
match.group(1)
|
|
for line in lines
|
|
if (match := re.match(r"^V: (Vocal|Ins) ", line))
|
|
]
|
|
if header_voice_ids != list(VOICE_IDS):
|
|
raise AbcRebuildError(f"Expected fixed Vocal/Ins voice definitions, got {header_voice_ids!r}")
|
|
|
|
header_key_index = next(
|
|
(
|
|
index
|
|
for index, line in enumerate(lines)
|
|
if index > 0
|
|
and line.startswith("K:")
|
|
and any(
|
|
header_index < index
|
|
for header_index, header_line in enumerate(lines)
|
|
if header_line.startswith("V: Ins ")
|
|
)
|
|
),
|
|
None,
|
|
)
|
|
if header_key_index is None:
|
|
raise AbcRebuildError("ABC header K: field is missing")
|
|
first_measure = score.measures[0]
|
|
expected_header_meter = (
|
|
f"M:{first_measure.abc_numerator}/{first_measure.abc_denominator}"
|
|
)
|
|
header_meters = [
|
|
line
|
|
for line in lines[:header_key_index + 1]
|
|
if line.startswith("M:")
|
|
]
|
|
if header_meters != [expected_header_meter]:
|
|
raise AbcRebuildError(
|
|
f"ABC header meters {header_meters!r} "
|
|
f"!= {[expected_header_meter]!r}"
|
|
)
|
|
expected_header_key = f"K:{score.key_arr[first_measure.start_t]}"
|
|
header_keys = [
|
|
line
|
|
for line in lines[:header_key_index + 1]
|
|
if line.startswith("K:")
|
|
]
|
|
if header_keys != [expected_header_key]:
|
|
raise AbcRebuildError(
|
|
f"ABC header keys {header_keys!r} "
|
|
f"!= {[expected_header_key]!r}"
|
|
)
|
|
|
|
unit_denominator = abc_unit_denominator(score)
|
|
expected_groups = _measure_groups(score)
|
|
cursor = header_key_index + 1
|
|
for group_index, group in enumerate(expected_groups):
|
|
structure_labels = []
|
|
while cursor < len(lines) and lines[cursor].startswith("% "):
|
|
structure_labels.append(lines[cursor][2:].strip())
|
|
cursor += 1
|
|
if structure_labels != group.structure_labels:
|
|
raise AbcRebuildError(
|
|
f"Group {group_index}: structure labels "
|
|
f"{structure_labels!r} != {group.structure_labels!r}"
|
|
)
|
|
|
|
cursor, vocal_fields, vocal_bars = _parse_voice_group(
|
|
lines,
|
|
cursor,
|
|
"Vocal",
|
|
group_index,
|
|
)
|
|
cursor, ins_fields, ins_bars = _parse_voice_group(
|
|
lines,
|
|
cursor,
|
|
"Ins",
|
|
group_index,
|
|
)
|
|
if vocal_fields != ins_fields:
|
|
raise AbcRebuildError(
|
|
f"Group {group_index}: meter/key changes must be scoped to both voices"
|
|
)
|
|
first_measure = group.measures[0]
|
|
expected_fields = {}
|
|
if group.meter_changed:
|
|
expected_fields["M"] = (
|
|
f"{first_measure.abc_numerator}/{first_measure.abc_denominator}"
|
|
)
|
|
if group.key_changed:
|
|
expected_fields["K"] = str(
|
|
score.key_arr[first_measure.start_t]
|
|
)
|
|
if vocal_fields != expected_fields:
|
|
raise AbcRebuildError(
|
|
f"Group {group_index}: fields {vocal_fields!r} "
|
|
f"!= required changes {expected_fields!r}"
|
|
)
|
|
if (
|
|
len(vocal_bars) != len(group.measures)
|
|
or len(ins_bars) != len(group.measures)
|
|
):
|
|
raise AbcRebuildError(
|
|
f"Group {group_index}: both voices must contain "
|
|
f"{len(group.measures)} measures"
|
|
)
|
|
|
|
for bar_index, measure in enumerate(group.measures):
|
|
expected_duration = (
|
|
measure.abc_numerator
|
|
* unit_denominator
|
|
// measure.abc_denominator
|
|
)
|
|
for voice_id, bars in (
|
|
("Vocal", vocal_bars),
|
|
("Ins", ins_bars),
|
|
):
|
|
quoted_events, key_events = _parse_music_measure(
|
|
bars[bar_index],
|
|
expected_duration,
|
|
f"measure {measure.index} {voice_id}",
|
|
)
|
|
expected_keys = _expected_measure_keys(
|
|
score,
|
|
measure,
|
|
unit_denominator,
|
|
)
|
|
if key_events != expected_keys:
|
|
raise AbcRebuildError(
|
|
f"Measure {measure.index} {voice_id}: inline keys "
|
|
f"{key_events!r} != {expected_keys!r}"
|
|
)
|
|
if voice_id == "Vocal":
|
|
expected_chords = _expected_measure_chords(
|
|
score,
|
|
measure,
|
|
unit_denominator,
|
|
)
|
|
if quoted_events != expected_chords:
|
|
raise AbcRebuildError(
|
|
f"Measure {measure.index}: chord symbols "
|
|
f"{quoted_events!r} != {expected_chords!r}"
|
|
)
|
|
elif quoted_events:
|
|
raise AbcRebuildError(
|
|
f"Measure {measure.index}: chords must only be in Vocal"
|
|
)
|
|
if cursor != len(lines):
|
|
raise AbcRebuildError(
|
|
f"Unexpected trailing ABC body lines: {lines[cursor:cursor + 5]!r}"
|
|
)
|
|
|
|
|
|
def abc_paths_from_melody(
|
|
melody_midi_path,
|
|
output_path=None,
|
|
*,
|
|
melody_only=False,
|
|
):
|
|
melody = Path(melody_midi_path)
|
|
if melody.name.endswith("_raw_full_melody.mid"):
|
|
raise AbcRebuildError(f"Raw melody MIDI is not a valid ABC input: {melody}")
|
|
if not melody.name.endswith("_melody.mid"):
|
|
raise AbcRebuildError(f"Expected an exact *_melody.mid input, got {melody}")
|
|
stem = melody.name[: -len("_melody.mid")]
|
|
prefix = melody.with_name(stem)
|
|
default_suffix = "_melody_only.abc" if melody_only else "_full.abc"
|
|
return {
|
|
"melody_midi": melody,
|
|
"beats": Path(str(prefix) + "_beats.txt"),
|
|
"chords": Path(str(prefix) + "_chords.txt"),
|
|
"keys": Path(str(prefix) + "_keys.txt"),
|
|
"structures": Path(str(prefix) + "_structures.txt"),
|
|
"output": (
|
|
Path(output_path)
|
|
if output_path is not None
|
|
else Path(str(prefix) + default_suffix)
|
|
),
|
|
}
|
|
|
|
|
|
def preflight_exports(
|
|
melody_midi_path,
|
|
output_path=None,
|
|
*,
|
|
melody_only=False,
|
|
):
|
|
paths = abc_paths_from_melody(
|
|
melody_midi_path,
|
|
output_path=output_path,
|
|
melody_only=melody_only,
|
|
)
|
|
required = {"melody_midi", "beats", "keys", "structures"}
|
|
if not melody_only:
|
|
required.add("chords")
|
|
missing = [
|
|
str(path)
|
|
for name, path in paths.items()
|
|
if name in required and not path.is_file()
|
|
]
|
|
if missing:
|
|
raise FileNotFoundError(
|
|
f"{paths['melody_midi']}: missing required companion file(s): {', '.join(missing)}"
|
|
)
|
|
return paths
|
|
|
|
|
|
def generate_abc_from_exports(
|
|
melody_midi_path,
|
|
*,
|
|
output_path=None,
|
|
meter_conflict="infer",
|
|
melody_only=False,
|
|
):
|
|
paths = preflight_exports(
|
|
melody_midi_path,
|
|
output_path=output_path,
|
|
melody_only=melody_only,
|
|
)
|
|
score = build_rebuilt_abc_score(
|
|
paths["melody_midi"],
|
|
paths["beats"],
|
|
paths["chords"],
|
|
paths["keys"],
|
|
paths["structures"],
|
|
meter_conflict=meter_conflict,
|
|
melody_only=melody_only,
|
|
)
|
|
return score_to_abc(score), score, paths
|
|
|
|
|
|
def generate_abc_from_data(melody_midi, beats, chords, keys, structures, *,
|
|
meter_conflict="infer", melody_only=False):
|
|
"""Return validated ABC text and its score without filesystem access."""
|
|
score = build_rebuilt_abc_score_from_data(
|
|
melody_midi, beats, chords, keys, structures,
|
|
meter_conflict=meter_conflict, melody_only=melody_only,
|
|
)
|
|
return score_to_abc(score), score
|