Gromov_IA/digital_microphone
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Инструменты ПК: прослушивание, перевод в WAV, список зависимостей

Browse Source 
- uart_audio.py: поток UART на звуковую карту ПК с подстройкой шага
  пересчёта по заполнению буфера, ФВЧ 20 Гц, --selftest;
- Bin_to_wav.py: записанный .bin -> WAV;
- requirements.txt: пакеты Python для всех скриптов.
This commit is contained in:
Gromov_IA
2026-10-02 14:21:07 +03:00
parent 2672f8b8a8
commit d6f3ee5ef8
3 changed files with 691 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/Bin_to_wav.py
+110
View File
@@ -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("<i4")
if WAV_BITS == 16:
frames = pcm.astype("<i2").tobytes()
elif WAV_BITS == 24:
# Младшие три байта каждого int32 little-endian
frames = pcm.view(np.uint8).reshape(-1, 4)[:, :3].tobytes()
else:
raise ValueError(f"WAV_BITS = {WAV_BITS}: поддерживаются 16 и 24")
with wave.open(output_filename, "wb") as w:
w.setnchannels(1)
w.setsampwidth(WAV_BITS // 8)
w.setframerate(FS)
w.writeframes(frames)
print(f"Готово! Файл сохранен:\n {output_filename}")
if __name__ == "__main__":
# Файл можно передать в командной строке, иначе -- выбор через окно
if len(sys.argv) > 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}")
tools/requirements.txt
+7
View File
@@ -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, вывод на звуковую карту
tools/uart_audio.py
+574
View File
@@ -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())