diff --git a/docs/micromate_client_spec.md b/docs/micromate_client_spec.md index b6a619f..6c57b57 100644 --- a/docs/micromate_client_spec.md +++ b/docs/micromate_client_spec.md @@ -277,7 +277,23 @@ front. --- -## `micromate/client.py` +## `micromate/client.py` — ✅ BUILT (read half) 2026-09-27 + +`connect()`, `get_state()`, `get_active_setup()`, `list_setups()` plus +`MicromateDeviceInfo` / `MicromateState` in `models.py`. 26 tests, every +response constant a real captured data section. + +⚠ **`connect()` is deliberately narrower than this spec asked for.** The spec +said to mirror Thor's `POLL → SERIAL → 0x49 → POLL` "because it is known-good". +Measurement showed the four-command form is Thor's *connection check*, present +in 3 of 8 sessions, and its fourth frame repeats its first — so `connect()` +sends the three reads that gather something. `0x01` is not read at all: Thor +never reads it, its layout is unmapped, and `firmware_line` comes free from any +response's flags byte. + +Event-chain methods (`list_events`, `download_event`, `get_event`) are step 4 +and not yet written; `MicromateProtocol.read_event_file()` already does the +download. ```python class MicromateClient: @@ -358,7 +374,7 @@ then `download_event()` and assert the bytes decode and match a 1. ✅ `framing.py` + its tests — **done 2026-09-27**, 31 tests, offline 2. ✅ `protocol.py` + its tests — **done 2026-09-27**, 35 tests, offline -3. `client.py` — `connect()`, `get_state()`, `list_setups()` +3. ✅ `client.py` + its tests — **done 2026-09-27**, 26 tests, offline 4. the event chain and `download_event()` 5. decode end-to-end and compare against a store event diff --git a/docs/micromate_protocol_reference.md b/docs/micromate_protocol_reference.md index f444e0f..7d46fed 100644 --- a/docs/micromate_protocol_reference.md +++ b/docs/micromate_protocol_reference.md @@ -1300,6 +1300,53 @@ through a record-type discriminator (`0x2C` partial vs `0x46` full) *on its event walk*, so partial records and events share one chain. Here the monitor log has its own cursor and the event chain never sees it. +#### 🔑 Every response has an 11-byte prefix — and its length byte lies + +**Content starts at `data[11]`.** One rule, every command. + +⚠ **`data[0]` is the content length `& 0xFF`, and there is no high byte +anywhere in the prefix.** `data[1]` is zero on every frame examined. So it +reads as a perfectly good length field for any response under 256 bytes — which +is most of them — and then: + +| command | true content | `data[0]` says | +|---|---|---| +| `0x1A` compliance | 2,092 | **44** | +| `0x0A` monitor log | 286 | **30** | +| `0x5A` download chunk | 1,024 | **0** | +| `0x41` setup name | 255 | 255 ✓ (only just) | + +182 of 251 captured responses agree with a naive uint16 LE read; the 69 that do +not are exactly the ones ≥ 256 bytes. + +**This is the third time a length in this protocol has been read too narrow** — +after `payload[9]` vs `payload[8:10]` in the probe response, and after +`data[0]` here. The pattern is worth naming rather than fixing case by case: +*take the content as `data[11:]` and let the frame's own length bound it.* +Nothing needs the declared length; the frame already knows how long it is. + +#### `SUB 0x5B` POLL content — where the strings actually sit + +``` +content[3] 0x50 ⚠ printable as "P", immediately before the vendor string +content[4] "Instantel\0" +content[17:26] binary — 06 00 c3 f0 4a 00 e4 19 4f 00 74 02 +content[26] "MM/ISEE/S/IO\0" ← "MM/ISEE/S" on the Thor line +``` + +⚠ **Do not parse this by scanning for printable runs.** That `0x50` at +content[3] is printable, so a run scan returns `PInstantel`. Nothing +distinguishes a length or tag byte from text by inspection. Read the vendor +from the fixed offset and find the model by searching for `MM/` — structural +rather than positional, which is what survives the model string's +firmware-line difference. + +#### `SUB 0x3F`/`0x40` setup walk — the terminator is an empty name + +23 responses on the bench unit: 22 names, then a record whose name field is all +zeros. The empty name **is** the end of list — not an error, not a setup. +`factory.MMB` first, `TEST1.mmb` last. + #### `SUB 0x01` has no Thor frame behind it Thor never reads device info in any captured session. `0xFFFF` for `0x01` comes diff --git a/micromate/client.py b/micromate/client.py new file mode 100644 index 0000000..997faca --- /dev/null +++ b/micromate/client.py @@ -0,0 +1,294 @@ +""" +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" + ) diff --git a/micromate/models.py b/micromate/models.py index 68a91a7..49e4c25 100644 --- a/micromate/models.py +++ b/micromate/models.py @@ -396,3 +396,85 @@ class IdfEvent: ) ev._waveform_key = waveform_key return ev + + +# ── Live-device models (2026-09-27) ─────────────────────────────────────────── +# +# These describe what a unit reports over the wire, not what Thor wrote to a +# file. Everything above this line came out of Thor's exports; everything below +# came out of Thor's *traffic*. Field offsets are recorded in +# ``micromate/client.py`` next to the code that reads them. + + +@dataclass +class MicromateDeviceInfo: + """Identity gathered by ``MicromateClient.connect()``. + + Sourced from three reads: + ``0x5B`` POLL → manufacturer, model + ``0x15`` SERIAL → serial + ``0x49`` STATE → monitoring + plus ``firmware_line``, which comes free from the flags byte of any + response and needs no read of its own. + """ + + serial: str + manufacturer: Optional[str] = None # "Instantel" + model: Optional[str] = None # "MM/ISEE/S/IO" (CB) / "MM/ISEE/S" (BD) + firmware_line: Optional[str] = None # "blastware" | "thor" | "unknown" + monitoring: Optional[bool] = None + active_setup: Optional[str] = None # e.g. "TEST1.mmb" + + def __str__(self) -> str: + bits = [self.serial] + if self.model: + bits.append(self.model) + if self.firmware_line: + bits.append(f"{self.firmware_line} fw") + if self.monitoring is not None: + bits.append("MONITORING" if self.monitoring else "idle") + if self.active_setup: + bits.append(f"setup={self.active_setup}") + return " ".join(bits) + + +@dataclass +class MicromateState: + """A unit's live state, from ``SUB 0x1C``. + + ``device_time`` is the unit's own clock, in its own local timezone — it is + NOT converted. Nothing else this protocol exposes reports the unit's time, + which makes it the only way to detect a drifted clock before it lands in + event timestamps. + """ + + monitoring: bool + device_time: Optional[datetime.datetime] = None + battery_volts: Optional[float] = None + memory_total_bytes: Optional[int] = None + memory_free_bytes: Optional[int] = None + raw: Optional[bytes] = field(default=None, repr=False) + + @property + def memory_used_bytes(self) -> Optional[int]: + if self.memory_total_bytes is None or self.memory_free_bytes is None: + return None + return self.memory_total_bytes - self.memory_free_bytes + + @property + def memory_used_fraction(self) -> Optional[float]: + used = self.memory_used_bytes + if used is None or not self.memory_total_bytes: + return None + return used / self.memory_total_bytes + + def __str__(self) -> str: + bits = ["MONITORING" if self.monitoring else "idle"] + if self.device_time: + bits.append(self.device_time.strftime("%Y-%m-%d %H:%M:%S")) + if self.battery_volts is not None: + bits.append(f"{self.battery_volts:.2f} V") + frac = self.memory_used_fraction + if frac is not None: + bits.append(f"memory {frac * 100:.1f}% used") + return " ".join(bits) diff --git a/tests/test_micromate_client.py b/tests/test_micromate_client.py new file mode 100644 index 0000000..ce0964c --- /dev/null +++ b/tests/test_micromate_client.py @@ -0,0 +1,396 @@ +"""Client-layer tests for the Micromate (series-4) live client. + +Every response constant below is a **real data section**, captured from UM12947 +(firmware 11.0CB) in ``bridges/captures/9-24-26 - micromate2/``. They are +embedded as hex because the captures are gitignored. + +Where a decoded value can be checked against something outside the bytes, it is: +the device clock against the capture's own filename timestamp, the battery +against Thor's event reports (3.8 V), the setup list against what the unit +displays. +""" +from __future__ import annotations + +import datetime +import os +import sys + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from micromate.client import CONTENT, MicromateClient, _content, _cstring +from micromate.framing import ETX, STX, checksum, stuff +from micromate.protocol import ProtocolError + +FLAGS_CB, FLAGS_THOR = 0xC5, 0x03 + + +# ── Captured response data sections ─────────────────────────────────────────── + +# Generated from the captures by hand-free extraction -- the hex below is +# verbatim response data, not reconstructed. The trailing comment on each +# names the capture it came from, which is what lets the clock assertions be +# checked against a wall-clock timestamp. + +POLL = bytes.fromhex( # 59 B, from_20260924_185113_ + "300000000000000000000000000050496e7374616e74656c" + "000600c3f04a00e4194f0074024d4d2f495345452f532f49" + "4f00001f603e7657603e76" +) +SERIAL = bytes.fromhex( # 21 B, from_20260924_191214_ + "0a00000000000000000000554d3132393437003100" +) +STATE_IDLE = bytes.fromhex( # 16 B, from_20260924_191214_ + "050000000000000000000000e8000b00" +) +STATE_MONITORING = bytes.fromhex( # 16 B, from_20260924_191214_ + "050000000000000000000002e8000b00" +) +MS_MONITORING = bytes.fromhex( # 55 B, from_20260924_191214_ + "2c00000000000000000000000e180907ea20130c19000000" + "000001000000000000000000000000000000000000017c00" + "e4e1c000e3f1c0" +) +MS_IDLE = bytes.fromhex( # 55 B, from_20260924_191214_ + "2c000000000000000000000000180907ea64130d22000000" + "000001000000000000000000000000000000000000017c00" + "e4e1c000e3e1c0" +) +MS_LATE = bytes.fromhex( # 55 B, from_20260925_011403_ + "2c000000000000000000000000190907ea74010e05000000" + "000001000000000000000000000000000000000000017c00" + "e4e1c000e3e1c0" +) + +_SETUP_PAD = 266 - CONTENT + + +def setup_response(name: str) -> bytes: + """A 0x41/0x3F/0x40 response: 11-byte prefix then a null-padded name.""" + body = name.encode("ascii").ljust(_SETUP_PAD, b"\x00") + return bytes([0xFF]) + bytes(10) + body + + +# The real 22 names, in the order the unit walked them. +SETUP_NAMES = [ + "factory.MMB", "TEST.MMB", "BUS TEST.MMB", "Walsh JV 241.mmb", + "Walsh JV 008.mmb", "Hawbaker 322.mmb", "Hawbaker 322 blasting.mmb", + "min.mmb", "Playhouse Loc 1.mmb", "Valley Rock Solution.MMB", + "RecordingSetup.mmb", "UPMC.mmb", "UPMC Loc 3.mmb", "Residence Inn.mmb", + "Micromate ext trigger.mmb", "Micromate remort alarm.mmb", + "Tree of Life - Loc 1 - 5861 Solway.mmb", "Mele-PWSA-Carroll -Loc 4.mmb", + "Micromate min trigger mmb.mmb", "Fay - Layton Bridge Project.mmb", + "Default Micromate ISEE.mmb", "TEST1.mmb", +] + + +# ── Test doubles ────────────────────────────────────────────────────────────── + +class ScriptedTransport: + def __init__(self, responses: list[bytes]) -> None: + self.queue = list(responses) + self.written: list[bytes] = [] + self._connected = False + + def connect(self) -> None: + self._connected = True + + def disconnect(self) -> None: + self._connected = False + + def is_connected(self) -> bool: + return self._connected + + def write(self, data: bytes) -> None: + self.written.append(data) + + def read(self, n: int) -> bytes: + return self.queue.pop(0) if self.queue else b"" + + +def frame(rsp_sub: int, data: bytes, *, flags: int = FLAGS_CB) -> bytes: + payload = bytes([0x00, flags, rsp_sub, 0x00, 0x00]) + data + return bytes([STX]) + stuff(payload + bytes([checksum(payload)])) + bytes([ETX]) + + +def client(responses: list[bytes], **kw) -> tuple[MicromateClient, ScriptedTransport]: + t = ScriptedTransport(responses) + return MicromateClient(t, recv_timeout=0.5, **kw), t + + +# ── The captured constants are what we think they are ───────────────────────── + +def test_captured_constants_have_the_expected_lengths(): + assert len(POLL) == 59 + assert len(SERIAL) == 21 + assert len(STATE_IDLE) == len(STATE_MONITORING) == 16 + assert len(MS_MONITORING) == len(MS_IDLE) == len(MS_LATE) == 55 + + +def test_the_prefix_length_byte_is_only_the_low_byte(): + """⚠ data[0] is `content_length & 0xFF`, with no high byte anywhere. + + True for every response under 256 bytes, which is why it reads as a working + length field -- and then loses 2,048 bytes on a setup block. The client + takes content as data[11:] for exactly this reason. + """ + for data in (POLL, SERIAL, STATE_IDLE, MS_MONITORING): + assert data[0] == (len(data) - CONTENT) & 0xFF + assert data[1] == 0x00, "no high byte is stored" + + # The two that prove it is not a real length: a 2,092-byte setup block + # reports 44, and a 1,024-byte download chunk reports 0. + assert (2092 & 0xFF) == 44 + assert (1024 & 0xFF) == 0 + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +def test_a_printable_byte_precedes_the_manufacturer_string(): + """content[2] is 0x50 -- "P". This is why the POLL parse cannot be a scan.""" + c = _content(POLL) + assert c[3] == 0x50 and chr(c[3]) == "P" + assert c[4:13] == b"Instantel" + + +def test_content_strips_exactly_eleven_bytes(): + assert _content(SERIAL) == bytes.fromhex("554d3132393437003100") + assert _content(b"short") == b"" + + +def test_cstring_stops_at_the_null(): + assert _cstring(bytes.fromhex("554d3132393437003100")) == "UM12947" + assert _cstring(b"\x00rest") == "" + + +# ── connect() ───────────────────────────────────────────────────────────────── + +def test_connect_decodes_identity(): + mm, t = client([ + frame(0xA4, POLL), frame(0xEA, SERIAL), frame(0xB6, STATE_IDLE), + frame(0xBE, setup_response("TEST1.mmb")), + ]) + info = mm.connect() + + assert info.serial == "UM12947" + assert info.manufacturer == "Instantel" + assert info.model == "MM/ISEE/S/IO" + assert info.firmware_line == "blastware" + assert info.monitoring is False + assert info.active_setup == "TEST1.mmb" + assert "UM12947" in str(info) and "idle" in str(info) + + +def test_connect_sends_three_reads_not_thors_four(): + """⚠ Deliberately narrower than Thor's POLL -> SERIAL -> 0x49 -> POLL. + + The trailing POLL repeats the first; measuring all 8 captured sessions + showed the four-command form is Thor's connection check (3 of 8 sessions), + not a handshake. Only "opens with POLL" is invariant. + """ + mm, t = client([ + frame(0xA4, POLL), frame(0xEA, SERIAL), frame(0xB6, STATE_IDLE), + frame(0xBE, setup_response("TEST1.mmb")), + ]) + mm.connect() + subs = [w[5] for w in t.written] # payload[2] lands at wire[5] + assert subs == [0x5B, 0x15, 0x49, 0x41] + assert 0x01 not in subs, "device info has no Thor precedent; do not read it" + + +def test_connect_can_skip_the_active_setup(): + mm, t = client([frame(0xA4, POLL), frame(0xEA, SERIAL), frame(0xB6, STATE_IDLE)]) + info = mm.connect(with_active_setup=False) + assert info.active_setup is None + assert len(t.written) == 3 + + +def test_connect_survives_an_unreadable_active_setup(): + """A unit with no setup loaded is a real state, not a failed connect.""" + mm, _ = client([ + frame(0xA4, POLL), frame(0xEA, SERIAL), frame(0xB6, STATE_IDLE), + frame(0x00, bytes(20)), # wrong SUB -> UnexpectedResponse + ]) + info = mm.connect() + assert info.serial == "UM12947" + assert info.active_setup is None + + +def test_connect_reports_a_monitoring_unit(): + mm, _ = client([ + frame(0xA4, POLL), frame(0xEA, SERIAL), frame(0xB6, STATE_MONITORING), + frame(0xBE, setup_response("TEST1.mmb")), + ]) + assert mm.connect().monitoring is True + + +def test_the_state_flag_is_tested_for_non_zero(): + """⚠ Never compared against 0x02 -- this flag family is not a stable enum. + + Its sibling in SUB 0x1C has read both 0x0E and 0x0C while monitoring. + """ + for value in (0x01, 0x02, 0x0C, 0x0E, 0xFF): + data = bytearray(STATE_IDLE) + data[CONTENT] = value + mm, _ = client([ + frame(0xA4, POLL), frame(0xEA, SERIAL), frame(0xB6, bytes(data)), + frame(0xBE, setup_response("x.mmb")), + ]) + assert mm.connect().monitoring is True, f"0x{value:02x} should read as monitoring" + + +def test_the_model_string_is_anchored_on_MM_not_on_an_offset(): + """The Thor firmware line reports a SHORTER model string, "MM/ISEE/S". + + Anchoring on b"MM/" survives that; a fixed end offset would not. A generic + printable-run scan fails for a different reason -- see _parse_poll. + """ + bd = bytearray(POLL) + assert bd[CONTENT + 26:CONTENT + 38] == b"MM/ISEE/S/IO" + bd[CONTENT + 26:CONTENT + 38] = b"MM/ISEE/S\x00\x00\x00" + mm, _ = client([ + frame(0xA4, bytes(bd), flags=FLAGS_THOR), frame(0xEA, SERIAL), + frame(0xB6, STATE_IDLE), frame(0xBE, setup_response("x.mmb")), + ]) + info = mm.connect() + assert info.model == "MM/ISEE/S" + assert info.manufacturer == "Instantel" + assert info.firmware_line == "thor" + + +# ── get_state() ─────────────────────────────────────────────────────────────── + +@pytest.mark.parametrize( + "data, monitoring, when, free", + [ + (MS_MONITORING, True, datetime.datetime(2026, 9, 24, 19, 12, 25), 0x00E3F1C0), + (MS_IDLE, False, datetime.datetime(2026, 9, 24, 19, 13, 34), 0x00E3E1C0), + (MS_LATE, False, datetime.datetime(2026, 9, 25, 1, 14, 5), 0x00E3E1C0), + ], + ids=["monitoring", "idle", "after-midnight"], +) +def test_get_state_decodes_the_real_reads(data, monitoring, when, free): + """⚠ There is an unidentified byte at content[6]; the hour is at content[7]. + + The protocol reference's 0x1C section has this right. Its one-line summary + in the divergences list reads as six contiguous fields and does not. + + Each expected time is checked against the capture filename that produced the + bytes: 19:12:14, 19:12:14 and 01:14:03. All three decode to seconds-to-a- + minute after their session opened, which is what a device clock should do. + Reading content[6] as the hour gives 32, 100 and 116. + """ + mm, _ = client([frame(0xE3, data)]) + st = mm.get_state() + + assert st.monitoring is monitoring + assert st.device_time == when + assert st.battery_volts == 3.80 # Thor's reports print 3.8 V + assert st.memory_total_bytes == 15_000_000 + assert st.memory_free_bytes == free + assert st.raw == data + + +def test_content_6_is_not_the_hour(): + """The byte the reference implies is the hour reads 32, 100 and 116.""" + for data in (MS_MONITORING, MS_IDLE, MS_LATE): + assert _content(data)[6] not in range(24) + + +def test_memory_derivations(): + mm, _ = client([frame(0xE3, MS_MONITORING)]) + st = mm.get_state() + assert st.memory_used_bytes == 15_000_000 - 0x00E3F1C0 + assert 0 < st.memory_used_fraction < 0.02 + assert "3.80 V" in str(st) + + +def test_battery_and_memory_are_read_forward_from_content_start(): + """⚠ NOT backward from the end. + + This block is 4 bytes longer on the Thor firmware line, and Series III's + from-the-end offsets give a 11.0BD unit a battery reading of 577.92 V. The + extra bytes are trailing, so appending four does not move anything. + """ + bd = MS_MONITORING + bytes.fromhex("0fa00000") + bd = bytes([0x30]) + bd[1:] # low-byte length becomes 48 + mm, _ = client([frame(0xE3, bd)]) + st = mm.get_state() + + assert st.battery_volts == 3.80, "forward offsets must survive the 4 extra bytes" + assert st.memory_total_bytes == 15_000_000 + # What the Series III from-the-end offsets would have produced: + assert int.from_bytes(bd[-10:-8], "big") / 100 == pytest.approx(577.92, abs=0.01) + + +def test_a_dead_clock_battery_does_not_fail_the_whole_read(): + """An impossible date is information; the rest of the block is still good.""" + broken = bytearray(MS_MONITORING) + broken[CONTENT + 3] = 0xFF # month 255 + mm, _ = client([frame(0xE3, bytes(broken))]) + st = mm.get_state() + assert st.device_time is None + assert st.battery_volts == 3.80 + + +def test_a_truncated_state_block_raises(): + mm, _ = client([frame(0xE3, bytes(20))]) + with pytest.raises(ProtocolError, match="need at least 44"): + mm.get_state() + + +# ── Setups ──────────────────────────────────────────────────────────────────── + +def test_list_setups_walks_to_the_empty_terminator(): + """22 real names then an empty one, exactly as the unit walked them.""" + responses = [frame(0xC0, setup_response(SETUP_NAMES[0]))] + responses += [frame(0xBF, setup_response(n)) for n in SETUP_NAMES[1:]] + responses += [frame(0xBF, setup_response(""))] + + mm, t = client(responses) + assert mm.list_setups() == SETUP_NAMES + assert len(t.written) == 23, "22 names plus the terminator" + assert t.written[0][5] == 0x3F + assert {w[5] for w in t.written[1:]} == {0x40} + + +def test_list_setups_handles_an_empty_unit(): + mm, _ = client([frame(0xC0, setup_response(""))]) + assert mm.list_setups() == [] + + +def test_list_setups_refuses_to_loop_forever(): + """A cursor that never advances is a bug, and must not hang the caller.""" + from micromate import client as C + + mm, _ = client([frame(0xC0, setup_response("a.mmb"))] + + [frame(0xBF, setup_response("a.mmb"))] * (C._MAX_SETUPS + 5)) + with pytest.raises(ProtocolError, match="not advancing"): + mm.list_setups() + + +def test_get_active_setup_handles_a_long_name(): + long_name = "Tree of Life - Loc 1 - 5861 Solway.mmb" + mm, _ = client([frame(0xBE, setup_response(long_name))]) + assert mm.get_active_setup() == long_name + + +# ── Lifecycle ───────────────────────────────────────────────────────────────── + +def test_the_client_owns_the_transport(): + mm, t = client([]) + assert not mm.is_open() + mm.open() + assert mm.is_open() and t.is_connected() + mm.close() + assert not mm.is_open() + + +def test_context_manager_opens_and_closes(): + t = ScriptedTransport([frame(0xA4, POLL), frame(0xEA, SERIAL), + frame(0xB6, STATE_IDLE), frame(0xBE, setup_response("x.mmb"))]) + with MicromateClient(t, recv_timeout=0.5) as mm: + assert t.is_connected() + assert mm.connect().serial == "UM12947" + assert not t.is_connected()