mike | July 14, 2023, 1:39 a.m.
If you're keeping track, we've now built 2 of 3 pieces of our minimal synthesizer. In part 1 we gave ourselves a way to play audio out loud. In part 2 we created a set of signal components capable of generating audio. In this part we'll give ourselves a way to control the sound. By the end of this tutorial you should have basic mono synth functionality. That is, you should be able to hook up a MIDI keyboard and play the notes one at a time.
This is the point at which you'll need a MIDI controller to continue using the program. If you don't have a MIDI controller, there are other ways to send MIDI messages to our synthesizer. The mido library, which we'll be using to receive MIDI messages, is also capable of sending them, for example.
To control our synthesizer, we'll implement a MIDI listener. The listener's job will be to listen for MIDI input messages and translate them into synthesizer commands. For a refresher on MIDI basics as they pertain to this synth, check out part 0.
Since listening for MIDI messages is a blocking operation, we'll need to run the listener in a separate thread. To do that we'll implement a class that extends threading.Thread
. The Thread class has a simple interface. We basically just need to implement a run
method that we can call to start the thread. In the run
method we'll write a loop that checks for MIDI messages and takes action on them. We can visualize how our MIDI listener thread works with a flowchart.
In addition to opening a MIDI port, our listener thread will also have two mailboxes: one for receiving messages from the main thread, and one for sending messages to the synthesizer. The only message the listener thread ever receives from the main thread is "exit"
. This tells the listener thread to exit its main loop and return because the user has initiated the shutdown process. The mailboxes are simply instances of the standard library Queue
class. The Queue
class is thread safe, so we can send and receive messages without worrying about synchronization issues between threads.
We'll use the mido library for MIDI communication. And since we want to use ports, we'll install a backend for mido as well. At your shell prompt, run:
$ pip install mido
$ pip install python-rtmidi
$ pip freeze > requirements.txt
Next, create a new submodule under the synth module called midi
. (Create a folder called midi
and put an __init__.py
in it.) We're actually going to put some code in the __init__.py
this time. Add the following:
import mido
def get_available_controllers():
return mido.get_input_names()
frequencies = [
8.176,8.662,9.177,9.723,
10.301,10.913,11.562,12.250,
12.978,13.750,14.568,15.434,
16.352,17.324,18.354,19.445,
20.601,21.826,23.124,24.499,
25.956,27.500,29.135,30.867,
32.703,34.648,36.708,38.890,
41.203,43.653,46.249,48.999,
51.913,55.000,58.270,61.735,
65.406,69.295,73.416,77.781,
82.406,87.307,92.499,97.998,
103.82,110.00,116.54,123.47,
130.81,138.59,146.83,155.56,
164.81,174.61,184.99,195.99,
207.65,220.00,233.08,246.94,
261.63,277.18,293.66,311.13,
329.63,349.23,369.99,391.99,
415.31,440.00,466.16,493.88,
523.25,554.37,587.33,622.25,
659.26,698.46,739.99,783.99,
830.61,880.00,932.32,987.77,
1046.5,1108.7,1174.7,1244.5,
1318.5,1396.9,1480.0,1568.0,
1661.2,1760.0,1864.7,1975.5,
2093.0,2217.5,2349.3,2489.0,
2637.0,2793.8,2960.0,3136.0,
3322.4,3520.0,3729.3,3951.1,
4186.0,4434.9,4698.6,4978.0,
5274.0,5587.7,5919.9,6271.9,
6644.9,7040.0,7458.6,7902.1,
8372.0,8869.8,9397.3,9956.1,
10548.1,11175.3,11839.8,12543.9
]
note_names = [
"C -1", "C#-1", "D -1", "D#-1",
"E -1", "F -1", "F#-1", "G -1",
"G#-1", "A -1", "Bb-1", "B -1",
"C 0", "C# 0", "D 0", "D#-1",
"E 0", "F 0", "F# 0", "G 0",
"G# 0", "A 0", "Bb 0", "B 0",
"C 1", "C# 1", "D 1", "D# 1",
"E 1", "F 1", "F# 1", "G 1",
"G# 1", "A 1", "Bb 1", "B 1",
"C 2", "C# 2", "D 2", "D# 2",
"E 2", "F 2", "F# 2", "G 2",
"G# 2", "A 2", "Bb 2", "B 2",
"C 3", "C# 3", "D 3", "D# 3",
"E 3", "F 3", "F# 3", "G 3",
"G# 3", "A 3", "Bb 3", "B 3",
"C 4", "C# 4", "D 4", "D# 4",
"E 4", "F 4", "F# 4", "G 4",
"G# 4", "A 4", "Bb 4", "B 4",
"C 5", "C# 5", "D 5", "D# 5",
"E 5", "F 5", "F# 5", "G 5",
"G# 5", "A 5", "Bb 5", "B 5",
"C 6", "C# 6", "D 6", "D# 6",
"E 6", "F 6", "F# 6", "G 6",
"G# 6", "A 6", "Bb 6", "B 6",
"C 7", "C# 7", "D 7", "D# 7",
"E 7", "F 7", "F# 7", "G 7",
"G# 7", "A 7", "Bb 7", "B 7",
"C 8", "C# 8", "D 8", "D# 8",
"E 8", "F 8", "F# 8", "G 8",
"G# 8", "A 8", "Bb 8", "B 8",
"C 9", "C# 9", "D 9", "D# 9",
"E 9", "F 9", "F# 9", "G 9"
]
These lists are the frequencies and names of all 128 MIDI notes. They'll be implicitly available to use anywhere we import the midi
submodule.
Before we jump into the next section of code, I think it's worth discussing why we're building it this way. We're going to be defining our own simple text based API for controlling the synth, rather than using MIDI directly. But why not just forward the MIDI messages straight to the synthesizer and parse them there? In short, because the synthesizer doesn't need to know about MIDI. We manipulate the synthesizer sound by changing the frequency, phase and amplitude of the oscillator waves. MIDI doesn't work directly with any of those parameters, but rather specifies note and control change numbers. Later we'll adjust other parameters, but none of them are inherently related to MIDI.
So instead of forwarding MIDI messages directly to the synthesizer mailbox, when we receive MIDI input we'll extract the information we want from the mido.Message
object that we receive and use it to build a text based message that the synth thread will know how to parse. For simplicity, we'll define an interface similar to MIDI, but there is no reason we have to.
We'll write a few classes to help us build the messages we'll send to the synth thread. Create a file in the midi module called message_builder.py
.
import logging
def builder():
return CommandBuilder()
class MessageBuilder():
"""
Base class for constructing messages to send to the controller.
"""
def __init__(self) -> None:
self.log = logging.getLogger(__name__)
self._message = ""
@property
def message(self):
return self._message
def build(self) -> str:
return str(self._message).strip()
class CommandBuilder(MessageBuilder):
"""
The main class for starting a command message to the controller.
Messages should be constructed by calling builder() and then chaining
the methods to build the message. When the message is complete, call .build()
"""
def __init__(self) -> None:
super().__init__()
def note_on(self):
self._message += " note_on"
return NoteParameterBuilder(self.message)
def note_off(self):
self._message += " note_off"
return NoteParameterBuilder(self.message)
def control_change(self):
self._message += " control_change"
return CCParameterBuilder(self.message)
You'll notice a few things here. We're defining a base MessageBuilder
class that initializes a message
property to an empty string. Below that, we've included another class called CommandBuilder
. This class provides an entry point to our message building functionality. The idea is that you get an instance of the CommandBuilder
class, then start a message by calling one of its methods. Each method returns an instance of a new class (which we'll implement below) which knows how to add to that message type.
For example, if you called
my_msg = builder().note_on()
you would get back an instance of NoteParameterBuilder
with its message
initialized to " note_on"
. Let's see how these builder classes can add to the message. Add the following below the CommandBuilder
class:
class NoteParameterBuilder(MessageBuilder):
"""
Note messages currently need to specify note and channel in that order.
"""
def __init__(self, message_base: str) -> None:
super().__init__()
self._message = message_base
def with_note(self, note):
try:
int_val = int(note)
if int_val < 0 or int_val > 127:
raise ValueError
self._message += f" -n {int_val}"
except ValueError:
self.log.error(f"Unable to set note: {note}")
raise
return NoteParameterBuilder(self._message)
def on_channel(self, channel):
try:
int_val = int(channel)
if int_val < 0 or int_val > 15:
raise ValueError
self._message += f" -c {int_val}"
except ValueError:
self.log.error(f"Unable to set channel: {channel}")
raise
return NoteParameterBuilder(self._message)
class CCParameterBuilder(MessageBuilder):
"""
Control Changes messages currently need to specify channel, control number, and value in that order.
"""
def __init__(self, message_base: str) -> None:
super().__init__()
self._message = message_base
def on_channel(self, channel):
try:
int_val = int(channel)
if int_val < 0 or int_val > 15:
raise ValueError
self._message += f" -c {int_val}"
except ValueError:
self.log.error(f"Unable to set channel: {channel}")
raise
return CCParameterBuilder(self._message)
def with_value(self, value):
try:
int_val = int(value)
if int_val < 0 or int_val > 127:
raise ValueError("MIDI values are from 0-127")
self._message += f" -v {int_val}"
except ValueError:
self.log.error(f"Unable to set channel: {value}")
raise
return CCParameterBuilder(self._message)
def with_control_num(self, value):
try:
int_val = int(value)
if int_val < 0 or int_val > 127:
raise ValueError("MIDI values are from 0-127")
self._message += f" -n {int_val}"
except ValueError:
self.log.error(f"Unable to set channel: {value}")
raise
return CCParameterBuilder(self._message)
As you can see, each class provides additional message options depending on the base message. It's not fool proof, but it's less error prone than constructing strings or objects by hand.
Time to build the listener! Create a file in the midi
submodule called midi_listener.py
. Add the imports we'll need.
import logging
import threading
import queue
import mido
from . import message_builder as mb
And below the imports, add the following class:
class MidiListener(threading.Thread):
"""
Listens for MIDI messages on a given port and sends them to the synth mailbox.
"""
def __init__(self, thread_mailbox: queue.Queue, synth_mailbox: queue.Queue, port_name: str):
super().__init__(name=f"{port_name}-listener")
self.log = logging.getLogger(__name__)
self.thread_mailbox = thread_mailbox # The mailbox that receives commands from the main thread. Namely the 'exit' command to shut down gracefully.
self.synth_mailbox = synth_mailbox # The OUT mailbox where we send the parsed commands to be played by the synth
self.port_name = port_name
def run(self):
should_run = True
inport = None
try:
inport = mido.open_input(self.port_name)
self.log.info(f"Opened port {self.port_name}")
except:
self.log.error(f"Failed to open MIDI port at {self.port_name}. Closing the listener thread.")
should_run = False
while should_run:
# Receive MIDI messages from the port and send them to the synth mailbox
if msg := inport.receive():
match msg.type:
case "note_on":
ctrl_msg = mb.builder().note_on().with_note(msg.note).on_channel(msg.channel).build()
self.synth_mailbox.put(ctrl_msg)
case "note_off":
ctrl_msg = mb.builder().note_off().with_note(msg.note).on_channel(msg.channel).build()
self.synth_mailbox.put(ctrl_msg)
case "control_change":
ctrl_msg = mb.builder().control_change().on_channel(msg.channel).with_control_num(msg.control).with_value(msg.value).build()
self.synth_mailbox.put(ctrl_msg)
case "stop":
self.log.info(f"Received midi STOP message")
case _:
self.log.info(f"Matched unknown MIDI message: {msg}")
# get_nowait raises queue.Empty exception if there is nothing in the queue
# We don't want to block this thread checking for thread command messages
try:
if mail := self.thread_mailbox.get_nowait():
match mail.split():
case ['exit']:
self.log.info("Got exit command.")
should_run = False
case _:
self.log.info(f"Matched unknown mailbox message: {mail}")
except queue.Empty:
pass
if inport is not None:
inport.close()
return
Let's dig in to how the listener works. The most interesting code is in the run
method. First, in the try-except
block, we open a MIDI port with the name passed in from the constructor. Then we enter the main thread loop. The line
if msg := inport.receive():
is doing a lot of heavy lifting for us. We're using Python's walrus operator to assign the msg
variable as part of a larger expression testing whether it exists. This line is essentially saying "if inport.receive()
returns a value, assign it to a variable named msg
".
It's also critical to note here that receive is a blocking call. That means if there is no input to receive, the thread execution will pause here until there is. (Other threads can continue their work.) By placing this line inside a while loop, we've created a mechanism that will wait for MIDI messages and respond to them almost immediately.
After we get a message, we match it based on its type
. This is where our message builder comes in. We get an instance of the builder and call the methods that correspond to the message type and parameters, then pass the newly constructed message to the synth mailbox.
Once we've handled the MIDI input, the listener thread has one more job before checking for more input. We have to check if the main thread has sent the "exit"
command. In this section, we don't want to block while waiting for a message. We simply want to check if it's there and if not carry on with our business. That's why we use the Queue
class's get_nowait method. Instead of blocking if the queue is empty, this method raises a queue.Empty
exception, which we can easily handle with pass
.
Alright! Now that we've written our MIDI listener, let's test it out. Make sure your MIDI controller is plugged in to your computer and not attached to any other programs, like a DAW.
We'll need to know the name of the port the controller opens. To do that we'll use the Python REPL. Open your terminal (with the virtual environment still active) and run
$ python
This will open the REPL. Now what you need to do is
>>> import synth.midi as midi
>>> midi.get_available_controllers()
This should return a list of MIDI controllers attached to your OS. For example: ['3- Focusrite USB MIDI 0', 'MPK mini 3 1']
. Copy the name of your controller. Now open up settings.py
in your project and add the line:
auto_attach = "MPK mini 3 1"
But replace MPK mini 3 1
with the name of your controller.
Now, open up your __main__.py
file. Add the following imports:
import queue
import sys
from optparse import OptionParser
import synth.midi as midi
from synth.midi.midi_listener import MidiListener
And directly under if __name__ == "__main__":
and above the logging configuration, add:
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()
This configures an option parser for our program that will allow us to pass in the name of the MIDI controller as a runtime flag instead of only hardcoding it in settings.py
. Right below the character art, add:
available_ports = midi.get_available_controllers()
log.info(f"Available MIDI ports: {available_ports}")
And above where we create the stream player, but above the try block, add the lines:
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)
Replace the try-except block and everything below it with
try:
stream_player.play()
midi_listener.start()
while True:
if synth_mail := synth_mailbox.get():
log.info(f"{synth_mail}")
except KeyboardInterrupt:
log.info("Caught keyboard interrupt. Exiting the program.")
stream_player.stop()
listener_mailbox.put("exit")
midi_listener.join()
sys.exit(0)
At this point if you run the program and play your MIDI controller, you'll be able to see events being logged in the console. You'll also notice a small quirk of the program: after we press ctrl-c to quit, we now have to press a key on the MIDI controller as well before the program will exit. This is because of the way we wrote the MIDI listener as a blocking operation. We'll just have to deal with it for now, but fixing it is one of the suggested improvements in part 5!
Let's connect the MIDI listener to our temporary signal tree. In __main__.py
, find the lines
osc_a.frequency = 440.0
osc_b.frequency = 440.0
and remove them. Then right above while True:
add the line
current_note = None
And below log.info(f"{synth_mail}")
, put:
match synth_mail.split():
case ["note_on", "-n", note, "-c", channel]:
int_note = int(note)
freq = midi.frequencies[int_note]
osc_a.frequency = float(freq)
osc_b.frequency = float(freq)
current_note = note
case ["note_off", "-n", note, "-c", channel]:
if current_note == note:
osc_a.frequency = 0.0
osc_b.frequency = 0.0
current_note = None
Congratulations! You've officially built a mono synth! __main__.py
now looks like:
import logging
import queue
import sys
from optparse import OptionParser
from time import sleep
import synth.midi as midi
from . import settings
from .playback.stream_player import StreamPlayer
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 .midi.midi_listener import MidiListener
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(
"""
__
| |
__| |___ ______ __ __
/__ __/ __|______|__ | | | |
| | | | | | |__|______| |
| | __ | | | | |______ |
|__|____|__| |__|______|__| ______|__|
|____| |______| |______|
__ __
| | | |
_______ __ __ __ ____ __| |___ | | ____
__|_______| | | | | | |__|____|__ /__ __/ | |__|____|__
|__|_______ |__|______| | | __| | | | | | __| | |
|_______|__ |______ | | | | | | | __ | | | |
__________|__| ______|__| | | | | |__|____|__| | | | |
|__________| |______| |__| |__| |____| |__| |__|
"""
)
# 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)
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)
try:
stream_player.play()
midi_listener.start()
current_note = None
while True:
if synth_mail := synth_mailbox.get():
log.info(f"{synth_mail}")
match synth_mail.split():
case ["note_on", "-n", note, "-c", channel]:
int_note = int(note)
freq = midi.frequencies[int_note]
osc_a.frequency = float(freq)
osc_b.frequency = float(freq)
current_note = note
case ["note_off", "-n", note, "-c", channel]:
if current_note == note:
osc_a.frequency = 0.0
osc_b.frequency = 0.0
current_note = None
except KeyboardInterrupt:
log.info("Caught keyboard interrupt. Exiting the program.")
stream_player.stop()
listener_mailbox.put("exit")
midi_listener.join()
sys.exit(0)
Now when you run the program with python -m synth, you should be able to play one note at a time using your MIDI keyboard. If you want to pass the name of your MIDI controller as a command line flag, run the program with python -m synth -p <controller-name>
.
In this tutorial we gave ourselves a way to control the sound. We implemented a MIDI listener, which allows us to take input from a keyboard or other MIDI device. We handled note_on
and note_off
messages. And finally, we set up a simple mono synth as a test. In the next part, we'll combine the pieces we've built so far to create a poly synth.