"""MIDI file import — list tracks and convert tracks to sloppak payloads. Three parallel flows live here: - **Keys path** (`list_midi_tracks` + `convert_midi_track_to_keys_wire`): filters channel-9 out and emits a standard guitar-style arrangement that the piano plugin decodes via `midi = string * 24 + fret`. - **Drums path** (`list_drum_tracks` + `convert_drum_track_from_midi`): keeps channel-9 only and emits the `drum_tab.json` shape documented in `docs/sloppak-spec.md` §5.3, ready to drop alongside the sloppak manifest's `drum_tab:` key. - **Lyrics path** (`extract_midi_lyrics`): reads SMF Lyric (0x05) meta events (with a Text-event fallback on vocal-ish tracks, covering karaoke `.kar` files) and emits the `lyrics.json` / `vocal_pitch.json` sidecar payloads documented in feedpak-spec §7.1 / §7.2, ready to drop alongside the manifest's `lyrics:` / `lyrics_source:` / `vocal_pitch:` keys. The editor's track picker uses the first two for the +Drums and +Keys modals. """ from __future__ import annotations import math import re from bisect import bisect_right from collections import deque from typing import Callable import mido import drums as drums_mod # General MIDI piano-family programs (0-7) plus chromatic percussion + organ. # Used to flag obvious keyboard tracks for the picker UI. _KEY_PROGRAMS = set(range(0, 24)) _KEYBOARD_NAME_HINTS = ( "piano", "keys", "keyboard", "synth", "organ", "rhodes", "harpsichord", "clavinet", "wurlitzer", "ep ", "epiano", ) def list_midi_tracks(midi_path: str) -> list[dict]: """Return a list of track descriptors suitable for the picker UI. Format-0 MIDI files store every channel in a single track; if we just enumerated `midi.tracks` we'd produce one picker entry that merged every part into a single Keys arrangement. For format-0 only, split that single track into one virtual entry per non-drum channel so the user can isolate the piano part. Type-1 (parallel tracks) and type-2 (independent sequences) keep their one-entry-per-track shape: their tracks already represent the parts the author intended, and a track that uses e.g. LH/RH on separate channels would otherwise lose half its notes when the user picked just one of the split entries with no way to recover the merged form. Drum (channel-9) channels are dropped from the listing here — the keys-import converter unconditionally skips channel-9 events, so a drums entry would yield an empty arrangement. Use `list_drum_tracks` + `convert_drum_track_from_midi` for the MIDI drum-import flow. Each item: {index, name, instrument, notes, channel, is_piano, is_drums, channel_filter}. For split entries `channel_filter` is set; for unsplit entries it's None. """ midi = mido.MidiFile(midi_path) tracks: list[dict] = [] midi_type = getattr(midi, "type", 1) # Only format-0 collapses every part into one track and therefore # benefits from per-channel splitting. Type-1/2 tracks already # represent author-defined parts. split_format = (midi_type == 0) for i, track in enumerate(midi.tracks): name = "" # Per-channel stats, populated by walking the track once. per_channel: dict[int, dict] = {} for msg in track: if msg.type == "track_name" and not name: name = msg.name or "" elif msg.type == "program_change": ch = int(getattr(msg, "channel", -1)) slot = per_channel.setdefault(ch, {"program": -1, "notes": 0}) if slot["program"] < 0: slot["program"] = int(msg.program) elif msg.type == "note_on" and int(getattr(msg, "velocity", 0)) > 0: ch = int(getattr(msg, "channel", -1)) slot = per_channel.setdefault(ch, {"program": -1, "notes": 0}) slot["notes"] += 1 # Drop channels that never produced a note (tempo/meta-only entries # would just clutter the picker) AND drop drum channels — the # keys-import converter skips channel-9 events unconditionally, # so a drums entry would always yield an empty arrangement. active_channels = sorted( ch for ch, info in per_channel.items() if info["notes"] > 0 and ch != 9 ) if not active_channels: # Track with no melodic notes (silent or drums-only). Skip. continue # Format-0 with multiple non-drum channels is the only case where # we split. Type-1/2 keep one-entry-per-track so the user can # always import the whole part. split = split_format and len(active_channels) > 1 if split: iter_channels = active_channels else: # One merged entry; channel comes from the first active one # for display purposes (and in case the converter ever needs # a hint, though channel_filter=None means "merge all"). iter_channels = [active_channels[0]] for ch in iter_channels: info = per_channel[ch] program = info["program"] note_count = ( info["notes"] if split else sum(per_channel[c]["notes"] for c in active_channels) ) if split: channel_label = f"Ch{ch + 1}" base = name or f"Track {i}" entry_name = f"{base} — {channel_label}" else: entry_name = name or f"Track {i}" # Classify on the per-channel program first. The track-level # name hint is a tiebreaker only when no program_change was # seen for this channel — otherwise a track named "Piano" # that hosts bass on ch2 would wrongly flag ch2 as piano in # the format-0 split case. For non-split tracks the name # hint still carries weight (single-channel tracks usually # share track name + program intent). if program in _KEY_PROGRAMS: is_piano = True elif program < 0 and not split: # Program unknown for this single-channel track — fall # back to the track-name heuristic. name_lower = entry_name.lower() is_piano = any(hint in name_lower for hint in _KEYBOARD_NAME_HINTS) else: is_piano = False tracks.append({ "index": i, # When set, the converter filters the track's events to this # channel only. None means "use every non-drum channel". "channel_filter": ch if split else None, "name": entry_name, "instrument": program, "notes": note_count, "channel": ch, "is_piano": bool(is_piano), # `is_drums` is always False on emitted entries because we # filter channel 9 above. Kept for shape compatibility # with the GP picker entries the frontend also reads. "is_drums": False, "strings": 0, "is_percussion": False, }) return tracks def convert_midi_track_to_keys_wire( midi_path: str, track_index: int, audio_offset: float = 0.0, name: str = "Keys", channel_filter: int | None = None, ) -> dict: """Convert a single MIDI track into a sloppak-format keys arrangement. Encodes each MIDI note as the piano plugin expects: string = pitch // 24, fret = pitch % 24 (so noteToMidi(s, f) = s * 24 + f recovers the pitch). Returns a wire-format arrangement dict ready to be written to arrangements/.json. audio_offset (seconds) is added to every note's start time. Useful as a coarse pre-sync handle; finer alignment happens in the editor. channel_filter (optional): when set, only events on this channel are processed. Used by the picker to isolate one channel out of a format-0 track that mixes multiple instruments. CC64 (sustain pedal) is honored: when a key is released while the pedal is held, the note's end time is extended to the pedal-up event on the same channel. Pedal-down/up transitions are tracked per channel. """ midi = mido.MidiFile(midi_path) if track_index < 0 or track_index >= len(midi.tracks): raise ValueError(f"track_index {track_index} out of range") # Build a tempo map. The right scope depends on the SMF format: # - type 0 (single track holding everything): the lone track is also # the source of tempo events. Walking it (and only it) is correct. # - type 1 (parallel tracks, shared timeline): tempo events live on # the conductor track (usually track 0) but the spec allows them # anywhere. Merge across all tracks so we don't miss any. # - type 2 (independent sequential tracks, each its own timeline): # a foreign track's tempo events do NOT apply to the chosen # track. Merging would mis-time the notes — restrict the tempo # scan to the selected track only. # ``ticks_per_beat`` is 0 for a malformed header and NEGATIVE for SMPTE # division (mido returns the signed short as-is). Both feed the two # divisions below (tempo-table build + tick_to_seconds), so guard here: # 0 would raise ZeroDivisionError and a negative value would yield # negative/garbage times. Use ``> 0`` (not ``or``) so the negative SMPTE # case also falls back to the SMF default. ticks_per_beat = midi.ticks_per_beat if midi.ticks_per_beat > 0 else 480 raw_events: list[tuple[int, int]] = [(0, 500000)] # default 120 BPM midi_type = getattr(midi, "type", 1) tempo_source = ( [midi.tracks[track_index]] if midi_type == 2 else midi.tracks ) for track in tempo_source: abs_tick = 0 for msg in track: abs_tick += msg.time if msg.type == "set_tempo": raw_events.append((abs_tick, int(msg.tempo))) raw_events.sort(key=lambda e: e[0]) # Deduplicate at same tick (keep the last one written). deduped: list[tuple[int, int]] = [] for ev in raw_events: if deduped and deduped[-1][0] == ev[0]: deduped[-1] = ev else: deduped.append(ev) # Precompute (tick, seconds_at_tick, microseconds_per_beat). seconds_at_tick # is the cumulative time up to that tempo-change event. tempo_table: list[tuple[int, float, int]] = [] cum_seconds = 0.0 prev_tick = 0 prev_tempo = deduped[0][1] for ev_tick, ev_tempo in deduped: cum_seconds += (ev_tick - prev_tick) * (prev_tempo / 1_000_000.0) / ticks_per_beat tempo_table.append((ev_tick, cum_seconds, ev_tempo)) prev_tick = ev_tick prev_tempo = ev_tempo tempo_ticks = [row[0] for row in tempo_table] def tick_to_seconds(tick: int) -> float: """O(log N) tempo-aware tick→seconds via cumulative table + bisect.""" i = bisect_right(tempo_ticks, tick) - 1 if i < 0: i = 0 base_tick, base_seconds, tempo = tempo_table[i] return base_seconds + (tick - base_tick) * (tempo / 1_000_000.0) / ticks_per_beat # Walk the requested track, collect note_on/note_off pairs. To handle # rapid retriggers (same pitch starting again before the previous # note_off), keep a stack of start ticks per (channel, pitch). # # Sustain pedal (CC64): when value >= 64, the channel is "pedal down" # and key-release events don't truncate the note — they move the # pending start onto `pedal_pending`, where it waits for the pedal-up # transition. Pedal-up finalises every pending note on that channel # using the pedal-up tick as the end. track = midi.tracks[track_index] abs_tick = 0 active: dict[tuple[int, int], deque[int]] = {} pedal_pending: dict[int, list[tuple[int, int]]] = {} # ch -> [(pitch, start_tick)] pedal_down: dict[int, bool] = {} notes_out: list[dict] = [] def _emit(pitch: int, start_tick: int, end_tick: int) -> None: t = tick_to_seconds(start_tick) + float(audio_offset) end = tick_to_seconds(end_tick) + float(audio_offset) notes_out.append({ "t": round(t, 3), "s": int(pitch // 24), "f": int(pitch % 24), "sus": round(max(0.0, end - t), 3), "sl": -1, "slu": -1, "bn": 0, "ho": False, "po": False, "hm": False, "hp": False, "pm": False, "mt": False, "tr": False, "ac": False, "tp": False, }) for msg in track: abs_tick += msg.time msg_ch = int(getattr(msg, "channel", -1)) # Channel filter: when the picker entry was a format-0 split, only # process events on the chosen channel. Channel-less meta events # (set_tempo, etc.) have channel == -1 and pass through unaffected # because the message types we act on below all have a channel. if channel_filter is not None and msg_ch != -1 and msg_ch != channel_filter: continue if msg.type == "note_on" and int(getattr(msg, "velocity", 0)) > 0: if msg_ch == 9: continue # skip percussion pitch = int(msg.note) active.setdefault((msg_ch, pitch), deque()).append(abs_tick) elif msg.type == "note_off" or ( msg.type == "note_on" and int(getattr(msg, "velocity", 0)) == 0 ): pitch = int(msg.note) stack = active.get((msg_ch, pitch)) if not stack: continue # FIFO match against the oldest still-active start so overlapping # retriggers each get a sensible end time. start_tick = stack.popleft() if not stack: active.pop((msg_ch, pitch), None) if pedal_down.get(msg_ch, False): # Defer: extend the note until pedal-up. pedal_pending.setdefault(msg_ch, []).append((pitch, start_tick)) else: _emit(pitch, start_tick, abs_tick) elif msg.type == "control_change" and int(getattr(msg, "control", -1)) == 64: was_down = pedal_down.get(msg_ch, False) now_down = int(getattr(msg, "value", 0)) >= 64 pedal_down[msg_ch] = now_down if was_down and not now_down: # Pedal-up: finalise every pending note on this channel. pending = pedal_pending.pop(msg_ch, []) for pitch, start_tick in pending: _emit(pitch, start_tick, abs_tick) # End-of-track: close anything still active or held by the pedal, # using abs_tick as the end. Pedaled notes that never saw a pedal-up # land here too. for (_ch, pitch), starts in active.items(): for start_tick in starts: _emit(pitch, start_tick, abs_tick) active.clear() for _ch, pending in pedal_pending.items(): for pitch, start_tick in pending: _emit(pitch, start_tick, abs_tick) pedal_pending.clear() notes_out.sort(key=lambda n: n["t"]) return { "name": name, "tuning": [0, 0, 0, 0, 0, 0], "capo": 0, "notes": notes_out, "chords": [], "anchors": [], "handshapes": [], "templates": [], } # ── Tempo + drum-import shared helpers ─────────────────────────────────────── def _build_tick_to_seconds(midi: mido.MidiFile, track_index: int) -> Callable[[int], float]: """Return an `(abs_tick) -> seconds` function for the chosen track. Tempo-event scope depends on the SMF format (mirrors the keys converter): - type 0: single track holds tempo + notes; walk it alone. - type 1: parallel tracks share the timeline; merge tempo events. - type 2: independent timelines; tempo only from the chosen track. """ # A metrical header carries positive ticks-per-beat. mido reads the SMF # division as a signed short, so an SMPTE-division file surfaces as a # negative value and a malformed header as 0 — both make the two division # sites below divide by a non-positive number (ZeroDivisionError, or # negative seconds that send the bar walk off the rails). Fall back to the # SMF default here, the single place every caller routes ticks through, so # each caller's own fallback is real rather than cosmetic. ticks_per_beat = midi.ticks_per_beat if midi.ticks_per_beat > 0 else 480 raw_events: list[tuple[int, int]] = [(0, 500000)] # default 120 BPM midi_type = getattr(midi, "type", 1) tempo_source = ( [midi.tracks[track_index]] if midi_type == 2 else midi.tracks ) for tr in tempo_source: abs_tick = 0 for msg in tr: abs_tick += msg.time if msg.type == "set_tempo": raw_events.append((abs_tick, int(msg.tempo))) raw_events.sort(key=lambda e: e[0]) deduped: list[tuple[int, int]] = [] for ev in raw_events: if deduped and deduped[-1][0] == ev[0]: deduped[-1] = ev else: deduped.append(ev) tempo_table: list[tuple[int, float, int]] = [] cum_seconds = 0.0 prev_tick = 0 prev_tempo = deduped[0][1] for ev_tick, ev_tempo in deduped: cum_seconds += (ev_tick - prev_tick) * (prev_tempo / 1_000_000.0) / ticks_per_beat tempo_table.append((ev_tick, cum_seconds, ev_tempo)) prev_tick = ev_tick prev_tempo = ev_tempo tempo_ticks = [row[0] for row in tempo_table] def tick_to_seconds(tick: int) -> float: i = bisect_right(tempo_ticks, tick) - 1 if i < 0: i = 0 base_tick, base_seconds, tempo = tempo_table[i] return base_seconds + (tick - base_tick) * (tempo / 1_000_000.0) / ticks_per_beat return tick_to_seconds # Safety valve for the bar walk below: a malformed SMF (absurd tempo + long # trailing meta) could otherwise imply millions of bars. Real charts sit # orders of magnitude below this. _TEMPO_MAP_MAX_BARS = 20000 def convert_midi_tempo_map(midi_path: str, track_index: int = 0) -> dict: """Extract the song-timeline grid a `.mid` file carries: tempos, time signatures, and a full beat grid — the data the note converters here always computed internally (to bake note times) and then threw away, which left every MIDI import with no bars, no measures, and an implied 4/4 no matter what the file said. Returns ``{"tempos": [...], "time_signatures": [...], "beats": [...]}``: - ``tempos``: ``{time, bpm}`` per tempo event (deduped per tick). - ``time_signatures``: ``{time, ts: [num, den]}`` per signature event — the song-timeline sidecar shape (feedpak-spec §7.4). - ``beats``: one row per beat on the editor grid shape — downbeats carry a running ``measure`` (1, 2, 3, …) plus a ``den`` hint (the signature denominator), interior beats carry ``measure: -1``. The beat unit follows the active signature (6/8 ⇒ six eighth-note rows per bar). Event scope mirrors ``_build_tick_to_seconds``: SMF type 0/1 merge meta from all tracks (shared timeline); type 2 reads ONLY ``track_index`` (independent timelines — callers must never share one grid across type-2 tracks). Signature changes apply at the NEXT bar boundary when a file places one mid-bar (ill-formed but seen in the wild). All times are computed from absolute ticks through the cumulative tempo table and rounded once at emit — rounding error never accumulates with song length. An SMF with no note events yields empty ``beats``. """ midi = mido.MidiFile(midi_path) # Positive for metrical files; 0 (malformed) or negative (SMPTE division, # read as a signed short) otherwise — fall back so beat_ticks below stays # sane, mirroring the guard inside _build_tick_to_seconds. ticks_per_beat = midi.ticks_per_beat if midi.ticks_per_beat > 0 else 480 midi_type = getattr(midi, "type", 1) # Same scope both converters use: type 2 reads only the chosen track # (independent timelines); type 0/1 merge all tracks (shared timeline). source_tracks = ( [midi.tracks[track_index]] if midi_type == 2 else midi.tracks ) tick_to_seconds = _build_tick_to_seconds(midi, track_index) # ── collect meta + the end of musical content in one pass ──────────── sig_events: list[tuple[int, int, int]] = [] tempo_events: list[tuple[int, int]] = [] end_tick = 0 for tr in source_tracks: abs_tick = 0 for msg in tr: abs_tick += msg.time if msg.type == "time_signature": num = int(getattr(msg, "numerator", 4) or 4) den = int(getattr(msg, "denominator", 4) or 4) if num > 0 and den > 0: sig_events.append((abs_tick, num, den)) elif msg.type == "set_tempo": tempo_events.append((abs_tick, int(msg.tempo))) elif msg.type in ("note_on", "note_off"): end_tick = max(end_tick, abs_tick) # Dedupe at equal ticks (last wins), matching the tempo-table rule. sig_events.sort(key=lambda e: e[0]) sigs: list[tuple[int, int, int]] = [] for ev in sig_events: if sigs and sigs[-1][0] == ev[0]: sigs[-1] = ev else: sigs.append(ev) if not sigs or sigs[0][0] > 0: sigs.insert(0, (0, 4, 4)) tempo_events.sort(key=lambda e: e[0]) seen_tempo_ticks: dict[int, int] = {} for ev_tick, ev_tempo in tempo_events: seen_tempo_ticks[ev_tick] = ev_tempo sorted_tempo_ticks = sorted(seen_tempo_ticks) tempos_out: list[dict] = [] # Seed the MIDI default (120 BPM) at time 0 when the first tempo event # lands after the start (or there are none). The beat grid already runs # at 120 for the head of the song, so the sidecar must say so too — # symmetric with the (0, 4, 4) default seeded into the signatures above. if not sorted_tempo_ticks or sorted_tempo_ticks[0] > 0: tempos_out.append({"time": 0.0, "bpm": 120.0}) for ev_tick in sorted_tempo_ticks: tempos_out.append({ "time": round(tick_to_seconds(ev_tick), 3), "bpm": round(60_000_000.0 / seen_tempo_ticks[ev_tick], 3), }) time_signatures_out = [ {"time": round(tick_to_seconds(t), 3), "ts": [num, den]} for t, num, den in sigs ] # ── walk bars from tick 0 to the end of the notes ──────────────────── beats: list[dict] = [] if end_tick > 0: cur_tick = 0.0 measure = 1 sig_idx = 0 while cur_tick < end_tick and measure <= _TEMPO_MAP_MAX_BARS: # Active signature: the latest event at or before this bar's # start. Mid-bar events wait for the next boundary by # construction (we only re-read between bars). while (sig_idx + 1 < len(sigs) and sigs[sig_idx + 1][0] <= cur_tick + 1e-6): sig_idx += 1 _, num, den = sigs[sig_idx] beat_ticks = ticks_per_beat * 4.0 / den beats.append({ "time": round(tick_to_seconds(int(round(cur_tick))), 3), "measure": measure, "den": den, }) for k in range(1, num): sub_tick = cur_tick + k * beat_ticks if sub_tick >= end_tick: break beats.append({ "time": round(tick_to_seconds(int(round(sub_tick))), 3), "measure": -1, }) cur_tick += num * beat_ticks measure += 1 return { "tempos": tempos_out, "time_signatures": time_signatures_out, "beats": beats, } # ── Drum track listing (channel-9 only) ────────────────────────────────────── # Velocity below this is treated as a ghost note. GM doesn't have an explicit # ghost flag; chartists encode dynamics through velocity. 40 is the same # threshold the drums plugin uses for ghost-note styling. _GHOST_VELOCITY = 40 # Two hits on the same piece closer together than this are interpreted as a # flam (the later one carries `f: true`). 30 ms matches the drums plugin's # leading-glyph offset for flam rendering. _FLAM_WINDOW_S = 0.030 # A cymbal note whose explicit note-off arrives within this window after the # note-on is treated as a choke (the chartist clamped the cymbal). 120 ms # matches the spec's mention of choke tail durations. _CHOKE_MAX_S = 0.120 def list_drum_tracks(midi_path: str) -> list[dict]: """List tracks that contain GM channel-9 (percussion) note_on events. For format-0 files (everything in one track, channels intermixed), this surfaces the lone track once when it has channel-9 hits. For format-1/2 files, each track that fires channel-9 notes shows up. Mirrors `list_midi_tracks` so the editor's +Drums modal can show the same shape of picker entry the +Keys modal does. Each item: {index, name, instrument, notes, channel, is_piano, is_drums, strings, is_percussion, channel_filter} — the same picker-entry shape `list_midi_tracks` emits, so the editor frontend can consume either list uniformly. For drum tracks the classification fields are fixed: `is_drums`/`is_percussion` True, `is_piano` False, `instrument` -1, `strings` 0. `channel_filter` is always 9 — the converter uses it to skip non-drum events on a mixed-channel track. """ midi = mido.MidiFile(midi_path) out: list[dict] = [] for i, track in enumerate(midi.tracks): name = "" note_count = 0 for msg in track: if msg.type == "track_name" and not name: name = msg.name or "" elif ( msg.type == "note_on" and int(getattr(msg, "velocity", 0)) > 0 and int(getattr(msg, "channel", -1)) == 9 ): note_count += 1 if note_count == 0: continue out.append({ "index": i, "channel_filter": 9, "name": name or f"Track {i} (drums)", "instrument": -1, "notes": note_count, "channel": 9, "is_piano": False, "is_drums": True, "strings": 0, "is_percussion": True, }) return out def convert_drum_track_from_midi( midi_path: str, track_index: int, audio_offset: float = 0.0, name: str = "Drums", *, out_unmapped: dict[int, dict] | None = None, ) -> dict: """Convert a MIDI drum track to a `drum_tab.json` dict. Reads channel-9 note_on events on the chosen track, maps each MIDI note to a piece-id via `lib.drums.midi_to_piece`, and emits hits with velocity preserved verbatim. Three heuristics encode articulations that GM MIDI doesn't have explicit flags for: - **Ghost**: velocity < 40 → `g: true`. - **Flam**: two hits on the same piece within 30 ms → the louder one (the main strike) carries `f: true` and the quieter grace note is dropped (the renderer draws the leading glyph itself from the `f` flag). Typically the grace note arrives first in time, but the heuristic is velocity-based to handle MIDI files where encoding order differs from chronological order. - **Choke**: a cymbal note-off arriving within 120 ms of its note-on sets `k` to the actual on→off duration. Callers can pass an empty dict as ``out_unmapped`` to receive a per-MIDI record of every channel-9 note_on that didn't resolve to a piece-id (``{midi: {"count": int, "times": [float, ...], "velocities": [int, ...]}}``, times/velocities index-aligned and capped at 100 samples per note — velocities carry the source notes' real dynamics so a hand-mapping UI doesn't have to flatten them to a default). The default path skips this capture entirely so MIDIs heavy with cowbell/tambourine/etc. take no extra work. """ offset = float(audio_offset) if not math.isfinite(offset): raise ValueError(f"audio_offset must be a finite number, got {audio_offset!r}") midi = mido.MidiFile(midi_path) if track_index < 0 or track_index >= len(midi.tracks): raise ValueError(f"track_index {track_index} out of range") tick_to_seconds = _build_tick_to_seconds(midi, track_index) track = midi.tracks[track_index] # Two passes: collect raw note_on/note_off pairs in pass 1 (so we know # each on's actual off time for choke detection), then apply the # flam-collapse + serialisation in pass 2. raw: list[dict] = [] # one entry per note_on # Use list[int] per MIDI note so overlapping/retriggered hits (note_on # before note_off for the same note) are each tracked independently # rather than the later one overwriting the earlier one's index. open_hits: dict[int, deque[int]] = {} # midi note -> FIFO queue of indices in `raw` abs_tick = 0 for msg in track: abs_tick += msg.time if int(getattr(msg, "channel", -1)) != 9: continue if msg.type == "note_on" and int(getattr(msg, "velocity", 0)) > 0: midi_note = int(msg.note) piece = drums_mod.midi_to_piece(midi_note) if piece is None: # Default path: drop silently. Only pay the tick->seconds # cost on the opt-in capture path so MIDIs full of unmapped # percussion don't take a perf hit when the caller didn't # ask for unmapped reporting. if out_unmapped is None: continue t = tick_to_seconds(abs_tick) + offset entry = out_unmapped.setdefault( midi_note, {"count": 0, "times": [], "velocities": []}) entry["count"] += 1 if len(entry["times"]) < 100: entry["times"].append(round(t, 3)) # Index-aligned with times: the note's real dynamics, # so hand-mapping doesn't flatten everything to 100. entry["velocities"].append(int(msg.velocity)) continue # Mapped note: compute t once for the raw entry. t = tick_to_seconds(abs_tick) + offset raw.append({ "t": t, "p": piece, "v": int(msg.velocity), "_midi": midi_note, "_on_tick": abs_tick, }) open_hits.setdefault(midi_note, deque()).append(len(raw) - 1) elif msg.type == "note_off" or ( msg.type == "note_on" and int(getattr(msg, "velocity", 0)) == 0 ): midi_note = int(msg.note) stack = open_hits.get(midi_note) if not stack: continue idx = stack.popleft() # FIFO: oldest note_on matches this note_off if not stack: del open_hits[midi_note] hit = raw[idx] if drums_mod.piece_category(hit["p"]) != "cymbal": continue on_secs = tick_to_seconds(hit["_on_tick"]) off_secs = tick_to_seconds(abs_tick) dur = off_secs - on_secs if 0.0 < dur <= _CHOKE_MAX_S: hit["k"] = round(dur, 3) raw.sort(key=lambda h: (h["t"], h["p"])) # Flam collapse: for each hit, compare it against the most-recent # previous hit of the SAME piece (not just the globally adjacent entry), # so an intervening hit from a different piece does not break flam # detection for densely-played patterns (e.g. kick + snare flam). flam_indices: set[int] = set() drop_indices: set[int] = set() last_by_piece: dict[str, int] = {} # piece-id -> index in raw[] for i, curr in enumerate(raw): piece = curr["p"] prev_i = last_by_piece.get(piece) if prev_i is not None and prev_i not in drop_indices: prev = raw[prev_i] if (curr["t"] - prev["t"]) <= _FLAM_WINDOW_S: # Prefer the louder hit as the "main"; the quieter is the # leading grace. The main hit receives `f: true` so the # renderer draws a small grace glyph slightly ahead of it. if prev["v"] <= curr["v"]: drop_indices.add(prev_i) flam_indices.add(i) else: drop_indices.add(i) flam_indices.add(prev_i) # Don't advance last_by_piece — keep prev_i as anchor so a # triple flam doesn't chain two drops. continue last_by_piece[piece] = i out_hits: list[dict] = [] for i, hit in enumerate(raw): if i in drop_indices: continue # Round here (not at append time) so flam comparisons above used full precision. new_hit: dict = {"t": round(hit["t"], 3), "p": hit["p"]} vel = hit["v"] if 1 <= vel <= 127: new_hit["v"] = vel if vel < _GHOST_VELOCITY: new_hit["g"] = True if i in flam_indices: new_hit["f"] = True if "k" in hit: new_hit["k"] = hit["k"] out_hits.append(new_hit) # Build kit legend from the union of piece-ids that survived. seen_pieces: list[str] = [] seen_set: set[str] = set() for h in out_hits: if h["p"] not in seen_set: seen_set.add(h["p"]) seen_pieces.append(h["p"]) return { "version": drums_mod.SCHEMA_VERSION, "name": name, "kit": [ {"id": pid, "name": pid.replace("_", " ").title()} for pid in seen_pieces ], "hits": out_hits, } # ── Lyrics + vocal-melody extraction ───────────────────────────────────────── # Vocal-track detection, mirroring the idiom in `lib/gp2rs_gpx.py`'s # `_is_vocal_track` (GM voice/choir/lead-voice programs + name keywords). # Kept as a local copy because that helper consumes gp2rs_gpx's own GP track # dicts, not raw MIDI tracks. "melody" is added to the name hints: karaoke # MIDIs commonly label the sung line "Melody" rather than "Vocals". _VOCAL_MIDI_PROGRAMS = {52, 53, 54, 85, 86, 87} # Choir Aahs, Voice Oohs, Synth Voice, Lead 5-7 (voice) _VOCAL_NAME_HINTS = ("vocal", "voice", "vox", "sing", "lyric", "choir", "melody") # A dedicated karaoke *text* track (SMF 0x01 Text events, `.kar` convention) # is usually noteless and named "Words" or "Soft Karaoke" — names the vocal # hints above don't catch. Only the Text-event fallback consults this wider # set; note-track detection sticks to the gp2rs_gpx idiom. _LYRIC_TEXT_TRACK_HINTS = _VOCAL_NAME_HINTS + ("words", "karaoke") # A lyric event pairs with a vocal note-on when their onsets sit within this # window. Karaoke files place the lyric event at (or a hair before) the # note-on tick, so real matches are ~0; the window only absorbs sloppy # authoring, and staying well under a typical syllable gap keeps a melisma's # extra notes from being stolen by the next syllable. _LYRIC_PAIR_TOLERANCE_S = 0.30 # Duration bounds for lyric entries with no pairable note (spoken lines, # lyrics-only files). "Until the next lyric event" is the natural display # duration, capped so a verse-final syllable before a long instrumental # break doesn't linger on screen, and floored so simultaneous/out-of-order # events can't produce a zero or negative duration. _UNPAIRED_LYRIC_MAX_D = 2.0 _UNPAIRED_LYRIC_MIN_D = 0.1 # Leading/word/trailing whitespace splitter for lyric tokens. DOTALL so # embedded newlines land in a group rather than killing the match. _LYRIC_TOKEN_RE = re.compile(r"^(\s*)(.*?)(\s*)$", re.S) def _scan_tracks_for_lyrics(midi: mido.MidiFile) -> list[dict]: """One pass per track collecting the raw material `extract_midi_lyrics` needs: name, per-channel programs, melodic (non-drum) notes with their on/off ticks, and Lyric/Text meta events. Each item: {name, channel_programs: {ch: program}, notes: [(start_tick, end_tick, pitch, channel)], lyric_events: [(tick, text)], text_events: [(tick, text)]}. Note pairing uses the same FIFO note_on/note_off matching as the keys converter so retriggers don't cross-wire durations. """ out: list[dict] = [] for track in midi.tracks: name = "" channel_programs: dict[int, int] = {} lyric_events: list[tuple[int, str]] = [] text_events: list[tuple[int, str]] = [] notes: list[tuple[int, int, int, int]] = [] active: dict[tuple[int, int], deque[int]] = {} abs_tick = 0 for msg in track: abs_tick += msg.time if msg.type == "track_name" and not name: name = msg.name or "" elif msg.type == "lyrics": lyric_events.append((abs_tick, msg.text or "")) elif msg.type == "text": text_events.append((abs_tick, msg.text or "")) elif msg.type == "program_change": ch = int(getattr(msg, "channel", -1)) if ch != 9 and ch not in channel_programs: channel_programs[ch] = int(msg.program) elif msg.type == "note_on" and int(getattr(msg, "velocity", 0)) > 0: ch = int(getattr(msg, "channel", -1)) if ch == 9: continue active.setdefault((ch, int(msg.note)), deque()).append(abs_tick) elif msg.type == "note_off" or ( msg.type == "note_on" and int(getattr(msg, "velocity", 0)) == 0 ): ch = int(getattr(msg, "channel", -1)) pitch = int(msg.note) stack = active.get((ch, pitch)) if not stack: continue start_tick = stack.popleft() if not stack: active.pop((ch, pitch), None) notes.append((start_tick, abs_tick, pitch, ch)) # Close anything left hanging at end-of-track, mirroring the keys # converter's end-of-track sweep. for (ch, pitch), starts in active.items(): for start_tick in starts: notes.append((start_tick, abs_tick, pitch, ch)) notes.sort(key=lambda n: n[0]) out.append({ "name": name, "channel_programs": channel_programs, "notes": notes, "lyric_events": lyric_events, "text_events": text_events, }) return out def _name_matches(name: str, hints: tuple[str, ...]) -> bool: name_l = (name or "").lower() return any(h in name_l for h in hints) def _normalize_lyric_tokens(events: list[tuple[int, str]]) -> list[dict]: """Turn raw lyric/text meta events into clean syllable tokens. Handles both encodings seen in the wild: - **`.kar` / karaoke convention**: `/` prefix = new line, `\\` prefix = new paragraph (both mean "the previous syllable ended a line"), `-` suffix = syllable joins the next one, `@`-prefixed tokens are file metadata (`@KMIDI`, `@T`, ...) and are dropped. - **Plain Lyric-event convention**: word boundaries carried by leading or trailing spaces; line breaks carried by embedded CR/LF. Each token: {tick, word, lead_ws, trail_ws, line_end}. Spacing-only and newline-only events don't emit a token — they fold their meaning (word-break / line-end) onto the previous one. """ toks: list[dict] = [] for tick, raw in events: text = "" if raw is None else str(raw) if not text: continue if text.lstrip().startswith(("@", "%")): # .kar metadata / sequencer directives, not sung text. continue kar_break = text[0] in ("/", "\\") if kar_break: text = text[1:] m = _LYRIC_TOKEN_RE.match(text) head, body, tail = m.group(1), m.group(2), m.group(3) nl_before = ("\n" in head) or ("\r" in head) nl_after = ("\n" in tail) or ("\r" in tail) if "\n" in body or "\r" in body: # Rare multi-line event: keep it one token, treat the break as # trailing so the line ends after this token. body = re.sub(r"[\r\n]+", " ", body).strip() nl_after = True if (kar_break or nl_before) and toks: toks[-1]["line_end"] = True if not body: # Pure spacing/newline token: fold onto the previous syllable. if nl_after and toks: toks[-1]["line_end"] = True if toks: toks[-1]["trail_ws"] = True continue toks.append({ "tick": tick, "word": body, "lead_ws": bool(head), "trail_ws": bool(tail), "line_end": nl_after, }) return toks def _apply_word_conventions(toks: list[dict]) -> list[str]: """Map tokens to spec §7.1 `w` strings: trailing ``-`` joins to the next syllable, trailing ``+`` ends a line. Which join convention the source used is detected per stream: - Any token already carrying a ``-`` suffix → the stream is hyphen-delimited (`.kar` style); those suffixes are the spec's own join marker and pass through untouched. - Otherwise, if the stream carries any spacing at all → space-delimited: a token with no trailing space followed by a token with no leading space is a mid-word syllable and gains a ``-``. - No hyphens and no spacing anywhere → the tokens are whole words (common for Text-event lyrics); no joins are synthesized. """ has_hyphens = any(t["word"].endswith("-") for t in toks) has_spacing = any(t["lead_ws"] or t["trail_ws"] for t in toks) words: list[str] = [] for i, tk in enumerate(toks): w = tk["word"] nxt = toks[i + 1] if i + 1 < len(toks) else None if tk["line_end"]: # A join can't cross a line break — the line marker wins. if w.endswith("-"): w = w[:-1] if w and not w.endswith("+"): w += "+" elif nxt is not None and not has_hyphens and has_spacing: if not tk["trail_ws"] and not nxt["lead_ws"] and not w.endswith("-"): w += "-" words.append(w) return words def _select_vocal_notes( scans: list[dict], lyric_track_index: int, midi_type: int, ) -> tuple[int, list[tuple[int, int, int, int]]] | None: """Pick the note pool the lyric syllables should be pitch-paired with. Returns ``(track_index, notes)`` or ``None`` when no vocal melody is identifiable (→ lyrics-only import). Selection order: 1. The lyric-carrying track itself, when it has notes: - channels with a vocal GM program → only those channels' notes (isolates the sung line inside a format-0 everything-in-one-track file); - vocal-ish track name → all its non-drum notes; - SMF type 1/2 with neither → still trusted: a track that interleaves per-syllable Lyric events with its own notes *is* the karaoke melody by construction. Format-0 files don't get this benefit of the doubt — there the single track holds every instrument, so without a vocal program/name there is no way to isolate the melody and we fall back to lyrics-only. 2. Otherwise (dedicated noteless "Words" track), the vocal-ish track — by name hint or vocal GM program, mirroring gp2rs_gpx — with the most notes; within it, vocal-program channels only when present. """ def _vocal_channels(scan: dict) -> set[int]: return { ch for ch, prog in scan["channel_programs"].items() if prog in _VOCAL_MIDI_PROGRAMS } def _pool(scan: dict) -> list[tuple[int, int, int, int]]: chans = _vocal_channels(scan) if chans: return [n for n in scan["notes"] if n[3] in chans] return scan["notes"] src = scans[lyric_track_index] if src["notes"]: if _vocal_channels(src) or _name_matches(src["name"], _VOCAL_NAME_HINTS): return lyric_track_index, _pool(src) if midi_type != 0: return lyric_track_index, src["notes"] return None best: tuple[int, list] | None = None for i, scan in enumerate(scans): if not scan["notes"]: continue if not (_vocal_channels(scan) or _name_matches(scan["name"], _VOCAL_NAME_HINTS)): continue pool = _pool(scan) if pool and (best is None or len(pool) > len(best[1])): best = (i, pool) return best def extract_midi_lyrics(midi_path: str, audio_offset: float = 0.0) -> dict | None: """Extract lyrics (and, when pairable, the vocal melody) from a `.mid`. Returns ``None`` when the file carries no usable lyric events — callers then change nothing, leaving any existing manifest keys and sidecar files untouched. Otherwise returns:: { "lyrics": [{"t": float, "d": float, "w": str}, ...], "lyrics_source": "authored", "vocal_pitch": {"version": 1, "notes": [{"t", "d", "midi"}, ...]} | None, } ``lyrics`` is the feedpak `lyrics.json` payload (spec §7.1: flat list, no version field; ``w`` uses trailing ``-`` for syllable joins and trailing ``+`` for line ends). ``vocal_pitch`` is the `vocal_pitch.json` payload (spec §7.2, same shape as gp2rs_gpx's ``convert_vocal_track_to_pitch_sidecar`` and the lyrics-karaoke plugin's ``_persist_pitch``) — ``None`` when no vocal note track could be identified, in which case the caller writes `lyrics.json` only. ``lyrics_source`` is always ``"authored"`` (spec §7.1 vocabulary): lyric meta events are chart-author data, not machine transcription. Callers assembling a pack write ``lyrics.json`` / ``vocal_pitch.json`` and set the manifest ``lyrics`` / ``lyrics_source`` / ``vocal_pitch`` keys — and should do so only for keys not already present, so an import never clobbers lyrics that arrived from another source. Sourcing rules: - Lyric text comes from SMF Lyric (0x05) meta events — the track with the most of them wins when several carry some. When the file has none at all, Text (0x01) events are accepted as a fallback, but only from a vocal-ish track (gp2rs_gpx-style name/program detection, widened with "words"/"karaoke" for `.kar` text tracks) — Text events elsewhere are copyright notices / markers, not lyrics. - `.kar` conventions are normalized (see ``_normalize_lyric_tokens`` / ``_apply_word_conventions``): ``/`` and ``\\`` line-break prefixes become the spec's ``+`` suffix on the previous syllable, ``@`` metadata tokens are dropped, ``-`` hyphen joins pass through. - Each syllable is paired with the vocal note (see ``_select_vocal_notes``) whose onset falls within ``_LYRIC_PAIR_TOLERANCE_S`` of the lyric event, greedily in time order, one note per syllable. Paired syllables snap ``t``/``d`` to the note (the authored melody is timing-authoritative, and keeps `lyrics.json` and `vocal_pitch.json` mirrored per §7.2); a melisma's extra notes are skipped. Unpaired syllables (talkies) keep the lyric event's own time and run until the next syllable, clamped to [``_UNPAIRED_LYRIC_MIN_D``, ``_UNPAIRED_LYRIC_MAX_D``] — they appear in ``lyrics`` only, which spec §7.2 explicitly allows (`vocal_pitch.notes` MAY be shorter than `lyrics.json`). ``audio_offset`` (seconds) shifts every emitted time, same handle as the keys/drums converters. Tempo-map scope per SMF type also matches them (type 2 reads only the involved track's tempo events). """ offset = float(audio_offset) if not math.isfinite(offset): raise ValueError(f"audio_offset must be a finite number, got {audio_offset!r}") midi = mido.MidiFile(midi_path) midi_type = getattr(midi, "type", 1) scans = _scan_tracks_for_lyrics(midi) # ── choose the lyric event stream ──────────────────────────────────── lyric_idx = -1 best_count = 0 for i, scan in enumerate(scans): if len(scan["lyric_events"]) > best_count: lyric_idx = i best_count = len(scan["lyric_events"]) if lyric_idx >= 0: toks = _normalize_lyric_tokens(scans[lyric_idx]["lyric_events"]) else: # Text-event fallback: vocal-ish tracks only (plus .kar "Words" / # "Soft Karaoke" text tracks). Normalize before counting so a track # of @-metadata can't outscore a real lyric track. toks = [] for i, scan in enumerate(scans): if not scan["text_events"]: continue vocal_prog = any( p in _VOCAL_MIDI_PROGRAMS for p in scan["channel_programs"].values() ) if not (vocal_prog or _name_matches(scan["name"], _LYRIC_TEXT_TRACK_HINTS)): continue cand = _normalize_lyric_tokens(scan["text_events"]) if len(cand) > len(toks): lyric_idx = i toks = cand if lyric_idx < 0 or not toks: return None words = _apply_word_conventions(toks) lyric_tick_to_seconds = _build_tick_to_seconds(midi, lyric_idx) # ── pick + time the vocal note pool ────────────────────────────────── picked = _select_vocal_notes(scans, lyric_idx, midi_type) vocal_notes: list[dict] = [] if picked is not None: note_idx, pool = picked # Type-2 tracks own independent timelines — time the notes through # their own track's tempo scope (same map as the lyric track for # type 0/1, where tempo is merged across tracks anyway). note_tick_to_seconds = ( lyric_tick_to_seconds if note_idx == lyric_idx else _build_tick_to_seconds(midi, note_idx) ) for start_tick, end_tick, pitch, _ch in pool: t = note_tick_to_seconds(start_tick) vocal_notes.append({ "t": t, "d": max(0.0, note_tick_to_seconds(end_tick) - t), "midi": int(pitch), }) vocal_notes.sort(key=lambda n: n["t"]) # ── pair syllables with notes (greedy, time-ordered) ───────────────── entries: list[dict] = [] # {t, d (None until resolved), w, paired} j = 0 for tk, w in zip(toks, words): t_lyric = lyric_tick_to_seconds(tk["tick"]) while (j < len(vocal_notes) and vocal_notes[j]["t"] < t_lyric - _LYRIC_PAIR_TOLERANCE_S): j += 1 if (j < len(vocal_notes) and vocal_notes[j]["t"] <= t_lyric + _LYRIC_PAIR_TOLERANCE_S): note = vocal_notes[j] j += 1 entries.append({ "t": note["t"], "d": note["d"], "w": w, "midi": note["midi"], "paired": True, }) else: entries.append({"t": t_lyric, "d": None, "w": w, "paired": False}) # Snapping can nudge a paired syllable past an unpaired neighbour; # sort so both sidecars stay chronological for downstream consumers. entries.sort(key=lambda e: e["t"]) # Unpaired durations: until the next syllable, clamped. Resolved after # the sort so "next" is the true chronological neighbour. for i, e in enumerate(entries): if e["d"] is None: if i + 1 < len(entries): gap = entries[i + 1]["t"] - e["t"] d = min(gap, _UNPAIRED_LYRIC_MAX_D) else: d = _UNPAIRED_LYRIC_MAX_D e["d"] = max(d, _UNPAIRED_LYRIC_MIN_D) lyrics_out = [ {"t": round(e["t"] + offset, 3), "d": round(e["d"], 3), "w": e["w"]} for e in entries ] pitch_notes = [ {"t": round(e["t"] + offset, 3), "d": round(e["d"], 3), "midi": int(e["midi"])} for e in entries if e["paired"] ] return { "lyrics": lyrics_out, "lyrics_source": "authored", "vocal_pitch": ( {"version": 1, "notes": pitch_notes} if pitch_notes else None ), }