[Python] Microcontroller as a DJ, measuring time between two onsets

I'm working on a larger project to automate the beatmatching of two vinyl turntables playing techno music. The system runs on a Raspberry Pi and uses a stepper motor connected to a linear axis table, which is then connected to the pitchfader of the turntable to adjust the tempo.

I capture the audio from both turntables via a sound card and do real-time signal processing in Python to extract the kick drums, which define the tempo and phase of the tracks.

My signal processing chain (for both channels):

• Lowpass filter at 180 Hz (4th order)
• RMS envelope
• Thresholding
• Gate/Trigger logic (detecting rising edges)

audio analysis

You find a signal graph attached. your text This reliably gives me a clean, artificial pulse train of kick positions (no noise, no false triggers).

My goal:


I want to determine if the two signals are in sync (same tempo & phase), and if not, how far apart the kicks are in time — ideally to feed a PI controller that adjusts the tempo of one deck using the motor.

My question:


What’s the most reliable way to determine if two signals (kick trains) are synchronized — and if not, how far apart they are in phase or time? The signals have different frequencies (e.g., 1150 BPM vs. 140 BPM) and a variable phase offset

I would appreciate pointers on robust methods or best practices — especially in a real-time context.

Thanks!

#!/usr/bin/env python3
import time
import numpy as np
import sounddevice as sd
from scipy.signal import butter, lfilter
import RPi.GPIO as GPIO

#filter
FS = 44100
BLOCK_SIZE = 1024
CUTOFF_HZ = 180
FILTER_ORDER = 4
RMS_FRAME = 512
RMS_HOP = 128
THRESH = 0.0006
HOLD_MS = 10
RETRIG_MS = 200

#motor para
PULSE_PIN = 20
DIR_PIN = 21
PULSES_PER_REV = 200
TRAVEL_REV_CM = 0.8
CM_PER_PULSE = TRAVEL_REV_CM / PULSES_PER_REV
MAX_POS_CM = 4.0
MIN_POS_CM = -4.0
current_pos_cm = 0.0

# --- PI-parameter
KP = 30.0
KI = 5

# ========== Setup GPIO ===========
GPIO.setmode(GPIO.BCM)
GPIO.setup(DIR_PIN, GPIO.OUT)
GPIO.setup(PULSE_PIN, GPIO.OUT)
# Software-PWM
pwm = GPIO.PWM(PULSE_PIN, 1)
pwm.start(0)


def butter_lowpass(cutoff, fs, order):
nyq = 0.5 * fs
norm = cutoff / nyq
return butter(order, norm, btype='low', analog=False)

b, a = butter_lowpass(CUTOFF_HZ, FS, FILTER_ORDER)
apply_filter = lambda x: lfilter(b, a, x)

def rms_envelope(signal, frame_size, hop_size):
env = []
for i in range(0, len(signal) - frame_size, hop_size):
frame = signal[i:i+frame_size]
env.append(np.sqrt(np.mean(frame**2)))
return np.repeat(env, hop_size)[:len(signal)]

def detect_kicks(env):
out = np.zeros_like(env)
hold = int(HOLD_MS/1000 * FS / RMS_HOP)
retrig = int(RETRIG_MS/1000 * FS / RMS_HOP)
last = -np.inf
for i, v in enumerate(env):
if v > THRESH and (i - last) > retrig:
out[i:i+hold] = 1
last = i
return out

# ========== PI control =============
class PIController:
def __init__(self, kp, ki):
self.kp = kp
self.ki = ki
self.integral = 0.0

def update(self, error, dt):
self.integral += error * dt
return self.kp * error + self.ki * self.integral

pi_controller = PIController(KP, KI)

# ========== Motor Control =============
def move(delta_cm):
global current_pos_cm
target = current_pos_cm + delta_cm
if target > MAX_POS_CM:
delta_cm = MAX_POS_CM - current_pos_cm
target = MAX_POS_CM
elif target < MIN_POS_CM:
delta_cm = MIN_POS_CM - current_pos_cm
target = MIN_POS_CM
if abs(delta_cm) < CM_PER_PULSE:
return
steps = abs(int(delta_cm / CM_PER_PULSE))
freq = 1000.0 # Hz
duration = steps / freq
GPIutput(DIR_PIN, GPIO.HIGH if delta_cm > 0 else GPIO.LOW)
pwm.ChangeFrequency(int(freq))
pwm.ChangeDutyCycle(50)
time.sleep(duration)
pwm.ChangeDutyCycle(0)
current_pos_cm = target

# ========== Kick Sync Callback =========
kick_times_L = []
kick_times_R = []


def audio_callback(indata, frames, time_info, status):
global kick_times_L, kick_times_R
now = time.time()
# Stereo input
L = indata[:,0]
R = indata[:,1]
env_L = detect_kicks(rms_envelope(apply_filter(L), RMS_FRAME, RMS_HOP))
env_R = detect_kicks(rms_envelope(apply_filter(R), RMS_FRAME, RMS_HOP))

for i in range(1, len(env_L)):
if env_L == 1 and env_L[i-1] == 0:
kick_times_L.append(now + i/FS)
if env_R == 1 and env_R[i-1] == 0:
kick_times_R.append(now + i/FS)

usedL, usedR = [], []
for tL in kick_times_L:
if not kick_times_R:
break
tR = min(kick_times_R, key=lambda x: abs(x-tL))
if abs(tL - tR) <= 1.0:
err = tL - tR
dt = abs(tL - tR)
ctrl = pi_controller.update(err, dt)
ctrl = max(-MAX_POS_CM, min(MAX_POS_CM, ctrl))
move(ctrl)
usedL.append(tL)
usedR.append(tR)

kick_times_L = [t for t in kick_times_L if t not in usedL]
kick_times_R = [t for t in kick_times_R if t not in usedR]

# ========== main==============
if __name__ == '__main__':
with sd.InputStream(channels=2, samplerate=FS, blocksize=BLOCK_SIZE,
callback=audio_callback):
print("Kick Sync & Motorsteuerung auf RPi aktiv…")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass
pwm.stop()
GPIO.cleanup()

Continue reading...