midi_diff.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 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 show`` 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 | from __future__ import annotations |
| 37 | |
| 38 | import hashlib |
| 39 | import io |
| 40 | import logging |
| 41 | from dataclasses import dataclass |
| 42 | from typing import TYPE_CHECKING, Literal, TypedDict |
| 43 | |
| 44 | if TYPE_CHECKING: |
| 45 | from muse.plugins.midi.entity import EntityIndex |
| 46 | |
| 47 | import mido |
| 48 | |
| 49 | from muse.core.diff_algorithms.lcs import myers_ses |
| 50 | from muse.domain import ( |
| 51 | DeleteOp, |
| 52 | DomainOp, |
| 53 | InsertOp, |
| 54 | StructuredDelta, |
| 55 | ) |
| 56 | |
| 57 | logger = logging.getLogger(__name__) |
| 58 | |
| 59 | #: Identifies the sub-domain for note-level operations inside a PatchOp. |
| 60 | _CHILD_DOMAIN = "midi_notes" |
| 61 | |
| 62 | _PITCH_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] |
| 63 | |
| 64 | |
| 65 | # --------------------------------------------------------------------------- |
| 66 | # NoteKey — the unit of LCS comparison |
| 67 | # --------------------------------------------------------------------------- |
| 68 | |
| 69 | |
| 70 | class NoteKey(TypedDict): |
| 71 | """Fully-specified MIDI note used as the LCS comparison unit. |
| 72 | |
| 73 | Two notes are considered identical in LCS iff all five fields match. |
| 74 | A pitch change, velocity change, timing shift, or channel change |
| 75 | counts as a delete of the old note and an insert of the new one. |
| 76 | This is conservative but correct — it means the LCS finds true |
| 77 | structural matches and surfaces real musical changes. |
| 78 | """ |
| 79 | |
| 80 | pitch: int |
| 81 | velocity: int |
| 82 | start_tick: int |
| 83 | duration_ticks: int |
| 84 | channel: int |
| 85 | |
| 86 | |
| 87 | # --------------------------------------------------------------------------- |
| 88 | # Helpers |
| 89 | # --------------------------------------------------------------------------- |
| 90 | |
| 91 | |
| 92 | def _pitch_name(midi_pitch: int) -> str: |
| 93 | """Return a human-readable pitch string, e.g. ``"C4"``, ``"F#5"``.""" |
| 94 | octave = midi_pitch // 12 - 1 |
| 95 | name = _PITCH_NAMES[midi_pitch % 12] |
| 96 | return f"{name}{octave}" |
| 97 | |
| 98 | |
| 99 | def _note_content_id(note: NoteKey) -> str: |
| 100 | """Return a deterministic SHA-256 for a note's five identity fields. |
| 101 | |
| 102 | This gives a stable ``content_id`` for use in ``InsertOp`` / ``DeleteOp`` |
| 103 | without requiring the note to be stored as a separate blob in the object |
| 104 | store. The hash uniquely identifies "this specific note event". |
| 105 | """ |
| 106 | payload = ( |
| 107 | f"{note['pitch']}:{note['velocity']}:" |
| 108 | f"{note['start_tick']}:{note['duration_ticks']}:{note['channel']}" |
| 109 | ) |
| 110 | return hashlib.sha256(payload.encode()).hexdigest() |
| 111 | |
| 112 | |
| 113 | def _note_summary(note: NoteKey, ticks_per_beat: int) -> str: |
| 114 | """Return a human-readable one-liner for a note, e.g. ``"C4 vel=80 @beat=1.00"``.""" |
| 115 | beat = note["start_tick"] / max(ticks_per_beat, 1) |
| 116 | dur = note["duration_ticks"] / max(ticks_per_beat, 1) |
| 117 | return ( |
| 118 | f"{_pitch_name(note['pitch'])} " |
| 119 | f"vel={note['velocity']} " |
| 120 | f"@beat={beat:.2f} " |
| 121 | f"dur={dur:.2f}" |
| 122 | ) |
| 123 | |
| 124 | |
| 125 | # --------------------------------------------------------------------------- |
| 126 | # Note extraction |
| 127 | # --------------------------------------------------------------------------- |
| 128 | |
| 129 | |
| 130 | def extract_notes(midi_bytes: bytes) -> tuple[list[NoteKey], int]: |
| 131 | """Parse *midi_bytes* and return ``(notes, ticks_per_beat)``. |
| 132 | |
| 133 | Notes are paired note_on / note_off events. A note_on with velocity=0 |
| 134 | is treated as note_off. Notes are sorted by start_tick then pitch for |
| 135 | deterministic ordering. |
| 136 | |
| 137 | Args: |
| 138 | midi_bytes: Raw bytes of a ``.mid`` file. |
| 139 | |
| 140 | Returns: |
| 141 | A tuple of (sorted NoteKey list, ticks_per_beat integer). |
| 142 | |
| 143 | Raises: |
| 144 | ValueError: When *midi_bytes* cannot be parsed as a MIDI file. |
| 145 | """ |
| 146 | try: |
| 147 | mid = mido.MidiFile(file=io.BytesIO(midi_bytes)) |
| 148 | except Exception as exc: |
| 149 | raise ValueError(f"Cannot parse MIDI bytes: {exc}") from exc |
| 150 | |
| 151 | ticks_per_beat: int = int(mid.ticks_per_beat) |
| 152 | # (channel, pitch) → (start_tick, velocity) |
| 153 | active: dict[tuple[int, int], tuple[int, int]] = {} |
| 154 | notes: list[NoteKey] = [] |
| 155 | |
| 156 | for track in mid.tracks: |
| 157 | abs_tick = 0 |
| 158 | for msg in track: |
| 159 | abs_tick += msg.time |
| 160 | if msg.type == "note_on" and msg.velocity > 0: |
| 161 | active[(msg.channel, msg.note)] = (abs_tick, msg.velocity) |
| 162 | elif msg.type == "note_off" or ( |
| 163 | msg.type == "note_on" and msg.velocity == 0 |
| 164 | ): |
| 165 | key = (msg.channel, msg.note) |
| 166 | if key in active: |
| 167 | start, vel = active.pop(key) |
| 168 | notes.append( |
| 169 | NoteKey( |
| 170 | pitch=msg.note, |
| 171 | velocity=vel, |
| 172 | start_tick=start, |
| 173 | duration_ticks=max(abs_tick - start, 1), |
| 174 | channel=msg.channel, |
| 175 | ) |
| 176 | ) |
| 177 | |
| 178 | # Close any notes still open at end of file with duration 1. |
| 179 | for (ch, pitch), (start, vel) in active.items(): |
| 180 | notes.append( |
| 181 | NoteKey( |
| 182 | pitch=pitch, |
| 183 | velocity=vel, |
| 184 | start_tick=start, |
| 185 | duration_ticks=1, |
| 186 | channel=ch, |
| 187 | ) |
| 188 | ) |
| 189 | |
| 190 | notes.sort(key=lambda n: (n["start_tick"], n["pitch"], n["channel"])) |
| 191 | return notes, ticks_per_beat |
| 192 | |
| 193 | |
| 194 | # --------------------------------------------------------------------------- |
| 195 | # NoteKey-level edit script — adapter over the core LCS |
| 196 | # --------------------------------------------------------------------------- |
| 197 | |
| 198 | EditKind = Literal["keep", "insert", "delete"] |
| 199 | |
| 200 | |
| 201 | @dataclass(frozen=True) |
| 202 | class EditStep: |
| 203 | """One step in the note-level edit script produced by :func:`lcs_edit_script`.""" |
| 204 | |
| 205 | kind: EditKind |
| 206 | base_index: int |
| 207 | target_index: int |
| 208 | note: NoteKey |
| 209 | |
| 210 | |
| 211 | def lcs_edit_script( |
| 212 | base: list[NoteKey], |
| 213 | target: list[NoteKey], |
| 214 | ) -> list[EditStep]: |
| 215 | """Compute the shortest edit script transforming *base* into *target*. |
| 216 | |
| 217 | Converts each ``NoteKey`` to its content ID, delegates to |
| 218 | :func:`~muse.core.diff_algorithms.lcs.myers_ses` for the SES, then maps |
| 219 | the result back to :class:`EditStep` entries carrying the original |
| 220 | ``NoteKey`` values. |
| 221 | |
| 222 | Two notes are matched iff all five ``NoteKey`` fields are equal. This is |
| 223 | correct: a pitch change, velocity change, or timing shift is a delete of |
| 224 | the old note and an insert of the new one. |
| 225 | |
| 226 | Args: |
| 227 | base: The base (ancestor) note sequence. |
| 228 | target: The target (newer) note sequence. |
| 229 | |
| 230 | Returns: |
| 231 | A list of :class:`EditStep` entries (keep / insert / delete). |
| 232 | """ |
| 233 | base_ids = [_note_content_id(n) for n in base] |
| 234 | target_ids = [_note_content_id(n) for n in target] |
| 235 | raw_steps = myers_ses(base_ids, target_ids) |
| 236 | |
| 237 | result: list[EditStep] = [] |
| 238 | for step in raw_steps: |
| 239 | if step.kind == "keep": |
| 240 | result.append(EditStep("keep", step.base_index, step.target_index, base[step.base_index])) |
| 241 | elif step.kind == "insert": |
| 242 | result.append(EditStep("insert", step.base_index, step.target_index, target[step.target_index])) |
| 243 | else: |
| 244 | result.append(EditStep("delete", step.base_index, step.target_index, base[step.base_index])) |
| 245 | return result |
| 246 | |
| 247 | |
| 248 | # --------------------------------------------------------------------------- |
| 249 | # Public diff entry point |
| 250 | # --------------------------------------------------------------------------- |
| 251 | |
| 252 | |
| 253 | def diff_midi_notes( |
| 254 | base_bytes: bytes, |
| 255 | target_bytes: bytes, |
| 256 | *, |
| 257 | file_path: str = "", |
| 258 | ) -> StructuredDelta: |
| 259 | """Compute a note-level ``StructuredDelta`` between two MIDI files. |
| 260 | |
| 261 | Parses both files, converts each note to its content ID, delegates to the |
| 262 | core :func:`~muse.core.diff_algorithms.lcs.myers_ses` for the SES, then |
| 263 | maps the edit steps to typed ``DomainOp`` instances. |
| 264 | |
| 265 | Args: |
| 266 | base_bytes: Raw bytes of the base (ancestor) MIDI file. |
| 267 | target_bytes: Raw bytes of the target (newer) MIDI file. |
| 268 | file_path: Workspace-relative path of the file being diffed (used |
| 269 | only in log messages and ``content_summary`` strings). |
| 270 | |
| 271 | Returns: |
| 272 | A ``StructuredDelta`` with ``InsertOp`` and ``DeleteOp`` entries for |
| 273 | each note added or removed. The ``summary`` field is human-readable, |
| 274 | e.g. ``"3 notes added, 1 note removed"``. |
| 275 | |
| 276 | Raises: |
| 277 | ValueError: When either byte string cannot be parsed as MIDI. |
| 278 | """ |
| 279 | base_notes, base_tpb = extract_notes(base_bytes) |
| 280 | target_notes, target_tpb = extract_notes(target_bytes) |
| 281 | tpb = base_tpb # use base ticks_per_beat for human-readable summaries |
| 282 | |
| 283 | # Convert NoteKey → content ID, then delegate LCS to the core algorithm. |
| 284 | base_ids = [_note_content_id(n) for n in base_notes] |
| 285 | target_ids = [_note_content_id(n) for n in target_notes] |
| 286 | steps = myers_ses(base_ids, target_ids) |
| 287 | |
| 288 | # Build a content-ID → NoteKey lookup so we can produce rich summaries. |
| 289 | base_by_id = {_note_content_id(n): n for n in base_notes} |
| 290 | target_by_id = {_note_content_id(n): n for n in target_notes} |
| 291 | |
| 292 | child_ops: list[DomainOp] = [] |
| 293 | inserts = 0 |
| 294 | deletes = 0 |
| 295 | |
| 296 | for step in steps: |
| 297 | if step.kind == "insert": |
| 298 | note = target_by_id.get(step.item) |
| 299 | summary = _note_summary(note, tpb) if note else step.item[:12] |
| 300 | child_ops.append( |
| 301 | InsertOp( |
| 302 | op="insert", |
| 303 | address=f"note:{step.target_index}", |
| 304 | position=step.target_index, |
| 305 | content_id=step.item, |
| 306 | content_summary=summary, |
| 307 | ) |
| 308 | ) |
| 309 | inserts += 1 |
| 310 | elif step.kind == "delete": |
| 311 | note = base_by_id.get(step.item) |
| 312 | summary = _note_summary(note, tpb) if note else step.item[:12] |
| 313 | child_ops.append( |
| 314 | DeleteOp( |
| 315 | op="delete", |
| 316 | address=f"note:{step.base_index}", |
| 317 | position=step.base_index, |
| 318 | content_id=step.item, |
| 319 | content_summary=summary, |
| 320 | ) |
| 321 | ) |
| 322 | deletes += 1 |
| 323 | # "keep" steps produce no ops — the note is unchanged. |
| 324 | |
| 325 | parts: list[str] = [] |
| 326 | if inserts: |
| 327 | parts.append(f"{inserts} note{'s' if inserts != 1 else ''} added") |
| 328 | if deletes: |
| 329 | parts.append(f"{deletes} note{'s' if deletes != 1 else ''} removed") |
| 330 | child_summary = ", ".join(parts) if parts else "no note changes" |
| 331 | |
| 332 | logger.debug( |
| 333 | "✅ MIDI diff %r: +%d -%d notes (%d SES steps)", |
| 334 | file_path, |
| 335 | inserts, |
| 336 | deletes, |
| 337 | len(steps), |
| 338 | ) |
| 339 | |
| 340 | return StructuredDelta( |
| 341 | domain=_CHILD_DOMAIN, |
| 342 | ops=child_ops, |
| 343 | summary=child_summary, |
| 344 | ) |
| 345 | |
| 346 | |
| 347 | # --------------------------------------------------------------------------- |
| 348 | # Entity-aware diff — wrapper that produces MutateOp for field-level mutations |
| 349 | # --------------------------------------------------------------------------- |
| 350 | |
| 351 | |
| 352 | |
| 353 | # --------------------------------------------------------------------------- |
| 354 | # MIDI reconstruction — inverse of extract_notes |
| 355 | # --------------------------------------------------------------------------- |
| 356 | |
| 357 | |
| 358 | def reconstruct_midi( |
| 359 | notes: list[NoteKey], |
| 360 | *, |
| 361 | ticks_per_beat: int = 480, |
| 362 | ) -> bytes: |
| 363 | """Produce raw MIDI bytes from a list of :class:`NoteKey` objects. |
| 364 | |
| 365 | Creates a Type 0 (single-track) MIDI file. One ``note_on`` and one |
| 366 | ``note_off`` event are emitted per note. Events are sorted by absolute |
| 367 | tick time so the output is a valid MIDI stream regardless of the input |
| 368 | order. |
| 369 | |
| 370 | This is the inverse of :func:`extract_notes`. Used by |
| 371 | :func:`~muse.plugins.midi.plugin._merge_patch_ops` after the OT |
| 372 | engine has confirmed that two branches' note sequences commute, allowing |
| 373 | the merged note list to be materialised as actual MIDI bytes. |
| 374 | |
| 375 | Args: |
| 376 | notes: Note events to write. May be in any order; the |
| 377 | function sorts by ``start_tick`` before writing. |
| 378 | ticks_per_beat: Timing resolution. Preserve the base file's value so |
| 379 | that beat positions remain meaningful. |
| 380 | |
| 381 | Returns: |
| 382 | Raw MIDI bytes ready to be written to the object store. |
| 383 | """ |
| 384 | mid = mido.MidiFile(ticks_per_beat=ticks_per_beat, type=0) |
| 385 | track = mido.MidiTrack() |
| 386 | mid.tracks.append(track) |
| 387 | |
| 388 | # Build flat (abs_tick, note_on, channel, pitch, velocity) event tuples. |
| 389 | raw_events: list[tuple[int, bool, int, int, int]] = [] |
| 390 | for note in notes: |
| 391 | raw_events.append( |
| 392 | (note["start_tick"], True, note["channel"], note["pitch"], note["velocity"]) |
| 393 | ) |
| 394 | raw_events.append( |
| 395 | ( |
| 396 | note["start_tick"] + note["duration_ticks"], |
| 397 | False, |
| 398 | note["channel"], |
| 399 | note["pitch"], |
| 400 | 0, |
| 401 | ) |
| 402 | ) |
| 403 | |
| 404 | # Sort: by tick, with note_off (False) before note_on (True) at the same |
| 405 | # tick so that retriggered notes are handled correctly. |
| 406 | raw_events.sort(key=lambda e: (e[0], e[1])) |
| 407 | |
| 408 | prev_tick = 0 |
| 409 | for abs_tick, is_on, channel, pitch, velocity in raw_events: |
| 410 | delta = abs_tick - prev_tick |
| 411 | if is_on: |
| 412 | track.append( |
| 413 | mido.Message("note_on", channel=channel, note=pitch, velocity=velocity, time=delta) |
| 414 | ) |
| 415 | else: |
| 416 | track.append( |
| 417 | mido.Message("note_off", channel=channel, note=pitch, velocity=0, time=delta) |
| 418 | ) |
| 419 | prev_tick = abs_tick |
| 420 | |
| 421 | buf = io.BytesIO() |
| 422 | mid.save(file=buf) |
| 423 | return buf.getvalue() |
| 424 | |
| 425 |
File History
5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e
Merge branch 'dev' into main
Human
29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce
Merge branch 'dev' into main
Human
48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa
feat: Muse — version control for the agent era
Human
100 days ago