74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
import ctypes
|
|
|
|
"""
|
|
Quick and dirty wrapper for ALSA.
|
|
"""
|
|
|
|
c_void_pp = ctypes.POINTER(ctypes.c_void_p)
|
|
_alsa = ctypes.cdll.LoadLibrary("libasound.so.2")
|
|
|
|
_alsa.snd_pcm_open.restype = ctypes.c_int
|
|
_alsa.snd_pcm_open.argtypes = [c_void_pp, ctypes.c_char_p, ctypes.c_int, ctypes.c_int]
|
|
|
|
_alsa.snd_pcm_set_params.restype = ctypes.c_int
|
|
_alsa.snd_pcm_set_params.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.c_uint, ctypes.c_uint, ctypes.c_int, ctypes.c_uint]
|
|
SND_PCM_FORMAT_S16_LE = 2
|
|
SND_PCM_ACCESS_RW_INTERLEAVED = 3
|
|
|
|
# snd_pcm_sframes_t snd_pcm_writei(snd_pcm_t *pcm, const void *buffer, snd_pcm_uframes_t size);
|
|
_alsa.snd_pcm_writei.restype = ctypes.c_long
|
|
_alsa.snd_pcm_writei.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_ulong]
|
|
|
|
# int snd_pcm_drain(snd_pcm_t *pcm);
|
|
_alsa.snd_pcm_drain.restype = ctypes.c_int
|
|
_alsa.snd_pcm_drain.argtypes = [ctypes.c_void_p]
|
|
|
|
|
|
class Alsa:
|
|
def __init__(self, sample_rate, latency):
|
|
self.sample_rate = sample_rate
|
|
self.handle = ctypes.c_void_p()
|
|
ret = _alsa.snd_pcm_open(ctypes.byref(self.handle), b"default", 0, 0)
|
|
assert ret >= 0, "snd_pcm_open failed"
|
|
ret = _alsa.snd_pcm_set_params(self.handle, SND_PCM_FORMAT_S16_LE, SND_PCM_ACCESS_RW_INTERLEAVED, 1, sample_rate, 1, latency)
|
|
|
|
def write(self, samples):
|
|
BufferType = ctypes.c_int16 * len(samples)
|
|
buf = BufferType(*samples)
|
|
ret = _alsa.snd_pcm_writei(self.handle, buf, len(samples))
|
|
assert ret == len(samples), "snd_pcm_writei failed"
|
|
return ret
|
|
|
|
def drain(self):
|
|
ret = _alsa.snd_pcm_drain(self.handle)
|
|
assert ret >= 0
|
|
|
|
|
|
#_alsa.snd_seq_open.argtypes = [c_void_pp, ctypes.c_char_p, ctypes.c_int, ctypes.c_int]
|
|
|
|
class AlsaSeq:
|
|
def __init__(self):
|
|
self.handle = ctypes.c_void_p()
|
|
ret = _alsa.snd_seq_open(ctypes.byref(self.handle), b"default", 2, 0)
|
|
if ret < 0:
|
|
raise RuntimeError("snd_seq_open failed")
|
|
ret = _alsa.snd_seq_set_client_name(self.handle, b"Midi Listener");
|
|
assert ret >= 0, "snd_seq_set_client_name failed"
|
|
self.port = _alsa.snd_seq_create_simple_port(self.handle, b"listen:in", 2 | 64, 1 << 20)
|
|
assert self.port >= 0, "snd_seq_create_simple_port failed"
|
|
print(self.handle, self.port)
|
|
err = _alsa.snd_seq_connect_from(self.handle, self.port, 20, 0)
|
|
assert err >= 0, "snd_seq_connect_from failed"
|
|
|
|
def get_event(self):
|
|
evt = ctypes.c_void_p()
|
|
err = _alsa.snd_seq_event_input(self.handle, ctypes.byref(evt))
|
|
assert err >= 0, "snd_seq_event_input failed"
|
|
return evt
|
|
|
|
if __name__ == "__main__":
|
|
seq = AlsaSeq()
|
|
while True:
|
|
evt = seq.get_event()
|
|
print(evt)
|