Initial commit

audio_python/pyalsa.py

@@ -0,0 +1,73 @@
+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)