muse_groove_check.py
python
| 1 | """Muse Groove-Check Service — rhythmic drift analysis across commits. |
| 2 | |
| 3 | Computes per-commit groove scores by measuring note-onset deviation |
| 4 | from the quantization grid, then detects which commits introduced |
| 5 | rhythmic inconsistency relative to their neighbors. |
| 6 | |
| 7 | "Groove drift" is the absolute change in average onset deviation between |
| 8 | adjacent commits. A commit with a large drift delta is the one that |
| 9 | "killed the groove." |
| 10 | |
| 11 | This is a stub implementation that demonstrates the correct CLI contract |
| 12 | and result schema. Full MIDI content analysis will be wired in once |
| 13 | Storpheus exposes a rhythmic quantization introspection route. |
| 14 | |
| 15 | Boundary rules: |
| 16 | - Pure data — no side effects, no external I/O. |
| 17 | - Must NOT import StateStore, EntityRegistry, or executor modules. |
| 18 | - Must NOT import LLM handlers or maestro_* pipeline modules. |
| 19 | """ |
| 20 | |
| 21 | from __future__ import annotations |
| 22 | |
| 23 | import logging |
| 24 | from dataclasses import dataclass, field |
| 25 | from enum import Enum |
| 26 | from typing import Optional |
| 27 | |
| 28 | logger = logging.getLogger(__name__) |
| 29 | |
| 30 | # --------------------------------------------------------------------------- |
| 31 | # Constants |
| 32 | # --------------------------------------------------------------------------- |
| 33 | |
| 34 | DEFAULT_THRESHOLD = 0.1 # beats — flag commits whose drift_delta exceeds this |
| 35 | DEFAULT_COMMIT_LIMIT = 10 # fallback window when no explicit range is given |
| 36 | |
| 37 | |
| 38 | # --------------------------------------------------------------------------- |
| 39 | # Result types (stable CLI contract) |
| 40 | # --------------------------------------------------------------------------- |
| 41 | |
| 42 | |
| 43 | class GrooveStatus(str, Enum): |
| 44 | """Per-commit groove assessment relative to the configured threshold. |
| 45 | |
| 46 | OK — drift_delta ≤ threshold; rhythm is consistent with neighbors. |
| 47 | WARN — drift_delta is between threshold and 2× threshold; mild drift. |
| 48 | FAIL — drift_delta > 2× threshold; likely culprit for groove regression. |
| 49 | """ |
| 50 | |
| 51 | OK = "OK" |
| 52 | WARN = "WARN" |
| 53 | FAIL = "FAIL" |
| 54 | |
| 55 | |
| 56 | @dataclass(frozen=True) |
| 57 | class CommitGrooveMetrics: |
| 58 | """Rhythmic groove metrics for a single commit. |
| 59 | |
| 60 | groove_score — average note-onset deviation from the quantization grid, |
| 61 | in beats. Lower is tighter (closer to the grid). |
| 62 | drift_delta — absolute change in groove_score relative to the prior |
| 63 | commit in the range. The first commit always has delta 0.0. |
| 64 | status — OK / WARN / FAIL classification against the threshold. |
| 65 | commit — short commit ref (8 hex chars or resolved ID). |
| 66 | track — track scope used for analysis, or "all". |
| 67 | section — section scope used for analysis, or "all". |
| 68 | midi_files — number of MIDI snapshots analysed for this commit. |
| 69 | """ |
| 70 | |
| 71 | commit: str |
| 72 | groove_score: float |
| 73 | drift_delta: float |
| 74 | status: GrooveStatus |
| 75 | track: str = "all" |
| 76 | section: str = "all" |
| 77 | midi_files: int = 0 |
| 78 | |
| 79 | |
| 80 | @dataclass(frozen=True) |
| 81 | class GrooveCheckResult: |
| 82 | """Aggregate result for a `muse groove-check` run. |
| 83 | |
| 84 | commit_range — the range string that was analysed (e.g. "HEAD~5..HEAD"). |
| 85 | threshold — drift threshold used for WARN/FAIL classification. |
| 86 | total_commits — total commits in the analysis window. |
| 87 | flagged_commits — number of commits with status WARN or FAIL. |
| 88 | worst_commit — commit ref with the highest drift_delta, or empty string. |
| 89 | entries — per-commit metrics, oldest-first. |
| 90 | """ |
| 91 | |
| 92 | commit_range: str |
| 93 | threshold: float |
| 94 | total_commits: int |
| 95 | flagged_commits: int |
| 96 | worst_commit: str |
| 97 | entries: tuple[CommitGrooveMetrics, ...] = field(default_factory=tuple) |
| 98 | |
| 99 | |
| 100 | # --------------------------------------------------------------------------- |
| 101 | # Status classification |
| 102 | # --------------------------------------------------------------------------- |
| 103 | |
| 104 | |
| 105 | def classify_status(drift_delta: float, threshold: float) -> GrooveStatus: |
| 106 | """Classify a drift delta against the threshold. |
| 107 | |
| 108 | Args: |
| 109 | drift_delta: Absolute change in groove_score vs. prior commit. |
| 110 | threshold: User-configurable WARN boundary in beats. |
| 111 | |
| 112 | Returns: |
| 113 | :class:`GrooveStatus` OK, WARN, or FAIL. |
| 114 | """ |
| 115 | if drift_delta <= threshold: |
| 116 | return GrooveStatus.OK |
| 117 | if drift_delta <= threshold * 2: |
| 118 | return GrooveStatus.WARN |
| 119 | return GrooveStatus.FAIL |
| 120 | |
| 121 | |
| 122 | # --------------------------------------------------------------------------- |
| 123 | # Stub data factories |
| 124 | # --------------------------------------------------------------------------- |
| 125 | |
| 126 | _STUB_COMMITS: tuple[tuple[str, float, int], ...] = ( |
| 127 | ("a1b2c3d4", 0.04, 3), |
| 128 | ("e5f6a7b8", 0.05, 3), |
| 129 | ("c9d0e1f2", 0.06, 3), |
| 130 | ("a3b4c5d6", 0.09, 3), |
| 131 | ("e7f8a9b0", 0.15, 3), # groove degraded here |
| 132 | ("c1d2e3f4", 0.13, 3), |
| 133 | ("a5b6c7d8", 0.08, 3), # recovered |
| 134 | ) |
| 135 | |
| 136 | |
| 137 | def build_stub_entries( |
| 138 | *, |
| 139 | threshold: float, |
| 140 | track: Optional[str], |
| 141 | section: Optional[str], |
| 142 | limit: int, |
| 143 | ) -> list[CommitGrooveMetrics]: |
| 144 | """Produce stub CommitGrooveMetrics for a commit window. |
| 145 | |
| 146 | Returns the last ``limit`` entries from the stub table with |
| 147 | drift_delta and status computed against ``threshold``. |
| 148 | |
| 149 | Args: |
| 150 | threshold: WARN/FAIL boundary in beats. |
| 151 | track: Track filter (stored in metadata; no content effect in stub). |
| 152 | section: Section filter (stored in metadata; no content effect in stub). |
| 153 | limit: Maximum number of commits to return. |
| 154 | |
| 155 | Returns: |
| 156 | List of :class:`CommitGrooveMetrics`, oldest-first. |
| 157 | """ |
| 158 | sample = list(_STUB_COMMITS[-limit:]) |
| 159 | entries: list[CommitGrooveMetrics] = [] |
| 160 | prev_score: Optional[float] = None |
| 161 | for commit, score, midi_files in sample: |
| 162 | delta = abs(score - prev_score) if prev_score is not None else 0.0 |
| 163 | status = classify_status(delta, threshold) |
| 164 | entries.append( |
| 165 | CommitGrooveMetrics( |
| 166 | commit=commit, |
| 167 | groove_score=round(score, 4), |
| 168 | drift_delta=round(delta, 4), |
| 169 | status=status, |
| 170 | track=track or "all", |
| 171 | section=section or "all", |
| 172 | midi_files=midi_files, |
| 173 | ) |
| 174 | ) |
| 175 | prev_score = score |
| 176 | return entries |
| 177 | |
| 178 | |
| 179 | def compute_groove_check( |
| 180 | *, |
| 181 | commit_range: str, |
| 182 | threshold: float = DEFAULT_THRESHOLD, |
| 183 | track: Optional[str] = None, |
| 184 | section: Optional[str] = None, |
| 185 | limit: int = DEFAULT_COMMIT_LIMIT, |
| 186 | ) -> GrooveCheckResult: |
| 187 | """Compute groove-check metrics for a commit range. |
| 188 | |
| 189 | Pure function — safe to call from tests without any repository context. |
| 190 | The stub implementation produces deterministic, musically-realistic results |
| 191 | using hardcoded representative data. |
| 192 | |
| 193 | Args: |
| 194 | commit_range: Commit range string used for display (e.g. "HEAD~5..HEAD"). |
| 195 | threshold: Drift threshold in beats (default 0.1). |
| 196 | track: Restrict analysis to a named instrument track. |
| 197 | section: Restrict analysis to a named musical section. |
| 198 | limit: Maximum number of commits to include (default 10). |
| 199 | |
| 200 | Returns: |
| 201 | A :class:`GrooveCheckResult` with per-commit metrics and summary fields. |
| 202 | """ |
| 203 | entries = build_stub_entries( |
| 204 | threshold=threshold, |
| 205 | track=track, |
| 206 | section=section, |
| 207 | limit=limit, |
| 208 | ) |
| 209 | flagged = [e for e in entries if e.status != GrooveStatus.OK] |
| 210 | worst = max(entries, key=lambda e: e.drift_delta, default=None) |
| 211 | worst_commit = worst.commit if worst and worst.drift_delta > 0 else "" |
| 212 | |
| 213 | result = GrooveCheckResult( |
| 214 | commit_range=commit_range, |
| 215 | threshold=threshold, |
| 216 | total_commits=len(entries), |
| 217 | flagged_commits=len(flagged), |
| 218 | worst_commit=worst_commit, |
| 219 | entries=tuple(entries), |
| 220 | ) |
| 221 | |
| 222 | logger.info( |
| 223 | "✅ Groove-check: range=%s threshold=%.3f flagged=%d/%d worst=%s", |
| 224 | commit_range, |
| 225 | threshold, |
| 226 | result.flagged_commits, |
| 227 | result.total_commits, |
| 228 | result.worst_commit or "none", |
| 229 | ) |
| 230 | return result |