From c1ac8d59ca6eb97a2b325c3607b3e613e0f2aebf Mon Sep 17 00:00:00 2001 From: nikit3s Date: Fri, 14 Aug 2026 16:10:57 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9F=D1=80=D0=B8=D1=91=D0=BC=D0=BD=D1=8B?= =?UTF-8?q?=D0=B5=20=D1=81=D0=BA=D1=80=D0=B8=D0=BF=D1=82=D1=8B:=20=D1=88?= =?UTF-8?q?=D0=BA=D0=B0=D0=BB=D0=B0=20dBFS=20=D0=B2=D0=BC=D0=B5=D1=81?= =?UTF-8?q?=D1=82=D0=BE=20=D0=B2=D0=BE=D0=BB=D1=8C=D1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Разбор кадров и запись .raw/.bin не трогали. Заменена только шкала: у микрофона нет опорного напряжения, уровни считаются от полной шкалы. Бит1 STATUS теперь показывается как насыщение дециматора. --- tools/uart_capture.py | 330 +++++++++++++++++++++++++++++++++ tools/uart_live.py | 419 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 749 insertions(+) create mode 100644 tools/uart_capture.py create mode 100644 tools/uart_live.py diff --git a/tools/uart_capture.py b/tools/uart_capture.py new file mode 100644 index 0000000..04e6011 --- /dev/null +++ b/tools/uart_capture.py @@ -0,0 +1,330 @@ +#!/usr/bin/env python3 +"""Приём кадров цифрового микрофона по UART. + +Пишет два файла: + .raw все байты из порта, до разбора -- отладка канала + .bin только отсчёты, 3 байта BE -- читается read_24bit_bin() + из F:\\PyPrj\\Signal_analayzer + +Кадр с несошедшейся CRC отбрасывается, пропуск восстанавливается по FRAME_CNT +и добивается нулями, чтобы не сдвинулась ось времени. +Формат кадра -- README.md. +""" + +import argparse +import math +import sys +import time + +import serial +from serial.tools import list_ports + +SYNC = b"\xA5\x5A" +SAMPLES_PER_FRAME = 50 +PAYLOAD_LEN = 2 + 1 + SAMPLES_PER_FRAME * 3 # FRAME_CNT + STATUS + DATA +FRAME_LEN = len(SYNC) + PAYLOAD_LEN + 2 # + CRC16 +FS_DEFAULT = 12500.0 +FS_CODE = 2 ** 23 # полная шкала: постоянная '1' на PDM + + +def dbfs(code) -> float: + """Код -> dBFS. Шкалу задаёт микрофон, вольт в тракте нет.""" + return 20.0 * math.log10(max(abs(code), 1e-9) / FS_CODE) + + +def _make_crc_table(): + tbl = [] + for b in range(256): + c = b << 8 + for _ in range(8): + c = ((c << 1) ^ 0x1021) & 0xFFFF if c & 0x8000 else (c << 1) & 0xFFFF + tbl.append(c) + return tbl + + +_CRC_TBL = _make_crc_table() + + +def crc16_ccitt(data: bytes) -> int: + """CRC-16/CCITT, poly 0x1021, init 0xFFFF, MSB-first -- как в ПЛИС.""" + crc = 0xFFFF + for byte in data: + crc = ((crc << 8) & 0xFFFF) ^ _CRC_TBL[(crc >> 8) ^ byte] + return crc + + +class Stats: + def __init__(self): + self.frames_ok = 0 + self.crc_errors = 0 + self.resyncs = 0 + self.padded_frames = 0 + self.status_flags = 0 + self.samples = 0 + self.code_min = None + self.code_max = None + self.code_sum = 0 + self.code_sqsum = 0 + self.bits_or = 0 # какие биты хоть раз были 1 + self.bits_and = 0xFFFFFF # какие биты всегда были 1 + + +def decode_samples(data: bytes): + """150 байт -> 50 знаковых отсчётов.""" + out = [] + for i in range(0, len(data), 3): + v = (data[i] << 16) | (data[i + 1] << 8) | data[i + 2] + if v & 0x800000: + v -= 0x1000000 + out.append(v) + return out + + +def open_port(port, baud): + ser = serial.Serial(port, baud, timeout=0.1) + try: + # Буфер COM-порта в Windows по умолчанию около 4 кБ -- это всего 100 мс + # потока, любая заминка в разборе стоит пачки байт + ser.set_buffer_size(rx_size=1 << 20, tx_size=1 << 12) + except Exception: + pass + ser.reset_input_buffer() + return ser + + +def run_dump(port, baud, duration, out_base): + """Только приём в файл, без разбора. Показывает, теряет ли байты сам канал.""" + raw_path = out_base + ".raw" + total = 0 + slowest = 0.0 + with open_port(port, baud) as ser, open(raw_path, "wb") as fraw: + t0 = time.time() + t_end = t0 + duration + while time.time() < t_end: + t1 = time.perf_counter() + chunk = ser.read(4096) + slowest = max(slowest, time.perf_counter() - t1) + if chunk: + fraw.write(chunk) + total += len(chunk) + elapsed = time.time() - t0 + + rate = total / elapsed + expect = FRAME_LEN * FS_DEFAULT / SAMPLES_PER_FRAME + print(f"принято : {total} байт за {elapsed:.2f} с") + print(f"скорость : {rate / 1000:.1f} кБ/с") + print(f"ожидалось : {expect / 1000:.1f} кБ/с ({FRAME_LEN} байт/кадр, " + f"{FS_DEFAULT / SAMPLES_PER_FRAME:.0f} кадров/с)") + print(f"дошло : {100.0 * rate / expect:.1f}%") + print(f"дольше всего ждали чтение: {slowest * 1000:.0f} мс") + if rate < expect * 0.95: + print("\nБайты теряются в канале, разбор тут ни при чём: " + "мост, драйвер или скорость порта.") + else: + print("\nКанал отдаёт весь поток. Значит потери возникают при разборе.") + return 0 + + +class FrameParser: + """Разбор потока кадров. Единственное место, где описан формат, -- + им пользуются и запись, и живой просмотр. + + feed() возвращает список (frame_cnt, status, data, gap): data -- 150 байт + отсчётов, gap -- сколько кадров потерялось перед этим. + """ + + def __init__(self): + self.buf = bytearray() + self.expected_cnt = None + self.frames_ok = 0 + self.crc_errors = 0 + self.resyncs = 0 + self.padded_frames = 0 + self.status_flags = 0 + + def feed(self, chunk): + out = [] + self.buf += chunk + while True: + idx = self.buf.find(SYNC) + if idx < 0: + # Оставляем хвост, в котором SYNC мог разорваться между чтениями + if len(self.buf) > 1: + del self.buf[:-1] + break + if idx > 0: + self.resyncs += 1 + del self.buf[:idx] + if len(self.buf) < FRAME_LEN: + break + + frame = bytes(self.buf[:FRAME_LEN]) + payload = frame[2:2 + PAYLOAD_LEN] + crc_rx = (frame[-2] << 8) | frame[-1] + + if crc16_ccitt(payload) != crc_rx: + self.crc_errors += 1 + # SYNC мог оказаться внутри данных: сдвигаемся и ищем дальше + del self.buf[:2] + continue + + del self.buf[:FRAME_LEN] + + cnt = (payload[0] << 8) | payload[1] + status = payload[2] + self.status_flags |= status + + gap = 0 + if self.expected_cnt is not None and cnt != self.expected_cnt: + gap = (cnt - self.expected_cnt) & 0xFFFF + self.padded_frames += gap + self.expected_cnt = (cnt + 1) & 0xFFFF + self.frames_ok += 1 + out.append((cnt, status, payload[3:], gap)) + return out + + +def run(port, baud, duration, out_base, quiet): + raw_path = out_base + ".raw" + bin_path = out_base + ".bin" + log_path = out_base + ".log" + + st = Stats() + log_lines = [] + + def log(msg): + log_lines.append(msg) + if not quiet: + print(msg) + + log(f"порт {port}, {baud} бод, длительность {duration} с") + + parser = FrameParser() + t_end = time.time() + duration + gaps_printed = 0 + + with open_port(port, baud) as ser, \ + open(raw_path, "wb") as fraw, open(bin_path, "wb") as fbin: + while time.time() < t_end: + chunk = ser.read(4096) + if not chunk: + continue + fraw.write(chunk) + + for cnt, status, data, gap in parser.feed(chunk): + if gap: + fbin.write(b"\x00" * (gap * SAMPLES_PER_FRAME * 3)) + st.samples += gap * SAMPLES_PER_FRAME + # Печатаем только первые сообщения: вывод в консоль медленный + # и на длинной записи сам становится причиной потерь + msg = (f"пропуск {gap} кадр(ов) перед FRAME_CNT={cnt}, " + f"добито {gap * SAMPLES_PER_FRAME} нулей") + log_lines.append(msg) + if gaps_printed < 10 and not quiet: + print(msg) + gaps_printed += 1 + + fbin.write(data) + st.samples += SAMPLES_PER_FRAME + + for v in decode_samples(data): + st.code_sum += v + st.code_sqsum += v * v + st.bits_or |= v & 0xFFFFFF + st.bits_and &= v & 0xFFFFFF + if st.code_min is None or v < st.code_min: + st.code_min = v + if st.code_max is None or v > st.code_max: + st.code_max = v + + st.frames_ok = parser.frames_ok + st.crc_errors = parser.crc_errors + st.resyncs = parser.resyncs + st.padded_frames = parser.padded_frames + st.status_flags = parser.status_flags + + log("") + log(f"кадров принято : {st.frames_ok}") + log(f"ошибок CRC : {st.crc_errors}") + log(f"ресинхронизаций : {st.resyncs}") + log(f"кадров добито нулями: {st.padded_frames}") + log(f"отсчётов в .bin : {st.samples} ({st.samples / FS_DEFAULT:.3f} с при {FS_DEFAULT:.0f} Гц)") + + if st.status_flags & 0x01: + log("STATUS: было переполнение FIFO в ПЛИС") + if st.status_flags & 0x02: + log("STATUS: выход дециматора упирался в шкалу") + if st.status_flags == 0 and st.frames_ok: + log("STATUS: чисто") + + if st.frames_ok: + n = st.frames_ok * SAMPLES_PER_FRAME + mean = st.code_sum / n + var = max(st.code_sqsum / n - mean * mean, 0.0) + sigma = var ** 0.5 + peak = max(abs(st.code_min), abs(st.code_max)) + log("") + log(f"код min/max/среднее : {st.code_min} / {st.code_max} / {mean:.1f}") + log(f"пик : {peak} код ({dbfs(peak):+.1f} dBFS)") + log(f"смещение нуля : {mean:.1f} код ({dbfs(mean):+.1f} dBFS)") + log(f"СКО : {sigma:.1f} код ({dbfs(sigma):+.1f} dBFS)") + + # Постоянные старшие биты -- норма: при малом размахе сигнала меняются + # только младшие. Тревожно обратное: залипшие младшие биты, которые + # обязан расшевелить хотя бы собственный шум микрофона. + frozen = ((~st.bits_or) | st.bits_and) & 0xFFFFFF + low_frozen = frozen & 0x0F + if frozen == 0xFFFFFF: + log("биты : НИ ОДИН из 24 не менялся — тракт отдаёт константу") + elif low_frozen: + log("биты : ВНИМАНИЕ, не менялись младшие " + ", ".join( + str(i) for i in range(3, -1, -1) if low_frozen >> i & 1) + + " — похоже на сбой чтения, а не на сигнал") + else: + top = 0 + for i in range(23, -1, -1): + if frozen >> i & 1: + top += 1 + else: + break + if top: + log(f"биты : младшие {24 - top} переключались, " + f"старшие {top} постоянны — норма при малом размахе") + else: + log("биты : все 24 переключались") + + with open(log_path, "w", encoding="utf-8") as flog: + flog.write("\n".join(log_lines) + "\n") + + return 1 if (st.crc_errors or st.status_flags or not st.frames_ok) else 0 + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--port", help="например COM5") + ap.add_argument("--baud", type=int, default=921600) + ap.add_argument("--seconds", type=float, default=10.0) + ap.add_argument("--out", default="capture", help="базовое имя файлов без расширения") + ap.add_argument("--quiet", action="store_true") + ap.add_argument("--list", action="store_true", help="показать доступные порты") + ap.add_argument("--dump", action="store_true", + help="только приём в .raw без разбора: проверка канала") + args = ap.parse_args() + + if args.list: + for p in list_ports.comports(): + print(f"{p.device} {p.description}") + return 0 + + if not args.port: + ap.error("нужен --port (список портов: --list)") + + if args.dump: + return run_dump(args.port, args.baud, args.seconds, args.out) + + return run(args.port, args.baud, args.seconds, args.out, args.quiet) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/uart_live.py b/tools/uart_live.py new file mode 100644 index 0000000..bbaf83a --- /dev/null +++ b/tools/uart_live.py @@ -0,0 +1,419 @@ +#!/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())