jack
5b186df888
- Add automatic format detection (binary, ASCII hex dump, or text line) in StreamReader based on stream preambles. - Implement fallback to 115200 baudrate when 888888 ioctl fails (critical for macOS PTYs). - Parse binary streams (X family) with little-endian sequence/length headers and Modbus CRC-16 checks. - Discover and prioritize virtual serial ports (from running socat instances) in enumerate_ports using lsof. - Update FakeTransport to mock read() and readline() calls for unit test compatibility.
47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
import time
|
|
from pygui.backend.models.frame import Frame
|
|
from pygui.serialcomm.stream_reader import StreamReader
|
|
|
|
class FakeTransport:
|
|
"""Yield canned data then blocks (returns b'')."""
|
|
def __init__(self, lines):
|
|
self._data = b"".join(lines)
|
|
self._pos = 0
|
|
self.closed = False
|
|
def read(self, n=1):
|
|
time.sleep(0.001)
|
|
if self._pos >= len(self._data):
|
|
return b""
|
|
res = self._data[self._pos:self._pos+n]
|
|
self._pos += n
|
|
return res
|
|
def readline(self):
|
|
time.sleep(0.001)
|
|
if self._pos >= len(self._data):
|
|
return b""
|
|
idx = self._data.find(b"\n", self._pos)
|
|
if idx == -1:
|
|
res = self._data[self._pos:]
|
|
self._pos = len(self._data)
|
|
return res
|
|
res = self._data[self._pos:idx+1]
|
|
self._pos = idx + 1
|
|
return res
|
|
def close(self): self.closed = True
|
|
|
|
def parse_line(raw: str, seq: int) -> Frame:
|
|
parts = [float(x) for x in raw.split(",")]
|
|
return Frame(seq = seq, time = parts[0], values = parts[1:])
|
|
|
|
def test_reads_frames_and_counts_errors():
|
|
got, errors = [], []
|
|
transport = FakeTransport([b"0.0,149,148\n", b"garbage\n", b"0.5,150,149\n"])
|
|
r = StreamReader(transport, parse_line,
|
|
on_frame = got.append, on_error = lambda e: errors.append(e))
|
|
|
|
r.start()
|
|
time.sleep(0.1)
|
|
r.stop()
|
|
assert [f.values for f in got] == [[149.0, 148.0], [150.0, 149.0]]
|
|
assert r.error_count == 1
|
|
assert transport.closed |