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())
|