From 5b1c924d35bcfd5d761fd108572daea6986fb088 Mon Sep 17 00:00:00 2001 From: jack Date: Mon, 15 Jun 2026 11:56:09 -0700 Subject: [PATCH] feat(stream_reader): optimize serial stream protocol detection. reduce CPU busy-spins --- src/pygui/serialcomm/stream_reader.py | 192 +++++++++++++++----------- 1 file changed, 113 insertions(+), 79 deletions(-) diff --git a/src/pygui/serialcomm/stream_reader.py b/src/pygui/serialcomm/stream_reader.py index e908cbf..5dc8cf3 100644 --- a/src/pygui/serialcomm/stream_reader.py +++ b/src/pygui/serialcomm/stream_reader.py @@ -32,9 +32,35 @@ class StreamReader: self._thread = threading.Thread(target=self._run, daemon = True) self._thread.start() + def _read_chunk(self, min_bytes: int = 1) -> bytes: + """Read a chunk of available bytes from the transport. + + Reads up to 256 bytes in one shot instead of byte-at-a-time, + falling back to single-byte reads with a yield when nothing is + immediately available.""" + chunk = self._transport.read(max(min_bytes, 256)) + if chunk: + return chunk + # Nothing available right now – yield briefly to avoid busy-spin. + b = self._transport.read(1) + if not b: + time.sleep(0.005) + return b + + def _accumulate(self, needed: int) -> bytearray: + """Accumulate *needed* bytes using buffered reads.""" + buf = bytearray() + while len(buf) < needed and not self._stop.is_set(): + chunk = self._transport.read(needed - len(buf)) + if not chunk: + time.sleep(0.005) + continue + buf.extend(chunk) + return buf + def _run(self) -> None: try: - # Check the first few bytes to determine if the wafer is streaming binary, ASCII hex dump, or text lines. + # Peek at the first few bytes to determine protocol. peek_buf = bytearray() while not self._stop.is_set() and len(peek_buf) < 32: b = self._transport.read(1) @@ -47,16 +73,24 @@ class StreamReader: if self._stop.is_set() or not peek_buf: return - detect_buf = bytes(c for c in peek_buf if c not in (10, 13, 0)) - if not detect_buf: - detect_buf = bytes(peek_buf) + raw = bytes(peek_buf) - if detect_buf.startswith(b"\xaa\x88") or (len(detect_buf) >= 2 and detect_buf[0] == 0xAA and detect_buf[1] == 0x88): - self._run_binary(bytes(peek_buf)) - elif b"," in detect_buf: - self._run_text_lines(bytes(peek_buf)) - else: - self._run_ascii(bytes(peek_buf)) + # Binary sync bytes are the strongest signal. + if raw.startswith(b"\xaa\x88"): + self._run_binary(raw) + return + + # Text-lines (CSV) requires BOTH a newline AND commas in the + # pre-newline content. This avoids misclassifying a binary + # payload that happens to contain 0x2C as text. + has_newline = b"\n" in raw or b"\r" in raw + line_before_nl = raw.split(b"\n")[0].split(b"\r")[0] + if has_newline and b"," in line_before_nl: + self._run_text_lines(raw) + return + + # Default: ASCII hex dump stream. + self._run_ascii(raw) finally: try: self._transport.close() @@ -66,41 +100,43 @@ class StreamReader: def _run_binary(self, initial_bytes: bytes) -> None: log.info("StreamReader: starting binary stream parsing") buf = bytearray(initial_bytes) - + while not self._stop.is_set(): + # Find sync marker in buffer. idx = 0 synced = False while idx < len(buf) - 1: - if buf[idx] == 0xAA and buf[idx+1] == 0x88: + if buf[idx] == 0xAA and buf[idx + 1] == 0x88: synced = True break idx += 1 - + if synced: buf = buf[idx:] + # Accumulate header (5 bytes after sync). while len(buf) < 7 and not self._stop.is_set(): - b = self._transport.read(1) - if b: - buf.extend(b) - + chunk = self._read_chunk() + buf.extend(chunk) + if self._stop.is_set(): return - + payload_len = int.from_bytes(buf[5:7], byteorder='little') total_packet_len = 2 + 5 + payload_len + 2 - - while len(buf) < total_packet_len and not self._stop.is_set(): - b = self._transport.read(1) - if b: - buf.extend(b) - + + # Accumulate the rest of the packet in one buffered read. + remaining = total_packet_len - len(buf) + if remaining > 0: + extra = self._accumulate(remaining) + buf.extend(extra) + if self._stop.is_set(): return - + header = buf[2:7] seq = int.from_bytes(header[1:3], byteorder='little') - payload = buf[7:7+payload_len] - + payload = buf[7:7 + payload_len] + try: frame = self._parse(payload, seq) self._on_frame(frame) @@ -109,25 +145,26 @@ class StreamReader: if self.error_count <= 10: log.warning("Parse error on binary payload: %s", exc) self._on_error(exc) - + buf = buf[total_packet_len:] else: + # Resync: keep only trailing 0xAA if present, then read more. if len(buf) > 0 and buf[-1] == 0xAA: buf = bytearray([0xAA]) else: buf.clear() - - b = self._transport.read(1) - if b: - buf.extend(b) + + chunk = self._read_chunk() + buf.extend(chunk) def _run_ascii(self, initial_bytes: bytes) -> None: log.info("StreamReader: starting ASCII hex dump stream parsing") - + valid_words = 80 if self._family_code == "X" else 244 block_words = 256 word_char_len = 4 - + chars_needed = block_words * word_char_len # total hex chars per block + hex_chars = bytearray() if initial_bytes: for b in initial_bytes: @@ -135,46 +172,45 @@ class StreamReader: hex_chars.append(b) elif b == 0xAA: hex_chars.append(b) - - def read_next_hex_char() -> str | None: - if hex_chars: - val = chr(hex_chars.pop(0)) - return val - while not self._stop.is_set(): - b = self._transport.read(1) - if not b: - continue - if b[0] == 0xAA: - hex_chars.extend(b) - return None - c = chr(b[0]) - if c in "0123456789ABCDEFabcdef": - return c - return None seq = 0 while not self._stop.is_set(): - word_buffer = [] - eof = False - for _ in range(block_words): - word_str = "" - for _ in range(word_char_len): - c = read_next_hex_char() - if c is None: - eof = True + # Accumulate a full block of hex characters using buffered reads. + while len(hex_chars) < chars_needed and not self._stop.is_set(): + chunk = self._read_chunk(min_bytes=chars_needed - len(hex_chars)) + for byte in chunk: + if byte == 0xAA: + # Sync marker appeared – hand off to binary parser. + hex_chars.append(byte) break - word_str += c - if eof or len(word_str) < word_char_len: - eof = True + c = chr(byte) + if c in "0123456789ABCDEFabcdef": + hex_chars.append(c) + else: + continue + break # broke out of inner loop due to 0xAA + + # Check for early termination (0xAA sync marker anywhere in buffer). + aa_idx = hex_chars.find(bytes([0xAA])) + if aa_idx >= 0: + log.info("StreamReader: Detected 0xAA at position %d, switching to binary mode.", aa_idx) + self._run_binary(hex_chars[aa_idx:]) + return + + # Parse words from the accumulated hex characters. + word_buffer = [] + for i in range(0, min(len(hex_chars), chars_needed), word_char_len): + word_str = hex_chars[i:i + word_char_len] + if len(word_str) < word_char_len: break - word_buffer.append(word_str) - - if eof: - if hex_chars and hex_chars[0] == 0xAA: - log.info("StreamReader: Detected 0xAA, switching to binary mode.") - self._run_binary(bytes(hex_chars)) - break - + word_buffer.append(bytes(word_str)) + + if not word_buffer: + time.sleep(0.01) + continue + + hex_chars = hex_chars[len(word_buffer) * word_char_len:] + valid_hex_words = word_buffer[:valid_words] values = [] for hex_val in valid_hex_words: @@ -184,7 +220,7 @@ class StreamReader: values.append(t) except Exception: values.append(0.0) - + try: frame = Frame(seq=seq, time=time.monotonic(), values=values) self._on_frame(frame) @@ -196,9 +232,10 @@ class StreamReader: log.info("StreamReader: starting line-by-line ASCII text parsing") buf = bytearray(initial_bytes) seq = 0 - + while not self._stop.is_set(): - if b"\n" in buf: + # Drain complete lines from the buffer. + while b"\n" in buf: line_bytes, _, buf = buf.partition(b"\n") line = line_bytes.decode('utf-8', errors='ignore').strip() if line: @@ -210,13 +247,10 @@ class StreamReader: except Exception as exc: self.error_count += 1 self._on_error(exc) - continue - - b = self._transport.read(1) - if not b: - time.sleep(0.005) - continue - buf.extend(b) + + # Read more data using buffered reads. + chunk = self._read_chunk() + buf.extend(chunk) def stop(self) -> None: self._stop.set()