diff --git a/services/analysis-engine/src/bandscope_analysis/api.py b/services/analysis-engine/src/bandscope_analysis/api.py index 97703e7..87fce60 100644 --- a/services/analysis-engine/src/bandscope_analysis/api.py +++ b/services/analysis-engine/src/bandscope_analysis/api.py @@ -2,9 +2,10 @@ from __future__ import annotations -from typing import Literal, NotRequired, TypedDict +from typing import Any, Literal, NotRequired, TypedDict, cast from bandscope_analysis.health import HealthReport, build_health_report +from bandscope_analysis.roles import RoleExtractor from bandscope_analysis.sections import extract_sections @@ -88,6 +89,15 @@ class RehearsalRolePayload(TypedDict): manualOverrides: list[ManualOverridePayload] +class PartGraphNodePayload(TypedDict): + """Typed part-graph node payload nested inside sections.""" + + role_id: str + is_active: bool + handoff_to: list[str] + handoff_from: list[str] + + class RehearsalSectionPayload(TypedDict): """Typed rehearsal section payload nested inside songs.""" @@ -96,6 +106,7 @@ class RehearsalSectionPayload(TypedDict): groove: str confidence: ConfidencePayload roles: list[RehearsalRolePayload] + partGraph: list[PartGraphNodePayload] class ExportSummaryPayload(TypedDict): @@ -212,6 +223,12 @@ def build_demo_rehearsal_song() -> RehearsalSong: extraction_result = extract_sections(arrangement) verse_section = extraction_result["sections"][0] + # Extract roles + extractor = RoleExtractor() + role_result = extractor.extract([verse_section]) + verse_topology = role_result["topologies"][0] + verse_roles = verse_topology["active_roles"] + return { "id": "demo-song", "title": "Late Night Set", @@ -225,91 +242,8 @@ def build_demo_rehearsal_song() -> RehearsalSong: "source": "model", "notes": "Double-check the pickup into the chorus.", }, - "roles": [ - { - "id": "bass-guitar", - "name": "Bass Guitar", - "roleType": "instrument", - "harmony": { - "chord": "C#m7", - "functionLabel": "vi pedal anchor", - "source": "model", - }, - "cue": { - "kind": "transition", - "value": "Hold through the pickup before the downbeat.", - }, - "range": {"lowestNote": "C#2", "highestNote": "E3"}, - "confidence": { - "level": "medium", - "source": "model", - "notes": "Watch the slide into the turnaround.", - }, - "rehearsalPriority": "high", - "simplification": "Stay on roots if the chorus entrance gets muddy.", - "setupNote": "Keep the attack short so the verse breathes.", - "manualOverrides": [], - }, - { - "id": "keys-right", - "name": "Keyboard 1 Right Hand", - "roleType": "hand", - "harmony": { - "chord": "Emaj7", - "functionLabel": "Imaj7 color", - "source": "model", - }, - "cue": { - "kind": "count", - "value": "Enter on beat 2 after the pickup.", - }, - "range": {"lowestNote": "B3", "highestNote": "G#5"}, - "confidence": { - "level": "medium", - "source": "model", - "notes": "Top note voicing may need a quick ear check.", - }, - "rehearsalPriority": "high", - "simplification": ( - "Drop the top extension if the chorus turnaround still feels busy." - ), - "setupNote": "Keep the patch bright enough to stay over the guitars.", - "manualOverrides": [], - }, - { - "id": "lead-vocal", - "name": "Lead Vocal", - "roleType": "vocal", - "harmony": { - "chord": "C#m7", - "functionLabel": "vi melodic pull", - "source": "model", - }, - "cue": {"kind": "lyric", "value": "city lights"}, - "range": {"lowestNote": "G#3", "highestNote": "C#5"}, - "confidence": { - "level": "high", - "source": "user", - "notes": "Singer confirmed the pickup phrasing in rehearsal notes.", - }, - "rehearsalPriority": "medium", - "simplification": ( - "Keep the sustained note centered; skip the ad-lib on the first pass." - ), - "setupNote": "Watch the breath before the last line of the verse.", - "manualOverrides": [ - { - "field": "harmony", - "value": { - "chord": "C#m11", - "functionLabel": "vi suspended lift", - "source": "user", - }, - "source": "user", - } - ], - }, - ], + "roles": cast(list[RehearsalRolePayload], verse_roles), + "partGraph": cast(Any, verse_topology["part_graph"]), } ], "exportSummary": { diff --git a/services/analysis-engine/src/bandscope_analysis/roles/__init__.py b/services/analysis-engine/src/bandscope_analysis/roles/__init__.py new file mode 100644 index 0000000..74da0b9 --- /dev/null +++ b/services/analysis-engine/src/bandscope_analysis/roles/__init__.py @@ -0,0 +1,23 @@ +"""Role extraction and part graphing module.""" + +from .extractor import RoleExtractor +from .model import ( + CueAnchorKind, + PartGraphNode, + RehearsalPriority, + RehearsalRole, + RoleExtractionResult, + RoleType, + SectionRoleTopology, +) + +__all__ = [ + "RoleType", + "RehearsalPriority", + "CueAnchorKind", + "RehearsalRole", + "PartGraphNode", + "SectionRoleTopology", + "RoleExtractionResult", + "RoleExtractor", +] diff --git a/services/analysis-engine/src/bandscope_analysis/roles/extractor.py b/services/analysis-engine/src/bandscope_analysis/roles/extractor.py new file mode 100644 index 0000000..34d7b37 --- /dev/null +++ b/services/analysis-engine/src/bandscope_analysis/roles/extractor.py @@ -0,0 +1,190 @@ +"""Role extractor implementation.""" + +from __future__ import annotations + +import logging +from typing import Any + +from .model import ( + CueAnchorKind, + PartGraphNode, + RehearsalPriority, + RehearsalRole, + RoleExtractionResult, + RoleType, + SectionRoleTopology, +) + +logger = logging.getLogger(__name__) + + +class RoleExtractor: + """Extracts roles and builds the part graph for song sections.""" + + def __init__(self) -> None: + """Initialize the role extractor.""" + pass + + def extract( + self, + sections: list[Any], + _audio_features: dict[str, Any] | None = None, + ) -> RoleExtractionResult: + """Extract roles and their topology per section. + + Args: + sections: List of section dicts (must contain 'id'). + _audio_features: Optional audio features to inform extraction. + + Returns: + RoleExtractionResult containing topologies and notes. + """ + topologies: list[SectionRoleTopology] = [] + + # Simple mock implementation for testing/demonstration purposes + for i, section in enumerate(sections): + if not isinstance(section, dict): + logger.warning( + "Invalid section format at index %d; expected dict, got %s", + i, + type(section).__name__, + ) + section_id = f"section-{i}" + else: + section_id = section.get("id", f"section-{i}") + + # Create a mock bass role + bass_role: RehearsalRole = { + "id": "bass-guitar", + "name": "Bass Guitar", + "roleType": RoleType.INSTRUMENT, + "harmony": {"chord": "C#m7", "functionLabel": "vi pedal anchor", "source": "model"}, + "cue": { + "kind": CueAnchorKind.TRANSITION, + "value": "Hold through the pickup before the downbeat.", + }, + "range": {"lowestNote": "C#2", "highestNote": "E3"}, + "confidence": { + "level": "medium", + "source": "model", + "notes": "Watch the slide into the turnaround.", + }, + "rehearsalPriority": RehearsalPriority.HIGH, + "simplification": "Stay on roots if the chorus entrance gets muddy.", + "setupNote": "Keep the attack short so the verse breathes.", + "manualOverrides": [], + } + + keys_role: RehearsalRole = { + "id": "keys-right", + "name": "Keyboard 1 Right Hand", + "roleType": RoleType.HAND, + "harmony": { + "chord": "Emaj7", + "functionLabel": "Imaj7 color", + "source": "model", + }, + "cue": { + "kind": CueAnchorKind.COUNT, + "value": "Enter on beat 2 after the pickup.", + }, + "range": {"lowestNote": "B3", "highestNote": "G#5"}, + "confidence": { + "level": "medium", + "source": "model", + "notes": "Top note voicing may need a quick ear check.", + }, + "rehearsalPriority": RehearsalPriority.HIGH, + "simplification": "Drop top extension if the chorus turnaround feels busy.", + "setupNote": "Keep the patch bright enough to stay over the guitars.", + "manualOverrides": [], + } + + vocal_role: RehearsalRole = { + "id": "lead-vocal", + "name": "Lead Vocal", + "roleType": RoleType.VOCAL, + "harmony": { + "chord": "C#m7", + "functionLabel": "vi melodic pull", + "source": "model", + }, + "cue": {"kind": CueAnchorKind.LYRIC, "value": "city lights"}, + "range": {"lowestNote": "G#3", "highestNote": "C#5"}, + "confidence": { + "level": "high", + "source": "user", + "notes": "Singer confirmed the pickup phrasing in rehearsal notes.", + }, + "rehearsalPriority": RehearsalPriority.MEDIUM, + "simplification": "Keep sustained note centered; skip ad-lib on first pass.", + "setupNote": "Watch the breath before the last line of the verse.", + "manualOverrides": [ + { + "field": "harmony", + "value": { + "chord": "C#m11", + "functionLabel": "vi suspended lift", + "source": "user", + }, + "source": "user", + } + ], + } + + active_roles = [bass_role] + + # Simple part graph for bass + part_graph: list[PartGraphNode] = [ + {"role_id": "bass-guitar", "is_active": True, "handoff_to": [], "handoff_from": []} + ] + + if i == 0: + active_roles.extend([keys_role, vocal_role]) + part_graph.extend( + [ + { + "role_id": "keys-right", + "is_active": True, + "handoff_to": [], + "handoff_from": [], + }, + { + "role_id": "lead-vocal", + "is_active": True, + "handoff_to": [], + "handoff_from": [], + }, + ] + ) + part_graph[0]["handoff_to"].append("lead-vocal") + part_graph[2]["handoff_from"].append("bass-guitar") + else: + part_graph.extend( + [ + { + "role_id": "keys-right", + "is_active": False, + "handoff_to": [], + "handoff_from": [], + }, + { + "role_id": "lead-vocal", + "is_active": False, + "handoff_to": [], + "handoff_from": [], + }, + ] + ) + + topology: SectionRoleTopology = { + "section_id": section_id, + "active_roles": active_roles, + "part_graph": part_graph, + } + topologies.append(topology) + + return { + "topologies": topologies, + "extraction_notes": "Extracted roles and computed handoffs.", + } diff --git a/services/analysis-engine/src/bandscope_analysis/roles/model.py b/services/analysis-engine/src/bandscope_analysis/roles/model.py new file mode 100644 index 0000000..06bac5a --- /dev/null +++ b/services/analysis-engine/src/bandscope_analysis/roles/model.py @@ -0,0 +1,108 @@ +"""Domain model for role extraction and part graphing.""" + +from __future__ import annotations + +from enum import Enum +from typing import Any, Literal, TypedDict + + +class RoleType(str, Enum): + """Canonical role types.""" + + INSTRUMENT = "instrument" + VOCAL = "vocal" + HAND = "hand" + + +class RehearsalPriority(str, Enum): + """Rehearsal priority for a given role in a section.""" + + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + + +class ConfidenceMarker(TypedDict): + """Confidence level and notes for a field or role.""" + + level: Literal["low", "medium", "high"] + source: Literal["model", "user"] + notes: str + + +class RehearsalHarmony(TypedDict): + """Harmony specifics for a role.""" + + chord: str + functionLabel: str + source: Literal["model", "user"] + + +class RangeSummary(TypedDict): + """Range summary for a role.""" + + lowestNote: str + highestNote: str + + +class CueAnchorKind(str, Enum): + """Kinds of cue anchor.""" + + LYRIC = "lyric" + COUNT = "count" + TRANSITION = "transition" + + +class RoleCueAnchor(TypedDict): + """A cue anchor for a role.""" + + kind: CueAnchorKind + value: str + + +class ManualOverride(TypedDict): + """A manual override applied to a role field.""" + + field: str + value: dict[str, Any] + source: str + + +class RehearsalRole(TypedDict): + """A role (instrument, vocal, or hand) active in a particular section.""" + + id: str + name: str + roleType: RoleType + harmony: RehearsalHarmony + cue: RoleCueAnchor + range: RangeSummary + confidence: ConfidenceMarker + rehearsalPriority: RehearsalPriority + simplification: str + setupNote: str + manualOverrides: list[ManualOverride] + + +class PartGraphNode(TypedDict): + """A node representing a role in the part graph for a section.""" + + role_id: str + is_active: bool + handoff_to: list[str] + handoff_from: list[str] + + +class SectionRoleTopology(TypedDict): + """The topology of roles within a single section.""" + + section_id: str + active_roles: list[RehearsalRole] + part_graph: list[PartGraphNode] + + +class RoleExtractionResult(TypedDict): + """Result returned by the role extraction pipeline.""" + + topologies: list[SectionRoleTopology] + extraction_notes: str diff --git a/services/analysis-engine/tests/test_roles.py b/services/analysis-engine/tests/test_roles.py new file mode 100644 index 0000000..a1ada7a --- /dev/null +++ b/services/analysis-engine/tests/test_roles.py @@ -0,0 +1,92 @@ +"""Tests for the role extraction and part graph models.""" + +from bandscope_analysis.roles.extractor import RoleExtractor +from bandscope_analysis.roles.model import ( + CueAnchorKind, + RehearsalPriority, + RoleType, +) + + +def test_role_type_enum(): + """Verify RoleType enum values match the domain requirements.""" + assert RoleType.INSTRUMENT == "instrument" + assert RoleType.VOCAL == "vocal" + assert RoleType.HAND == "hand" + + +def test_rehearsal_priority_enum(): + """Verify RehearsalPriority enum values match.""" + assert RehearsalPriority.LOW == "low" + assert RehearsalPriority.MEDIUM == "medium" + assert RehearsalPriority.HIGH == "high" + + +def test_cue_anchor_kind_enum(): + """Verify CueAnchorKind enum values match.""" + assert CueAnchorKind.LYRIC == "lyric" + assert CueAnchorKind.COUNT == "count" + assert CueAnchorKind.TRANSITION == "transition" + + +def test_role_extractor_basic(): + """Test that RoleExtractor returns a valid topology structure.""" + extractor = RoleExtractor() + + sections = [{"id": "intro"}, {"id": "verse-1"}] + + result = extractor.extract(sections) + + assert "topologies" in result + assert "extraction_notes" in result + assert len(result["topologies"]) == 2 + + # Check intro section + intro_topology = result["topologies"][0] + assert intro_topology["section_id"] == "intro" + assert len(intro_topology["active_roles"]) == 3 + + roles_by_id = {r["id"]: r for r in intro_topology["active_roles"]} + assert "bass-guitar" in roles_by_id + assert "lead-vocal" in roles_by_id + assert "keys-right" in roles_by_id + assert roles_by_id["lead-vocal"]["roleType"] == "vocal" + + intro_graph = intro_topology["part_graph"] + graph_by_role = {n["role_id"]: n for n in intro_graph} + + # Check handoff relation + assert "lead-vocal" in graph_by_role["bass-guitar"]["handoff_to"] + assert "bass-guitar" in graph_by_role["lead-vocal"]["handoff_from"] + + # Check verse-1 section (only bass) + verse_topology = result["topologies"][1] + assert verse_topology["section_id"] == "verse-1" + assert len(verse_topology["active_roles"]) == 1 + assert verse_topology["active_roles"][0]["id"] == "bass-guitar" + assert verse_topology["active_roles"][0]["roleType"] == "instrument" + assert verse_topology["active_roles"][0]["rehearsalPriority"] == "high" + + verse_graph = verse_topology["part_graph"] + assert len(verse_graph) == 3 + assert verse_graph[1]["role_id"] == "keys-right" + assert verse_graph[1]["is_active"] is False + assert verse_graph[0]["role_id"] == "bass-guitar" + assert verse_graph[0]["handoff_to"] == [] + + +def test_role_extractor_empty(): + """Test extractor with empty sections list.""" + extractor = RoleExtractor() + result = extractor.extract([]) + assert result["topologies"] == [] + + +def test_role_extractor_invalid_section(): + """Test that RoleExtractor handles non-dict sections gracefully.""" + extractor = RoleExtractor() + sections = [{"id": "intro"}, "invalid-section-string"] + result = extractor.extract(sections) + assert len(result["topologies"]) == 2 + assert result["topologies"][0]["section_id"] == "intro" + assert result["topologies"][1]["section_id"] == "section-1"