serversdown/seismo-relay
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(micromate): client layer -- connect, state, setups; read-only

Browse Source 
Step 3 of docs/micromate_client_spec.md: micromate/client.py, two models in
micromate/models.py, 26 offline tests.  Every response constant in the tests
is a real captured data section from UM12947.

Field offsets were measured rather than taken from the spec, which turned up
one general rule and one trap:

THE RESPONSE SHAPE.  Every response carries an 11-byte prefix and content
starts at data[11].  One rule, every command.

THE TRAP: data[0] is the content length & 0xFF, with no high byte anywhere in
the prefix.  It is therefore correct for every response under 256 bytes --
most of them -- and then reports 44 for a 2,092-byte setup block, 30 for a
286-byte monitor-log record, and 0 for a 1,024-byte download chunk.  182 of
251 captured responses agree with a naive read; the 69 that disagree are
exactly the ones >= 256 bytes.

That is the third length in this protocol read too narrow, after
payload[9]-vs-payload[8:10] in the probe response.  The client takes content
as data[11:] and lets the frame's own length bound it -- nothing needs the
declared length, since the frame already knows how long it is.

Also measured:

- POLL content[3] is 0x50, printable as "P", immediately before "Instantel".
  A generic printable-run scan therefore returns "PInstantel" -- it caught a
  test, not a unit.  Vendor comes from a fixed offset; the model is found by
  searching for "MM/", which is structural rather than positional and so
  survives the Thor line's shorter "MM/ISEE/S".
- The setup walk terminates on an EMPTY NAME, not an error: 23 responses, 22
  names, factory.MMB first through TEST1.mmb last.
- The 0x1C clock has an unidentified byte at content[6]; the hour is at
  content[7].  The protocol reference's 0x1C section already had this right
  and names the byte -- its one-line summary in the divergences list reads as
  six contiguous fields and is the version not to trust.  Re-verified against
  three captures: 19:12:25, 19:13:34 and 01:14:05 against filenames stamped
  19:12:14, 19:12:14 and 01:14:03.
- Battery and memory are read FORWARD from content start, never backward from
  the end.  This block is 4 bytes longer on the Thor line; the Series III
  from-the-end offsets give a 11.0BD unit 577.92 V.  A test appends the four
  trailing bytes and asserts the forward offsets survive.

connect() is narrower than the spec asked.  The spec said to mirror Thor's
POLL -> SERIAL -> 0x49 -> POLL "because it is known-good"; measurement showed
that is Thor's connection check (3 of 8 sessions) and its fourth frame repeats
its first.  So connect() sends the three reads that gather something, and
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.  If a unit ever
refuses the next command after a cold connect, put the fourth POLL back and
record it.

A dead clock battery yields device_time=None rather than failing the whole
state read; an unreadable active setup yields active_setup=None rather than
failing connect.  Both are real device states.

Still verified only against 11.0CB and only over USB.  The BD offsets follow
from the extra bytes being trailing, which is documented but not something
this code has seen.

Full suite unchanged at 16 pre-existing failures; 445 passed, up 26.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ru8Lg9HkkYvX9VWWo65SmL
This commit is contained in:
serversdown
2026-09-28 20:25:40 -04:00
co-authored by Claude Opus 5
parent a7e3a8f20a
commit f09b7dcaaf
5 changed files with 837 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/micromate_client_spec.md
+18 -2
View File
@@ -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
docs/micromate_protocol_reference.md
+47
View File
@@ -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
micromate/client.py
+294
View File
@@ -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"
)
micromate/models.py
+82
View File
@@ -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)
tests/test_micromate_client.py
+396
View File
@@ -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()