For more information about the examples, such as how the Python and Mojo files interact with each other, see the Examples Overview

WavetableOscSIMD

Example of a wavetable oscillator using custom wavetables loaded from files.

This example uses SIMDBuffer instead of Buffer to load the wavetable. This allows for more efficient processing for wavetables with a small number of channels (2-8), where the number of channels is known ahead of time, but it should not be used with wavetables that have a large number of waveforms.

This example also uses Mojo-side Poly vs PVoiceAllocator.

Python Code

import sys
from pathlib import Path

# In order to do this, it needs to add the parent directory to the path
# (the next line here) so that it can find the mmm_src and mmm_utils packages.
# If you want to run it line by line in a REPL, skip this line!
sys.path.insert(0, str(Path(__file__).parent.parent))
from mmm_python import *

def main():
    mmm_audio = MMMAudio(128, graph_name="WavetableOscSIMD", package_name="examples")
    mmm_audio.start_audio() 

    import threading, mido, time

    # open your midi device - you may need to change the device name
    in_port = mido.open_input('Oxygen Pro Mini USB MIDI')

    # PolyPal correctly formats messages to be sent to a Synth that uses a Poly object
    poly_pal = PolyPal(mmm_audio, "poly", 10)

    # Create stop event
    global stop_event
    stop_event = threading.Event()
    def start_midi():
        while not stop_event.is_set():
            for msg in in_port.iter_pending():
                if stop_event.is_set():  # Check if we should stop
                    return

                if msg.type in ["note_on", "note_off", "control_change"]:
                    if msg.type == "note_on":
                        poly_pal.send_ints([msg.note, (msg.velocity)])  
                    if msg.type == "note_off":
                        poly_pal.send_ints([msg.note, 0.0])  
                    if msg.type == "control_change":
                        print(f"Control Change: {msg.control} Value: {msg.value}")
                        # Example: map CC 1 to wubb_rate of all voices
                        if msg.control == 1:
                            wubb_rate = linexp(msg.value, 0, 127, 0.1, 10.0)
                            mmm_audio.send_float("wubb_rate", wubb_rate)
                        if msg.control == 33:
                            mmm_audio.send_float("filter_cutoff", linexp(msg.value, 0, 127, 20.0, 20000.0))
                        if msg.control == 34:
                            mmm_audio.send_float("filter_resonance", linexp(msg.value, 0, 127, 0.1, 1.0))

            time.sleep(0.01)
    # Start the thread
    midi_thread = threading.Thread(target=start_midi, daemon=False)
    midi_thread.start()

if __name__ == "__main__":
    main()

Mojo Code

from mmm_audio import *

struct OscVoice(PolyObject):
    var osc: Osc[1,Interp.sinc,0]
    var tri: LFTri[]
    var world: World
    var env: ASREnv
    var gate: Bool
    var freq: Float64
    var vol: Float64
    var wubb_rate: Float64
    var messenger: Messenger
    var triggered: Bool
    var just_offset: List[Float64]

    fn check_active(mut self) -> Bool:
        return self.env.is_active

    # Poly will use this function to release the voice when it receives a note off message for the note that this voice is playing. 
    fn set_gate(mut self, gate: Bool):
        self.gate = gate

    # necessary to ensure a fresh env when the voice is copied by Poly
    fn reset_env(mut self):
        self.env = ASREnv(self.world)

    fn __init__(out self, world: World, name_space: String = ""):
        self.osc = Osc[1,Interp.sinc,0](world)
        self.tri = LFTri(world)
        self.env = ASREnv(world)
        self.gate = False
        self.freq = 440.0
        self.vol = 1.0
        self.wubb_rate = 0.5
        self.messenger = Messenger(world, name_space)
        self.world = world
        self.triggered = False
        self.just_offset = [0.0, 0.1173, 0.0391, 0.1564, -0.1369, -0.0196, -0.0978, 0.0196, 0.1369, -0.1564, 0.1760, -0.1173]

    fn next(mut self, ref buffer: SIMDBuffer) -> MFloat[1]:
        osc_frac = self.tri.next(self.wubb_rate, 0.75, trig=self.gate) * 0.5 + 0.5
        return self.osc.next_vwt(buffer, self.freq, osc_frac = osc_frac) * self.env.next(0.01,0.2,0.7,self.gate,2) * self.vol

struct WavetableOscSIMD(Movable, Copyable):
    comptime wavetables_per_channel = 8
    comptime num_messages = 10

    var world: World  
    var voices: List[OscVoice]
    var buffer: SIMDBuffer[Self.wavetables_per_channel]
    var file_name: String
    var messenger: Messenger
    var filter_cutoff: Float64
    var filter_resonance: Float64
    var moog_filter: VAMoogLadder[1,1]
    var poly: PolyGate


    fn __init__(out self, world: World):
        self.world = world
        self.file_name = "resources/small_wavetable8.wav"

        self.buffer = SIMDBuffer[Self.wavetables_per_channel].load(self.file_name, num_wavetables=self.wavetables_per_channel)
        self.voices = [OscVoice(self.world, "voice_"+String(i)) for i in range(8)]

        self.messenger = Messenger(world)
        self.filter_cutoff = 20000.0
        self.filter_resonance = 0.5
        self.moog_filter = VAMoogLadder[1,1](self.world)
        self.poly = PolyGate(8, 16, world, "poly")

    fn __repr__(self) -> String:
        return String("Default")

    fn loadBuffer(mut self):
        self.buffer = SIMDBuffer[Self.wavetables_per_channel].load(self.file_name, num_wavetables=self.wavetables_per_channel)

    fn next(mut self) -> MFloat[2]:
        if self.messenger.notify_update(self.file_name, "load_file"):
            self.loadBuffer()

        # the callback function sent to the Poly, to be called whenever a new trigger is received from Python.
        # the kinds of messages the Messenger can receive are defined by the type of the `note` argument in the callback function
        fn callback(mut poly_object: OscVoice, mut vals: List[Int]):
            if vals[1] > 0: # the call_back will be called for both note on and note off messages
                midi = vals[0] + poly_object.just_offset[vals[0] % 12]
                print(vals[0], midi)
                poly_object.freq = midicps(midi)
                poly_object.vol = Float64(vals[1]) / 127.0
        # the poly has an internal Messenger that receives messages from Python. these have to be in the form of a List[Float64] or a List[Int]
        # for next_gate, the first value in the list is the note to trigger and the second value is the velocity or volume of the note, where 0 denotes a note off message. the callback function receives the list of ints or floats as the second argument, so the PolyObject can be controlled by the message from Python.
        self.poly.next(self.voices, call_back=callback)

        # add the output of all the voices
        var out = 0.0
        for ref voice in self.voices:
            out += voice.next(self.buffer)

        self.messenger.update(self.filter_cutoff, "filter_cutoff")
        self.messenger.update(self.filter_resonance, "filter_resonance")
        wubb_rate = 0.0
        got_wubbed = self.messenger.notify_update(wubb_rate, "wubb_rate")
        if got_wubbed:
            for ref voice in self.voices:
                voice.wubb_rate = wubb_rate
        sample = self.moog_filter.next(out, self.filter_cutoff, self.filter_resonance)

        return sample * 0.5