Gist created for the Making A Synth With Python series of posts.

adsr_envelope.py

class ADSREnvelope:
    def __init__(self, attack_duration=0.05, decay_duration=0.2, sustain_level=0.7, \
                 release_duration=0.3, sample_rate=44100):
        self.attack_duration = attack_duration
        self.decay_duration = decay_duration
        self.sustain_level = sustain_level
        self.release_duration = release_duration
        self._sample_rate = sample_rate
        
    def get_ads_stepper(self):
        steppers = []
        if self.attack_duration > 0:
            steppers.append(itertools.count(start=0, \
                step= 1 / (self.attack_duration * self._sample_rate)))
        if self.decay_duration > 0:
            steppers.append(itertools.count(start=1, \
            step=-(1 - self.sustain_level) / (self.decay_duration  * self._sample_rate)))
        while True:
            l = len(steppers)
            if l > 0:
                val = next(steppers[0])
                if l == 2 and val > 1:
                    steppers.pop(0)
                    val = next(steppers[0])
                elif l == 1 and val < self.sustain_level:
                    steppers.pop(0)
                    val = self.sustain_level
            else:
                val = self.sustain_level
            yield val
    
    def get_r_stepper(self):
        val = 1
        if self.release_duration > 0:
            release_step = - self.val / (self.release_duration * self._sample_rate)
            stepper = itertools.count(self.val, step=release_step)
        else:
            val = -1
        while True:
            if val <= 0:
                self.ended = True
                val = 0
            else:
                val = next(stepper)
            yield val
    
    def __iter__(self):
        self.val = 0
        self.ended = False
        self.stepper = self.get_ads_stepper()
        return self
    
    def __next__(self):
        self.val = next(self.stepper)
        return self.val
        
    def trigger_release(self):
        self.stepper = self.get_r_stepper()

chain.py

class Chain:
    def __init__(self, generator, *modifiers):
        self.generator = generator
        self.modifiers = modifiers
        
    def __getattr__(self, attr):
        val = None
        if hasattr(self.generator, attr):
            val = getattr(self.generator, attr)
        else:
            for modifier in self.modifiers:
                if hasattr(modifier, attr):
                    val = getattr(modifier, attr)
                    break
            else:
                raise AttributeError(f"attribute '{attr}' does not exist")
        return val
    
    def trigger_release(self):
        tr = "trigger_release"
        if hasattr(self.generator, tr):
            self.generator.trigger_release()
        for modifier in self.modifiers:
            if hasattr(modifier, tr):
                modifier.trigger_release()
                
    @property
    def ended(self):
        ended = []; e = "ended"
        if hasattr(self.generator, e):
            ended.append(self.generator.ended)
        ended.extend([m.ended for m in self.modifiers if hasattr(m, e)])
        return all(ended)
    
    def __iter__(self):
        iter(self.generator)
        [iter(mod) for mod in self.modifiers if hasattr(mod, "__iter__")]
        return self
        
    def __next__(self):
        val = next(self.generator)
        [next(mod) for mod in self.modifiers if hasattr(mod, "__iter__")]
        for modifier in self.modifiers:
            val = modifier(val)
        return val

get_glider.py

def get_glider(old_freq,new_freq, glide, sample_rate):
    np.linspace(old_freq, new_freq, int(sample_rate * glide))
    if old_freq == new_freq:
        glider = itertools.cycle([new_freq])
    else: 
        glider = itertools.count(old_freq, (new_freq-old_freq)/(glide*sample_rate))
    for f in glider:
        if old_freq > new_freq:
            yield max(f, new_freq)
        else:
            yield min(f, new_freq)

get_samples.py

def get_samples(notes_dict, num_samples=256):
    return [sum([int(next(osc) * 32767) \
            for _, osc in notes_dict.items()]) \
            for _ in range(num_samples)]

get_simple_sine_amp.py

def get_sin_oscillator(freq=55, amp=1, sample_rate=44100):
    increment = (2 * math.pi * freq)/ sample_rate
    return (math.sin(v) * amp for v in itertools.count(start=0, step=increment))

less_simple_sine_generator.py

def get_sin_oscillator(freq, amp=1, phase=0, sample_rate=44100):
    phase = (phase / 360) * 2 * math.pi
    increment = (2 * math.pi * freq)/ sample_rate
    return (math.sin(phase + v) * amp for v in itertools.count(start=0, step=increment))

modulated_oscillator.py

class ModulatedOscillator:
    def __init__(self, oscillator, *modulators, amp_mod=None, freq_mod=None, phase_mod=None):
        self.oscillator = oscillator
        self.modulators = modulators # list
        self.amp_mod = amp_mod
        self.freq_mod = freq_mod
        self.phase_mod = phase_mod
        self._modulators_count = len(modulators)
    
    def __iter__(self):
        iter(self.oscillator)
        [iter(modulator) for modulator in self.modulators]
        return self
    
    def _modulate(self, mod_vals):
        if self.amp_mod is not None:
            new_amp = self.amp_mod(self.oscillator.init_amp, mod_vals[0])
            self.oscillator.amp = new_amp
            
        if self.freq_mod is not None:
            if self._modulators_count == 2:
                mod_val = mod_vals[1]
            else:
                mod_val = mod_vals[0]
            new_freq = self.freq_mod(self.oscillator.init_freq, mod_val)
            self.oscillator.freq = new_freq
            
        if self.phase_mod is not None:
            if self._modulators_count == 3:
                mod_val = mod_vals[2]
            else:
                mod_val = mod_vals[-1]
            new_phase = self.phase_mod(self.oscillator.init_phase, mod_val)
            self.oscillator.phase = new_phase
    
    def trigger_release(self):
        tr = "trigger_release"
        for modulator in self.modulators:
            if hasattr(modulator, tr):
                modulator.trigger_release()
        if hasattr(self.oscillator, tr):
            self.oscillator.trigger_release()
            
    @property
    def ended(self):
        e = "ended"
        ended = []
        for modulator in self.modulators:
            if hasattr(modulator, e):
                ended.append(modulator.ended)
        if hasattr(self.oscillator, e):
            ended.append(self.oscillator.ended)
        return all(ended)

    def __next__(self):
        mod_vals = [next(modulator) for modulator in self.modulators]
        self._modulate(mod_vals)
        return next(self.oscillator)

modulated_panner.py

class ModulatedPanner(Panner):
    def __init__(self, modulator):
        super().__init__(r=0)
        self.modulator = modulator
        
    def __iter__(self):
        iter(self.modulator)
        return self
    
    def __next__(self):
        self.r = (next(self.modulator) + 1) / 2
        return self.r

modulated_volume.py

class ModulatedVolume(Volume):
    def __init__(self, modulator):
        super().__init__(0.)
        self.modulator = modulator
        
    def __iter__(self):
        iter(self.modulator)
        return self
    
    def __next__(self):
        self.amp = next(self.modulator)
        return self.amp
    
    def trigger_release(self):
        if hasattr(self.modulator, "trigger_release"):
            self.modulator.trigger_release()
    
    @property
    def ended(self):
        ended = False
        if hasattr(self.modulator, "ended"):
            ended = self.modulator.ended
        return ended

monosynth.py

class MonoSynth(PolySynth):
    def _get_samples(self, osc, glider):
        samples = []
        for _ in range(self.num_samples):
            freq = next(glider)
            if freq != osc.freq:osc.freq = freq
            samples.append(next(osc))
        samples = (np.array(samples) * self.amp_scale).clip(-self.max_amp, self.max_amp)
        return np.int16(samples * 32767)
        
    def play(self, osc, close=False, glide=0.1):
        self._init_stream(1)

        try:
            play = False
            note = None
            glider = None
            
            while True:
                if play:
                    # Play the notes
                    samples = self._get_samples(osc, glider)
                    
                    # Stream Samples
                    self.stream.write(samples.tobytes())
                    
                if self.midi_input.poll():
                    # Add or remove notes from notes_dict
                    for event in self.midi_input.read(num_events=16):
                        (status, note_, vel, _), _ = event
                        freq = midi.midi_to_frequency(note_)
                        
                        if status == 0x80 and note == note_:
                            play = False
                            
                        elif status == 0x90:
                            if not play:
                                play = True
                                osc.freq = freq
                            glider = get_glider(osc.freq, freq, glide, self.sample_rate)
                            note = note_

        except KeyboardInterrupt as err:
            self.stream.close()
            if close:
                self.midi_input.close()

oscillator.py

from abc import ABC, abstractmethod

class Oscillator(ABC):
    def __init__(self, freq=440, phase=0, amp=1, \
                 sample_rate=44_100, wave_range=(-1, 1)):
        self._freq = freq
        self._amp = amp
        self._phase = phase
        self._sample_rate = sample_rate
        self._wave_range = wave_range
        
        # Properties that will be changed
        self._f = freq
        self._a = amp
        self._p = phase
        
    @property
    def init_freq(self):
        return self._freq
    
    @property
    def init_amp(self):
        return self._amp
    
    @property
    def init_phase(self):
        return self._phase
    
    @property
    def freq(self):
        return self._f
    
    @freq.setter
    def freq(self, value):
        self._f = value
        self._post_freq_set()
        
    @property
    def amp(self):
        return self._a
    
    @amp.setter
    def amp(self, value):
        self._a = value
        self._post_amp_set()
        
    @property
    def phase(self):
        return self._p
    
    @phase.setter
    def phase(self, value):
        self._p = value
        self._post_phase_set()
    
    def _post_freq_set(self):
        pass
    
    def _post_amp_set(self):
        pass
    
    def _post_phase_set(self):
        pass
    
    @abstractmethod
    def _initialize_osc(self):
        pass
    
    @staticmethod
    def squish_val(val, min_val=0, max_val=1):
        return (((val + 1) / 2 ) * (max_val - min_val)) + min_val

    @abstractmethod
    def __next__(self):
        return None
    
    def __iter__(self):
        self.freq = self._freq
        self.phase = self._phase
        self.amp = self._amp
        self._initialize_osc()
        return self

panner.py

class Panner:
    def __init__(self, r=0.5):
        self.r = r
        
    def __call__(self, val):
        r = self.r * 2
        l = 2 - r
        return (l * val, r * val)

param_mod.py

def amp_mod(init_amp, env):
    return env * init_amp
    
def freq_mod(init_freq, env, mod_amt=0.01, sustain_level=0.7):
    return init_freq + ((env - sustain_level) * init_freq * mod_amt)

polysynth.py

class PolySynth:
    def __init__(self, amp_scale=0.3, max_amp=0.8, sample_rate=44100, num_samples=64):
        # Initialize MIDI
        midi.init()
        if midi.get_count() > 0:
            self.midi_input = midi.Input(midi.get_default_input_id())
        else: 
            raise Exception("no midi devices detected")
        
        # Constants
        self.num_samples = num_samples
        self.sample_rate = sample_rate
        self.amp_scale = amp_scale
        self.max_amp = max_amp
    
    def _init_stream(self, nchannels):
        # Initialize the Stream object
        self.stream = pyaudio.PyAudio().open(
            rate=self.sample_rate,
            channels=nchannels,
            format=pyaudio.paInt16,
            output=True,
            frames_per_buffer=self.num_samples
        )
        
    def _get_samples(self, notes_dict):
        # Return samples in int16 format
        samples = []
        for _ in range(self.num_samples):
            samples.append(
                [next(osc[0]) for _, osc in notes_dict.items()]
            )
        samples = np.array(samples).sum(axis=1) * self.amp_scale
        samples = np.int16(samples.clip(-self.max_amp, self.max_amp) * 32767)
        return samples.reshape(self.num_samples, -1)

    def play(self, osc_function=get_sin_oscillator, close=False):
        # Check for release trigger, number of channels and init Stream
        tempcf = osc_function(1, 1, self.sample_rate)
        has_trigger = hasattr(tempcf, "trigger_release")
        tempsm = self._get_samples({-1: [tempcf, False]})
        nchannels = tempsm.shape[1]
        self._init_stream(nchannels)

        try:
            notes_dict = {}
            while True:
                if notes_dict:
                    # Play the notes
                    samples = self._get_samples(notes_dict)
                    self.stream.write(samples.tobytes())
                    
                if self.midi_input.poll():
                    # Add or remove notes from notes_dict
                    for event in self.midi_input.read(num_events=16):
                        (status, note, vel, _), _ = event
                        
                        # Note OFF
                        if status == 0x80 and note in notes_dict:
                            if has_trigger:
                                notes_dict[note][0].trigger_release()
                                notes_dict[note][1] = True
                            else:
                                del notes_dict[note]
                        
                        # Note ON
                        elif status == 0x90:
                            freq = midi.midi_to_frequency(note)
                            notes_dict[note] = [
                                osc_function(freq=freq, amp=vel/127, sample_rate=self.sample_rate), 
                                False
                            ]

                if has_trigger:
                    # Delete notes if ended
                    ended_notes = [k for k,o in notes_dict.items() if o[0].ended and o[1]]
                    for note in ended_notes:
                        del notes_dict[note]

        except KeyboardInterrupt as err:
            self.stream.close()
            if close:
                self.midi_input.close()

sawtooth_oscillator.py

class SawtoothOscillator(Oscillator):
    def _post_freq_set(self):
        self._period = self._sample_rate / self._f
        self._post_phase_set
        
    def _post_phase_set(self):
        self._p = ((self._p + 90)/ 360) * self._period
    
    def _initialize_osc(self):
        self._i = 0
    
    def __next__(self):
        div = (self._i + self._p )/self._period
        val = 2 * (div - math.floor(0.5 + div))
        self._i = self._i + 1
        if self._wave_range is not (-1, 1):
            val = self.squish_val(val, *self._wave_range)
        return val * self._a

simple_sine_generator.py

def get_sin_oscillator(freq, sample_rate):
    increment = (2 * math.pi * freq)/ sample_rate
    return (math.sin(v) for v in itertools.count(start=0, step=increment))

sine_oscillator.py

class SineOscillator(Oscillator):
    def _post_freq_set(self):
        self._step = (2 * math.pi * self._f) / self._sample_rate
        
    def _post_phase_set(self):
        self._p = (self._p / 360) * 2 * math.pi
        
    def _initialize_osc(self):
        self._i = 0
        
    def __next__(self):
        val = math.sin(self._i + self._p)
        self._i = self._i + self._step
        if self._wave_range is not (-1, 1):
            val = self.squish_val(val, *self._wave_range)
        return val * self._a

square_oscillator.py

class SquareOscillator(SineOscillator):
    def __init__(self, freq=440, phase=0, amp=1, \
                 sample_rate=44_100, wave_range=(-1, 1), threshold=0):
        super().__init__(freq, phase, amp, sample_rate, wave_range)
        self.threshold = threshold
    
    def __next__(self):
        val = math.sin(self._i + self._p)
        self._i = self._i + self._step
        if val < self.threshold:
            val = self._wave_range[0]
        else:
            val = self._wave_range[1]
        return val * self._a

synth_loop.py

try: 
    notes_dict = {}
    while True:
        if notes_dict:
            # Play the notes
            samples = get_samples(notes_dict)
            samples = np.int16(samples).tobytes()
            stream.write(samples)
            
        if midi_input.poll():
            # Add or remove notes from notes_dict
            for event in midi_input.read(num_events=16):
                (status, note, vel, _), _ = event
                if status == 0x80 and note in notes_dict:
                    del notes_dict[note]
                elif status == 0x90 and note not in notes_dict:
                    freq = midi.midi_to_frequency(note)
                    notes_dict[note] = get_sin_oscillator(freq=freq, amp=vel/127)
                    
except KeyboardInterrupt as err:
    print("Stopping...")

triangle_oscillator.py

class TriangleOscillator(SawtoothOscillator):
    def __next__(self):
        div = (self._i + self._p)/self._period
        val = 2 * (div - math.floor(0.5 + div))
        val = (abs(val) - 0.5) * 2
        self._i = self._i + 1
        if self._wave_range is not (-1, 1):
            val = self.squish_val(val, *self._wave_range)
        return val * self._a

volume.py

class Volume:
    def __init__(self, amp=1.):
        self.amp = amp
        
    def __call__(self, val):
        _val = None
        if isinstance(val, Iterable):
            _val = tuple(v * self.amp for v in val)
        elif isinstance(val, (int, float)):
            _val = val * self.amp
        return _val

wave_adder.py

class WaveAdder:
    def __init__(self, *oscillators):
        self.oscillators = oscillators
        self.n = len(oscillators)
    
    def __iter__(self):
        [iter(osc) for osc in self.oscillators]
        return self
    
    def __next__(self):
        return sum(next(osc) for osc in self.oscillators) / self.n

wave_adder_recode.py

from collections.abc import Iterable

class WaveAdder:
    def __init__(self, *generators, stereo=False):
        self.generators = generators
        self.stereo = stereo
        
    def _mod_channels(self, _val):
        val = _val
        if isinstance(_val, (int, float)) and self.stereo:
            val = (_val, _val)
        elif isinstance(_val, Iterable) and not self.stereo:
            val = sum(_val)/len(_val)
        return val
    
    def trigger_release(self):
        [gen.trigger_release() for gen in self.generators if hasattr(gen, "trigger_release")]
    
    @property
    def ended(self):
        ended = [gen.ended for gen in self.generators if hasattr(gen, "ended")]
        return all(ended)
    
    def __iter__(self):
        [iter(gen) for gen in self.generators]
        return self
            
    def __next__(self):
        vals = [self._mod_channels(next(gen)) for gen in self.generators]
        if self.stereo:
            l, r = zip(*vals)
            val = (sum(l)/len(l), sum(r)/len(r))
        else:
            val = sum(vals)/ len(vals)
        return val

wave_to_file.py

import numpy as np
from scipy.io import wavfile

def wave_to_file(wav, wav2=None, fname="temp.wav", amp=0.1, sample_rate=44100):
    wav = np.array(wav)
    wav = np.int16(wav * amp * (2**15 - 1))
    
    if wav2 is not None:
        wav2 = np.array(wav2)
        wav2 = np.int16(wav2 * amp * (2 ** 15 - 1))
        wav = np.stack([wav, wav2]).T
    
    wavfile.write(fname, sample_rate, wav)