Generating sounds live.

[botoxparty]

Hi,

I'm trying to run Surge with the python bindings in a way that it can generate sounds Live (whereas the existing demos show it creating waveforms to be saved or viewed in a plot.)

This is what I have so far but there's clearly something wrong with my code. Anyone able to test this locally and is more familiar with audio processing? Would love to get this working and add this to the examples.

import surgepy
import sounddevice as sd
import threading
import time

class SoundGenerator:
    def __init__(self, sample_rate=44100, buffer_size=512):
        # Initialize the synth
        self.synth = surgepy.createSurge(sample_rate)
        self.sample_rate = self.synth.getSampleRate()
        self.buffer_size = buffer_size

        # Create buffer
        twosecondsInBlocks = int(2 * self.sample_rate / self.synth.getBlockSize())
        self.buf = self.synth.createMultiBlock(twosecondsInBlocks)

        # Set up sounddevice
        sd.default.samplerate = self.sample_rate
        sd.play(self.buf[0], self.sample_rate)

        # Calculate the timer interval
        self.timer_interval = self.buffer_size / self.sample_rate

        # Start a separate thread for the audio processing loop
        self.processing_thread = threading.Thread(target=self.process_audio_loop)
        self.processing_thread.daemon = True
        self.processing_thread.start()

        self.synth.playNote(0, 60, 127, 0)

    def process_audio_loop(self):
        while True:
            start_time = time.time()
            self.synth.processMultiBlock(self.buf)  # Process the audio block
            elapsed_time = time.time() - start_time
            sleep_time = self.timer_interval - elapsed_time
            if sleep_time > 0:
                time.sleep(sleep_time)  # Sleep for the remaining time in the interval
            else:
                print("Warning: Processing is taking longer than the interval!")

SoundGenerator()

while True:
    # Your loop code here
    print("Looping...")

[baconpaul]

So I haven’t done this with python but the trick really is to generate 32 sample blocks by calling g surge process in your audio callback and have a separate midi thread which pushes the messages over on a lock free structure

you can see how the cli does it here

https://github.com/surge-synthesizer/surge/blob/6c59e385daed31f8118fa312133340a856893b7d/src/surge-xt/cli/cli-main.cpp#L176

and you want to basicalky do “the same”

[botoxparty]

It sounds a little funny on my hardware, I think it might be my audio codec so will try and test another setup.

If you have surgepy running locally could you test this python script on your end?

God bless ChatGPT:

import surgepy
import sounddevice as sd
import threading
import time
import numpy as np

class Synth:
    def __init__(self, sample_rate=48000, buffer_size=512):
        # Initialize the synth processor
        self.synth = surgepy.createSurge(sample_rate)
        self.sample_rate = int(self.synth.getSampleRate())  # Ensure sample_rate is an integer
        self.buffer_size = int(buffer_size)  # Ensure buffer_size is an integer

        # Create buffers
        self.block_size = int(self.synth.getBlockSize())  # Ensure block_size is an integer
        self.num_blocks = 2 * (self.sample_rate // self.block_size)  # Ensure num_blocks is an integer
        self.audio_buffer = np.zeros((self.num_blocks, 2, self.block_size), dtype=np.float32)

        self.midi_buffer = []

        # Set up sounddevice
        sd.default.samplerate = self.sample_rate
        sd.default.blocksize = self.block_size

    def play_note(self, channel, pitch, velocity, time):
        self.synth.playNote(channel, pitch, velocity, time)

    def set_parameter(self, param_name, value):
        self.synth.setParameter(param_name, value)

    def load_preset(self, preset_name):
        # Assuming there is a method to load a preset
        self.synth.loadPreset(preset_name)

    def process_audio(self, outdata, frames, time, status):
        # Handles audio processing similar to audioDeviceIOCallbackWithContext

        if self.midi_buffer:
            for message in self.midi_buffer:
                self.synth.applyMidi(message)
            self.midi_buffer.clear()

        self.synth.process()
       # Retrieve the processed audio from the internal buffer
        output = self.synth.getOutput()

        # Fill the output buffer
        outdata[:, 0] = output[0, :frames]  # Left channel
        outdata[:, 1] = output[1, :frames]  # Right channel

    def start_audio(self):
        # Start the audio stream
        self.stream = sd.OutputStream(callback=self.process_audio)
        self.stream.start()

    def stop_audio(self):
        # Stop the audio stream
        self.stream.stop()
        self.stream.close()

    def handle_midi_message(self, message):
        # Add the incoming MIDI message to the buffer
        self.midi_buffer.append(message)

# Example usage in a SynthController class:
class SynthController:
    def __init__(self):
        # Initialize the synth
        self.synth = Synth()

        # Start the audio processing loop
        self.synth.start_audio()

        # Example note
        self.synth.play_note(0, 60, 127, 0)

    def on_slider_value_change(self, value):
        self.synth.set_parameter('filter_cutoff', value / 127.0)

    def on_button_press(self):
        self.synth.load_preset('Preset Name')

    def stop(self):
        # Stop the synth audio processing when done
        self.synth.stop_audio()

if __name__ == "__main__":
    controller = SynthController()
    try:
        # Keep the program running to allow audio processing
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        controller.stop()

[baconpaul]

just looking, if process_audio gets called with frames != 32 it will definitely be wrong. You need to actually loop and populate since the audio card will ask for an arbitray number of samples and surge makes them 32 at a time which is why the CLI has blockpos counter etc

[botoxparty]

In this version, we set the block_size that the audio card will be requesting so frames is always == 32.

It still has the same issue with the sound, when playing a sine wave theres some kind of noise at a high pitch. Have verified this on multiple machines with different hardware now. (x86 and arm)

import surgepy
import sounddevice as sd
import numpy as np
import os

class Synth:
    def __init__(self, sample_rate=48000, buffer_size=512):
        # Initialize the synth processor
        self.synth = surgepy.createSurge(sample_rate)
        self.sample_rate = int(self.synth.getSampleRate())
        self.buffer_size = int(buffer_size)
        self.patch_name = ""

        # Create buffers
        self.block_size = int(self.synth.getBlockSize())
        self.audio_buffer = np.zeros((2, self.block_size), dtype=np.float32)  # 2 for stereo output

        # Position tracking within the block
        self.pos = self.block_size

        # Set up sounddevice
        sd.default.samplerate = self.sample_rate
        sd.default.blocksize = self.block_size

    def process_audio(self, outdata, frames, time, status):
        self.synth.process()
       # Retrieve the processed audio from the internal buffer
        output = self.synth.getOutput()

        # Fill the output buffer
        outdata[:, 0] = output[0, :frames]  # Left channel
        outdata[:, 1] = output[1, :frames]  # Right channel

    def start_audio(self):
        # Start the audio stream
        self.stream = sd.OutputStream(callback=self.process_audio, blocksize=self.block_size)
        self.stream.start()

    def stop_audio(self):
        # Stop the audio stream
        self.stream.stop()
        self.stream.close()

    def play_note(self, channel, pitch, velocity, time):
        self.synth.playNote(channel, pitch, velocity, time)

[botoxparty]

Ok so i asked chatgpt to use processmultiblock and that's fixed the audio issue.

  def process_audio(self, outdata, frames, time, status):
        # Calculate the number of blocks needed
        blocks_needed = (frames + self.block_size - 1) // self.block_size

        # Ensure the internal buffer is large enough
        output_buffer = np.zeros((2, blocks_needed * self.block_size), dtype=np.float32)

        # Process the required number of blocks
        self.synth.processMultiBlock(output_buffer, 0, blocks_needed)

        # Fill the output buffer with the processed audio
        outdata[:, 0] = output_buffer[0, :frames]  # Left channel
        outdata[:, 1] = output_buffer[1, :frames]  # Right channel

Should i add a basic example to the docs?

[baconpaul]

That would be great sure! Thanks!