Deterministic Parsing of Mixed Binary/ASCII Instrument Outputs in Python

Mixed-format instrument streams routinely multiplex human-readable configuration blocks with high-throughput binary measurement payloads over a single transport layer such as UART, TCP, or GPIB. In laboratory automation, ad-hoc string splitting or naive socket.recv() loops collapse under partial-read conditions, delimiter collisions, and transport jitter, silently reinterpreting one dropped byte as a shifted struct field. This page works the exact single-frame case end to end: a bounded, three-phase finite-state machine that enforces strict framing invariants, validates payloads at the byte level, and hands each payload to downstream consumers as a read-only memoryview slice with a single bounded copy.

Framing Contract & Synchronization Scope

This is the focused, one-frame companion to the broader treatment in Binary & ASCII Format Parsing, the parent within the Data Capture, Validation & Metadata Sync pipeline. Where the parent surveys frame anatomy, byte-order tables, and hex/ASCII hybrids across an instrument fleet, this page pins down a single wire contract — an ASCII header, a length-prefixed little- or big-endian binary payload, and a CRC-32 trailer — and proves a parser correct against it. The assumptions are narrow on purpose: exactly one start marker per frame, an explicit LEN field, a declared endianness flag, and a fixed 4-byte trailer. It applies to any byte-stream instrument that fits that shape: digital storage oscilloscopes returning definite-length blocks, spectrometers emitting ASCII diagnostic headers before raw scan buffers, and packed-array DAQ cards.

Every frame must carry an unambiguous ASCII start marker, an explicit payload length, a deterministic endianness declaration, and a verifiable trailer. Timestamps and sequence counters are injected at the ingestion boundary, not derived post-parse, so clock drift cannot leak in during a multi-chunk partial read. Any violation of these invariants triggers immediate frame discard and a hard state reset; speculative recovery introduces silent data corruption downstream. The wire contract this page assumes:

Copy
*HDR\r\n
DEV:OSC-4000
MODE:RAW_ACQ
ENDIAN:L
LEN:4096
;END\r\n
<4096 bytes binary payload>
<CRC32 trailer (4 bytes)>

Buffer Residency Bounds & Deterministic Consumption

The correctness argument rests on two guarantees: the ingestion buffer stays bounded regardless of transport noise, and every feed() call makes forward progress or stops. When no start marker is present, the parser must not accumulate unbounded noise, so it trims all but a trailing window N_trim (large enough to hold a start marker split across two reads). Once a header is located, at most one in-flight frame is resident. That gives a hard ceiling on buffered bytes:

$$B_{\text{resident}}\le N_{\text{trim}}+|H|+L_{\max }+C$$

where N_trim is the noise-retention window (2048 B here), |H| the header length, L_max the configured max_payload, and C the 4-byte CRC trailer. Because L_max is validated before the parser transitions into payload reading, a malformed LEN field can never provoke an allocation larger than this bound — the struct overflow and out-of-memory failure modes are closed off at the header stage rather than caught after the fact.

Termination is enforced structurally. Each feed() snapshots (state, len(buffer)) before every state step and breaks the loop if a step produces no change, so a partial header, payload, or trailer parks the residual bytes for the next call instead of spinning. Amortized cost is linear: each byte is appended once and deleted once, giving O(n) total consumption across the stream. The one non-linear risk is repeated bytearray.find() over retained noise; capping the retained window at N_trim bounds each scan and keeps sustained-noise latency flat.

State-Machine Architecture & Buffer Discipline

The parser consumes bytes deterministically and transitions only when framing invariants are satisfied, in three discrete phases:

1. Header Acquisition (AWAITING_HEADER): Scan for the fixed ASCII start marker. Discard all preceding bytes as transport noise. Parse key/value metadata until the deterministic terminator (;END\r\n).
2. Length & Format Resolution: Extract payload length and endianness from the header. Validate against the hard max_payload boundary before allocating, using an explicit struct byte-order prefix (< for little-endian, > for big-endian).
3. Binary Extraction & Trailer Validation (READING_PAYLOAD → VALIDATING_TRAILER): Slice a fixed-width payload, copy it once into an immutable buffer, verify the CRC-32 trailer, and yield a zero-copy memoryview to consumers. Reset state immediately upon successful emission.

This deterministic model eliminates regex backtracking and speculative buffering. The CRC step is deliberately the same integrity gate documented under Checksum & CRC Validation — a bit-flipped block is rejected here, at the transport boundary, rather than decoded into plausible-looking noise. Because a read timeout can leave a half-received frame, the partial-frame discipline mirrors Timeout Handling & Retry Logic: the machine must be able to complete an interrupted frame on the next poll, never discard it.

Production Parser Implementation

The following implementation enforces explicit error boundaries, deterministic state transitions, and single-copy payload delivery. It is designed for synchronous or asyncio-compatible ingestion loops.

Copy
import struct
import time
import zlib
import logging
from enum import Enum, auto
from dataclasses import dataclass
from typing import Iterator, Optional

logger = logging.getLogger(__name__)

class ParseError(Exception):
    """Raised on deterministic framing violations or struct mismatches."""
    pass

class FrameState(Enum):
    AWAITING_HEADER = auto()
    READING_PAYLOAD = auto()
    VALIDATING_TRAILER = auto()

@dataclass(frozen=True)
class InstrumentFrame:
    metadata: dict[str, str]
    payload: memoryview
    ingestion_ts: float
    seq_id: int
    transport_id: str

class MixedStreamParser:
    """Bounded three-phase parser for ASCII-header + binary-payload frames."""

    def __init__(
        self,
        start_marker: bytes = b"*HDR\r\n",
        header_term: bytes = b";END\r\n",
        max_payload: int = 10_000_000,
        transport_id: str = "primary_bus",
    ):
        self.start_marker = start_marker
        self.header_term = header_term
        self.max_payload = max_payload
        self.transport_id = transport_id

        self._state = FrameState.AWAITING_HEADER
        self._buffer = bytearray()
        self._payload_len = 0
        self._payload_fmt = "<"
        self._current_meta: dict[str, str] = {}
        self._current_payload: Optional[memoryview] = None
        self._ingestion_ts = 0.0
        self._seq_counter = 0

    def feed(self, chunk: bytes) -> Iterator[InstrumentFrame]:
        """Ingest raw bytes and yield fully validated frames."""
        if not chunk:
            return

        # Inject ingestion boundary metadata before parsing.
        self._ingestion_ts = time.perf_counter()
        self._seq_counter += 1
        self._buffer.extend(chunk)

        while self._buffer:
            # Snapshot state+length to detect stalls (partial frame in flight)
            # and break instead of spinning on incomplete data.
            progress_marker = (self._state, len(self._buffer))

            if self._state == FrameState.AWAITING_HEADER:
                self._acquire_header()
            elif self._state == FrameState.READING_PAYLOAD:
                self._extract_payload()
            elif self._state == FrameState.VALIDATING_TRAILER:
                frame = self._validate_and_emit()
                if frame is not None:
                    yield frame
            else:
                break

            if (self._state, len(self._buffer)) == progress_marker:
                break  # No forward progress; await more bytes.

    def _acquire_header(self) -> None:
        idx = self._buffer.find(self.start_marker)
        if idx == -1:
            # Discard transport noise aggressively to bound memory.
            if len(self._buffer) > 4096:
                self._buffer = bytearray(self._buffer[-2048:])
            return

        del self._buffer[:idx]  # Strip preceding noise.

        term_idx = self._buffer.find(self.header_term)
        if term_idx == -1:
            return  # Partial header; await more data.

        raw_header = bytes(self._buffer[:term_idx + len(self.header_term)])
        del self._buffer[:term_idx + len(self.header_term)]

        self._current_meta = self._parse_metadata(raw_header)
        self._payload_len = int(self._current_meta.get("LEN", 0))
        endian_flag = self._current_meta.get("ENDIAN", "L")
        self._payload_fmt = "<" if endian_flag.upper() == "L" else ">"

        if self._payload_len <= 0 or self._payload_len > self.max_payload:
            logger.error("Frame discarded: invalid LEN=%s", self._payload_len)
            self._reset()
            return

        self._state = FrameState.READING_PAYLOAD

    def _parse_metadata(self, raw: bytes) -> dict[str, str]:
        content = raw[len(self.start_marker):-len(self.header_term)]
        meta: dict[str, str] = {}
        for line in content.split(b"\r\n"):
            if b":" in line:
                k, v = line.split(b":", 1)
                meta[k.decode("ascii").strip()] = v.decode("ascii").strip()
        return meta

    def _extract_payload(self) -> None:
        if len(self._buffer) < self._payload_len:
            return  # Partial payload.

        # Detach the payload into an immutable buffer. A memoryview over the
        # mutable bytearray would block the subsequent del (BufferError), so we
        # copy once here and expose a read-only memoryview to consumers.
        self._current_payload = memoryview(bytes(self._buffer[:self._payload_len]))
        del self._buffer[:self._payload_len]
        self._state = FrameState.VALIDATING_TRAILER

    def _validate_and_emit(self) -> Optional[InstrumentFrame]:
        if len(self._buffer) < 4:
            return None  # Awaiting CRC32 trailer.

        trailer = bytes(self._buffer[:4])
        del self._buffer[:4]

        computed_crc = zlib.crc32(self._current_payload) & 0xFFFFFFFF
        received_crc = struct.unpack(">I", trailer)[0]

        if computed_crc != received_crc:
            logger.critical(
                "CRC mismatch: computed=%#010x, received=%#010x. Discarding frame.",
                computed_crc, received_crc,
            )
            self._reset()
            return None

        frame = InstrumentFrame(
            metadata=self._current_meta,
            payload=self._current_payload,
            ingestion_ts=self._ingestion_ts,
            seq_id=self._seq_counter,
            transport_id=self.transport_id,
        )

        # Reset for next frame.
        self._state = FrameState.AWAITING_HEADER
        self._current_payload = None
        return frame

    def _reset(self) -> None:
        """Hard state reset on framing violation."""
        self._state = FrameState.AWAITING_HEADER
        self._buffer.clear()
        self._payload_len = 0
        self._current_payload = None

Live-Lab Validation & Verification

Before trusting this parser against a live instrument, confirm correct behavior with a deterministic loopback so that a failure isolates to your framing assumptions rather than the hardware. Build one known-good frame from bytes you control, then feed it in adversarial chunk boundaries — one byte at a time — to prove the machine reassembles across partial reads:

Copy
def build_frame(payload: bytes, endian: str = "L") -> bytes:
    """Construct a spec-compliant frame for loopback verification."""
    header = (
        b"*HDR\r\n"
        b"DEV:OSC-4000\r\n"
        b"MODE:RAW_ACQ\r\n"
        + f"ENDIAN:{endian}\r\n".encode("ascii")
        + f"LEN:{len(payload)}\r\n".encode("ascii")
        + b";END\r\n"
    )
    crc = zlib.crc32(payload) & 0xFFFFFFFF
    return header + payload + struct.pack(">I", crc)


def test_byte_at_a_time() -> None:
    payload = struct.pack("<1024f", *[float(i) for i in range(1024)])
    frame = build_frame(payload)
    parser = MixedStreamParser(max_payload=1 << 20)

    emitted = []
    for i in range(len(frame)):
        emitted.extend(parser.feed(frame[i:i + 1]))  # Worst-case fragmentation.

    assert len(emitted) == 1, "exactly one frame must survive reassembly"
    assert emitted[0].metadata["LEN"] == str(len(payload))
    assert bytes(emitted[0].payload) == payload
    print("loopback OK:", emitted[0].seq_id, emitted[0].metadata)

In a live control loop, watch three signals. First, the CRC mismatch line logs at CRITICAL — a healthy link produces zero; a nonzero-but-flat rate points to a specific cable or channel, while a climbing rate points to progressive EMI or connector degradation. Second, monitor seq_id continuity per transport_id: gaps mean frames were discarded upstream, not corrupted. Third, on the instrument side, cross-check the emitted metadata["LEN"] against the device’s own block-length preamble (for USB-TMC / IEEE 488.2 sources, the #<n><length> header) — a mismatch there means the wire contract drifted with a firmware revision. Timestamp each emission with ingestion_ts and confirm inter-frame deltas track the instrument’s configured acquisition rate rather than the transport’s arrival jitter.

Failure Modes & Diagnosis

Four failure modes are specific to this narrow header/payload/trailer machine. Map the observable symptom to a root cause and a concrete action:

For faults that recur across an instrument array rather than a single link, route the typed ParseError through Error Code Categorization so recovery policy is driven by fault class, and encode each device’s byte order behind a Protocol Abstraction Layer instead of a global flag.

Pipeline Integration & Cross-Links

This parser is the ingestion boundary for the wider pipeline. Upstream, the serial or TCP transport configured through PySerial Configuration & Tuning must guarantee byte-order consistency and a read granularity finer than one frame. Downstream, each emitted InstrumentFrame feeds Checksum & CRC Validation audit trails — the concrete streaming variant is worked in Implementing CRC32 Validation for Sensor Data Streams — where per-channel CRC rates drive hardware health monitoring.

The metadata dictionary routes into Metadata Injection Workflows, attaching calibration coefficients and instrument state before data reaches storage. The ingestion_ts field drives Threshold Tuning & Alerting for real-time deviation detection against baseline acquisition rates. In concurrent deployments, wrap feed() behind an Async Command Queuing System so one slow instrument cannot stall the shared loop — one parser instance per session, since the bytearray buffer and FSM state are not reentrant. The read-only memoryview payload casts directly into a NumPy array with no further copy for analytics.

Conclusion

Mixed ASCII/binary instrument streams demand a deterministic framing contract, strict state transitions, and a single bounded copy from the resizable ingestion buffer into an immutable, read-only payload. By enforcing ingestion-boundary timestamping, explicit length and endianness resolution before allocation, and hard CRC validation, this machine bounds its own memory, guarantees forward progress on partial reads, and prevents silent corruption from reaching the analysis layer. Deploy it with bounded buffers, explicit struct prefixes, and immediate state resets on violation to hold pipeline integrity across high-throughput laboratory environments.