#!/usr/bin/env python3 """Прослушивание микрофона: поток UART -> звуковая карта -> джек. ЦАП здесь -- сама звуковая карта, внешнего железа не нужно. Кадры разбирает тот же FrameParser, что и запись, отсчёты 24 бит / 12 500 Гц пересчитываются в частоту устройства вывода и уходят на выход. Отдельный процесс от uart_live.py не для удобства, а по необходимости: COM-порт держит кто-то один. Либо смотреть, либо слушать, либо -- вообще без порта -- проиграть уже записанный .bin ключом --file. Два места, на которых такая программа обычно и ломается: * частоты не совпадают. 12 500 Гц отмеряет кварц платы, частоту вывода -- кварц звуковой карты; расхождение в десятки ppm за минуту накапливает миллисекунды и рано или поздно опустошает либо переполняет буфер. Поэтому шаг пересчёта не постоянный: его подстраивает регулятор по заполнению буфера, и постоянного отношения частот в программе нет нигде; * у микрофона заметная постоянная составляющая -- она съедает запас по шкале и даёт щелчок при старте, поэтому на входе стоит ФВЧ 20 Гц. Полоса та же, что у тракта, 5 кГц: звук телефонного качества -- это норма, а не признак поломки. Слушать лучше в наушниках, динамик рядом с микрофоном даёт самовозбуждение. py -3 tools/uart_audio.py --port COM10 py -3 tools/uart_audio.py --file live_01.bin --gain 12 py -3 tools/uart_audio.py --list """ import argparse import collections import sys import threading import time import numpy as np from scipy.signal import butter, lfilter, lfilter_zi from scipy.signal.windows import blackmanharris import uart_capture as uc try: import sounddevice as sd except Exception as _e: # пакета может не быть -- --selftest и --list живут без него sd = None _SD_ERR = _e FS_IN = uc.FS_DEFAULT # 12 500 Гц, задаёт кварц платы FS_CODE = float(uc.FS_CODE) # полная шкала 24 бит HP_HZ = 20.0 # ФВЧ: постоянная составляющая микрофона BUF_MS = 120.0 # запас буфера: перекрывает заминки Windows TRIM_K = 0.05 # П-регулятор шага, постоянная времени ~2.5 с TRIM_MAX = 0.01 # предел подстройки, 1% = 17 центов FILL_TAU = 0.5 # сглаживание оценки заполнения, с class Resampler: """Пересчёт частоты с произвольным и меняющимся на ходу шагом. Постоянного отношения мало: именно шагом компенсируется расхождение кварцев. Поэтому таблица фаз считается один раз, а дробная позиция чтения двигается тем шагом, который на этот блок назначил регулятор. """ TAPS = 32 # окно свёртки, в отсчётах входа PHASES = 2048 # квантование дробной задержки, спуры ниже -80 дБ OFF = TAPS // 2 - 1 # отводов слева от позиции чтения LOOK = TAPS - 1 - OFF # и справа: столько входа всегда лежит впрок BETA = 7.857 # окно Кайзера на 80 дБ def __init__(self): # Срез на Найквисте входа. Давить надо не полосу 6250...12500 Гц -- её # во входе нет, -- а образы, ближайший из них на 12500-5000 = 7500 Гц. # На переход 5000...7500 Гц 32 отводов хватает с запасом. mu = np.arange(self.PHASES) / self.PHASES d = mu[:, None] + self.OFF - np.arange(self.TAPS)[None, :] arg = np.maximum(0.0, 1.0 - (2.0 * d / self.TAPS) ** 2) bank = np.sinc(d) * (np.i0(self.BETA * np.sqrt(arg)) / np.i0(self.BETA)) # Нормировка построчно: иначе усиление гуляет от фазы к фазе, и # медленный дрейф шага слышен как дрожание громкости. self.bank = bank / bank.sum(axis=1, keepdims=True) self.buf = np.zeros(self.TAPS) self.pos = float(self.OFF) def feed(self, x): self.buf = np.concatenate((self.buf, x)) @property def pending(self): """Сколько отсчётов входа ещё не проиграно. Отводы справа не в счёт: они лежат всегда, и по ним конец потока не отличить от середины.""" return max(0.0, len(self.buf) - self.pos - self.LOOK) def avail(self, step): """Сколько отсчётов выхода можно выдать из того, что уже лежит.""" imax = len(self.buf) - 1 - self.LOOK return max(int((imax - self.pos) / step) + 1, 0) def pull(self, n, step): if n <= 0: return np.zeros(0) t = self.pos + step * np.arange(n) i = t.astype(np.int64) p = ((t - i) * self.PHASES).astype(np.int64) idx = i[:, None] - self.OFF + np.arange(self.TAPS)[None, :] y = (self.buf[idx] * self.bank[p]).sum(axis=1) self.pos = t[-1] + step drop = int(self.pos) - self.OFF if drop > 0: self.buf = self.buf[drop:] self.pos -= drop return y class Source(threading.Thread): """Приём отсчётов: из порта или из записанного .bin. Кладёт в очередь. ФВЧ стоит здесь, а не в звуковом обратном вызове: он работает на частоте входа, где считать вчетверо дешевле, и не в потоке реального времени. """ daemon = True def __init__(self, port=None, baud=921600, path=None, hp=HP_HZ): super().__init__() self.port, self.baud, self.path = port, baud, path self.q = collections.deque() self.lock = threading.Lock() self.n_queued = 0 self.stop_evt = threading.Event() self.parser = uc.FrameParser() self.err = None self.eof = False self.gaps = 0 self.hp_zi = None self.hp_b = self.hp_a = None if hp > 0: self.hp_b, self.hp_a = butter(2, hp, "highpass", fs=FS_IN) # --- очередь --- def put(self, y): if self.hp_b is not None and len(y): if self.hp_zi is None: # Начальное состояние по первому отсчёту: иначе постоянная # составляющая уезжает к нулю ступенькой, и это слышно щелчком self.hp_zi = lfilter_zi(self.hp_b, self.hp_a) * y[0] y, self.hp_zi = lfilter(self.hp_b, self.hp_a, y, zi=self.hp_zi) with self.lock: self.q.append(y) self.n_queued += len(y) def take(self, n): """Не больше n отсчётов; сколько есть, столько и отдаёт.""" out, got = [], 0 with self.lock: while self.q and got < n: c = self.q[0] if got + len(c) <= n: self.q.popleft() out.append(c) got += len(c) else: k = n - got out.append(c[:k]) self.q[0] = c[k:] got = n self.n_queued -= got return np.concatenate(out) if out else np.zeros(0) @property def queued(self): with self.lock: return self.n_queued # --- источники --- def run(self): try: self._run_file() if self.path else self._run_port() except Exception as e: self.err = str(e) self.eof = True def _run_port(self): ser = uc.open_port(self.port, self.baud) try: while not self.stop_evt.is_set(): chunk = ser.read(4096) if not chunk: continue for _cnt, _st, data, gap in self.parser.feed(chunk): if gap: # Как и в записи: дыру добиваем нулями, иначе звук # уедет по времени относительно источника self.gaps += gap self.put(np.zeros(gap * uc.SAMPLES_PER_FRAME)) self.put(np.asarray(uc.decode_samples(data), float) / FS_CODE) finally: ser.close() def _run_file(self): with open(self.path, "rb") as f: raw = f.read() a = np.frombuffer(raw[:len(raw) // 3 * 3], dtype=np.uint8).reshape(-1, 3).astype(np.int32) v = (a[:, 0] << 16) | (a[:, 1] << 8) | a[:, 2] y = np.where(v & 0x800000, v - 0x1000000, v) / FS_CODE # Файл отдаётся в темпе реального времени: иначе он влетел бы в буфер # целиком, и проверялся бы не тот режим, что на живом порту block = 250 t0 = time.perf_counter() for i in range(0, len(y), block): if self.stop_evt.is_set(): return self.put(y[i:i + block]) dt = t0 + (i + block) / FS_IN - time.perf_counter() if dt > 0: self.stop_evt.wait(dt) class Player: """Звуковой обратный вызов и регулятор шага пересчёта.""" def __init__(self, src, fs_out, channels, gain_db, target): self.src = src self.rs = Resampler() self.fs_out = fs_out self.nominal = FS_IN / fs_out self.channels = channels self.gain = 10.0 ** (gain_db / 20.0) self.target = target # желаемый запас, отсчётов входа self.fill_avg = target self.fill_min = target self.trim = 0.0 self.underruns = 0 self.resyncs = 0 self.clipped = 0 self.late = 0 self.peak = 0.0 def fill(self): return self.src.queued + self.rs.pending def callback(self, outdata, frames, _time, status): if status: self.late += 1 fill = self.fill() self.fill_min = min(self.fill_min, fill) if fill > 3 * self.target: # После долгой заминки выбросить дешевле, чем догонять регулятором: # 1% в секунду разгребал бы полсекунды задержки почти минуту self.resyncs += 1 self.src.take(int(fill - self.target)) fill = self.fill() self.fill_avg = fill # Регулятор ведём по сглаженному заполнению, а не по мгновенному: # порт и планировщик Windows отдают данные рывками в единицы # миллисекунд, и на мгновенном шаг гулял бы на слышимые доли процента, # хотя расхождение кварцев, которое и надо ловить, -- десятки ppm a = min(1.0, frames / (self.fs_out * FILL_TAU)) self.fill_avg += a * (fill - self.fill_avg) self.trim = float(np.clip(TRIM_K * (self.fill_avg - self.target) / self.target, -TRIM_MAX, TRIM_MAX)) step = self.nominal * (1.0 + self.trim) need = int(frames * step) + self.rs.TAPS + 2 - len(self.rs.buf) if need > 0: self.rs.feed(self.src.take(need)) n = min(frames, self.rs.avail(step)) y = self.rs.pull(n, step) if n < frames: self.underruns += frames - n y = np.concatenate((y, np.zeros(frames - n))) y = y * self.gain if len(y): self.peak = max(self.peak, float(np.abs(y).max())) n_clip = int(np.count_nonzero(np.abs(y) > 1.0)) if n_clip: self.clipped += n_clip y = np.clip(y, -1.0, 1.0) outdata[:] = np.repeat(y[:, None], self.channels, axis=1).astype(np.float32) def pick_device(name): """Без указания -- то устройство, что выбрано в Windows, но по возможности оно же через WASAPI: там поток идёт на родной частоте устройства и не проходит через ресемплер Windows, а пересчёт у нас свой. Просто взять устройство WASAPI по умолчанию нельзя: список WASAPI в PortAudio неполон, и джек может там не значиться вовсе. """ if name is not None: return int(name) if str(name).isdigit() else name dev = sd.default.device[1] try: want = sd.query_devices(dev)["name"][:31] # MME режет имена до 31 знака except Exception: return dev for i, d in enumerate(sd.query_devices()): if (d["max_output_channels"] > 0 and "WASAPI" in sd.query_hostapis(d["hostapi"])["name"] and d["name"][:31] == want): return i return dev def list_all(): from serial.tools import list_ports print("порты:") for p in list_ports.comports(): print(f" {p.device} {p.description}") if sd is None: print(f"\nустройства вывода: нет пакета sounddevice ({_SD_ERR})") return 0 print("\nустройства вывода:") apis = sd.query_hostapis() for i, d in enumerate(sd.query_devices()): if d["max_output_channels"] > 0: print(f" {i:3d} {d['name']} [{apis[d['hostapi']]['name']}] " f"{d['default_samplerate']:.0f} Гц") return 0 def run(args): if sd is None: print(f"нужен пакет sounddevice: py -3 -m pip install sounddevice\n({_SD_ERR})") return 2 dev = pick_device(args.device) info = sd.query_devices(dev, "output") fs_out = float(args.rate or info["default_samplerate"]) channels = min(2, int(info["max_output_channels"])) if fs_out < FS_IN: print(f"частота устройства {fs_out:.0f} Гц ниже входной: будет алиасинг, " f"задайте --rate") target = args.buffer / 1000.0 * FS_IN try: latency = float(args.latency) except ValueError: latency = args.latency src = Source(args.port, args.baud, args.file, args.hp) src.start() print(f"вход : {'файл ' + args.file if args.file else args.port}, " f"{FS_IN:.0f} Гц / 24 бита") print(f"выход : {info['name']}, {fs_out:.0f} Гц, " f"{channels} кан., буфер {args.buffer:.0f} мс") # Накопить запас до старта: иначе первые доли секунды идёт тишина, а # регулятор стартует с заведомо пустого буфера и зря разгоняет шаг t0 = time.time() while src.queued < target and not src.err and not src.eof and time.time() - t0 < 5.0: time.sleep(0.02) if src.err: print(f"источник: {src.err}") return 1 if src.queued == 0: print("данных нет: проверьте порт, скорость и питание платы") return 1 pl = Player(src, fs_out, channels, args.gain, target) try: stream = sd.OutputStream(device=dev, samplerate=fs_out, channels=channels, dtype="float32", latency=latency, callback=pl.callback) except Exception as e: print(f"устройство не открылось: {e}\nсписок выходов: --list, выбор: --device") return 1 t_start = time.time() try: with stream: while True: time.sleep(0.5) if src.err: print(f"\nисточник: {src.err}") return 1 if src.eof and pl.fill() < 2: break peak, pl.peak = pl.peak, 0.0 # Минимум за интервал, а не мгновенное: до недобора буфер # доводит именно провал, а его мгновенное чтение не поймает lo, pl.fill_min = pl.fill_min, pl.fill() db = 20.0 * np.log10(max(peak, 1e-9)) sys.stdout.write( f"\r{time.time() - t_start:6.1f} с буфер {pl.fill() / FS_IN * 1000:5.1f}" f" мс (мин {lo / FS_IN * 1000:5.1f})" f" шаг {pl.trim * 1e6:+6.0f} ppm пик {db:6.1f} dBFS" f" кадры {src.parser.frames_ok} CRC {src.parser.crc_errors}" f" дыры {src.gaps} недобор {pl.underruns}" f" клип {pl.clipped} сброс {pl.resyncs} опоздания {pl.late} ") sys.stdout.flush() except KeyboardInterrupt: pass finally: src.stop_evt.set() print() return 0 def _peak_hz(sp, f): """Пик спектра с параболической интерполяцией: корзина тут 6 Гц, и без интерполяции числа в отчёте выглядят промахом, хотя промаха нет.""" k = int(np.argmax(sp)) if 0 < k < len(sp) - 1: a, b, c = np.log(sp[k - 1:k + 2] + 1e-30) return k, float(f[k] + 0.5 * (a - c) / (a - 2 * b + c) * (f[1] - f[0])) return k, float(f[k]) def selftest(): """Полная проверка тракта без порта и без звука.""" import struct # --- разбор кадров: формат общий с записью --- parser = uc.FrameParser() got = [] for k in range(200): s = [int(4_000_000 * np.sin(2 * np.pi * 1000 * (k * 50 + i) / FS_IN)) for i in range(uc.SAMPLES_PER_FRAME)] data = b"".join(struct.pack(">i", v & 0xFFFFFF)[1:] for v in s) payload = struct.pack(">HB", k, 0) + data for _c, _st, d, _g in parser.feed(uc.SYNC + payload + struct.pack(">H", uc.crc16_ccitt(payload))): got.extend(np.asarray(uc.decode_samples(d), float) / FS_CODE) assert parser.frames_ok == 200 and parser.crc_errors == 0 x = np.asarray(got) assert len(x) == 200 * uc.SAMPLES_PER_FRAME print(f"кадры : {parser.frames_ok}, отсчётов {len(x)}, " f"размах {x.min():+.3f}...{x.max():+.3f}") # --- пересчёт 12500 -> 48000: частота, уровень, паразитные составляющие --- fs_out = 48000.0 step = FS_IN / fs_out rs = Resampler() rs.feed(x) y = rs.pull(int(len(x) / step) - 40, step) # Окно Блэкмена-Харриса, а не Ханна: у Ханна боковые лепестки сами по себе # около -60 дБ и меряться будет окно, а не пересчёт z = y[256:256 + 8192] sp = np.abs(np.fft.rfft(z * blackmanharris(len(z)))) f = np.fft.rfftfreq(len(z), 1 / fs_out) k, f_pk = _peak_hz(sp, f) assert abs(f_pk - 1000.0) < 1.0, f_pk amp = np.abs(y[256:]).max() assert abs(amp - x.max()) < 0.02 * x.max(), (amp, x.max()) spur = sp.copy() spur[max(k - 12, 0):k + 13] = 0.0 j = int(np.argmax(spur)) db = 20.0 * np.log10(spur[j] / sp[k]) assert db < -70.0, (f[j], db) print(f"пересчёт : пик {f_pk:.1f} Гц, уровень {amp:.4f} против " f"{x.max():.4f}, паразитные {db:.1f} дБ на {f[j]:.0f} Гц") # --- непрерывность на границах блоков: в звуке разрыв слышен щелчком --- rs2 = Resampler() parts = [] pos = 0 while pos < len(x): rs2.feed(x[pos:pos + 137]) pos += 137 parts.append(rs2.pull(min(97, rs2.avail(step)), step)) yb = np.concatenate(parts) n = min(len(y), len(yb)) # На один шаг фазовой таблицы разойтись они вправе: позиция чтения копится # по-разному, и индекс фазы может отличаться на единицу. Слышимого это не # даёт, поэтому проверяется другое -- что на стыках нет скачка, то есть # приращение выхода не больше, чем у того же сигнала, посчитанного разом. bound = x.max() * 2 * np.pi * 1000 / FS_IN / Resampler.PHASES err = float(np.abs(y[:n] - yb[:n]).max()) assert err < 2 * bound, (err, bound) jump = float(np.abs(np.diff(yb[300:n])).max()) smooth = float(np.abs(np.diff(y[300:n])).max()) assert jump < smooth * 1.05, (jump, smooth) print(f"блоки : {len(parts)} кусков, расхождение {err:.1e} " f"при пороге {2 * bound:.1e}, скачок {jump:.4f} против {smooth:.4f}") # --- шаг действительно тянет частоту: этим и компенсируется дрейф --- rs3 = Resampler() rs3.feed(x) y3 = rs3.pull(int(len(x) / (step * 1.01)) - 40, step * 1.01) z3 = y3[256:256 + 8192] sp3 = np.abs(np.fft.rfft(z3 * blackmanharris(len(z3)))) _k3, f3 = _peak_hz(sp3, np.fft.rfftfreq(len(z3), 1 / fs_out)) assert abs(f3 - 1010.0) < 1.0, f3 print(f"подстройка : шаг +1% дал {f3:.1f} Гц вместо 1000") # --- регулятор: пустой буфер замедляет чтение, полный ускоряет --- target = 1500.0 blk = np.zeros((512, 2), dtype=np.float32) src = Source(hp=0.0) pl = Player(src, fs_out, 2, 0.0, target) pl.callback(blk, 512, None, None) lo = pl.trim assert pl.underruns > 500, "пустой буфер обязан дать недобор, а не тишину молча" src.put(np.zeros(3000)) pl.callback(blk, 512, None, None) hi = pl.trim assert lo < 0 < hi, (lo, hi) # Завал буфера разгребается выбросом, а не подстройкой: 1% в секунду # тянул бы полсекунды задержки почти минуту src.put(np.zeros(int(10 * target))) pl.callback(blk, 512, None, None) assert pl.resyncs == 1, pl.resyncs assert abs(pl.fill() - target) < 0.2 * target, pl.fill() print(f"регулятор : пусто {lo * 1e6:+.0f} ppm, полно {hi * 1e6:+.0f} ppm, " f"недобор {pl.underruns}, сброс {pl.resyncs} до {pl.fill():.0f} " f"из {target:.0f}") # --- ФВЧ: постоянная составляющая уходит, синус остаётся --- s2 = Source(hp=HP_HZ) s2.put(0.25 + 0.1 * np.sin(2 * np.pi * 1000 * np.arange(12500) / FS_IN)) d = s2.take(12500) assert abs(d[-2500:].mean()) < 1e-3, d[-2500:].mean() assert abs(np.abs(d[-2500:]).max() - 0.1) < 0.005 print(f"ФВЧ : постоянная {d[-2500:].mean():+.1e}, " f"размах {np.abs(d[-2500:]).max():.4f} из 0.1") if sd is None: print(f"устройство : нет пакета sounddevice ({_SD_ERR})") else: try: dev = pick_device(None) info = sd.query_devices(dev, "output") with sd.OutputStream(device=dev, dtype="float32", channels=min(2, int(info["max_output_channels"])), samplerate=float(info["default_samplerate"])): pass print(f"устройство : {info['name']}, " f"{info['default_samplerate']:.0f} Гц -- открывается") except Exception as e: print(f"устройство : не открылось ({e})") print("\nвсё сошлось") return 0 def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--port", help="например COM10") ap.add_argument("--baud", type=int, default=921600) ap.add_argument("--file", help="проиграть записанный .bin вместо порта") ap.add_argument("--device", help="устройство вывода: номер или часть имени") ap.add_argument("--rate", type=float, help="частота вывода, по умолчанию своя у устройства") ap.add_argument("--gain", type=float, default=0.0, help="усиление, дБ") ap.add_argument("--buffer", type=float, default=BUF_MS, help="запас буфера, мс") ap.add_argument("--latency", default="high", help="запрос к PortAudio: low, high или секунды") ap.add_argument("--hp", type=float, default=HP_HZ, help="срез ФВЧ, Гц; 0 -- выключить") ap.add_argument("--list", action="store_true", help="порты и устройства вывода") ap.add_argument("--selftest", action="store_true") args = ap.parse_args() if args.list: return list_all() if args.selftest: return selftest() if not args.port and not args.file: ap.error("нужен --port или --file (список портов и устройств: --list)") return run(args) if __name__ == "__main__": sys.exit(main())