aaronrene / gabriel-muse public
_invariants.py python
597 lines 19.7 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago
1 """Musical invariants engine for the Muse MIDI plugin.
2
3 Invariants are semantic rules that a MIDI track must satisfy. They are
4 evaluated at commit time, merge time, or on-demand via ``muse music-check``.
5 Violations are reported with human-readable descriptions, severity levels,
6 and structured addresses for programmatic consumers.
7
8 Rule file format (TOML)
9 -----------------------
10 Rules are declared in ``.muse/midi_invariants.toml`` (default path).
11 Example::
12
13 [[rule]]
14 name = "max_polyphony"
15 severity = "error"
16 scope = "track"
17 rule_type = "max_polyphony"
18
19 [rule.params]
20 max_simultaneous = 6
21
22 [[rule]]
23 name = "keep_in_range"
24 severity = "warning"
25 scope = "track"
26 rule_type = "pitch_range"
27
28 [rule.params]
29 min_pitch = 24
30 max_pitch = 108
31
32 [[rule]]
33 name = "no_fifths"
34 severity = "warning"
35 scope = "voice_pair"
36 rule_type = "no_parallel_fifths"
37
38 [[rule]]
39 name = "consistent_key"
40 severity = "info"
41 scope = "track"
42 rule_type = "key_consistency"
43
44 [rule.params]
45 threshold = 0.15
46
47 Built-in rule types
48 -------------------
49
50 ``max_polyphony``
51 Detects bars where more than *max_simultaneous* notes overlap at any
52 tick position. Uses a sweep-line algorithm over start/end tick events.
53
54 ``pitch_range``
55 Detects any note with ``pitch < min_pitch`` or ``pitch > max_pitch``.
56
57 ``key_consistency``
58 Detects notes whose pitch class is highly inconsistent with the key
59 estimated by the Krumhansl-Schmuckler algorithm. Fires when the ratio
60 of "foreign" pitch classes exceeds *threshold*.
61
62 ``no_parallel_fifths``
63 Detects consecutive bars where the lowest voice and the second-lowest
64 voice both move by a perfect fifth in parallel (a classical counterpoint
65 violation). Best-effort heuristic — voice assignment is implicit.
66
67 Severity levels
68 ---------------
69 - ``"error"`` — must be resolved before committing (when ``--strict`` is set).
70 - ``"warning"`` — reported but does not block commits.
71 - ``"info"`` — informational; surfaced in ``muse music-check`` output only.
72
73 Public API
74 ----------
75 - :class:`InvariantRule` — rule declaration TypedDict.
76 - :class:`InvariantViolation` — single violation record TypedDict.
77 - :class:`InvariantReport` — full report for one commit / track.
78 - :func:`load_invariant_rules` — load from TOML file with defaults fallback.
79 - :func:`run_invariants` — evaluate all rules against a commit.
80 """
81
82 from __future__ import annotations
83
84 import logging
85 import pathlib
86 from typing import Literal, TypedDict
87
88 from muse.core.invariants import BaseReport, BaseViolation, make_report
89 from muse.core.object_store import read_object
90 from muse.core.store import get_commit_snapshot_manifest
91 from muse.plugins.midi._query import NoteInfo, key_signature_guess, notes_by_bar
92 from muse.plugins.midi.midi_diff import extract_notes
93
94
95
96
97 type _ScopeMap = dict[str, "Literal['track', 'bar', 'voice_pair', 'global']"]
98 type _SeverityMap = dict[str, "Literal['info', 'warning', 'error']"]
99 type _ParamMap = dict[str, str | int | float]
100 logger = logging.getLogger(__name__)
101
102 _DEFAULT_RULES_FILE = ".muse/midi_invariants.toml"
103
104
105 # ---------------------------------------------------------------------------
106 # Types
107 # ---------------------------------------------------------------------------
108
109
110 class _InvariantRuleRequired(TypedDict):
111 name: str
112 severity: Literal["info", "warning", "error"]
113 scope: Literal["track", "bar", "voice_pair", "global"]
114 rule_type: str
115
116
117 class InvariantRule(_InvariantRuleRequired, total=False):
118 """Declaration of one MIDI invariant rule.
119
120 ``name`` Human-readable rule identifier (unique within a rule set).
121 ``severity`` Violation severity: ``"info"``, ``"warning"``, or ``"error"``.
122 ``scope`` Granularity: ``"track"``, ``"bar"``, ``"voice_pair"``, ``"global"``.
123 ``rule_type`` Built-in type string: ``"max_polyphony"``, ``"pitch_range"``,
124 ``"key_consistency"``, ``"no_parallel_fifths"``.
125 ``params`` Rule-specific parameter dict.
126 """
127
128 params: _ParamMap
129
130
131 class InvariantViolation(TypedDict):
132 """A single invariant violation record.
133
134 ``rule_name`` The name of the rule that fired.
135 ``severity`` Severity level from the rule declaration.
136 ``track`` Workspace-relative MIDI file path.
137 ``bar`` 1-indexed bar number (0 for track-level violations).
138 ``description`` Human-readable explanation of what was violated.
139 ``addresses`` Note addresses or other domain addresses involved.
140 """
141
142 rule_name: str
143 severity: Literal["info", "warning", "error"]
144 track: str
145 bar: int
146 description: str
147 addresses: list[str]
148
149
150 class InvariantReport(TypedDict):
151 """Full invariant check report for one commit.
152
153 ``commit_id`` The commit that was checked.
154 ``violations`` All violations found, sorted by track then bar.
155 ``rules_checked`` Number of rules evaluated.
156 ``has_errors`` True when any violation has severity ``"error"``.
157 ``has_warnings`` True when any violation has severity ``"warning"``.
158 """
159
160 commit_id: str
161 violations: list[InvariantViolation]
162 rules_checked: int
163 has_errors: bool
164 has_warnings: bool
165
166
167 # ---------------------------------------------------------------------------
168 # Built-in rule implementations
169 # ---------------------------------------------------------------------------
170
171
172 def check_max_polyphony(
173 notes: list[NoteInfo],
174 track: str,
175 rule_name: str,
176 severity: Literal["info", "warning", "error"],
177 *,
178 max_simultaneous: int = 6,
179 ) -> list[InvariantViolation]:
180 """Find bars where simultaneous note count exceeds *max_simultaneous*.
181
182 Uses a tick-based sweep-line over (start_tick, end_tick) intervals.
183 Reports one violation per offending bar.
184
185 Args:
186 notes: All notes in the track.
187 track: Track file path for violation records.
188 rule_name: Rule identifier string.
189 severity: Violation severity.
190 max_simultaneous: Maximum allowed simultaneous notes.
191
192 Returns:
193 List of :class:`InvariantViolation` records.
194 """
195 violations: list[InvariantViolation] = []
196 bars = notes_by_bar(notes)
197
198 for bar_num, bar_notes in sorted(bars.items()):
199 # Collect all tick events: +1 for note_on, -1 for note_off.
200 events: list[tuple[int, int]] = []
201 for n in bar_notes:
202 events.append((n.start_tick, 1))
203 events.append((n.start_tick + n.duration_ticks, -1))
204 events.sort(key=lambda e: (e[0], e[1])) # off before on at same tick
205
206 current = 0
207 peak = 0
208 peak_tick = 0
209 for tick, delta in events:
210 current += delta
211 if current > peak:
212 peak = current
213 peak_tick = tick
214
215 if peak > max_simultaneous:
216 violations.append(
217 InvariantViolation(
218 rule_name=rule_name,
219 severity=severity,
220 track=track,
221 bar=bar_num,
222 description=(
223 f"Polyphony reached {peak} simultaneous notes at tick {peak_tick} "
224 f"(max allowed: {max_simultaneous})"
225 ),
226 addresses=[f"bar:{bar_num}:tick:{peak_tick}"],
227 )
228 )
229
230 return violations
231
232
233 def check_pitch_range(
234 notes: list[NoteInfo],
235 track: str,
236 rule_name: str,
237 severity: Literal["info", "warning", "error"],
238 *,
239 min_pitch: int = 0,
240 max_pitch: int = 127,
241 ) -> list[InvariantViolation]:
242 """Find notes outside the allowed MIDI pitch range.
243
244 Args:
245 notes: All notes in the track.
246 track: Track file path.
247 rule_name: Rule identifier.
248 severity: Violation severity.
249 min_pitch: Lowest allowed MIDI pitch (inclusive).
250 max_pitch: Highest allowed MIDI pitch (inclusive).
251
252 Returns:
253 One :class:`InvariantViolation` per out-of-range note.
254 """
255 violations: list[InvariantViolation] = []
256 for note in notes:
257 if note.pitch < min_pitch or note.pitch > max_pitch:
258 violations.append(
259 InvariantViolation(
260 rule_name=rule_name,
261 severity=severity,
262 track=track,
263 bar=note.bar,
264 description=(
265 f"Note {note.pitch_name} (MIDI {note.pitch}) is outside "
266 f"allowed range [{min_pitch}, {max_pitch}]"
267 ),
268 addresses=[f"bar:{note.bar}:pitch:{note.pitch}"],
269 )
270 )
271 return violations
272
273
274 def check_key_consistency(
275 notes: list[NoteInfo],
276 track: str,
277 rule_name: str,
278 severity: Literal["info", "warning", "error"],
279 *,
280 threshold: float = 0.15,
281 ) -> list[InvariantViolation]:
282 """Detect notes whose pitch class is inconsistent with the guessed key.
283
284 Estimates the key using the Krumhansl-Schmuckler algorithm, then counts
285 the fraction of notes that use a pitch class not diatonic to that key.
286 Fires when the foreign-note ratio exceeds *threshold*.
287
288 Args:
289 notes: All notes in the track.
290 track: Track file path.
291 rule_name: Rule identifier.
292 severity: Violation severity.
293 threshold: Maximum allowed ratio of foreign pitch classes (0.0–1.0).
294
295 Returns:
296 Zero or one :class:`InvariantViolation` for the track.
297 """
298 if not notes:
299 return []
300
301 key_guess = key_signature_guess(notes)
302 # Parse key guess string e.g. "G major" or "D minor".
303 parts = key_guess.split()
304 if len(parts) < 2:
305 return []
306
307 root_name = parts[0]
308 mode = parts[1]
309
310 pitch_classes = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
311 root_idx = pitch_classes.index(root_name) if root_name in pitch_classes else -1
312 if root_idx < 0:
313 return []
314
315 # Diatonic pitch classes for major and natural minor scales.
316 major_steps = [0, 2, 4, 5, 7, 9, 11]
317 minor_steps = [0, 2, 3, 5, 7, 8, 10]
318 steps = major_steps if mode == "major" else minor_steps
319 diatonic_pcs = frozenset((root_idx + s) % 12 for s in steps)
320
321 foreign = sum(1 for n in notes if n.pitch_class not in diatonic_pcs)
322 ratio = foreign / len(notes)
323
324 if ratio > threshold:
325 return [
326 InvariantViolation(
327 rule_name=rule_name,
328 severity=severity,
329 track=track,
330 bar=0,
331 description=(
332 f"{foreign}/{len(notes)} notes ({ratio:.0%}) use pitch classes "
333 f"foreign to estimated key {key_guess} "
334 f"(threshold: {threshold:.0%})"
335 ),
336 addresses=[track],
337 )
338 ]
339 return []
340
341
342 def check_no_parallel_fifths(
343 notes: list[NoteInfo],
344 track: str,
345 rule_name: str,
346 severity: Literal["info", "warning", "error"],
347 ) -> list[InvariantViolation]:
348 """Detect consecutive bars with parallel perfect fifth motion.
349
350 Heuristic: for each pair of consecutive bars, find the two lowest-pitched
351 notes (approximating bass and tenor voices) and check whether both voices
352 move by a perfect fifth (7 semitones) in the same direction.
353
354 This is a best-effort approximation — accurate voice separation would
355 require dedicated voice-leading analysis beyond this scope.
356
357 Args:
358 notes: All notes in the track.
359 track: Track file path.
360 rule_name: Rule identifier.
361 severity: Violation severity.
362
363 Returns:
364 One :class:`InvariantViolation` per detected parallel-fifth bar pair.
365 """
366 violations: list[InvariantViolation] = []
367 bars = notes_by_bar(notes)
368 sorted_bars = sorted(bars.keys())
369
370 for i in range(len(sorted_bars) - 1):
371 bar_a = sorted_bars[i]
372 bar_b = sorted_bars[i + 1]
373 notes_a = sorted(bars[bar_a], key=lambda n: n.pitch)
374 notes_b = sorted(bars[bar_b], key=lambda n: n.pitch)
375
376 if len(notes_a) < 2 or len(notes_b) < 2:
377 continue
378
379 # Take two lowest pitches as approximated bass + tenor voices.
380 v1_a, v2_a = notes_a[0].pitch, notes_a[1].pitch
381 v1_b, v2_b = notes_b[0].pitch, notes_b[1].pitch
382
383 # Interval between voices in each bar.
384 interval_a = abs(v2_a - v1_a) % 12
385 interval_b = abs(v2_b - v1_b) % 12
386
387 # Both form a perfect fifth (7 semitones modulo octave)?
388 if interval_a == 7 and interval_b == 7:
389 # Both voices moved in the same direction?
390 motion_v1 = v1_b - v1_a
391 motion_v2 = v2_b - v2_a
392 if (motion_v1 > 0 and motion_v2 > 0) or (motion_v1 < 0 and motion_v2 < 0):
393 violations.append(
394 InvariantViolation(
395 rule_name=rule_name,
396 severity=severity,
397 track=track,
398 bar=bar_b,
399 description=(
400 f"Parallel fifths between bars {bar_a} and {bar_b}: "
401 f"lower voice {notes_a[0].pitch_name}→{notes_b[0].pitch_name}, "
402 f"upper voice {notes_a[1].pitch_name}→{notes_b[1].pitch_name}"
403 ),
404 addresses=[f"bar:{bar_a}", f"bar:{bar_b}"],
405 )
406 )
407
408 return violations
409
410
411 # ---------------------------------------------------------------------------
412 # Rule loading
413 # ---------------------------------------------------------------------------
414
415 _DEFAULT_RULE_SET: list[InvariantRule] = [
416 InvariantRule(
417 name="max_polyphony",
418 severity="warning",
419 scope="track",
420 rule_type="max_polyphony",
421 params={"max_simultaneous": 8},
422 ),
423 InvariantRule(
424 name="pitch_range",
425 severity="warning",
426 scope="track",
427 rule_type="pitch_range",
428 params={"min_pitch": 0, "max_pitch": 127},
429 ),
430 ]
431
432
433 def load_invariant_rules(rules_file: pathlib.Path | None = None) -> list[InvariantRule]:
434 """Load invariant rules from a TOML file, falling back to defaults.
435
436 Requires ``tomllib`` (Python 3.11+) for TOML parsing. If the file does
437 not exist or cannot be parsed, the default rule set is returned.
438
439 Args:
440 rules_file: Path to the TOML rule file. ``None`` means use defaults.
441
442 Returns:
443 List of :class:`InvariantRule` dicts.
444 """
445 if rules_file is None or not rules_file.exists():
446 return list(_DEFAULT_RULE_SET)
447
448 try:
449 import tomllib
450
451 with rules_file.open("rb") as fh:
452 data = tomllib.load(fh)
453
454 rules: list[InvariantRule] = []
455 for raw in data.get("rule", []):
456 _valid_severities: _SeverityMap = {
457 "info": "info", "warning": "warning", "error": "error",
458 }
459 _valid_scopes: _ScopeMap = {
460 "track": "track", "bar": "bar", "voice_pair": "voice_pair", "global": "global",
461 }
462 sev = _valid_severities.get(str(raw.get("severity", "")), "warning")
463 scope = _valid_scopes.get(str(raw.get("scope", "")), "track")
464 rule = InvariantRule(
465 name=str(raw.get("name", "unnamed")),
466 severity=sev,
467 scope=scope,
468 rule_type=str(raw.get("rule_type", "")),
469 )
470 if "params" in raw:
471 rule["params"] = raw["params"]
472 rules.append(rule)
473 return rules if rules else list(_DEFAULT_RULE_SET)
474
475 except Exception as exc:
476 logger.warning("⚠️ Could not load invariant rules from %s: %s", rules_file, exc)
477 return list(_DEFAULT_RULE_SET)
478
479
480 # ---------------------------------------------------------------------------
481 # Main runner
482 # ---------------------------------------------------------------------------
483
484
485 def run_invariants(
486 root: "pathlib.Path",
487 commit_id: str,
488 rules: list[InvariantRule],
489 *,
490 track_filter: str | None = None,
491 ) -> InvariantReport:
492 """Evaluate all *rules* against every MIDI track in *commit_id*.
493
494 Args:
495 root: Repository root.
496 commit_id: Commit to check.
497 rules: List of :class:`InvariantRule` declarations.
498 track_filter: Restrict check to a single MIDI file path.
499
500 Returns:
501 An :class:`InvariantReport` with all violations found.
502 """
503 import pathlib as _pathlib
504
505 all_violations: list[InvariantViolation] = []
506 manifest = get_commit_snapshot_manifest(root, commit_id) or {}
507
508 midi_paths = [
509 p for p in manifest
510 if p.lower().endswith(".mid")
511 and (track_filter is None or p == track_filter)
512 ]
513
514 for track_path in sorted(midi_paths):
515 obj_hash = manifest.get(track_path)
516 if obj_hash is None:
517 continue
518 raw = read_object(root, obj_hash)
519 if raw is None:
520 continue
521 try:
522 keys, tpb = extract_notes(raw)
523 except ValueError as exc:
524 logger.debug("Cannot parse MIDI %r: %s", track_path, exc)
525 continue
526
527 notes = [NoteInfo.from_note_key(k, tpb) for k in keys]
528
529 for rule in rules:
530 rt = rule["rule_type"]
531 sev = rule["severity"]
532 params = rule.get("params", {})
533 name = rule["name"]
534
535 if rt == "max_polyphony":
536 max_sim = int(params.get("max_simultaneous", 8))
537 all_violations.extend(
538 check_max_polyphony(notes, track_path, name, sev, max_simultaneous=max_sim)
539 )
540 elif rt == "pitch_range":
541 min_p = int(params.get("min_pitch", 0))
542 max_p = int(params.get("max_pitch", 127))
543 all_violations.extend(
544 check_pitch_range(notes, track_path, name, sev, min_pitch=min_p, max_pitch=max_p)
545 )
546 elif rt == "key_consistency":
547 thresh = float(params.get("threshold", 0.15))
548 all_violations.extend(
549 check_key_consistency(notes, track_path, name, sev, threshold=thresh)
550 )
551 elif rt == "no_parallel_fifths":
552 all_violations.extend(
553 check_no_parallel_fifths(notes, track_path, name, sev)
554 )
555 else:
556 logger.debug("Unknown rule_type %r in rule %r — skipped", rt, name)
557
558 all_violations.sort(key=lambda v: (v["track"], v["bar"]))
559 has_errors = any(v["severity"] == "error" for v in all_violations)
560 has_warnings = any(v["severity"] == "warning" for v in all_violations)
561
562 return InvariantReport(
563 commit_id=commit_id,
564 violations=all_violations,
565 rules_checked=len(rules) * len(midi_paths),
566 has_errors=has_errors,
567 has_warnings=has_warnings,
568 )
569
570
571 class MidiChecker:
572 """Satisfies :class:`~muse.core.invariants.InvariantChecker` for the MIDI domain.
573
574 Wraps :func:`run_invariants` so that the generic ``muse check`` command
575 can dispatch to the MIDI checker without knowing MIDI internals.
576 """
577
578 def check(
579 self,
580 repo_root: pathlib.Path,
581 commit_id: str,
582 *,
583 rules_file: pathlib.Path | None = None,
584 ) -> BaseReport:
585 """Run MIDI invariant checks against *commit_id* and return a :class:`~muse.core.invariants.BaseReport`."""
586 rules = load_invariant_rules(rules_file)
587 midi_report = run_invariants(repo_root, commit_id, rules)
588 base_violations: list[BaseViolation] = [
589 BaseViolation(
590 rule_name=v["rule_name"],
591 severity=v["severity"],
592 address=v["track"],
593 description=v["description"],
594 )
595 for v in midi_report["violations"]
596 ]
597 return make_report(commit_id, "midi", base_violations, midi_report["rules_checked"])
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago