Phase 2: MIDI control — synth definitions, tools, and deps

.github/workflows/test.yml

@@ -1,4 +1,4 @@
-name: Tests
+name: CI
 
 on:
   push:
@@ -7,12 +7,27 @@ on:
     branches: [main]
 
 jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: astral-sh/setup-uv@v5
+        with:
+          python-version: "3.14"
+      - name: Install system dependencies for python-rtmidi
+        run: sudo apt-get update && sudo apt-get install -y libasound2-dev libjack-jackd2-dev
+      - run: uv sync
+      - run: uv run ruff check .
+      - run: uv run ruff format --check .
+
   test:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
       - uses: astral-sh/setup-uv@v5
         with:
           python-version: "3.14"
+      - name: Install system dependencies for python-rtmidi
+        run: sudo apt-get update && sudo apt-get install -y libasound2-dev libjack-jackd2-dev
       - run: uv sync
       - run: uv run pytest .

patchwork/agent.py

@@ -1,4 +1,13 @@
-from pydantic_ai import Agent
+from pydantic_ai import Agent, RunContext
+
+from patchwork.deps import PatchworkDeps
+from patchwork.tools.midi_control import (
+    connect_midi,
+    list_midi_ports,
+    list_synths,
+    send_cc,
+    send_patch,
+)
 
 SYSTEM_PROMPT = """You are Patchwork, an expert synthesizer sound design assistant.
 
@@ -16,8 +25,15 @@
 - 1010 Music Blackbox (sampler/groovebox)
 - Arturia Minibrute 2S (analog, semi-modular)
 
-You will eventually control these synths via MIDI CC messages. For now, describe
-settings conceptually using parameter names and values.
+You have tools to control synths via MIDI:
+- list_midi_ports: Show available MIDI output devices
+- connect_midi: Connect to a MIDI port
+- list_synths: Show loaded synth definitions and their controllable parameters
+- send_cc: Send a single MIDI CC value to a synth parameter
+- send_patch: Send multiple MIDI CC values at once to set a full patch
+
+When the user asks you to set a sound or tweak a parameter, use these tools.
+Always confirm what you sent after a successful tool call.
 
 Tone: conversational but concise. Use musical and technical terminology naturally.
 When describing a patch, be specific enough that someone could recreate it by hand.
@@ -26,5 +42,19 @@
 agent = Agent(
     "anthropic:claude-sonnet-4-6",
     system_prompt=SYSTEM_PROMPT,
+    deps_type=PatchworkDeps,
     defer_model_check=True,
+    tools=[list_midi_ports, connect_midi, list_synths, send_cc, send_patch],
 )
+
+
+@agent.system_prompt
+async def add_synth_context(ctx: RunContext[PatchworkDeps]) -> str:
+    """Append loaded synth definitions to the system prompt at runtime."""
+    if not ctx.deps.synths:
+        return "No synth definitions loaded yet."
+    lines = []
+    for synth in ctx.deps.synths.values():
+        params = ", ".join(synth.cc_map.keys())
+        lines.append(f"- {synth.manufacturer} {synth.name} (ch.{synth.midi_channel}): {params}")
+    return "Loaded synths and their controllable parameters:\n" + "\n".join(lines)

patchwork/cli.py

@@ -4,39 +4,54 @@
 from rich.console import Console
 
 from patchwork.agent import agent
+from patchwork.deps import PatchworkDeps
+from patchwork.midi import MidiConnection
+from patchwork.synth_definitions import load_synth_definitions
 
 console = Console()
 
 
 async def main():
+    midi = MidiConnection()
+    synths = load_synth_definitions()
+    deps = PatchworkDeps(midi=midi, synths=synths)
+
     console.print("[bold]patchwork[/bold] — synth research agent\n")
+    if synths:
+        synth_names = ", ".join(s.name for s in synths.values())
+        console.print(f"[dim]loaded {len(synths)} synth(s): {synth_names}[/dim]\n")
+    else:
+        console.print("[dim]no synth definitions found in synths/[/dim]\n")
     message_history = []
 
-    while True:
-        try:
-            user_input = console.input("[bold cyan]patch>[/bold cyan] ")
-        except (KeyboardInterrupt, EOFError):
-            console.print("\n[dim]goodbye[/dim]")
-            break
-
-        if not user_input.strip():
-            continue
-
-        if user_input.strip().lower() in ("quit", "exit"):
-            console.print("[dim]goodbye[/dim]")
-            break
-
-        try:
-            async with agent.run_stream(
-                user_input, message_history=message_history
-            ) as result:
-                async for chunk in result.stream_text(delta=True):
-                    console.print(chunk, end="", markup=False, highlight=False)
-                console.print()  # newline after stream
-
-            message_history = result.all_messages()
-        except Exception as e:
-            console.print(f"\n[bold red]error:[/bold red] {e}")
+    try:
+        while True:
+            try:
+                user_input = console.input("[bold cyan]patch>[/bold cyan] ")
+            except KeyboardInterrupt, EOFError:
+                console.print("\n[dim]goodbye[/dim]")
+                break
+
+            if not user_input.strip():
+                continue
+
+            if user_input.strip().lower() in ("quit", "exit"):
+                console.print("[dim]goodbye[/dim]")
+                break
+
+            try:
+                async with agent.run_stream(
+                    user_input, message_history=message_history, deps=deps
+                ) as result:
+                    async for chunk in result.stream_text(delta=True):
+                        console.print(chunk, end="", markup=False, highlight=False)
+                    console.print()  # newline after stream
+
+                message_history = result.all_messages()
+            except Exception as e:
+                console.print(f"\n[bold red]error:[/bold red] {e}")
+    finally:
+        midi.close()
 
 
 def main_cli():

patchwork/deps.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+from patchwork.midi import MidiConnection
+from patchwork.synth_definitions import SynthDefinition
+
+
+@dataclass
+class PatchworkDeps:
+    midi: MidiConnection
+    synths: dict[str, SynthDefinition]

patchwork/midi.py

@@ -0,0 +1,56 @@
+import rtmidi
+
+
+class MidiConnection:
+    def __init__(self):
+        self._out: rtmidi.MidiOut | None = None
+        self._port_name: str | None = None
+
+    @property
+    def is_connected(self) -> bool:
+        return self._out is not None
+
+    @property
+    def port_name(self) -> str | None:
+        return self._port_name
+
+    def list_ports(self) -> list[str]:
+        """List available MIDI output port names."""
+        out = rtmidi.MidiOut()
+        ports = out.get_ports()
+        del out
+        return ports
+
+    def open(self, port_index: int = 0) -> str:
+        """Open a MIDI output port by index. Returns the port name."""
+        self.close()
+        self._out = rtmidi.MidiOut()
+        ports = self._out.get_ports()
+        if not ports:
+            raise RuntimeError("No MIDI output ports available")
+        if not (0 <= port_index < len(ports)):
+            raise ValueError(f"Port index {port_index} out of range (0-{len(ports) - 1})")
+        self._out.open_port(port_index)
+        self._port_name = ports[port_index]
+        return self._port_name
+
+    def close(self):
+        """Close the current MIDI connection."""
+        if self._out is not None:
+            self._out.close_port()
+            del self._out
+            self._out = None
+            self._port_name = None
+
+    def send_cc(self, channel: int, cc_number: int, value: int):
+        """Send a MIDI CC message. Channel is 1-16 (converted to 0-15 internally)."""
+        if self._out is None:
+            raise RuntimeError("MIDI port not open — call open() first")
+        if not (1 <= channel <= 16):
+            raise ValueError(f"MIDI channel must be 1-16, got {channel}")
+        if not (0 <= cc_number <= 127):
+            raise ValueError(f"CC number must be 0-127, got {cc_number}")
+        if not (0 <= value <= 127):
+            raise ValueError(f"CC value must be 0-127, got {value}")
+        status = 0xB0 | (channel - 1)
+        self._out.send_message([status, cc_number, value])

patchwork/synth_definitions.py

@@ -0,0 +1,46 @@
+from pathlib import Path
+
+import yaml
+from pydantic import BaseModel, Field, model_validator
+
+
+class CCParameter(BaseModel):
+    cc: int = Field(ge=0, le=127)
+    value_range: tuple[int, int] = (0, 127)
+    notes: str | None = None
+
+    @model_validator(mode="after")
+    def check_range_order(self) -> CCParameter:
+        low, high = self.value_range
+        if low > high:
+            raise ValueError(f"value_range low ({low}) must be <= high ({high})")
+        return self
+
+
+class SynthDefinition(BaseModel):
+    name: str
+    manufacturer: str
+    midi_channel: int = Field(ge=1, le=16)
+    cc_map: dict[str, CCParameter]
+
+
+_DEFAULT_SYNTHS_DIR = Path(__file__).resolve().parent.parent / "synths"
+
+
+def load_synth_definitions(synths_dir: Path = _DEFAULT_SYNTHS_DIR) -> dict[str, SynthDefinition]:
+    """Load and validate all synth YAML files from the given directory."""
+    definitions: dict[str, SynthDefinition] = {}
+    if not synths_dir.exists():
+        return definitions
+    yaml_files = sorted([*synths_dir.glob("*.yaml"), *synths_dir.glob("*.yml")])
+    for yaml_file in yaml_files:
+        data = yaml.safe_load(yaml_file.read_text())
+        synth = SynthDefinition(**data)
+        key = synth.name.lower()
+        if key in definitions:
+            raise ValueError(
+                f"Duplicate synth name '{synth.name}' "
+                f"(from {yaml_file.name} and existing definition)"
+            )
+        definitions[key] = synth
+    return definitions