midi_diff.py
python
sha256:ac29b8ba9021514a03ab2782d92bf67671f0efa5b3b70d46f7598c5d4e923378
docs: record muse#74 Phase 4 live verification (SCR_13, SCR_14)
Sonnet 5
13 days ago
| 1 | """MIDI note-level diff for the Muse MIDI plugin. |
| 2 | |
| 3 | Produces a ``StructuredDelta`` with note-level ``InsertOp`` and ``DeleteOp`` |
| 4 | entries from two MIDI byte strings. This is what lets ``muse read`` display |
| 5 | "C4 added at beat 3.5" rather than "tracks/drums.mid modified". |
| 6 | |
| 7 | Algorithm |
| 8 | --------- |
| 9 | 1. Parse MIDI bytes and extract paired note events (note_on + note_off) |
| 10 | sorted by start tick. |
| 11 | 2. Represent each note as a ``NoteKey`` TypedDict with five fields. |
| 12 | 3. Convert each ``NoteKey`` to its deterministic content ID (SHA-256 of the |
| 13 | five fields). |
| 14 | 4. Delegate to :func:`~muse.core.diff_algorithms.lcs.myers_ses` — the shared |
| 15 | LCS implementation from the diff algorithm library — for the SES. |
| 16 | 5. Map edit steps to typed ``DomainOp`` instances using the note's content |
| 17 | ID and a human-readable summary string. |
| 18 | 6. Wrap the ops in a ``StructuredDelta``. |
| 19 | |
| 20 | Additional features |
| 21 | ----------------- |
| 22 | :func:`reconstruct_midi` — the inverse of :func:`extract_notes`. Given a list |
| 23 | of :class:`NoteKey` objects and a ticks_per_beat value, produces raw MIDI bytes |
| 24 | for a Type 0 single-track file. Used by ``MidiPlugin.merge_ops()`` to |
| 25 | materialise a merged MIDI file after the OT engine has determined that |
| 26 | two branches' note-level operations commute. |
| 27 | |
| 28 | Public API |
| 29 | ---------- |
| 30 | - :class:`NoteKey` — typed MIDI note identity. |
| 31 | - :func:`extract_notes` — MIDI bytes → sorted ``list[NoteKey]``. |
| 32 | - :func:`reconstruct_midi` — ``list[NoteKey]`` → MIDI bytes. |
| 33 | - :func:`diff_midi_notes` — top-level: MIDI bytes × 2 → ``StructuredDelta``. |
| 34 | """ |
| 35 | |
| 36 | import hashlib |
| 37 | import io |
| 38 | import logging |
| 39 | from dataclasses import dataclass |
| 40 | from typing import TYPE_CHECKING, Literal, TypedDict |
| 41 | |
| 42 | if TYPE_CHECKING: |
| 43 | from muse.plugins.midi.entity import EntityIndex |
| 44 | |
| 45 | import mido |
| 46 | |
| 47 | from muse.core.diff_algorithms.lcs import myers_ses |
| 48 | from muse.domain import ( |
| 49 | DeleteOp, |
| 50 | DomainOp, |
| 51 | InsertOp, |
| 52 | StructuredDelta, |
| 53 | ) |
| 54 | from muse.plugins.midi._midi_keys import NoteKey, _note_content_id, _note_summary, _pitch_name |
| 55 | |
| 56 | logger = logging.getLogger(__name__) |
| 57 | |
| 58 | #: Identifies the sub-domain for note-level operations inside a PatchOp. |
| 59 | _CHILD_DOMAIN = "midi_notes" |
| 60 | |
| 61 | # --------------------------------------------------------------------------- |
| 62 | # Note extraction |
| 63 | # --------------------------------------------------------------------------- |
| 64 | |
| 65 | def extract_notes(midi_bytes: bytes) -> tuple[list[NoteKey], int]: |
| 66 | """Parse *midi_bytes* and return ``(notes, ticks_per_beat)``. |
| 67 | |
| 68 | Notes are paired note_on / note_off events. A note_on with velocity=0 |
| 69 | is treated as note_off. Notes are sorted by start_tick then pitch for |
| 70 | deterministic ordering. |
| 71 | |
| 72 | Args: |
| 73 | midi_bytes: Raw bytes of a ``.mid`` file. |
| 74 | |
| 75 | Returns: |
| 76 | A tuple of (sorted NoteKey list, ticks_per_beat integer). |
| 77 | |
| 78 | Raises: |
| 79 | ValueError: When *midi_bytes* cannot be parsed as a MIDI file. |
| 80 | """ |
| 81 | try: |
| 82 | mid = mido.MidiFile(file=io.BytesIO(midi_bytes)) |
| 83 | except Exception as exc: |
| 84 | raise ValueError(f"Cannot parse MIDI bytes: {exc}") from exc |
| 85 | |
| 86 | ticks_per_beat: int = int(mid.ticks_per_beat) |
| 87 | # (channel, pitch) → (start_tick, velocity) |
| 88 | active: dict[tuple[int, int], tuple[int, int]] = {} |
| 89 | notes: list[NoteKey] = [] |
| 90 | |
| 91 | for track in mid.tracks: |
| 92 | abs_tick = 0 |
| 93 | for msg in track: |
| 94 | abs_tick += msg.time |
| 95 | if msg.type == "note_on" and msg.velocity > 0: |
| 96 | active[(msg.channel, msg.note)] = (abs_tick, msg.velocity) |
| 97 | elif msg.type == "note_off" or ( |
| 98 | msg.type == "note_on" and msg.velocity == 0 |
| 99 | ): |
| 100 | key = (msg.channel, msg.note) |
| 101 | if key in active: |
| 102 | start, vel = active.pop(key) |
| 103 | notes.append( |
| 104 | NoteKey( |
| 105 | pitch=msg.note, |
| 106 | velocity=vel, |
| 107 | start_tick=start, |
| 108 | duration_ticks=max(abs_tick - start, 1), |
| 109 | channel=msg.channel, |
| 110 | ) |
| 111 | ) |
| 112 | |
| 113 | # Close any notes still open at end of file with duration 1. |
| 114 | for (ch, pitch), (start, vel) in active.items(): |
| 115 | notes.append( |
| 116 | NoteKey( |
| 117 | pitch=pitch, |
| 118 | velocity=vel, |
| 119 | start_tick=start, |
| 120 | duration_ticks=1, |
| 121 | channel=ch, |
| 122 | ) |
| 123 | ) |
| 124 | |
| 125 | notes.sort(key=lambda n: (n["start_tick"], n["pitch"], n["channel"])) |
| 126 | return notes, ticks_per_beat |
| 127 | |
| 128 | # --------------------------------------------------------------------------- |
| 129 | # NoteKey-level edit script — adapter over the core LCS |
| 130 | # --------------------------------------------------------------------------- |
| 131 | |
| 132 | EditKind = Literal["keep", "insert", "delete"] |
| 133 | |
| 134 | @dataclass(frozen=True) |
| 135 | class EditStep: |
| 136 | """One step in the note-level edit script produced by :func:`lcs_edit_script`.""" |
| 137 | |
| 138 | kind: EditKind |
| 139 | base_index: int |
| 140 | target_index: int |
| 141 | note: NoteKey |
| 142 | |
| 143 | def lcs_edit_script( |
| 144 | base: list[NoteKey], |
| 145 | target: list[NoteKey], |
| 146 | ) -> list[EditStep]: |
| 147 | """Compute the shortest edit script transforming *base* into *target*. |
| 148 | |
| 149 | Converts each ``NoteKey`` to its content ID, delegates to |
| 150 | :func:`~muse.core.diff_algorithms.lcs.myers_ses` for the SES, then maps |
| 151 | the result back to :class:`EditStep` entries carrying the original |
| 152 | ``NoteKey`` values. |
| 153 | |
| 154 | Two notes are matched iff all five ``NoteKey`` fields are equal. This is |
| 155 | correct: a pitch change, velocity change, or timing shift is a delete of |
| 156 | the old note and an insert of the new one. |
| 157 | |
| 158 | Args: |
| 159 | base: The base (ancestor) note sequence. |
| 160 | target: The target (newer) note sequence. |
| 161 | |
| 162 | Returns: |
| 163 | A list of :class:`EditStep` entries (keep / insert / delete). |
| 164 | """ |
| 165 | base_ids = [_note_content_id(n) for n in base] |
| 166 | target_ids = [_note_content_id(n) for n in target] |
| 167 | raw_steps = myers_ses(base_ids, target_ids) |
| 168 | |
| 169 | result: list[EditStep] = [] |
| 170 | for step in raw_steps: |
| 171 | if step.kind == "keep": |
| 172 | result.append(EditStep("keep", step.base_index, step.target_index, base[step.base_index])) |
| 173 | elif step.kind == "insert": |
| 174 | result.append(EditStep("insert", step.base_index, step.target_index, target[step.target_index])) |
| 175 | else: |
| 176 | result.append(EditStep("delete", step.base_index, step.target_index, base[step.base_index])) |
| 177 | return result |
| 178 | |
| 179 | # --------------------------------------------------------------------------- |
| 180 | # Public diff entry point |
| 181 | # --------------------------------------------------------------------------- |
| 182 | |
| 183 | def diff_midi_notes( |
| 184 | base_bytes: bytes, |
| 185 | target_bytes: bytes, |
| 186 | *, |
| 187 | file_path: str = "", |
| 188 | ) -> StructuredDelta: |
| 189 | """Compute a note-level ``StructuredDelta`` between two MIDI files. |
| 190 | |
| 191 | Parses both files, converts each note to its content ID, delegates to the |
| 192 | core :func:`~muse.core.diff_algorithms.lcs.myers_ses` for the SES, then |
| 193 | maps the edit steps to typed ``DomainOp`` instances. |
| 194 | |
| 195 | Args: |
| 196 | base_bytes: Raw bytes of the base (ancestor) MIDI file. |
| 197 | target_bytes: Raw bytes of the target (newer) MIDI file. |
| 198 | file_path: Workspace-relative path of the file being diffed (used |
| 199 | only in log messages and ``content_summary`` strings). |
| 200 | |
| 201 | Returns: |
| 202 | A ``StructuredDelta`` with ``InsertOp`` and ``DeleteOp`` entries for |
| 203 | each note added or removed. The ``summary`` field is human-readable, |
| 204 | e.g. ``"3 notes added, 1 note removed"``. |
| 205 | |
| 206 | Raises: |
| 207 | ValueError: When either byte string cannot be parsed as MIDI. |
| 208 | """ |
| 209 | base_notes, base_tpb = extract_notes(base_bytes) |
| 210 | target_notes, target_tpb = extract_notes(target_bytes) |
| 211 | tpb = base_tpb # use base ticks_per_beat for human-readable summaries |
| 212 | |
| 213 | # Convert NoteKey → content ID, then delegate LCS to the core algorithm. |
| 214 | base_ids = [_note_content_id(n) for n in base_notes] |
| 215 | target_ids = [_note_content_id(n) for n in target_notes] |
| 216 | steps = myers_ses(base_ids, target_ids) |
| 217 | |
| 218 | # Build a content-ID → NoteKey lookup so we can produce rich summaries. |
| 219 | base_by_id = {_note_content_id(n): n for n in base_notes} |
| 220 | target_by_id = {_note_content_id(n): n for n in target_notes} |
| 221 | |
| 222 | child_ops: list[DomainOp] = [] |
| 223 | inserts = 0 |
| 224 | deletes = 0 |
| 225 | |
| 226 | for step in steps: |
| 227 | if step.kind == "insert": |
| 228 | note = target_by_id.get(step.item) |
| 229 | summary = _note_summary(note, tpb) if note else step.item[:12] |
| 230 | child_ops.append( |
| 231 | InsertOp( |
| 232 | op="insert", |
| 233 | address=f"note:{step.target_index}", |
| 234 | position=step.target_index, |
| 235 | content_id=step.item, |
| 236 | content_summary=summary, |
| 237 | ) |
| 238 | ) |
| 239 | inserts += 1 |
| 240 | elif step.kind == "delete": |
| 241 | note = base_by_id.get(step.item) |
| 242 | summary = _note_summary(note, tpb) if note else step.item[:12] |
| 243 | child_ops.append( |
| 244 | DeleteOp( |
| 245 | op="delete", |
| 246 | address=f"note:{step.base_index}", |
| 247 | position=step.base_index, |
| 248 | content_id=step.item, |
| 249 | content_summary=summary, |
| 250 | ) |
| 251 | ) |
| 252 | deletes += 1 |
| 253 | # "keep" steps produce no ops — the note is unchanged. |
| 254 | |
| 255 | parts: list[str] = [] |
| 256 | if inserts: |
| 257 | parts.append(f"{inserts} note{'s' if inserts != 1 else ''} added") |
| 258 | if deletes: |
| 259 | parts.append(f"{deletes} note{'s' if deletes != 1 else ''} removed") |
| 260 | child_summary = ", ".join(parts) if parts else "no note changes" |
| 261 | |
| 262 | logger.debug( |
| 263 | "✅ MIDI diff %r: +%d -%d notes (%d SES steps)", |
| 264 | file_path, |
| 265 | inserts, |
| 266 | deletes, |
| 267 | len(steps), |
| 268 | ) |
| 269 | |
| 270 | return StructuredDelta( |
| 271 | domain=_CHILD_DOMAIN, |
| 272 | ops=child_ops, |
| 273 | summary=child_summary, |
| 274 | ) |
| 275 | |
| 276 | # --------------------------------------------------------------------------- |
| 277 | # Entity-aware diff — wrapper that produces MutateOp for field-level mutations |
| 278 | # --------------------------------------------------------------------------- |
| 279 | |
| 280 | # --------------------------------------------------------------------------- |
| 281 | # MIDI reconstruction — inverse of extract_notes |
| 282 | # --------------------------------------------------------------------------- |
| 283 | |
| 284 | def reconstruct_midi( |
| 285 | notes: list[NoteKey], |
| 286 | *, |
| 287 | ticks_per_beat: int = 480, |
| 288 | ) -> bytes: |
| 289 | """Produce raw MIDI bytes from a list of :class:`NoteKey` objects. |
| 290 | |
| 291 | Creates a Type 0 (single-track) MIDI file. One ``note_on`` and one |
| 292 | ``note_off`` event are emitted per note. Events are sorted by absolute |
| 293 | tick time so the output is a valid MIDI stream regardless of the input |
| 294 | order. |
| 295 | |
| 296 | This is the inverse of :func:`extract_notes`. Used by |
| 297 | :func:`~muse.plugins.midi.plugin._merge_patch_ops` after the OT |
| 298 | engine has confirmed that two branches' note sequences commute, allowing |
| 299 | the merged note list to be materialised as actual MIDI bytes. |
| 300 | |
| 301 | Args: |
| 302 | notes: Note events to write. May be in any order; the |
| 303 | function sorts by ``start_tick`` before writing. |
| 304 | ticks_per_beat: Timing resolution. Preserve the base file's value so |
| 305 | that beat positions remain meaningful. |
| 306 | |
| 307 | Returns: |
| 308 | Raw MIDI bytes ready to be written to the object store. |
| 309 | """ |
| 310 | mid = mido.MidiFile(ticks_per_beat=ticks_per_beat, type=0) |
| 311 | track = mido.MidiTrack() |
| 312 | mid.tracks.append(track) |
| 313 | |
| 314 | # Build flat (abs_tick, note_on, channel, pitch, velocity) event tuples. |
| 315 | raw_events: list[tuple[int, bool, int, int, int]] = [] |
| 316 | for note in notes: |
| 317 | raw_events.append( |
| 318 | (note["start_tick"], True, note["channel"], note["pitch"], note["velocity"]) |
| 319 | ) |
| 320 | raw_events.append( |
| 321 | ( |
| 322 | note["start_tick"] + note["duration_ticks"], |
| 323 | False, |
| 324 | note["channel"], |
| 325 | note["pitch"], |
| 326 | 0, |
| 327 | ) |
| 328 | ) |
| 329 | |
| 330 | # Sort: by tick, with note_off (False) before note_on (True) at the same |
| 331 | # tick so that retriggered notes are handled correctly. |
| 332 | raw_events.sort(key=lambda e: (e[0], e[1])) |
| 333 | |
| 334 | prev_tick = 0 |
| 335 | for abs_tick, is_on, channel, pitch, velocity in raw_events: |
| 336 | delta = abs_tick - prev_tick |
| 337 | if is_on: |
| 338 | track.append( |
| 339 | mido.Message("note_on", channel=channel, note=pitch, velocity=velocity, time=delta) |
| 340 | ) |
| 341 | else: |
| 342 | track.append( |
| 343 | mido.Message("note_off", channel=channel, note=pitch, velocity=0, time=delta) |
| 344 | ) |
| 345 | prev_tick = abs_tick |
| 346 | |
| 347 | buf = io.BytesIO() |
| 348 | mid.save(file=buf) |
| 349 | return buf.getvalue() |
| 350 |
File History
2 commits
sha256:ac29b8ba9021514a03ab2782d92bf67671f0efa5b3b70d46f7598c5d4e923378
docs: record muse#74 Phase 4 live verification (SCR_13, SCR_14)
Sonnet 5
13 days ago
sha256:2562dffa0a0822ac1bdea854f9b267843c6ce95b497a9dc5c55837c80a3ebd0a
feat: domain_command_registry — Phase 1 of muse#74 (SCR_01-03)
Sonnet 4.6
patch
13 days ago