""" framing.py — frame codec for the Instantel Micromate (Series IV) wire protocol. A Micromate answers Series III *command* frames, so the request side looks familiar. The framing underneath is not the same, and the differences are all of the kind that produce a silently-ignored frame rather than an error: Series III response: [DLE 0x10] [STX 0x02] … [chk] [ETX 0x03] Micromate response: [STX 0x02] … [chk] [ETX 0x03] ^ no leading DLE That missing byte is why `minimateplus.framing.S3FrameParser` returns *nothing* on Micromate traffic — it locates frames by scanning for `DLE STX`, which never occurs. A capture holding 12 acknowledged writes reads as 12 unanswered requests. De-stuffed payload layout (both directions): request response [0] CMD 0x10 [0] CMD 0x00 [1] flags 0x00 [1] flags 0xC5 / 0x03 ← firmware line [2] SUB [2] SUB 0xFF − request_SUB [3] 0x00 [3] PAGE_HI [4] offset_hi [4] PAGE_LO [5] offset_lo [5+] data [6:16] params (10 bytes) Everything below was established against the 251 request and 251 response frames in `bridges/captures/9-24-26 - micromate2/` (UM12947, firmware 11.0CB). Where a rule is asserted, the number of frames it was checked on is given — the two rules that look like small details cost 26% and 22% of frames respectively when guessed wrong, so the counts are the point. """ from __future__ import annotations from dataclasses import dataclass from typing import Optional # ── Protocol byte constants ─────────────────────────────────────────────────── DLE = 0x10 # Data Link Escape STX = 0x02 # Start of text — begins a frame ETX = 0x03 # End of text — ends a frame ACK = 0x41 # Frame-start marker on the request side MM_CMD = 0x10 # payload[0] in a request MM_RSP_CMD = 0x00 # payload[0] in a response # payload[1] of a response identifies the firmware line it came from. # ⚠ Two units, one of each — a strong hypothesis, not a proven encoding. FLAGS_BLASTWARE = 0xC5 # the 11.0CB line (UM12947) FLAGS_THOR = 0x03 # the 11.0BD line (UM20147) # ⚠ THE ESCAPE SET. A Micromate escapes exactly these four byte values, # prefixing each with a DLE — and nothing else. Established by re-stuffing # every captured frame and comparing to the wire: 251/251 responses and 251/251 # requests reproduce byte-for-byte with this set, and no other candidate set # reproduces even 200 of either. # # The two near-misses are worth naming, because both look plausible: # * `{0x10}` alone — the Series III rule — reproduces 130/251 responses and # 177/251 requests. # * adding ACK (0x41) reproduces only 196/251 responses: a literal 0x41 in # the data is NOT escaped. _ESCAPED = frozenset({STX, ETX, 0x04, DLE}) # A response header is 5 bytes; a frame must also carry its checksum. _MIN_PAYLOAD = 5 _REQUEST_PAYLOAD_SIZE = 16 # ── Stuffing ────────────────────────────────────────────────────────────────── def stuff(data: bytes) -> bytes: """Escape every byte the Micromate escapes: `XX` → `10 XX` for the four.""" out = bytearray() for b in data: if b in _ESCAPED: out.append(DLE) out.append(b) return bytes(out) def unstuff(data: bytes) -> bytes: """Reverse `stuff()`: `10 XX` → `XX`, for any XX. Uniform, with no inner-frame carve-out — which is a real simplification over Series III, where `DLE+ETX` inside a frame is literal data that must survive de-stuffing. Since only four byte values are ever escaped, taking *any* `10 XX` as `XX` is exact rather than merely convenient. """ out = bytearray() i = 0 while i < len(data): if data[i] == DLE and i + 1 < len(data): out.append(data[i + 1]) i += 2 else: out.append(data[i]) i += 1 return bytes(out) # ── Checksum ────────────────────────────────────────────────────────────────── def checksum(payload: bytes) -> int: """SUM8 of the **de-stuffed** payload, mod 256. 251/251 both directions. ⚠ Do NOT exclude `0x10` bytes from this sum. The DLE-aware checksum that Series III uses for its `5A` and write frames is the right answer to a *different* question: it pairs with Series III de-stuffing, which leaves escaped bytes in the payload as two bytes. De-stuffing uniformly already removes the DLE, so excluding `0x10` as well subtracts the correction twice. That combination — uniform de-stuffing *and* an exclusive sum — is what `scratch/mm_frame_parse.py` shipped with. It disagrees with the wire on **55 of 251** captured response frames, all of them frames whose payload holds a literal `0x10`. The script only ever looked correct because it accepts a frame that matches *either* rule, so it reported those 55 as plain SUM8 and never flagged one bad. """ return sum(payload) & 0xFF # ── Request builder ─────────────────────────────────────────────────────────── def build_request(sub: int, offset: int = 0, params: bytes = bytes(10)) -> bytes: """Build a host→unit command frame. ⚠ Do **not** substitute `minimateplus.framing.build_bw_frame()` here, even though the payload layout is identical. That builder escapes only `0x10`, so it reproduces just **161 of Thor's 218** captured read frames. The 57 it gets wrong are not edge cases: * every `SUB 0x5A` bulk download — `offset = 0x0400` puts a literal `0x04` in `offset_hi`, which must go out as `10 04` * `SUB 0x47` (scheduler enable), whose params carry a `0x03` An unescaped `0x03` or `0x04` reads as a frame terminator, so the unit sees a truncated frame and simply does not answer. That is indistinguishable from a dead unit, and event download would have hit it on the first try. With the correct escape set this builder reproduces **218/218**. Args: sub: command SUB byte. offset: uint16 at payload[4:5]. Micromate reads are single-step — Thor asks for `0xFFFF` and gets the whole block — so this is usually `0xFFFF`, not Series III's probe-then-data pair. params: exactly 10 bytes at payload[6:16]. A `0x10` inside `params` is fine and needs no special handling: Thor sends `SUB 0x5A` with `params = 00 00 10 00 …` and the wire carries `10 10`. (This was the spec's one open question; five captured frames settle it.) """ if len(params) != 10: raise ValueError(f"params must be exactly 10 bytes, got {len(params)}") if not 0 <= offset <= 0xFFFF: raise ValueError(f"offset must fit in uint16, got {offset:#x}") if not 0 <= sub <= 0xFF: raise ValueError(f"sub must be a single byte, got {sub:#x}") payload = bytes([MM_CMD, 0x00, sub, 0x00, (offset >> 8) & 0xFF, offset & 0xFF]) + params body = payload + bytes([checksum(payload)]) return bytes([ACK, STX]) + stuff(body) + bytes([ETX]) # ── Response frame ──────────────────────────────────────────────────────────── @dataclass class MicromateFrame: """A parsed, de-stuffed unit→host response frame.""" sub: int # response SUB; the request was 0xFF − this flags: int # payload[1] — 0xC5 Blastware line, 0x03 Thor line page_hi: int page_lo: int data: bytes # payload[5:], checksum stripped checksum_valid: bool chk_byte: int = 0 # the checksum byte as received @property def request_sub(self) -> int: """The SUB this is answering. No known exception to `0xFF − SUB`.""" return 0xFF - self.sub @property def page_key(self) -> int: """payload[3:5] as a uint16 BE — a page/address on `0x5A` responses.""" return (self.page_hi << 8) | self.page_lo @property def firmware_line(self) -> str: return {FLAGS_BLASTWARE: "blastware", FLAGS_THOR: "thor"}.get(self.flags, "unknown") @property def probe_length(self) -> Optional[int]: """Data length declared by a **probe** response: uint16 BE at data[3:5]. ⚠ Only meaningful in the reply to an `offset = 0` probe. Series III hardcodes a `DATA_LENGTHS` table; a Micromate will tell you instead, which already caught one divergence (call-home config is `0x7E`, where Series III has `0x7C`). ⚠ It is a **uint16 BE**, not a byte. Read as `data[3]` alone it is right only while the high byte is zero, and wrong by 47x for `SUB 0x1A`: a true `0x082C` (2092) reads as 44. Returns None on a frame too short to hold the field. Note this reads as 0 on the single-step reads Thor actually uses — those are not probes, and `page_key` is the meaningful field there. """ if len(self.data) < 5: return None return (self.data[3] << 8) | self.data[4] # ── Streaming parser ────────────────────────────────────────────────────────── class MicromateFrameParser: """Incremental parser for unit→host frames. Mirrors `S3FrameParser`. Feed bytes with `feed()`; completed frames are returned and also collected in `.frames`. IDLE — scanning for a bare STX IN_FRAME — collecting; bare ETX terminates AFTER_DLE — the next byte is literal, whatever it is Request frames are rejected rather than parsed: a frame whose `payload[0]` is not `0x00` is dropped, so feeding a bidirectional capture yields only the responses. """ _IDLE, _IN_FRAME, _AFTER_DLE = 0, 1, 2 def __init__(self) -> None: self._state = self._IDLE self._body = bytearray() self.frames: list[MicromateFrame] = [] # Distinguishes "no bytes at all" from "bytes but no complete frame" on # a timeout. That distinction earned its keep during the Series III # work and costs one integer here. self.bytes_fed: int = 0 def reset(self) -> None: self._state = self._IDLE self._body.clear() self.bytes_fed = 0 def feed(self, data: bytes) -> list[MicromateFrame]: self.bytes_fed += len(data) completed: list[MicromateFrame] = [] for b in data: frame = self._step(b) if frame is not None: completed.append(frame) self.frames.append(frame) return completed def _step(self, b: int) -> Optional[MicromateFrame]: if self._state == self._IDLE: if b == STX: self._body.clear() self._state = self._IN_FRAME # Boot strings, modem RING/CONNECT chatter and stray ACKs land here # and are discarded. elif self._state == self._IN_FRAME: if b == DLE: self._state = self._AFTER_DLE elif b == ETX: self._state = self._IDLE return self._finalise() else: self._body.append(b) elif self._state == self._AFTER_DLE: # Uniform rule: the escaped byte is itself, including 0x03. self._body.append(b) self._state = self._IN_FRAME return None def _finalise(self) -> Optional[MicromateFrame]: body = bytes(self._body) if len(body) < _MIN_PAYLOAD + 1: return None payload, chk_received = body[:-1], body[-1] if payload[0] != MM_RSP_CMD: return None # a request frame, or garbage that framed by accident return MicromateFrame( sub = payload[2], flags = payload[1], page_hi = payload[3], page_lo = payload[4], data = payload[5:], checksum_valid = (chk_received == checksum(payload)), chk_byte = chk_received, )