cgcardona / muse public
similarity.py python
465 lines 14.8 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """muse similarity — compute musical similarity score between two commits.
2
3 Compares two Muse commits across up to five musical dimensions and
4 produces per-dimension scores plus a weighted overall score. An AI
5 agent can use this output to calibrate how much new material to generate:
6 a similarity of 0.9 suggests a small variation; 0.4 suggests a major
7 rework.
8
9 Dimensions
10 ----------
11 - harmonic — key, chord vocabulary, chord progression similarity
12 - rhythmic — tempo, time signature, rhythmic density
13 - melodic — motif reuse, interval contour, pitch range
14 - structural — section layout (intro, verse, bridge, outro lengths)
15 - dynamic — velocity profile, crescendo/decrescendo patterns
16
17 Scores are normalized to [0.0, 1.0]:
18 1.0 = identical
19 0.0 = completely different
20
21 Output (default)::
22
23 Similarity: HEAD~10 vs HEAD
24
25 Harmonic: 0.45 ██████████░░░░░░░░░░ (key modulation, new chords)
26 Rhythmic: 0.89 ████████████████████ (same tempo, slightly more swing)
27 Melodic: 0.72 ██████████████████░░ (same motifs, extended range)
28 Structural: 0.65 █████████████░░░░░░░ (bridge added, intro shortened)
29 Dynamic: 0.55 ███████████░░░░░░░░░ (much louder, crescendo added)
30
31 Overall: 0.65 (Significantly different — major rework)
32
33 Flags
34 -----
35 COMMIT-A First commit ref (required).
36 COMMIT-B Second commit ref (required).
37 --dimensions TEXT Comma-separated subset of dimensions (default: all five).
38 --section TEXT Scope comparison to a named section/region.
39 --track TEXT Scope comparison to a specific track.
40 --json Emit machine-readable JSON.
41 --threshold FLOAT Exit 1 if overall similarity < threshold (for scripting).
42 """
43 from __future__ import annotations
44
45 import asyncio
46 import json
47 import logging
48 import pathlib
49 from typing import Optional
50
51 import typer
52 from sqlalchemy.ext.asyncio import AsyncSession
53 from typing_extensions import TypedDict
54
55 from maestro.muse_cli._repo import require_repo
56 from maestro.muse_cli.db import open_session
57 from maestro.muse_cli.errors import ExitCode
58
59 logger = logging.getLogger(__name__)
60
61 app = typer.Typer()
62
63 # ---------------------------------------------------------------------------
64 # Constants
65 # ---------------------------------------------------------------------------
66
67 DIMENSION_NAMES: tuple[str, ...] = (
68 "harmonic",
69 "rhythmic",
70 "melodic",
71 "structural",
72 "dynamic",
73 )
74
75 _ALL_DIMENSIONS: frozenset[str] = frozenset(DIMENSION_NAMES)
76
77 # Dimension weights for computing the overall score.
78 # Harmonic and melodic carry the most musical identity.
79 _DIMENSION_WEIGHTS: dict[str, float] = {
80 "harmonic": 0.25,
81 "rhythmic": 0.20,
82 "melodic": 0.25,
83 "structural": 0.15,
84 "dynamic": 0.15,
85 }
86
87 # Score thresholds for the human-readable quality label.
88 _LABEL_THRESHOLDS: tuple[tuple[float, str], ...] = (
89 (0.90, "Nearly identical — minimal change"),
90 (0.75, "Highly similar — subtle variation"),
91 (0.60, "Moderately similar — noticeable changes"),
92 (0.40, "Significantly different — major rework"),
93 (0.00, "Completely different — new direction"),
94 )
95
96 _BAR_WIDTH = 20 # characters in the progress bar
97
98
99 # ---------------------------------------------------------------------------
100 # Named result types (stable CLI contract)
101 # ---------------------------------------------------------------------------
102
103
104 class DimensionScore(TypedDict):
105 """Score for a single musical dimension.
106
107 Contract:
108 dimension — one of the five canonical dimension names
109 score — normalized similarity in [0.0, 1.0]
110 note — brief human-readable interpretation of the difference
111 """
112
113 dimension: str
114 score: float
115 note: str
116
117
118 class SimilarityResult(TypedDict):
119 """Overall similarity result between two commits.
120
121 Contract:
122 commit_a — first commit ref as provided by the caller
123 commit_b — second commit ref as provided by the caller
124 dimensions — list of per-dimension scores (may be a subset)
125 overall — weighted overall similarity in [0.0, 1.0]
126 label — human-readable summary of the overall score
127 max_divergence — dimension name with the lowest score
128 """
129
130 commit_a: str
131 commit_b: str
132 dimensions: list[DimensionScore]
133 overall: float
134 label: str
135 max_divergence: str
136
137
138 # ---------------------------------------------------------------------------
139 # Stub data — realistic placeholder until MIDI data is queryable per-commit
140 # ---------------------------------------------------------------------------
141
142 # Stub per-dimension scores and interpretive notes.
143 _STUB_DIMENSION_DATA: dict[str, tuple[float, str]] = {
144 "harmonic": (0.45, "key modulation, new chords"),
145 "rhythmic": (0.89, "same tempo, slightly more swing"),
146 "melodic": (0.72, "same motifs, extended range"),
147 "structural": (0.65, "bridge added, intro shortened"),
148 "dynamic": (0.55, "much louder, crescendo added"),
149 }
150
151
152 def _stub_dimension_scores(dimensions: frozenset[str]) -> list[DimensionScore]:
153 """Return stub DimensionScore rows for the requested dimensions.
154
155 The ordering mirrors DIMENSION_NAMES so output is always stable.
156 """
157 return [
158 DimensionScore(
159 dimension=dim,
160 score=_STUB_DIMENSION_DATA[dim][0],
161 note=_STUB_DIMENSION_DATA[dim][1],
162 )
163 for dim in DIMENSION_NAMES
164 if dim in dimensions
165 ]
166
167
168 # ---------------------------------------------------------------------------
169 # Score computation helpers
170 # ---------------------------------------------------------------------------
171
172
173 def _weighted_overall(scores: list[DimensionScore]) -> float:
174 """Compute a weighted overall similarity score.
175
176 Uses _DIMENSION_WEIGHTS when a dimension is in the standard set;
177 falls back to equal weighting for any custom/unknown dimension.
178 """
179 if not scores:
180 return 0.0
181 total_weight = sum(_DIMENSION_WEIGHTS.get(s["dimension"], 1.0) for s in scores)
182 weighted_sum = sum(
183 s["score"] * _DIMENSION_WEIGHTS.get(s["dimension"], 1.0) for s in scores
184 )
185 if total_weight == 0.0:
186 return 0.0
187 return round(weighted_sum / total_weight, 4)
188
189
190 def _overall_label(overall: float) -> str:
191 """Return a human-readable label for an overall similarity score."""
192 for threshold, label in _LABEL_THRESHOLDS:
193 if overall >= threshold:
194 return label
195 return _LABEL_THRESHOLDS[-1][1]
196
197
198 def _max_divergence_dimension(scores: list[DimensionScore]) -> str:
199 """Return the name of the dimension with the lowest similarity score."""
200 if not scores:
201 return ""
202 return min(scores, key=lambda s: s["score"])["dimension"]
203
204
205 def build_similarity_result(
206 commit_a: str,
207 commit_b: str,
208 scores: list[DimensionScore],
209 ) -> SimilarityResult:
210 """Assemble a complete SimilarityResult from scored dimensions.
211
212 Separated from the async core so tests can validate the computation
213 without I/O.
214
215 Args:
216 commit_a: First commit ref.
217 commit_b: Second commit ref.
218 scores: Per-dimension scores (may be a subset of all five).
219
220 Returns:
221 A SimilarityResult with all fields populated.
222 """
223 overall = _weighted_overall(scores)
224 return SimilarityResult(
225 commit_a=commit_a,
226 commit_b=commit_b,
227 dimensions=scores,
228 overall=overall,
229 label=_overall_label(overall),
230 max_divergence=_max_divergence_dimension(scores),
231 )
232
233
234 # ---------------------------------------------------------------------------
235 # Rendering helpers
236 # ---------------------------------------------------------------------------
237
238
239 def _bar(score: float, width: int = _BAR_WIDTH) -> str:
240 """Render a Unicode block progress bar for a score in [0, 1]."""
241 filled = round(score * width)
242 return "\u2588" * filled + "\u2591" * (width - filled)
243
244
245 def render_similarity_text(result: SimilarityResult) -> str:
246 """Render a human-readable similarity report.
247
248 Called by the CLI and by tests so the rendering contract can be
249 validated independently of Typer.
250
251 Args:
252 result: A fully populated SimilarityResult.
253
254 Returns:
255 Multi-line string ready to echo to stdout.
256 """
257 lines: list[str] = [
258 f"Similarity: {result['commit_a']} vs {result['commit_b']}",
259 "",
260 ]
261
262 label_width = max((len(s["dimension"]) for s in result["dimensions"]), default=0) + 1
263 for score in result["dimensions"]:
264 dim_label = f"{score['dimension'].capitalize()}:".ljust(label_width + 1)
265 bar = _bar(score["score"])
266 lines.append(
267 f" {dim_label} {score['score']:.2f} {bar} ({score['note']})"
268 )
269
270 lines.append("")
271 lines.append(f" Overall: {result['overall']:.2f} ({result['label']})")
272
273 if result["max_divergence"]:
274 lines.append(
275 f" Max divergence: {result['max_divergence']} dimension"
276 )
277
278 return "\n".join(lines)
279
280
281 def render_similarity_json(result: SimilarityResult) -> str:
282 """Render a SimilarityResult as indented JSON."""
283 return json.dumps(dict(result), indent=2)
284
285
286 # ---------------------------------------------------------------------------
287 # Testable async core
288 # ---------------------------------------------------------------------------
289
290
291 async def _similarity_async(
292 *,
293 root: pathlib.Path,
294 session: AsyncSession,
295 commit_a: str,
296 commit_b: str,
297 dimensions: frozenset[str],
298 section: Optional[str],
299 track: Optional[str],
300 threshold: Optional[float],
301 as_json: bool,
302 ) -> int:
303 """Core similarity logic — fully injectable for tests.
304
305 Resolves both commit refs against the .muse/ directory, produces
306 stub per-dimension scores for the requested dimensions, assembles a
307 SimilarityResult, renders output, and returns an exit code.
308
309 Args:
310 root: Repository root (directory containing .muse/).
311 session: Open async DB session (reserved for full implementation).
312 commit_a: First commit ref.
313 commit_b: Second commit ref.
314 dimensions: Set of dimension names to compute.
315 section: Named section to scope comparison (stub: noted).
316 track: Named track to scope comparison (stub: noted).
317 threshold: Exit 1 if overall < threshold; None means no check.
318 as_json: Emit JSON instead of text.
319
320 Returns:
321 Integer exit code — 0 on success, 1 if below threshold.
322 """
323 muse_dir = root / ".muse"
324 _head_ref = (muse_dir / "HEAD").read_text().strip()
325
326 if section:
327 typer.echo(
328 f"WARNING: --section {section}: section-scoped comparison not yet implemented."
329 )
330 if track:
331 typer.echo(
332 f"WARNING: --track {track}: track-scoped comparison not yet implemented."
333 )
334
335 scores = _stub_dimension_scores(dimensions)
336 result = build_similarity_result(commit_a, commit_b, scores)
337
338 if as_json:
339 typer.echo(render_similarity_json(result))
340 else:
341 typer.echo(render_similarity_text(result))
342
343 if threshold is not None and result["overall"] < threshold:
344 return int(1)
345
346 return int(ExitCode.SUCCESS)
347
348
349 # ---------------------------------------------------------------------------
350 # Typer command
351 # ---------------------------------------------------------------------------
352
353
354 @app.callback(invoke_without_command=True)
355 def similarity(
356 ctx: typer.Context,
357 commit_a: str = typer.Argument(
358 ...,
359 help="First commit ref to compare.",
360 metavar="COMMIT-A",
361 ),
362 commit_b: str = typer.Argument(
363 ...,
364 help="Second commit ref to compare.",
365 metavar="COMMIT-B",
366 ),
367 dimensions: Optional[str] = typer.Option(
368 None,
369 "--dimensions",
370 help=(
371 "Comma-separated list of dimensions to compare. "
372 "Valid: harmonic,rhythmic,melodic,structural,dynamic. "
373 "Default: all five."
374 ),
375 metavar="DIMS",
376 ),
377 section: Optional[str] = typer.Option(
378 None,
379 "--section",
380 help="Scope comparison to a named section/region.",
381 metavar="TEXT",
382 ),
383 track: Optional[str] = typer.Option(
384 None,
385 "--track",
386 help="Scope comparison to a specific track.",
387 metavar="TEXT",
388 ),
389 as_json: bool = typer.Option(
390 False,
391 "--json",
392 help="Emit machine-readable JSON output.",
393 ),
394 threshold: Optional[float] = typer.Option(
395 None,
396 "--threshold",
397 help=(
398 "Exit 1 if the overall similarity score is below this value. "
399 "Useful in scripts to detect major reworks."
400 ),
401 metavar="FLOAT",
402 ),
403 ) -> None:
404 """Compute musical similarity score between two commits.
405
406 Produces per-dimension scores (harmonic, rhythmic, melodic, structural,
407 dynamic) and a weighted overall score in [0.0, 1.0].
408
409 Example::
410
411 muse similarity HEAD~10 HEAD
412 muse similarity HEAD~10 HEAD --dimensions harmonic,rhythmic
413 muse similarity HEAD~10 HEAD --json
414 muse similarity HEAD~10 HEAD --threshold 0.5
415 """
416 # -- Validate flags first — before repo detection so bad input fails fast --
417 active_dimensions: frozenset[str]
418 if dimensions is not None:
419 requested = frozenset(
420 d.strip().lower() for d in dimensions.split(",") if d.strip()
421 )
422 invalid = requested - _ALL_DIMENSIONS
423 if invalid:
424 typer.echo(
425 f"Unknown dimension(s): {', '.join(sorted(invalid))}. "
426 f"Valid: {', '.join(DIMENSION_NAMES)}"
427 )
428 raise typer.Exit(code=ExitCode.USER_ERROR)
429 if not requested:
430 typer.echo("--dimensions must specify at least one dimension.")
431 raise typer.Exit(code=ExitCode.USER_ERROR)
432 active_dimensions = requested
433 else:
434 active_dimensions = _ALL_DIMENSIONS
435
436 if threshold is not None and not (0.0 <= threshold <= 1.0):
437 typer.echo(f"--threshold {threshold!r} out of range [0.0, 1.0].")
438 raise typer.Exit(code=ExitCode.USER_ERROR)
439
440 root = require_repo()
441
442 async def _run() -> int:
443 async with open_session() as session:
444 return await _similarity_async(
445 root=root,
446 session=session,
447 commit_a=commit_a,
448 commit_b=commit_b,
449 dimensions=active_dimensions,
450 section=section,
451 track=track,
452 threshold=threshold,
453 as_json=as_json,
454 )
455
456 try:
457 exit_code = asyncio.run(_run())
458 if exit_code != 0:
459 raise typer.Exit(code=exit_code)
460 except typer.Exit:
461 raise
462 except Exception as exc:
463 typer.echo(f"muse similarity failed: {exc}")
464 logger.error("muse similarity error: %s", exc, exc_info=True)
465 raise typer.Exit(code=ExitCode.INTERNAL_ERROR)