cgcardona / muse public
muse_emotion_diff.py python
564 lines 20.2 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Muse Emotion-Diff Engine — compare emotion vectors between two commits.
2
3 Answers: "How did the emotional character of a composition change between
4 two points in history?" An agent composing a new section uses this to detect
5 whether the current creative direction is drifting from the intended emotional
6 arc, and to decide whether to reinforce or contrast the mood.
7
8 Two sourcing strategies are supported:
9
10 1. **Explicit tags** — ``emotion:*`` tags attached via ``muse tag add``.
11 When both commits carry an emotion tag, their vectors are looked up from
12 the canonical :data:`EMOTION_VECTORS` table and compared directly.
13
14 2. **Inferred** — When one or both commits lack an emotion tag, the engine
15 infers a vector from available musical metadata (tempo, commit metadata)
16 stored in the :class:`~maestro.muse_cli.models.MuseCliCommit` row.
17 Full MIDI-feature inference (mode, note density, velocity) is tracked as a
18 follow-up; the current implementation uses tempo and tag-derived proxies.
19
20 Boundary rules
21 --------------
22 - Must NOT import StateStore, executor, MCP tools, or handlers.
23 - Must NOT import live streaming or SSE modules.
24 - May import ``muse_cli.{db, models}``.
25 """
26 from __future__ import annotations
27
28 import logging
29 import math
30 from dataclasses import dataclass
31
32 from sqlalchemy.ext.asyncio import AsyncSession
33 from sqlalchemy.future import select
34
35 from maestro.muse_cli.models import MuseCliCommit, MuseCliTag
36
37 logger = logging.getLogger(__name__)
38
39 # ---------------------------------------------------------------------------
40 # Emotion vector catalogue
41 # ---------------------------------------------------------------------------
42
43 #: Canonical 4-D emotion vectors keyed by ``emotion:<label>`` suffix.
44 #:
45 #: Dimensions:
46 #: energy — activity / rhythmic intensity (0.0 = still, 1.0 = frenetic)
47 #: valence — positivity / happiness (0.0 = dark/sad, 1.0 = bright/joyful)
48 #: tension — harmonic / rhythmic tension (0.0 = resolved, 1.0 = highly tense)
49 #: darkness — heaviness / weight (0.0 = light, 1.0 = heavy/dark)
50 EMOTION_VECTORS: dict[str, tuple[float, float, float, float]] = {
51 "joyful": (0.80, 0.90, 0.20, 0.10),
52 "melancholic": (0.30, 0.30, 0.40, 0.60),
53 "anxious": (0.60, 0.20, 0.80, 0.50),
54 "cinematic": (0.55, 0.50, 0.50, 0.40),
55 "peaceful": (0.20, 0.70, 0.10, 0.20),
56 "dramatic": (0.80, 0.30, 0.70, 0.60),
57 "hopeful": (0.60, 0.70, 0.30, 0.20),
58 "tense": (0.70, 0.20, 0.90, 0.50),
59 "dark": (0.40, 0.20, 0.50, 0.80),
60 "euphoric": (0.90, 0.90, 0.30, 0.10),
61 "serene": (0.25, 0.65, 0.15, 0.25),
62 "epic": (0.85, 0.55, 0.65, 0.45),
63 "mysterious": (0.35, 0.40, 0.60, 0.55),
64 "aggressive": (0.90, 0.25, 0.85, 0.70),
65 "nostalgic": (0.35, 0.50, 0.35, 0.50),
66 }
67
68 #: Ordered tuple of dimension names (index-stable for vector arithmetic).
69 EMOTION_DIMENSIONS: tuple[str, ...] = ("energy", "valence", "tension", "darkness")
70
71
72 # ---------------------------------------------------------------------------
73 # Result types
74 # ---------------------------------------------------------------------------
75
76
77 @dataclass(frozen=True)
78 class EmotionVector:
79 """4-dimensional emotion representation in [0.0, 1.0] per dimension.
80
81 Attributes:
82 energy: Activity / rhythmic intensity.
83 valence: Positivity / happiness.
84 tension: Harmonic / rhythmic tension.
85 darkness: Heaviness / weight.
86 """
87
88 energy: float
89 valence: float
90 tension: float
91 darkness: float
92
93 def as_tuple(self) -> tuple[float, float, float, float]:
94 """Return dimensions in :data:`EMOTION_DIMENSIONS` order."""
95 return (self.energy, self.valence, self.tension, self.darkness)
96
97 def drift_from(self, other: EmotionVector) -> float:
98 """Euclidean distance between *self* and *other* in emotion space.
99
100 Range: [0.0, 2.0] (maximum when all four dimensions flip from 0 to 1).
101 A drift > 0.5 is considered a significant emotional shift.
102
103 Args:
104 other: The reference vector (commit A).
105
106 Returns:
107 Euclidean distance rounded to 4 decimal places.
108 """
109 return round(
110 math.sqrt(sum((a - b) ** 2 for a, b in zip(self.as_tuple(), other.as_tuple()))),
111 4,
112 )
113
114
115 @dataclass(frozen=True)
116 class EmotionDimDelta:
117 """Delta for a single emotion dimension between two commits.
118
119 Attributes:
120 dimension: Dimension name (one of :data:`EMOTION_DIMENSIONS`).
121 value_a: Value at commit A.
122 value_b: Value at commit B.
123 delta: ``value_b - value_a``; positive = increased, negative = decreased.
124 """
125
126 dimension: str
127 value_a: float
128 value_b: float
129 delta: float
130
131
132 @dataclass(frozen=True)
133 class EmotionDiffResult:
134 """Full emotion-diff report between two Muse commits.
135
136 Attributes:
137 commit_a: Short (8-char) ref of the first commit.
138 commit_b: Short (8-char) ref of the second commit.
139 source: ``"explicit_tags"`` | ``"inferred"`` | ``"mixed"``
140 how the emotion vectors were obtained.
141 label_a: Emotion label for commit A (e.g. ``"melancholic"``),
142 or ``None`` when inferred without a known label.
143 label_b: Emotion label for commit B, or ``None``.
144 vector_a: Emotion vector at commit A, or ``None`` if unavailable.
145 vector_b: Emotion vector at commit B, or ``None`` if unavailable.
146 dimensions: Per-dimension deltas between the two vectors.
147 drift: Euclidean distance in emotion space.
148 narrative: Human-readable summary of the emotional shift.
149 track: Track filter applied (or ``None``).
150 section: Section filter applied (or ``None``).
151 """
152
153 commit_a: str
154 commit_b: str
155 source: str
156 label_a: str | None
157 label_b: str | None
158 vector_a: EmotionVector | None
159 vector_b: EmotionVector | None
160 dimensions: tuple[EmotionDimDelta, ...]
161 drift: float
162 narrative: str
163 track: str | None
164 section: str | None
165
166
167 # ---------------------------------------------------------------------------
168 # Pure helpers
169 # ---------------------------------------------------------------------------
170
171
172 def vector_from_label(label: str) -> EmotionVector | None:
173 """Look up the canonical :class:`EmotionVector` for an emotion label.
174
175 The *label* should be the suffix of an ``emotion:*`` tag (e.g. ``"melancholic"``).
176 Returns ``None`` for unknown labels so callers can fall back to inference.
177
178 Args:
179 label: Lowercase emotion label string.
180
181 Returns:
182 :class:`EmotionVector` if known, ``None`` otherwise.
183 """
184 entry = EMOTION_VECTORS.get(label.lower())
185 if entry is None:
186 return None
187 energy, valence, tension, darkness = entry
188 return EmotionVector(energy=energy, valence=valence, tension=tension, darkness=darkness)
189
190
191 def infer_vector_from_metadata(commit_metadata: dict[str, object] | None) -> EmotionVector:
192 """Infer an emotion vector from available commit metadata.
193
194 Uses ``tempo_bpm`` (from ``muse tempo --set``) as the primary signal:
195 - Higher tempo → higher energy, lower darkness.
196 - Absent metadata → returns a neutral midpoint vector.
197
198 Full MIDI-feature inference (mode detection, note density, velocity
199 analysis) is tracked as a follow-up and will supersede this stub when
200 MIDI content is queryable at commit time.
201
202 Args:
203 commit_metadata: The ``commit_metadata`` JSON blob from
204 :class:`~maestro.muse_cli.models.MuseCliCommit`, or ``None``.
205
206 Returns:
207 An :class:`EmotionVector` inferred from available signals.
208 """
209 if not commit_metadata:
210 # Neutral midpoint — no musical signal available
211 return EmotionVector(energy=0.50, valence=0.50, tension=0.50, darkness=0.50)
212
213 tempo_bpm = commit_metadata.get("tempo_bpm")
214 if tempo_bpm is None or not isinstance(tempo_bpm, (int, float)):
215 return EmotionVector(energy=0.50, valence=0.50, tension=0.50, darkness=0.50)
216
217 # Normalize tempo: 60 BPM = 0.0 energy, 180 BPM = 1.0 energy
218 tempo_f = float(tempo_bpm)
219 energy = min(1.0, max(0.0, (tempo_f - 60.0) / 120.0))
220 # Fast tempo correlates slightly with higher valence (major-feel dance music)
221 valence = min(1.0, max(0.0, 0.3 + energy * 0.4))
222 # Fast tempo can increase rhythmic tension up to a point
223 tension = min(1.0, max(0.0, 0.2 + energy * 0.5))
224 # Darkness inversely correlates with energy at moderate tempos
225 darkness = min(1.0, max(0.0, 0.7 - energy * 0.6))
226
227 return EmotionVector(
228 energy=round(energy, 4),
229 valence=round(valence, 4),
230 tension=round(tension, 4),
231 darkness=round(darkness, 4),
232 )
233
234
235 def compute_dimension_deltas(
236 vec_a: EmotionVector,
237 vec_b: EmotionVector,
238 ) -> tuple[EmotionDimDelta, ...]:
239 """Compute per-dimension deltas between two emotion vectors.
240
241 Args:
242 vec_a: Vector at commit A (baseline).
243 vec_b: Vector at commit B (target).
244
245 Returns:
246 Tuple of :class:`EmotionDimDelta` in :data:`EMOTION_DIMENSIONS` order.
247 """
248 dims = zip(EMOTION_DIMENSIONS, vec_a.as_tuple(), vec_b.as_tuple())
249 return tuple(
250 EmotionDimDelta(
251 dimension=dim,
252 value_a=round(a, 4),
253 value_b=round(b, 4),
254 delta=round(b - a, 4),
255 )
256 for dim, a, b in dims
257 )
258
259
260 def build_narrative(
261 label_a: str | None,
262 label_b: str | None,
263 dimensions: tuple[EmotionDimDelta, ...],
264 drift: float,
265 source: str,
266 ) -> str:
267 """Produce a human-readable narrative of the emotional shift.
268
269 The narrative describes the direction and magnitude of change using
270 production-vocabulary language. Agents use this to decide whether a
271 compositional decision is reinforcing or subverting the intended arc.
272
273 Args:
274 label_a: Emotion label at commit A (or ``None``).
275 label_b: Emotion label at commit B (or ``None``).
276 dimensions: Per-dimension deltas from :func:`compute_dimension_deltas`.
277 drift: Euclidean drift distance.
278 source: Sourcing strategy (``"explicit_tags"`` | ``"inferred"`` | ``"mixed"``).
279
280 Returns:
281 Human-readable narrative string.
282 """
283 if drift < 0.05:
284 magnitude = "minimal"
285 verdict = "Emotional character unchanged."
286 elif drift < 0.25:
287 magnitude = "subtle"
288 verdict = "Slight emotional shift."
289 elif drift < 0.50:
290 magnitude = "moderate"
291 verdict = "Noticeable emotional change."
292 elif drift < 0.80:
293 magnitude = "significant"
294 verdict = "Strong emotional shift — compositional direction changed."
295 else:
296 magnitude = "major"
297 verdict = "Dramatic emotional departure — a fundamentally different mood."
298
299 # Build label transition string
300 if label_a and label_b:
301 transition = f"{label_a} → {label_b}"
302 elif label_a:
303 transition = f"{label_a} → (inferred)"
304 elif label_b:
305 transition = f"(inferred) → {label_b}"
306 else:
307 transition = "(inferred) → (inferred)"
308
309 # Dominant dimension change
310 biggest = max(dimensions, key=lambda d: abs(d.delta))
311 sign = "+" if biggest.delta > 0 else ""
312 dim_note = f"+{biggest.dimension}" if biggest.delta > 0 else f"-{biggest.dimension}"
313 if abs(biggest.delta) < 0.02:
314 dim_note = "no dominant shift"
315
316 source_note = " [inferred from metadata]" if source != "explicit_tags" else ""
317
318 return (
319 f"{verdict} {transition} (drift={drift:.3f}, {magnitude}, "
320 f"dominant: {dim_note}){source_note}"
321 )
322
323
324 # ---------------------------------------------------------------------------
325 # Async DB helpers
326 # ---------------------------------------------------------------------------
327
328
329 async def get_emotion_tag(
330 session: AsyncSession,
331 repo_id: str,
332 commit_id: str,
333 ) -> str | None:
334 """Return the first ``emotion:*`` tag for *commit_id*, or ``None``.
335
336 Args:
337 session: Open async DB session.
338 repo_id: Repository identifier.
339 commit_id: Full 64-char commit hash.
340
341 Returns:
342 The label portion of the ``emotion:<label>`` tag (e.g. ``"melancholic"``),
343 or ``None`` if no emotion tag is attached.
344 """
345 result = await session.execute(
346 select(MuseCliTag.tag)
347 .where(
348 MuseCliTag.repo_id == repo_id,
349 MuseCliTag.commit_id == commit_id,
350 MuseCliTag.tag.like("emotion:%"),
351 )
352 .limit(1)
353 )
354 row = result.scalar_one_or_none()
355 if row is None:
356 return None
357 # Strip "emotion:" prefix
358 return row[len("emotion:"):]
359
360
361 async def resolve_commit_id(
362 session: AsyncSession,
363 repo_id: str,
364 ref: str,
365 branch: str,
366 ) -> str | None:
367 """Resolve a commit ref to a full commit ID.
368
369 Supported refs:
370 - Full 64-char hash — returned as-is (after existence check).
371 - ``HEAD`` — resolves to the most recent commit on *branch*.
372 - ``HEAD~N`` — walks N parents back from HEAD (e.g. ``HEAD~1``).
373 - 8-char abbreviated hash — matches any commit ID with that prefix.
374
375 Args:
376 session: Open async DB session.
377 repo_id: Repository identifier.
378 ref: Commit reference string.
379 branch: Current branch name (used for HEAD resolution).
380
381 Returns:
382 Full 64-char commit ID, or ``None`` if the ref cannot be resolved.
383 """
384 # ── HEAD~N shorthand ─────────────────────────────────────────────────
385 head_tilde_n = 0
386 lookup_ref = ref
387 if ref.upper() == "HEAD" or ref.upper().startswith("HEAD~"):
388 if ref.upper() == "HEAD":
389 head_tilde_n = 0
390 else:
391 try:
392 head_tilde_n = int(ref[5:]) # strip "HEAD~"
393 except ValueError:
394 return None
395 lookup_ref = "HEAD"
396
397 if lookup_ref.upper() == "HEAD":
398 # Find HEAD commit for branch
399 result = await session.execute(
400 select(MuseCliCommit)
401 .where(
402 MuseCliCommit.repo_id == repo_id,
403 MuseCliCommit.branch == branch,
404 )
405 .order_by(MuseCliCommit.committed_at.desc())
406 .limit(1)
407 )
408 commit = result.scalar_one_or_none()
409 if commit is None:
410 return None
411 # Walk N parents back
412 for _ in range(head_tilde_n):
413 if commit.parent_commit_id is None:
414 return None
415 parent = await session.get(MuseCliCommit, commit.parent_commit_id)
416 if parent is None:
417 return None
418 commit = parent
419 return commit.commit_id
420
421 # ── Full 64-char hash ─────────────────────────────────────────────────
422 if len(ref) == 64:
423 commit = await session.get(MuseCliCommit, ref)
424 return ref if commit is not None else None
425
426 # ── Abbreviated hash (prefix match) ──────────────────────────────────
427 result = await session.execute(
428 select(MuseCliCommit.commit_id)
429 .where(
430 MuseCliCommit.repo_id == repo_id,
431 MuseCliCommit.commit_id.like(f"{ref}%"),
432 )
433 .limit(1)
434 )
435 return result.scalar_one_or_none() # type: ignore[return-value] # SQLAlchemy scalar() -> Any
436
437
438 # ---------------------------------------------------------------------------
439 # Public API
440 # ---------------------------------------------------------------------------
441
442
443 async def compute_emotion_diff(
444 session: AsyncSession,
445 *,
446 repo_id: str,
447 commit_a: str,
448 commit_b: str,
449 branch: str,
450 track: str | None = None,
451 section: str | None = None,
452 ) -> EmotionDiffResult:
453 """Compute an emotion-diff between two Muse commits.
454
455 Sourcing strategy (in priority order):
456 1. Both commits have ``emotion:*`` tags → ``"explicit_tags"``.
457 2. One has a tag, the other is inferred → ``"mixed"``.
458 3. Neither has a tag → ``"inferred"`` from commit metadata.
459
460 Args:
461 session: Open async DB session.
462 repo_id: Repository identifier (from ``.muse/repo.json``).
463 commit_a: Commit reference for the baseline (e.g. ``"HEAD~1"``).
464 commit_b: Commit reference for the target (e.g. ``"HEAD"``).
465 branch: Current branch name (used for HEAD resolution).
466 track: Optional track name filter (noted in result; full filtering
467 requires MIDI content access — tracked as follow-up).
468 section: Optional section name filter (same stub note as *track*).
469
470 Returns:
471 :class:`EmotionDiffResult` with vectors, per-dimension deltas,
472 drift distance, and a human-readable narrative.
473
474 Raises:
475 ValueError: If *commit_a* or *commit_b* cannot be resolved to a
476 commit that exists in the database.
477 """
478 # ── Resolve commit refs ───────────────────────────────────────────────
479 # Read branch from HEAD file if needed — callers should pass branch
480 resolved_a = await resolve_commit_id(session, repo_id, commit_a, branch)
481 if resolved_a is None:
482 raise ValueError(
483 f"Cannot resolve commit ref '{commit_a}' in repo '{repo_id}' "
484 f"on branch '{branch}'."
485 )
486 resolved_b = await resolve_commit_id(session, repo_id, commit_b, branch)
487 if resolved_b is None:
488 raise ValueError(
489 f"Cannot resolve commit ref '{commit_b}' in repo '{repo_id}' "
490 f"on branch '{branch}'."
491 )
492
493 short_a = resolved_a[:8]
494 short_b = resolved_b[:8]
495
496 # ── Load commit rows for metadata ────────────────────────────────────
497 row_a = await session.get(MuseCliCommit, resolved_a)
498 row_b = await session.get(MuseCliCommit, resolved_b)
499
500 # Both rows are guaranteed to exist because resolve_commit_id checked them
501 meta_a: dict[str, object] | None = row_a.commit_metadata if row_a else None
502 meta_b: dict[str, object] | None = row_b.commit_metadata if row_b else None
503
504 # ── Read explicit emotion tags ────────────────────────────────────────
505 label_a = await get_emotion_tag(session, repo_id, resolved_a)
506 label_b = await get_emotion_tag(session, repo_id, resolved_b)
507
508 # ── Resolve vectors ───────────────────────────────────────────────────
509 vec_a: EmotionVector | None = None
510 vec_b: EmotionVector | None = None
511
512 if label_a:
513 vec_a = vector_from_label(label_a)
514 if label_b:
515 vec_b = vector_from_label(label_b)
516
517 # Fall back to inference for commits without explicit tags
518 if vec_a is None:
519 vec_a = infer_vector_from_metadata(meta_a)
520 if vec_b is None:
521 vec_b = infer_vector_from_metadata(meta_b)
522
523 # ── Determine sourcing label ─────────────────────────────────────────
524 if label_a and label_b:
525 source = "explicit_tags"
526 elif label_a or label_b:
527 source = "mixed"
528 else:
529 source = "inferred"
530
531 # ── Compute deltas and drift ─────────────────────────────────────────
532 dimensions = compute_dimension_deltas(vec_a, vec_b)
533 drift = vec_b.drift_from(vec_a)
534 narrative = build_narrative(label_a, label_b, dimensions, drift, source)
535
536 if track:
537 logger.info("⚠️ --track %r: per-track emotion scoping not yet implemented", track)
538 if section:
539 logger.info(
540 "⚠️ --section %r: section-scoped emotion analysis not yet implemented", section
541 )
542
543 logger.info(
544 "✅ muse emotion-diff: %s → %s drift=%.4f source=%s",
545 short_a,
546 short_b,
547 drift,
548 source,
549 )
550
551 return EmotionDiffResult(
552 commit_a=short_a,
553 commit_b=short_b,
554 source=source,
555 label_a=label_a,
556 label_b=label_b,
557 vector_a=vec_a,
558 vector_b=vec_b,
559 dimensions=dimensions,
560 drift=drift,
561 narrative=narrative,
562 track=track,
563 section=section,
564 )