The examples
The package's python folder ships these classes. Each is the file itself, included here as it
ships; open one in a text editor beside Max, save a change, and the objects using it reload.
For tap.python~
default.py — a gain, per sample
What both objects load with no argument: a gain attribute, the messages greet, int, float and
bang, and process() once per sample. In tap.python, process() is a message like the others.
from attrs import define, field
@define
class default:
"""The class loaded by tap.python~ and tap.python when no source argument is given.
Demonstrates the mapping from Python to Max:
- the annotated 'gain' field becomes a Max attribute
- any public method (like 'greet') becomes a Max message
- methods named 'int' and 'float' answer those standard Max messages
- process() runs on the audio signal in tap.python~, once per sample (see
numpy_gain.py for the much cheaper once-per-vector form)
- in tap.python, what a method returns is output: bang outputs the gain
"""
gain: float = field(default = 1.0)
def greet(self, name: str) -> None:
print(f"hello {name}, from python!")
def process(self, x: float) -> float:
return x * self.gain
def bang(self) -> float:
return self.gain
# These two come last on purpose: once `def float` has run, the name `float` in
# the class body means this method, so an annotation written after it would no
# longer mean the built-in type. (Inside method bodies the built-ins are unaffected.)
def int(self, new_gain: int) -> None:
self.gain = float(new_gain)
def float(self, new_gain: float) -> None:
self.gain = new_gain
numpy_gain.py — a gain, per vector
The same gain written for numpy: process() is hinted with arrays, so it runs once per signal
vector. See Performance for what that saves.
import numpy as np
from attrs import define, field
@define
class numpy_gain:
"""A gain, processed a whole signal vector at a time with numpy.
Because process() is annotated np.ndarray, tap.python~ calls it once per signal
vector instead of once per sample — far cheaper, and the way to write anything
that has to run in real time. x is reused from one call to the next: copy it
if you want to keep it.
"""
gain: float = field(default = 1.0)
def process(self, x: np.ndarray) -> np.ndarray:
return x * self.gain
allpass.py — a filter that knows the sample rate
A Schroeder allpass filter, per sample: prepare() turns its delay in milliseconds into samples, and
attrs validators guard its fields. The notebook runs it the way the object does.
import numpy as np
from attrs import define, field
@define
class allpass:
"""Schroeder allpass filter over a delay line, for use with tap.python~.
y[n] = alpha * x[n] + x[n - D] - alpha * y[n - D], with D the delay in samples:
unity gain at every frequency, with a frequency-dependent phase shift.
"""
delay: float = field(default = 1.0) # delay time in ms
alpha: float = field(default = 0.5) # allpass coefficient, strictly between -1 and 1
_fs: float = field(init = False, default = 48000.0) # set by prepare()
_delay_in_samples: int = field(init = False, default = 1)
_x: np.ndarray = field(init = False, factory = lambda: np.zeros(1))
_y: np.ndarray = field(init = False, factory = lambda: np.zeros(1))
_write_index: int = field(init = False, default = 0)
_read_index: int = field(init = False, default = 0)
# Validators also run on plain attribute assignment, which is how tap.python~
# sets attributes from Max. During __init__ they fire before the private
# fields exist, so guard with hasattr and do the initial setup in post-init.
@delay.validator
def _delay_changed(self, attribute, value):
if value < 0.0:
raise ValueError("delay must be non-negative")
if hasattr(self, "_delay_in_samples"):
self._update_delay(value, self._fs)
@alpha.validator
def _check_alpha(self, attribute, value):
# |alpha| = 1 puts the feedback pole on the unit circle: the filter no longer decays
if not -1.0 < value < 1.0:
raise ValueError("alpha must be strictly between -1.0 and 1.0")
def __attrs_post_init__(self):
self._update_delay(self.delay, self._fs)
self.clear()
def prepare(self, sample_rate: float, vector_size: int) -> None:
"""Called by tap.python~ with Max's audio settings, before audio and whenever they change."""
self._fs = sample_rate
self._update_delay(self.delay, sample_rate)
def _update_delay(self, delay_in_ms, sampling_frequency):
new_delay_in_samples = max(1, int((delay_in_ms / 1000.0) * sampling_frequency))
if new_delay_in_samples != self._delay_in_samples:
self._delay_in_samples = new_delay_in_samples
print(f"Setting delay to {delay_in_ms} ms ({new_delay_in_samples} samples @ {sampling_frequency:g} Hz)")
self.clear()
def clear(self) -> None:
self._x = np.zeros(self._delay_in_samples)
self._y = np.zeros(self._delay_in_samples)
# process() reads before it writes, so reading the slot about to be
# overwritten yields the value written exactly _delay_in_samples ago
self._write_index = 0
self._read_index = 0
def process(self, x: float) -> float:
x1 = self._x[self._read_index]
y1 = self._y[self._read_index]
y = x1 + (x - y1) * self.alpha
self._x[self._write_index] = x
self._y[self._write_index] = y
self._read_index = (self._read_index + 1) % self._delay_in_samples
self._write_index = (self._write_index + 1) % self._delay_in_samples
return y
numpy_allpass.py — the same filter, per vector
The same filter a vector at a time — the core battery checks the two sample for sample.
import numpy as np
from attrs import define, field
@define
class numpy_allpass:
"""The allpass filter of allpass.py, processed a whole signal vector at a time with numpy.
y[n] = alpha * x[n] + x[n - D] - alpha * y[n - D], with D the delay in samples. The
feedback reaches back D samples, so any stretch of up to D samples depends only on samples
before it: each stretch is one numpy expression, and a vector is split only when the delay
is shorter than it. The output is the same as allpass.py's, sample for sample, at a small
fraction of the cost — compare the two in the PythonTap book's Performance chapter.
"""
delay: float = field(default = 1.0) # delay time in ms
alpha: float = field(default = 0.5) # allpass coefficient, strictly between -1 and 1
_fs: float = field(init = False, default = 48000.0) # set by prepare()
_vector_size: int = field(init = False, default = 64) # set by prepare()
_delay_in_samples: int = field(init = False, default = 1)
# the last D inputs and outputs, then room for one vector
_x: np.ndarray = field(init = False, factory = lambda: np.zeros(1))
_y: np.ndarray = field(init = False, factory = lambda: np.zeros(1))
# As in allpass.py: validators also run when tap.python~ sets an attribute, and during
# __init__ they fire before the private fields exist, hence the hasattr guard.
@delay.validator
def _delay_changed(self, attribute, value):
if value < 0.0:
raise ValueError("delay must be non-negative")
if hasattr(self, "_delay_in_samples"):
self._update_delay(value, self._fs)
@alpha.validator
def _check_alpha(self, attribute, value):
# |alpha| = 1 puts the feedback pole on the unit circle: the filter no longer decays
if not -1.0 < value < 1.0:
raise ValueError("alpha must be strictly between -1.0 and 1.0")
def __attrs_post_init__(self):
self._update_delay(self.delay, self._fs)
self.clear()
def prepare(self, sample_rate: float, vector_size: int) -> None:
"""Called by tap.python~ with Max's audio settings, before audio and whenever they change."""
self._fs = sample_rate
if vector_size != self._vector_size:
self._vector_size = vector_size
self.clear()
self._update_delay(self.delay, sample_rate)
def _update_delay(self, delay_in_ms, sampling_frequency):
new_delay_in_samples = max(1, int((delay_in_ms / 1000.0) * sampling_frequency))
if new_delay_in_samples != self._delay_in_samples:
self._delay_in_samples = new_delay_in_samples
print(f"Setting delay to {delay_in_ms} ms ({new_delay_in_samples} samples @ {sampling_frequency:g} Hz)")
self.clear()
def clear(self) -> None:
self._x = np.zeros(self._delay_in_samples + self._vector_size)
self._y = np.zeros(self._delay_in_samples + self._vector_size)
def process(self, x: np.ndarray) -> np.ndarray:
d, n = self._delay_in_samples, len(x)
if d + n > len(self._x): # a longer vector than prepare() announced: make room, keeping the history
self._x = np.concatenate((self._x[:d], np.zeros(n)))
self._y = np.concatenate((self._y[:d], np.zeros(n)))
xs, ys, alpha = self._x, self._y, self.alpha
xs[d:d + n] = x
for start in range(0, n, d): # at most d samples at a time: each y[n - d] is already computed
end = min(start + d, n)
# written as allpass.py computes it, so the two agree to the last bit
ys[d + start:d + end] = xs[start:end] + (xs[d + start:d + end] - ys[start:end]) * alpha
y = ys[d:d + n].copy()
xs[:d] = xs[n:n + d] # keep the last d inputs and outputs for the next vector
ys[:d] = ys[n:n + d]
return y
stereo_width.py — two inputs, two outputs
process()'s parameters are the object's signal inlets and its return hint its outlets.
import numpy as np
from attrs import define, field
@define
class stereo_width:
"""Stereo width: two signal inputs and two outputs, left and right.
process()'s parameters are the object's signal inlets — [tap.python~ stereo_width] has two —
and its return hint, tuple[np.ndarray, np.ndarray], gives it two outlets. The signal is split
into what the channels share (mid) and what differs (side); width scales the side: 0 folds
the pair to mono, 1 leaves it as it was, more than 1 widens it.
"""
width: float = field(default = 1.0)
@width.validator
def _check_width(self, attribute, value):
if value < 0.0:
raise ValueError("width must be 0 or more")
def process(self, left: np.ndarray, right: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
mid = (left + right) * 0.5
side = (left - right) * (0.5 * self.width)
return mid + side, mid - side
For tap.python
euclid.py — a rhythm from a bang
from attrs import define, field
@define
class euclid:
"""A Euclidean rhythm, for tap.python: a bang outputs `steps` 0s and 1s with `pulses` of the 1s
spread as evenly as they go — 8 steps and 3 pulses is 1 0 0 1 0 0 1 0 — a list to drive a step
sequencer.
@steps and @pulses are its attributes, so `steps 16` or an attrui changes the pattern; what
bang() returns, a list of ints, is what the object outputs.
"""
steps: int = field(default = 8)
pulses: int = field(default = 3)
def bang(self) -> list[int]:
steps = max(1, self.steps)
return [int((i * self.pulses) % steps < self.pulses) for i in range(steps)]
scale.py — a list through numpy
import numpy as np
from attrs import define, field
@define
class scale:
"""A list in, scaled and offset with numpy, a list out, for tap.python: with @factor 2 and
@offset 1, the list 1 2 3 outputs 3 5 7.
A last parameter hinted np.ndarray takes every atom of the message as one float64 array, and an
array returned is output as a list — so a list message runs through numpy and back.
"""
factor: float = field(default = 1.0)
offset: float = field(default = 0.0)
# `list` comes last on purpose: once `def list` has run, the name `list` in the class body means
# this method, so an annotation written after it would no longer mean the built-in type. (Inside
# method bodies the built-ins are unaffected.)
def list(self, values: np.ndarray) -> np.ndarray:
return values * self.factor + self.offset
note_name.py — two outlets
from attrs import define
NAMES = ("C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B")
@define
class note_name:
"""A MIDI note number in, its name out, for tap.python: the pitch class from the left outlet and
the octave from the right — 60 outputs C and 4, middle C being C4.
The return hint tuple[str, int] gives the object two outlets, one value each, output right to
left as Max objects do: the octave, then the name (as `symbol C`).
"""
# `int` comes last on purpose: once `def int` has run, the name `int` in the class body means this
# method, so an annotation written after it would no longer mean the built-in type. (Inside method
# bodies the built-ins are unaffected.)
def int(self, note: int) -> tuple[str, int]:
return NAMES[note % 12], note // 12 - 1