How to build a synthesizer with python: part 4
Synthesizer
mike | July 17, 2023, 2:58 a.m.
In this tutorial, we'll use the modules we built in the preceding parts to build a Synthesizer class. In part 1 we built a stream player to play audio out loud. In part 2 we wrote the core synthesis code, giving our stream player something to play. In part 3 we gave ourselves a way to control the signal components with MIDI. By the end of this part, you will have a 4 voice, 2 oscillator-per-voice poly synth. We'll be able to adjust the oscillator mix via MIDI CC messages as well.
Putting the pieces together
Remember what our minimal synthesizer consists of?
- A way to generate audio
- A way to play audio out loud
- A way to control the audio
Now we've built all 3 pieces, let's put them together. The goal here is to build a 4 voice synthesizer, so let's think about what we'll need to do that. Recall how we constructed a single voice, or "mono", synth by arranging our signal components in a tree structure and manipulating the components. You can think of that tree as comprising a single voice. Since we want our synth to have 4 voices, it makes sense to encapsulate this data structure and functionality into a class, which we'll call Voice.
In addition to the Voice class, we'll make a supporting class called Chain which will act as an interface to our tree of signal components. Each instance of Voice will have a signal.Chain instance. The Chain class's job is to provide access to the functionality of the signal components without exposing their structure.
And lastly, we'll need the Synthesizer class itself. If everything we've built so far are the bread, fillings, and condiments, this class is the sandwich. Calling run on an instance of this class is analogous to flipping the power switch of a physical synth to the On position.
A heading about chains
Let's build from the bottom up. Create a file in the signal submodule called chain.py. Add the following imports:
import logging
from copy import deepcopy
from .component import Component
from .oscillator import Oscillator
And below that, add the class definition:
class Chain():
def __init__(self, root_component: Component):
self.log = logging.getLogger(__name__)
self._root_component = root_component
def __iter__(self):
self.root_iter = iter(self._root_component)
return self
def __next__(self):
chunk = next(self.root_iter)
return chunk
def __deepcopy__(self, memo):
return Chain(deepcopy(self._root_component, memo))
def __str__(self):
string = "--- Signal Chain ---\n"
string += str(self._root_component)
return string
@property
def active(self):
"""
The active status.
The chain is considered active when the root component is active
"""
return self._root_component.active
def get_components_by_class(self, cls):
components = []
def search_subcomponents(component):
if isinstance(component, cls):
components.append(component)
if hasattr(component, "subcomponents") and len(component.subcomponents) > 0:
for subcomponent in component.subcomponents:
search_subcomponents(subcomponent)
search_subcomponents(self._root_component)
return components
def get_components_by_control_tag(self, control_tag):
components = []
def search_subcomponents(component):
if hasattr(component, "control_tag") and component.control_tag == control_tag:
components.append(component)
if hasattr(component, "subcomponents") and len(component.subcomponents) > 0:
for subcomponent in component.subcomponents:
search_subcomponents(subcomponent)
search_subcomponents(self._root_component)
return components
def note_on(self, frequency):
for osc in self.get_components_by_class(Oscillator):
osc.frequency = frequency
self._root_component.active = True
def note_off(self):
# Setting the root component active status should propagate down the tree
self._root_component.active = False
You'll notice that we're essentially acting as a wrapper for Component types and giving the note_on and note_off methods to control them. I'll be the first to admit this section of code could be cleaned up or consolidated, but hey, sometimes you go with what works. The original idea here was to gate access to the signal component tree structure, but as you'll see later we end up accessing signal components through the Voice class anyway.
Vox
Now that we've got that out of the way, let's add a nice short class. Create a file in the synthesis submodule and call it voice.py.
from .signal.chain import Chain
class Voice:
def __init__(self, signal_chain: Chain):
self.signal_chain = iter(signal_chain)
self.note_id = None
@property
def active(self):
return self.signal_chain.active
def note_on(self, frequency, id):
self._active = True
self.note_id = id
self.signal_chain.note_on(frequency)
def note_off(self):
self.signal_chain.note_off()
As you can see, we only need a few members in our Voice class. The class encapsulates a small set of actions: turning a musical note on and off. By having multiple instances of this class, our synth can play multiple notes at the same time.
The penultimate event
The next class we want to write is the Synthesizer class. Before we dive into the code, let's review what we've built so far and how the Synthesizer class will fit in.
First, we built a stream player, which gives us a way to play audio out loud. The stream player fetches audio data chunk by chunk via a callback function. This callback function is the hook where we can attach our synthesizer functionality to the audio output stream. Our Synthesizer will attach a generator function to the stream player. The generator function mixes the output from our 4 voices and yields a bufferful of audio data. You can think of this as a continuous process. Even when the synth is silent, the generator will yield chunks full of zeros.
After the stream player, we built a set of signal components that can generate and modulate audio. Our Synthesizer class uses these components through the Voice class. When the synth starts up it creates a prototype signal chain and then makes a deep copy of it for each voice.
And finally, we built a way to control our synth. The MIDI listener interacts with the synth only indirectly, by placing messages into a mailbox. This is the main way we control our Synthesizer instance. The synth thread's main loop is to wait for mailbox messages and act on them.
The main event
Alright, let's get to the code! Create a file in the synth folder called synthesizer.py. Add the following imports:
import threading
import logging
from queue import Queue
from copy import deepcopy
import numpy as np
from . import midi
from .synthesis.voice import Voice
from .synthesis.signal.chain import Chain
from .synthesis.signal.sine_wave_oscillator import SineWaveOscillator
from .synthesis.signal.square_wave_oscillator import SquareWaveOscillator
from .synthesis.signal.gain import Gain
from .synthesis.signal.mixer import Mixer
from .playback.stream_player import StreamPlayer
Next, add the class constructor.
class Synthesizer(threading.Thread):
def __init__(self, sample_rate: int, frames_per_chunk: int, mailbox: Queue, num_voices: int=4) -> None:
super().__init__(name="Synthesizer Thread")
self.log = logging.getLogger(__name__)
self.sample_rate = sample_rate
self.frames_per_chunk = frames_per_chunk
self.mailbox = mailbox
self.num_voices = num_voices
self.should_run = True
# Set up the voices
signal_chain_prototype = self.setup_signal_chain()
self.log.info(f"Signal Chain Prototype:\n{str(signal_chain_prototype)}")
self.voices = [Voice(deepcopy(signal_chain_prototype)) for _ in range(self.num_voices)]
# Set up the stream player
self.stream_player = StreamPlayer(self.sample_rate, self.frames_per_chunk, self.generator())
The two most important sections are setting up the voices and setting up the stream player. As mentioned above, we create a signal chain prototype, then create deep copies of it for each voice. After that we instantiate a stream player and pass in the generator function which we'll define below.
For now, let's define the message handling behavior. Add the following code below the constructor:
def run(self):
self.stream_player.play()
while self.should_run and self.stream_player.is_active():
# get() is a blocking call
if message := self.mailbox.get():
self.message_handler(message)
return
def message_handler(self, message: str):
"""Handles messages from the mailbox."""
match message.split():
case ["exit"]:
self.log.info("Got exit command.")
self.stream_player.stop()
self.should_run = False
case ["note_on", "-n", note, "-c", channel]:
int_note = int(note)
chan = int(channel)
note_name = midi.note_names[int_note]
self.note_on(int_note, chan)
self.log.info(f"Note on {note_name} ({int_note}), chan {chan}")
case ["note_off", "-n", note, "-c", channel]:
int_note = int(note)
chan = int(channel)
note_name = midi.note_names[int_note]
self.note_off(int_note, chan)
self.log.info(f"Note off {note_name} ({int_note}), chan {chan}")
case ["control_change", "-c", channel, "-n", cc_num, "-v", control_val]:
chan = int(channel)
int_cc_num = int(cc_num)
int_cc_val = int(control_val)
self.control_change_handler(chan, int_cc_num, int_cc_val)
case _:
self.log.info(f"Matched unknown command: {message}")
def control_change_handler(self, channel: int, cc_number: int, val: int):
self.log.info(f"Control Change: channel {channel}, number {cc_number}, value {val}")
As you can see, the run method is quite simple for this thread. We're simply checking for mailbox messages and then sending them to a handler. So far, we're handling note_on and note_off messages, but simply printing out a log for MIDI CC messages.
Let's define the methods that will take care of the sound synthesis:
def setup_signal_chain(self) -> Chain:
"""Build the signal chain prototype."""
osc_a = SineWaveOscillator(self.sample_rate, self.frames_per_chunk)
osc_b = SquareWaveOscillator(self.sample_rate, self.frames_per_chunk)
gain_a = Gain(self.sample_rate, self.frames_per_chunk, [osc_a], control_tag="gain_a")
gain_b = Gain(self.sample_rate, self.frames_per_chunk, [osc_b], control_tag="gain_b")
mixer = Mixer(self.sample_rate, self.frames_per_chunk, [gain_a, gain_b])
signal_chain = Chain(mixer)
return signal_chain
def generator(self):
"""
Generate the signal by mixing the voice outputs
"""
mix = np.zeros(self.frames_per_chunk, np.float32)
num_active_voices = 0
while True:
for i in range(self.num_voices):
voice = self.voices[i]
mix += next(voice.signal_chain)
if voice.active:
num_active_voices += 1
# Prevent the mix from going outside the range (-1, 1)
mix = np.clip(mix, -1.0, 1.0)
yield mix
mix = np.zeros(self.frames_per_chunk, np.float32)
num_active_voices = 0
We used setup_signal_chain and generator in the class constructor. The setup_signal_chain method is where we define the structure of our signal chain. This is where we decide what oscillator types to use and what order the components are arranged in. If you built the extra sawtooth and triangle wave oscillator types in part 2, this is where you can substitute them for the sine or square types.
generator is where our synthesizer produces sound. You may notice it simply gets the next chunk from each voice and sums them together. It is essentially a simple mixer.
And finally (for now), let's add the code our message handler will use to turn notes on and off:
def note_on(self, note: int, chan: int):
"""
Set a voice on with the given note.
If there are no unused voices, drop the voice that has been on for the longest and use that voice
"""
note_id = self.get_note_id(note, chan)
freq = midi.frequencies[note]
for i in range(len(self.voices)):
voice = self.voices[i]
if not voice.active:
voice.note_on(freq, note_id)
self.voices.append(self.voices.pop(i)) # Move this voice to the back of the list. It should be popped last
break
if i == len(self.voices) - 1:
self.log.debug(f"Had no unused voices!")
self.voices[0].note_off()
self.voices[0].note_on(freq, note_id)
self.voices.append(self.voices.pop(0))
def note_off(self, note: int, chan: int):
"""
Find the voice playing the given note and turn it off.
"""
note_id = self.get_note_id(note, chan)
for i in range(len(self.voices)):
voice = self.voices[i]
if voice.active and voice.note_id == note_id:
voice.note_off()
def get_note_id(self, note: int, chan: int):
"""
Generate an id for a given note and channel
By hashing the note and channel we can ensure that we are turning off the exact note
that was turned on
"""
note_id = hash(f"{note}{chan}")
return note_id
The note_on and note_off methods have to do a bit more than the ones inside the Voice class. Since we are able to play multiple notes at the same time, we are faced with the problem of knowing which voice to use when turning on a new note. We could simply iterate through the list of voices and turn on the first unused one, but then what happens when there are no unused voices? We probably want to drop the note that has been on the longest and turn on the new note. But we need to know which note has been on for the longest in that case. By maintaining the list of voices as a FIFO stack, we can solve the problem.
Yet another test
We should be at the point where we can try out the note_on/note_off functionality. But first we need to delete some code we no longer need. Open up __main__.py and delete all of the signal components and the stream player. That's roughly the code section that looks like:
# create a sine wave oscillator
osc_a = SineWaveOscillator(settings.sample_rate, settings.frames_per_chunk)
osc_b = SquareWaveOscillator(settings.sample_rate, settings.frames_per_chunk)
gain_a = Gain(settings.sample_rate, settings.frames_per_chunk, subcomponents=[osc_a])
gain_b = Gain(settings.sample_rate, settings.frames_per_chunk, subcomponents=[osc_b])
mixer = Mixer(settings.sample_rate, settings.frames_per_chunk, subcomponents=[gain_a, gain_b])
# Create a stream player
stream_player = StreamPlayer(sample_rate=settings.sample_rate, frames_per_chunk=settings.frames_per_chunk, input_delegate=mixer)
Also delete everything inside the try block. We'll just rewrite it.
Delete all of the unused imports (everything from the synthesis and playback modules), and add the following import:
from .synthesizer import Synthesizer
And right above the try block add:
synthesizer = Synthesizer(settings.sample_rate, settings.frames_per_chunk, synth_mailbox)
Now add the following inside the try block:
midi_listener.start()
synthesizer.start()
while True:
sleep(1)
A little bit cleaner, right? Right below the except clause, delete the line
stream_player.stop()
And below listener_mailbox.put("exit") add:
synth_mailbox.put("exit")
synthesizer.join()
That was a lot
In case you didn't follow that, your __main__.py file should look like:
import logging
import queue
import sys
from optparse import OptionParser
from time import sleep
from . import settings
from .midi.midi_listener import MidiListener
from .synthesizer import Synthesizer
if __name__ == "__main__":
parser = OptionParser()
parser.add_option("-p", "--port", dest="midi_port", default=None, help="MIDI port to listen on", metavar="MIDI_PORT")
(options, args) = parser.parse_args()
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s [%(levelname)s] %(module)s [%(funcName)s]: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
log = logging.getLogger(__name__)
log.info(
"""
__
| |
__| |___ ______ __ __
/__ __/ __|______|__ | | | |
| | | | | | |__|______| |
| | __ | | | | |______ |
|__|____|__| |__|______|__| ______|__|
|____| |______| |______|
__ __
| | | |
_______ __ __ __ ____ __| |___ | | ____
__|_______| | | | | | |__|____|__ /__ __/ | |__|____|__
|__|_______ |__|______| | | __| | | | | | __| | |
|_______|__ |______ | | | | | | | __ | | | |
__________|__| ______|__| | | | | |__|____|__| | | | |
|__________| |______| |__| |__| |____| |__| |__|
"""
)
listener_mailbox = queue.Queue()
synth_mailbox = queue.Queue()
midi_listen_port = options.midi_port if options.midi_port else settings.auto_attach
log.info(f"Using MIDI port {midi_listen_port}")
midi_listener = MidiListener(listener_mailbox, synth_mailbox, midi_listen_port)
synthesizer = Synthesizer(settings.sample_rate, settings.frames_per_chunk, synth_mailbox)
try:
midi_listener.start()
synthesizer.start()
while True:
sleep(1)
except KeyboardInterrupt:
log.info("Caught keyboard interrupt. Exiting the program.")
listener_mailbox.put("exit")
synth_mailbox.put("exit")
synthesizer.join()
midi_listener.join()
sys.exit(0)
You can now run the program with python -m synth. You should be able to play up to 4 notes at once this time.
Poly graph
But how does that affect our output? Remember our plain old sine wave from before? Let's see what it looks like to play 2 notes at once. Here's an example of a minor third:
And just for good measure, here's the same minor third, but with triangle waves instead of sine.
Notice in both the same pattern of big peaks and small peaks? If you look closely, you can see that the small peak pattern is slightly different every time. This suggests some dissonance. Here's what a perfect fifth, a more consonant diad looks like:
See how the pattern is even more consistent? The interference pattern of the waves falls into a stable cycle.
Mixing it up
Okay! Right now each voice has a dual oscillator set up, with one sine and one square wave oscillator. Currently the mix is 50/50, but let's make it adjustable. This will involve parsing MIDI control change messages, which will be similar to note on/off.
Before we go any further, let's figure out which MIDI CC number we'll be using to adjust the mix. Run the program.
$ python -m synth
Now, assuming your MIDI controller has a few knobs or faders, pick the one you want to use and try moving it. You should get a log message like:
2023-06-28 21:13:20 [INFO] midi_listener [run]: Matched unknown MIDI message: control_change channel=0 control=70 value=42 time=0
See where it says control= ? Take the number that follows. That's the CC number we'll use to adjust the oscillator mix.
The next thing we need to do is give the synth a way to handle MIDI control change messages. We'll make a new file to handle MIDI implementation values. Create a file in the midi submodule called implementation.py. Add an enum:
from enum import Enum
class Implementation(Enum):
OSCILLATOR_MIX = 70
But replace 70 with the value you got when you moved your knob or fader. The values you place in this enum will roughly correspond to the MIDI implementation chart that comes with most synthesizers.
Connecting the dots
Let's go back to the synthesizer.py file. Import the enum we just created:
from .midi.implementation import Implementation
And add a new section in the constructor:
# Set up the lookup values
self.osc_mix_vals = np.linspace(0, 1, 128, endpoint=True, dtype=np.float32)
This gives us a 128 element array of values evenly spaced between 0 and 1 (the range of our Gain component). Since MIDI control change values are between 0-127, we can directly map them to the array. Each time we get a MIDI control change message, we can look up the correspeonding gain value.
In our case, we want the sum of the two oscillator gains to equal 1. By varying the ratio, we can control the oscillator mix. Add code to the control_change_handler method so it looks like this:
def control_change_handler(self, channel: int, cc_number: int, val: int):
self.log.info(f"Control Change: channel {channel}, number {cc_number}, value {val}")
if cc_number == Implementation.OSCILLATOR_MIX:
gain_a_mix_val = self.osc_mix_vals[val]
gain_b_mix_val = 1 - gain_a_mix_val
self.set_gain_a(gain_a_mix_val)
self.set_gain_b(gain_b_mix_val)
self.log.info(f"Gain A: {gain_a_mix_val}")
self.log.info(f"Gain B: {gain_b_mix_val}")
We've referenced set_gain_a and set_gain_b but we haven't implemented them yet. Go ahead and add two methods to the bottom of the file:
def set_gain_a(self, gain):
for voice in self.voices:
gain_a_components = voice.signal_chain.get_components_by_control_tag("gain_a")
for gain_a in gain_a_components:
gain_a.amp = gain
def set_gain_b(self, gain):
for voice in self.voices:
gain_b_components = voice.signal_chain.get_components_by_control_tag("gain_b")
for gain_b in gain_b_components:
gain_b.amp = gain
And that's all we need! We've now created a basic poly synth with adjustable oscillator mix. As always, run the program with python -m synth. If you move the control you set up while playing some notes, you should be able to hear the texture of the sound change as the wave shifts from sine to square shape.
The synthesizer.py file should look like:
import threading
import logging
from queue import Queue
from copy import deepcopy
import numpy as np
from . import midi
from .midi.implementation import Implementation
from .synthesis.voice import Voice
from .synthesis.signal.chain import Chain
from .synthesis.signal.sine_wave_oscillator import SineWaveOscillator
from .synthesis.signal.square_wave_oscillator import SquareWaveOscillator
from .synthesis.signal.gain import Gain
from .synthesis.signal.mixer import Mixer
from .playback.stream_player import StreamPlayer
class Synthesizer(threading.Thread):
def __init__(self, sample_rate: int, frames_per_chunk: int, mailbox: Queue, num_voices: int=4) -> None:
super().__init__(name="Synthesizer Thread")
self.log = logging.getLogger(__name__)
self.sample_rate = sample_rate
self.frames_per_chunk = frames_per_chunk
self.mailbox = mailbox
self.num_voices = num_voices
self.should_run = True
# Set up the voices
signal_chain_prototype = self.setup_signal_chain()
self.log.info(f"Signal Chain Prototype:\n{str(signal_chain_prototype)}")
self.voices = [Voice(deepcopy(signal_chain_prototype)) for _ in range(self.num_voices)]
# Set up the stream player
self.stream_player = StreamPlayer(self.sample_rate, self.frames_per_chunk, self.generator())
# Set up the lookup values
self.osc_mix_vals = np.linspace(0, 1, 128, endpoint=True, dtype=np.float32)
def run(self):
self.stream_player.play()
while self.should_run and self.stream_player.is_active():
# get() is a blocking call
if message := self.mailbox.get():
self.message_handler(message)
return
def message_handler(self, message: str):
"""Handles messages from the mailbox."""
match message.split():
case ["exit"]:
self.log.info("Got exit command.")
self.stream_player.stop()
self.should_run = False
case ["note_on", "-n", note, "-c", channel]:
int_note = int(note)
chan = int(channel)
note_name = midi.note_names[int_note]
self.note_on(int_note, chan)
self.log.info(f"Note on {note_name} ({int_note}), chan {chan}")
case ["note_off", "-n", note, "-c", channel]:
int_note = int(note)
chan = int(channel)
note_name = midi.note_names[int_note]
self.note_off(int_note, chan)
self.log.info(f"Note off {note_name} ({int_note}), chan {chan}")
case ["control_change", "-c", channel, "-n", cc_num, "-v", control_val]:
chan = int(channel)
int_cc_num = int(cc_num)
int_cc_val = int(control_val)
self.control_change_handler(chan, int_cc_num, int_cc_val)
case _:
self.log.info(f"Matched unknown command: {message}")
def control_change_handler(self, channel: int, cc_number: int, val: int):
self.log.info(f"Control Change: channel {channel}, number {cc_number}, value {val}")
if cc_number == Implementation.OSCILLATOR_MIX.value:
gain_b_mix_val = self.osc_mix_vals[val]
gain_a_mix_val = 1 - gain_b_mix_val
self.set_gain_a(gain_a_mix_val)
self.set_gain_b(gain_b_mix_val)
self.log.info(f"Gain A: {gain_a_mix_val}")
self.log.info(f"Gain B: {gain_b_mix_val}")
def setup_signal_chain(self) -> Chain:
"""Build the signal chain prototype."""
osc_a = SineWaveOscillator(self.sample_rate, self.frames_per_chunk)
osc_b = SquareWaveOscillator(self.sample_rate, self.frames_per_chunk)
gain_a = Gain(self.sample_rate, self.frames_per_chunk, [osc_a], control_tag="gain_a")
gain_b = Gain(self.sample_rate, self.frames_per_chunk, [osc_b], control_tag="gain_b")
mixer = Mixer(self.sample_rate, self.frames_per_chunk, [gain_a, gain_b])
signal_chain = Chain(mixer)
return signal_chain
def generator(self):
"""
Generate the signal by mixing the voice outputs
"""
mix = np.zeros(self.frames_per_chunk, np.float32)
num_active_voices = 0
while True:
for i in range(self.num_voices):
voice = self.voices[i]
mix += next(voice.signal_chain)
if voice.active:
num_active_voices += 1
# Prevent the mix from going outside the range (-1, 1)
mix = np.clip(mix, -1.0, 1.0)
yield mix
mix = np.zeros(self.frames_per_chunk, np.float32)
num_active_voices = 0
def note_on(self, note: int, chan: int):
"""
Set a voice on with the given note.
If there are no unused voices, drop the voice that has been on for the longest and use that voice
"""
note_id = self.get_note_id(note, chan)
freq = midi.frequencies[note]
for i in range(len(self.voices)):
voice = self.voices[i]
if not voice.active:
voice.note_on(freq, note_id)
self.voices.append(self.voices.pop(i)) # Move this voice to the back of the list. It should be popped last
break
if i == len(self.voices) - 1:
self.log.debug(f"Had no unused voices!")
self.voices[0].note_off()
self.voices[0].note_on(freq, note_id)
self.voices.append(self.voices.pop(0))
def note_off(self, note: int, chan: int):
"""
Find the voice playing the given note and turn it off.
"""
note_id = self.get_note_id(note, chan)
for i in range(len(self.voices)):
voice = self.voices[i]
if voice.active and voice.note_id == note_id:
voice.note_off()
def get_note_id(self, note: int, chan: int):
"""
Generate an id for a given note and channel
By hashing the note and channel we can ensure that we are turning off the exact note
that was turned on
"""
note_id = hash(f"{note}{chan}")
return note_id
def set_gain_a(self, gain):
for voice in self.voices:
gain_a_components = voice.signal_chain.get_components_by_control_tag("gain_a")
for gain_a in gain_a_components:
gain_a.amp = gain
def set_gain_b(self, gain):
for voice in self.voices:
gain_b_components = voice.signal_chain.get_components_by_control_tag("gain_b")
for gain_b in gain_b_components:
gain_b.amp = gain
Conclusion
In this part we put the pieces together. We connected our stream player, signal components, and MIDI listener to create a Synthesizer class. By now, we've achieved the main goal we set out to accomplish: we built a proof-of-concept poly synth with Python. Hopefully you've enjoyed building it! In the next part, we'll build a few additional FX components and go over some suggestions for taking the project further. As always, thanks for reading and happy coding!