-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathaudio_engine.py
More file actions
55 lines (40 loc) · 1.33 KB
/
audio_engine.py
File metadata and controls
55 lines (40 loc) · 1.33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import sounddevice as sd
import numpy as np
import soundfile as sf
from queue import Queue
# Load a sample
BUFFER_SIZE = 256
SR = 44100
pending = Queue()
active_samples = [] # list of (audio_array, current_position)
def load_sample(path):
audio, sr = sf.read(path, dtype='float32')
# Convert mono to stereo
if audio.ndim == 1:
audio = np.stack([audio, audio], axis=1) # (n,) -> (n, 2)
return audio, sr
clap, sr = load_sample("samples/TR-808 Kit/Clap.wav")
hihat, sr1 = load_sample("samples/TR-808 Kit/Hihat.wav")
kick, sr2 = load_sample("samples/TR-808 Kit/Kick Mid.wav")
def play_sample(audio):
pending.put([audio, 0]) # safe to call from any thread
def callback(outdata, frames, time, status):
mixed = np.zeros((frames, 2))
while not pending.empty():
active_samples.append(pending.get_nowait())
for sample in active_samples:
audio, pos = sample
end = pos + frames
chunk = audio[pos:end]
mixed[:len(chunk)] += chunk
sample[1] += frames # advance position
# Remove finished samples
active_samples[:] = [s for s in active_samples if s[1] < len(s[0])]
outdata[:] = np.clip(mixed, -1.0, 1.0)
stream = sd.OutputStream(
samplerate=SR,
blocksize=BUFFER_SIZE,
channels=2,
dtype='float32',
callback=callback
)