Apply pyaudio; Add reedsolo encode&decode; Add stop_event during send

amodem/async_reader.py

@@ -14,7 +14,7 @@ def __init__(self, stream, bufsize):
         self.queue = Queue()
         self.stop = threading.Event()
         args = (stream, bufsize, self.queue, self.stop)
-        self.thread = threading.Thread(target=AsyncReader._thread,
+        self.thread = threading.Thread(target=AsyncReader._thread, daemon=True,
                                        args=args, name='AsyncReader')
         self.thread.start()
         self.buf = b''

amodem/audio.py

@@ -2,13 +2,23 @@
 
 import ctypes
 import logging
+import platform
+import string
 import time
 
 log = logging.getLogger(__name__)
 
-
-class AudioError(Exception):
-    pass
+PLATFORM_IS_WIN = platform.system() == 'Windows'
+WIN_ERR_CODES = ['pa.paNoError', 'pa.paNotInitialized', 'pa.paUnanticipatedHostError', 'pa.paInvalidChannelCount',
+                 'pa.paInvalidSampleRate', 'pa.paInvalidDevice', 'pa.paInvalidFlag', 'pa.paSampleFormatNotSupported',
+                 'pa.paBadIODeviceCombination', 'pa.paInsufficientMemory', 'pa.paBufferTooBig', 'pa.paBufferTooSmall',
+                 'pa.paNullCallback', 'pa.paBadStreamPtr', 'pa.paTimedOut', 'pa.paInternalError',
+                 'pa.paDeviceUnavailable', 'pa.paIncompatibleHostApiSpecificStreamInfo', 'pa.paStreamIsStopped',
+                 'pa.paStreamIsNotStopped', 'pa.paInputOverflowed', 'pa.paOutputUnderflowed', 'pa.paHostApiNotFound',
+                 'pa.paInvalidHostApi', 'pa.paCanNotReadFromACallbackStream', 'pa.paCanNotWriteToACallbackStream',
+                 'pa.paCanNotReadFromAnOutputOnlyStream', 'pa.paCanNotWriteToAnInputOnlyStream',
+                 'pa.paIncompatibleStreamHostApi']
+WIN_ERR_CODE_MAP = {}
 
 
 class Interface:
@@ -18,28 +28,69 @@ def __init__(self, config, debug=False):
         self.streams = []
         self.lib = None
 
+        self.win_pyaudio = None
+        self.win_err_code_map = {}
+        self.win_func_names = []
+        self.win_mapped_func = {}
+
     def load(self, name):
-        self.lib = ctypes.CDLL(name)
+        if PLATFORM_IS_WIN:
+            import pyaudio as pa
+            self.lib = pa._portaudio
+            self.win_pyaudio = pa.PyAudio()
+            self.win_func_names = dir(self.lib)
+            for code_str in WIN_ERR_CODES:
+                WIN_ERR_CODE_MAP[eval(code_str)] = code_str[3:]
+            WIN_ERR_CODE_MAP[0] = b'Success'
+        else:
+            self.lib = ctypes.CDLL(name)
         assert self._error_string(0) == b'Success'
         version = self.call('GetVersionText', restype=ctypes.c_char_p)
         log.info('%s loaded', version)
         return self
 
     def _error_string(self, code):
-        return self.call('GetErrorText', code, restype=ctypes.c_char_p)
+        if PLATFORM_IS_WIN:
+            return WIN_ERR_CODE_MAP.get(code, "Unknown Error")
+        else:
+            return self.call('GetErrorText', code, restype=ctypes.c_char_p)
 
     def call(self, name, *args, **kwargs):
         assert self.lib is not None
-        func_name = f'Pa_{name}'
-        if self.debug:
-            log.debug('API: %s%s', name, args)
-        func = getattr(self.lib, func_name)
-        func.restype = kwargs.get('restype', self._error_check)
-        return func(*args)
+        if PLATFORM_IS_WIN:
+            if name in self.win_mapped_func:
+                func_name = self.win_mapped_func[name]
+            else:
+                func_name = ""
+                for c, ch in enumerate(name):
+                    if c > 0 and ch in string.ascii_uppercase:
+                        func_name += f"_{ch.lower()}"
+                    else:
+                        func_name += ch.lower()
+            if self.debug:
+                log.debug('API: %s%s', name, args)
+            if func_name in self.win_func_names:
+                func = getattr(self.lib, func_name)
+                if hasattr(func, "restype"):
+                    func.restype = kwargs.get('restype', self._error_check)
+                try:
+                    return func(*args)
+                except Exception as e:
+                    log.error(f"call [{name}], args [{args}], kwargs [{kwargs}]")
+                    raise e
+            else:
+                raise Exception("No such method", func_name)
+        else:
+            func_name = f'Pa_{name}'
+            if self.debug:
+                log.debug('API: %s%s', name, args)
+            func = getattr(self.lib, func_name)
+            func.restype = kwargs.get('restype', self._error_check)
+            return func(*args)
 
     def _error_check(self, res):
         if res != 0:
-            raise AudioError(res, self._error_string(res))
+            raise Exception(res, self._error_string(res))
 
     def __enter__(self):
         self.call('Initialize')
@@ -58,7 +109,6 @@ def player(self):
 
 
 class Stream:
-
     timer = time.time
 
     class Parameters(ctypes.Structure):
@@ -77,51 +127,78 @@ def __init__(self, interface, config, read=False, write=False):
         self.stream_callback = ctypes.c_void_p(None)
         self.bytes_per_sample = config.sample_size
         self.latency = float(config.latency)  # in seconds
-        self.bufsize = int(self.latency * config.Fs * self.bytes_per_sample)
+        self.bufsize = int(2 * self.latency * config.Fs * self.bytes_per_sample)
         assert config.bits_per_sample == 16  # just to make sure :)
 
         read = bool(read)
         write = bool(write)
         assert read != write  # don't support full duplex
 
-        direction = 'Input' if read else 'Output'
-        api_name = f'GetDefault{direction}Device'
-        index = interface.call(api_name, restype=ctypes.c_int)
-        self.params = Stream.Parameters(
-            device=index,               # choose default device
-            channelCount=1,             # mono audio
-            sampleFormat=0x00000008,    # 16-bit samples (paInt16)
-            suggestedLatency=self.latency,
-            hostApiSpecificStreamInfo=None)
-
-        self.interface.call(
-            'OpenStream',
-            ctypes.byref(self.stream),
-            ctypes.byref(self.params) if read else None,
-            ctypes.byref(self.params) if write else None,
-            ctypes.c_double(config.Fs),
-            ctypes.c_ulong(0),  # (paFramesPerBufferUnspecified)
-            ctypes.c_ulong(0),  # no flags (paNoFlag)
-            self.stream_callback,
-            self.user_data)
-
-        self.interface.streams.append(self)
-        self.interface.call('StartStream', self.stream)
+        if PLATFORM_IS_WIN:
+            import pyaudio as pa
+            arguments = {
+                'rate': int(config.Fs),
+                'channels': 1,
+                'format': pa.paInt16,
+                'input': read,
+                'output': write,
+                'frames_per_buffer': pa.paFramesPerBufferUnspecified,
+            }
+            self.stream = self.interface.win_pyaudio.open(**arguments)
+            self.interface.streams.append(self)
+            self.stream.start_stream()
+        else:
+            direction = 'Input' if read else 'Output'
+            api_name = f'GetDefault{direction}Device'
+            index = interface.call(api_name, restype=ctypes.c_int)
+            self.params = Stream.Parameters(
+                device=index,  # choose default device
+                channelCount=1,  # mono audio
+                sampleFormat=0x00000008,  # 16-bit samples (paInt16)
+                suggestedLatency=self.latency,
+                hostApiSpecificStreamInfo=None)
+
+            self.interface.call(
+                'OpenStream',
+                ctypes.byref(self.stream),
+                ctypes.byref(self.params) if read else None,
+                ctypes.byref(self.params) if write else None,
+                ctypes.c_double(config.Fs),
+                ctypes.c_ulong(0),  # (paFramesPerBufferUnspecified)
+                ctypes.c_ulong(0),  # no flags (paNoFlag)
+                self.stream_callback,
+                self.user_data)
+
+            self.interface.streams.append(self)
+            self.interface.call('StartStream', self.stream)
         self.start_time = self.timer()
         self.io_time = 0
 
     def close(self):
         if self.stream:
-            self.interface.call('StopStream', self.stream)
-            self.interface.call('CloseStream', self.stream)
+            if PLATFORM_IS_WIN:
+                self.stream.stop_stream()
+                self.stream.close()
+            else:
+                self.interface.call('StopStream', self.stream)
+                self.interface.call('CloseStream', self.stream)
             self.stream = None
 
     def read(self, size):
         assert size % self.bytes_per_sample == 0
-        buf = ctypes.create_string_buffer(size)
-        frames = ctypes.c_ulong(size // self.bytes_per_sample)
-        t0 = self.timer()
-        self.interface.call('ReadStream', self.stream, buf, frames)
+        if PLATFORM_IS_WIN:
+            class _tmp:
+                def __init__(self):
+                    self.raw = None
+
+            t0 = self.timer()
+            buf = _tmp()
+            buf.raw = self.stream.read((size // self.bytes_per_sample))
+        else:
+            buf = ctypes.create_string_buffer(size)
+            frames = ctypes.c_ulong(size // self.bytes_per_sample)
+            t0 = self.timer()
+            self.interface.call('ReadStream', self.stream, buf, frames)
         t1 = self.timer()
         self.io_time += (t1 - t0)
         if self.interface.debug:
@@ -132,6 +209,9 @@ def read(self, size):
     def write(self, data):
         data = bytes(data)
         assert len(data) % self.bytes_per_sample == 0
-        buf = ctypes.c_char_p(data)
-        frames = ctypes.c_ulong(len(data) // self.bytes_per_sample)
-        self.interface.call('WriteStream', self.stream, buf, frames)
+        if PLATFORM_IS_WIN:
+            self.stream.write(data, num_frames=(len(data) // self.bytes_per_sample))
+        else:
+            buf = ctypes.c_char_p(data)
+            frames = ctypes.c_ulong(len(data) // self.bytes_per_sample)
+            self.interface.call('WriteStream', self.stream, buf, frames)

amodem/config.py

@@ -8,6 +8,7 @@ class Configuration:
     Tsym = 0.001  # symbol duration [seconds]
     Npoints = 64
     frequencies = [1e3, 8e3]  # use 1..8 kHz carriers
+    negotiate_frequencies = [12e3]  # negotiate carrier
 
     # audio config
     bits_per_sample = 16
@@ -71,6 +72,7 @@ def __init__(self, **kwargs):
     24: Configuration(Fs=16e3, Npoints=16, frequencies=[1e3, 6e3]),
     28: Configuration(Fs=32e3, Npoints=16, frequencies=[3e3, 9e3]),
     32: Configuration(Fs=32e3, Npoints=16, frequencies=[2e3, 9e3]),
+    40: Configuration(Fs=32e3, Npoints=16, frequencies=[2e3, 11e3]),
     36: Configuration(Fs=32e3, Npoints=64, frequencies=[4e3, 9e3]),
     42: Configuration(Fs=32e3, Npoints=64, frequencies=[4e3, 10e3]),
     48: Configuration(Fs=32e3, Npoints=64, frequencies=[3e3, 10e3]),

amodem/detect.py

@@ -3,6 +3,7 @@
 import collections
 import itertools
 import logging
+import threading
 
 import numpy as np
 
@@ -14,7 +15,6 @@
 
 
 class Detector:
-
     COHERENCE_THRESHOLD = 0.9
 
     CARRIER_DURATION = sum(equalizer.prefix)
@@ -31,10 +31,12 @@ def __init__(self, config, pylab):
         self.max_offset = config.timeout * config.Fs
         self.plt = pylab
 
-    def _wait(self, samples):
+    def _wait(self, samples, stop_event: threading.Event = None):
         counter = 0
         bufs = collections.deque([], maxlen=self.maxlen)
         for offset, buf in common.iterate(samples, self.Nsym, index=True):
+            if stop_event is not None and stop_event.is_set():
+                raise StopIteration('Detector stop iteration by stop_event')
             if offset > self.max_offset:
                 raise ValueError('Timeout waiting for carrier')
             bufs.append(buf)
@@ -50,8 +52,8 @@ def _wait(self, samples):
 
         raise ValueError('No carrier detected')
 
-    def run(self, samples):
-        offset, bufs = self._wait(samples)
+    def run(self, samples, stop_event: threading.Event = None):
+        offset, bufs = self._wait(samples, stop_event=stop_event)
 
         length = (self.CARRIER_THRESHOLD - 1) * self.Nsym
         begin = offset - length
@@ -62,7 +64,7 @@ def run(self, samples):
 
         log.debug('Buffered %d ms of audio', len(bufs))
 
-        bufs = list(bufs)[-self.CARRIER_THRESHOLD-self.SEARCH_WINDOW:]
+        bufs = list(bufs)[-self.CARRIER_THRESHOLD - self.SEARCH_WINDOW:]
         n = self.SEARCH_WINDOW + self.CARRIER_DURATION - self.CARRIER_THRESHOLD
         trailing = list(itertools.islice(samples, n * self.Nsym))
         bufs.append(np.array(trailing))
@@ -86,7 +88,7 @@ def find_start(self, buf):
         signal = (2 ** 0.5) * signal / dsp.norm(signal)
 
         corr = np.abs(np.correlate(buf, signal))
-        norm_b = np.sqrt(np.correlate(np.abs(buf)**2, np.ones(len(signal))))
+        norm_b = np.sqrt(np.correlate(np.abs(buf) ** 2, np.ones(len(signal))))
         coeffs = np.zeros_like(corr)
         coeffs[norm_b > 0.0] = corr[norm_b > 0.0] / norm_b[norm_b > 0.0]