gabriel / muse public
midi_merge.py python
588 lines 22.7 KB
Raw
9ee9c39c refactor: rename music→midi domain, strip all 5-dim backward compat Gabriel Cardona <gabriel@tellurstori.com> 5d ago
1 """MIDI dimension-aware merge for the Muse music plugin.
2
3 This module implements the multidimensional merge that makes Muse meaningfully
4 different from git. Git treats every file as an opaque byte sequence: any
5 two-branch change to the same file is a conflict. Muse understands that a
6 MIDI file has *independent orthogonal axes*, and two collaborators can touch
7 different axes of the same file without conflicting.
8
9 Dimensions
10 ----------
11
12 MIDI carries far more independent axes than a naive "notes vs. everything else"
13 split. The full dimension taxonomy maps every MIDI event type to exactly one
14 internal bucket:
15
16 +----------------------+--------------------------------------------------+
17 | Internal dimension | MIDI event types / CC numbers |
18 +======================+==================================================+
19 | ``notes`` | ``note_on`` / ``note_off`` |
20 +----------------------+--------------------------------------------------+
21 | ``pitch_bend`` | ``pitchwheel`` |
22 +----------------------+--------------------------------------------------+
23 | ``channel_pressure`` | ``aftertouch`` (mono channel pressure) |
24 +----------------------+--------------------------------------------------+
25 | ``poly_pressure`` | ``polytouch`` (per-note polyphonic aftertouch) |
26 +----------------------+--------------------------------------------------+
27 | ``cc_modulation`` | CC 1 — modulation wheel |
28 +----------------------+--------------------------------------------------+
29 | ``cc_volume`` | CC 7 — channel volume |
30 +----------------------+--------------------------------------------------+
31 | ``cc_pan`` | CC 10 — stereo pan |
32 +----------------------+--------------------------------------------------+
33 | ``cc_expression`` | CC 11 — expression controller |
34 +----------------------+--------------------------------------------------+
35 | ``cc_sustain`` | CC 64 — damper / sustain pedal |
36 +----------------------+--------------------------------------------------+
37 | ``cc_sostenuto`` | CC 66 — sostenuto pedal |
38 +----------------------+--------------------------------------------------+
39 | ``cc_soft_pedal`` | CC 67 — soft pedal (una corda) |
40 +----------------------+--------------------------------------------------+
41 | ``cc_portamento`` | CC 65 — portamento on/off |
42 +----------------------+--------------------------------------------------+
43 | ``cc_reverb`` | CC 91 — reverb send level |
44 +----------------------+--------------------------------------------------+
45 | ``cc_chorus`` | CC 93 — chorus send level |
46 +----------------------+--------------------------------------------------+
47 | ``cc_other`` | All other CC events (numbered controllers) |
48 +----------------------+--------------------------------------------------+
49 | ``program_change`` | ``program_change`` (patch / instrument select) |
50 +----------------------+--------------------------------------------------+
51 | ``tempo_map`` | ``set_tempo`` meta events |
52 +----------------------+--------------------------------------------------+
53 | ``time_signatures`` | ``time_signature`` meta events |
54 +----------------------+--------------------------------------------------+
55 | ``key_signatures`` | ``key_signature`` meta events |
56 +----------------------+--------------------------------------------------+
57 | ``markers`` | ``marker``, ``cue_marker``, ``text``, |
58 | | ``lyrics``, ``copyright`` meta events |
59 +----------------------+--------------------------------------------------+
60 | ``track_structure`` | ``track_name``, ``instrument_name``, ``sysex``, |
61 | | ``sequencer_specific`` and unknown meta events |
62 +----------------------+--------------------------------------------------+
63
64 Why fine-grained dimensions matter
65 -----------------------------------
66 With the old 4-bucket model, changing sustain pedal (CC64) and changing channel
67 volume (CC7) were the same dimension: they always conflicted. With 21 internal
68 dimensions they are independent — two agents can edit different aspects of the
69 same MIDI file without ever conflicting.
70
71 Independence rules
72 ------------------
73 - **Independent** (``independent_merge=True``): notes, pitch_bend, all CC
74 dimensions, channel_pressure, poly_pressure, program_change, key_signatures,
75 markers. Conflicts in these dimensions never block merging others.
76 - **Non-independent** (``independent_merge=False``): tempo_map, time_signatures,
77 track_structure. A conflict here blocks merging other dimensions until
78 resolved, because a tempo change shifts the musical meaning of every subsequent
79 tick position, and track structure changes affect routing.
80
81 Merge algorithm
82 ---------------
83 1. Parse ``base``, ``left``, and ``right`` MIDI bytes into event streams.
84 2. Convert to absolute-tick representation and bucket by dimension.
85 3. Hash each bucket; compare ``base ↔ left`` and ``base ↔ right`` to detect
86 per-dimension changes.
87 4. For each dimension apply the winning side determined by ``.museattributes``
88 strategy (or the standard one-sided-change rule when no conflict exists).
89 5. Reconstruct a valid MIDI file by merging winning dimension slices, sorting
90 by absolute tick, converting back to delta-time, and writing to bytes.
91
92 Public API
93 ----------
94 - :func:`extract_dimensions` — parse MIDI bytes → ``MidiDimensions``
95 - :func:`merge_midi_dimensions` — three-way dimension merge → bytes or ``None``
96 - :func:`dimension_conflict_detail` — per-dimension change report for logging
97 - :data:`INTERNAL_DIMS` — ordered list of all internal dimension names
98 - :data:`DIM_ALIAS` — user-facing ``.museattributes`` name → internal bucket
99 - :data:`NON_INDEPENDENT_DIMS` — dimensions that block others on conflict
100 """
101 from __future__ import annotations
102
103 import hashlib
104 import io
105 import json
106 from dataclasses import dataclass, field
107
108 import mido
109
110 from muse.core.attributes import AttributeRule, resolve_strategy
111
112 # ---------------------------------------------------------------------------
113 # Dimension constants — the complete MIDI dimension taxonomy
114 # ---------------------------------------------------------------------------
115
116 #: Internal dimension names, ordered canonically.
117 #: Each MIDI event type maps to exactly one of these buckets.
118 INTERNAL_DIMS: list[str] = [
119 # --- Expressive note content ---
120 "notes", # note_on / note_off
121 "pitch_bend", # pitchwheel
122 "channel_pressure", # aftertouch (mono)
123 "poly_pressure", # polytouch (per-note)
124 # --- Named CC controllers (individually mergeable) ---
125 "cc_modulation", # CC 1
126 "cc_volume", # CC 7
127 "cc_pan", # CC 10
128 "cc_expression", # CC 11
129 "cc_sustain", # CC 64
130 "cc_portamento", # CC 65
131 "cc_sostenuto", # CC 66
132 "cc_soft_pedal", # CC 67
133 "cc_reverb", # CC 91
134 "cc_chorus", # CC 93
135 "cc_other", # all remaining CC numbers
136 # --- Patch / program selection ---
137 "program_change",
138 # --- Timeline / notation metadata (non-independent) ---
139 "tempo_map", # set_tempo — non-independent: affects all tick positions
140 "time_signatures", # time_signature — non-independent: affects bar structure
141 # --- Tonal context and notation ---
142 "key_signatures", # key_signature
143 "markers", # marker, cue_marker, text, lyrics, copyright
144 # --- Track structure (non-independent) ---
145 "track_structure", # track_name, instrument_name, sysex, unknown meta
146 ]
147
148 #: Dimensions whose conflicts block merging all other dimensions until resolved.
149 #: All other dimensions are merged in parallel regardless of conflicts here.
150 NON_INDEPENDENT_DIMS: frozenset[str] = frozenset({
151 "tempo_map",
152 "time_signatures",
153 "track_structure",
154 })
155
156 #: User-facing dimension names from .museattributes mapped to internal buckets.
157 #: Agents and humans use these names in merge strategy declarations.
158 DIM_ALIAS: dict[str, str] = {
159 "pitch_bend": "pitch_bend",
160 "aftertouch": "channel_pressure",
161 "poly_aftertouch": "poly_pressure",
162 "modulation": "cc_modulation",
163 "volume": "cc_volume",
164 "pan": "cc_pan",
165 "expression": "cc_expression",
166 "sustain": "cc_sustain",
167 "portamento": "cc_portamento",
168 "sostenuto": "cc_sostenuto",
169 "soft_pedal": "cc_soft_pedal",
170 "reverb": "cc_reverb",
171 "chorus": "cc_chorus",
172 "automation": "cc_other",
173 "program": "program_change",
174 "tempo": "tempo_map",
175 "time_sig": "time_signatures",
176 "key_sig": "key_signatures",
177 "markers": "markers",
178 "track_structure": "track_structure",
179 }
180
181 #: All valid names (aliases + internal) → internal bucket.
182 _CANONICAL: dict[str, str] = {**DIM_ALIAS, **{d: d for d in INTERNAL_DIMS}}
183
184 #: CC number → internal dimension name for named controllers.
185 _CC_DIM: dict[int, str] = {
186 1: "cc_modulation",
187 7: "cc_volume",
188 10: "cc_pan",
189 11: "cc_expression",
190 64: "cc_sustain",
191 65: "cc_portamento",
192 66: "cc_sostenuto",
193 67: "cc_soft_pedal",
194 91: "cc_reverb",
195 93: "cc_chorus",
196 }
197
198
199 # ---------------------------------------------------------------------------
200 # Data types
201 # ---------------------------------------------------------------------------
202
203
204 @dataclass
205 class DimensionSlice:
206 """Events belonging to one dimension of a MIDI file.
207
208 ``events`` is a list of ``(abs_tick, mido.Message)`` pairs sorted by
209 ascending absolute tick. ``content_hash`` is the SHA-256 digest of the
210 canonical JSON serialisation of the event list (used for change detection
211 without loading file bytes).
212 """
213
214 name: str
215 events: list[tuple[int, mido.Message]] = field(default_factory=list)
216 content_hash: str = ""
217
218 def __post_init__(self) -> None:
219 if not self.content_hash:
220 self.content_hash = _hash_events(self.events)
221
222
223 @dataclass
224 class MidiDimensions:
225 """All dimension slices extracted from one MIDI file.
226
227 ``slices`` maps internal dimension name → :class:`DimensionSlice`.
228 Every internal dimension in :data:`INTERNAL_DIMS` has an entry, even if
229 the corresponding event list is empty (hash of empty list is stable).
230 """
231
232 ticks_per_beat: int
233 file_type: int
234 slices: dict[str, DimensionSlice]
235
236 def get(self, dim: str) -> DimensionSlice:
237 """Return the slice for a user-facing or internal dimension name."""
238 internal = _CANONICAL.get(dim, dim)
239 return self.slices[internal]
240
241
242 # ---------------------------------------------------------------------------
243 # Internal helpers
244 # ---------------------------------------------------------------------------
245
246
247 def _classify_event(msg: mido.Message) -> str | None:
248 """Map a mido Message to an internal dimension bucket.
249
250 Returns ``None`` for events that should be excluded from all buckets
251 (e.g. ``end_of_track`` is handled during reconstruction, not stored here).
252 Unknown messages that are meta events fall back to ``"track_structure"``.
253 True unknowns (no ``is_meta`` attribute) are discarded.
254 """
255 t = msg.type
256
257 # --- Note events ---
258 if t in ("note_on", "note_off"):
259 return "notes"
260
261 # --- Pitch / pressure ---
262 if t == "pitchwheel":
263 return "pitch_bend"
264 if t == "aftertouch":
265 return "channel_pressure"
266 if t == "polytouch":
267 return "poly_pressure"
268
269 # --- CC — split by controller number ---
270 if t == "control_change":
271 return _CC_DIM.get(msg.control, "cc_other")
272
273 # --- Program change ---
274 if t == "program_change":
275 return "program_change"
276
277 # --- Timeline metadata ---
278 if t == "set_tempo":
279 return "tempo_map"
280 if t == "time_signature":
281 return "time_signatures"
282 if t == "key_signature":
283 return "key_signatures"
284
285 # --- Section markers and text annotations ---
286 if t in ("marker", "cue_marker", "text", "lyrics", "copyright"):
287 return "markers"
288
289 # --- Track structure and routing ---
290 if t in ("track_name", "instrument_name", "sysex", "sequencer_specific"):
291 return "track_structure"
292
293 # --- End-of-track is reconstructed, not stored ---
294 if t == "end_of_track":
295 return None
296
297 # --- Unknown meta events → track structure (safe default) ---
298 if getattr(msg, "is_meta", False):
299 return "track_structure"
300
301 return None
302
303
304 type _MsgVal = int | str | list[int]
305
306
307 def _msg_to_dict(msg: mido.Message) -> dict[str, _MsgVal]:
308 """Serialise a mido Message to a JSON-compatible dict."""
309 d: dict[str, _MsgVal] = {"type": msg.type}
310 for attr in (
311 "channel", "note", "velocity", "control", "value",
312 "pitch", "program", "numerator", "denominator",
313 "clocks_per_click", "notated_32nd_notes_per_beat",
314 "tempo", "key", "scale", "text", "data",
315 ):
316 if hasattr(msg, attr):
317 raw = getattr(msg, attr)
318 if isinstance(raw, (bytes, bytearray)):
319 d[attr] = list(raw)
320 elif isinstance(raw, str):
321 d[attr] = raw
322 elif isinstance(raw, int):
323 d[attr] = raw
324 return d
325
326
327 def _hash_events(events: list[tuple[int, mido.Message]]) -> str:
328 """SHA-256 of the canonical JSON representation of an event list."""
329 payload = json.dumps(
330 [(tick, _msg_to_dict(msg)) for tick, msg in events],
331 sort_keys=True,
332 separators=(",", ":"),
333 ).encode()
334 return hashlib.sha256(payload).hexdigest()
335
336
337 def _to_absolute(track: mido.MidiTrack) -> list[tuple[int, mido.Message]]:
338 """Convert a delta-time track to a list of ``(abs_tick, msg)`` pairs."""
339 result: list[tuple[int, mido.Message]] = []
340 abs_tick = 0
341 for msg in track:
342 abs_tick += msg.time
343 result.append((abs_tick, msg))
344 return result
345
346
347 # ---------------------------------------------------------------------------
348 # Public: extract_dimensions
349 # ---------------------------------------------------------------------------
350
351
352 def extract_dimensions(midi_bytes: bytes) -> MidiDimensions:
353 """Parse *midi_bytes* and bucket events by dimension.
354
355 Every event type in the MIDI spec maps to exactly one of the
356 :data:`INTERNAL_DIMS` buckets. Empty buckets are present with an empty
357 event list so that callers can always index by dimension name.
358
359 Args:
360 midi_bytes: Raw bytes of a ``.mid`` file.
361
362 Returns:
363 A :class:`MidiDimensions` with one :class:`DimensionSlice` per
364 internal dimension. Events within each slice are sorted by ascending
365 absolute tick, then by event type for determinism when multiple events
366 share the same tick.
367
368 Raises:
369 ValueError: If *midi_bytes* cannot be parsed as a MIDI file.
370 """
371 try:
372 mid = mido.MidiFile(file=io.BytesIO(midi_bytes))
373 except Exception as exc:
374 raise ValueError(f"Failed to parse MIDI data: {exc}") from exc
375
376 buckets: dict[str, list[tuple[int, mido.Message]]] = {
377 dim: [] for dim in INTERNAL_DIMS
378 }
379
380 for track in mid.tracks:
381 for abs_tick, msg in _to_absolute(track):
382 bucket = _classify_event(msg)
383 if bucket is not None:
384 buckets[bucket].append((abs_tick, msg))
385
386 for dim in INTERNAL_DIMS:
387 buckets[dim].sort(key=lambda x: (x[0], x[1].type))
388
389 slices = {
390 dim: DimensionSlice(name=dim, events=events)
391 for dim, events in buckets.items()
392 }
393 return MidiDimensions(
394 ticks_per_beat=mid.ticks_per_beat,
395 file_type=mid.type,
396 slices=slices,
397 )
398
399
400 # ---------------------------------------------------------------------------
401 # Public: dimension_conflict_detail
402 # ---------------------------------------------------------------------------
403
404
405 def dimension_conflict_detail(
406 base: MidiDimensions,
407 left: MidiDimensions,
408 right: MidiDimensions,
409 ) -> dict[str, str]:
410 """Return a per-dimension change report for a conflicting file.
411
412 Returns a dict mapping internal dimension name to one of:
413
414 - ``"unchanged"`` — neither side changed this dimension.
415 - ``"left_only"`` — only the left (ours) side changed.
416 - ``"right_only"`` — only the right (theirs) side changed.
417 - ``"both"`` — both sides changed; a dimension-level conflict.
418
419 This is used by :func:`merge_midi_dimensions` and surfaced in
420 ``muse merge`` output for human-readable conflict diagnostics.
421 """
422 report: dict[str, str] = {}
423 for dim in INTERNAL_DIMS:
424 base_hash = base.slices[dim].content_hash
425 left_hash = left.slices[dim].content_hash
426 right_hash = right.slices[dim].content_hash
427 left_changed = base_hash != left_hash
428 right_changed = base_hash != right_hash
429 if left_changed and right_changed:
430 report[dim] = "both"
431 elif left_changed:
432 report[dim] = "left_only"
433 elif right_changed:
434 report[dim] = "right_only"
435 else:
436 report[dim] = "unchanged"
437 return report
438
439
440 # ---------------------------------------------------------------------------
441 # Reconstruction helpers
442 # ---------------------------------------------------------------------------
443
444
445 def _events_to_track(
446 events: list[tuple[int, mido.Message]],
447 ) -> mido.MidiTrack:
448 """Convert absolute-tick events to a mido MidiTrack with delta times."""
449 track = mido.MidiTrack()
450 prev_tick = 0
451 for abs_tick, msg in sorted(events, key=lambda x: (x[0], x[1].type)):
452 delta = abs_tick - prev_tick
453 new_msg = msg.copy(time=delta)
454 track.append(new_msg)
455 prev_tick = abs_tick
456 if not track or track[-1].type != "end_of_track":
457 track.append(mido.MetaMessage("end_of_track", time=0))
458 return track
459
460
461 def _reconstruct(
462 ticks_per_beat: int,
463 winning_slices: dict[str, list[tuple[int, mido.Message]]],
464 ) -> bytes:
465 """Build a type-0 MIDI file from winning dimension event lists.
466
467 All dimension events are merged into a single track (type-0) for
468 maximum compatibility. The absolute-tick ordering is preserved and
469 duplicate end_of_track messages are removed.
470 """
471 all_events: list[tuple[int, mido.Message]] = []
472 for events in winning_slices.values():
473 all_events.extend(events)
474
475 all_events = [
476 (tick, msg) for tick, msg in all_events
477 if msg.type != "end_of_track"
478 ]
479 all_events.sort(key=lambda x: (x[0], x[1].type))
480
481 track = _events_to_track(all_events)
482 mid = mido.MidiFile(type=0, ticks_per_beat=ticks_per_beat)
483 mid.tracks.append(track)
484
485 buf = io.BytesIO()
486 mid.save(file=buf)
487 return buf.getvalue()
488
489
490 # ---------------------------------------------------------------------------
491 # Public: merge_midi_dimensions
492 # ---------------------------------------------------------------------------
493
494
495 def merge_midi_dimensions(
496 base_bytes: bytes,
497 left_bytes: bytes,
498 right_bytes: bytes,
499 attrs_rules: list[AttributeRule],
500 path: str,
501 ) -> tuple[bytes, dict[str, str]] | None:
502 """Attempt a dimension-level three-way merge of a MIDI file.
503
504 For each internal dimension (all 21 of them):
505
506 - If neither side changed → keep base.
507 - If only one side changed → take that side (clean auto-merge).
508 - If both sides changed → consult ``.museattributes`` strategy:
509
510 * ``ours`` / ``theirs`` → take the specified side; record in report.
511 * ``manual`` / ``auto`` / ``union`` → unresolvable; return ``None``.
512
513 Non-independent dimensions (``tempo_map``, ``time_signatures``,
514 ``track_structure``) that have bilateral conflicts cause an immediate
515 ``None`` return — their conflicts cannot be auto-resolved because they
516 affect the semantic meaning of all other dimensions.
517
518 Args:
519 base_bytes: MIDI bytes for the common ancestor.
520 left_bytes: MIDI bytes for the ours (left) branch.
521 right_bytes: MIDI bytes for the theirs (right) branch.
522 attrs_rules: Rule list from :func:`muse.core.attributes.load_attributes`.
523 path: Workspace-relative POSIX path (used for strategy lookup).
524
525 Returns:
526 A ``(merged_bytes, dimension_report)`` tuple when all dimension
527 conflicts can be resolved, or ``None`` when at least one dimension
528 conflict has no resolvable strategy.
529
530 *dimension_report* maps each internal dimension name to the side
531 chosen: ``"base"``, ``"left"``, ``"right"``, or the strategy string.
532 Only dimensions with non-empty event lists or conflicts are included.
533
534 Raises:
535 ValueError: If any of the byte strings cannot be parsed as MIDI.
536 """
537 base_dims = extract_dimensions(base_bytes)
538 left_dims = extract_dimensions(left_bytes)
539 right_dims = extract_dimensions(right_bytes)
540
541 detail = dimension_conflict_detail(base_dims, left_dims, right_dims)
542
543 winning_slices: dict[str, list[tuple[int, mido.Message]]] = {}
544 dimension_report: dict[str, str] = {}
545
546 for dim in INTERNAL_DIMS:
547 change = detail[dim]
548
549 if change == "unchanged":
550 winning_slices[dim] = base_dims.slices[dim].events
551 if base_dims.slices[dim].events:
552 dimension_report[dim] = "base"
553
554 elif change == "left_only":
555 winning_slices[dim] = left_dims.slices[dim].events
556 dimension_report[dim] = "left"
557
558 elif change == "right_only":
559 winning_slices[dim] = right_dims.slices[dim].events
560 dimension_report[dim] = "right"
561
562 else:
563 # Both sides changed — resolve via .museattributes strategy.
564 # Look up by user-facing aliases first, then internal name.
565 user_dim_names = [k for k, v in DIM_ALIAS.items() if v == dim]
566 user_dim_names.append(dim) # internal name is also a valid alias
567
568 strategy = "auto"
569 for user_dim in user_dim_names:
570 s = resolve_strategy(attrs_rules, path, user_dim)
571 if s != "auto":
572 strategy = s
573 break
574 if strategy == "auto":
575 strategy = resolve_strategy(attrs_rules, path, "*")
576
577 if strategy == "ours":
578 winning_slices[dim] = left_dims.slices[dim].events
579 dimension_report[dim] = f"ours ({dim})"
580 elif strategy == "theirs":
581 winning_slices[dim] = right_dims.slices[dim].events
582 dimension_report[dim] = f"theirs ({dim})"
583 else:
584 # Unresolvable conflict. Non-independent dims fail fast.
585 return None
586
587 merged_bytes = _reconstruct(base_dims.ticks_per_beat, winning_slices)
588 return merged_bytes, dimension_report