#!/usr/bin/env python3 """Живой просмотр потока с цифрового микрофона: осциллограмма, спектр, счётчики. Нужен, чтобы глазами проконтролировать сигнал до того, как писать запись: видно и форму, и шум, и наводки, и целостность кадров. Разбор кадров общий с uart_capture.py -- формат описан там в одном месте. Запись сюда встроена не для удобства, а по необходимости: COM-порт держит один процесс, и параллельно запустить uart_capture.py на том же порту нельзя. Управление -- кнопками внизу окна, те же действия продублированы клавишами: r запись d убрать постоянную составляющую a автомасштаб s спектр: плотность или амплитуда f пауза q выход Зум и панорама -- штатной панелью инструментов matplotlib. Автомасштаб при этом выключается сам, иначе он затирал бы выбранный масштаб на каждом кадре. """ import argparse import collections import sys import threading import time import numpy as np import matplotlib matplotlib.use("TkAgg") import matplotlib.pyplot as plt from matplotlib.widgets import Button import uart_capture as uc FS_CODE = uc.FS_CODE # полная шкала, шкала тракта -- dBFS NFFT = 8192 DB_FLOOR = 160.0 # на сколько дБ вниз растягиваем ось спектра def decode(data: bytes) -> np.ndarray: """150 байт кадра -> 50 знаковых отсчётов.""" a = np.frombuffer(data, dtype=np.uint8).reshape(-1, 3).astype(np.int32) v = (a[:, 0] << 16) | (a[:, 1] << 8) | a[:, 2] return np.where(v & 0x800000, v - 0x1000000, v) def minmax_decimate(y, buckets): """Прореживание для отрисовки с сохранением огибающей: на каждый столбец пикселей берём минимум и максимум, иначе выбросы шума пропадут.""" n = len(y) if n <= buckets * 2: return np.arange(n), y step = n // buckets blocks = y[:step * buckets].reshape(buckets, step) idx = np.arange(buckets) * step x = np.empty(buckets * 2) v = np.empty(buckets * 2) x[0::2], x[1::2] = idx, idx + step - 1 v[0::2], v[1::2] = blocks.min(axis=1), blocks.max(axis=1) return x, v class Reader(threading.Thread): """Читает порт, разбирает кадры, копит отсчёты для отрисовки.""" daemon = True def __init__(self, port, baud, depth): super().__init__() self.port, self.baud = port, baud self.samples = collections.deque(maxlen=depth) self.lock = threading.Lock() self.stop_evt = threading.Event() self.parser = uc.FrameParser() self.err = None self.bytes_total = 0 self.t0 = None self.rec_path = None # запрос от интерфейса: путь или None self.rec_active = None # что пишется сейчас self.rec_samples = 0 def run(self): try: ser = uc.open_port(self.port, self.baud) except Exception as e: self.err = str(e) return self.t0 = time.time() fraw = fbin = None try: while not self.stop_evt.is_set(): # Файлы открывает и закрывает только тот поток, что в них пишет if self.rec_path != self.rec_active: if fraw: fraw.close(); fbin.close() fraw = fbin = None if self.rec_path: fraw = open(self.rec_path + ".raw", "wb") fbin = open(self.rec_path + ".bin", "wb") self.rec_samples = 0 self.rec_active = self.rec_path chunk = ser.read(4096) if not chunk: continue self.bytes_total += len(chunk) if fraw: fraw.write(chunk) for _cnt, _status, data, gap in self.parser.feed(chunk): if fbin: if gap: fbin.write(b"\x00" * (gap * uc.SAMPLES_PER_FRAME * 3)) self.rec_samples += gap * uc.SAMPLES_PER_FRAME fbin.write(data) self.rec_samples += uc.SAMPLES_PER_FRAME with self.lock: self.samples.extend(decode(data)) except Exception as e: self.err = str(e) finally: if fraw: fraw.close(); fbin.close() ser.close() def snapshot(self): with self.lock: return np.fromiter(self.samples, dtype=np.int32, count=len(self.samples)) class Live: def __init__(self, reader, fs, window_s, out_base): self.rd = reader self.fs = fs self.out_base = out_base self.window_s = window_s self.sub_dc = False self.frozen = False self.psd_mode = True self.autoscale = True self.rec_n = 0 self.fig = plt.figure(figsize=(11, 8)) self.fig.canvas.manager.set_window_title("Цифровой микрофон -- живой просмотр") self.ax_t = self.fig.add_axes([0.08, 0.50, 0.89, 0.33]) self.ax_f = self.fig.add_axes([0.08, 0.14, 0.89, 0.27]) self.ln_t, = self.ax_t.plot([], [], lw=0.8) self.ax_t.set_xlabel("время, с") self.ax_t.set_ylabel("доли шкалы") self.ax_t.grid(alpha=0.3) self.ax_t.set_xlim(-window_s, 0) self.ln_f, = self.ax_f.plot([], [], lw=0.8) self.ax_f.set_xlabel("частота, Гц") self.ax_f.set_ylabel("dBFS/√Гц") self.ax_f.grid(alpha=0.3, which="both") # Ось частоты логарифмическая, поэтому нижний предел строго больше нуля self.ax_f.set_xscale("log") self.ax_f.set_xlim(1, fs / 2) self.ax_f.set_ylim(-DB_FLOOR, 0) self.txt = self.fig.text(0.01, 0.985, "", family="monospace", fontsize=9, va="top") self.buttons = [] self.b_rec = self._button(0.08, "● Запись", self.toggle_rec) self.b_auto = self._button(0.24, "Автомасштаб: вкл", self.toggle_auto) self.b_dc = self._button(0.40, "Как есть", self.toggle_dc) self.b_sp = self._button(0.56, "Спектр: плотность", self.toggle_sp) self.b_frz = self._button(0.72, "Пауза", self.toggle_frz) self.fig.canvas.mpl_connect("key_press_event", self.on_key) def _button(self, x, label, cb): ax = self.fig.add_axes([x, 0.025, 0.15, 0.05]) b = Button(ax, label) b.on_clicked(cb) self.buttons.append(b) return b # --- переключатели --- def toggle_rec(self, _=None): if self.rd.rec_path: self.rd.rec_path = None self.b_rec.label.set_text("● Запись") self.b_rec.ax.set_facecolor("0.85") else: self.rec_n += 1 self.rd.rec_path = f"{self.out_base}_{self.rec_n:02d}" self.b_rec.label.set_text("■ Стоп") self.b_rec.ax.set_facecolor("#ff9c9c") def toggle_auto(self, _=None): self.autoscale = not self.autoscale self.b_auto.label.set_text( "Автомасштаб: вкл" if self.autoscale else "Автомасштаб: выкл") def toggle_dc(self, _=None): self.sub_dc = not self.sub_dc self.b_dc.label.set_text("Без постоянной" if self.sub_dc else "Как есть") def toggle_sp(self, _=None): self.psd_mode = not self.psd_mode self.b_sp.label.set_text( "Спектр: плотность" if self.psd_mode else "Спектр: амплитуда") def toggle_frz(self, _=None): self.frozen = not self.frozen self.b_frz.label.set_text("Продолжить" if self.frozen else "Пауза") def on_key(self, ev): {"q": lambda: plt.close(self.fig), "r": self.toggle_rec, "a": self.toggle_auto, "d": self.toggle_dc, "s": self.toggle_sp, "f": self.toggle_frz}.get(ev.key, lambda: None)() def _toolbar_busy(self): """Пользователь работает лупой или рукой -- пределы осей не трогаем.""" tb = getattr(self.fig.canvas, "toolbar", None) return bool(getattr(tb, "mode", "")) if tb else False # --- отрисовка --- def update(self, _=None): if self._toolbar_busy() and self.autoscale: self.toggle_auto() if self.frozen: return self.ln_t, self.ln_f y = self.rd.snapshot() if len(y) < NFFT: self.txt.set_text(f"ждём данные... принято {len(y)} отсчётов" + (f"\nОШИБКА: {self.rd.err}" if self.rd.err else "")) return self.ln_t, self.ln_f v = y / FS_CODE mean = v.mean() vv = v - mean if self.sub_dc else v x, d = minmax_decimate(vv, 1200) self.ln_t.set_data((x - len(vv)) / self.fs, d) seg = v[-NFFT:] - v[-NFFT:].mean() w = np.hanning(NFFT) sp = np.abs(np.fft.rfft(seg * w)) ** 2 f = np.fft.rfftfreq(NFFT, 1 / self.fs) # Логарифм от нуля недопустим, поэтому подрезаем снизу ничтожным уровнем if self.psd_mode: psd = sp / ((w ** 2).sum() * self.fs) * 2 self.ln_f.set_data(f[1:], 10 * np.log10(np.maximum(psd[1:], 1e-30))) self.ax_f.set_ylabel("dBFS/√Гц") else: amp = np.sqrt(sp[1:]) / w.sum() * 2 self.ln_f.set_data(f[1:], 20 * np.log10(np.maximum(amp, 1e-15))) self.ax_f.set_ylabel("dBFS") if self.autoscale: lo, hi = d.min(), d.max() pad = max((hi - lo) * 0.1, 1e-7) self.ax_t.set_ylim(lo - pad, hi + pad) self.ax_t.set_xlim(-len(vv) / self.fs, 0) # Пределы спектра считаем сами. relim() с autoscale_view() здесь # не годятся: лупа снимает у оси внутренний флаг autoscale_on, # после чего эта пара молча перестаёт работать навсегда. yd = np.asarray(self.ln_f.get_ydata()) if len(yd): top = yd.max() + 10 bot = max(np.percentile(yd, 1) - 10, top - DB_FLOOR) self.ax_f.set_ylim(bot, top) self.ax_f.set_xlim(1, self.fs / 2) p = self.rd.parser el = max(time.time() - (self.rd.t0 or time.time()), 1e-3) rec = (f"ЗАПИСЬ {self.rd.rec_active} {self.rd.rec_samples} отсч." if self.rd.rec_active else "не пишем") st = ("чисто" if p.status_flags == 0 else f"FIFO={p.status_flags & 1} SAT={(p.status_flags >> 1) & 1}") peak = max(abs(int(y.max())), abs(int(y.min()))) self.txt.set_text( f"смещение {uc.dbfs(mean * FS_CODE):+7.1f} dBFS " f"СКО {y.std():8.1f} код ({uc.dbfs(y.std()):+7.1f} dBFS) " f"пик {peak:8d} код ({uc.dbfs(peak):+7.1f} dBFS)\n" f"кадров {p.frames_ok:<8d} CRC {p.crc_errors:<5d} потеряно {p.padded_frames:<6d} " f"ресинхр {p.resyncs:<5d} STATUS {st}\n" f"канал {self.rd.bytes_total / el / 1000:5.1f} кБ/с " f"кадров/с {p.frames_ok / el:5.1f} окно {len(vv) / self.fs:.2f} с " f"автомасштаб {'вкл' if self.autoscale else 'ВЫКЛ'} {rec}") return self.ln_t, self.ln_f def selftest(): """Прогон конвейера и всех веток отрисовки без порта и без окна.""" import struct import types matplotlib.use("Agg", force=True) parser = uc.FrameParser() samples = [] for k in range(400): s = [int(8000 * np.sin(2 * np.pi * 300 * (k * 50 + i) / 12500)) 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 frame = uc.SYNC + payload + struct.pack(">H", uc.crc16_ccitt(payload)) for _c, _st, d, _g in parser.feed(frame): samples.extend(decode(d)) y = np.asarray(samples, dtype=np.int32) assert parser.frames_ok == 400 and parser.crc_errors == 0 assert len(y) == 400 * uc.SAMPLES_PER_FRAME xs, dd = minmax_decimate(y, 500) assert len(xs) == 1000 and dd.max() == y.max() and dd.min() == y.min() sp = np.abs(np.fft.rfft((y[:NFFT] - y[:NFFT].mean()) * np.hanning(NFFT))) peak = np.fft.rfftfreq(NFFT, 1 / 12500.0)[np.argmax(sp)] assert abs(peak - 300) < 3, peak print(f"конвейер: 400 кадров, {len(y)} отсчётов, пик спектра {peak:.1f} Гц") rd = types.SimpleNamespace( parser=parser, bytes_total=390000, t0=time.time(), rec_path=None, rec_active=None, rec_samples=0, err=None, snapshot=lambda: np.zeros(10, dtype=np.int32)) live = Live(rd, 12500.0, 2.0, "live") # Рисуем именно холст, а не только считаем: логарифмическая шкала падает # на отрисовке, а не на подготовке данных. Пустое состояние -- первым. live.update() live.fig.canvas.draw() print(" отрисовка без данных -- ок") rd.snapshot = lambda: y live.update() live.fig.canvas.draw() print(" отрисовка с данными -- ок") lim0 = live.ax_t.get_ylim() for name, fn in (("автомасштаб", live.toggle_auto), ("постоянная", live.toggle_dc), ("спектр", live.toggle_sp), ("пауза", live.toggle_frz)): fn() live.update() live.fig.canvas.draw() print(f" переключение: {name} -- ок") live.toggle_frz() # С выключенным автомасштабом пределы обязаны остаться нашими live.ax_t.set_ylim(-1.0, 1.0) live.update() assert live.ax_t.get_ylim() == (-1.0, 1.0), "автомасштаб затирает зум" print(" зум при выключенном автомасштабе сохраняется -- ок") live.toggle_auto() live.update() assert live.ax_t.get_ylim() != (-1.0, 1.0) assert live.ax_t.get_ylim() == lim0 print(" автомасштаб обратно включается -- ок") # Спектр обязан следовать за уровнем сигнала f_small = live.ax_f.get_ylim() rd.snapshot = lambda: y * 100 live.update() f_big = live.ax_f.get_ylim() assert f_big[1] > f_small[1] + 30, (f_small, f_big) # сто раз -- это +40 дБ print(" спектр масштабируется за сигналом -- ок") # Лупа снимает у оси внутренний флаг autoscale_on -- после этого # масштабирование обязано продолжать работать live.ax_f.set_autoscaley_on(False) live.ax_t.set_autoscaley_on(False) rd.snapshot = lambda: y live.update() assert live.ax_f.get_ylim()[1] < f_big[1] - 30, live.ax_f.get_ylim() assert live.ax_t.get_ylim() == lim0 print(" масштабирование переживает работу лупой -- ок") live.toggle_rec() assert rd.rec_path == "live_01" and "Стоп" in live.b_rec.label.get_text() live.toggle_rec() assert rd.rec_path is None and "Запись" in live.b_rec.label.get_text() print(" кнопка записи -- ок") return 0 def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--port") ap.add_argument("--baud", type=int, default=921600) ap.add_argument("--fs", type=float, default=uc.FS_DEFAULT) ap.add_argument("--window", type=float, default=2.0, help="окно осциллограммы, с") ap.add_argument("--out", default="live", help="базовое имя для записей") ap.add_argument("--selftest", action="store_true") args = ap.parse_args() if args.selftest: return selftest() if not args.port: ap.error("нужен --port (список портов: uart_capture.py --list)") rd = Reader(args.port, args.baud, int(args.fs * max(args.window, 1.0))) rd.start() live = Live(rd, args.fs, args.window, args.out) from matplotlib.animation import FuncAnimation # Ссылку держим на объекте: локальную переменную сборщик может забрать live.anim = FuncAnimation(live.fig, live.update, interval=100, blit=False, cache_frame_data=False) plt.show() rd.stop_evt.set() rd.join(timeout=2) if rd.err: print(f"ошибка чтения порта: {rd.err}") return 1 return 0 if __name__ == "__main__": sys.exit(main())