ChTheo/RFG_stm32_ADC_receiver_GUI
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

New tty parser: accepts binary format. Enable arg: --bin

Browse Source
This commit is contained in:
Theodor Chikin
2026-02-17 18:51:12 +03:00
parent 34d151aef1
commit 8b1d424cbe
5 changed files with 171 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
rfg_adc_plotter/gui/matplotlib_backend.py
View File

@ -89,7 +89,14 @@ def run_matplotlib(args):
q: Queue[SweepPacket] = Queue(maxsize=1000)
stop_event = threading.Event()
reader = SweepReader(args.port, args.baud, q, stop_event, fancy=bool(args.fancy))
reader = SweepReader(
args.port,
args.baud,
q,
stop_event,
fancy=bool(args.fancy),
bin_mode=bool(getattr(args, "bin_mode", False)),
)
reader.start()
max_sweeps = int(max(10, args.max_sweeps))

9
rfg_adc_plotter/gui/pyqtgraph_backend.py
View File

@ -106,7 +106,14 @@ def run_pyqtgraph(args):
q: Queue[SweepPacket] = Queue(maxsize=1000)
stop_event = threading.Event()
reader = SweepReader(args.port, args.baud, q, stop_event, fancy=bool(args.fancy))
reader = SweepReader(
args.port,
args.baud,
q,
stop_event,
fancy=bool(args.fancy),
bin_mode=bool(getattr(args, "bin_mode", False)),
)
reader.start()
max_sweeps = int(max(10, args.max_sweeps))

212
rfg_adc_plotter/io/sweep_reader.py
View File

@ -24,6 +24,7 @@ class SweepReader(threading.Thread):
out_queue: "Queue[SweepPacket]",
stop_event: threading.Event,
fancy: bool = False,
bin_mode: bool = False,
):
super().__init__(daemon=True)
self._port_path = port_path
@ -32,11 +33,17 @@ class SweepReader(threading.Thread):
self._stop = stop_event
self._src: Optional[SerialLineSource] = None
self._fancy = bool(fancy)
self._bin_mode = bool(bin_mode)
self._max_width: int = 0
self._sweep_idx: int = 0
self._last_sweep_ts: Optional[float] = None
self._n_valid_hist = deque()
@staticmethod
def _u16_to_i16(v: int) -> int:
"""Преобразование 16-bit слова в знаковое значение."""
return v - 0x10000 if (v & 0x8000) else v
def _finalize_current(self, xs, ys, channels: Optional[set]):
if not xs:
return
@ -135,11 +142,145 @@ class SweepReader(threading.Thread):
except Exception:
pass
def run(self):
xs: list = []
ys: list = []
def _run_ascii_stream(self, chunk_reader: SerialChunkReader):
xs: list[int] = []
ys: list[int] = []
cur_channel: Optional[int] = None
cur_channels: set = set()
cur_channels: set[int] = set()
buf = bytearray()
while not self._stop.is_set():
data = chunk_reader.read_available()
if data:
buf += data
else:
time.sleep(0.0005)
continue
while True:
nl = buf.find(b"\n")
if nl == -1:
break
line = bytes(buf[:nl])
del buf[: nl + 1]
if line.endswith(b"\r"):
line = line[:-1]
if not line:
continue
if line.startswith(b"Sweep_start"):
self._finalize_current(xs, ys, cur_channels)
xs.clear()
ys.clear()
cur_channel = None
cur_channels.clear()
continue
if len(line) >= 3:
parts = line.split()
if len(parts) >= 3 and (parts[0].lower() == b"s" or parts[0].lower().startswith(b"s")):
try:
if parts[0].lower() == b"s":
if len(parts) >= 4:
ch = int(parts[1], 10)
x = int(parts[2], 10)
y = int(parts[3], 10)
else:
ch = 0
x = int(parts[1], 10)
y = int(parts[2], 10)
else:
ch = int(parts[0][1:], 10)
x = int(parts[1], 10)
y = int(parts[2], 10)
except Exception:
continue
if cur_channel is None:
cur_channel = ch
cur_channels.add(ch)
xs.append(x)
ys.append(y)
if len(buf) > 1_000_000:
del buf[:-262144]
self._finalize_current(xs, ys, cur_channels)
def _run_binary_stream(self, chunk_reader: SerialChunkReader):
xs: list[int] = []
ys: list[int] = []
cur_channel: Optional[int] = None
cur_channels: set[int] = set()
waiting_channel = False
waiting_first_point = False
point_word: Optional[int] = None
value_word: Optional[int] = None
buf = bytearray()
while not self._stop.is_set():
data = chunk_reader.read_available()
if data:
buf += data
else:
time.sleep(0.0005)
continue
usable = len(buf) & ~1
if usable == 0:
continue
i = 0
while i < usable:
w = int(buf[i]) | (int(buf[i + 1]) << 8)
i += 2
if waiting_channel:
cur_channel = int(w)
cur_channels.add(cur_channel)
waiting_channel = False
waiting_first_point = True
continue
if w == 0xFFFF:
self._finalize_current(xs, ys, cur_channels)
xs.clear()
ys.clear()
cur_channel = None
cur_channels.clear()
waiting_channel = True
waiting_first_point = False
point_word = None
value_word = None
continue
if point_word is None:
if waiting_first_point and (w == 0x0A0A or w == 0x000A):
continue
point_word = int(w)
waiting_first_point = False
continue
if value_word is None:
value_word = int(w)
continue
is_point_end = (w == 0x000A) or ((w & 0x00FF) == 0x0A) or ((w >> 8) == 0x0A)
if is_point_end:
if cur_channel is not None:
cur_channels.add(cur_channel)
xs.append(point_word)
ys.append(self._u16_to_i16(value_word))
point_word = None
value_word = None
del buf[:usable]
if len(buf) > 1_000_000:
del buf[:-262144]
self._finalize_current(xs, ys, cur_channels)
def run(self):
try:
self._src = SerialLineSource(self._port_path, self._baud, timeout=1.0)
@ -150,66 +291,11 @@ class SweepReader(threading.Thread):
try:
chunk_reader = SerialChunkReader(self._src)
buf = bytearray()
while not self._stop.is_set():
data = chunk_reader.read_available()
if data:
buf += data
else:
time.sleep(0.0005)
continue
while True:
nl = buf.find(b"\n")
if nl == -1:
break
line = bytes(buf[:nl])
del buf[: nl + 1]
if line.endswith(b"\r"):
line = line[:-1]
if not line:
continue
if line.startswith(b"Sweep_start"):
self._finalize_current(xs, ys, cur_channels)
xs.clear()
ys.clear()
cur_channel = None
cur_channels.clear()
continue
if len(line) >= 3:
parts = line.split()
if len(parts) >= 3 and (parts[0].lower() == b"s" or parts[0].lower().startswith(b"s")):
try:
if parts[0].lower() == b"s":
if len(parts) >= 4:
ch = int(parts[1], 10)
x = int(parts[2], 10)
y = int(parts[3], 10)
else:
ch = 0
x = int(parts[1], 10)
y = int(parts[2], 10)
else:
ch = int(parts[0][1:], 10)
x = int(parts[1], 10)
y = int(parts[2], 10)
except Exception:
continue
if cur_channel is None:
cur_channel = ch
cur_channels.add(ch)
xs.append(x)
ys.append(y)
if len(buf) > 1_000_000:
del buf[:-262144]
if self._bin_mode:
self._run_binary_stream(chunk_reader)
else:
self._run_ascii_stream(chunk_reader)
finally:
try:
self._finalize_current(xs, ys, cur_channels)
except Exception:
pass
try:
if self._src is not None:
self._src.close()

6
rfg_adc_plotter/main.py Normal file → Executable file
View File

@ -77,6 +77,12 @@ def build_parser() -> argparse.ArgumentParser:
default="projector",
help="Тип нормировки: projector (по огибающим в [-1000,+1000]) или simple (raw/calib)",
)
parser.add_argument(
"--bin",
dest="bin_mode",
action="store_true",
help="Бинарный протокол: 16-bit поток, 0xFFFF+канал для старта свипа, точки point,value,'\\n'",
)
return parser