422 lines
18 KiB
Python
422 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""Живой просмотр потока с AD7766: осциллограмма, спектр и счётчики.
|
||
|
||
Нужен, чтобы глазами проконтролировать сигнал до того, как писать запись:
|
||
видно и форму, и шум, и наводки, и целостность кадров.
|
||
|
||
Разбор кадров общий с 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
|
||
|
||
VREF = 5.0
|
||
LSB = 2 * VREF / 2 ** 24 # 596.0 нВ
|
||
NFFT = 8192
|
||
|
||
|
||
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("AD7766 -- живой просмотр")
|
||
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("нВ/√Гц")
|
||
self.ax_f.grid(alpha=0.3, which="both")
|
||
self.ax_f.set_xscale("log")
|
||
self.ax_f.set_yscale("log")
|
||
# Пределы обеих осей задаём сразу положительными: до прихода данных
|
||
# линия пуста, а логарифмическая шкала с нулём в пределах падает
|
||
self.ax_f.set_xlim(1, fs / 2)
|
||
self.ax_f.set_ylim(1, 1e6)
|
||
|
||
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 * LSB
|
||
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:], np.maximum(np.sqrt(psd[1:]) * 1e9, 1e-3))
|
||
self.ax_f.set_ylabel("нВ/√Гц")
|
||
else:
|
||
amp = np.sqrt(sp[1:]) / w.sum() * 2 * 1e6
|
||
self.ln_f.set_data(f[1:], np.maximum(amp, 1e-6))
|
||
self.ax_f.set_ylabel("амплитуда, мкВ")
|
||
|
||
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())
|
||
pos = yd[yd > 0]
|
||
if len(pos):
|
||
top = pos.max() * 3
|
||
bot = max(np.percentile(pos, 1) / 3, top / 1e6)
|
||
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} DRDY={(p.status_flags >> 1) & 1}")
|
||
self.txt.set_text(
|
||
f"среднее {mean * 1e3:+9.3f} мВ СКО {v.std() / LSB:7.1f} код "
|
||
f"({v.std() * 1e6:8.2f} мкВ) размах {(v.max() - v.min()) * 1e6:9.1f} мкВ\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] * 10, (f_small, f_big)
|
||
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] / 10, 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())
|