Live engine: real-time MIDI-to-audio synthesis

LiveEngine listens for MIDI input and synthesizes audio in real-time. Each MIDI channel maps to a pytheory instrument with its own synth, envelope, and effects. Supports polyphony, voice stealing, and GM drum channel (10). Adds python-rtmidi as a dependency. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-06-05 23:00:20 +00:00 · 2026-03-29 18:29:51 -04:00
parent 417d9a6908
commit bc652c37d0
3 changed files with 372 additions and 0 deletions
@@ -23,6 +23,7 @@ classifiers = [
 dependencies = [
    "sounddevice",
    "scipy",
+    "python-rtmidi>=1.5.8",
 ]

 [project.urls]
@@ -0,0 +1,346 @@
+"""Real-time MIDI-driven synthesis engine.
+
+Listens for MIDI input (e.g. from an OP-XY, keyboard, or DAW) and
+synthesizes audio in real-time through pytheory's synth engine.
+
+Usage::
+
+    from pytheory.live import LiveEngine
+
+    engine = LiveEngine()
+    engine.channel(1, instrument="electric_piano")
+    engine.channel(2, instrument="bass_guitar", lowpass=800)
+    engine.channel(10, drums=True)
+    engine.start()  # blocks until Ctrl-C
+"""
+
+import threading
+import numpy
+import sounddevice as sd
+import rtmidi
+
+from .play import (
+    _SYNTH_FUNCTIONS, _resolve_synth, _resolve_envelope,
+    _apply_envelope, _apply_lowpass, _render_drum_hit_cached,
+    SAMPLE_RATE, SAMPLE_PEAK,
+)
+from .rhythm import INSTRUMENTS, DrumSound
+
+
+# ── Voice ────────────────────────────────────────────────────────────────
+
+class _Voice:
+    """A single sounding note — holds a pre-rendered wavetable and
+    tracks playback position + envelope state."""
+    __slots__ = ('wave', 'pos', 'velocity', 'active', 'releasing',
+                 'release_pos', 'release_len', 'note')
+
+    def __init__(self, wave, velocity, note):
+        self.wave = wave           # float32 array — one shot or looped
+        self.pos = 0               # current read position
+        self.velocity = velocity
+        self.active = True
+        self.releasing = False
+        self.release_pos = 0
+        self.release_len = int(SAMPLE_RATE * 0.05)  # 50ms release
+        self.note = note           # MIDI note number
+
+
+# ── Channel ──────────────────────────────────────────────────────────────
+
+class _Channel:
+    """One MIDI channel — has a synth, effects, and a voice pool."""
+
+    def __init__(self, synth_name="sine", envelope_name="piano",
+                 is_drums=False, max_voices=12, **kwargs):
+        self.synth_fn = _resolve_synth(synth_name)
+        self.synth_name = synth_name
+        self.envelope_name = envelope_name
+        self.env_tuple = _resolve_envelope(envelope_name)
+        self.is_drums = is_drums
+        self.max_voices = max_voices
+        self.kwargs = kwargs
+        self.lowpass = kwargs.get('lowpass', 0)
+        self.lowpass_q = kwargs.get('lowpass_q', 0.707)
+        self.reverb = kwargs.get('reverb', 0)
+        self.volume = kwargs.get('volume', 0.5)
+
+        self.voices = []           # active _Voice objects
+        self._cache = {}           # MIDI note → pre-rendered wave
+        self._lock = threading.Lock()
+
+    def _get_wave(self, midi_note, n_samples):
+        """Get or render a waveform for a MIDI note."""
+        if self.is_drums:
+            return _render_drum_hit_cached(midi_note, n_samples)
+
+        if midi_note in self._cache:
+            cached = self._cache[midi_note]
+            if len(cached) >= n_samples:
+                return cached[:n_samples]
+
+        hz = 440.0 * (2 ** ((midi_note - 69) / 12.0))
+
+        # Synth kwargs
+        skw = {}
+        if self.synth_name in ("fm",):
+            skw["mod_ratio"] = self.kwargs.get("fm_ratio", 2.0)
+            skw["mod_index"] = self.kwargs.get("fm_index", 3.0)
+
+        wave = self.synth_fn(hz, n_samples=n_samples, **skw)
+        wave_f = wave.astype(numpy.float32) / SAMPLE_PEAK
+
+        # Apply envelope
+        a, d, s, r = self.env_tuple
+        if a > 0 or d > 0 or s < 1.0 or r > 0:
+            wave_f = _apply_envelope(wave_f, a, d, s, r)
+
+        # Apply lowpass
+        if self.lowpass > 0:
+            wave_f = _apply_lowpass(wave_f, self.lowpass, q=self.lowpass_q)
+
+        self._cache[midi_note] = wave_f
+        return wave_f
+
+    def note_on(self, midi_note, velocity):
+        """Start a new voice."""
+        vel_scale = velocity / 127.0
+        # Render 2 seconds of audio
+        n_samples = SAMPLE_RATE * 2
+        wave = self._get_wave(midi_note, n_samples)
+
+        with self._lock:
+            # Voice stealing — kill oldest if at max
+            if len(self.voices) >= self.max_voices:
+                self.voices.pop(0)
+            self.voices.append(_Voice(wave, vel_scale, midi_note))
+
+    def note_off(self, midi_note):
+        """Trigger release on voices playing this note."""
+        with self._lock:
+            for v in self.voices:
+                if v.note == midi_note and v.active and not v.releasing:
+                    v.releasing = True
+                    v.release_pos = 0
+
+    def render(self, n_frames):
+        """Mix all active voices into a buffer."""
+        buf = numpy.zeros(n_frames, dtype=numpy.float32)
+        dead = []
+
+        with self._lock:
+            for i, v in enumerate(self.voices):
+                if not v.active:
+                    dead.append(i)
+                    continue
+
+                remaining = len(v.wave) - v.pos
+                chunk = min(n_frames, remaining)
+
+                if chunk <= 0:
+                    v.active = False
+                    dead.append(i)
+                    continue
+
+                samples = v.wave[v.pos:v.pos + chunk] * v.velocity * self.volume
+
+                # Release fade
+                if v.releasing:
+                    fade_chunk = min(chunk, v.release_len - v.release_pos)
+                    if fade_chunk > 0:
+                        fade = numpy.linspace(
+                            1.0 - v.release_pos / v.release_len,
+                            1.0 - (v.release_pos + fade_chunk) / v.release_len,
+                            fade_chunk
+                        ).astype(numpy.float32)
+                        samples[:fade_chunk] *= fade
+                        v.release_pos += fade_chunk
+                    if v.release_pos >= v.release_len:
+                        v.active = False
+                        samples[fade_chunk:] = 0
+
+                buf[:chunk] += samples
+                v.pos += chunk
+
+            # Clean up dead voices
+            for i in reversed(dead):
+                if i < len(self.voices):
+                    self.voices.pop(i)
+
+        return buf
+
+
+# ── LiveEngine ───────────────────────────────────────────────────────────
+
+class LiveEngine:
+    """Real-time MIDI-to-audio engine.
+
+    Maps MIDI channels to pytheory instruments and synthesizes
+    audio in real-time via sounddevice.
+
+    Example::
+
+        engine = LiveEngine()
+        engine.channel(1, instrument="electric_piano")
+        engine.channel(2, instrument="bass_guitar")
+        engine.channel(10, drums=True)
+        engine.start()
+    """
+
+    def __init__(self, buffer_size=512, sample_rate=SAMPLE_RATE):
+        self.buffer_size = buffer_size
+        self.sample_rate = sample_rate
+        self.channels = {}   # MIDI channel (1-16) → _Channel
+        self._midi_in = None
+        self._stream = None
+
+    def channel(self, ch, *, instrument=None, synth=None, envelope=None,
+                drums=False, **kwargs):
+        """Configure a MIDI channel.
+
+        Args:
+            ch: MIDI channel number (1-16). Channel 10 = drums by convention.
+            instrument: Instrument preset name (e.g. "electric_piano").
+            synth: Synth waveform name (overrides instrument).
+            envelope: Envelope name (overrides instrument).
+            drums: If True, this channel triggers drum sounds by MIDI note.
+            **kwargs: Any Part parameter (lowpass, reverb, volume, etc.)
+        """
+        # Build params from instrument preset
+        params = {}
+        if instrument:
+            preset = INSTRUMENTS.get(instrument)
+            if preset:
+                params.update(preset)
+        if synth:
+            params["synth"] = synth
+        if envelope:
+            params["envelope"] = envelope
+        params.update(kwargs)
+
+        synth_name = params.pop("synth", "sine")
+        env_name = params.pop("envelope", "piano")
+
+        self.channels[ch] = _Channel(
+            synth_name=synth_name,
+            envelope_name=env_name,
+            is_drums=drums or ch == 10,
+            **params,
+        )
+        return self
+
+    def _midi_callback(self, event, data=None):
+        """Handle incoming MIDI messages."""
+        msg, _ = event
+        if len(msg) < 3:
+            return
+
+        status = msg[0]
+        ch = (status & 0x0F) + 1  # MIDI channels are 0-indexed in protocol
+        msg_type = status & 0xF0
+
+        if ch not in self.channels:
+            return
+
+        channel = self.channels[ch]
+        note = msg[1]
+        velocity = msg[2]
+
+        if msg_type == 0x90 and velocity > 0:
+            channel.note_on(note, velocity)
+        elif msg_type == 0x80 or (msg_type == 0x90 and velocity == 0):
+            channel.note_off(note)
+
+    def _audio_callback(self, outdata, frames, time_info, status):
+        """sounddevice callback — mix all channels."""
+        buf = numpy.zeros(frames, dtype=numpy.float32)
+        for channel in self.channels.values():
+            buf += channel.render(frames)
+
+        # Soft clip
+        buf = numpy.tanh(buf)
+
+        # Stereo output
+        outdata[:, 0] = buf
+        outdata[:, 1] = buf
+
+    def list_ports(self):
+        """List available MIDI input ports."""
+        midi_in = rtmidi.MidiIn()
+        ports = midi_in.get_ports()
+        for i, name in enumerate(ports):
+            print(f"  {i}: {name}")
+        midi_in.delete()
+        return ports
+
+    def start(self, port=None):
+        """Start the engine — opens MIDI input and audio output.
+
+        Args:
+            port: MIDI port index or name. None = first available.
+
+        Blocks until Ctrl-C.
+        """
+        if not self.channels:
+            # Default: Rhodes on channel 1
+            self.channel(1, instrument="electric_piano")
+
+        # Open MIDI
+        self._midi_in = rtmidi.MidiIn()
+        ports = self._midi_in.get_ports()
+
+        if not ports:
+            print("  No MIDI input ports found.")
+            print("  Connect a MIDI device and try again.")
+            return
+
+        if port is None:
+            port = 0
+        elif isinstance(port, str):
+            for i, name in enumerate(ports):
+                if port.lower() in name.lower():
+                    port = i
+                    break
+
+        self._midi_in.open_port(port)
+        self._midi_in.set_callback(self._midi_callback)
+        port_name = ports[port] if isinstance(port, int) else port
+
+        print(f"  PyTheory Live Engine")
+        print(f"  MIDI: {port_name}")
+        print(f"  Buffer: {self.buffer_size} samples ({self.buffer_size/self.sample_rate*1000:.1f}ms)")
+        print(f"  Channels:")
+        for ch, channel in sorted(self.channels.items()):
+            kind = "drums" if channel.is_drums else channel.synth_name
+            print(f"    {ch:2d}: {kind} (vol={channel.volume})")
+        print()
+        print("  Playing... (Ctrl-C to stop)")
+        print()
+
+        # Open audio
+        self._stream = sd.OutputStream(
+            samplerate=self.sample_rate,
+            blocksize=self.buffer_size,
+            channels=2,
+            dtype='float32',
+            callback=self._audio_callback,
+        )
+
+        try:
+            self._stream.start()
+            # Block forever
+            threading.Event().wait()
+        except KeyboardInterrupt:
+            print("\n  Stopped.")
+        finally:
+            self._stream.stop()
+            self._stream.close()
+            self._midi_in.close_port()
+            self._midi_in.delete()
+
+    def stop(self):
+        """Stop the engine."""
+        if self._stream:
+            self._stream.stop()
+        if self._midi_in:
+            self._midi_in.close_port()
@@ -693,6 +693,7 @@ name = "pytheory"
 version = "0.40.0"
 source = { editable = "." }
 dependencies = [
+    { name = "python-rtmidi" },
    { name = "scipy", version = "1.15.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
    { name = "scipy", version = "1.17.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
    { name = "sounddevice" },
@@ -712,6 +713,7 @@ docs = [

 [package.metadata]
 requires-dist = [
+    { name = "python-rtmidi", specifier = ">=1.5.8" },
    { name = "scipy" },
    { name = "sounddevice" },
 ]
@@ -723,6 +725,29 @@ docs = [
    { name = "sphinx" },
 ]

+[[package]]
+name = "python-rtmidi"
+version = "1.5.8"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/dd/ee/0f91965dcc471714c69df21e5ca3d94dc81411b7dee2d31ff1184bea07c9/python_rtmidi-1.5.8.tar.gz", hash = "sha256:7f9ade68b068ae09000ecb562ae9521da3a234361ad5449e83fc734544d004fa", size = 368130, upload-time = "2023-11-20T21:55:02.192Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/46/9c/95c0a6a43bd24a17568e1e31008b1fab7e9a2e54c0ed7301e8d5cc9fa109/python_rtmidi-1.5.8-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:efc07413b30b0039c0d35abe25a81d740c7405124eb58eed141a8f24388e6fe0", size = 148826, upload-time = "2023-11-20T21:54:20.658Z" },
+    { url = "https://files.pythonhosted.org/packages/bf/9b/8e452d6edc2c04e3407f542d3185c66ffc2d39c8811cf2b117653a0a4d63/python_rtmidi-1.5.8-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:844bd12840c9d4e03dfc89b2cd57c55dcbf5ed7246504d69c6c661732249b19c", size = 145363, upload-time = "2023-11-20T21:54:23.023Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/48/aa1d4924f7aa238a192d69aa565b315af0037f684c9475e8b860c679a655/python_rtmidi-1.5.8-cp310-cp310-manylinux_2_28_aarch64.whl", hash = "sha256:8bbaf7c7164471712a93ac60c8f9ed146b336a294a5103223bbaf8f10709a0bf", size = 293253, upload-time = "2023-11-20T21:54:24.899Z" },
+    { url = "https://files.pythonhosted.org/packages/bf/24/32dc239047a56f44d8d8090d55010f85a38ed959ffe517c2e87a2aa34190/python_rtmidi-1.5.8-cp310-cp310-manylinux_2_28_x86_64.whl", hash = "sha256:878ce085dfb65c0974810a7e919f73708cbb4c0430c7924b78f25aea1dd4ebee", size = 304051, upload-time = "2023-11-20T21:54:26.82Z" },
+    { url = "https://files.pythonhosted.org/packages/9b/0c/cf771eca1b64610e627ca1e67be8390ecdf5e0e1914efbdd9d50ac4c5986/python_rtmidi-1.5.8-cp310-cp310-win_amd64.whl", hash = "sha256:f2138005c6bd3d8b9af05df383679f6d0827d16056e68a941110732310dcb7dd", size = 132157, upload-time = "2023-11-20T21:54:30.059Z" },
+    { url = "https://files.pythonhosted.org/packages/4b/0c/23be16b75c90946784b8d233e61db14cf0482def5396821a1ae0bdcd2739/python_rtmidi-1.5.8-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:30d117193dcad8af67c600c405f53eb096e4ff84849760be14c97270af334922", size = 150205, upload-time = "2023-11-20T21:54:31.849Z" },
+    { url = "https://files.pythonhosted.org/packages/4b/12/37d41151b08a292719f05dbeae15475537f8aa291cda34c6634b35916dff/python_rtmidi-1.5.8-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4e234dca7f9d783dd3f1e9c9c5c2f295f02b7af3085301d6eed3b428cf49d327", size = 146737, upload-time = "2023-11-20T21:54:33.299Z" },
+    { url = "https://files.pythonhosted.org/packages/53/5e/b866491545135c699bfbed62f54b93c4d6587afc2bba6e2cbbe898570c32/python_rtmidi-1.5.8-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:271d625c489fffb39b3edc5aba67f7c8e29a04a0a0f056ce19e5a888a08b4c59", size = 294847, upload-time = "2023-11-20T21:54:35.017Z" },
+    { url = "https://files.pythonhosted.org/packages/a0/79/1ddb4fb1bdb1a8b8bd62007ca4980344a53b7f29633a7bca1088eed964ce/python_rtmidi-1.5.8-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:46bbf32c8a4bf6c8f0df1c02a68689d0757f13cb7a69f27ccbbed3d7b2365918", size = 305433, upload-time = "2023-11-20T21:54:36.322Z" },
+    { url = "https://files.pythonhosted.org/packages/13/ff/2c55797dbf020d462132d1bc5b34d596b400fa197e2a259b8dd2ea2e5954/python_rtmidi-1.5.8-cp311-cp311-win_amd64.whl", hash = "sha256:cfea32c91752fa7aecfe3d6827535c190ba0e646a9accd6604f4fc70cf4b780f", size = 132937, upload-time = "2023-11-20T21:54:38.3Z" },
+    { url = "https://files.pythonhosted.org/packages/51/27/887b0378e0a907489a07bdeb808fa5ed349675245c6ee14d9f6d00304f96/python_rtmidi-1.5.8-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:5443634597eb340cdec0734f76267a827c2d366f00a6f9195141c78828016ac2", size = 158861, upload-time = "2023-11-20T21:54:39.549Z" },
+    { url = "https://files.pythonhosted.org/packages/4d/ec/57cecde253daab896ce53778520cd41eb062641862ebdb0ee6f97511b1d9/python_rtmidi-1.5.8-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:29d9c9d9f82ce679fecad7bb4cb79f3a24574ea84600e377194b4cc1baacec0e", size = 153416, upload-time = "2023-11-20T21:54:40.835Z" },
+    { url = "https://files.pythonhosted.org/packages/6f/5b/dc19c53d9d512b74dc2cca3725591cc612b9465645695a0696352a8c8b54/python_rtmidi-1.5.8-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:25f5a5db7be98911c41ca5bebb262fcf9a7c89600b88fd3c207ceafd3101e721", size = 305696, upload-time = "2023-11-20T21:54:42.037Z" },
+    { url = "https://files.pythonhosted.org/packages/f6/92/5a60f56dfb2740e644e932233928947423cd2101895319b331f84527eb31/python_rtmidi-1.5.8-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:cec30924e305f55284594ccf35a71dee7216fd308dfa2dec1b3ed03e6f243803", size = 315579, upload-time = "2023-11-20T21:54:43.339Z" },
+    { url = "https://files.pythonhosted.org/packages/93/46/6af077d262f521ea2bf1ab60b8aad72f34fe6dd55af739176605369d449c/python_rtmidi-1.5.8-cp312-cp312-win_amd64.whl", hash = "sha256:052c89933cae4fca354012d8ca7248f4f9e1e3f062471409d48415a7f7d7e59e", size = 129755, upload-time = "2023-11-20T21:54:44.935Z" },
+]
+
 [[package]]
 name = "pyyaml"
 version = "6.0.3"