aaronrene / gabriel-muse public
midi_merge.py python
607 lines 23.5 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """MIDI dimension-aware merge for the Muse MIDI 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
102 from __future__ import annotations
103
104 import hashlib
105 import io
106 import json
107 import logging
108 from dataclasses import dataclass, field
109
110 import mido
111
112 logger = logging.getLogger(__name__)
113
114 from muse.core.attributes import AttributeRule, resolve_strategy
115
116 # ---------------------------------------------------------------------------
117 # Dimension constants — the complete MIDI dimension taxonomy
118 # ---------------------------------------------------------------------------
119
120 #: Internal dimension names, ordered canonically.
121 #: Each MIDI event type maps to exactly one of these buckets.
122 INTERNAL_DIMS: list[str] = [
123 # --- Expressive note content ---
124 "notes", # note_on / note_off
125 "pitch_bend", # pitchwheel
126 "channel_pressure", # aftertouch (mono)
127 "poly_pressure", # polytouch (per-note)
128 # --- Named CC controllers (individually mergeable) ---
129 "cc_modulation", # CC 1
130 "cc_volume", # CC 7
131 "cc_pan", # CC 10
132 "cc_expression", # CC 11
133 "cc_sustain", # CC 64
134 "cc_portamento", # CC 65
135 "cc_sostenuto", # CC 66
136 "cc_soft_pedal", # CC 67
137 "cc_reverb", # CC 91
138 "cc_chorus", # CC 93
139 "cc_other", # all remaining CC numbers
140 # --- Patch / program selection ---
141 "program_change",
142 # --- Timeline / notation metadata (non-independent) ---
143 "tempo_map", # set_tempo — non-independent: affects all tick positions
144 "time_signatures", # time_signature — non-independent: affects bar structure
145 # --- Tonal context and notation ---
146 "key_signatures", # key_signature
147 "markers", # marker, cue_marker, text, lyrics, copyright
148 # --- Track structure (non-independent) ---
149 "track_structure", # track_name, instrument_name, sysex, unknown meta
150 ]
151
152 #: Dimensions whose conflicts block merging all other dimensions until resolved.
153 #: All other dimensions are merged in parallel regardless of conflicts here.
154 NON_INDEPENDENT_DIMS: frozenset[str] = frozenset({
155 "tempo_map",
156 "time_signatures",
157 "track_structure",
158 })
159
160 #: User-facing dimension names from .museattributes mapped to internal buckets.
161 #: Agents and humans use these names in merge strategy declarations.
162 DIM_ALIAS: _DimAliasMap = {
163 "pitch_bend": "pitch_bend",
164 "aftertouch": "channel_pressure",
165 "poly_aftertouch": "poly_pressure",
166 "modulation": "cc_modulation",
167 "volume": "cc_volume",
168 "pan": "cc_pan",
169 "expression": "cc_expression",
170 "sustain": "cc_sustain",
171 "portamento": "cc_portamento",
172 "sostenuto": "cc_sostenuto",
173 "soft_pedal": "cc_soft_pedal",
174 "reverb": "cc_reverb",
175 "chorus": "cc_chorus",
176 "automation": "cc_other",
177 "program": "program_change",
178 "tempo": "tempo_map",
179 "time_sig": "time_signatures",
180 "key_sig": "key_signatures",
181 "markers": "markers",
182 "track_structure": "track_structure",
183 }
184
185 #: All valid names (aliases + internal) → internal bucket.
186 _CANONICAL: _DimAliasMap = {**DIM_ALIAS, **{d: d for d in INTERNAL_DIMS}}
187
188 #: CC number → internal dimension name for named controllers.
189 _CC_DIM: dict[int, str] = {
190 1: "cc_modulation",
191 7: "cc_volume",
192 10: "cc_pan",
193 11: "cc_expression",
194 64: "cc_sustain",
195 65: "cc_portamento",
196 66: "cc_sostenuto",
197 67: "cc_soft_pedal",
198 91: "cc_reverb",
199 93: "cc_chorus",
200 }
201
202
203 # ---------------------------------------------------------------------------
204 # Data types
205 # ---------------------------------------------------------------------------
206
207
208 @dataclass
209 class DimensionSlice:
210 """Events belonging to one dimension of a MIDI file.
211
212 ``events`` is a list of ``(abs_tick, mido.Message)`` pairs sorted by
213 ascending absolute tick. ``content_hash`` is the SHA-256 digest of the
214 canonical JSON serialisation of the event list (used for change detection
215 without loading file bytes).
216 """
217
218 name: str
219 events: list[tuple[int, mido.Message]] = field(default_factory=list)
220 content_hash: str = ""
221
222 def __post_init__(self) -> None:
223 if not self.content_hash:
224 self.content_hash = _hash_events(self.events)
225
226
227 @dataclass
228 class MidiDimensions:
229 """All dimension slices extracted from one MIDI file.
230
231 ``slices`` maps internal dimension name → :class:`DimensionSlice`.
232 Every internal dimension in :data:`INTERNAL_DIMS` has an entry, even if
233 the corresponding event list is empty (hash of empty list is stable).
234 """
235
236 ticks_per_beat: int
237 file_type: int
238 slices: _DimSliceMap
239
240 def get(self, dim: str) -> DimensionSlice:
241 """Return the slice for a user-facing or internal dimension name."""
242 internal = _CANONICAL.get(dim, dim)
243 return self.slices[internal]
244
245
246 # ---------------------------------------------------------------------------
247 # Internal helpers
248 # ---------------------------------------------------------------------------
249
250
251 def _classify_event(msg: mido.Message) -> str | None:
252 """Map a mido Message to an internal dimension bucket.
253
254 Returns ``None`` for events that should be excluded from all buckets
255 (e.g. ``end_of_track`` is handled during reconstruction, not stored here).
256 Unknown messages that are meta events fall back to ``"track_structure"``.
257 True unknowns (no ``is_meta`` attribute) are discarded.
258 """
259 t = msg.type
260
261 # --- Note events ---
262 if t in ("note_on", "note_off"):
263 return "notes"
264
265 # --- Pitch / pressure ---
266 if t == "pitchwheel":
267 return "pitch_bend"
268 if t == "aftertouch":
269 return "channel_pressure"
270 if t == "polytouch":
271 return "poly_pressure"
272
273 # --- CC — split by controller number ---
274 if t == "control_change":
275 return _CC_DIM.get(msg.control, "cc_other")
276
277 # --- Program change ---
278 if t == "program_change":
279 return "program_change"
280
281 # --- Timeline metadata ---
282 if t == "set_tempo":
283 return "tempo_map"
284 if t == "time_signature":
285 return "time_signatures"
286 if t == "key_signature":
287 return "key_signatures"
288
289 # --- Section markers and text annotations ---
290 if t in ("marker", "cue_marker", "text", "lyrics", "copyright"):
291 return "markers"
292
293 # --- Track structure and routing ---
294 if t in ("track_name", "instrument_name", "sysex", "sequencer_specific"):
295 return "track_structure"
296
297 # --- End-of-track is reconstructed, not stored ---
298 if t == "end_of_track":
299 return None
300
301 # --- Unknown meta events → track structure (safe default) ---
302 if getattr(msg, "is_meta", False):
303 return "track_structure"
304
305 return None
306
307
308 type _MsgVal = int | str | list[int]
309 type _MsgDict = dict[str, _MsgVal] # serialised mido.Message
310 type _DimAliasMap = dict[str, str] # dimension alias → canonical name
311 type _DimSliceMap = dict[str, "DimensionSlice"] # dim name → slice
312 type _DimBuckets = dict[str, list[tuple[int, mido.Message]]] # dim → (tick, msg) list
313 type _DimReport = dict[str, str] # dim name → resolution summary
314
315
316 def _msg_to_dict(msg: mido.Message) -> _MsgDict:
317 """Serialise a mido Message to a JSON-compatible dict."""
318 from muse.core.validation import MAX_SYSEX_BYTES
319
320 d: _MsgDict = {"type": msg.type}
321 for attr in (
322 "channel", "note", "velocity", "control", "value",
323 "pitch", "program", "numerator", "denominator",
324 "clocks_per_click", "notated_32nd_notes_per_beat",
325 "tempo", "key", "scale", "text", "data",
326 ):
327 if hasattr(msg, attr):
328 raw = getattr(msg, attr)
329 if isinstance(raw, (bytes, bytearray)):
330 # Cap sysex / large byte payloads to prevent memory exhaustion
331 # when a crafted MIDI contains a giant sysex blob.
332 if len(raw) > MAX_SYSEX_BYTES:
333 logger.warning(
334 "⚠️ Sysex payload %d bytes exceeds cap (%d) — truncating",
335 len(raw), MAX_SYSEX_BYTES,
336 )
337 raw = raw[:MAX_SYSEX_BYTES]
338 d[attr] = list(raw)
339 elif isinstance(raw, str):
340 d[attr] = raw
341 elif isinstance(raw, int):
342 d[attr] = raw
343 return d
344
345
346 def _hash_events(events: list[tuple[int, mido.Message]]) -> str:
347 """SHA-256 of the canonical JSON representation of an event list."""
348 payload = json.dumps(
349 [(tick, _msg_to_dict(msg)) for tick, msg in events],
350 sort_keys=True,
351 separators=(",", ":"),
352 ).encode()
353 return hashlib.sha256(payload).hexdigest()
354
355
356 def _to_absolute(track: mido.MidiTrack) -> list[tuple[int, mido.Message]]:
357 """Convert a delta-time track to a list of ``(abs_tick, msg)`` pairs."""
358 result: list[tuple[int, mido.Message]] = []
359 abs_tick = 0
360 for msg in track:
361 abs_tick += msg.time
362 result.append((abs_tick, msg))
363 return result
364
365
366 # ---------------------------------------------------------------------------
367 # Public: extract_dimensions
368 # ---------------------------------------------------------------------------
369
370
371 def extract_dimensions(midi_bytes: bytes) -> MidiDimensions:
372 """Parse *midi_bytes* and bucket events by dimension.
373
374 Every event type in the MIDI spec maps to exactly one of the
375 :data:`INTERNAL_DIMS` buckets. Empty buckets are present with an empty
376 event list so that callers can always index by dimension name.
377
378 Args:
379 midi_bytes: Raw bytes of a ``.mid`` file.
380
381 Returns:
382 A :class:`MidiDimensions` with one :class:`DimensionSlice` per
383 internal dimension. Events within each slice are sorted by ascending
384 absolute tick, then by event type for determinism when multiple events
385 share the same tick.
386
387 Raises:
388 ValueError: If *midi_bytes* cannot be parsed as a MIDI file.
389 """
390 try:
391 mid = mido.MidiFile(file=io.BytesIO(midi_bytes))
392 except Exception as exc:
393 raise ValueError(f"Failed to parse MIDI data: {exc}") from exc
394
395 buckets: _DimBuckets = {
396 dim: [] for dim in INTERNAL_DIMS
397 }
398
399 for track in mid.tracks:
400 for abs_tick, msg in _to_absolute(track):
401 bucket = _classify_event(msg)
402 if bucket is not None:
403 buckets[bucket].append((abs_tick, msg))
404
405 for dim in INTERNAL_DIMS:
406 buckets[dim].sort(key=lambda x: (x[0], x[1].type))
407
408 slices = {
409 dim: DimensionSlice(name=dim, events=events)
410 for dim, events in buckets.items()
411 }
412 return MidiDimensions(
413 ticks_per_beat=mid.ticks_per_beat,
414 file_type=mid.type,
415 slices=slices,
416 )
417
418
419 # ---------------------------------------------------------------------------
420 # Public: dimension_conflict_detail
421 # ---------------------------------------------------------------------------
422
423
424 def dimension_conflict_detail(
425 base: MidiDimensions,
426 left: MidiDimensions,
427 right: MidiDimensions,
428 ) -> _DimReport:
429 """Return a per-dimension change report for a conflicting file.
430
431 Returns a dict mapping internal dimension name to one of:
432
433 - ``"unchanged"`` — neither side changed this dimension.
434 - ``"left_only"`` — only the left (ours) side changed.
435 - ``"right_only"`` — only the right (theirs) side changed.
436 - ``"both"`` — both sides changed; a dimension-level conflict.
437
438 This is used by :func:`merge_midi_dimensions` and surfaced in
439 ``muse merge`` output for human-readable conflict diagnostics.
440 """
441 report: _DimReport = {}
442 for dim in INTERNAL_DIMS:
443 base_hash = base.slices[dim].content_hash
444 left_hash = left.slices[dim].content_hash
445 right_hash = right.slices[dim].content_hash
446 left_changed = base_hash != left_hash
447 right_changed = base_hash != right_hash
448 if left_changed and right_changed:
449 report[dim] = "both"
450 elif left_changed:
451 report[dim] = "left_only"
452 elif right_changed:
453 report[dim] = "right_only"
454 else:
455 report[dim] = "unchanged"
456 return report
457
458
459 # ---------------------------------------------------------------------------
460 # Reconstruction helpers
461 # ---------------------------------------------------------------------------
462
463
464 def _events_to_track(
465 events: list[tuple[int, mido.Message]],
466 ) -> mido.MidiTrack:
467 """Convert absolute-tick events to a mido MidiTrack with delta times."""
468 track = mido.MidiTrack()
469 prev_tick = 0
470 for abs_tick, msg in sorted(events, key=lambda x: (x[0], x[1].type)):
471 delta = abs_tick - prev_tick
472 new_msg = msg.copy(time=delta)
473 track.append(new_msg)
474 prev_tick = abs_tick
475 if not track or track[-1].type != "end_of_track":
476 track.append(mido.MetaMessage("end_of_track", time=0))
477 return track
478
479
480 def _reconstruct(
481 ticks_per_beat: int,
482 winning_slices: _DimBuckets,
483 ) -> bytes:
484 """Build a type-0 MIDI file from winning dimension event lists.
485
486 All dimension events are merged into a single track (type-0) for
487 maximum compatibility. The absolute-tick ordering is preserved and
488 duplicate end_of_track messages are removed.
489 """
490 all_events: list[tuple[int, mido.Message]] = []
491 for events in winning_slices.values():
492 all_events.extend(events)
493
494 all_events = [
495 (tick, msg) for tick, msg in all_events
496 if msg.type != "end_of_track"
497 ]
498 all_events.sort(key=lambda x: (x[0], x[1].type))
499
500 track = _events_to_track(all_events)
501 mid = mido.MidiFile(type=0, ticks_per_beat=ticks_per_beat)
502 mid.tracks.append(track)
503
504 buf = io.BytesIO()
505 mid.save(file=buf)
506 return buf.getvalue()
507
508
509 # ---------------------------------------------------------------------------
510 # Public: merge_midi_dimensions
511 # ---------------------------------------------------------------------------
512
513
514 def merge_midi_dimensions(
515 base_bytes: bytes,
516 left_bytes: bytes,
517 right_bytes: bytes,
518 attrs_rules: list[AttributeRule],
519 path: str,
520 ) -> tuple[bytes, dict[str, str]] | None:
521 """Attempt a dimension-level three-way merge of a MIDI file.
522
523 For each internal dimension (all 21 of them):
524
525 - If neither side changed → keep base.
526 - If only one side changed → take that side (clean auto-merge).
527 - If both sides changed → consult ``.museattributes`` strategy:
528
529 * ``ours`` / ``theirs`` → take the specified side; record in report.
530 * ``manual`` / ``auto`` / ``union`` → unresolvable; return ``None``.
531
532 Non-independent dimensions (``tempo_map``, ``time_signatures``,
533 ``track_structure``) that have bilateral conflicts cause an immediate
534 ``None`` return — their conflicts cannot be auto-resolved because they
535 affect the semantic meaning of all other dimensions.
536
537 Args:
538 base_bytes: MIDI bytes for the common ancestor.
539 left_bytes: MIDI bytes for the ours (left) branch.
540 right_bytes: MIDI bytes for the theirs (right) branch.
541 attrs_rules: Rule list from :func:`muse.core.attributes.load_attributes`.
542 path: Workspace-relative POSIX path (used for strategy lookup).
543
544 Returns:
545 A ``(merged_bytes, dimension_report)`` tuple when all dimension
546 conflicts can be resolved, or ``None`` when at least one dimension
547 conflict has no resolvable strategy.
548
549 *dimension_report* maps each internal dimension name to the side
550 chosen: ``"base"``, ``"left"``, ``"right"``, or the strategy string.
551 Only dimensions with non-empty event lists or conflicts are included.
552
553 Raises:
554 ValueError: If any of the byte strings cannot be parsed as MIDI.
555 """
556 base_dims = extract_dimensions(base_bytes)
557 left_dims = extract_dimensions(left_bytes)
558 right_dims = extract_dimensions(right_bytes)
559
560 detail = dimension_conflict_detail(base_dims, left_dims, right_dims)
561
562 winning_slices: _DimBuckets = {}
563 dimension_report: _DimReport = {}
564
565 for dim in INTERNAL_DIMS:
566 change = detail[dim]
567
568 if change == "unchanged":
569 winning_slices[dim] = base_dims.slices[dim].events
570 if base_dims.slices[dim].events:
571 dimension_report[dim] = "base"
572
573 elif change == "left_only":
574 winning_slices[dim] = left_dims.slices[dim].events
575 dimension_report[dim] = "left"
576
577 elif change == "right_only":
578 winning_slices[dim] = right_dims.slices[dim].events
579 dimension_report[dim] = "right"
580
581 else:
582 # Both sides changed — resolve via .museattributes strategy.
583 # Look up by user-facing aliases first, then internal name.
584 user_dim_names = [k for k, v in DIM_ALIAS.items() if v == dim]
585 user_dim_names.append(dim) # internal name is also a valid alias
586
587 strategy = "auto"
588 for user_dim in user_dim_names:
589 s = resolve_strategy(attrs_rules, path, user_dim)
590 if s != "auto":
591 strategy = s
592 break
593 if strategy == "auto":
594 strategy = resolve_strategy(attrs_rules, path, "*")
595
596 if strategy == "ours":
597 winning_slices[dim] = left_dims.slices[dim].events
598 dimension_report[dim] = f"ours ({dim})"
599 elif strategy == "theirs":
600 winning_slices[dim] = right_dims.slices[dim].events
601 dimension_report[dim] = f"theirs ({dim})"
602 else:
603 # Unresolvable conflict. Non-independent dims fail fast.
604 return None
605
606 merged_bytes = _reconstruct(base_dims.ticks_per_beat, winning_slices)
607 return merged_bytes, dimension_report
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago