gabriel / musehub public
musehub_analysis.py python
1809 lines 66.1 KB
Raw
51c9a460 chore(deps): upgrade to Python 3.12 with latest dependencies Gabriel Cardona <gabriel@tellurstori.com> 6d ago
1 """Muse Hub Analysis Service — structured musical analysis for agent consumption.
2
3 This module is the single orchestration point for all 13 analysis dimensions.
4 Route handlers delegate here; no business logic lives in routes.
5
6 Why this exists
7 ---------------
8 AI agents need structured, typed JSON data to make informed composition
9 decisions. HTML analysis pages are not machine-readable. This service
10 bridges the gap by returning fully-typed Pydantic models for every musical
11 dimension of a Muse commit.
12
13 Stub implementation
14 -------------------
15 Full MIDI content analysis will be wired in once Storpheus exposes a
16 per-dimension introspection route. Until then, the service returns
17 deterministic stub data keyed on the ``ref`` value — deterministic so that
18 agents receive consistent responses across retries and across sessions.
19
20 The stub data is musically realistic: values are drawn from realistic ranges
21 for jazz/soul/pop production and are internally consistent within each
22 dimension (e.g. the key reported by ``harmony`` matches the key reported by
23 ``key``).
24
25 Boundary rules
26 --------------
27 - Pure data — no side effects, no external I/O beyond reading ``ref``.
28 - Must NOT import StateStore, EntityRegistry, or executor modules.
29 - Must NOT import LLM handlers or muse_* pipeline modules.
30 """
31 from __future__ import annotations
32
33 import hashlib
34 import logging
35 from collections import defaultdict
36 from collections.abc import Callable
37 from datetime import datetime, timezone
38 from typing import Literal
39
40 from sqlalchemy.ext.asyncio import AsyncSession
41
42 from musehub.models.musehub import (
43 ArrangementCellData,
44 ArrangementColumnSummary,
45 ArrangementMatrixResponse,
46 ArrangementRowSummary,
47 )
48 from musehub.models.musehub_analysis import (
49 ALL_DIMENSIONS,
50 AggregateAnalysisResponse,
51 AlternateKey,
52 AnalysisFilters,
53 AnalysisResponse,
54 CadenceEvent,
55 ChangedDimension,
56 ChordEvent,
57 ChordMapData,
58 CommitEmotionSnapshot,
59 CompareDimension,
60 CompareResult,
61 ContourData,
62 ContextResult,
63 DimensionData,
64 DivergenceData,
65 DivergenceDimension,
66 DivergenceResult,
67 DynamicArc,
68 DynamicsData,
69 DynamicsPageData,
70 EmotionData,
71 EmotionDelta8D,
72 EmotionDiffResponse,
73 EmotionDrift,
74 EmotionMapPoint,
75 EmotionMapResponse,
76 EmotionVector,
77 EmotionVector8D,
78 FormData,
79 FormStructureResponse,
80 GrooveData,
81 HarmonyAnalysisResponse,
82 HarmonyData,
83 HarmonyModulationEvent,
84 IrregularSection,
85 KeyData,
86 MeterData,
87 ModulationPoint,
88 MotifEntry,
89 MotifRecurrenceCell,
90 MotifTransformation,
91 MotifsData,
92 RecallMatch,
93 RecallResponse,
94 RepetitionEntry,
95 RomanNumeralEvent,
96 SectionEntry,
97 SectionMapEntry,
98 SectionSimilarityHeatmap,
99 RefSimilarityDimensions,
100 RefSimilarityResponse,
101 SimilarCommit,
102 SimilarityData,
103 TempoChange,
104 TempoData,
105 TrackDynamicsProfile,
106 VelocityEvent,
107 )
108
109 logger = logging.getLogger(__name__)
110
111 # ---------------------------------------------------------------------------
112 # Stub data constants — musically realistic, deterministic by ref hash
113 # ---------------------------------------------------------------------------
114
115 _MODES = ["major", "minor", "dorian", "mixolydian", "lydian", "phrygian"]
116 _EMOTIONS = ["joyful", "melancholic", "tense", "serene", "energetic", "brooding"]
117 _FORMS = ["AABA", "verse-chorus", "through-composed", "rondo", "binary", "ternary"]
118 _GROOVES = ["straight", "swing", "shuffled", "half-time", "double-time"]
119 _TONICS = ["C", "F", "G", "D", "Bb", "Eb"]
120 _DYNAMIC_ARCS: list[DynamicArc] = [
121 "flat", "terraced", "crescendo", "decrescendo", "swell", "hairpin",
122 ]
123 _DEFAULT_TRACKS = ["bass", "keys", "drums", "melody", "pads"]
124
125
126 def _ref_hash(ref: str) -> int:
127 """Derive a stable integer seed from a ref string for deterministic stubs."""
128 return int(hashlib.md5(ref.encode()).hexdigest(), 16) # noqa: S324 — non-crypto use
129
130
131 def _pick(seed: int, items: list[str], offset: int = 0) -> str:
132 return items[(seed + offset) % len(items)]
133
134
135 def _utc_now() -> datetime:
136 return datetime.now(tz=timezone.utc)
137
138
139 # ---------------------------------------------------------------------------
140 # Per-dimension stub builders
141 # ---------------------------------------------------------------------------
142
143
144 def _build_harmony(ref: str, track: str | None, section: str | None) -> HarmonyData:
145 """Build stub harmonic analysis. Deterministic for a given ref."""
146 seed = _ref_hash(ref)
147 tonic = _pick(seed, _TONICS)
148 mode = _pick(seed, _MODES)
149 total_beats = 32
150
151 progression = [
152 ChordEvent(beat=0.0, chord=f"{tonic}maj7", function="Imaj7", tension=0.1),
153 ChordEvent(beat=4.0, chord="Am7", function="VIm7", tension=0.2),
154 ChordEvent(beat=8.0, chord="Dm7", function="IIm7", tension=0.25),
155 ChordEvent(beat=12.0, chord="G7", function="V7", tension=0.6),
156 ChordEvent(beat=16.0, chord=f"{tonic}maj7", function="Imaj7", tension=0.1),
157 ChordEvent(beat=20.0, chord="Em7b5", function="VIIm7b5", tension=0.7),
158 ChordEvent(beat=24.0, chord="A7", function="V7/IIm", tension=0.65),
159 ChordEvent(beat=28.0, chord="Dm7", function="IIm7", tension=0.25),
160 ]
161
162 tension_curve = [
163 round(0.1 + 0.5 * abs((i - total_beats / 2) / (total_beats / 2)) * (seed % 3 + 1) / 3, 4)
164 for i in range(total_beats)
165 ]
166
167 modulation_points = (
168 [ModulationPoint(beat=16.0, from_key=f"{tonic} {mode}", to_key=f"G {mode}", confidence=0.72)]
169 if seed % 3 == 0
170 else []
171 )
172
173 return HarmonyData(
174 tonic=tonic,
175 mode=mode,
176 key_confidence=round(0.7 + (seed % 30) / 100, 4),
177 chord_progression=progression,
178 tension_curve=tension_curve,
179 modulation_points=modulation_points,
180 total_beats=total_beats,
181 )
182
183
184 def _build_dynamics(ref: str, track: str | None, section: str | None) -> DynamicsData:
185 seed = _ref_hash(ref)
186 base_vel = 64 + (seed % 32)
187 peak = min(127, base_vel + 30)
188 low = max(20, base_vel - 20)
189
190 curve = [
191 VelocityEvent(
192 beat=float(i * 2),
193 velocity=min(127, max(20, base_vel + (seed >> i) % 20 - 10)),
194 )
195 for i in range(16)
196 ]
197
198 events = ["crescendo@8", "sfz@16"] if seed % 2 == 0 else ["diminuendo@12", "fp@0"]
199
200 return DynamicsData(
201 peak_velocity=peak,
202 mean_velocity=round(float(base_vel), 2),
203 min_velocity=low,
204 dynamic_range=peak - low,
205 velocity_curve=curve,
206 dynamic_events=events,
207 )
208
209
210 _CONTOUR_LABELS = [
211 "ascending-step",
212 "descending-step",
213 "arch",
214 "valley",
215 "oscillating",
216 "static",
217 ]
218 _TRANSFORMATION_TYPES = ["inversion", "retrograde", "retrograde-inversion", "transposition"]
219 _MOTIF_TRACKS = ["melody", "bass", "keys", "strings", "brass"]
220 _MOTIF_SECTIONS = ["intro", "verse_1", "chorus", "verse_2", "outro"]
221
222
223 def _invert_intervals(intervals: list[int]) -> list[int]:
224 """Return the melodic inversion (negate all semitone intervals)."""
225 return [-x for x in intervals]
226
227
228 def _retrograde_intervals(intervals: list[int]) -> list[int]:
229 """Return the retrograde (reversed interval sequence)."""
230 return list(reversed(intervals))
231
232
233 def _build_motifs(ref: str, track: str | None, section: str | None) -> MotifsData:
234 """Build stub motif analysis with transformations, contour, and recurrence grid.
235
236 Deterministic for a given ``ref`` value. Produces 2–4 motifs, each with:
237 - Original interval sequence and occurrence beats
238 - Melodic contour label (arch, valley, oscillating, etc.)
239 - All tracks where the motif or its transformations appear
240 - Up to 3 transformations (inversion, retrograde, transposition)
241 - Flat track×section recurrence grid for heatmap rendering
242 """
243 seed = _ref_hash(ref)
244 n_motifs = 2 + (seed % 3)
245 all_tracks = _MOTIF_TRACKS[: 2 + (seed % 3)]
246 sections = _MOTIF_SECTIONS
247
248 motifs: list[MotifEntry] = []
249 for i in range(n_motifs):
250 intervals = [2, -1, 3, -2][: 2 + i]
251 occurrences = [float(j * 8 + i * 2) for j in range(2 + (seed % 2))]
252 contour_label = _pick(seed, _CONTOUR_LABELS, offset=i)
253 primary_track = track or all_tracks[i % len(all_tracks)]
254
255 # Cross-track sharing: motif appears in 1–3 tracks
256 n_sharing_tracks = 1 + (seed + i) % min(3, len(all_tracks))
257 sharing_tracks = [all_tracks[(i + k) % len(all_tracks)] for k in range(n_sharing_tracks)]
258 if primary_track not in sharing_tracks:
259 sharing_tracks = [primary_track] + sharing_tracks[: n_sharing_tracks - 1]
260
261 # Build transformations
262 transformations: list[MotifTransformation] = []
263 inv_occurrences = [float(j * 8 + i * 2 + 4) for j in range(1 + (seed % 2))]
264 transformations.append(
265 MotifTransformation(
266 transformation_type="inversion",
267 intervals=_invert_intervals(intervals),
268 transposition_semitones=0,
269 occurrences=inv_occurrences,
270 track=sharing_tracks[0],
271 )
272 )
273 if len(intervals) >= 2:
274 retro_occurrences = [float(j * 8 + i * 2 + 2) for j in range(1 + (seed % 2))]
275 transformations.append(
276 MotifTransformation(
277 transformation_type="retrograde",
278 intervals=_retrograde_intervals(intervals),
279 transposition_semitones=0,
280 occurrences=retro_occurrences,
281 track=sharing_tracks[-1],
282 )
283 )
284 if (seed + i) % 2 == 0:
285 transpose_by = 5 if (seed % 2 == 0) else 7
286 transpo_occurrences = [float(j * 16 + i * 2) for j in range(1 + (seed % 2))]
287 transformations.append(
288 MotifTransformation(
289 transformation_type="transposition",
290 intervals=[x for x in intervals],
291 transposition_semitones=transpose_by,
292 occurrences=transpo_occurrences,
293 track=sharing_tracks[min(1, len(sharing_tracks) - 1)],
294 )
295 )
296
297 # Build recurrence grid: track × section
298 recurrence_grid: list[MotifRecurrenceCell] = []
299 for t in all_tracks:
300 for s in sections:
301 # Original present in primary track, first two sections
302 if t == primary_track and s in sections[:2]:
303 recurrence_grid.append(
304 MotifRecurrenceCell(
305 track=t,
306 section=s,
307 present=True,
308 occurrence_count=1 + (seed % 2),
309 transformation_types=["original"],
310 )
311 )
312 # Inversion in sharing tracks at chorus
313 elif t in sharing_tracks and s == "chorus":
314 recurrence_grid.append(
315 MotifRecurrenceCell(
316 track=t,
317 section=s,
318 present=True,
319 occurrence_count=1,
320 transformation_types=["inversion"],
321 )
322 )
323 # Transposition in bridge / outro for certain motifs
324 elif (seed + i) % 2 == 0 and t in sharing_tracks and s == "outro":
325 recurrence_grid.append(
326 MotifRecurrenceCell(
327 track=t,
328 section=s,
329 present=True,
330 occurrence_count=1,
331 transformation_types=["transposition"],
332 )
333 )
334 else:
335 recurrence_grid.append(
336 MotifRecurrenceCell(
337 track=t,
338 section=s,
339 present=False,
340 occurrence_count=0,
341 transformation_types=[],
342 )
343 )
344
345 motifs.append(
346 MotifEntry(
347 motif_id=f"M{i + 1:02d}",
348 intervals=intervals,
349 length_beats=float(2 + i),
350 occurrence_count=len(occurrences),
351 occurrences=occurrences,
352 track=primary_track,
353 contour_label=contour_label,
354 tracks=sharing_tracks,
355 transformations=transformations,
356 recurrence_grid=recurrence_grid,
357 )
358 )
359
360 return MotifsData(
361 total_motifs=len(motifs),
362 motifs=motifs,
363 sections=sections,
364 all_tracks=all_tracks,
365 )
366
367
368 def _build_form(ref: str, track: str | None, section: str | None) -> FormData:
369 seed = _ref_hash(ref)
370 form_label = _pick(seed, _FORMS)
371 sections = [
372 SectionEntry(label="intro", function="exposition", start_beat=0.0, end_beat=8.0, length_beats=8.0),
373 SectionEntry(label="verse_1", function="statement", start_beat=8.0, end_beat=24.0, length_beats=16.0),
374 SectionEntry(label="chorus", function="climax", start_beat=24.0, end_beat=40.0, length_beats=16.0),
375 SectionEntry(label="verse_2", function="restatement", start_beat=40.0, end_beat=56.0, length_beats=16.0),
376 SectionEntry(label="outro", function="resolution", start_beat=56.0, end_beat=64.0, length_beats=8.0),
377 ]
378 return FormData(form_label=form_label, total_beats=64, sections=sections)
379
380
381 def _build_groove(ref: str, track: str | None, section: str | None) -> GrooveData:
382 seed = _ref_hash(ref)
383 style = _pick(seed, _GROOVES)
384 swing = 0.5 if style == "straight" else round(0.55 + (seed % 20) / 100, 4)
385 bpm = round(80.0 + (seed % 80), 1)
386 return GrooveData(
387 swing_factor=swing,
388 grid_resolution="1/16" if style == "straight" else "1/8T",
389 onset_deviation=round(0.01 + (seed % 10) / 200, 4),
390 groove_score=round(0.6 + (seed % 40) / 100, 4),
391 style=style,
392 bpm=bpm,
393 )
394
395
396 def _build_emotion(ref: str, track: str | None, section: str | None) -> EmotionData:
397 seed = _ref_hash(ref)
398 emotion = _pick(seed, _EMOTIONS)
399 valence_map: dict[str, float] = {
400 "joyful": 0.8, "melancholic": -0.5, "tense": -0.3,
401 "serene": 0.4, "energetic": 0.6, "brooding": -0.7,
402 }
403 arousal_map: dict[str, float] = {
404 "joyful": 0.7, "melancholic": 0.3, "tense": 0.8,
405 "serene": 0.2, "energetic": 0.9, "brooding": 0.5,
406 }
407 return EmotionData(
408 valence=valence_map[emotion],
409 arousal=arousal_map[emotion],
410 tension=round(0.1 + (seed % 60) / 100, 4),
411 primary_emotion=emotion,
412 confidence=round(0.65 + (seed % 35) / 100, 4),
413 )
414
415
416 def _build_chord_map(ref: str, track: str | None, section: str | None) -> ChordMapData:
417 harmony = _build_harmony(ref, track, section)
418 return ChordMapData(
419 progression=harmony.chord_progression,
420 total_chords=len(harmony.chord_progression),
421 total_beats=harmony.total_beats,
422 )
423
424
425 def _build_contour(ref: str, track: str | None, section: str | None) -> ContourData:
426 seed = _ref_hash(ref)
427 shapes = ["arch", "ascending", "descending", "flat", "wave"]
428 shape = _pick(seed, shapes)
429 base_pitch = 60 + (seed % 12)
430 pitch_curve = [
431 round(base_pitch + 5 * (i / 16) * (1 if seed % 2 == 0 else -1) + (seed >> i) % 3, 1)
432 for i in range(16)
433 ]
434 return ContourData(
435 shape=shape,
436 direction_changes=1 + (seed % 4),
437 peak_beat=float(4 + (seed % 12)),
438 valley_beat=float(seed % 8),
439 overall_direction="up" if seed % 3 == 0 else ("down" if seed % 3 == 1 else "flat"),
440 pitch_curve=pitch_curve,
441 )
442
443
444 def _build_key(ref: str, track: str | None, section: str | None) -> KeyData:
445 seed = _ref_hash(ref)
446 tonic = _pick(seed, _TONICS)
447 mode = _pick(seed, _MODES[:2])
448 rel_choices = ["A", "D", "E", "B", "G", "C"]
449 relative = f"{_pick(seed + 3, rel_choices)}m" if mode == "major" else f"{tonic}m"
450 alternates = [
451 AlternateKey(
452 tonic=_pick(seed + 2, ["G", "D", "A", "E", "Bb"]),
453 mode="dorian",
454 confidence=round(0.3 + (seed % 20) / 100, 4),
455 )
456 ]
457 return KeyData(
458 tonic=tonic,
459 mode=mode,
460 confidence=round(0.75 + (seed % 25) / 100, 4),
461 relative_key=relative,
462 alternate_keys=alternates,
463 )
464
465
466 def _build_tempo(ref: str, track: str | None, section: str | None) -> TempoData:
467 seed = _ref_hash(ref)
468 bpm = round(80.0 + (seed % 80), 1)
469 stability = round(0.7 + (seed % 30) / 100, 4)
470 feels = ["straight", "laid-back", "rushing"]
471 feel = _pick(seed, feels)
472 changes = (
473 [TempoChange(beat=32.0, bpm=round(bpm * 1.05, 1))]
474 if seed % 4 == 0
475 else []
476 )
477 return TempoData(bpm=bpm, stability=stability, time_feel=feel, tempo_changes=changes)
478
479
480 def _build_meter(ref: str, track: str | None, section: str | None) -> MeterData:
481 seed = _ref_hash(ref)
482 sigs = ["4/4", "3/4", "6/8", "5/4", "7/8"]
483 sig = _pick(seed, sigs[:2])
484 is_compound = sig in ("6/8", "12/8")
485 profile_44 = [1.0, 0.2, 0.6, 0.2]
486 profile_34 = [1.0, 0.3, 0.5]
487 profile = profile_44 if sig == "4/4" else profile_34
488 irregular: list[IrregularSection] = (
489 [IrregularSection(start_beat=24.0, end_beat=25.0, time_signature="5/4")]
490 if seed % 5 == 0
491 else []
492 )
493 return MeterData(
494 time_signature=sig,
495 irregular_sections=irregular,
496 beat_strength_profile=profile,
497 is_compound=is_compound,
498 )
499
500
501 def _build_similarity(ref: str, track: str | None, section: str | None) -> SimilarityData:
502 seed = _ref_hash(ref)
503 n = 1 + (seed % 3)
504 similar = [
505 SimilarCommit(
506 ref=f"commit_{hashlib.md5(f'{ref}{i}'.encode()).hexdigest()[:8]}", # noqa: S324
507 score=round(0.5 + (seed >> i) % 50 / 100, 4),
508 shared_motifs=[f"M{j + 1:02d}" for j in range(1 + i % 2)],
509 commit_message=f"Add {'bridge' if i == 0 else 'variation'} section",
510 )
511 for i in range(n)
512 ]
513 return SimilarityData(similar_commits=similar, embedding_dimensions=128)
514
515
516 def _build_divergence(ref: str, track: str | None, section: str | None) -> DivergenceData:
517 seed = _ref_hash(ref)
518 score = round((seed % 60) / 100, 4)
519 changed = [
520 ChangedDimension(
521 dimension="harmony",
522 change_magnitude=round(0.2 + (seed % 40) / 100, 4),
523 description="Key shifted from C major to F major",
524 ),
525 ChangedDimension(
526 dimension="tempo",
527 change_magnitude=round(0.1 + (seed % 20) / 100, 4),
528 description="BPM increased by ~8%",
529 ),
530 ]
531 return DivergenceData(
532 divergence_score=score,
533 base_ref=f"parent:{ref[:8]}",
534 changed_dimensions=changed,
535 )
536
537
538 # ---------------------------------------------------------------------------
539 # Dimension dispatch table
540 # ---------------------------------------------------------------------------
541
542 # Each builder has signature (ref: str, track: str | None, section: str | None) -> DimensionData
543 _DimBuilder = Callable[[str, str | None, str | None], DimensionData]
544
545 _BUILDERS: dict[str, _DimBuilder] = {
546 "harmony": _build_harmony,
547 "dynamics": _build_dynamics,
548 "motifs": _build_motifs,
549 "form": _build_form,
550 "groove": _build_groove,
551 "emotion": _build_emotion,
552 "chord-map": _build_chord_map,
553 "contour": _build_contour,
554 "key": _build_key,
555 "tempo": _build_tempo,
556 "meter": _build_meter,
557 "similarity": _build_similarity,
558 "divergence": _build_divergence,
559 }
560
561
562 # ---------------------------------------------------------------------------
563 # Public API
564 # ---------------------------------------------------------------------------
565
566
567 def compute_dimension(
568 dimension: str,
569 ref: str,
570 track: str | None = None,
571 section: str | None = None,
572 ) -> DimensionData:
573 """Compute analysis data for a single musical dimension.
574
575 Dispatches to the appropriate stub builder based on ``dimension``.
576 Returns a fully-typed Pydantic model for the given dimension.
577
578 Args:
579 dimension: One of the 13 supported dimension names.
580 ref: Muse commit ref (branch name, commit ID, or tag).
581 track: Optional instrument track filter.
582 section: Optional musical section filter.
583
584 Returns:
585 Dimension-specific Pydantic data model.
586
587 Raises:
588 ValueError: If ``dimension`` is not a supported analysis dimension.
589 """
590 builder = _BUILDERS.get(dimension)
591 if builder is None:
592 raise ValueError(f"Unknown analysis dimension: {dimension!r}")
593 return builder(ref, track, section)
594
595
596 def compute_analysis_response(
597 *,
598 repo_id: str,
599 dimension: str,
600 ref: str,
601 track: str | None = None,
602 section: str | None = None,
603 ) -> AnalysisResponse:
604 """Build a complete :class:`AnalysisResponse` envelope for one dimension.
605
606 This is the primary entry point for the single-dimension endpoint.
607
608 Args:
609 repo_id: Muse Hub repo UUID.
610 dimension: Analysis dimension name.
611 ref: Muse commit ref.
612 track: Optional track filter.
613 section: Optional section filter.
614
615 Returns:
616 :class:`AnalysisResponse` with typed ``data`` and filter metadata.
617 """
618 data = compute_dimension(dimension, ref, track, section)
619 response = AnalysisResponse(
620 dimension=dimension,
621 ref=ref,
622 computed_at=_utc_now(),
623 data=data,
624 filters_applied=AnalysisFilters(track=track, section=section),
625 )
626 logger.info("✅ analysis/%s repo=%s ref=%s", dimension, repo_id[:8], ref)
627 return response
628
629
630 def compute_form_structure(*, repo_id: str, ref: str) -> FormStructureResponse:
631 """Build a combined form and structure response for the UI visualisation.
632
633 Derives three complementary structural views from the ``form`` and ``meter``
634 stubs so the form-structure page can render a section map, a repetition
635 panel, and a section-comparison heatmap in a single API call.
636
637 Stub implementation: values are deterministic for a given ``ref`` and
638 musically consistent with the results of the ``form`` and ``meter``
639 analysis dimensions.
640
641 Args:
642 repo_id: Muse Hub repo UUID (used for logging only).
643 ref: Muse commit ref.
644
645 Returns:
646 :class:`FormStructureResponse` with section_map, repetition_structure,
647 and section_comparison fields populated.
648 """
649 seed = _ref_hash(ref)
650 form_data = _build_form(ref, None, None)
651 meter_data = _build_meter(ref, None, None)
652
653 time_sig = meter_data.time_signature
654 beats_per_bar = 3 if time_sig == "3/4" else (6 if time_sig == "6/8" else 4)
655
656 # Colour palette for section types — stable mapping
657 _SECTION_COLORS: dict[str, str] = {
658 "intro": "#1f6feb",
659 "verse": "#3fb950",
660 "chorus": "#f0883e",
661 "bridge": "#bc8cff",
662 "outro": "#8b949e",
663 "pre-chorus": "#ff7b72",
664 "breakdown": "#56d364",
665 }
666
667 def _section_color(label: str) -> str:
668 for key, color in _SECTION_COLORS.items():
669 if key in label.lower():
670 return color
671 return "#58a6ff"
672
673 # Build section map (convert beats → bars, 1-indexed)
674 section_map: list[SectionMapEntry] = []
675 for sec in form_data.sections:
676 start_bar = max(1, int(sec.start_beat / beats_per_bar) + 1)
677 end_bar = max(start_bar, int((sec.end_beat - 1) / beats_per_bar) + 1)
678 bar_count = end_bar - start_bar + 1
679 section_map.append(
680 SectionMapEntry(
681 label=sec.label,
682 function=sec.function,
683 start_bar=start_bar,
684 end_bar=end_bar,
685 bar_count=bar_count,
686 color_hint=_section_color(sec.label),
687 )
688 )
689
690 total_bars = max(1, int(form_data.total_beats / beats_per_bar))
691
692 # Build repetition structure — group sections by base label (strip _N suffix)
693 def _base_label(label: str) -> str:
694 """Strip numeric suffix from section label, e.g. 'verse_1' → 'verse'."""
695 parts = label.rsplit("_", 1)
696 if len(parts) == 2 and parts[1].isdigit():
697 return parts[0]
698 return label
699
700 groups: dict[str, list[int]] = defaultdict(list)
701 for entry in section_map:
702 groups[_base_label(entry.label)].append(entry.start_bar)
703
704 repetition_structure = [
705 RepetitionEntry(
706 pattern_label=pattern,
707 occurrences=bars,
708 occurrence_count=len(bars),
709 similarity_score=round(0.85 + (seed % 15) / 100, 4) if len(bars) > 1 else 1.0,
710 )
711 for pattern, bars in groups.items()
712 if len(bars) >= 1
713 ]
714
715 # Build section-comparison heatmap — symmetric cosine-similarity stub
716 labels = [s.label for s in section_map]
717 n = len(labels)
718 matrix: list[list[float]] = []
719 for i in range(n):
720 row: list[float] = []
721 for j in range(n):
722 if i == j:
723 row.append(1.0)
724 else:
725 # Sections with the same base label score higher
726 same_base = _base_label(labels[i]) == _base_label(labels[j])
727 base_sim = 0.75 if same_base else round(0.1 + (seed >> (i + j)) % 50 / 100, 4)
728 row.append(min(1.0, base_sim))
729 matrix.append(row)
730
731 section_comparison = SectionSimilarityHeatmap(labels=labels, matrix=matrix)
732
733 logger.info("✅ form-structure repo=%s ref=%s sections=%d", repo_id[:8], ref, len(section_map))
734 return FormStructureResponse(
735 repo_id=repo_id,
736 ref=ref,
737 form_label=form_data.form_label,
738 time_signature=time_sig,
739 beats_per_bar=beats_per_bar,
740 total_bars=total_bars,
741 section_map=section_map,
742 repetition_structure=repetition_structure,
743 section_comparison=section_comparison,
744 )
745
746
747 def _build_track_dynamics_profile(
748 ref: str,
749 track: str,
750 track_index: int,
751 ) -> TrackDynamicsProfile:
752 """Build a deterministic per-track dynamic profile for the dynamics page.
753
754 Seed is derived from ``ref`` XOR ``track_index`` so each track gets a
755 distinct but reproducible curve for the same ref.
756 """
757 seed = _ref_hash(ref) ^ (track_index * 0x9E3779B9)
758 base_vel = 50 + (seed % 50)
759 peak = min(127, base_vel + 20 + (seed % 30))
760 low = max(10, base_vel - 20 - (seed % 20))
761 mean = round(float((peak + low) / 2), 2)
762
763 curve = [
764 VelocityEvent(
765 beat=float(i * 2),
766 velocity=min(127, max(10, base_vel + (seed >> (i % 16)) % 25 - 12)),
767 )
768 for i in range(16)
769 ]
770
771 arc: DynamicArc = _DYNAMIC_ARCS[(seed + track_index) % len(_DYNAMIC_ARCS)]
772
773 return TrackDynamicsProfile(
774 track=track,
775 peak_velocity=peak,
776 min_velocity=low,
777 mean_velocity=mean,
778 velocity_range=peak - low,
779 arc=arc,
780 velocity_curve=curve,
781 )
782
783
784 def compute_dynamics_page_data(
785 *,
786 repo_id: str,
787 ref: str,
788 track: str | None = None,
789 section: str | None = None,
790 ) -> DynamicsPageData:
791 """Build per-track dynamics data for the Dynamics Analysis page.
792
793 Returns one :class:`TrackDynamicsProfile` per active track, or a single
794 entry when ``track`` filter is applied. Each profile includes a velocity
795 curve suitable for rendering a profile graph, an arc classification badge,
796 and peak/range metrics for the loudness comparison bar chart.
797
798 Args:
799 repo_id: Muse Hub repo UUID.
800 ref: Muse commit ref (branch name, commit ID, or tag).
801 track: Optional track filter — if set, only that track is returned.
802 section: Optional section filter (recorded in ``filters_applied``).
803
804 Returns:
805 :class:`DynamicsPageData` with per-track profiles.
806 """
807 tracks_to_include = [track] if track else _DEFAULT_TRACKS
808 profiles = [
809 _build_track_dynamics_profile(ref, t, i)
810 for i, t in enumerate(tracks_to_include)
811 ]
812 now = _utc_now()
813 logger.info(
814 "✅ dynamics/page repo=%s ref=%s tracks=%d",
815 repo_id[:8], ref, len(profiles),
816 )
817 return DynamicsPageData(
818 ref=ref,
819 repo_id=repo_id,
820 computed_at=now,
821 tracks=profiles,
822 filters_applied=AnalysisFilters(track=track, section=section),
823 )
824
825
826 def compute_emotion_map(
827 *,
828 repo_id: str,
829 ref: str,
830 track: str | None = None,
831 section: str | None = None,
832 ) -> EmotionMapResponse:
833 """Build a complete :class:`EmotionMapResponse` for an emotion map page.
834
835 Returns per-beat intra-ref evolution, cross-commit trajectory, drift
836 distances between consecutive commits, a generated narrative, and source
837 attribution. All data is deterministic for a given ``ref`` so agents
838 receive consistent results across retries.
839
840 Why separate from ``compute_dimension('emotion', ...)``
841 -------------------------------------------------------
842 The generic emotion dimension returns a single aggregate snapshot. This
843 function returns the *temporal* and *cross-commit* shape of the emotional
844 arc — the information needed to render line charts and trajectory plots.
845
846 Args:
847 repo_id: Muse Hub repo UUID (used for logging).
848 ref: Head Muse commit ref (branch name or commit ID).
849 track: Optional instrument track filter.
850 section: Optional musical section filter.
851
852 Returns:
853 :class:`EmotionMapResponse` with evolution, trajectory, drift, and narrative.
854 """
855 seed = _ref_hash(ref)
856 now = _utc_now()
857
858 # ── Per-beat evolution within this ref ─────────────────────────────────
859 total_beats = 32
860 evolution: list[EmotionMapPoint] = []
861 for i in range(total_beats):
862 phase = i / total_beats
863 # Each axis follows a gentle sinusoidal arc seeded by ref hash
864 energy = round(0.4 + 0.4 * abs((phase - 0.5) * 2) * (1 + (seed >> (i % 16)) % 3) / 4, 4)
865 valence = round(max(0.0, min(1.0, 0.5 + 0.3 * ((seed + i) % 7 - 3) / 3)), 4)
866 tension_val = round(min(1.0, 0.2 + 0.6 * phase * (1 + (seed % 3) / 3)), 4)
867 darkness = round(max(0.0, min(1.0, 1.0 - valence * 0.6 - energy * 0.2)), 4)
868 evolution.append(
869 EmotionMapPoint(
870 beat=float(i),
871 vector=EmotionVector(
872 energy=energy,
873 valence=valence,
874 tension=tension_val,
875 darkness=darkness,
876 ),
877 )
878 )
879
880 # Summary vector = mean across all evolution points
881 n = len(evolution)
882 summary_vector = EmotionVector(
883 energy=round(sum(p.vector.energy for p in evolution) / n, 4),
884 valence=round(sum(p.vector.valence for p in evolution) / n, 4),
885 tension=round(sum(p.vector.tension for p in evolution) / n, 4),
886 darkness=round(sum(p.vector.darkness for p in evolution) / n, 4),
887 )
888
889 # ── Cross-commit trajectory (5 synthetic ancestor snapshots + head) ───
890 _COMMIT_EMOTIONS = ["serene", "tense", "brooding", "joyful", "melancholic", "energetic"]
891 trajectory: list[CommitEmotionSnapshot] = []
892 n_commits = 5
893 for j in range(n_commits):
894 commit_seed = _ref_hash(f"{ref}:{j}")
895 em = _COMMIT_EMOTIONS[(seed + j) % len(_COMMIT_EMOTIONS)]
896 valence_traj = round(max(0.0, min(1.0, 0.3 + (commit_seed % 70) / 100)), 4)
897 energy_traj = round(max(0.0, min(1.0, 0.2 + (commit_seed % 80) / 100)), 4)
898 tension_traj = round(max(0.0, min(1.0, 0.1 + (commit_seed % 90) / 100)), 4)
899 darkness_traj = round(max(0.0, min(1.0, 1.0 - valence_traj * 0.7)), 4)
900 trajectory.append(
901 CommitEmotionSnapshot(
902 commit_id=hashlib.md5(f"{ref}:{j}".encode()).hexdigest()[:16], # noqa: S324
903 message=f"Ancestor commit {n_commits - j}: {em} passage",
904 timestamp=f"2026-0{1 + j % 9}-{10 + j:02d}T12:00:00Z",
905 vector=EmotionVector(
906 energy=energy_traj,
907 valence=valence_traj,
908 tension=tension_traj,
909 darkness=darkness_traj,
910 ),
911 primary_emotion=em,
912 )
913 )
914 # Append head commit snapshot
915 trajectory.append(
916 CommitEmotionSnapshot(
917 commit_id=ref[:16] if len(ref) >= 16 else ref,
918 message="HEAD — current composition state",
919 timestamp=now.strftime("%Y-%m-%dT%H:%M:%SZ"),
920 vector=summary_vector,
921 primary_emotion=_COMMIT_EMOTIONS[seed % len(_COMMIT_EMOTIONS)],
922 )
923 )
924
925 # ── Drift between consecutive commits ──────────────────────────────────
926 _AXES = ["energy", "valence", "tension", "darkness"]
927 drift: list[EmotionDrift] = []
928 for k in range(len(trajectory) - 1):
929 a = trajectory[k].vector
930 b = trajectory[k + 1].vector
931 diff = [
932 abs(b.energy - a.energy),
933 abs(b.valence - a.valence),
934 abs(b.tension - a.tension),
935 abs(b.darkness - a.darkness),
936 ]
937 euclidean = round((sum(d**2 for d in diff) ** 0.5), 4)
938 dominant_change = _AXES[diff.index(max(diff))]
939 drift.append(
940 EmotionDrift(
941 from_commit=trajectory[k].commit_id,
942 to_commit=trajectory[k + 1].commit_id,
943 drift=euclidean,
944 dominant_change=dominant_change,
945 )
946 )
947
948 # ── Narrative generation ───────────────────────────────────────────────
949 head_em = trajectory[-1].primary_emotion
950 first_em = trajectory[0].primary_emotion
951 max_drift_entry = max(drift, key=lambda d: d.drift) if drift else None
952 narrative_parts = [
953 f"This composition begins with a {first_em} character",
954 f"and arrives at a {head_em} state at the head commit.",
955 ]
956 if max_drift_entry is not None:
957 narrative_parts.append(
958 f"The largest emotional shift occurs between commits "
959 f"{max_drift_entry.from_commit[:8]} and {max_drift_entry.to_commit[:8]}, "
960 f"with a {max_drift_entry.dominant_change} shift of {max_drift_entry.drift:.2f}."
961 )
962 narrative = " ".join(narrative_parts)
963
964 # ── Source attribution ─────────────────────────────────────────────────
965 source: Literal["explicit", "inferred", "mixed"] = "inferred" # Full implementation will check commit metadata for explicit tags
966
967 logger.info("✅ emotion-map repo=%s ref=%s beats=%d commits=%d", repo_id[:8], ref, n, len(trajectory))
968 return EmotionMapResponse(
969 repo_id=repo_id,
970 ref=ref,
971 computed_at=now,
972 filters_applied=AnalysisFilters(track=track, section=section),
973 evolution=evolution,
974 summary_vector=summary_vector,
975 trajectory=trajectory,
976 drift=drift,
977 narrative=narrative,
978 source=source,
979 )
980
981
982 def compute_aggregate_analysis(
983 *,
984 repo_id: str,
985 ref: str,
986 track: str | None = None,
987 section: str | None = None,
988 ) -> AggregateAnalysisResponse:
989 """Build a complete :class:`AggregateAnalysisResponse` for all 13 dimensions.
990
991 This is the primary entry point for the aggregate endpoint. All 13
992 dimensions are computed in a single call so agents can retrieve the full
993 musical picture without issuing 13 sequential requests.
994
995 Args:
996 repo_id: Muse Hub repo UUID.
997 ref: Muse commit ref.
998 track: Optional track filter (applied to all dimensions).
999 section: Optional section filter (applied to all dimensions).
1000
1001 Returns:
1002 :class:`AggregateAnalysisResponse` with one entry per dimension.
1003 """
1004 now = _utc_now()
1005 dimensions = [
1006 AnalysisResponse(
1007 dimension=dim,
1008 ref=ref,
1009 computed_at=now,
1010 data=compute_dimension(dim, ref, track, section),
1011 filters_applied=AnalysisFilters(track=track, section=section),
1012 )
1013 for dim in ALL_DIMENSIONS
1014 ]
1015 logger.info("✅ analysis/aggregate repo=%s ref=%s dims=%d", repo_id[:8], ref, len(dimensions))
1016 return AggregateAnalysisResponse(
1017 ref=ref,
1018 repo_id=repo_id,
1019 computed_at=now,
1020 dimensions=dimensions,
1021 filters_applied=AnalysisFilters(track=track, section=section),
1022 )
1023
1024
1025 # ---------------------------------------------------------------------------
1026 # Dedicated harmony endpoint — muse harmony command
1027 # ---------------------------------------------------------------------------
1028
1029 _ROMAN_NUMERALS_BY_MODE: dict[str, list[tuple[str, str, str, str]]] = {
1030 # (roman, quality, function, root-offset-label)
1031 # root-offset-label is relative; actual root derived from tonic + offset
1032 "major": [
1033 ("I", "major", "tonic", "P1"),
1034 ("IIm7", "minor", "pre-dominant", "M2"),
1035 ("IIIm", "minor", "tonic", "M3"),
1036 ("IV", "major", "subdominant", "P4"),
1037 ("V7", "dominant","dominant", "P5"),
1038 ("VIm", "minor", "tonic", "M6"),
1039 ("VIIø", "half-diminished", "dominant", "M7"),
1040 ],
1041 "minor": [
1042 ("Im", "minor", "tonic", "P1"),
1043 ("IIø", "half-diminished", "pre-dominant", "M2"),
1044 ("bIII", "major", "tonic", "m3"),
1045 ("IVm", "minor", "subdominant", "P4"),
1046 ("V7", "dominant","dominant", "P5"),
1047 ("bVI", "major", "subdominant", "m6"),
1048 ("bVII", "major", "subdominant", "m7"),
1049 ],
1050 }
1051
1052 # Semitone offsets for scale degrees so we can compute the actual root pitch.
1053 _SEMITONE_OFFSETS: dict[str, int] = {
1054 "P1": 0, "M2": 2, "M3": 4, "P4": 5, "P5": 7,
1055 "M6": 9, "M7": 11, "m3": 3, "m6": 8, "m7": 10,
1056 }
1057
1058 _CHROMATIC_SCALE = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
1059
1060 _CADENCE_TYPES = ["authentic", "half", "plagal", "deceptive", "perfect-authentic"]
1061
1062 _HARMONY_MODES = ["major", "minor"]
1063
1064
1065 def _transpose_root(tonic: str, semitones: int) -> str:
1066 """Return the pitch class that is ``semitones`` above ``tonic``."""
1067 try:
1068 base_idx = _CHROMATIC_SCALE.index(tonic)
1069 except ValueError:
1070 # Fallback for flat spellings like Bb, Eb — map to sharp equivalent.
1071 _FLAT_TO_SHARP = {"Bb": "A#", "Eb": "D#", "Ab": "G#", "Db": "C#", "Gb": "F#"}
1072 base_idx = _CHROMATIC_SCALE.index(_FLAT_TO_SHARP.get(tonic, "C"))
1073 return _CHROMATIC_SCALE[(base_idx + semitones) % 12]
1074
1075
1076 def compute_harmony_analysis(
1077 *,
1078 repo_id: str,
1079 ref: str,
1080 track: str | None = None,
1081 section: str | None = None,
1082 ) -> HarmonyAnalysisResponse:
1083 """Build a dedicated harmonic analysis for a Muse commit ref.
1084
1085 Returns a Roman-numeral-centric view of the harmonic content. Unlike the
1086 generic ``harmony`` dimension (which returns :class:`HarmonyData` with raw
1087 chord symbols and a tension curve), this response is structured for tonal
1088 reasoning: Roman numerals with function labels, cadence positions, and
1089 detected modulations.
1090
1091 Maps to the ``muse harmony --ref {ref}`` CLI command.
1092
1093 The stub data is deterministic for a given ``ref`` so agents receive
1094 consistent responses across retries. Harmonic content is keyed on
1095 tonic/mode derived from the ref hash — the same tonic and mode that the
1096 generic harmony dimension uses, ensuring cross-endpoint consistency.
1097
1098 Args:
1099 repo_id: Muse Hub repo UUID (used only for logging).
1100 ref: Muse commit ref (seeds the deterministic data).
1101 track: Optional track filter (recorded in response; stub ignores it).
1102 section: Optional section filter (recorded in response; stub ignores it).
1103
1104 Returns:
1105 :class:`HarmonyAnalysisResponse` ready for the harmony endpoint.
1106 """
1107 seed = _ref_hash(ref)
1108 tonic = _pick(seed, _TONICS)
1109 mode = _pick(seed, _HARMONY_MODES, offset=1)
1110 key_label = f"{tonic} {mode}"
1111
1112 # Build Roman numeral events — use the first 5 chords from the mode table.
1113 rn_table = _ROMAN_NUMERALS_BY_MODE.get(mode, _ROMAN_NUMERALS_BY_MODE["major"])
1114 chord_count = 4 + (seed % 3) # 4–6 chord events
1115 roman_numerals: list[RomanNumeralEvent] = []
1116 beat = 0.0
1117 for i in range(min(chord_count, len(rn_table))):
1118 roman, quality, function, offset_label = rn_table[i]
1119 semitones = _SEMITONE_OFFSETS.get(offset_label, 0)
1120 root = _transpose_root(tonic, semitones)
1121 roman_numerals.append(
1122 RomanNumeralEvent(
1123 beat=beat,
1124 chord=roman,
1125 root=root,
1126 quality=quality,
1127 function=function,
1128 )
1129 )
1130 beat += 4.0
1131
1132 # Build cadences — one or two, positioned at phrase boundaries.
1133 cadence_type = _pick(seed, _CADENCE_TYPES, offset=2)
1134 cadence_beat = float((seed % 4) * 4 + 4)
1135 cadences: list[CadenceEvent] = [
1136 CadenceEvent.model_validate({"from": "V", "to": "I", "beat": cadence_beat, "type": cadence_type}),
1137 ]
1138 if seed % 3 == 0:
1139 cadences.append(
1140 CadenceEvent.model_validate({"from": "IV", "to": "I", "beat": cadence_beat + 16.0, "type": "plagal"}),
1141 )
1142
1143 # Build modulations — 0 or 1, depending on ref seed.
1144 modulations: list[HarmonyModulationEvent] = []
1145 if seed % 4 == 0:
1146 dom_root = _transpose_root(tonic, 7) # dominant (P5)
1147 modulations.append(
1148 HarmonyModulationEvent(
1149 beat=32.0,
1150 from_key=key_label,
1151 to_key=f"{dom_root} {mode}",
1152 pivot_chord=dom_root,
1153 )
1154 )
1155
1156 # Harmonic rhythm: chord changes per minute. Assumes ~120 BPM tempo.
1157 # With chords every 4 beats at 120 BPM → 30 chord changes per minute.
1158 # Varies slightly per ref to feel alive.
1159 base_rhythm = 2.0
1160 harmonic_rhythm_bpm = round(base_rhythm + (seed % 5) * 0.25, 2)
1161
1162 logger.info("✅ harmony/analysis repo=%s ref=%s key=%s", repo_id[:8], ref[:8], key_label)
1163 return HarmonyAnalysisResponse(
1164 key=key_label,
1165 mode=mode,
1166 roman_numerals=roman_numerals,
1167 cadences=cadences,
1168 modulations=modulations,
1169 harmonic_rhythm_bpm=harmonic_rhythm_bpm,
1170 )
1171
1172
1173 # ---------------------------------------------------------------------------
1174 # Arrangement matrix
1175 # ---------------------------------------------------------------------------
1176
1177 _ARRANGEMENT_INSTRUMENTS: list[str] = ["bass", "keys", "guitar", "drums", "lead", "pads"]
1178 _ARRANGEMENT_SECTIONS: list[str] = ["intro", "verse_1", "chorus", "bridge", "outro"]
1179
1180 # Beat positions for each section (start, end). Realistic 4/4 structure with
1181 # 8-bar sections at 120 BPM (32 beats per section).
1182 _SECTION_BEATS: dict[str, tuple[float, float]] = {
1183 "intro": (0.0, 32.0),
1184 "verse_1": (32.0, 64.0),
1185 "chorus": (64.0, 96.0),
1186 "bridge": (96.0, 112.0),
1187 "outro": (112.0, 128.0),
1188 }
1189
1190 # Probability that an instrument is active in a given section (realistic
1191 # arrangement logic — drums always play, bass almost always, pads lighter).
1192 _ACTIVE_PROBABILITY: dict[str, dict[str, float]] = {
1193 "bass": {"intro": 0.7, "verse_1": 1.0, "chorus": 1.0, "bridge": 0.8, "outro": 0.6},
1194 "keys": {"intro": 0.5, "verse_1": 0.8, "chorus": 1.0, "bridge": 0.7, "outro": 0.5},
1195 "guitar": {"intro": 0.3, "verse_1": 0.7, "chorus": 0.9, "bridge": 0.6, "outro": 0.3},
1196 "drums": {"intro": 0.5, "verse_1": 1.0, "chorus": 1.0, "bridge": 0.8, "outro": 0.4},
1197 "lead": {"intro": 0.2, "verse_1": 0.5, "chorus": 0.8, "bridge": 0.9, "outro": 0.3},
1198 "pads": {"intro": 0.8, "verse_1": 0.6, "chorus": 0.7, "bridge": 1.0, "outro": 0.9},
1199 }
1200
1201
1202 def compute_arrangement_matrix(*, repo_id: str, ref: str) -> ArrangementMatrixResponse:
1203 """Build a deterministic :class:`ArrangementMatrixResponse` for a Muse commit ref.
1204
1205 Returns instrument × section density data so the arrangement matrix page can
1206 render a colour-coded grid without downloading any audio or MIDI files.
1207
1208 The stub data is deterministically seeded by ``ref`` so that agents receive
1209 consistent responses across retries. Note counts and density values are
1210 drawn from realistic ranges for a 6-instrument soul/pop arrangement.
1211
1212 Args:
1213 repo_id: Muse Hub repo UUID (used only for logging).
1214 ref: Muse commit ref (seeds the deterministic RNG).
1215
1216 Returns:
1217 :class:`ArrangementMatrixResponse` ready for the arrange endpoint.
1218 """
1219 seed_int = int(hashlib.md5(ref.encode()).hexdigest()[:8], 16) # noqa: S324 — non-crypto
1220
1221 instruments = _ARRANGEMENT_INSTRUMENTS
1222 sections = _ARRANGEMENT_SECTIONS
1223
1224 cells: list[ArrangementCellData] = []
1225 raw_counts: dict[tuple[str, str], int] = {}
1226
1227 # Generate note counts deterministically per (instrument, section).
1228 for i_idx, instrument in enumerate(instruments):
1229 for s_idx, section in enumerate(sections):
1230 prob = _ACTIVE_PROBABILITY.get(instrument, {}).get(section, 0.5)
1231 # Mix ref seed with cell position for per-cell variation.
1232 cell_seed = (seed_int + i_idx * 31 + s_idx * 97) % (2**32)
1233 # Deterministic "random" value in [0, 1) via cheap LCG step.
1234 lcg = (cell_seed * 1664525 + 1013904223) % (2**32)
1235 roll = lcg / (2**32)
1236 active = roll < prob
1237 if active:
1238 # Note count: 8–64, skewed toward busier instruments.
1239 note_count = 8 + int(roll * 56)
1240 else:
1241 note_count = 0
1242 raw_counts[(instrument, section)] = note_count
1243
1244 # Normalise counts to [0, 1] density (max across the whole matrix).
1245 max_count = max(raw_counts.values()) or 1
1246
1247 for i_idx, instrument in enumerate(instruments):
1248 for s_idx, section in enumerate(sections):
1249 note_count = raw_counts[(instrument, section)]
1250 beat_start, beat_end = _SECTION_BEATS[section]
1251 active = note_count > 0
1252 # Pitch range: realistic MIDI range per instrument.
1253 pitch_base = {"bass": 28, "keys": 48, "guitar": 40, "drums": 36, "lead": 60, "pads": 52}.get(
1254 instrument, 48
1255 )
1256 pitch_low = pitch_base if active else 0
1257 pitch_high = pitch_base + 24 if active else 0
1258 cells.append(
1259 ArrangementCellData(
1260 instrument=instrument,
1261 section=section,
1262 note_count=note_count,
1263 note_density=round(note_count / max_count, 4),
1264 beat_start=beat_start,
1265 beat_end=beat_end,
1266 pitch_low=pitch_low,
1267 pitch_high=pitch_high,
1268 active=active,
1269 )
1270 )
1271
1272 # Row summaries (per instrument).
1273 row_summaries: list[ArrangementRowSummary] = []
1274 for instrument in instruments:
1275 inst_cells = [c for c in cells if c.instrument == instrument]
1276 total = sum(c.note_count for c in inst_cells)
1277 active_secs = sum(1 for c in inst_cells if c.active)
1278 mean_d = round(sum(c.note_density for c in inst_cells) / len(inst_cells), 4) if inst_cells else 0.0
1279 row_summaries.append(
1280 ArrangementRowSummary(
1281 instrument=instrument,
1282 total_notes=total,
1283 active_sections=active_secs,
1284 mean_density=mean_d,
1285 )
1286 )
1287
1288 # Column summaries (per section).
1289 column_summaries: list[ArrangementColumnSummary] = []
1290 for section in sections:
1291 sec_cells = [c for c in cells if c.section == section]
1292 total = sum(c.note_count for c in sec_cells)
1293 active_inst = sum(1 for c in sec_cells if c.active)
1294 beat_start, beat_end = _SECTION_BEATS[section]
1295 column_summaries.append(
1296 ArrangementColumnSummary(
1297 section=section,
1298 total_notes=total,
1299 active_instruments=active_inst,
1300 beat_start=beat_start,
1301 beat_end=beat_end,
1302 )
1303 )
1304
1305 total_beats = max(end for _, end in _SECTION_BEATS.values())
1306 logger.info("✅ arrangement/matrix repo=%s ref=%s cells=%d", repo_id[:8], ref[:8], len(cells))
1307 return ArrangementMatrixResponse(
1308 repo_id=repo_id,
1309 ref=ref,
1310 instruments=instruments,
1311 sections=sections,
1312 cells=cells,
1313 row_summaries=row_summaries,
1314 column_summaries=column_summaries,
1315 total_beats=total_beats,
1316 )
1317
1318
1319 # ---------------------------------------------------------------------------
1320 # Semantic recall
1321 # ---------------------------------------------------------------------------
1322
1323 _RECALL_DIMENSIONS: list[str] = ["harmony", "groove", "emotion", "motifs", "contour", "tempo"]
1324
1325 _RECALL_MESSAGES: list[str] = [
1326 "Add jazzy chord progression with swing feel",
1327 "Introduce minor-key bridge with tension build",
1328 "Refine melodic contour — ascending arch in chorus",
1329 "Adjust groove: add half-time feel in verse",
1330 "Add layered pad texture for emotional depth",
1331 "Modulate to dominant for climactic section",
1332 "Tighten rhythmic grid — straight feel throughout",
1333 "Add secondary dominant walkdown before chorus",
1334 ]
1335
1336 _RECALL_BRANCHES: list[str] = ["main", "feature/bridge", "feature/chorus", "experiment/jazz", "develop"]
1337
1338
1339 def compute_recall(
1340 *,
1341 repo_id: str,
1342 ref: str,
1343 query: str,
1344 limit: int = 10,
1345 ) -> RecallResponse:
1346 """Query the musical feature vector space for commits semantically matching ``query``.
1347
1348 Why this exists
1349 ---------------
1350 Agents and producers need to surface past commits that are musically relevant
1351 to a natural-language description (e.g. ``"a jazzy chord progression with swing
1352 groove"``). This endpoint bridges semantic intent and the vector index so that
1353 retrieval is based on musical meaning rather than exact keyword matching.
1354
1355 Implementation note
1356 -------------------
1357 Returns deterministic stub results keyed on the XOR of the ref and query
1358 hashes so agents receive consistent responses across retries.
1359
1360 Args:
1361 repo_id: Muse Hub repo UUID (used for scoping and logging).
1362 ref: Muse commit ref to scope the search to (only reachable commits).
1363 query: Natural-language search string, e.g. ``"swing groove with jazz harmony"``.
1364 limit: Maximum number of matches to return (default 10, max 50).
1365
1366 Returns:
1367 :class:`RecallResponse` with a ranked list of :class:`RecallMatch` entries,
1368 sorted descending by cosine similarity score.
1369 """
1370 limit = max(1, min(limit, 50))
1371 q_seed = _ref_hash(query)
1372 r_seed = _ref_hash(ref)
1373 combined_seed = q_seed ^ r_seed
1374
1375 # Deterministic total count — varies by query so results feel realistic.
1376 total_matches = 4 + (combined_seed % 12)
1377 n_to_return = min(limit, total_matches)
1378
1379 matches: list[RecallMatch] = []
1380 for i in range(n_to_return):
1381 item_seed = combined_seed ^ (i * 0x9E3779B9)
1382 # Score: highest for i=0, decaying with rank (deterministic)
1383 base_score = 0.92 - i * 0.06
1384 noise = ((item_seed >> (i % 16)) % 8) / 100.0
1385 score = round(max(0.0, min(1.0, base_score - noise)), 4)
1386
1387 commit_hash = hashlib.md5(f"{ref}:{query}:{i}".encode()).hexdigest()[:16] # noqa: S324
1388 message = _RECALL_MESSAGES[(combined_seed + i) % len(_RECALL_MESSAGES)]
1389 branch = _RECALL_BRANCHES[(combined_seed + i) % len(_RECALL_BRANCHES)]
1390
1391 # Pick 1–3 matched dimensions — most relevant first.
1392 n_dims = 1 + i % 3
1393 dims = [
1394 _RECALL_DIMENSIONS[(combined_seed + i + k) % len(_RECALL_DIMENSIONS)]
1395 for k in range(n_dims)
1396 ]
1397 # Deduplicate while preserving order.
1398 seen: set[str] = set()
1399 unique_dims = [d for d in dims if not (d in seen or seen.add(d))] # type: ignore[func-returns-value]
1400
1401 matches.append(
1402 RecallMatch(
1403 commit_id=commit_hash,
1404 commit_message=message,
1405 branch=branch,
1406 score=score,
1407 matched_dimensions=unique_dims,
1408 )
1409 )
1410
1411 logger.info(
1412 "✅ recall repo=%s ref=%s query=%r matches=%d/%d",
1413 repo_id[:8], ref, query[:40], len(matches), total_matches,
1414 )
1415 return RecallResponse(
1416 repo_id=repo_id,
1417 ref=ref,
1418 query=query,
1419 matches=matches,
1420 total_matches=total_matches,
1421 embedding_dimensions=128,
1422 )
1423
1424
1425 # ---------------------------------------------------------------------------
1426 # Cross-ref similarity
1427 # ---------------------------------------------------------------------------
1428
1429 # Interpretation thresholds: the weighted mean of 10 dimension scores maps
1430 # to a qualitative label. Weights are equal (0.1 each) so the overall score
1431 # is a simple mean, but kept in this lookup to support future re-weighting
1432 # without changing the response shape.
1433 _SIMILARITY_WEIGHTS: dict[str, float] = {
1434 "pitch_distribution": 0.10,
1435 "rhythm_pattern": 0.10,
1436 "tempo": 0.10,
1437 "dynamics": 0.10,
1438 "harmonic_content": 0.10,
1439 "form": 0.10,
1440 "instrument_blend": 0.10,
1441 "groove": 0.10,
1442 "contour": 0.10,
1443 "emotion": 0.10,
1444 }
1445
1446
1447 def _interpret_similarity(score: float, dims: RefSimilarityDimensions) -> str:
1448 """Generate a human-readable interpretation of a cross-ref similarity score.
1449
1450 The interpretation names the dominant divergence axis when the overall
1451 score is below 0.9, giving agents and UIs actionable language without
1452 requiring further API calls.
1453 """
1454 dim_values = {
1455 "pitch distribution": dims.pitch_distribution,
1456 "rhythm pattern": dims.rhythm_pattern,
1457 "tempo": dims.tempo,
1458 "dynamics": dims.dynamics,
1459 "harmonic content": dims.harmonic_content,
1460 "form": dims.form,
1461 "instrument blend": dims.instrument_blend,
1462 "groove": dims.groove,
1463 "contour": dims.contour,
1464 "emotion": dims.emotion,
1465 }
1466 lowest_dim = min(dim_values, key=lambda k: dim_values[k])
1467 lowest_score = dim_values[lowest_dim]
1468
1469 if score >= 0.90:
1470 return "Nearly identical arrangements — only subtle differences detected."
1471 if score >= 0.75:
1472 return (
1473 f"Highly similar arrangement with divergent {lowest_dim} choices "
1474 f"(score: {lowest_score:.2f})."
1475 )
1476 if score >= 0.55:
1477 return (
1478 f"Moderately similar — significant divergence in {lowest_dim} "
1479 f"(score: {lowest_score:.2f}) and related dimensions."
1480 )
1481 return (
1482 f"Low similarity — the two refs differ substantially, "
1483 f"especially in {lowest_dim} (score: {lowest_score:.2f})."
1484 )
1485
1486
1487 def compute_ref_similarity(
1488 *,
1489 repo_id: str,
1490 base_ref: str,
1491 compare_ref: str,
1492 ) -> RefSimilarityResponse:
1493 """Compute cross-ref similarity between two Muse refs.
1494
1495 Returns a :class:`~musehub.models.musehub_analysis.RefSimilarityResponse`
1496 with per-dimension scores and an overall weighted mean.
1497
1498 Scores are deterministic stubs derived from both ref hashes so that:
1499 - The same pair always returns the same result (idempotent for agents).
1500 - Swapping base/compare yields scores of the same magnitude (symmetry).
1501
1502 When real MIDI content analysis is available, replace the stub derivation
1503 below with actual per-dimension comparison logic while preserving this
1504 function's signature and return type.
1505
1506 Args:
1507 repo_id: Muse repository identifier (used for log context only).
1508 base_ref: The baseline ref (branch name, tag, or commit hash).
1509 compare_ref: The ref to compare against ``base_ref``.
1510
1511 Returns:
1512 :class:`~musehub.models.musehub_analysis.RefSimilarityResponse`
1513 containing 10 dimension scores, an overall similarity, and an
1514 auto-generated interpretation string.
1515 """
1516 base_seed = _ref_hash(base_ref)
1517 compare_seed = _ref_hash(compare_ref)
1518
1519 def _dim_score(offset: int) -> float:
1520 """Derive a deterministic 0–1 similarity score for one dimension."""
1521 combined = (base_seed + compare_seed + offset) % (2**16)
1522 raw = (combined / (2**16 - 1))
1523 return round(0.50 + raw * 0.50, 4)
1524
1525 dims = RefSimilarityDimensions(
1526 pitch_distribution=_dim_score(0),
1527 rhythm_pattern=_dim_score(1),
1528 tempo=_dim_score(2),
1529 dynamics=_dim_score(3),
1530 harmonic_content=_dim_score(4),
1531 form=_dim_score(5),
1532 instrument_blend=_dim_score(6),
1533 groove=_dim_score(7),
1534 contour=_dim_score(8),
1535 emotion=_dim_score(9),
1536 )
1537
1538 overall = round(
1539 sum(
1540 getattr(dims, k.replace(" ", "_").replace("-", "_")) * w
1541 for k, w in {
1542 "pitch_distribution": 0.10,
1543 "rhythm_pattern": 0.10,
1544 "tempo": 0.10,
1545 "dynamics": 0.10,
1546 "harmonic_content": 0.10,
1547 "form": 0.10,
1548 "instrument_blend": 0.10,
1549 "groove": 0.10,
1550 "contour": 0.10,
1551 "emotion": 0.10,
1552 }.items()
1553 ),
1554 4,
1555 )
1556
1557 interpretation = _interpret_similarity(overall, dims)
1558 logger.info(
1559 "✅ similarity repo=%s base=%s compare=%s overall=%.2f",
1560 repo_id[:8],
1561 base_ref[:8],
1562 compare_ref[:8],
1563 overall,
1564 )
1565 return RefSimilarityResponse(
1566 base_ref=base_ref,
1567 compare_ref=compare_ref,
1568 overall_similarity=overall,
1569 dimensions=dims,
1570 interpretation=interpretation,
1571 )
1572
1573
1574 # ---------------------------------------------------------------------------
1575 # Emotion diff
1576 # ---------------------------------------------------------------------------
1577
1578 # Axis labels in declaration order — used for delta interpretation
1579 _EMOTION_8D_AXES: list[str] = [
1580 "valence", "energy", "tension", "complexity",
1581 "warmth", "brightness", "darkness", "playfulness",
1582 ]
1583
1584
1585 def _build_emotion_vector_8d(ref: str) -> EmotionVector8D:
1586 """Build a deterministic 8-axis emotion vector for a ref.
1587
1588 All eight axes are derived from independent bit-slices of the ref hash so
1589 they vary independently across refs — avoids correlated stubs.
1590
1591 Why 8D not 4D: the emotion-diff endpoint uses an extended radar chart that
1592 separates warmth/brightness/playfulness/complexity from the core
1593 valence/energy/tension/darkness axes used in the emotion-map endpoint.
1594 """
1595 seed = _ref_hash(ref)
1596
1597 def _axis(shift: int, base: float = 0.1, spread: float = 0.8) -> float:
1598 return round(base + ((seed >> shift) % 100) * spread / 100, 4)
1599
1600 return EmotionVector8D(
1601 valence=_axis(0),
1602 energy=_axis(8),
1603 tension=_axis(16),
1604 complexity=_axis(24),
1605 warmth=_axis(32),
1606 brightness=_axis(40),
1607 darkness=_axis(48),
1608 playfulness=_axis(56),
1609 )
1610
1611
1612 def _clamp(value: float) -> float:
1613 """Clamp a delta value to [-1, 1] for the signed delta field."""
1614 return max(-1.0, min(1.0, round(value, 4)))
1615
1616
1617 def compute_emotion_diff(
1618 *,
1619 repo_id: str,
1620 head_ref: str,
1621 base_ref: str,
1622 ) -> EmotionDiffResponse:
1623 """Compute an 8-axis emotional diff between two Muse commit refs.
1624
1625 Returns the per-axis emotion vectors for ``base_ref`` and ``head_ref``,
1626 their signed delta (``head - base``), and a natural-language interpretation
1627 of the most significant shifts.
1628
1629 Why this is separate from the generic emotion dimension
1630 -------------------------------------------------------
1631 The generic ``emotion`` dimension returns a single aggregate snapshot with
1632 a 2-axis (valence/arousal) model. This endpoint uses an extended 8-axis
1633 radar model and computes a *comparative* diff between two refs — the
1634 information the ``muse emotion-diff`` CLI command and the PR detail page
1635 need to answer "how did this commit change the emotional character?"
1636
1637 Args:
1638 repo_id: Muse Hub repo UUID (used for logging).
1639 head_ref: The ref being evaluated (the head commit).
1640 base_ref: The ref used as comparison baseline (e.g. parent commit, ``main``).
1641
1642 Returns:
1643 :class:`EmotionDiffResponse` with base, head, delta vectors, and interpretation.
1644 """
1645 base_vec = _build_emotion_vector_8d(base_ref)
1646 head_vec = _build_emotion_vector_8d(head_ref)
1647
1648 delta = EmotionDelta8D(
1649 valence=_clamp(head_vec.valence - base_vec.valence),
1650 energy=_clamp(head_vec.energy - base_vec.energy),
1651 tension=_clamp(head_vec.tension - base_vec.tension),
1652 complexity=_clamp(head_vec.complexity - base_vec.complexity),
1653 warmth=_clamp(head_vec.warmth - base_vec.warmth),
1654 brightness=_clamp(head_vec.brightness - base_vec.brightness),
1655 darkness=_clamp(head_vec.darkness - base_vec.darkness),
1656 playfulness=_clamp(head_vec.playfulness - base_vec.playfulness),
1657 )
1658
1659 raw_deltas: list[tuple[str, float]] = [
1660 ("valence", delta.valence),
1661 ("energy", delta.energy),
1662 ("tension", delta.tension),
1663 ("complexity", delta.complexity),
1664 ("warmth", delta.warmth),
1665 ("brightness", delta.brightness),
1666 ("darkness", delta.darkness),
1667 ("playfulness", delta.playfulness),
1668 ]
1669 sorted_deltas = sorted(raw_deltas, key=lambda x: abs(x[1]), reverse=True)
1670 dominant_axis, dominant_value = sorted_deltas[0]
1671
1672 if abs(dominant_value) < 0.05:
1673 interpretation = (
1674 "This commit introduced minimal emotional change — the character of the "
1675 "piece is nearly identical to the base ref across all eight perceptual axes."
1676 )
1677 else:
1678 direction = "increased" if dominant_value > 0 else "decreased"
1679 secondary_parts: list[str] = []
1680 for axis, value in sorted_deltas[1:3]:
1681 if abs(value) >= 0.05:
1682 secondary_parts.append(
1683 f"{axis} {'rose' if value > 0 else 'fell'} by {abs(value):.2f}"
1684 )
1685 secondary_text = (
1686 f" Notable secondary shifts: {', '.join(secondary_parts)}." if secondary_parts else ""
1687 )
1688 interpretation = (
1689 f"This commit {direction} {dominant_axis} by {abs(dominant_value):.2f} "
1690 f"(the dominant emotional shift).{secondary_text}"
1691 )
1692
1693 logger.info(
1694 "✅ emotion-diff repo=%s head=%s base=%s dominant=%s delta=%.3f",
1695 repo_id[:8], head_ref, base_ref, dominant_axis, dominant_value,
1696 )
1697 return EmotionDiffResponse(
1698 repo_id=repo_id,
1699 base_ref=base_ref,
1700 head_ref=head_ref,
1701 computed_at=_utc_now(),
1702 base_emotion=base_vec,
1703 head_emotion=head_vec,
1704 delta=delta,
1705 interpretation=interpretation,
1706 )
1707
1708
1709 # ---------------------------------------------------------------------------
1710 # SSR aggregation functions — compare / divergence / context pages
1711 # ---------------------------------------------------------------------------
1712
1713 _MUSICAL_DIMENSIONS = ["Melodic", "Harmonic", "Rhythmic", "Structural", "Dynamic"]
1714
1715
1716 async def compare_refs(
1717 db: AsyncSession,
1718 repo_id: str,
1719 base: str,
1720 head: str,
1721 ) -> CompareResult:
1722 """Return a per-dimension comparison between two refs for SSR rendering.
1723
1724 Produces deterministic stub scores keyed on the ref values. Callers
1725 should treat this as a realistic approximation until Storpheus exposes
1726 per-ref MIDI introspection.
1727
1728 The returned :class:`CompareResult` is consumed directly by
1729 ``pages/analysis/compare.html`` — no client-side fetch required.
1730 """
1731 base_seed = _ref_hash(base)
1732 head_seed = _ref_hash(head)
1733 dimensions: list[CompareDimension] = []
1734 for i, name in enumerate(_MUSICAL_DIMENSIONS):
1735 base_val = round(((base_seed + i * 31) % 100) / 100.0, 4)
1736 head_val = round(((head_seed + i * 31) % 100) / 100.0, 4)
1737 delta = round(head_val - base_val, 4)
1738 dimensions.append(CompareDimension(name=name, base_value=base_val, head_value=head_val, delta=delta))
1739 logger.info("✅ compare-refs repo=%s base=%s head=%s", repo_id[:8], base[:8], head[:8])
1740 return CompareResult(base=base, head=head, dimensions=dimensions)
1741
1742
1743 async def compute_divergence(
1744 db: AsyncSession,
1745 repo_id: str,
1746 fork_repo_id: str | None = None,
1747 ) -> DivergenceResult:
1748 """Return a musical divergence score between a repo and its fork for SSR rendering.
1749
1750 Produces deterministic stub scores. When ``fork_repo_id`` is ``None`` the
1751 divergence is computed relative to the repo's own HEAD (self-comparison → score=0).
1752
1753 Consumed directly by ``pages/analysis/divergence.html``.
1754 """
1755 seed = _ref_hash(fork_repo_id or repo_id)
1756 repo_seed = _ref_hash(repo_id)
1757 dimensions: list[DivergenceDimension] = []
1758 for i, name in enumerate(_MUSICAL_DIMENSIONS):
1759 raw = abs(((seed + i * 37) % 100) - ((repo_seed + i * 37) % 100)) / 100.0
1760 divergence = round(min(raw, 1.0), 4)
1761 dimensions.append(DivergenceDimension(name=name, divergence=divergence))
1762 overall = round(sum(d.divergence for d in dimensions) / len(dimensions), 4)
1763 logger.info(
1764 "✅ compute-divergence repo=%s fork=%s score=%.3f",
1765 repo_id[:8], (fork_repo_id or "self")[:8], overall,
1766 )
1767 return DivergenceResult(score=overall, dimensions=dimensions)
1768
1769
1770 async def get_context(
1771 db: AsyncSession,
1772 repo_id: str,
1773 ref: str,
1774 ) -> ContextResult:
1775 """Return a musical context summary for the given ref for SSR rendering.
1776
1777 Produces deterministic stub data keyed on the ref value. Full LLM-generated
1778 summaries will replace this once the Muse pipeline is wired to the context
1779 endpoint.
1780
1781 Consumed directly by ``pages/analysis/context.html``.
1782 """
1783 seed = _ref_hash(ref)
1784 missing_pool = [
1785 "bass line",
1786 "kick drum",
1787 "reverb tail",
1788 "chord voicings",
1789 "melodic counter-line",
1790 "dynamic variation",
1791 ]
1792 suggestion_pool = {
1793 "Groove": "Introduce a 16th-note hi-hat pattern to add rhythmic density.",
1794 "Harmony": "Extend the chord to a 9th to enrich the harmonic texture.",
1795 "Melody": "Add a pentatonic counter-melody in the upper register.",
1796 "Dynamics": "Apply a decrescendo into the final bar for a softer landing.",
1797 }
1798 n_missing = (seed % 3) + 1
1799 missing = missing_pool[: n_missing]
1800 n_suggestions = (seed % 3) + 2
1801 suggestions = dict(list(suggestion_pool.items())[:n_suggestions])
1802 summary = (
1803 f"Ref {ref[:8]} establishes a {_pick(seed, _MODES)}-mode foundation "
1804 f"at {60 + (seed % 60)} BPM with a {_pick(seed + 1, _GROOVES)} groove. "
1805 f"The arrangement currently features {5 - n_missing} of the expected core elements. "
1806 f"Muse suggests {n_suggestions} compositional refinements."
1807 )
1808 logger.info("✅ get-context repo=%s ref=%s", repo_id[:8], ref[:8])
1809 return ContextResult(summary=summary, missing_elements=missing, suggestions=suggestions)