muse_models.py
python
| 1 | """SQLAlchemy ORM models for Muse persistent variation history. |
| 2 | |
| 3 | Tables: |
| 4 | - muse_variations: Top-level variation proposals with lineage tracking |
| 5 | - muse_phrases: Independently reviewable musical phrases within a variation |
| 6 | - muse_note_changes: Individual note-level diffs within a phrase |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | from datetime import datetime, timezone |
| 12 | from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, Text |
| 13 | from sqlalchemy.orm import Mapped, mapped_column, relationship |
| 14 | from sqlalchemy.types import JSON |
| 15 | |
| 16 | from maestro.contracts.json_types import AftertouchDict, CCEventDict, NoteDict, PitchBendDict |
| 17 | from maestro.db.database import Base |
| 18 | from maestro.db.models import generate_uuid, utc_now |
| 19 | |
| 20 | |
| 21 | class Variation(Base): |
| 22 | """A persisted variation proposal with lineage tracking.""" |
| 23 | |
| 24 | __tablename__ = "muse_variations" |
| 25 | |
| 26 | variation_id: Mapped[str] = mapped_column(String(36), primary_key=True) |
| 27 | project_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) |
| 28 | base_state_id: Mapped[str] = mapped_column(String(36), nullable=False) |
| 29 | conversation_id: Mapped[str] = mapped_column(String(36), nullable=False, default="") |
| 30 | intent: Mapped[str] = mapped_column(Text, nullable=False) |
| 31 | explanation: Mapped[str | None] = mapped_column(Text, nullable=True) |
| 32 | status: Mapped[str] = mapped_column(String(20), nullable=False, default="created") |
| 33 | affected_tracks: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) |
| 34 | affected_regions: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) |
| 35 | beat_range_start: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) |
| 36 | beat_range_end: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) |
| 37 | |
| 38 | # ── Lineage (Phase 5) ──────────────────────────────────────────── |
| 39 | parent_variation_id: Mapped[str | None] = mapped_column( |
| 40 | String(36), |
| 41 | ForeignKey("muse_variations.variation_id", ondelete="SET NULL"), |
| 42 | nullable=True, |
| 43 | index=True, |
| 44 | ) |
| 45 | parent2_variation_id: Mapped[str | None] = mapped_column( |
| 46 | String(36), |
| 47 | ForeignKey("muse_variations.variation_id", ondelete="SET NULL"), |
| 48 | nullable=True, |
| 49 | ) |
| 50 | commit_state_id: Mapped[str | None] = mapped_column(String(36), nullable=True) |
| 51 | is_head: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) |
| 52 | |
| 53 | created_at: Mapped[datetime] = mapped_column( |
| 54 | DateTime(timezone=True), default=utc_now, nullable=False, |
| 55 | ) |
| 56 | updated_at: Mapped[datetime] = mapped_column( |
| 57 | DateTime(timezone=True), default=utc_now, onupdate=utc_now, nullable=False, |
| 58 | ) |
| 59 | |
| 60 | phrases: Mapped[list["Phrase"]] = relationship( |
| 61 | "Phrase", |
| 62 | back_populates="variation", |
| 63 | cascade="all, delete-orphan", |
| 64 | order_by="Phrase.sequence", |
| 65 | ) |
| 66 | children: Mapped[list["Variation"]] = relationship( |
| 67 | "Variation", |
| 68 | backref="parent", |
| 69 | remote_side=[variation_id], |
| 70 | foreign_keys=[parent_variation_id], |
| 71 | ) |
| 72 | |
| 73 | def __repr__(self) -> str: |
| 74 | return f"<Variation {self.variation_id[:8]} status={self.status} head={self.is_head}>" |
| 75 | |
| 76 | |
| 77 | class Phrase(Base): |
| 78 | """A persisted musical phrase within a variation.""" |
| 79 | |
| 80 | __tablename__ = "muse_phrases" |
| 81 | |
| 82 | phrase_id: Mapped[str] = mapped_column(String(36), primary_key=True) |
| 83 | variation_id: Mapped[str] = mapped_column( |
| 84 | String(36), |
| 85 | ForeignKey("muse_variations.variation_id", ondelete="CASCADE"), |
| 86 | nullable=False, |
| 87 | index=True, |
| 88 | ) |
| 89 | sequence: Mapped[int] = mapped_column(Integer, nullable=False) |
| 90 | track_id: Mapped[str] = mapped_column(String(36), nullable=False) |
| 91 | region_id: Mapped[str] = mapped_column(String(36), nullable=False) |
| 92 | start_beat: Mapped[float] = mapped_column(Float, nullable=False) |
| 93 | end_beat: Mapped[float] = mapped_column(Float, nullable=False) |
| 94 | label: Mapped[str] = mapped_column(String(255), nullable=False) |
| 95 | tags: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) |
| 96 | explanation: Mapped[str | None] = mapped_column(Text, nullable=True) |
| 97 | cc_events: Mapped[list[CCEventDict] | None] = mapped_column(JSON, nullable=True) |
| 98 | pitch_bends: Mapped[list[PitchBendDict] | None] = mapped_column(JSON, nullable=True) |
| 99 | aftertouch: Mapped[list[AftertouchDict] | None] = mapped_column(JSON, nullable=True) |
| 100 | |
| 101 | region_start_beat: Mapped[float | None] = mapped_column(Float, nullable=True) |
| 102 | region_duration_beats: Mapped[float | None] = mapped_column(Float, nullable=True) |
| 103 | region_name: Mapped[str | None] = mapped_column(String(255), nullable=True) |
| 104 | |
| 105 | variation: Mapped["Variation"] = relationship("Variation", back_populates="phrases") |
| 106 | note_changes: Mapped[list["NoteChange"]] = relationship( |
| 107 | "NoteChange", |
| 108 | back_populates="phrase", |
| 109 | cascade="all, delete-orphan", |
| 110 | ) |
| 111 | |
| 112 | def __repr__(self) -> str: |
| 113 | return f"<Phrase {self.phrase_id[:8]} {self.label}>" |
| 114 | |
| 115 | |
| 116 | class NoteChange(Base): |
| 117 | """A persisted note-level diff within a phrase.""" |
| 118 | |
| 119 | __tablename__ = "muse_note_changes" |
| 120 | |
| 121 | id: Mapped[str] = mapped_column(String(36), primary_key=True, default=generate_uuid) |
| 122 | phrase_id: Mapped[str] = mapped_column( |
| 123 | String(36), |
| 124 | ForeignKey("muse_phrases.phrase_id", ondelete="CASCADE"), |
| 125 | nullable=False, |
| 126 | index=True, |
| 127 | ) |
| 128 | change_type: Mapped[str] = mapped_column(String(20), nullable=False) |
| 129 | before_json: Mapped[NoteDict | None] = mapped_column(JSON, nullable=True) |
| 130 | after_json: Mapped[NoteDict | None] = mapped_column(JSON, nullable=True) |
| 131 | |
| 132 | phrase: Mapped["Phrase"] = relationship("Phrase", back_populates="note_changes") |
| 133 | |
| 134 | def __repr__(self) -> str: |
| 135 | return f"<NoteChange {self.id[:8]} {self.change_type}>" |