""" client.py — high-level API for a live Micromate (Series IV). Owns the transport, turns raw payloads into models. Read-only, like the layer below it: nothing here writes, erases, or changes monitoring state. with MicromateClient(TcpTransport("63.45.161.30", 9034)) as mm: info = mm.connect() print(info) # UM12947 MM/ISEE/S/IO blastware fw idle print(mm.get_state()) # idle 2026-09-25 01:14:05 3.80 V memory 0.4% used for name in mm.list_setups(): print(name) The response layout, measured rather than assumed ------------------------------------------------- **Every response carries an 11-byte prefix, and the content starts at ``data[11]``.** That one rule covers every command. ⚠ **``data[0]`` looks like the content length and is only its low byte.** A 2,092-byte setup block (`SUB 0x1A`) reports 44, and a 1,024-byte download chunk reports 0. It happens to be right for every response shorter than 256 bytes, which is most of them — so it reads as a working length field right up until it silently loses 2,048 bytes. There is no high byte anywhere in the prefix; it is ``length & 0xFF`` and nothing more. This is the same trap as ``MicromateFrame.probe_length``, in a different place, and it is now the third time a length in this protocol has been read too narrow. **Take the content as ``data[11:]`` and let the frame's own length bound it.** """ from __future__ import annotations import datetime import logging from typing import Optional from minimateplus.transport import BaseTransport from .models import MicromateDeviceInfo, MicromateState from .protocol import MicromateProtocol, ProtocolError log = logging.getLogger(__name__) # The content of every response begins here; the first 11 bytes are a prefix # whose only decoded field is an unreliable low-byte length (see module docstring). CONTENT = 11 # Field offsets, relative to the start of content. Sources are named because # two of them disagree with docs/micromate_protocol_reference.md. _STATE_FLAG = 0 # 0x49: 0x00 idle, 0x02 monitoring _MS_FLAG = 1 # 0x1C: monitoring flag — test NON-ZERO _MS_DAY, _MS_MONTH, _MS_YEAR = 2, 3, slice(4, 6) _MS_UNKNOWN_6 = 6 # ⚠ NOT the hour — see read note below _MS_HOUR, _MS_MIN, _MS_SEC = 7, 8, 9 _MS_BATTERY = slice(34, 36) # uint16 BE, volts × 100 _MS_MEM_TOTAL = slice(36, 40) # uint32 BE _MS_MEM_FREE = slice(40, 44) # uint32 BE # A setup-list walk that does not terminate is a bug, not a big fleet. The # bench unit holds 22 setups; this is a generous ceiling, not a limit. _MAX_SETUPS = 512 def _content(data: bytes) -> bytes: """Strip the 11-byte response prefix.""" return data[CONTENT:] if len(data) > CONTENT else b"" def _cstring(buf: bytes, offset: int = 0) -> str: """A null-terminated ASCII run, stripped.""" return buf[offset:].split(b"\x00")[0].decode("ascii", "replace").strip() class MicromateClient: """High-level read-only client for one Micromate. Owns the transport, unlike ``MicromateProtocol``, which borrows it. """ def __init__( self, transport: BaseTransport, recv_timeout: float = 10.0, strict_checksums: bool = True, ) -> None: self._transport = transport self._proto = MicromateProtocol( transport, recv_timeout=recv_timeout, strict_checksums=strict_checksums ) self._firmware_line: Optional[str] = None # ── Lifecycle ───────────────────────────────────────────────────────────── def open(self) -> None: self._transport.connect() def close(self) -> None: self._transport.disconnect() def is_open(self) -> bool: return self._transport.is_connected() def __enter__(self) -> "MicromateClient": self.open() return self def __exit__(self, *_) -> None: self.close() @property def protocol(self) -> MicromateProtocol: """The wire layer, for anything this class does not wrap yet.""" return self._proto # ── Identity ────────────────────────────────────────────────────────────── def connect(self, *, with_active_setup: bool = True) -> MicromateDeviceInfo: """`POLL → SERIAL → state`, plus the active setup name. ⚠ **This is deliberately not Thor's full preamble.** Thor sends `POLL → SERIAL → 0x49 → POLL` and the client spec said to copy it verbatim on the grounds that it is known-good. Measuring all 8 captured sessions showed the only invariant is that a session **opens with POLL** — the four-command form appears in 3 of 8 and is Thor's *connection check*, run where it wants to refresh what it displays. The trailing POLL is a repeat of the first. So this sends the three reads that actually gather something. Dropping the fourth is a judgement call on measured evidence, not a proof that nothing depends on it; if a unit ever refuses the next command after a cold connect, put it back and say so in the protocol reference. `SUB 0x01` (device info) is **not** read. Thor never reads it in any captured session, its field layout is unmapped beyond eight `1.0f` floats, and `firmware_line` — the one thing we would want from it — comes free from the flags byte of any response. """ poll = self._proto.poll() self._firmware_line = poll.firmware_line manufacturer, model = self._parse_poll(poll.data) serial = _cstring(_content(self._proto.read_serial())) monitoring = self._parse_state(self._proto.read_state()) info = MicromateDeviceInfo( serial=serial, manufacturer=manufacturer, model=model, firmware_line=poll.firmware_line, monitoring=monitoring, ) if with_active_setup: try: info.active_setup = self.get_active_setup() except ProtocolError as e: # Not worth failing a connect over: a unit with no setup loaded # is a real state, and the caller can still read everything else. log.warning("active setup unreadable: %s", e) log.info("connected: %s", info) return info @staticmethod def _parse_poll(data: bytes) -> tuple[Optional[str], Optional[str]]: """Manufacturer and model out of the POLL block. `Instantel` sits at content[4] and the model at content[26], with 13 binary bytes between them. ⚠ A generic "find the printable runs" scan does **not** work here, which cost a test failure before it cost anything worse. content[3] is `0x50` — printable as `P` — sitting immediately before `Instantel`, so a run scan returns `PInstantel`. Nothing distinguishes a length or tag byte from text by inspection. So: the manufacturer comes from a fixed offset, and the model is found by searching for `MM/`. That anchor is structural rather than positional, which matters because the model string **differs by firmware line** — `MM/ISEE/S/IO` on the Blastware build, `MM/ISEE/S` on the Thor build — and only its tail changes. """ c = _content(data) manufacturer = _cstring(c, 4) or None idx = c.find(b"MM/") model = _cstring(c, idx) if idx >= 0 else None return manufacturer, model @staticmethod def _parse_state(data: bytes) -> Optional[bool]: """`SUB 0x49` content[0]: 0x00 idle, 0x02 monitoring. ⚠ Tested for non-zero, never against `0x02`. The sibling flag in `SUB 0x1C` has read both `0x0E` and `0x0C` while monitoring, so this family of flags is not a stable enum. """ c = _content(data) return bool(c[_STATE_FLAG]) if c else None # ── State ───────────────────────────────────────────────────────────────── def get_state(self) -> MicromateState: """`SUB 0x1C` — monitoring, device clock, battery, memory. ⚠ Every offset here is **forward from the start of content**, never backward from the end. Series III reads battery and memory from the end of this block, and this block is **4 bytes longer on the Thor firmware line** — applying from-the-end offsets to a `11.0BD` unit yields a battery voltage of 577.92 V. The four extra bytes are trailing, so from-the-start offsets hold for both lines. ⚠ Verified on `11.0CB` only. That the same offsets hold on `11.0BD` follows from the extra bytes being trailing, which is documented but not something this code has seen. """ data = self._proto.read_monitor_status() c = _content(data) if len(c) < 44: raise ProtocolError( f"monitor status content is {len(c)} B, need at least 44" ) battery = int.from_bytes(c[_MS_BATTERY], "big") / 100.0 return MicromateState( monitoring=bool(c[_MS_FLAG]), device_time=self._parse_clock(c), battery_volts=battery, memory_total_bytes=int.from_bytes(c[_MS_MEM_TOTAL], "big"), memory_free_bytes=int.from_bytes(c[_MS_MEM_FREE], "big"), raw=data, ) @staticmethod def _parse_clock(c: bytes) -> Optional[datetime.datetime]: """The unit's own clock, in its own local time. ⚠ **content[6] is not part of the time.** The layout is day, month, year, *one unidentified byte*, then h/m/s — so the hour is at content[7]. The protocol reference's `SUB 0x1C` section has this right and names `data[17]` as unidentified; its one-line summary in the divergences list ("day/month/year/h/m/s at `data[13:21]`") reads as six contiguous fields and is the version worth not trusting. Re-measured here across three captures: content[6] read 32, 100 and 116, none a valid hour, while content[7:10] gave 19:12:25, 19:13:34 and 01:14:05 against capture filenames stamped 19:12:14, 19:12:14 and 01:14:03 — each seconds to a minute after its session opened, which is what a device clock should do. content[6] is undecoded and deliberately not exposed. """ try: return datetime.datetime( year=int.from_bytes(c[_MS_YEAR], "big"), month=c[_MS_MONTH], day=c[_MS_DAY], hour=c[_MS_HOUR], minute=c[_MS_MIN], second=c[_MS_SEC], ) except ValueError as e: # A unit with a dead clock battery reports an impossible date. That # is information, not a reason to fail the whole state read. log.warning("device clock unreadable (%s): %s", e, c[2:10].hex(" ")) return None # ── Setups ──────────────────────────────────────────────────────────────── def get_active_setup(self) -> str: """`SUB 0x41` — the loaded `.MMB` file name, e.g. `TEST1.mmb`.""" return _cstring(_content(self._proto.read_active_setup_name())) def list_setups(self) -> list[str]: """`0x3F` then `0x40`… — every setup file stored on the unit. A cursor walk: the device holds the position, so the same `0x40` request returns the next name. **An empty name terminates the list** — it is not an error and not a real setup. Measured on the bench unit: 23 responses, 22 names then the empty one, `factory.MMB` first through `TEST1.mmb` last. """ names: list[str] = [] raw = self._proto.read_first_setup() for _ in range(_MAX_SETUPS): name = _cstring(_content(raw)) if not name: return names names.append(name) raw = self._proto.read_next_setup() raise ProtocolError( f"setup list did not terminate after {_MAX_SETUPS} entries — the " f"device cursor is not advancing" )