Testing pulse audio using ctypes

sound_test.py

#Run this with python3 interactively like
#python3 -i sound_test.py

#In the python interactive console you can play with
#env, wave and rb to adjust envelope, waveform and reverb settings


import ctypes, sys, math, threading
PA = ctypes.CDLL('libpulse-simple.so')

math.tau = math.pi * 2.0

void_pointer = ctypes.POINTER(None)
int_pointer = ctypes.POINTER(ctypes.c_int)

pa_sample_format = ctypes.c_int
pa_sample_format.PA_SAMPLE_FLOAT32LE = ctypes.c_int(0x5)

pa_simple_write = PA.pa_simple_write

pa_simple_write.argtypes = [void_pointer, void_pointer, ctypes.c_int, int_pointer]


PA_STREAM_PLAYBACK = ctypes.c_int(1)

NULL = None



class pa_sample_format_t(ctypes.Structure):
	_fields_ = [
		('format', pa_sample_format),
		('rate', ctypes.c_uint32),
		('channels', ctypes.c_uint8),
	]



ss_v = pa_sample_format_t(pa_sample_format.PA_SAMPLE_FLOAT32LE, 48000, 2)
error = ctypes.c_int(0)


ss = ctypes.pointer(ss_v)

pa_simple_new  = PA.pa_simple_new

pa_simple_new.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_char_p, void_pointer, void_pointer, void_pointer, int_pointer]
pa_simple_new.restype = void_pointer

output = pa_simple_new(NULL, bytes(sys.argv[0], 'utf-8'), PA_STREAM_PLAYBACK, NULL, b"playback", ss, NULL, NULL, ctypes.pointer(error))

buflen = 32




SR = 48000.0

class mono_to_stereo:
	def __init__(self, target):
		self.target = target

	def __call__(self, data):
		for i in range(len(data)):
			self.target[i] = data[i], data[i]



class sine_wave:
	def __init__(self, freq, amp):
		self.rate = freq / SR
		self.amp = amp
		self.a = 0.0

	@property
	def freq(self):
		return self.rate * SR

	@freq.setter
	def freq(self, value):
		self.rate = value / SR

	def __call__(self, data):
		for i in range(len(data)):
			self.a += self.rate % math.tau			
			data[i] = math.sin(self.a)*self.amp[i]


class envelope:
	def __init__(self, attack, hold, decay, pause):
		self.attack = attack
		self.hold = hold
		self.decay = decay
		self.pause = pause

		self.mode = 1
		self.time_left = self.attack

	def __call__(self, data):

		tick = 1.0 / SR
		for i in range(len(data)):
			self.time_left -= tick
			if self.time_left <= 0:				

				if self.mode == 1:
					self.mode = 2
					self.time_left = self.hold
				elif self.mode == 2:
					self.mode = 3
					self.time_left = self.decay
				elif self.mode == 3:
					self.mode = 4
					self.time_left = self.pause
				elif self.mode == 4:
					self.mode = 1
					self.time_left = self.attack

			if self.mode == 1:
				if self.attack:
					data[i] = (self.attack - self.time_left) / self.attack
				else:
					data[i] = 1.0
			elif self.mode == 2:
				data[i] = 1.0
			elif self.mode == 3:
				if self.decay:
					data[i] = self.time_left / self.decay
				else:
					data[i] = 0.0

			elif self.mode == 4:
				data[i] = 0.0


class reverb:
	def __init__(self, size):
		self.size = size
		self.buffer = (ctypes.c_float * self.size)()
		self.pos = 0
		self.wet = 0.5
		self.dry = 0.5

	def __call__(self, buffer):

		for i in range(len(buffer)):
			buffer[i] = self.buffer[self.pos] * self.wet + buffer[i] * self.dry
			self.buffer[self.pos] = buffer[i]
			self.pos = (self.pos + 1) % self.size			
			

env = envelope(0.01, 0.2, 0.01, 1.5)
env_buffer =  (ctypes.c_float * buflen)()
wave = sine_wave(4200.0, env_buffer)

rb = reverb(int(0.25 * SR))

def sound_thread(output):

	output_buffer = (ctypes.c_float * 2 * buflen)()
	mono_buffer =  (ctypes.c_float * buflen)()
	

	mixer = mono_to_stereo(output_buffer)

	while 1:
		env(env_buffer)
		wave(mono_buffer)
		rb(mono_buffer)
		mixer(mono_buffer)

		pa_simple_write(output, output_buffer, ctypes.sizeof(output_buffer), ctypes.pointer(error))

t = threading.Thread(target=sound_thread, args=(output,))
t.start()