diff --git a/tools/Bin_to_wav.py b/tools/Bin_to_wav.py new file mode 100644 index 0000000..a6be95c --- /dev/null +++ b/tools/Bin_to_wav.py @@ -0,0 +1,110 @@ +import numpy as np +import wave +import sys +import tkinter as tk +from tkinter import filedialog +import os + +# ================= НАСТРОЙКИ ================= +FS = 12500 # Частота дискретизации (должна совпадать с записью) +ADC_BITS = 24 # Разрядность исходных данных + +# 📢 КОЭФФИЦИЕНТ УСИЛЕНИЯ (GAIN) +# 1.0 = без изменений +# 10.0 = усилить в 10 раз (20 дБ) +# 100.0 = усилить в 100 раз (40 дБ) +GAIN = 5 + +# Разрядность WAV. Пишется обычный PCM: float-WAV открывают далеко не все +# программы, а 64-битный float, который получался раньше, -- почти никакие. +# 16 -- открывается везде; 24 -- полная разрядность записи +WAV_BITS = 16 + + +def convert_bin_to_wav(input_filename, output_filename, gain): + print(f"Чтение файла: {os.path.basename(input_filename)}...") + + # --- 1. Чтение 24-битного Big Endian файла --- + # (Используем ту же логику чтения, что и в анализаторе) + raw = np.fromfile(input_filename, dtype=np.uint8) + n_samples = len(raw) // 3 + raw = raw[:n_samples * 3] + data = raw.reshape(-1, 3) + + # Сборка 3 байт в int32 + samples = (data[:, 0].astype(np.int32) << 16) | \ + (data[:, 1].astype(np.int32) << 8) | \ + data[:, 2].astype(np.int32) + + # Обработка знака (Two's complement) + mask = (samples & 0x800000) != 0 + samples[mask] -= 0x1000000 + + # --- 2. Нормализация (перевод в диапазон -1.0 ... 1.0) --- + # Сразу float64: numpy 1.x всё равно приводит к нему при делении на 8388607 + max_val = 2 ** (ADC_BITS - 1) - 1 # 8,388,607 + audio_float = samples.astype(np.float64) / max_val + + # Убираем постоянную составляющую (DC Offset), чтобы динамик не "выпирало" + audio_float -= np.mean(audio_float) + + # --- 3. Применение усиления --- + print(f"Применяем усиление: x{gain}") + audio_float *= gain + + # --- 4. Проверка на клиппинг (перегрузку) --- + max_amp = np.max(np.abs(audio_float)) + if max_amp > 1.0: + clipped_samples = np.sum(np.abs(audio_float) > 1.0) + percent_clipped = (clipped_samples / len(audio_float)) * 100 + print(f"ВНИМАНИЕ: Перегрузка сигнала! Пик: {max_amp:.2f}") + print(f" Будет обрезано {percent_clipped:.2f}% сэмплов (Hard Clipping).") + print(" Рекомендуется уменьшить GAIN.") + + # Обрезаем все, что выше 1.0 или ниже -1.0 + audio_float = np.clip(audio_float, -1.0, 1.0) + else: + print(f"Сигнал в норме. Пиковый уровень: {max_amp:.2f} (из 1.0)") + + # --- 5. Сохранение WAV (PCM) --- + print(f"Сохранение в WAV ({WAV_BITS}-bit PCM, {FS} Гц)...") + full = 2 ** (WAV_BITS - 1) - 1 + pcm = np.round(audio_float * full).astype(" 1: + input_path = sys.argv[1] + else: + root = tk.Tk() + root.withdraw() + + input_path = filedialog.askopenfilename( + title="Выберите .bin файл для конвертации", + filetypes=[("Binary files", "*.bin")] + ) + + if input_path: + # Формируем имя выходного файла + folder = os.path.dirname(input_path) + name = os.path.basename(input_path).rsplit('.', 1)[0] + output_path = os.path.join(folder, f"{name}_gain{GAIN}.wav") + + try: + convert_bin_to_wav(input_path, output_path, GAIN) + except Exception as e: + print(f"Ошибка при конвертации: {e}") diff --git a/tools/requirements.txt b/tools/requirements.txt new file mode 100644 index 0000000..998855e --- /dev/null +++ b/tools/requirements.txt @@ -0,0 +1,7 @@ +# Скрипты ПК для стенда цифрового микрофона. +# Установка: python -m pip install -r tools/requirements.txt +pyserial>=3.5 # uart_capture, uart_live, uart_audio +numpy>=1.24 # все скрипты +scipy>=1.10 # gen_fir_coefs, uart_audio +matplotlib>=3.7 # uart_live, окно на TkAgg +sounddevice>=0.4 # uart_audio, вывод на звуковую карту diff --git a/tools/uart_audio.py b/tools/uart_audio.py new file mode 100644 index 0000000..400fd68 --- /dev/null +++ b/tools/uart_audio.py @@ -0,0 +1,574 @@ +#!/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())