Gromov_IA
d6f3ee5ef8
- uart_audio.py: поток UART на звуковую карту ПК с подстройкой шага пересчёта по заполнению буфера, ФВЧ 20 Гц, --selftest; - Bin_to_wav.py: записанный .bin -> WAV; - requirements.txt: пакеты Python для всех скриптов.
575 lines
27 KiB
Python
575 lines
27 KiB
Python
#!/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())
|