Author SHA1 Message Date
serversdown d2258dad4e Merge pull request 'Release v0.31.0 — report parity + the inverted rescue (0.29.0 → 0.31.0)' (#40) from dev into main
Reviewed-on: #40
2026-09-18 16:46:41 -04:00
serversdown e050e602c2 Merge pull request 'Release v0.29.0 — offset detector + false_trigger_reason + BlastMate serials (0.27.0→0.29.0)' (#36) from dev into main
Reviewed-on: #36
2026-09-07 15:57:37 -04:00
serversdown fc1c7c7936 Merge pull request 'Docs/claude.md corrections' (#35) from dev into main
Reviewed-on: #35
2026-08-29 15:48:33 -04:00
serversdown 7d0d12079b Merge pull request 'v0.27.0 Decoder fixes, offset exploration and testing.' (#34) from dev into main
Reviewed-on: #34
2026-08-28 22:40:34 -04:00
serversdown 5203aab849 Merge pull request 'update to 0.26.0. Big chonking update including 0.23, 0.24, and 0.25 as well.' (#33) from dev into main
Reviewed-on: #33
2026-08-27 13:43:00 -04:00
serversdown 5b65718b72 Merge pull request 'v0.23.0 - ZC freq in events store' (#31) from dev into main
Reviewed-on: #31
2026-08-06 12:12:19 -04:00
serversdown 483762607e Merge pull request 'Update to v0.22.0' (#30) from dev into main
Reviewed-on: #30
2026-07-03 15:35:20 -04:00
serversdown d0b66368d5 Merge pull request 'update to v0.21.1, thor data import successful' (#29) from dev into main
Reviewed-on: #29
2026-06-01 16:54:23 -04:00
serversdown 2eb1d25028 Merge pull request 'v0.20.0 -- Full s3 event parse and PDF creation.' (#28) from dev into main
Reviewed-on: #28
2026-05-28 17:54:31 -04:00
claude cc821f9ee3 hotfix: fix dockerfile on main to fix import bug on prod 2026-05-21 20:42:15 +00:00
24 changed files with 24 additions and 9123 deletions
-215
View File
@@ -4,221 +4,6 @@ All notable changes to seismo-relay are documented here.
---
## Unreleased
### Fixed
- **Waveform event times were the monitoring-session start, not the trigger
(~hours off).** `read_blastware_file` stamped events with footer `ts1`, which
for a waveform is the session start a unit shares across every event that day
(a unit arming at 06:00 stamped 06:00 on all of them — the modal and PDF both
showed it, since it's the stored value). The event time is footer `ts2` (the
recording stop), and Blastware's trigger = `ts2 - record time`. The record
time is a big-endian float32 in the recording-setup config block (30 bytes
before the `Standard Recording Setup` marker), so the **exact trigger is now
recovered from the binary alone** — all 7 BE12844 oracle events decode to
their exact Blastware time (e.g. N844LQHB 10:33:29), no paired `.TXT` needed.
Histograms keep `ts1` (the ~24 h window start). A paired report's
`event_datetime` stays authoritative (unit-clock drift).
⚠ **Needs a re-decode backfill** to correct existing stored events' timestamps.
- **`scratch/mm_frame_parse.py` accepted either checksum rule, so it could not
falsify either.** It tried plain SUM8 *and* a DLE-aware variant and reported
whichever matched, which is why it never flagged a bad frame and why the
protocol reference carried the wrong rule for two days — "zero bad checksums
across three sessions" was true and carried no information. It now validates
against SUM8 alone and names the DLE-aware result only as a near-miss, never as
a pass. Still zero bad frames across all captures, strictly tighter.
### Added
- **Diagnostics tab in the SFM standalone webapp.** Surfaces the device
endpoints that previously existed only as `curl`: `events/storage_range` and
`events/index` alongside `monitor/status`, then stop monitoring, disable ACH
(`rescue?erase=false`, so stored events survive), and erase. The wedged-unit
ladder — slow drip and blind stop — sits under its own heading pointing at
`docs/runbooks/wedged_unit_recovery.md`, with the reminder that `slow_drip`'s
success signal is `bytes_received > 0` and not a clean duration. Erase is
guarded by typing the unit's serial: auth answers *who*, not *did you mean
it*, and Swagger's try-it-out button on `/device/events/erase` is live on
`:8200/docs`.
- **`docs/sfm_tool_status.md`** — an honest per-capability maturity assessment:
what is production-grade (the codec library, the data side), what is
emergency-grade (the device side), what is a research artifact, the
known-issues table, and the gap to a real tool. Also records the **5A
page-boundary bug** as known: `parse_strt_end_offset()` discards the key's
page byte, so once a unit has recorded more than 64 KB since its last erase,
an event spanning the boundary reads an `end_offset` *behind* its own start —
the chunk loop fetches nothing and TERM packs a negative `offset_word`, which
500s. Reproduced on BE12599. Production is unaffected: it ingests complete
files via the watcher path and never runs this walk.
- **The Micromate (Series IV) live wire protocol, reverse-engineered end to
end** — `docs/micromate_protocol_reference.md`. Worked out against a bench
UM12947 over USB and a recording relay, with THOR driving every write so that
no command has ever been originated against a unit by this project. **A
Micromate answers Series III command frames**, with three framing differences:
responses carry no leading `DLE`, `payload[1]` is `0xC5` (Blastware firmware)
or `0x03` (Thor firmware) rather than `0x10`, and the data length is a
**uint16 BE at `payload[8:10]`** — read as a single byte it under-reads
`SUB 0x1A` by 47x. Read path, event chain, and `SUB 0x5A` streaming the
`.IDFW` file verbatim are all confirmed.
- **Series IV setup management, fully mapped.** `0x41` reads the active setup
name, `0x1A` its config block, `0xDA` names the target `.MMB`, `0x71`/`0x72`
write it back. **Setups are read-modify-write** — the written block is the
read block, 91% byte-identical at a fixed 11-byte shift. `0xDA` **creates**
files rather than only overwriting, confirmed on the unit's own screen, and an
overwrite is protocol-identical to a create: no handshake, no warning, and no
protection even over the *active* setup of a monitoring unit.
- **The scheduler file decoded** — `\system\schedule\schedule.dat`, 260-byte
records carrying an action bitmask (2 start, 4 stop, 8 self-check, 16 ACH), a
half-hour slot (48/day), day-of-week (0 = Sunday) and a length-prefixed setup
name. Verified entry-for-entry against the operator's own THOR screen.
- **A generic file transfer addressed by full path** — `0x94`/`0x48` read,
`0x8D`/`0x8E` write. This **retracts** an earlier conclusion in the same
document that no such command existed; that was inferred from absent firmware
strings and was wrong.
- **Monitoring control and per-event delete.** `0x96`/`0x97` start and stop as
on Series III, but the monitoring flag at `SUB 0x1C` `data[12]` must be tested
for **non-zero** (observed as both `0x0E` and `0x0C`) rather than compared to a
constant. Deletion is **per-event** — `0xA8` with the event key, then `0xAA` —
which is safer than Series III's erase-everything. `SUB 0x1C` also carries the
device clock.
- **`bridges/mm_probe.py`** — distinguishes the four faults THOR reports
identically as "disconnected": refused, connect timeout (the silent-drop
signature of a trusted-IP whitelist), **connected but no reply** (the modem
answered and the unit did not), and replied. Each verdict names what to try
next. `--slots N` tests single-session modem behaviour.
- **`bridges/mm_link.py`** — a bench stand-in for a cellular modem, with a
decoded timestamped log and fault injection (`blackhole`, `drop`, `delay`,
`onewaydev`) driven by a control file. No pyserial; stdlib `termios` only.
- **`scratch/mm_frame_parse.py`**, **`socat_log_split.py`** and **`fake_unit.py`**
— a Micromate-aware frame parser (`S3FrameParser` cannot see these responses at
all, since it scans for `DLE+STX`), byte-exact capture recovery from a
`socat -x` relay log, and a serial-port stand-in that answers as a unit.
- **A live client for Series IV — `micromate/{framing,protocol,client}.py`.**
`micromate/` was codec-only; it can now talk to a unit. Connect over TCP (a
cellular modem) or serial/USB, identify a unit, read its state, clock, battery,
memory and setups, walk the event chain and download events as `.IDFW`/`.IDFH`
bytes the existing codec already decodes. **98 offline tests**, every response
constant a real captured data section. `minimateplus/transport.py` is reused
as-is; `minimateplus/framing.py` deliberately is **not** — see *Changed*.
⚠ **Read-only.** Setups, schedules, call-home config, monitoring start/stop
and per-event delete are all mapped and none are implemented. No command has
ever been originated against a unit by this project; every write in the
protocol reference was performed by THOR while we recorded.
- **Verified on real hardware, both transports, both firmware lines.** Identical
wire bytes over USB CDC-ACM and an RX55 in PAD mode — 11,580 B in and 761 B out
to the byte. Every `11.0BD` inference confirmed on UM20147, including the four
extra trailing bytes in `SUB 0x1C` that make Series III's from-the-end offsets
report a battery voltage of **577.92 V**. A monitoring unit answers reads with
no `SESSION_RESET`, which Series III requires. All six bench events decode
with the existing IDF codec — 4 waveforms at 12,288/12,288/12,288/8,192 samples
and 2 histograms — so `/db/import/idf_file` ingests a directly downloaded event
unchanged.
- **`bridges/mm_client_check.py`** — drives the read client against a unit and
reports per-command timings, read counts and byte totals, so two transports or
two firmware lines can be diffed. `--capture DIR` writes a `raw_bw_*`/`raw_s3_*`
pair in the layout `scratch/mm_frame_parse.py` reads, turning a field run into
a test fixture. No pyserial — stdlib `termios`, because `pip install` is
refused outright by PEP 668 on the distros the bench hosts run.
### Changed
- **Connecting to a unit no longer walks its event chain.** `/device/events`
reads every event header over the cellular link; on a unit with a large or
wrapped chain that takes minutes or fails outright, and it fired
automatically on every connect. Connect now uses only ~2 s probes —
`/device/info` (which already carried the compliance config the walk was
re-reading) plus `events/storage_range` — and the Device tab gains an Event
Chain card. The walk moved behind a **Load events** button in the Events
toolbar. Knowing whether a unit's ACH is on no longer requires reading every
event it has stored.
- **Recorded what THOR actually does on the wire**, measured rather than assumed.
A "status check" is **eleven commands, ~2.2 KB including TCP setup** — 18.8
MB/day per unit at a 10 s cadence, against ~0.2 MB/day for a `POLL` +
`MONITOR_STATUS` check at 60 s. The **status interval is honoured; the
connection interval is not** — it sets `(status / connection) - 1` checks per
cycle, so equal values yield *zero* cheap checks and every connection becomes
the expensive one.
- **Two THOR defects reproduced with timestamps.** After a connection drops
mid-download it retries **once**, stops polling entirely and **never resumes**,
while displaying `Connected` for as long as it is left alone — and `Idle` for a
unit that is actively recording. Separately, THOR's own log shows a
**subscription leak**: one logical event dispatched to a growing number of
handlers, **1 to 12 over ten hours** of uptime, consistent with the field
report that only a restart recovers it.
- **The Micromate's USB host supports FTDI and CDC-ACM only** — no Prolific, in
either firmware line. A PL2303 cable (Benfei) leaves a unit with no working
modem port; an FTDI cable (Sabrent) works. Both are in circulation and
indistinguishable by eye — identify by `lsusb` VID, `0403` against `067b`.
- **Six rules in the Series-4 client spec were wrong, and measuring against the
captures caught all six before any code shipped.** Each fails *silently* —
a frame the unit ignores, or a checksum that reads as bad:
requests escape **four** byte values (`0x02 0x03 0x04 0x10`), not one, so the
Series III builder reproduces only **161 of THOR's 218** read frames and misses
*every* `SUB 0x5A` download; the response checksum is **plain SUM8** of the
de-stuffed payload, not the DLE-aware variant, which disagrees with the wire on
**55 of 251** frames; `SUB 0x5A` is a **1024-byte chunk loop**, not one request
per event; `SUB 0x0A` is the **monitor-log walk** (same request repeated,
device-side cursor), not a keyed event-header read; `1E`/`1F` carry **token
`0xFE`** at `params[7]`; and `SUB 0x01` has **no THOR precedent at all**.
Recorded in `docs/micromate_protocol_reference.md` as corrections in place.
- **`SUB 0x0C`'s peak float is the per-sample peak vector sum** — resolved after
being marked *do-not-rely-on*. Exact to **0.000%** on all four bench waveforms
against a PVS recomputed from decoded samples, so its offset (`Tran` label − 12)
is established rather than inferred. The earlier "not the vector sum" reading
compared against `sqrt(Σpeak²)` from the three *reported* peaks, which is an
upper bound: the channel maxima do not occur at the same instant. **This gives
the decoder a free self-check** — the device computed that number from the same
samples, independently of our codec, so a mismatch means the decode is wrong.
Worth having in a codebase whose channel truncations have historically been
silent.
- **Cellular costs ~0.65 s per round trip, independent of payload size.**
Measured on UM12947 over an RX55: `list_setups()` on a unit with 24 setups takes
**16.0 s**, against 0.46 s over USB, for 24 commands. A 1,024-byte chunk and a
16-byte state read cost the same. **Over cellular, minimise round trips, not
bytes** — cache the setup list rather than refreshing it on a timer, and note a
13 KB event is 14 chunks ≈ 8.4 s of latency against ~0.03 s of data. The modem
also needs **fewer** reads than USB, not more: it buffers ~1 s then forwards one
large segment where CDC-ACM delivers many small ones.
- **Event keys collide across units.** UM20147's first event and UM12947's first
event are both `055d4a81` — different sizes, different contents, identical key.
The counter starts from the same value on every unit, so **a key is meaningless
without its serial**. Anything that stores, deduplicates or addresses Series IV
events must key on `(serial, event_key)`; keyed on the event key alone, one
unit's event is silently treated as a duplicate of another's and simply never
ingested. Series III has the cousin of this — its counter resets after an
erase, so keys are reused *within* a unit, which is why `ach_state.json` already
tracks `max_downloaded_key` per serial.
- **Setup file names are spelled three ways by three commands.** In one session
on one unit: `0x41` reported `test2.MMB`, `0x40` reported `test2.mmb`, and `0x0C`
reported `test2`. **Compare setup names case-insensitively and without the
extension** — an exact-string test of "is the active setup one I know about?"
answers *no*.
### Migration
**None.** No codec change, no waveform-store change, no DB or schema change,
and **no `TOOL_VERSION` bump** — so no `backfill_sidecars.py` /
`backfill_event_shape.py` run is owed. The webapp is served from the image, so
its changes appear after the next `sfm` rebuild.
The Series-4 live client adds **new** modules under `micromate/`
(`framing.py`, `protocol.py`, `client.py`) and appends two dataclasses to
`micromate/models.py`; `micromate/idf_file.py` — the codec — is untouched, and
nothing under `sfm/` or `minimateplus/` changed. The new modules are additive
and nothing imports them yet, so an existing deployment behaves identically.
---
## v0.31.0 — 2026-09-18
**Report parity, and a second way to rescue a runaway unit.** Two threads.
+3 -70
View File
@@ -10,38 +10,10 @@ pair — lives in `../terra-view/docs/tmi-stack.md`, which is also loaded as
---
## Where things stand (updated 2026-09-26)
## Where things stand (updated 2026-08-28)
Read this first when picking the project back up.
- **The Series-4 LIVE wire protocol is reverse-engineered end to end
(2026-09-25).** `docs/micromate_protocol_reference.md` is the Series-4
Rosetta Stone, sibling to `instantel_protocol_reference.md`. **A Micromate
answers Series III command frames** — three framing differences: responses
have **no leading `DLE`** (a bare `STX`), `payload[1]` is `0xC5` (Blastware
firmware) or `0x03` (Thor firmware) rather than `0x10`, and the data length
is a **uint16 BE at `payload[8:10]`** (as a byte it under-reads `SUB 0x1A`
by 47x). Read path, event chain, setups, scheduler, monitoring control and
per-event delete are all mapped; **the inbound call-home session is the only
protocol unknown left.**
⚠ **No command has ever been originated against a unit by this project.**
Every write was performed by THOR while we recorded. That line is worth
keeping.
⚠ `micromate/` still has **no live client** — it is codec-only. The
`minimateplus/` stack (transport/framing/protocol/client) has no Series-4
counterpart yet. `minimateplus.transport` is protocol-agnostic and reusable.
- **Bench tooling for device diagnosis (2026-09-25).** `bridges/mm_probe.py`
distinguishes the four faults THOR reports identically as "disconnected"
(refused / connect timeout / **connected but no reply** / replied) and names
what to try next. `bridges/mm_link.py` is a stand-in for a cellular modem
with a decoded log and fault injection. `scratch/mm_frame_parse.py` exists
because **`S3FrameParser` cannot see Micromate responses at all** — it scans
for `DLE+STX`, which never appears in Series-4 traffic.
- **A Micromate's USB-A host port drives FTDI and CDC-ACM only** — no Prolific,
in either firmware line. TMI buys both Sabrent (FTDI) and Benfei (PL2303)
cables and they are indistinguishable by eye. A PL2303 cable leaves a unit
with **no working modem port at all**; identify by `lsusb` VID, `0403` vs
`067b`. This accounted for a unit that could not be deployed.
- **Series-3 decode is verified per-sample at scale (v0.27.0).** The full DL2
archive decodes **14,338 / 14,338** paired files exactly against their
preserved Blastware ASCII exports — 1,249 waveform + 13,089 histogram, 45
@@ -89,24 +61,6 @@ Read this first when picking the project back up.
4th-decimal tick and are Thor's own rounding — no single linear LSB can
reproduce every printed value (the constraints are infeasible by 7e-5
relative), so do NOT retune `_GEO_LSB_IPS`.
- **⚠ KNOWN BUG — the 5A walk breaks once a unit's buffer crosses 64 KB.**
`parse_strt_end_offset()` returns only `(end_key[2] << 8) | end_key[3]`,
discarding the key's page byte. An event starting at `0x0111F2A2` and ending
at `0x0112_1010` therefore reads `end_offset = 0x1010` — *behind* its own
start. The chunk loop then exits before fetching anything and TERM computes
a negative `offset_word`, which `struct.pack(">H", ...)` rejects: the
`/device/events` walk 500s. Reproduced on BE12599 (2026-09-19), which had
78 KB stored and had rolled into page `0x12`.
**Why it hid so long:** every 5A capture the walk was verified against came
from a freshly-erased BE11529 — all three confirmed TERM examples in
`framing.py` (`0x1ABE`, `0x21F2`, `0x417E`) sit inside page `0x11`. Prod is
unaffected: it ingests complete files via BW ACH, never this walk.
**Fixing it has two layers** — the arithmetic (`if end < start: end +=
0x10000`) stops the crash and bounds the loop correctly; carrying the page
byte through the chunk requests (`params[1]` 0x11 -> 0x12, counter rolling
over) needs a BW capture of a spanning event first. Do not ship layer one
alone without a loud truncation warning — a silently short event is the
failure mode this codec has been bitten by repeatedly.
- **Open, not blocking:** 14 sensitive-range files show an exact 8x
(= 10.0/1.25) units discrepancy; `scripts/backfill_sidecars.py --force` also
inserts DB rows for store files that have none (one-time per store) and the
@@ -120,9 +74,6 @@ Read this first when picking the project back up.
**v0.27.0 does NOT owe prod a backfill** — verified: the partial-final-block
fix changes 0 of the 10,215 histograms in the prod store (the 4 recovered
files are archive-only and were never ingested).
✅ **The v0.30.0 Series-4 backfill HAS been run on prod (2026-09-25).** Every
stored Series-4 geophone value was ~3.3% low until then; that is corrected and
the job does not need repeating.
- **The "offset" hardware fault has its own journal** --
`docs/offset_investigation.md`. **5 of 45 units (11%)**, and the fault is
**persistent** — it stays until the geophone is serviced. Detect it with
@@ -134,18 +85,7 @@ Read this first when picking the project back up.
`SUB 0x0E` (unimplemented), which may carry those very numbers.
When new information about a protocol is discovered, record it in the matching
reference **in addition to** this document:
| series | document |
|---|---|
| Series III (MiniMate Plus / BlastMate) | `docs/instantel_protocol_reference.md` |
| **Series IV (Micromate / THOR)** | **`docs/micromate_protocol_reference.md`** |
| Thor IDF file format | `docs/idf_protocol_reference.md` |
Both protocol references carry retractions in place rather than deleting what
turned out to be wrong — that convention has already saved re-deriving the same
mistakes twice, so keep it.
When new information about the protocol is discovered, please update the instantel_protocol_reference.md with the findings in addition to this document
---
@@ -318,15 +258,8 @@ minimateplus/ ← Python client library (primary focus)
sfm/server.py ← FastAPI REST server exposing device data over HTTP
seismo_lab.py ← Tkinter GUI (Bridge + Analyzer + Console tabs)
bridges/
mm_probe.py ← name the fault behind a dead unit (4 verdicts, read-only)
mm_link.py ← bench stand-in for a cellular modem, with fault injection
ach_mitm.py ← TCP relay for recording a Series-3 ACH session
docs/
instantel_protocol_reference.md ← Series III protocol spec ("the Rosetta Stone")
micromate_protocol_reference.md ← Series IV protocol spec + THOR's measured behaviour
idf_protocol_reference.md ← Thor IDF file format
instantel_protocol_reference.md ← reverse-engineered protocol spec ("the Rosetta Stone")
CHANGELOG.md ← version history
```
-5
View File
@@ -496,11 +496,6 @@ Use **com0com** or **VSPD** to create the virtual COM pair on Windows.
## Roadmap (Future)
> **Where it stands *today*** — an honest per-capability maturity assessment,
> what to rely on, known issues, and the gap to a real tool:
> [`docs/sfm_tool_status.md`](docs/sfm_tool_status.md). This section covers
> where it is *going*.
### Strategic direction — where this is going
seismo-relay is being built as a **suite of cooperating components**
-393
View File
@@ -1,393 +0,0 @@
#!/usr/bin/env python3
"""
mm_client_check.py — exercise the Micromate read client against a real unit.
**Read-only.** It sends POLL, SERIAL, state, monitor status, the setup walk and
(optionally) one event download. It never writes, never erases, never starts or
stops monitoring.
Why it exists
-------------
`micromate/{framing,protocol,client}.py` are verified against captures taken
**over USB**, on **one firmware line** (`11.0CB`). Two things that cannot be
verified that way:
* **the modem path.** An RX55/RV55 bridges serial to TCP transparently, but
it buffers up to ~1 s before forwarding, so a single logical response can
arrive as many small reads. The client reads to frame completion rather
than using idle-gap detection, which should be strictly more robust — but
"should be" is the point of this script.
* **the other firmware line.** `11.0BD` reports `flags = 0x03`, a shorter
model string, and a `SUB 0x1C` block 4 bytes longer. Everything about that
is currently inference from one 2026-09-23 sweep whose captures never
landed in the repo.
Run it over both paths and diff the two reports. Anything that differs beyond
timings is a finding.
Usage
-----
# over the modem
python3 bridges/mm_client_check.py 63.45.161.30:9034
# over USB / direct serial
python3 bridges/mm_client_check.py /dev/ttyACM0 --baud 115200
# include one event download (still read-only)
python3 bridges/mm_client_check.py <target> --download
⚠ These modems bridge ONE TCP session to serial at a time. If THOR holds the
unit, this will connect and then see nothing — that is contention, not a fault.
`bridges/mm_probe.py` explains that case; disconnect THOR first.
"""
from __future__ import annotations
import argparse
import errno
import os
import select
import sys
import termios
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from micromate.client import MicromateClient, _content # noqa: E402
from micromate.protocol import ProtocolError # noqa: E402
from minimateplus.transport import TcpTransport # noqa: E402
class StdlibSerial:
"""Raw serial on stdlib `termios` — no pyserial.
`minimateplus.SerialTransport` needs pyserial, and a bench host is whatever
is to hand. On a PEP 668 distro (Mint 22, Ubuntu 24.04, Debian 12) a plain
`pip install pyserial` is refused outright, so a diagnostic that depends on
it is one you cannot run at the moment you need it. `bridges/mm_link.py`
and `scratch/fake_unit.py` already take this approach; this is the same
~30 lines, and it means the tool runs on a stock Python 3 anywhere.
Not a general replacement for SerialTransport — no flow control, no
parity options, Linux/macOS only. Enough for a Micromate, which is 8N1
with no handshaking.
"""
_BAUD = {9600: termios.B9600, 19200: termios.B19200, 38400: termios.B38400,
57600: termios.B57600, 115200: termios.B115200}
def __init__(self, path: str, baud: int = 115200) -> None:
if baud not in self._BAUD:
raise ValueError(f"unsupported baud {baud}; pick from {sorted(self._BAUD)}")
self.path, self.baud, self.fd = path, baud, None
def connect(self) -> None:
if self.fd is not None:
return
self.fd = os.open(self.path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
a = termios.tcgetattr(self.fd)
a[0] = a[1] = a[3] = 0 # raw in/out, non-canonical
a[2] = termios.CS8 | termios.CREAD | termios.CLOCAL # 8N1, ignore modem lines
a[4] = a[5] = self._BAUD[self.baud]
a[6] = list(a[6])
a[6][termios.VMIN] = 0
a[6][termios.VTIME] = 0
termios.tcsetattr(self.fd, termios.TCSANOW, a)
termios.tcflush(self.fd, termios.TCIOFLUSH)
def disconnect(self) -> None:
if self.fd is not None:
os.close(self.fd)
self.fd = None
def is_connected(self) -> bool:
return self.fd is not None
def read(self, n: int) -> bytes:
if self.fd is None:
return b""
r, _, _ = select.select([self.fd], [], [], 0.05)
if not r:
return b""
try:
return os.read(self.fd, n)
except OSError as e:
if e.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
return b""
raise
def write(self, data: bytes) -> None:
if self.fd is None:
raise OSError("port is not open")
while data:
data = data[os.write(self.fd, data):]
class _Timed:
"""Count bytes and time each read, so the two transports can be compared.
With `capture`, also writes the raw byte streams to a `raw_bw_*` /
`raw_s3_*` pair in the layout `scratch/mm_frame_parse.py` already reads --
so a run on an unfamiliar unit can be turned into test fixtures without
setting up a relay.
"""
def __init__(self, inner, capture: str | None = None) -> None:
self._inner = inner
self.reads = 0
self.bytes_in = 0
self.bytes_out = 0
self._bw = self._s3 = None
if capture:
stamp = time.strftime("%Y%m%d_%H%M%S")
d = Path(capture)
d.mkdir(parents=True, exist_ok=True)
self.bw_path = d / f"raw_bw_{stamp}_mm_client_check.bin"
self.s3_path = d / f"raw_s3_{stamp}_mm_client_check.bin"
self._bw = open(self.bw_path, "wb")
self._s3 = open(self.s3_path, "wb")
def close_capture(self) -> None:
for f in (self._bw, self._s3):
if f:
f.close()
def connect(self):
return self._inner.connect()
def disconnect(self):
return self._inner.disconnect()
def is_connected(self):
return self._inner.is_connected()
def write(self, data: bytes):
self.bytes_out += len(data)
if self._bw:
self._bw.write(data); self._bw.flush()
return self._inner.write(data)
def read(self, n: int) -> bytes:
chunk = self._inner.read(n)
if chunk:
self.reads += 1
self.bytes_in += len(chunk)
if self._s3:
self._s3.write(chunk); self._s3.flush()
return chunk
# ⚠ HYPOTHESIS, 6 events. content[11] of the 0x0C record separated 4 waveforms
# from 2 histograms cleanly and was constant within each group. A 4/2 split is
# thin evidence for a byte that could be anything, so _decode() below does NOT
# trust it -- it tries the other suffix on failure and says when the guess was
# wrong. The protocol reference states no type field is known; this may be it.
_TYPE_BYTE = 11
_TYPES = {0x07: ".IDFW", 0x08: ".IDFH"}
def _event_type(record: bytes) -> str:
if len(record) <= _TYPE_BYTE:
return "?"
b = record[_TYPE_BYTE]
return {0x07: "waveform", 0x08: "histogram"}.get(b, f"unknown(0x{b:02x})")
def _decode(blob: bytes, key: bytes, record: bytes) -> None:
"""Decode the downloaded bytes, proving they are a real event file.
read_idf_file() picks waveform vs histogram from the FILENAME SUFFIX, and a
wire download has no filename -- so the suffix has to come from somewhere.
This tries the 0x0C type byte first and the other suffix second; getting a
decode either way proves the chunk assembly, and which one worked is itself
the finding.
"""
import tempfile
from micromate.idf_file import read_idf_file
guess = _TYPES.get(record[_TYPE_BYTE] if len(record) > _TYPE_BYTE else -1, ".IDFW")
order = [guess] + [e for e in (".IDFW", ".IDFH") if e != guess]
for n, ext in enumerate(order):
with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as f:
f.write(blob)
tmp = f.name
try:
res = read_idf_file(tmp)
samples = sum(len(v) for v in getattr(res, "samples", {}).values())
note = "" if n == 0 else f" *** the 0x0C type byte guessed {guess} — WRONG ***"
print(f" decoded OK as {ext}: {samples} samples{note}")
os.unlink(tmp)
return
except Exception as e:
last = f"{ext}: {type(e).__name__}: {e}"
finally:
if os.path.exists(tmp):
os.unlink(tmp)
print(f" decode failed BOTH ways — last: {last}")
out = Path(f"./{key.hex()}.bin")
out.write_bytes(blob)
print(f" saved to {out} for offline analysis")
def step(label: str, fn):
"""Run one read, report how long it took and what it returned."""
t0 = time.monotonic()
try:
value = fn()
except Exception as e:
print(f" {label:.<26} FAILED {type(e).__name__}: {e}")
return None
ms = 1000 * (time.monotonic() - t0)
shown = value if isinstance(value, str) else repr(value)
if isinstance(value, list):
shown = f"{len(value)} entries"
print(f" {label:.<26} {ms:7.0f} ms {shown}")
return value
def main() -> int:
ap = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
ap.add_argument("target", help="host:port for TCP, or a serial device path")
ap.add_argument("--baud", type=int, default=115200,
help="serial only; the USB-A/FTDI path runs at 115200. "
"Ignored by the USB-B 'PC' port, which is CDC-ACM "
"and negotiates its own rate.")
ap.add_argument("--timeout", type=float, default=10.0)
ap.add_argument("--download", action="store_true",
help="also download the first stored event (read-only)")
ap.add_argument("--capture", metavar="DIR",
help="also write a raw_bw_*/raw_s3_*.bin pair to DIR, so "
"this run can become a test fixture. Worth doing on "
"any unit whose firmware line is new to us.")
ap.add_argument("--lenient", action="store_true",
help="do not raise on a bad checksum — for diagnosis only")
a = ap.parse_args()
if ":" in a.target and not Path(a.target).exists():
host, _, port = a.target.rpartition(":")
inner = TcpTransport(host, int(port), connect_timeout=a.timeout)
path = f"TCP {host}:{port}"
else:
inner = StdlibSerial(a.target, baud=a.baud)
path = f"serial {a.target} @ {a.baud}"
transport = _Timed(inner, capture=a.capture)
mm = MicromateClient(transport, recv_timeout=a.timeout,
strict_checksums=not a.lenient)
print(f"\n{path} (read-only: POLL, SERIAL, state, status, setups)\n")
t0 = time.monotonic()
try:
mm.open()
except OSError as e:
print(f" connect.................... FAILED {e}")
return 2
print(f" {'connect':.<26} {1000*(time.monotonic()-t0):7.0f} ms")
try:
info = step("connect() identity", mm.connect)
if info:
print(f" serial={info.serial} model={info.model} "
f"fw={info.firmware_line} monitoring={info.monitoring}")
print(f" active setup={info.active_setup!r}")
if info.firmware_line == "thor":
print(" *** 11.0BD unit — the FIRST one this code has met. ***")
print(" *** Check the battery and clock below carefully: ***")
print(" *** its 0x1C block is 4 bytes longer. ***")
state = step("get_state()", mm.get_state)
if state:
print(f" {state}")
if state.battery_volts and not 2.5 < state.battery_volts < 9.0:
print(f" *** battery {state.battery_volts} V is impossible — "
f"this is the from-the-end offset bug. ***")
if state.device_time is None:
print(" *** device clock did not decode — dump raw below. ***")
print(f" raw 0x1C content: {_content(state.raw).hex(' ')}")
setups = step("list_setups()", mm.list_setups)
if setups:
print(f" first={setups[0]!r} last={setups[-1]!r}")
# ── SUB 0x06: is content[0:4] really the event count? ─────────────
# Two samples on one unit said yes, and THOR reads it BEFORE the chain
# walk then stops without ever reading the sentinel. A third value
# either confirms it or kills it.
claimed = None
raw06 = step("0x06 storage range", mm.protocol.read_storage_range)
if raw06:
c = _content(raw06)
claimed = int.from_bytes(c[0:4], "big")
print(f" content[0:4] = {claimed} <- CANDIDATE: event count")
print(f" content[4:8] = {int.from_bytes(c[4:8],'big')} "
f"<- unexplained (read 9 alongside a 6 on UM12947)")
if a.download:
print("\n event chain (read-only), via MicromateClient:")
t1 = time.monotonic()
refs = mm.list_events() # walks to the sentinel
walk = time.monotonic() - t1
print(f" {len(refs)} events in {walk:.1f} s "
f"({walk/max(len(refs),1):.2f} s each, 3 round trips per event)")
if claimed is not None:
verdict = ("✓ AGREES" if claimed == len(refs)
else f"✗ DISAGREES (0x06 said {claimed})")
print(f" 0x06 count vs chain length: {verdict}")
for ref in refs:
print(f" {ref}")
print(f" would be filed as {ref.filename}")
if refs:
# Download via iter_events, which reproduces THOR's interleaved
# order -- the one with captures behind it.
print("\n download (first event, THOR's interleaved order):")
for ref in mm.iter_events():
t1 = time.monotonic()
try:
result = mm.get_event(ref) # verify=True
except Exception as e:
print(f" {ref.key_hex}: {type(e).__name__}: {e}")
break
dt = max(time.monotonic() - t1, 1e-6)
n = sum(len(v) for v in getattr(result, "samples", {}).values())
err = mm.decode_error(ref, result)
check = ("PVS agrees to %+.4f%%" % (100 * err)
if err is not None else "PVS check n/a (histogram)")
print(f" {ref.key_hex} {ref.record_type:9} {ref.size:6} B "
f"in {dt:.1f} s -> {n} samples, {check}")
break
except ProtocolError as e:
print(f"\n ABORTED {type(e).__name__}: {e}")
return 3
finally:
mm.close()
transport.close_capture()
if a.capture:
print(f"\n capture written:\n {transport.bw_path}\n {transport.s3_path}")
print(" parse it with: python3 scratch/mm_frame_parse.py "
f"{transport.bw_path} {transport.s3_path}")
elapsed = time.monotonic() - t0
print(f"\n transport: {transport.reads} reads, "
f"{transport.bytes_in} B in, {transport.bytes_out} B out, "
f"{elapsed:.1f} s total")
print(" Measured 2026-09-29, UM12947, same unit both ways:")
print(" USB-B (CDC-ACM) 83 reads list_setups 0.46 s download 394 KiB/s")
print(" RX55 (TCP) 36 reads list_setups 16.05 s download 1.6 KiB/s")
print(" The modem needs FEWER reads, not more -- it buffers ~1 s and then")
print(" forwards one large segment, where CDC-ACM delivers many small ones.")
print(" Cost is ~0.65 s PER ROUND TRIP regardless of payload size, so what")
print(" matters over cellular is the number of commands, not the bytes.\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())
-338
View File
@@ -1,338 +0,0 @@
#!/usr/bin/env python3
"""
mm_link.py — a "perfect modem" between THOR and a Micromate, with a readable
log and deliberate fault injection.
Why
---
THOR gives almost no visibility into a connection: a refresh button, two poll
intervals, and no way to see whether a check succeeded, timed out, or was never
sent. When a unit "won't stay connected" there is nothing to look at.
This sits where the cellular modem would sit and answers the question directly:
* **What is THOR actually doing?** Every frame is decoded and timestamped —
`POLL`, `MONITOR_STATUS`, `SETUP_NAME_READ` — not a hex dump.
* **Is it even trying?** Silence is visible: the log shows gaps.
* **How does it behave when the link misbehaves?** Faults can be injected on
demand, which a real cell link will not do on cue.
Point THOR at this host and port exactly as if it were a modem (Communication:
TCP, IP: <this host>, Port: <--listen>).
Fault injection
---------------
Write a mode into the control file (default `mm_link.ctl`) and it takes effect
on the next byte:
echo pass > mm_link.ctl # normal relay
echo blackhole > mm_link.ctl # TCP stays up, bytes are swallowed
echo drop > mm_link.ctl # close the connection abruptly (RST-ish)
echo delay:2.0 > mm_link.ctl # forward, but 2 s late in both directions
echo onewaydev > mm_link.ctl # THOR->unit passes, unit->THOR is swallowed
**`blackhole` is the one that matters.** It reproduces the classic cellular
failure: the socket is still open as far as both ends are concerned, but nothing
crosses. A client that relies on TCP to tell it the peer is gone will sit there
until the OS keepalive fires — which by default is about two hours.
Usage
-----
python3 bridges/mm_link.py --serial /dev/ttyACM0 --baud 115200 \\
--listen 12345 --logdir ~/mm-captures
Writes, per session:
<logdir>/mmlink_<ts>/session.log decoded, timestamped, human-readable
<logdir>/mmlink_<ts>/raw_bw.bin THOR -> unit, raw
<logdir>/mmlink_<ts>/raw_s3.bin unit -> THOR, raw
The raw pair loads straight into `scratch/mm_frame_parse.py`.
"""
from __future__ import annotations
import argparse
import datetime
import os
import socket
import sys
import threading
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scratch"))
try:
from mm_frame_parse import SUBNAME, destuff # noqa: F401
except Exception: # pragma: no cover
SUBNAME = {}
import errno
import select
import termios
DLE, STX, ETX, ACK = 0x10, 0x02, 0x03, 0x41
_BAUD = {9600: termios.B9600, 19200: termios.B19200, 38400: termios.B38400,
57600: termios.B57600, 115200: termios.B115200}
class SerialPort:
"""Minimal raw serial port on stdlib termios — no pyserial dependency.
The bench hosts are whatever is to hand; requiring a pip install on someone
else's machine is a poor trade for the ~30 lines this saves.
"""
def __init__(self, path: str, baud: int):
if baud not in _BAUD:
raise ValueError(f"unsupported baud {baud}; pick one of {sorted(_BAUD)}")
self.fd = os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
a = termios.tcgetattr(self.fd)
a[0] = 0 # iflag: no translation
a[1] = 0 # oflag: raw
a[2] = termios.CS8 | termios.CREAD | termios.CLOCAL # cflag: 8N1, ignore modem lines
a[3] = 0 # lflag: non-canonical, no echo
a[4] = a[5] = _BAUD[baud]
a[6] = list(a[6])
a[6][termios.VMIN] = 0
a[6][termios.VTIME] = 0
termios.tcsetattr(self.fd, termios.TCSANOW, a)
termios.tcflush(self.fd, termios.TCIOFLUSH)
def read(self, n: int) -> bytes:
r, _, _ = select.select([self.fd], [], [], 0.2)
if not r:
return b""
try:
return os.read(self.fd, n)
except OSError as e:
if e.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
return b""
raise
def write(self, data: bytes) -> None:
while data:
try:
data = data[os.write(self.fd, data):]
except OSError as e:
if e.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
select.select([], [self.fd], [], 0.2)
continue
raise
def close(self) -> None:
try:
os.close(self.fd)
except OSError:
pass
def name_of(sub: int, is_request: bool) -> str:
if is_request:
return SUBNAME.get(sub, f"SUB_{sub:02X}")
return "rsp " + SUBNAME.get(0xFF - sub, f"SUB_{0xFF - sub:02X}")
class FrameSniffer:
"""Accumulate bytes and report complete frames, without altering the stream."""
def __init__(self, is_request: bool):
self.is_request = is_request
self.buf = bytearray()
def feed(self, data: bytes):
"""Yield (sub, payload_len) for each complete frame seen."""
self.buf.extend(data)
while True:
start = -1
for i, b in enumerate(self.buf):
if self.is_request and b == ACK and i + 1 < len(self.buf) and self.buf[i + 1] == STX:
start = i
break
if not self.is_request and b == STX:
start = i
break
if start < 0:
if len(self.buf) > 8192:
del self.buf[:-16]
return
j = start + (2 if self.is_request else 1)
end = -1
while j < len(self.buf):
if self.buf[j] == DLE and j + 1 < len(self.buf):
j += 2
continue
if self.buf[j] == ETX:
end = j
break
j += 1
if end < 0:
return # wait for more bytes
body = self.buf[start:end + 1]
del self.buf[:end + 1]
# SUB sits at a fixed spot past the leading framing -- but it is
# DLE-escaped when its own value is 0x02/0x03/0x04/0x10, so a raw
# read reports 0x10 for those. SUB 0x02 was being logged as
# "SUB_10" until this was handled.
off = 5 if self.is_request else 3
if len(body) > off:
sub = body[off]
if sub == DLE and len(body) > off + 1:
sub = body[off + 1]
yield sub, len(body)
class Link:
def __init__(self, args):
self.args = args
self.mode = "pass"
self.delay = 0.0
self.ctl = Path(args.control)
self.session: Path | None = None
self.log_fh = None
self.raw = {}
self.t0 = time.time()
self.counts = {}
# ── logging ────────────────────────────────────────────────────────────
def open_session(self):
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
self.session = Path(self.args.logdir) / f"mmlink_{ts}"
self.session.mkdir(parents=True, exist_ok=True)
self.log_fh = open(self.session / "session.log", "a", buffering=1)
self.raw = {
"bw": open(self.session / "raw_bw.bin", "ab"),
"s3": open(self.session / "raw_s3.bin", "ab"),
}
self.say(f"=== session {ts} — serial {self.args.serial} @ {self.args.baud} ===")
def say(self, text: str):
line = f"{datetime.datetime.now().strftime('%H:%M:%S.%f')[:-3]} {text}"
print(line, flush=True)
if self.log_fh:
self.log_fh.write(line + "\n")
# ── control file ───────────────────────────────────────────────────────
def poll_control(self):
while True:
try:
if self.ctl.exists():
want = self.ctl.read_text().strip().lower()
if want.startswith("delay:"):
d = float(want.split(":", 1)[1])
if ("delay", d) != (self.mode, self.delay):
self.mode, self.delay = "delay", d
self.say(f"*** MODE -> delay {d}s ***")
elif want and want != self.mode:
self.mode, self.delay = want, 0.0
self.say(f"*** MODE -> {want} ***")
except Exception:
pass
time.sleep(0.25)
# ── the relay ──────────────────────────────────────────────────────────
def pump(self, src, dst, tag: str, is_request: bool, stop: threading.Event):
sniff = FrameSniffer(is_request)
arrow = "THOR->unit" if is_request else "unit->THOR"
last = time.time()
while not stop.is_set():
timed_out = False
try:
data = src.recv(4096) if isinstance(src, socket.socket) else src.read(4096)
except TimeoutError:
timed_out = True
# socket.timeout subclasses OSError, so it MUST be caught first.
# Treating it as a dead socket closes the connection after 200 ms
# of quiet -- which is exactly what `blackhole` produces, so the
# relay killed the link it was supposed to be faking a fault on.
data = b""
except OSError:
break
if isinstance(src, socket.socket) and data == b"" and not timed_out:
self.say(f"{arrow}: peer closed the connection")
break
if not data:
if time.time() - last > self.args.quiet_after and self.counts:
self.say(f"--- {self.args.quiet_after:.0f}s with no traffic ---")
last = time.time()
continue
last = time.time()
self.raw[tag].write(data)
self.raw[tag].flush()
for sub, ln in sniff.feed(data):
label = name_of(sub, is_request)
self.counts[label] = self.counts.get(label, 0) + 1
self.say(f"{arrow} {label:<20} ({ln} B)"
+ ("" if self.mode == "pass" else f" [mode={self.mode}]"))
mode = self.mode
if mode == "drop":
self.say(f"{arrow}: DROPPING the connection (fault injection)")
stop.set()
break
if mode == "blackhole":
continue # swallow, keep the socket open
if mode == "onewaydev" and not is_request:
continue # unit's replies never reach THOR
if mode == "delay" and self.delay:
time.sleep(self.delay)
try:
if isinstance(dst, socket.socket):
dst.sendall(data)
else:
dst.write(data)
except OSError:
break
stop.set()
def serve(self):
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("0.0.0.0", self.args.listen))
srv.listen(5)
self.open_session()
self.say(f"listening on 0.0.0.0:{self.args.listen} control file: {self.ctl}")
self.say("point THOR at this host/port as Communication=TCP")
threading.Thread(target=self.poll_control, daemon=True).start()
while True:
conn, addr = srv.accept()
conn.settimeout(0.2)
self.say(f"+++ THOR connected from {addr[0]}:{addr[1]} +++")
try:
ser = SerialPort(self.args.serial, self.args.baud)
except OSError as e:
self.say(f"!!! cannot open {self.args.serial}: {e}")
conn.close()
continue
stop = threading.Event()
ts = [
threading.Thread(target=self.pump, args=(conn, ser, "bw", True, stop), daemon=True),
threading.Thread(target=self.pump, args=(ser, conn, "s3", False, stop), daemon=True),
]
for t in ts:
t.start()
for t in ts:
t.join()
conn.close()
ser.close()
summary = ", ".join(f"{k}x{v}" for k, v in sorted(self.counts.items()))
self.say(f"--- connection closed. frames this session: {summary or 'none'} ---")
def main():
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--serial", default="/dev/ttyACM0")
ap.add_argument("--baud", type=int, default=115200)
ap.add_argument("--listen", type=int, default=12345)
ap.add_argument("--logdir", default=os.path.expanduser("~/mm-captures"))
ap.add_argument("--control", default="mm_link.ctl")
ap.add_argument("--quiet-after", type=float, default=30.0,
help="log a marker after this many seconds of silence")
Link(ap.parse_args()).serve()
if __name__ == "__main__":
main()
-239
View File
@@ -1,239 +0,0 @@
#!/usr/bin/env python3
"""
mm_probe.py — answer "why can't we reach this unit?" in one command.
THOR reports a failed connection as "disconnected" and nothing else. That single
word covers at least four completely different faults with four different fixes,
and telling them apart is the difference between a modem reboot and a site visit:
* **connection refused** something answered and said no — wrong port, or the
modem is refusing a further session
* **connect timed out** nothing answered at all — trusted-IP whitelist,
firewall, or the modem is off the network
* **connected, no reply** the MODEM answered but the unit did not. The TCP
path is fine; the modem is not forwarding to serial.
This is the signature of a wedged transparent-TCP
session, and it is the one THOR cannot distinguish
from any of the others
* **replied** the unit is alive; the problem is upstream software
Read-only. It sends `POLL`, then optionally `SERIAL` and the state read — the
same three commands THOR's own connection check uses — and never writes.
Usage
-----
python3 bridges/mm_probe.py 63.45.161.30:9034
python3 bridges/mm_probe.py 10.0.0.8:12345 --timeout 5
python3 bridges/mm_probe.py <host:port> --slots 3
`--slots N` opens N connections at once and reports how many the far end accepts.
A transparent-TCP modem typically serves **one** session; if the first succeeds
and the rest are refused or hang, that confirms the single-slot behaviour and
explains why a leaked session takes a unit offline until the slot frees.
Works for both series: a Series III reply opens `DLE STX`, a Micromate reply
opens with a bare `STX`, so the probe also tells you which one answered.
"""
from __future__ import annotations
import argparse
import socket
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from minimateplus.framing import build_bw_frame # noqa: E402
DLE, STX, ETX = 0x10, 0x02, 0x03
def destuff(raw: bytes) -> bytes:
"""Strip framing and DLE escapes; return the payload without its checksum."""
i = 1 if raw and raw[0] == STX else (2 if len(raw) > 1 and raw[1] == STX else 0)
out = bytearray()
while i < len(raw):
b = raw[i]
if b == DLE and i + 1 < len(raw):
out.append(raw[i + 1])
i += 2
continue
if b == ETX:
break
out.append(b)
i += 1
return bytes(out[:-1]) if len(out) > 1 else b""
# Reads are two-step on Series III: a probe at offset 0, then a data read at the
# block's length. THOR sends these offsets, and they also work on a Micromate.
OFFSETS = {0x5B: 0x0030, 0x15: 0x000A, 0x49: 0xFFFF}
def exchange(sock: socket.socket, sub: int, timeout: float) -> tuple[bytes, float]:
sock.sendall(build_bw_frame(sub, OFFSETS.get(sub, 0)))
t0 = time.time()
buf, deadline = b"", t0 + timeout
sock.settimeout(0.3)
while time.time() < deadline:
try:
chunk = sock.recv(4096)
if not chunk:
break
buf += chunk
if buf.endswith(bytes([ETX])) and len(buf) > 8:
break
except TimeoutError:
continue
except OSError:
break
return buf, time.time() - t0
def step(n: int, label: str, result: str) -> None:
print(f" [{n}] {label:.<28} {result}")
def probe(host: str, port: int, timeout: float) -> int:
print(f"\ntarget {host}:{port} (read-only: POLL, SERIAL, state)\n")
# ── 1. TCP ────────────────────────────────────────────────────────────
t0 = time.time()
try:
sock = socket.create_connection((host, port), timeout=timeout)
except ConnectionRefusedError:
step(1, "TCP connect", f"REFUSED after {1000*(time.time()-t0):.0f} ms")
print("\nverdict: something answered and actively refused.")
print(" Not a silent firewall drop — the host is reachable.")
print(" Wrong port, the service is down, or the modem is refusing")
print(" an additional session because its one slot is in use.")
return 2
except (TimeoutError, socket.timeout):
step(1, "TCP connect", f"TIMED OUT after {time.time()-t0:.1f} s")
print("\nverdict: nothing answered at all.")
print(" A silent drop, which is what a trusted-IP whitelist looks")
print(" like — it discards rather than refuses. Check the modem's")
print(" Trusted IPs (and note a VPN changes the IP you arrive from),")
print(" the firewall, and whether the modem is on the network.")
return 3
except OSError as e:
step(1, "TCP connect", f"FAILED: {e}")
return 4
step(1, "TCP connect", f"ok ({1000*(time.time()-t0):.0f} ms)")
# ── 2. POLL ───────────────────────────────────────────────────────────
try:
raw, dt = exchange(sock, 0x5B, timeout)
except OSError as e:
step(2, "POLL", f"send failed: {e}")
sock.close()
return 4
if not raw:
step(2, "POLL", f"NO REPLY in {timeout:.1f} s")
print("\nverdict: the MODEM answered but the unit did not.")
print(" TCP is fine end to end — something accepted the connection.")
print(" What is missing is the serial side. Two quite different")
print(" causes produce this, and they are NOT distinguishable from")
print(" here:")
print("\n 1. SOMEONE ELSE HOLDS THE SESSION. These modems bridge ONE")
print(" TCP session to serial at a time. A second connection is")
print(" accepted and then simply not forwarded. Confirmed 2026-09-26:")
print(" with THOR connected this probe saw exactly this; the moment")
print(" THOR disconnected the same probe returned the serial number.")
print(" ** Check whether THOR (or anything else) has the unit first. **")
print("\n 2. The serial path is genuinely broken — a stale session the")
print(" modem never released, a cable the unit cannot enumerate, or")
print(" a unit that is off.")
print("\n Try, in order:")
print(" 1. Disconnect any other client and re-probe. If it answers,")
print(" it was contention, not a fault.")
print(" 2. The cable's chipset. A Micromate drives FTDI and CDC-ACM")
print(" only — a Prolific PL2303 gives it no serial port at all.")
print(" lsusb: FTDI is 0403, Prolific 067b.")
print(" 3. Power-cycle the UNIT with the cable attached (hold power")
print(" 5 s, through the two-stage prompt). Its USB host rescans")
print(" on cold boot; it may not on hot-swap.")
print(" 4. AirLink OS -> TCP Idle Timeout. If 0/disabled, a stale")
print(" session holds the slot indefinitely. 2 minutes is the")
print(" value this project standardised on.")
sock.close()
return 5
series = "Series III (DLE STX)" if raw[0] == DLE else "Micromate (bare STX)"
step(2, "POLL", f"reply {len(raw)} B in {1000*dt:.0f} ms")
p = destuff(raw)
ok = len(p) > 3 and p[2] == 0xFF - 0x5B
step(3, "frame", f"{'valid' if ok else 'MALFORMED'}, {series}")
if not ok:
print("\nverdict: something replied, but not a seismograph.")
print(" Another service is on this port, or the modem is in a mode")
print(" that injects its own text (check Quiet Mode / AT echo).")
print(f" first bytes: {raw[:16].hex(' ')}")
sock.close()
return 6
# ── 3. identity + state ───────────────────────────────────────────────
for n, (sub, label) in enumerate(((0x15, "serial"), (0x49, "state")), start=4):
try:
r, dt = exchange(sock, sub, timeout)
d = destuff(r)[5:]
if sub == 0x15:
# serial is a null-terminated run; a further field follows it
serial = bytes(d[11:]).split(b"\x00")[0]
step(n, label, serial.decode("ascii", "replace") or "(empty)")
else:
step(n, label, "MONITORING" if len(d) > 11 and d[11] else "idle")
except OSError:
step(n, label, "no reply")
sock.close()
print("\nverdict: the unit is alive and answering.")
print(" If THOR still shows it disconnected, the fault is in THOR, not")
print(" the network or the device.")
return 0
def slots(host: str, port: int, n: int, timeout: float) -> None:
print(f"\nopening {n} simultaneous connections to {host}:{port}\n")
held = []
for i in range(n):
try:
s = socket.create_connection((host, port), timeout=timeout)
held.append(s)
step(i + 1, f"connection {i+1}", "accepted")
except ConnectionRefusedError:
step(i + 1, f"connection {i+1}", "REFUSED")
except (TimeoutError, socket.timeout):
step(i + 1, f"connection {i+1}", "timed out")
except OSError as e:
step(i + 1, f"connection {i+1}", f"failed: {e}")
print(f"\n{len(held)} of {n} accepted.")
if len(held) == 1:
print(" Single-slot behaviour confirmed — this far end serves ONE")
print(" session at a time. A connection that is never closed takes")
print(" the unit offline until the idle timeout frees the slot.")
for s in held:
s.close()
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("target", help="host:port, e.g. 63.45.161.30:9034")
ap.add_argument("--timeout", type=float, default=10.0)
ap.add_argument("--slots", type=int, metavar="N",
help="open N simultaneous connections to test single-slot behaviour")
a = ap.parse_args()
host, _, port = a.target.rpartition(":")
if not host:
ap.error("target must be host:port")
if a.slots:
slots(host, int(port), a.slots, a.timeout)
return 0
return probe(host, int(port), a.timeout)
if __name__ == "__main__":
raise SystemExit(main())
-408
View File
@@ -1,408 +0,0 @@
# Spec — a live client for Series IV (Micromate)
Drafted 2026-09-26, ahead of implementation. The protocol work is finished; this
is the plan for turning `docs/micromate_protocol_reference.md` into code SFM can
run.
**Read that document first.** Everything here assumes it, and every constant
below is sourced from it rather than restated with justification.
---
## Goal and scope
`micromate/` is codec-only today — `idf_file.py`, `models.py`, the report
writers. There is no way to talk to a unit. This adds the live half, mirroring
`minimateplus/`.
**In scope, first pass:**
- connect over TCP (a field modem) or serial/USB (a bench unit)
- identify a unit, read its state, clock, memory and setups
- walk the event chain and download events
- return `Event` objects the existing codec already understands
**Explicitly out of scope, first pass:**
- ⚠ **Any write.** Setups, schedules, call-home config, monitoring start/stop,
and per-event delete are all mapped, and none of them will be implemented
here. **No command has ever been originated against a unit by this project**
— every write observed was performed by THOR while we recorded. Keeping that
true through the read client is deliberate: it means the first thing we ever
send to a customer's instrument is a decision someone made on purpose, not a
side effect of a client that happened to grow a method.
- the inbound call-home session — still the one protocol unknown
---
## Layout
```
micromate/
framing.py NEW frame building, response parsing, checksum
protocol.py NEW one method per wire command, returns raw payloads
client.py NEW high-level API, returns models
idf_file.py (existing — decodes what 0x5A returns, unchanged)
models.py (existing — extend, do not fork)
```
**Transport is reused, not rewritten.** `minimateplus/transport.py` is
byte-level and protocol-agnostic — `BaseTransport`, `SerialTransport`,
`TcpTransport`, plus `read_until_idle()` which already handles the RV50/RV55
habit of emitting `\r\nRING\r\n\r\nCONNECT\r\n` to a caller. Import it.
⚠ Do **not** import `minimateplus.framing`. The two framings differ in ways
that look small and are not, and a shared module would accumulate `if series ==`
branches until neither case is readable.
---
## `micromate/framing.py` — ✅ BUILT 2026-09-27
Implemented, with `tests/test_micromate_framing.py` (31 tests, all passing).
**Two things in this section as originally drafted were wrong**, and both were
caught by measuring against the captures before writing code rather than after.
They are left in place below, struck through, because both are the kind of
mistake that would be made again.
### Requests
~~Series IV accepts Series III request frames unmodified. The simplest correct
implementation re-exports the builder rather than duplicating it:~~
```python
from minimateplus.framing import build_bw_frame # ✗ WRONG — 161/218
```
⚠ **`build_bw_frame` reproduces only 161 of Thor's 218 captured read frames.**
The payload layout is identical; the stuffing is not. A Micromate escapes
**four** byte values — `0x02`, `0x03`, `0x04`, `0x10` — where Series III
escapes one. The frames this breaks are **every `SUB 0x5A` download**
(`offset = 0x0400` → a literal `0x04` in `offset_hi`) and the scheduler enable.
An unescaped `0x03`/`0x04` terminates the frame early, so the unit just does not
answer.
`build_request()` with the correct escape set reproduces **218/218**.
✅ **Settled: `0x10` inside params needs no special handling.** The planned
`NotImplementedError` guard is unnecessary — Thor sends
`params = 00 00 10 00 …` on `SUB 0x5A` in five captured frames and the wire
carries an ordinary doubled `10 10`. One rule covers the whole payload.
### Responses — where Series III's parser cannot follow
| | Series III | Micromate |
|---|---|---|
| frame start | `DLE STX` | **bare `STX`** |
| `payload[1]` | `0x10` | `0xC5` (Blastware fw) / `0x03` (Thor fw) |
| destuffing | `DLE+ETX` kept as literal inner-frame data | **`10 XX` → `XX`, uniformly** |
The first row is why `S3FrameParser` returns nothing at all on Series IV traffic:
it scans for `DLE+STX`, which never appears.
The third is a genuine **simplification** — no inner-frame carve-out. Validated
by checksum across every capture in `bridges/captures/9-24-26 - micromate2/`:
four candidate destuffing rules were tried, and only this one makes all frames
validate.
### Checksum
```python
def checksum(payload: bytes) -> int:
return sum(payload) & 0xFF # payload already de-stuffed
```
~~The DLE-aware variant, same as Series III's `5A` and write frames.~~
⚠ **Plain SUM8, not the DLE-aware variant** — 251/251 both directions. The
DLE-aware form is the right answer paired with *Series III* de-stuffing, which
leaves an escaped byte in the payload as two bytes. De-stuffing `10 XX → XX`
already removes the `0x10`, so excluding it again subtracts the correction
twice, and the result disagrees with the wire on **55 of 251** captured
responses — every frame holding a literal `0x10`.
`scratch/mm_frame_parse.py` shipped with exactly that pairing. It looked clean
only because it accepts a frame matching *either* rule, so it labelled those 55
`SUM8` and never flagged one bad. "Zero bad checksums" was true and carried no
information. Fixed there too.
### ⚠ The SUB byte can be escaped
When a SUB's value is `0x02`, `0x03`, `0x04` or `0x10` it arrives as `10 XX`.
Reading it positionally without destuffing reports `0x10`. This bit once
already — `SUB 0x02` was logged as `SUB_10` for an afternoon. Destuff first,
then index.
### Response shape
```python
@dataclass
class MicromateFrame:
sub: int # response SUB; request = 0xFF - sub
flags: int # 0xC5 Blastware line, 0x03 Thor line
page_hi: int
page_lo: int
data: bytes # payload[5:], checksum stripped
checksum_valid: bool
@property
def request_sub(self) -> int: # 0xFF - sub
@property
def page_key(self) -> int: # uint16 BE at payload[3:5]
@property
def firmware_line(self) -> str: # "blastware" | "thor" | "unknown"
@property
def probe_length(self) -> int | None: # uint16 BE at data[3:5] (= payload[8:10])
```
⚠ **`probe_length` is a uint16 BE.** Read as a single byte it under-reads
`SUB 0x1A` by 47x — 44 against a true 2092. This is the single most expensive
mistake available in this protocol and it has already been made once.
Renamed from `declared_length`, because it is **only meaningful in the reply to
an `offset = 0` probe** — and Thor never probes. Across all 251 captured
responses the field reads 0 or a page count, never a length, precisely because
that session is single-step reads throughout. `page_key` is the field that
carries meaning there. The only genuine probe reply we hold is the POLL one
preserved in `scratch/fake_unit.py`.
`MicromateFrameParser` mirrors `S3FrameParser`: `feed(bytes) -> list[frame]`,
accumulates in `.frames`, `reset()`, and keeps the `bytes_fed` counter (it is
what distinguishes "no bytes at all" from "bytes but no complete frame" on a
timeout, and that distinction earned its keep during the Series III work).
---
## `micromate/protocol.py` — ✅ BUILT 2026-09-27
Implemented, with `tests/test_micromate_protocol.py` (35 tests). Reads only;
nothing here writes, erases or changes monitoring state.
The tests replay Thor's captured responses through a scripted transport and
assert **the bytes we emit are the bytes Thor emits** — including a full replay
of the six-event download session, all 56 `0x5A` frames byte-for-byte. That is
a stronger guarantee than "our parser understands the device": a passing test
means a real unit has already answered exactly that frame.
One method per command, returning raw payload bytes. No interpretation — that
belongs in `client.py`.
**Reads use `offset = 0xFFFF`** and return the whole block in one response;
Series III's two-step probe/data dance is unnecessary. `POLL` is the exception,
taking its data length. Per-command offsets, all observed:
| command | SUB | rsp | offset | returns |
|---|---|---|---|---|
| poll | `0x5B` | `0xA4` | `0x0030` | device string, model |
| serial | `0x15` | `0xEA` | `0x000A` | `UM12947` |
| device info | `0x01` | `0xFE` | `0xFFFF` | firmware, calibration |
| state | `0x49` | `0xB6` | `0xFFFF` | `data[11]`: non-zero = monitoring |
| monitor status | `0x1C` | `0xE3` | `0xFFFF` | flag, **device clock**, battery, memory |
| storage range | `0x06` | `0xF9` | `0xFFFF` | event storage extent |
| active setup name | `0x41` | `0xBE` | `0xFFFF` | `TEST1.mmb` |
| first setup | `0x3F` | `0xC0` | `0xFFFF` | setup-list walk head |
| next setup | `0x40` | `0xBF` | `0xFFFF` | …until an empty name |
| compliance config | `0x1A` | `0xE5` | `0xFFFF` | ~2103 B setup block |
| call-home config | `0x2C` | `0xD3` | `0xFFFF` | 137 B |
| arm event | `0x93` | `0x6C` | — | before every event |
| first event | `0x1E` | `0xE1` | `0xFFFF` | key + size |
| next event | `0x1F` | `0xE0` | `0xFFFF` | key + size |
| event record | `0x0C` | `0xF3` | `0xFFFF` | 221 B — project, location, peaks |
| ~~event header~~ **monitor log** | `0x0A` | `0xF5` | `0xFFFF` | ⚠ 297 B, a **walk** — see below |
| bulk download | `0x5A` | `0xA5` | computed | **the `.IDFW` verbatim**, 1024 B at a time |
⚠ **Corrected 2026-09-27, from Thor's frames.** Three rows of the table above
were wrong or incomplete, and the last one is a different command than labelled:
- **`0x0A` is the monitor-log walk**, not a keyed "30 B list record" read. The
*same request repeated* returns successive 297-byte records — serial, mode,
thresholds — until an 11-byte ack ends the list. The device holds the cursor;
nothing in the request selects a record. Series III reaches this data through
a record-type discriminator on its event chain; here it has its own cursor and
the event chain never sees it.
- **`0x1E`/`0x1F` carry token `0xFE` at `params[7]`.** The reference documents
all-zero params (our own probing, which also worked). Thor's form is the one
with mileage.
- **`0x01` has no Thor frame behind it** — it is never read in any captured
session. Its `0xFFFF` comes from our probes.
And one useful negative: **no `SESSION_RESET` (`41 03`)**. Series III needs that
2-byte signal or a monitoring unit will not answer `POLL` over TCP. Thor never
sends it — zero occurrences across 8 sessions, including 40 frames exchanged
with a unit that *was* monitoring.
⚠ **`SUB 0x1C` is 4 bytes longer on the Thor firmware line** (`0x30` vs `0x2C`).
Parse **forward** from `declared_length`, never backward from the end — Series
III reads battery and memory from the end of that block, and doing so on a BD
unit yields a battery voltage of **577.92 V**.
⚠ **Test the monitoring flag for non-zero**, never against a constant. It has
read both `0x0E` and `0x0C` while monitoring.
### `0x5A` — a bounded chunk loop, and much simpler than Series III
⚠ **Corrected 2026-09-27.** This section said "no chunk loop — one request
returns the whole event", with `offset_word = 0x1000 + 2 * ceil(size / 512)`.
That describes our own 2026-09-23 probes, which set `offset_hi = 0x10`. **Thor
chunks**, and Thor's form is the one verified from bytes on disk:
```python
n = ceil(size / 1024) # size from the chain walk
for i in range(n):
offset = min(1024, size - 1024 * i) # a BYTE COUNT
params = key4 + bytes(6) if i == 0 else bytes(2) + pack(">H", 1024*i) + bytes(6)
file_bytes += response.data[11:] # response data is exactly offset + 11
```
Verified on all six bench events (4,076 → 13,424 B): `sum(offsets) == size`
exactly, with the predicted chunk count and final offset every time.
Still no arming ritual for `0x5A` itself, no `STRT` end-offset parsing and no
`TERM` frame — the simplification the original claim celebrated is real, it just
is not single-shot. (`SUB 0x93` arms the *chain walk*, before `1E`/`1F`, not
the download.)
The concatenated payload **is** the `.IDFW` file, byte for byte — so it feeds
`micromate.idf_file.read_idf_file()` and `/db/import/idf_file` unchanged.
⚠ Do not port the Series III `5A` walk. Its address arithmetic caused a 5x
over-read and a `> 64 KB` page-boundary bug that is *still open* on the Series
III side. None of that applies here: the chunk index is a byte offset into the
file, bounded by a size the device told us, and it cannot run past the event.
⚠ **`assert sum(len(chunk) - 11 for chunk in chunks) == size`.** A silently
short event is the failure mode this project has been bitten by repeatedly on
the Series III side, and here the check is free because the size is known up
front.
---
## `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:
def __init__(self, transport: BaseTransport): ...
def open(self) / close(self) / is_open(self)
# identity and state
def connect(self) -> DeviceInfo # poll → serial → device info → state
def get_state(self) -> UnitState # monitoring?, clock, battery, memory
def get_active_setup(self) -> str
def list_setups(self) -> list[str] # 0x3F → 0x40… until empty
# events
def list_events(self) -> list[EventRef] # 0x93 → 0x1E → 0x1F… (key + size)
def download_event(self, ref) -> bytes # raw .IDFW/.IDFH
def get_event(self, ref) -> Event # download + decode via idf_file
```
`connect()` should mirror THOR's preamble (`POLL → SERIAL → 0x49 → POLL`) —
⚠ but note the reference records that **whether the unit requires it is
untested**. Do it because it is known-good, not because it is known-necessary,
and say so in the docstring.
`list_events()` returns the key *and* the size, because `download_event()` needs
the size to compute its offset word.
---
## Tests
**Offline, from captured bytes — no hardware.** This is the part worth doing
first, because it can be fully verified tonight's-captures-style before any unit
is involved.
```
tests/test_micromate_framing.py
```
⚠ `bridges/captures/` and `tests/fixtures/` are both gitignored, so tests must
not depend on files being present. **Embed the frames as hex constants** — they
are 19–138 bytes each. ✅ Done; what actually landed:
| case | source | why |
|---|---|---|
| POLL probe reply, 19 B | captured (via `fake_unit.py`) | shortest valid frame; the only real probe reply we hold |
| `0x5A` chunk, 138 B | captured | holds literal `0x10` **and** literal `0x41` — the checksum case, and proves ACK is not escaped |
| `0x49` state reply, 25 B | captured | a literal `0x02`, escaped |
| `0x48` file reply, 24 B | captured | escaped `0x04` in `data[0]` — one byte late without destuffing |
| 8 Thor request frames | captured | byte-for-byte against `build_request()`, incl. both `0x5A` forms and the scheduler enable |
| `probe_length = 0x082C` | **synthesised** | no probe reply for `0x1A` exists on disk — the 9-24-26 session never probes |
| Thor-line reply, `flags = 0x03` | **synthesised** | no 11.0BD capture is in the repo; built by flipping one byte of the real POLL reply |
| escaped checksum byte | **synthesised** | the shortest real one is 1,070 B, too long to embed for one assertion |
| truncated / corrupt / split-across-feeds | derived | parser must return nothing, flag rather than swallow, and survive any split point |
⚠ Synthesised frames are marked `SYNTH_`-style in the test and each says what it
stands in for and why no capture was available. Do not let that set grow
quietly: the `flags = 0x03` case in particular is the **only** coverage of half
the fleet, and it deserves a real 11.0BD capture the next time UM20147 is on a
bench.
Two corpus-backed tests run when the captures happen to be on the box and skip
cleanly otherwise: **251 response frames parse with zero bad checksums**, and
**`build_request()` reproduces 218/218 read frames**. The second is the test
that would have caught the escape-set error, so it is worth the skip marker.
⚠ Do **not** assert against `scratch/mm_frame_parse.py`'s output as the original
plan proposed. That script accepts either checksum rule and is wrong about
which one is right — using it as an oracle would have pinned the bug.
**Live, second:** against the bench unit on mint-mac via `mm_link.py`.
`connect()`, `list_setups()` (should return the 23 known names), `list_events()`,
then `download_event()` and assert the bytes decode and match a
`/db/import/idf_file` ingest of the same event.
---
## Order of work
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` + 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
Steps 1–2 need no hardware at all.
**Worth carrying forward.** Both steps began by measuring against the captures
rather than trusting this document, and both found errors in it — three in the
framing rules (the escape set, 26% of frames; the checksum, 22%; the `0x5A`
chunk model) and three more in the command table (`0x0A`'s meaning, the
`1E`/`1F` token, `0x01`'s provenance). All six fail quietly. The captures are on
disk and a measure-then-write loop costs about two minutes per rule, so keep
doing it for `client.py`'s field offsets — and treat this spec as a plan, not a
source.
---
## Open questions to settle while implementing
- ~~**Request param stuffing**~~ — ✅ settled 2026-09-27; no special handling.
- **Is the single-request `0x5A` form real?** Our 2026-09-23 probes set
`offset_hi = 0x10` and appeared to get a whole 11 KB event back, where Thor
chunks at 1024 B. Plausibly a distinct streaming mode that returns several
frames. One bench test settles it; implement Thor's form regardless.
- **Is THOR's preamble required?** Try one command cold and find out; it is a
two-minute test with the bench unit and it removes a ritual if unnecessary.
- **`Event` model fit** — Series IV carries fields Series III lacks (setup file
name, `LMic`/`SMic` channels). Extend `micromate/models.py`; do not fork the
shared `Event`.
- **Which `0x0C` fields to trust.** The peak float there runs 2–5% above
`max(T,V,L)` and is **not** the vector sum; its offset was inferred, not
established. The reference marks it do-not-rely-on — prefer decoded samples.
File diff suppressed because it is too large Load Diff
-150
View File
@@ -1,150 +0,0 @@
# SFM — where it actually stands as a tool
**Status as of 2026-09-20 (v0.31.0).** This is the honest assessment, not the
roadmap — `README.md § Roadmap` covers where it is *going*. Expect this file to
go stale; re-date it when you revise it.
---
## The framing
SFM is **three different things wearing one name**, at three very different
levels of maturity:
| | what it is | maturity |
|---|---|---|
| **The codec library** | `minimateplus/`, `micromate/` — bytes in, `Event` out | **Production.** Verified per-sample at scale. |
| **SDM — the data side** | the DB, waveform store, `/db/*`, ingest | **Production.** Terra-View depends on it daily. |
| **SFM — the device side** | `/device/*`, live connections to units | **Emergency-grade.** Works, but manual, unauthenticated, and thinly tested. |
| **The lab** | `seismo_lab.py`, `scratch/`, the Inspector | **Research artifacts.** Useful, not products. |
Brian's own description — *"right now it's an emergency tool and a research
project"* — is accurate, and it applies specifically to the **device side**.
The data side is not an emergency tool; it has been carrying production for
months.
Most confusion about "is SFM reliable?" comes from answering for the wrong
tier.
---
## 1. What you can rely on
### Production-grade — trust it
- **Series-3 decode.** 14,338 / 14,338 files decode per-sample exact against
preserved Blastware ASCII exports, 45 units, files back to 2018.
- **Series-4 (Thor) decode.** 1,057,536 / 1,057,536 geo samples exact against
Thor's own CSV exports; production IDFW 575/575 with zero truncations.
- **Histogram decode.** 1,211 / 1,211 production histograms exact, including
842,442 per-interval frequency comparisons with zero mismatches.
- **The ingest path.** `/db/import/blastware_file` and `/db/import/idf_file`
fed by the watchers — this is how prod actually gets its data, and it has
been running unattended for months.
- **`/db/*` read API.** Always-on, consumed by Terra-View for every fleet
listing, event detail and report.
- **The waveform store** — `.h5` + `.sfm.json` sidecars + retained raw
binaries, with operator review state preserved across regeneration.
- **`bridges/ach_server.py`** — speaks the full BW protocol to calling units.
Proven in the field, including as a rescue tool (see the runbook).
### Emergency-grade — works, but you are the error handling
- **`/device/*` live endpoints.** They do what they say. But they are
synchronous, unauthenticated, and a single cellular download can exceed the
60 s timeouts that sit in front of them.
- **The rescue ladder** (`rescue`, `stop_monitoring_*`, `events/erase`).
Each has worked in a real incident — but each has been used a handful of
times, by one person, with the runbook open.
- **The standalone webapp.** Perfectly usable, and as of v0.31.0 the cheap
probes and rescue actions are reachable without curl. No auth of any kind.
### Research artifacts — useful, not products
- **`seismo_lab.py`** — 2,789 lines of Tkinter (Bridge / Analyzer / Query DB /
Inspector). Desktop-only, single-user, no tests.
- **`scratch/`** — the verification harnesses (`verify_against_ascii.py`,
`verify_thor_against_csv.py`) and the offset detector (`offset_scan3.py`).
These produced the numbers the production claims rest on, so they matter —
but they are analysis scripts, not maintained code.
- **`docs/offset_investigation.md`** — an open investigation, not a feature.
---
## 2. What to use when
| you want to… | use | notes |
|---|---|---|
| Know if a unit is monitoring / its battery / memory | `GET /device/monitor/status?force=true` | ~2 s |
| Know whether ACH is on | `GET /device/call_home` | ~2 s. **Not** `/device/events`. |
| See how full a unit's buffer is | `GET /device/events/storage_range` | ~2 s, no chain walk |
| Stop a runaway unit | Diagnostics tab → Stop Monitoring | see the runbook first |
| Reach a unit that will not answer | **point its modem at an `ach_server` and answer its call** | runbook Method A — do not race it |
| List a unit's stored events | Events tab → Load events | **slow**, and broken past 64 KB (below) |
| Get event data into the DB | the watcher → `/db/import/*` path | not the live walk |
The single most useful habit: **the cheap probes are cheap and the event walk
is not.** Reaching for `/device/events` to answer a yes/no question about a
unit is the mistake that motivated the v0.31.0 webapp changes.
---
## 3. Known issues
| issue | impact | status |
|---|---|---|
| **5A walk dies once a unit's buffer crosses 64 KB** | `/device/events` 500s; event body never downloads | Known, documented in `CLAUDE.md`. Needs a BW capture of a spanning event to fix properly. |
| **No auth on SFM at all** | 21 `/device/*` endpoints, including destructive ones, open to anything that reaches the port | Design agreed (Terra-View as authenticated jump host); not built. |
| **Swagger try-it-out is live on destructive endpoints** | `POST /device/events/erase` is one click away at `:8200/docs` | Partially mitigated: the webapp's erase now requires typing the serial. `/docs` itself is unguarded. |
| **`SUB 0x08` lifetime counter reads 0** | `/device/events/index` returns a meaningless number | Suspected field-offset bug. Surfaced in the UI as "unreliable". |
| **Long device operations are synchronous** | 60 s timeouts in `routers/sfm.py` and the reverse proxy; a full download exceeds both | Known design constraint. Must be POST-starts-job / GET-polls before any remote lab. |
| **`backfill_sidecars.py --force` silently inserts DB rows** | store files with no DB row get one; the dry-run does not report the count | Known. Avoid `--force` — `TOOL_VERSION` gates regeneration anyway. |
| **14 sensitive-range files show an exact 8× discrepancy** | 10.0 / 1.25 — a units problem, not a decode problem | Open, not blocking. |
| **16 failing tests on `dev`** | 15 need gitignored fixture bundles; 1 is real (`sc["peak_values"]["transverse"]` returns `None` where `0.0` is expected) | The real one shipped in v0.31.0. |
---
## 4. What stands between this and a real tool
Roughly in dependency order — each unblocks the ones below it.
**1. Authentication.** Everything else is gated on this. SFM has none, and
the modem IP whitelist gives zero protection because SFM *is* the whitelisted
origin. The agreed design delegates rather than builds: Terra-View becomes the
authenticated jump host (`/api/sfm/*` already inherits deny-by-default operator
auth), and the `8200:8200` publish is dropped so Terra-View is the only door.
**2. Async long operations.** POST starts a job, GET polls. Retrofitting this
after building a remote lab on top of synchronous endpoints would be far worse
than designing for it now.
**3. Confirm-guards on the remaining destructive endpoints.** Auth answers
*who*, not *did you mean it*. The webapp's erase is guarded; the other seven
destructive POSTs and `/docs` are not.
**4. The 5A page-boundary fix.** Until this lands, live event download is
unreliable on exactly the units most likely to need attention — the ones that
have been recording heavily. Wants a Blastware capture of an event spanning a
page boundary before the chunk-addressing half is trustworthy.
**5. A live Thor / Micromate client.** The device side is MiniMate-only.
Series-4 units can only be read from forwarded files, so half the fleet has no
live path at all.
**6. Test coverage that runs from a clean checkout.** 15 of 16 current
failures are missing fixture bundles. A test suite that cannot go green on a
fresh clone cannot gate anything.
**7. The SDM rename.** Cosmetic relative to the above, but the longer `sfm/`
holds the data-side code the more the tiers blur. ~30–50 files here, ~10–15 in
Terra-View, plus a Docker volume migration. Do it when the codebase is quiet.
---
## The short version
The **data side is a real tool already**. The **device side is a set of sharp
instruments** that work in the hands of the person who wrote them, with the
runbook open. The gap between those two states is mostly **auth, async, and
guardrails** — not protocol work. The protocol is the part that is actually
finished.
-568
View File
@@ -1,568 +0,0 @@
"""
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
import math
import os
import struct
from typing import Optional
from minimateplus.transport import BaseTransport
from .models import MicromateDeviceInfo, MicromateEventRef, 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
# `SUB 0x0C` record, relative to content. Established against 7 events across
# both firmware lines.
_REC_DAY, _REC_MONTH, _REC_YEAR = 0, 1, slice(2, 4)
_REC_UNKNOWN_4 = 4 # ⚠ same shape as 0x1C's content[6]; undecoded
_REC_HOUR, _REC_MIN, _REC_SEC = 5, 6, 7
_REC_TYPE = 11 # 0x07 waveform, 0x08 histogram
_REC_LOCATION = 12
_REC_SETUP = 34
_REC_SERIAL = 76
_RECORD_TYPES = {0x07: "waveform", 0x08: "histogram"}
# The channel labels the record carries, in the order they appear. The peak
# float sits `label + 6`; the peak vector sum sits 12 bytes BEFORE "Tran".
_REC_CHANNELS = (b"Tran", b"Vert", b"Long", b"Mic")
_REC_PVS_BACK = 12
# A chain walk terminates on an all-zero key. The ceilings below are guards
# against a device cursor that never advances, not fleet limits.
_NULL_KEY = bytes(4)
_MAX_EVENTS = 4096
# 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 DecodeMismatch(ProtocolError):
"""Our decoded peak disagrees with the one the device computed itself."""
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
self._serial: 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()))
self._serial = 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.
✅ **Confirmed on `11.0BD` 2026-09-30.** UM20147 read back 3.55 V and
a clock correct to the second over USB, so the from-the-start offsets do
survive the four extra trailing bytes. Had they not, the battery would
have read 577.92 V — which is what makes this cheap to check.
"""
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"
)
# ── Events ────────────────────────────────────────────────────────────────
def serial(self) -> str:
"""The unit's serial, cached from `connect()` or read on demand.
Needed by anything that handles an event, because an event key is
ambiguous without it — see `MicromateEventRef`.
"""
if self._serial is None:
self._serial = _cstring(_content(self._proto.read_serial()))
return self._serial
def list_events(self, *, with_records: bool = True) -> list[MicromateEventRef]:
"""Walk the event chain. `0x93 → 1E`, then `0x93 → 1F` until the sentinel.
THOR sends `0x93` before **every** chain read, and this mirrors that.
The chain ends on an all-zero key.
⚠ `with_records=True` costs **one extra round trip per event** for the
`0x0C` read, and over cellular a round trip is ~0.65 s regardless of
size. On a unit with 40 events that is the difference between ~52 s and
~78 s. Pass False when you only need "what is here and how big" — but
note the **record is where the type and timestamp live**, so without it
`ref.filename` is None and `get_event()` cannot pick a suffix.
⚠ **To download, prefer `iter_events()`.** This walks the whole chain
first; THOR interleaves, downloading each event before advancing.
`0x5A` addresses an event by key, so downloading afterwards *should*
work — but "should" is doing real work in that sentence, and the
interleaved order is the one with captures behind it. See
`iter_events()`.
"""
serial = self.serial()
refs: list[MicromateEventRef] = []
for i in range(_MAX_EVENTS):
self._proto.arm_event()
raw = (self._proto.read_event_first() if i == 0
else self._proto.read_event_next())
c = _content(raw)
if len(c) < 8:
raise ProtocolError(
f"chain entry {i}: {len(c)} B of content, need 8 (key + size)"
)
key, size = c[0:4], int.from_bytes(c[4:8], "big")
if key == _NULL_KEY:
return refs # the sentinel, not an error
ref = MicromateEventRef(index=i, key=key, size=size, serial=serial)
if with_records:
self._read_record_into(ref)
refs.append(ref)
raise ProtocolError(
f"event chain did not terminate after {_MAX_EVENTS} entries — the "
f"device cursor is not advancing"
)
def iter_events(self, *, with_records: bool = True):
"""Walk the chain, yielding each event **at the cursor position THOR uses.**
for ref in mm.iter_events():
if ref.is_histogram:
continue
data = mm.download_event(ref) # ← safe here
Why this exists alongside `list_events()`: THOR's captured order is
0x93 → 1E → 0C → 5A×n → 0x93 → 1F → 0C → 5A×n → …
— it downloads each event *before* advancing the chain. `list_events()`
walks to the end first, which is fine for browsing (the browse walk is
separately attested) but means a later download happens with the device
cursor parked past the event. `0x5A` is key-addressed, so it very
probably does not care; nothing observed says it does, and nothing
observed says it does not.
Downloading inside this loop reproduces THOR's sequence exactly, so it
is the path to use when it matters. ⚠ Do not advance the generator
before finishing with the event it yielded.
"""
serial = self.serial()
for i in range(_MAX_EVENTS):
self._proto.arm_event()
raw = (self._proto.read_event_first() if i == 0
else self._proto.read_event_next())
c = _content(raw)
if len(c) < 8:
raise ProtocolError(
f"chain entry {i}: {len(c)} B of content, need 8 (key + size)"
)
key, size = c[0:4], int.from_bytes(c[4:8], "big")
if key == _NULL_KEY:
return
ref = MicromateEventRef(index=i, key=key, size=size, serial=serial)
if with_records:
self._read_record_into(ref)
yield ref
raise ProtocolError(
f"event chain did not terminate after {_MAX_EVENTS} entries — the "
f"device cursor is not advancing"
)
def _read_record_into(self, ref: MicromateEventRef) -> None:
"""`SUB 0x0C` — 210 B of content: timestamp, type, names, peaks."""
raw = self._proto.read_event_record(ref.key)
c = _content(raw)
if len(c) <= _REC_SERIAL:
raise ProtocolError(f"event record is {len(c)} B, too short to decode")
ref.raw_record = raw
ref.record_type = _RECORD_TYPES.get(c[_REC_TYPE])
if ref.record_type is None:
# Worth saying out loud rather than filing the event as a waveform:
# the suffix decides which codec runs.
log.warning("event %s: unknown record type 0x%02x at content[%d]",
ref.key_hex, c[_REC_TYPE], _REC_TYPE)
try:
ref.timestamp = datetime.datetime(
year=int.from_bytes(c[_REC_YEAR], "big"),
month=c[_REC_MONTH], day=c[_REC_DAY],
hour=c[_REC_HOUR], minute=c[_REC_MIN], second=c[_REC_SEC],
)
except ValueError as e:
log.warning("event %s: bad timestamp (%s): %s",
ref.key_hex, e, c[:8].hex(" "))
ref.sensor_location = _cstring(c, _REC_LOCATION) or None
ref.setup = _cstring(c, _REC_SETUP) or None
# Prefer the record's own serial over the cached one — they have always
# agreed, but the record is the event's own account of where it came from.
if rec_serial := _cstring(c, _REC_SERIAL):
ref.serial = rec_serial
peaks: dict[str, float] = {}
for label in _REC_CHANNELS:
i = c.find(label)
if i < 0 or i + len(label) + 10 > len(c):
continue
peaks[label.decode()] = struct.unpack(
">f", c[i + len(label) + 2: i + len(label) + 6])[0]
ref.peaks_ips = peaks or None
tran = c.find(b"Tran")
if tran >= _REC_PVS_BACK:
ref.peak_vector_sum_ips = struct.unpack(
">f", c[tran - _REC_PVS_BACK: tran - _REC_PVS_BACK + 4])[0]
def download_event(self, ref: MicromateEventRef) -> bytes:
"""The raw `.IDFW`/`.IDFH` bytes, exactly as THOR would have stored them.
Feeds `micromate.idf_file.read_idf_file()` and `/db/import/idf_file`
unchanged — no new codec work is needed for a directly downloaded event.
"""
return self._proto.read_event_file(ref.key, ref.size)
def get_event(self, ref: MicromateEventRef, *, verify: bool = True,
tolerance: float = 0.01):
"""Download and decode one event.
Returns the codec's `IdfReadResult`. Needs `ref.record_type`, since
`read_idf_file()` dispatches on the filename suffix and there is no
filename on the wire — so call `list_events(with_records=True)` first.
⚠ `verify=True` re-computes the **peak vector sum** from the decoded
samples and compares it against the float the *device* put in the `0x0C`
record. Those are two independent computations over the same samples —
the device's from its own firmware, ours from our codec — so a
disagreement means our decode is wrong. Measured agreement on the bench
events is **0.000%**.
This is cheap insurance in a codebase whose decode failures have
historically been *silent*: unhandled block tags shorten a channel and
nothing raises. ⚠ It is a decode-correctness check, **not** a
truncation detector — a channel cut after its peak still yields the
right PVS.
Histograms are not verified: `samples` is empty for them.
"""
if not ref.record_type:
raise ValueError(
f"event {ref.key_hex}: record_type is unknown, so the codec "
f"cannot be dispatched. Use list_events(with_records=True)."
)
blob = self.download_event(ref)
import tempfile
from .idf_file import read_idf_file
# read_idf_file dispatches on the suffix, so the bytes need a name.
with tempfile.NamedTemporaryFile(suffix=ref.suffix, delete=False) as f:
f.write(blob)
tmp = f.name
try:
result = read_idf_file(tmp)
finally:
os.unlink(tmp)
if verify and not ref.is_histogram:
err = self.decode_error(ref, result)
if err is not None and abs(err) > tolerance:
raise DecodeMismatch(
f"event {ref.uid}: decoded peak vector sum differs from the "
f"device's own by {100 * err:+.3f}% (tolerance "
f"{100 * tolerance:.1f}%) — the decode is suspect, not the "
f"device. Stored {ref.peak_vector_sum_ips:.5f} in/s."
)
if err is not None:
log.debug("event %s: PVS agrees to %+.4f%%", ref.uid, 100 * err)
return result
@staticmethod
def decode_error(ref: MicromateEventRef, result) -> Optional[float]:
"""Relative error between our decoded PVS and the device's stored one.
None when either side is unavailable. Positive means the device's
figure is higher than ours.
"""
from .idf_file import geo_count_to_ips
stored = ref.peak_vector_sum_ips
if not stored or not getattr(result, "samples", None):
return None
ch = {k.lower(): v for k, v in result.samples.items()}
try:
t, v, l = ch["tran"], ch["vert"], ch["long"]
except KeyError:
return None
n = min(len(t), len(v), len(l))
if not n:
return None
pvs = max(
math.sqrt(geo_count_to_ips(t[i]) ** 2
+ geo_count_to_ips(v[i]) ** 2
+ geo_count_to_ips(l[i]) ** 2)
for i in range(n)
)
return (stored - pvs) / pvs if pvs else None
-304
View File
@@ -1,304 +0,0 @@
"""
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,
)
-155
View File
@@ -396,158 +396,3 @@ 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)
@dataclass
class MicromateEventRef:
"""One entry in a unit's event chain, from `1E`/`1F` and optionally `0x0C`.
⚠ **`key` is NOT unique across units.** The event counter starts from the
same value on every Micromate — UM12947 and UM20147 both have an event
`055d4a81`, with different sizes and different contents. Use `uid`, or key
on `(serial, key_hex)`, for anything that stores or deduplicates. A store
keyed on the event key alone silently treats one unit's event as a duplicate
of another's, and nothing raises.
"""
index: int
key: bytes # 4-byte event key from the chain walk
size: int # bytes the device will send for this event
serial: Optional[str] = None # the unit, because `key` alone is ambiguous
# From `SUB 0x0C` — one extra round trip per event, so optional.
record_type: Optional[str] = None # "waveform" | "histogram"
timestamp: Optional[datetime.datetime] = None
setup: Optional[str] = None # setup file name, no extension
sensor_location: Optional[str] = None
peak_vector_sum_ips: Optional[float] = None # per-sample PVS, device-computed
peaks_ips: Optional[Dict[str, float]] = None # {"Tran": …, "Vert": …, …}
raw_record: Optional[bytes] = field(default=None, repr=False)
@property
def key_hex(self) -> str:
return self.key.hex()
@property
def uid(self) -> str:
"""`SERIAL:key` — safe to use as a primary key. See the class note."""
return f"{self.serial or '?'}:{self.key_hex}"
@property
def is_histogram(self) -> Optional[bool]:
if self.record_type is None:
return None
return self.record_type == "histogram"
@property
def suffix(self) -> Optional[str]:
return {"waveform": ".IDFW", "histogram": ".IDFH"}.get(self.record_type or "")
@property
def filename(self) -> Optional[str]:
"""The name THOR would have given this event.
`<serial>_<YYYYMMDDHHMMSS>.IDF{W,H}` — e.g.
`UM12947_20260923163319.IDFW`. Verified against the production store
for all five bench events.
⚠ The type comes from the **protocol**, not the payload, so it has to be
carried here from the `0x0C` read. Returns None without it: guessing
the suffix would file a histogram as a waveform, and `read_idf_file()`
dispatches on exactly that.
"""
if not (self.serial and self.timestamp and self.suffix):
return None
return f"{self.serial}_{self.timestamp:%Y%m%d%H%M%S}{self.suffix}"
def __str__(self) -> str:
bits = [self.uid, f"{self.size} B"]
if self.record_type:
bits.append(self.record_type)
if self.timestamp:
bits.append(self.timestamp.strftime("%Y-%m-%d %H:%M:%S"))
if self.peak_vector_sum_ips is not None:
bits.append(f"PVS {self.peak_vector_sum_ips:.4f} in/s")
return " ".join(bits)
-501
View File
@@ -1,501 +0,0 @@
"""
protocol.py — one method per Micromate (Series IV) wire command.
Returns raw payload bytes. Interpretation belongs in ``client.py``; this layer
knows frames, offsets and sequencing, and nothing about what a field means.
Scope: **reads only.** Nothing here writes, erases, or changes monitoring
state. That is deliberate and worth keeping — no command has ever been
originated against a unit by this project; every write in
``docs/micromate_protocol_reference.md`` was performed by THOR while we
recorded. The first thing this codebase ever sends to a customer's instrument
should be a decision someone made on purpose, not a side effect of a client
that grew a method.
Every offset and params layout below was read off THOR's own frames in
``bridges/captures/9-24-26 - micromate2/`` rather than taken from the spec
table, because the same exercise during the framing work found three of that
table's rules wrong. It found three more here:
* ``0x0A`` is the **monitor-log walk** — the same request repeated, the
device advancing its own cursor, terminated by a short response — not the
keyed "event header, 30 B list record" the spec describes.
* ``0x1E``/``0x1F`` carry **token 0xFE** at ``params[7]``. The protocol
reference documents all-zero params for the browse walk; that was our own
probing, and THOR does not do it that way.
* There is **no fixed preamble**. Sessions open with ``POLL`` and go
straight to the operation. ``POLL → SERIAL → 0x49 → POLL`` appears in 3 of
8 captured sessions and is THOR's *connection check*, not a handshake.
And one useful negative: **no ``SESSION_RESET`` (``41 03``).** Series III
needs that 2-byte signal to wake a monitoring unit or it will not answer POLL
over TCP. THOR never sends it — 0 occurrences across all 8 sessions, including
40 frames exchanged with a unit that *was* monitoring.
"""
from __future__ import annotations
import logging
import math
import struct
import time
from typing import Optional
from minimateplus.transport import BaseTransport
from .framing import MicromateFrame, MicromateFrameParser, build_request
log = logging.getLogger(__name__)
DEFAULT_RECV_TIMEOUT = 10.0
# An acknowledgement carries an 11-byte data section and nothing else. It is
# also how the monitor-log walk says "no more records".
ACK_DATA_LEN = 11
# ── Command SUBs ──────────────────────────────────────────────────────────────
SUB_DEVICE_INFO = 0x01
SUB_STORAGE_RANGE = 0x06
SUB_EVENT_INDEX = 0x08
SUB_MONITOR_LOG = 0x0A
SUB_EVENT_RECORD = 0x0C
SUB_SERIAL = 0x15
SUB_COMPLIANCE_CONFIG = 0x1A
SUB_MONITOR_STATUS = 0x1C
SUB_EVENT_FIRST = 0x1E
SUB_EVENT_NEXT = 0x1F
SUB_CALL_HOME_CONFIG = 0x2C
SUB_TRIGGER_CONFIG = 0x2E
SUB_SETUP_FIRST = 0x3F
SUB_SETUP_NEXT = 0x40
SUB_SETUP_ACTIVE = 0x41
SUB_STATE = 0x49
SUB_BULK_DOWNLOAD = 0x5A
SUB_POLL = 0x5B
SUB_ARM_EVENT = 0x93
# ⚠ Reads are SINGLE-STEP. Series III probes at offset 0 to learn the length,
# then reads again at that length; a Micromate returns the whole block when
# asked for 0xFFFF. THOR never probes, which is why `MicromateFrame.probe_length`
# reads 0 on live traffic.
READ_ALL = 0xFFFF
# The two commands that do NOT use READ_ALL, and the data length each returned
# on UM12947 (firmware 11.0CB).
_OFFSETS = {
SUB_POLL: 0x0030, # 59 B — the one offset THOR treats as a constant
SUB_SERIAL: 0x000A, # 21 B
}
# Data-section lengths observed, for orientation only — deliberately NOT
# asserted. `SUB 0x1C` is 4 bytes longer on the Thor firmware line (0x30 vs
# 0x2C declared), so a length check here would fire spuriously on half the
# fleet. See the protocol reference, "A/B: Blastware build vs Thor build".
OBSERVED_DATA_LEN = {
SUB_STORAGE_RANGE: 47, SUB_EVENT_INDEX: 101, SUB_MONITOR_LOG: 297,
SUB_EVENT_RECORD: 221, SUB_SERIAL: 21, SUB_COMPLIANCE_CONFIG: 2103,
SUB_MONITOR_STATUS: 55, SUB_EVENT_FIRST: 19, SUB_EVENT_NEXT: 19,
SUB_CALL_HOME_CONFIG: 137, SUB_TRIGGER_CONFIG: 39,
SUB_SETUP_FIRST: 266, SUB_SETUP_NEXT: 266, SUB_SETUP_ACTIVE: 266,
SUB_STATE: 16, SUB_POLL: 59, SUB_ARM_EVENT: ACK_DATA_LEN,
}
# `1E`/`1F` carry this at params[7]. Series III uses the same value to arm its
# bulk stream; here THOR sends it on every chain read, browse or download.
EVENT_TOKEN = 0xFE
# `SUB 0x5A` chunk size, in bytes of file payload per response.
CHUNK_SIZE = 1024
# Every `0x5A` response prefixes the file bytes with 11 bytes of header.
_CHUNK_PREFIX = 11
# ── Exceptions ────────────────────────────────────────────────────────────────
class ProtocolError(Exception):
"""The device violated the expected protocol."""
class TimeoutError(ProtocolError):
"""No response arrived within the allowed time."""
class ChecksumError(ProtocolError):
"""A received frame failed its checksum."""
class UnexpectedResponse(ProtocolError):
"""The response SUB did not match the request."""
class ShortRead(ProtocolError):
"""A bulk download returned fewer bytes than the device promised."""
# ── Params builders ───────────────────────────────────────────────────────────
def token_params(token: int = EVENT_TOKEN) -> bytes:
"""`1E`/`1F`: the token sits at params[7]."""
return bytes(7) + bytes([token]) + bytes(2)
def key_params(key4: bytes) -> bytes:
"""`0x0C`: the full 4-byte event key at params[4:8]."""
if len(key4) != 4:
raise ValueError(f"key4 must be 4 bytes, got {len(key4)}")
return bytes(4) + key4 + bytes(2)
def key_lo_params(key4: bytes) -> bytes:
"""`0x0A`: only the key's **low two bytes**, at params[6:8].
⚠ Inferred from a single key value. All nine captured `0x0A` frames carry
`4a 81`, and the event key in play was `055d4a81` — so this is consistent
with "the low half of the current key" and equally consistent with "a
cursor handle that happened to equal it". Both readings produce the same
bytes for that key, so one event cannot separate them.
It does not matter much in practice: the walk works with the same params
repeated, so whichever it is, passing the current key is right.
"""
if len(key4) != 4:
raise ValueError(f"key4 must be 4 bytes, got {len(key4)}")
return bytes(6) + key4[2:4] + bytes(2)
def chunk_params(key4: bytes, byte_offset: int) -> bytes:
"""`0x5A`: the key opens the file, then a byte offset walks it.
Chunk 0 carries the event key at params[0:4] — that is what says "from the
beginning". Later chunks carry a uint16 BE byte offset at params[2:4].
"""
if byte_offset == 0:
if len(key4) != 4:
raise ValueError(f"key4 must be 4 bytes, got {len(key4)}")
return key4 + bytes(6)
if not 0 <= byte_offset <= 0xFFFF:
raise ValueError(f"byte_offset must fit in uint16, got {byte_offset}")
return bytes(2) + struct.pack(">H", byte_offset) + bytes(6)
# ── Protocol ──────────────────────────────────────────────────────────────────
class MicromateProtocol:
"""Wire-level command set for one open connection to a Micromate.
Does not own the transport; lifetime belongs to the client.
proto = MicromateProtocol(transport)
proto.poll()
serial = proto.read_serial()
"""
def __init__(
self,
transport: BaseTransport,
recv_timeout: float = DEFAULT_RECV_TIMEOUT,
strict_checksums: bool = True,
) -> None:
"""
Args:
strict_checksums: raise on a bad checksum. **Defaults to True,
unlike the Series III sibling**, which logs and continues
because its parser cannot reliably tell an inner-frame
delimiter from a checksum byte. That excuse does not apply
here: the Micromate rule is plain SUM8 over the de-stuffed
payload and it holds on 251 of 251 captured frames, so a
mismatch means something real — line noise, a desync, or a rule
we have wrong — and all three are worth hearing about.
The lenient Series III default is instructive: it hid the fact
that the documented checksum rule was wrong for two days. Set
False only to get a field diagnosis unstuck.
"""
self._transport = transport
self._recv_timeout = recv_timeout
self._strict = strict_checksums
self._parser = MicromateFrameParser()
self._pending: list[MicromateFrame] = []
# ── Identity and state ────────────────────────────────────────────────────
def poll(self) -> MicromateFrame:
"""`0x5B` → `0xA4`. Handshake; carries the ID block and model string.
Every captured session opens with this and nothing before it.
"""
return self._exchange(SUB_POLL)
def read_serial(self) -> bytes:
"""`0x15` → `0xEA`. ASCII, null-terminated — e.g. `UM12947`."""
return self._read(SUB_SERIAL)
def read_device_info(self) -> bytes:
"""`0x01` → `0xFE`. Firmware, calibration, per-channel float block.
⚠ THOR never sends this in any captured session, so the `0xFFFF` offset
is from our own 2026-09-23 probes rather than from THOR's behaviour. It
answered correctly on both firmware lines, but it is the one read here
with no THOR frame behind it.
"""
return self._read(SUB_DEVICE_INFO)
def read_state(self) -> bytes:
"""`0x49` → `0xB6`. A cheap monitoring check; 16 B.
⚠ Test `data[11]` for **non-zero**, never against a constant — it has
read both `0x0E` and `0x0C` while monitoring.
"""
return self._read(SUB_STATE)
def read_monitor_status(self) -> bytes:
"""`0x1C` → `0xE3`. Flag, **device clock**, battery, memory.
⚠ Parse **forward** from the declared length, never backward from the
end. This block is 4 bytes longer on the Thor firmware line, and Series
III's relative-to-end offsets yield a battery voltage of 577.92 V on a
`11.0BD` unit.
"""
return self._read(SUB_MONITOR_STATUS)
def read_storage_range(self) -> bytes:
"""`0x06` → `0xF9`. Event storage extent; 47 B."""
return self._read(SUB_STORAGE_RANGE)
def read_event_index(self) -> bytes:
"""`0x08` → `0xF7`. 101 B. Contents not yet mapped."""
return self._read(SUB_EVENT_INDEX)
def read_trigger_config(self) -> bytes:
"""`0x2E` → `0xD1`. 39 B. Series IV only; no Series III equivalent."""
return self._read(SUB_TRIGGER_CONFIG)
def read_compliance_config(self) -> bytes:
"""`0x1A` → `0xE5`. The whole active setup — 2103 B on UM12947.
One response. Series III needs a 4-frame sequence for the same thing.
"""
return self._read(SUB_COMPLIANCE_CONFIG)
def read_call_home_config(self) -> bytes:
"""`0x2C` → `0xD3`. 137 B — Series III's is 124, so do not reuse its map."""
return self._read(SUB_CALL_HOME_CONFIG)
# ── Setups ────────────────────────────────────────────────────────────────
def read_active_setup_name(self) -> bytes:
"""`0x41` → `0xBE`. 266 B; carries the active `.MMB` name."""
return self._read(SUB_SETUP_ACTIVE)
def read_first_setup(self) -> bytes:
"""`0x3F` → `0xC0`. Head of the setup-file list."""
return self._read(SUB_SETUP_FIRST)
def read_next_setup(self) -> bytes:
"""`0x40` → `0xBF`. Repeat until the record carries an empty name.
Stateful: the device holds the cursor, so the same request walks the
list. 22 of these appear back to back in one captured session.
"""
return self._read(SUB_SETUP_NEXT)
# ── Event chain ───────────────────────────────────────────────────────────
def arm_event(self) -> MicromateFrame:
"""`0x93` → `0x6C`. THOR sends this before **every** `1E`/`1F`.
It replaces Series III's `1E(token=0xFE)` arming step. No params, no
offset payload — an 11-byte ack.
⚠ Whether a unit actually requires it is untested. Do it because it is
known-good, not because it is known-necessary.
"""
return self._exchange(SUB_ARM_EVENT, offset=READ_ALL)
def read_event_first(self) -> bytes:
"""`0x1E` → `0xE1`. First event key + size; 19 B."""
return self._read(SUB_EVENT_FIRST, params=token_params())
def read_event_next(self) -> bytes:
"""`0x1F` → `0xE0`. Next key + size, or the all-zero null sentinel."""
return self._read(SUB_EVENT_NEXT, params=token_params())
def read_event_record(self, key4: bytes) -> bytes:
"""`0x0C` → `0xF3`. 221 B — project, client, operator, timestamp, peaks.
⚠ The peak float in here runs 2–5% above `max(T,V,L)` and is **not** the
vector sum; its offset was inferred, not established. Prefer decoded
samples.
"""
return self._read(SUB_EVENT_RECORD, params=key_params(key4))
def read_monitor_log_next(self, key4: bytes) -> Optional[bytes]:
"""`0x0A` → `0xF5`. One monitor-log record, or None at end of list.
⚠ Not the keyed single read the spec describes. This is a **walk**:
the same request repeated, the device advancing its own cursor, each
response a 297-byte record carrying serial, mode and thresholds. The
list ends with a bare 11-byte ack — nine captured frames, eight records
then the terminator.
Series III reaches the same data through a record-type discriminator on
its event walk (`0x2C` partial vs `0x46` full). Here it is a separate
cursor and the event chain does not see it at all.
"""
data = self._read(SUB_MONITOR_LOG, params=key_lo_params(key4))
return None if len(data) <= ACK_DATA_LEN else data
# ── Bulk download ─────────────────────────────────────────────────────────
def read_event_file(self, key4: bytes, size: int) -> bytes:
"""`0x5A` → `0xA5`. The `.IDFW`/`.IDFH` file, byte for byte.
`size` is the 4 bytes after the key in the `1E`/`1F` response. Returns
exactly that many bytes, or raises `ShortRead`.
A bounded chunk walk — `ceil(size / 1024)` requests, each asking for
`min(1024, remaining)` bytes:
offset = the byte count wanted (NOT an address)
params = the key on chunk 0, then a uint16 BE byte offset
response = exactly `offset + 11` bytes; file bytes are data[11:]
Verified against THOR on all six bench events (4,076 → 13,424 B):
`sum(offsets) == size` exactly, every time.
⚠ Do not port the Series III `5A` walk. Its address arithmetic caused a
5x over-read and a `>64 KB` page-boundary bug that is *still open* on
that side. Neither applies here — the cursor is a byte offset into the
file, bounded by a size the device supplied, so it cannot run past the
event.
The result feeds `micromate.idf_file.read_idf_file()` and
`/db/import/idf_file` unchanged; no new codec work is needed.
"""
if size <= 0:
raise ValueError(f"size must be positive, got {size}")
out = bytearray()
n_chunks = math.ceil(size / CHUNK_SIZE)
for i in range(n_chunks):
want = min(CHUNK_SIZE, size - i * CHUNK_SIZE)
data = self._read(
SUB_BULK_DOWNLOAD,
offset=want,
params=chunk_params(key4, i * CHUNK_SIZE),
)
if len(data) < _CHUNK_PREFIX:
raise ShortRead(
f"chunk {i + 1}/{n_chunks} of {key4.hex()}: "
f"{len(data)} B is too short to hold a chunk header"
)
body = data[_CHUNK_PREFIX:]
if len(body) != want:
# Worth being loud: a silently short event is the failure mode
# this project has been bitten by repeatedly on the Series III
# side, and here the expected length is known up front.
raise ShortRead(
f"chunk {i + 1}/{n_chunks} of {key4.hex()}: asked for "
f"{want} B, got {len(body)}"
)
out += body
if len(out) != size:
raise ShortRead(
f"{key4.hex()}: assembled {len(out)} B, device promised {size}"
)
log.debug("downloaded %s: %d B in %d chunks", key4.hex(), len(out), n_chunks)
return bytes(out)
# ── Plumbing ──────────────────────────────────────────────────────────────
def _read(
self,
sub: int,
*,
params: bytes = bytes(10),
offset: Optional[int] = None,
timeout: Optional[float] = None,
) -> bytes:
"""Send one read command, return the response's data section."""
return self._exchange(sub, params=params, offset=offset, timeout=timeout).data
def _exchange(
self,
sub: int,
*,
params: bytes = bytes(10),
offset: Optional[int] = None,
timeout: Optional[float] = None,
) -> MicromateFrame:
if offset is None:
offset = _OFFSETS.get(sub, READ_ALL)
# Start every exchange clean: drop any half-frame and any frame left
# stashed by the last one. This is a strict request/response protocol,
# so anything already buffered when we send is by definition stale, and
# `expected_sub` would reject it anyway — better to discard it here than
# to raise a confusing UnexpectedResponse one command later.
self._parser.reset()
self._pending.clear()
self._send(build_request(sub, offset, params))
return self._recv_one(expected_sub=0xFF - sub, timeout=timeout,
reset_parser=False)
def _send(self, frame: bytes) -> None:
log.debug("TX %d bytes: %s", len(frame), frame.hex())
self._transport.write(frame)
def _recv_one(
self,
expected_sub: Optional[int] = None,
timeout: Optional[float] = None,
reset_parser: bool = True,
) -> MicromateFrame:
"""Read until one complete frame is parsed."""
deadline = time.monotonic() + (timeout or self._recv_timeout)
if reset_parser:
self._parser.reset()
self._pending.clear()
if self._pending:
return self._validate(self._pending.pop(0), expected_sub)
while time.monotonic() < deadline:
chunk = self._transport.read(4096)
if not chunk:
time.sleep(0.005)
continue
log.debug("RX %d bytes", len(chunk))
frames = self._parser.feed(chunk)
if frames:
self._pending.extend(frames[1:])
return self._validate(frames[0], expected_sub)
raise TimeoutError(
f"no frame in {timeout or self._recv_timeout:.1f}s"
+ (f" (expected SUB 0x{expected_sub:02X})" if expected_sub is not None else "")
+ f"; {self._parser.bytes_fed} bytes were received"
# That byte count is the whole point: it separates "nothing came
# back at all" from "bytes arrived but never framed", and those have
# completely different causes. It earned its keep on Series III.
)
def _validate(
self, frame: MicromateFrame, expected_sub: Optional[int]
) -> MicromateFrame:
if not frame.checksum_valid:
msg = (
f"SUB 0x{frame.sub:02X}: checksum mismatch "
f"(got 0x{frame.chk_byte:02X}, {len(frame.data)} B data)"
)
if self._strict:
raise ChecksumError(msg)
log.warning("%s — continuing (strict_checksums=False)", msg)
if expected_sub is not None and frame.sub != expected_sub:
raise UnexpectedResponse(
f"expected SUB 0x{expected_sub:02X}, got 0x{frame.sub:02X}"
)
return frame
+1 -63
View File
@@ -296,16 +296,6 @@ def apply_report_to_event(event: Event, report: BwAsciiReport) -> None:
event.sample_rate = report.sample_rate_sps
if report.record_time_s is not None:
event.rectime_seconds = report.record_time_s
# The report's event_datetime is Blastware's exact trigger time (parsed
# from Event Time + Event Date). Prefer it over the binary footer's stop
# time so a report-paired import matches BW to the second.
edt = report.event_datetime
if edt is not None:
event.timestamp = Timestamp(
raw=b"", flag=0x10,
year=edt.year, unknown_byte=0, month=edt.month, day=edt.day,
hour=edt.hour, minute=edt.minute, second=edt.second,
)
def apply_bw_report_dict_to_event(event: Event, bw_report: dict) -> None:
@@ -818,30 +808,6 @@ def derive_record_type_from_filename(filename, default: str = "Waveform") -> str
return _RECORD_TYPE_BY_EXT_SUFFIX.get(ext[-1].upper(), default)
# Marker for the recording-setup config block, and the offset of the record-time
# float32 within it. The configured post-trigger record time (seconds) is a
# big-endian float32 exactly 30 bytes before the "Standard Recording Setup"
# label. Verified across the corpus reading 1.0 / 2.0 / 3.0 s on different
# setups — and ts2 - record_time reproduces Blastware's trigger to the second
# (N844LQHB: stop 10:33:32 - 3.0 = 10:33:29).
_RECSETUP_MARKER = b"Standard Recording Setup"
_RECTIME_OFFSET_BEFORE_MARKER = 30
def _parse_record_time_seconds(raw: bytes) -> Optional[float]:
"""The configured post-trigger record time in seconds, from the recording-
setup config block, or None when absent / implausible."""
a = raw.find(_RECSETUP_MARKER)
if a < _RECTIME_OFFSET_BEFORE_MARKER:
return None
off = a - _RECTIME_OFFSET_BEFORE_MARKER
try:
rt = struct.unpack(">f", raw[off:off + 4])[0]
except struct.error:
return None
return rt if 0.05 <= rt <= 600.0 else None
def read_blastware_file(path: Union[str, Path]) -> Event:
"""
Parse a Blastware waveform file into an Event.
@@ -951,10 +917,6 @@ def read_blastware_file(path: Union[str, Path]) -> Event:
# rest of the event (timestamp, waveform_key, project strings) is
# still recoverable and useful.
decoded = decode_waveform_v2(body)
# Discriminator for the timestamp logic below: a waveform (trigger) event
# vs a histogram window. Keyed on the codec, not the filename — the
# save_imported_bw path passes a tmp ".bw" name whose extension lies.
is_waveform_body = decoded is not None
if decoded is None:
decoded = decode_histogram_body(body)
if decoded is None:
@@ -986,31 +948,7 @@ def read_blastware_file(path: Union[str, Path]) -> Event:
ev.total_samples = strt_fields.get("total_samples")
ev.pretrig_samples = strt_fields.get("pretrig_samples")
# Event timestamp. The footer's two timestamps mean different things by
# record type:
# * Waveform: ts1 = the monitoring-SESSION start (shared across every
# event that day — a unit arming at 06:00 stamps 06:00 on all of them),
# ts2 = THIS event's recording STOP. Blastware's Date/Time is the
# TRIGGER = ts2 - record time, and the record time is a float32 in the
# recording-setup config block (see _parse_record_time_seconds), so the
# exact trigger is recoverable from the binary alone. Falls back to ts2
# (the stop, within the record duration) if the config block is absent.
# (Stamping ts1 showed the session start, hours off.)
# * Histogram / undecodable: ts1 = the window start, which IS the event
# time — keep it.
# Discriminate by ``is_waveform_body`` (the codec), not the filename.
if is_waveform_body and ts2 is not None:
_stop = datetime.datetime(ts2.year, ts2.month, ts2.day,
ts2.hour, ts2.minute, ts2.second)
_rt = _parse_record_time_seconds(raw)
_trig = _stop - datetime.timedelta(seconds=_rt) if _rt is not None else _stop
ev.timestamp = Timestamp(
raw=footer[10:18],
flag=0x10,
year=_trig.year, unknown_byte=0, month=_trig.month, day=_trig.day,
hour=_trig.hour, minute=_trig.minute, second=_trig.second,
)
elif ts1 is not None:
if ts1 is not None:
ev.timestamp = Timestamp(
raw=footer[2:10],
flag=0x10,
-33
View File
@@ -1,33 +0,0 @@
"""Pretend to be a Micromate on a serial port: log what arrives, reply to POLL.
Proves the modem's return path (serial -> TCP) independently of the real unit.
"""
import os, select, sys, termios, time
path, baud = sys.argv[1], int(sys.argv[2]) if len(sys.argv) > 2 else 115200
B = {9600: termios.B9600, 38400: termios.B38400, 115200: termios.B115200}[baud]
fd = os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
a = termios.tcgetattr(fd)
a[0] = a[1] = a[3] = 0
a[2] = termios.CS8 | termios.CREAD | termios.CLOCAL
a[4] = a[5] = B
a[6] = list(a[6]); a[6][termios.VMIN] = 0; a[6][termios.VTIME] = 0
termios.tcsetattr(fd, termios.TCSANOW, a)
termios.tcflush(fd, termios.TCIOFLUSH)
# A real POLL probe reply, captured from UM12947 on 2026-09-24.
REPLY = bytes.fromhex("0200c5a4000000000000300000000000000099") + b"\x03"
print(f"fake unit on {path} @ {baud}; will answer any inbound frame", flush=True)
while True:
r, _, _ = select.select([fd], [], [], 1.0)
if not r:
continue
data = os.read(fd, 4096)
if not data:
continue
ts = time.strftime("%H:%M:%S")
print(f"{ts} IN {len(data):3} B {data.hex(' ')}", flush=True)
time.sleep(0.02)
os.write(fd, REPLY)
print(f"{ts} OUT {len(REPLY):3} B {REPLY.hex(' ')} <- canned POLL reply", flush=True)
-223
View File
@@ -1,223 +0,0 @@
#!/usr/bin/env python3
"""
mm_frame_parse.py — parse Micromate (Series IV) frames out of a seismo_lab
raw capture pair.
Why this exists
---------------
`minimateplus.framing.S3FrameParser` cannot see Micromate traffic. It locates
frames by scanning for `DLE STX`, and a Micromate response has **no leading
DLE** — it starts at a bare `STX`. It also expects `payload[1] == 0x10`, where
the Micromate sends `0xC5` (Blastware firmware) or `0x03` (Thor firmware).
The practical consequence, seen on the 9-24-26 setup-push capture: the
Blastware-side requests parse fine (Thor emits Series III request frames), but
**every device response is silently dropped or mis-framed** — so a capture that
actually contains 12 acked writes looks like 12 unanswered requests.
Destuffing
----------
One rule covers both directions: after the leading doubled `BW_CMD`, every
`10 XX` pair on the wire destuffs to `XX`. That includes `10 03` — Thor
escapes literal `0x03` bytes in write data so they are not mistaken for ETX,
exactly as Blastware does.
A Micromate escapes exactly four byte values — `0x02 0x03 0x04 0x10` — and
nothing else, which is what makes the uniform rule exact rather than merely
convenient. Established by re-stuffing all 502 captured frames and comparing
to the wire: 251/251 each direction, where `{0x10}` alone gets 130 and 177.
⚠ Checksum, corrected 2026-09-27
--------------------------------
With uniform destuffing the checksum is **plain SUM8 of the destuffed
payload**. This script used to try SUM8 *and* a "DLE-aware" variant that
excludes `0x10` bytes, and report whichever matched — which is why it never
flagged a bad frame and why the protocol reference carried the wrong rule for
two days. The DLE-aware form belongs with *Series III* destuffing, where an
escaped byte survives as two bytes; applying it after uniform destuffing
subtracts the correction twice and disagrees with the wire on 55 of 251
responses.
The lesson generalises: a tool that accepts any of N candidate rules cannot
falsify any of them. It now validates against SUM8 alone, and reports
`DLE-aware` only to name what a mismatch *would* have been — never as a pass.
See `docs/micromate_protocol_reference.md` → *Checksum*, and
`tests/test_micromate_framing.py`, which pins it.
`micromate/framing.py` is the production implementation; this stays as the
one-shot capture-inspection tool.
Usage
-----
python scratch/mm_frame_parse.py <capture-dir>
python scratch/mm_frame_parse.py <raw_bw.bin> <raw_s3.bin>
python scratch/mm_frame_parse.py <capture-dir> --dump 0x71
"""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
DLE, STX, ETX, ACK = 0x10, 0x02, 0x03, 0x41
# Request SUB -> short name. Series III names where they carry over; the
# Series IV additions are marked.
SUBNAME = {
0x01: "DEVICE_INFO",
0x06: "STORAGE_RANGE",
0x08: "EVENT_INDEX",
0x0A: "WAVEFORM_HDR",
0x0C: "WAVEFORM_REC",
0x15: "SERIAL",
0x1A: "COMPLIANCE_CFG",
0x1C: "MONITOR_STATUS",
0x1E: "EVENT_HDR",
0x2C: "CALLHOME_CFG",
0x2E: "TRIGGER_CFG_READ", # Series IV
0x3E: "OPERATOR",
0x41: "SETUP_NAME_READ", # Series IV
0x5A: "BULK_DOWNLOAD",
0x5B: "POLL",
0x68: "EVENT_INDEX_WRITE",
0x69: "WAVEFORM_WRITE",
0x71: "COMPLIANCE_WRITE",
0x72: "CONFIRM_A",
0x73: "CONFIRM_B",
0x74: "CONFIRM_C",
0x82: "TRIGGER_WRITE",
0x83: "TRIGGER_CONFIRM",
0xDA: "SETUP_FILE_DECL", # Series IV — names the target .MMB
0xFE: "FULL_CFG",
}
def destuff(blob: bytes, start: int, *, is_request: bool) -> tuple[bytes, int, int]:
"""Destuff one frame starting at `start`.
Returns (payload, checksum, index_of_terminating_ETX). `payload` excludes
the trailing checksum byte. A request frame opens `ACK STX 10 10`; a
response opens with a bare `STX`.
"""
i = start + (2 if is_request else 1)
out = bytearray()
if is_request:
# The doubled BW_CMD is the one guaranteed stuffed byte.
if blob[i : i + 2] != bytes([DLE, DLE]):
raise ValueError(f"@0x{start:04x}: request does not open with 10 10")
out.append(DLE)
i += 2
while i < len(blob):
b = blob[i]
if b == DLE and i + 1 < len(blob):
out.append(blob[i + 1])
i += 2
continue
if b == ETX:
break
out.append(b)
i += 1
if len(out) < 2:
raise ValueError(f"@0x{start:04x}: frame too short")
return bytes(out[:-1]), out[-1], i
def frames(blob: bytes, *, is_request: bool):
"""Yield (offset, payload, chk, checksum_kind)."""
i, n = 0, len(blob)
while i < n:
if is_request:
if not (blob[i] == ACK and i + 1 < n and blob[i + 1] == STX):
i += 1
continue
elif blob[i] != STX:
i += 1
continue
try:
payload, chk, end = destuff(blob, i, is_request=is_request)
except ValueError:
i += 1
continue
# SUM8 of the destuffed payload is THE rule -- 502/502 captured frames.
# The DLE-aware variant is reported only to name a near-miss; it is
# never a pass. See the module docstring.
if (sum(payload) & 0xFF) == chk:
kind = "ok"
elif (sum(b for b in payload if b != DLE) & 0xFF) == chk:
kind = "BAD(dle-aware)"
else:
kind = "BAD"
yield i, payload, chk, kind
i = end + 1
def describe(payload: bytes, is_request: bool) -> str:
if len(payload) < 3:
return "??"
sub = payload[2]
if is_request:
return SUBNAME.get(sub, f"SUB_{sub:02X}")
req = 0xFF - sub
return "rsp<-" + SUBNAME.get(req, f"SUB_{req:02X}")
def report(path: Path, *, is_request: bool, dump_sub: int | None) -> None:
blob = path.read_bytes()
side = "Thor" if is_request else "unit"
print(f"== {side:4} {path.name} ({len(blob)} bytes)")
n_bad = 0
for idx, (off, p, chk, kind) in enumerate(frames(blob, is_request=is_request)):
if kind == "BAD":
n_bad += 1
sub = p[2] if len(p) > 2 else -1
flags = p[1] if len(p) > 1 else -1
# Requests carry offset at payload[4:6]; responses page at [3:5].
word = int.from_bytes(p[4:6] if is_request else p[3:5], "big")
data = len(p) - 16 if is_request else max(len(p) - 5, 0)
print(
f" [{idx:2}] @0x{off:04x} payload={len(p):5} data={data:5} "
f"flags=0x{flags:02x} SUB=0x{sub:02x} {describe(p, is_request):18} "
f"{'offset' if is_request else 'page'}=0x{word:04x} chk={kind}"
)
if dump_sub is not None and sub == dump_sub:
body = p[16:] if is_request else p[5:]
print(f" ---- data ({len(body)} bytes) ----")
for o in range(0, len(body), 16):
chunk = body[o : o + 16]
txt = "".join(chr(c) if 32 <= c < 127 else "." for c in chunk)
print(f" {o:06x} {chunk.hex(' '):<47} |{txt}|")
print(f" -- {idx + 1} frames, {n_bad} bad checksum\n")
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("paths", nargs="+",
help="a capture directory, or raw_bw.bin and raw_s3.bin")
ap.add_argument("--dump", default=None,
help="hex-dump the data section of this SUB (e.g. 0x71)")
args = ap.parse_args()
dump_sub = int(args.dump, 0) if args.dump else None
if len(args.paths) == 1 and Path(args.paths[0]).is_dir():
d = Path(args.paths[0])
bw = sorted(d.glob("raw_bw_*.bin"))
s3 = sorted(d.glob("raw_s3_*.bin"))
if not bw or not s3:
print(f"{d}: need one raw_bw_*.bin and one raw_s3_*.bin", file=sys.stderr)
return 2
pairs = [(bw[0], True), (s3[0], False)]
elif len(args.paths) == 2:
pairs = [(Path(args.paths[0]), True), (Path(args.paths[1]), False)]
else:
ap.error("pass a capture directory, or exactly two .bin files")
for path, is_request in pairs:
report(path, is_request=is_request, dump_sub=dump_sub)
return 0
if __name__ == "__main__":
sys.exit(main())
-119
View File
@@ -1,119 +0,0 @@
#!/usr/bin/env python3
"""
socat_log_split.py — recover a capture pair from a `socat -x` relay log.
Why this exists
---------------
The bench relay that puts Thor in front of a USB-attached Micromate is:
socat -d -d -x TCP-LISTEN:12345,reuseaddr,fork /dev/ttyACM0,raw,echo=0,b115200 \
> ~/mm-captures/socat_<ts>.log 2>&1
`-x` makes socat hex-dump every byte it forwards, in both directions, with
timestamps. That log is therefore a **complete second copy of every capture**
taken through the relay — independent of whether seismo_lab was recording.
On 2026-09-25 that mattered: a capture's `.bin` files never made it off the
Windows machine, and the session was rebuilt from this log instead. When the
real bins turned up later, the reconstruction was **byte-for-byte identical in
both directions** (3,595 and 4,004 bytes). So this is a validated fallback, not
a lossy approximation.
Log format
----------
```
> 2026/09/25 00:30:35.000276659 length=21 from=0 to=20
41 02 10 10 00 5b 00 00 30 00 ...
2026/09/25 00:30:35 socat[32190] N write(5, 0x..., 21) completed
< 2026/09/25 00:30:35.000384100 length=64 from=0 to=63
02 00 c5 a4 00 00 30 00 ...
```
`>` is data heading toward the serial device (Thor → unit). `<` is data coming
back (unit → Thor). Hex lines are space-separated and indented; socat's own
status lines start with a date and carry no payload.
Usage
-----
# whole log
python scratch/socat_log_split.py socat_20260924_181248.log --out-dir ./recovered
# one session — line numbers from the "accepting connection" markers
grep -n "accepting connection" socat_*.log
python scratch/socat_log_split.py socat_*.log --from-line 919 --out-dir ./recovered
Then parse the result as usual:
python scratch/mm_frame_parse.py recovered/raw_bw.bin recovered/raw_s3.bin
⚠ A log spanning several sessions concatenates them. Split by line number using
the `accepting connection` markers, or the frame walk will run sessions together.
"""
from __future__ import annotations
import argparse
import re
from pathlib import Path
_HEX = re.compile(r"\A[0-9a-f]{2}\Z")
_SOCAT_STATUS = re.compile(r"\A\d{4}/\d{2}/\d{2}")
def split(lines) -> tuple[bytes, bytes]:
"""Return (to_device, from_device) byte streams."""
to_dev, from_dev = bytearray(), bytearray()
cur = None
for line in lines:
if line.startswith(">"):
cur = to_dev
continue
if line.startswith("<"):
cur = from_dev
continue
if _SOCAT_STATUS.match(line):
# socat's own status line ends the current dump block.
cur = None
continue
if cur is None or not line.startswith(" "):
continue
toks = line.split()
if toks and all(_HEX.match(t) for t in toks):
cur.extend(int(t, 16) for t in toks)
return bytes(to_dev), bytes(from_dev)
def main() -> int:
ap = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
ap.add_argument("log", help="a socat -x log file")
ap.add_argument("--out-dir", default=".", help="where to write the .bin pair")
ap.add_argument("--from-line", type=int, default=1,
help="first log line to read (1-based) — use the "
"'accepting connection' marker of the session you want")
ap.add_argument("--to-line", type=int, default=None,
help="last log line to read (1-based, inclusive)")
ap.add_argument("--prefix", default="raw", help="output basename prefix")
args = ap.parse_args()
lines = Path(args.log).read_text(errors="replace").splitlines()
lo = max(args.from_line - 1, 0)
hi = args.to_line if args.to_line is not None else len(lines)
to_dev, from_dev = split(lines[lo:hi])
out = Path(args.out_dir)
out.mkdir(parents=True, exist_ok=True)
bw = out / f"{args.prefix}_bw.bin"
s3 = out / f"{args.prefix}_s3.bin"
bw.write_bytes(to_dev)
s3.write_bytes(from_dev)
print(f"Thor -> unit {len(to_dev):>7} bytes {bw}")
print(f"unit -> Thor {len(from_dev):>7} bytes {s3}")
if not to_dev or not from_dev:
print("⚠ one direction is empty — check --from-line / --to-line")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+20 -295
View File
@@ -108,12 +108,6 @@
color: var(--text);
}
.btn-ghost:hover { border-color: var(--blue-lt); color: var(--blue-lt); }
.btn-danger { background: var(--red); color: #fff; }
.btn-danger:hover:not(:disabled) { filter: brightness(1.15); }
.diag-result { display:block; margin-top:6px; font-size:12px; opacity:.85;
white-space:pre-wrap; word-break:break-word; }
.diag-result.ok { color: var(--green); }
.diag-result.error { color: var(--red); }
.btn:disabled { background: var(--surface2) !important; color: var(--text-mute) !important; cursor: not-allowed; border-color: var(--border2) !important; }
/* #connect-btn styles moved to #live-connect-bar block */
@@ -916,7 +910,6 @@
<button class="tab-btn" data-tab="events" onclick="switchTab('events')">Events</button>
<button class="tab-btn" data-tab="config" onclick="switchTab('config')">Config</button>
<button class="tab-btn" data-tab="call-home" onclick="switchTab('call-home')">Call Home</button>
<button class="tab-btn" data-tab="diagnostics" onclick="switchTab('diagnostics')">Diagnostics</button>
</div>
<!-- ════════════════════════════════════════════════════════════════
@@ -945,10 +938,6 @@
<div id="tab-events" class="tab-pane" style="display:flex; flex-direction:column; overflow:hidden;">
<div class="event-toolbar">
<button class="btn btn-ghost" id="load-events-btn" onclick="loadEventList()" disabled
title="Walk the device's event chain and list its stored events. This is the slow one — it reads every event header over the cellular link.">
⟳ Load events
</button>
<button class="btn btn-ghost" id="load-btn" onclick="loadWaveform()" disabled>Load Waveform</button>
<button class="btn btn-ghost" id="save-btn" onclick="saveEventToDb()" disabled
title="Download the full waveform from the device and save it to the SFM database + waveform store. Honors the Force refresh toggle.">
@@ -1216,77 +1205,6 @@
</div><!-- end #tab-call-home -->
<!-- ════════════════════════════════════════════════════════════════
TAB: Diagnostics
═══════════════════════════════════════════════════════════════════ -->
<div id="tab-diagnostics" class="tab-pane">
<div class="cfg-grid">
<div class="cfg-section">
<div class="cfg-section-title">Device State</div>
<div class="hint" style="margin-bottom:10px">
Fast probes — POLL plus one read each, about 2 s. None of these walk the event chain.
</div>
<div class="dev-table" id="diag-table"></div>
<div class="cfg-actions" style="margin-top:12px">
<button class="btn btn-ghost" id="diag-refresh-btn" onclick="refreshDiagnostics()" disabled>Refresh</button>
<span id="diag-status"></span>
</div>
</div>
<div class="cfg-section">
<div class="cfg-section-title">Actions</div>
<div class="cfg-field">
<label>Stop Monitoring</label>
<button class="btn btn-ghost" id="diag-stop-btn" onclick="diagStopMonitoring()" disabled>Send Stop (SUB 0x97)</button>
<div class="hint">Halts recording. On a unit triggering continuously, this is what breaks the call-home loop.</div>
<span class="diag-result" id="diag-stop-result"></span>
</div>
<div class="cfg-field">
<label>Disable Auto Call Home</label>
<button class="btn btn-ghost" id="diag-ach-btn" onclick="diagDisableAch()" disabled>Disable ACH</button>
<div class="hint">Stored events are left untouched (<code>rescue?erase=false</code>). The unit stops dialing out until ACH is re-enabled.</div>
<span class="diag-result" id="diag-ach-result"></span>
</div>
<div class="cfg-field">
<label>Erase All Events</label>
<input type="text" id="diag-erase-confirm" placeholder="Type the serial to enable"
oninput="diagCheckEraseConfirm()" autocomplete="off" />
<button class="btn btn-danger" id="diag-erase-btn" onclick="diagEraseEvents()" disabled>Erase Events</button>
<div class="hint">⚠ Permanent, and resets the event chain to key <code>0x01110000</code>. Download anything worth keeping first.</div>
<span class="diag-result" id="diag-erase-result"></span>
</div>
</div>
<div class="cfg-section">
<div class="cfg-section-title">Unresponsive Unit</div>
<div class="hint" style="margin-bottom:10px">
The escalation ladder from <code>docs/runbooks/wedged_unit_recovery.md</code>, for a unit too busy
to answer normal request/response. Prefer <b>Method A</b> — point the modem at an
<code>ach_server</code> and answer its call — before racing it with these.
</div>
<div class="cfg-field">
<label>Slow drip <span class="hint" style="display:inline">(one held session, a stop every 3 s)</span></label>
<button class="btn btn-ghost" id="diag-drip-btn" onclick="diagSlowDrip()" disabled>Run 120 s drip</button>
<div class="hint">Success is <code>bytes_received &gt; 0</code>. A full duration with <code>send_error: null</code> is <b>not</b> success on its own.</div>
<span class="diag-result" id="diag-drip-result"></span>
</div>
<div class="cfg-field">
<label>Blind stop <span class="hint" style="display:inline">(fire-and-forget, one attempt)</span></label>
<button class="btn btn-ghost" id="diag-blind-btn" onclick="diagBlindStop()" disabled>Send blind stop</button>
<span class="diag-result" id="diag-blind-result"></span>
</div>
</div>
</div>
</div><!-- end #tab-diagnostics -->
</div><!-- end #section-live -->
<!-- ════════════════════════════════════════════════════════════════
@@ -1443,8 +1361,6 @@
// ── State ──────────────────────────────────────────────────────────────────────
let unitInfo = null;
let eventList = [];
let storageInfo = null; // /device/events/storage_range — cheap, read on connect
let eventsLoaded = false; // the event chain walk is opt-in; see loadEventList()
let currentEvent = 0;
let charts = {};
let geoAdcScale = 6.206;
@@ -1542,7 +1458,6 @@ function switchTab(name) {
if (name === 'units') { if (!unitsLoaded) loadUnits(); }
if (name === 'monlog') { if (!monlogLoaded) loadMonitorLog(); }
if (name === 'sessions') { if (!sessLoaded) loadSessions(); }
if (name === 'diagnostics' && devHost() && unitInfo) refreshDiagnostics();
}
// ── Connect ────────────────────────────────────────────────────────────────────
@@ -1563,13 +1478,18 @@ async function connectUnit() {
btn.disabled = false; btn.textContent = 'Connect'; return;
}
// Connecting deliberately does NOT walk the event chain. That walk reads
// every event header over the cellular link and can take minutes — or fail
// outright on a unit whose buffer has wrapped past 0xFFFF. Use the ~2 s
// probes instead; the event list is opt-in via loadEventList().
eventList = []; eventsLoaded = false;
setStatus('Reading device state…', 'loading');
storageInfo = await fetchJson(`/device/events/storage_range`).catch(() => null);
setStatus('Fetching event list…', 'loading');
try {
const r = await fetch(`${api()}/device/events?${deviceParams()}`);
if (!r.ok) { const e = await r.json().catch(() => ({})); throw new Error(e.detail || r.statusText); }
const evData = await r.json();
eventList = evData.events || [];
// Merge compliance from /device/events response (it re-reads it)
if (evData.device) unitInfo = { ...unitInfo, ...evData.device };
} catch (e) {
setStatus(`Event fetch failed: ${e.message}`, 'error');
btn.disabled = false; btn.textContent = 'Reconnect'; return;
}
populateDeviceBar();
populateDeviceTab();
@@ -1578,9 +1498,11 @@ async function connectUnit() {
document.getElementById('device-bar').style.display = 'flex';
document.getElementById('monitor-panel').style.display = 'flex';
setEventButtonsEnabled();
document.getElementById('load-events-btn').disabled = false;
setDiagButtonsEnabled(true);
document.getElementById('load-btn').disabled = eventList.length === 0;
document.getElementById('save-btn').disabled = eventList.length === 0;
document.getElementById('download-btn').disabled = eventList.length === 0;
document.getElementById('prev-btn').disabled = true;
document.getElementById('next-btn').disabled = eventList.length <= 1;
document.getElementById('cfg-read-btn').disabled = false;
document.getElementById('cfg-write-btn').disabled = false;
document.getElementById('ch-read-btn').disabled = false;
@@ -1588,9 +1510,7 @@ async function connectUnit() {
btn.disabled = false; btn.textContent = 'Reconnect';
setStatus(storageInfo && storageInfo.is_empty
? 'Connected — no events stored.'
: 'Connected. Event list not loaded (Events → Load events).', 'ok');
setStatus(`Connected — ${eventList.length} event${eventList.length !== 1 ? 's' : ''} stored.`, 'ok');
// Fetch monitor status in background (non-blocking)
refreshMonitorStatus().catch(() => {});
@@ -1602,48 +1522,6 @@ async function connectUnit() {
}
}
// ── Shared fetch helper ────────────────────────────────────────────────────────
async function fetchJson(path, opts) {
const sep = path.includes('?') ? '&' : '?';
const r = await fetch(`${api()}${path}${sep}${deviceParams()}`, opts);
const body = await r.json().catch(() => ({}));
if (!r.ok) throw new Error(body.detail || r.statusText);
return body;
}
function setEventButtonsEnabled() {
const n = eventList.length;
document.getElementById('load-btn').disabled = n === 0;
document.getElementById('save-btn').disabled = n === 0;
document.getElementById('download-btn').disabled = n === 0;
document.getElementById('prev-btn').disabled = true;
document.getElementById('next-btn').disabled = n <= 1;
}
// ── Event list (opt-in — this is the slow chain walk) ──────────────────────────
async function loadEventList() {
if (!devHost()) { setStatus('Connect to a device first.', 'error'); return; }
const btn = document.getElementById('load-events-btn');
btn.disabled = true;
setStatus('Walking the event chain — this can take a while…', 'loading');
try {
const evData = await fetchJson('/device/events');
eventList = evData.events || [];
eventsLoaded = true;
// /device/events re-reads compliance; fold it in.
if (evData.device) unitInfo = { ...unitInfo, ...evData.device };
} catch (e) {
setStatus(`Event fetch failed: ${e.message}`, 'error');
btn.disabled = false; return;
}
populateDeviceBar();
populateDeviceTab();
populateEventChips();
setEventButtonsEnabled();
btn.disabled = false;
setStatus(`${eventList.length} event${eventList.length !== 1 ? 's' : ''} stored.`, 'ok');
}
// ── Device bar ─────────────────────────────────────────────────────────────────
function populateDeviceBar() {
qs('di-serial').textContent = unitInfo.serial || '—';
@@ -1652,7 +1530,7 @@ function populateDeviceBar() {
qs('di-sr').textContent = cc.sample_rate ? `${cc.sample_rate} sps` : '—';
qs('di-rt').textContent = cc.record_time != null ? `${cc.record_time.toFixed(1)} s` : '—';
qs('di-trig').textContent = cc.trigger_level_geo != null ? `${cc.trigger_level_geo.toFixed(3)} in/s` : '—';
qs('di-count').textContent = eventsLoaded ? eventList.length : '—';
qs('di-count').textContent = eventList.length;
qs('di-project').textContent = cc.project || '—';
qs('di-client').textContent = cc.client || '—';
qs('di-operator').textContent = cc.operator || '—';
@@ -1782,8 +1660,7 @@ function populateDeviceTab() {
{ label:'DSP', value: unitInfo.dsp_version || '—' },
{ label:'Model', value: unitInfo.model || '—' },
{ label:'Manufacturer', value: unitInfo.manufacturer || '—' },
{ label:'Stored Events', value: eventsLoaded ? eventList.length : 'not loaded' },
{ label:'Storage Used', value: storageUsedLabel() },
{ label:'Stored Events', value: eventList.length },
];
for (const {label, value} of cardData) {
const c = document.createElement('div');
@@ -1830,158 +1707,6 @@ function renderTable(id, rows) {
}
}
// ── Diagnostics ────────────────────────────────────────────────────────────────
// Everything here is a cheap probe (POLL + one read) or a single write. None of
// it walks the event chain. See docs/runbooks/wedged_unit_recovery.md.
function storageUsedLabel() {
if (!storageInfo) return '—';
if (storageInfo.is_empty) return 'empty';
const f = storageInfo.first_key, l = storageInfo.last_key;
return (f && l) ? `${f} → ${l}` : '—';
}
function setDiagButtonsEnabled(on) {
for (const id of ['diag-refresh-btn','diag-stop-btn','diag-ach-btn',
'diag-drip-btn','diag-blind-btn']) {
const el = document.getElementById(id);
if (el) el.disabled = !on;
}
diagCheckEraseConfirm();
}
// Erase is guarded by typing the serial — auth answers "who", not "did you mean it".
function diagCheckEraseConfirm() {
const box = document.getElementById('diag-erase-confirm');
const btn = document.getElementById('diag-erase-btn');
if (!box || !btn) return;
const serial = (unitInfo && unitInfo.serial) || '';
btn.disabled = !serial || box.value.trim().toUpperCase() !== serial.toUpperCase();
}
function diagResult(id, text, cls) {
const el = document.getElementById(id);
if (!el) return;
el.textContent = text;
el.className = 'diag-result' + (cls ? ' ' + cls : '');
}
async function refreshDiagnostics() {
if (!devHost()) return;
const st = document.getElementById('diag-status');
if (st) { st.textContent = 'Reading…'; st.className = 'loading'; }
const [mon, store, idx] = await Promise.all([
fetchJson('/device/monitor/status?force=true').catch(e => ({ _err: e.message })),
fetchJson('/device/events/storage_range').catch(e => ({ _err: e.message })),
fetchJson('/device/events/index').catch(e => ({ _err: e.message })),
]);
if (!store._err) storageInfo = store;
const err = v => `<span style="color:var(--red)">${v}</span>`;
const rows = [];
rows.push(['Monitoring', mon._err ? err(mon._err)
: (mon.is_monitoring ? '<b>MONITORING</b>' : 'idle')]);
if (!mon._err) {
rows.push(['Battery', mon.battery_v != null ? `${mon.battery_v.toFixed(2)} V` : '—']);
if (mon.memory_total_bytes) {
const used = mon.memory_total_bytes - (mon.memory_free_bytes ?? 0);
const pct = (used / mon.memory_total_bytes * 100).toFixed(1);
rows.push(['Memory used', `${used.toLocaleString()} / ${mon.memory_total_bytes.toLocaleString()} bytes (${pct}%)`]);
}
}
rows.push(['Event chain', store._err ? err(store._err) : storageUsedLabel()]);
if (!store._err) rows.push(['Chain empty', store.is_empty ? 'yes' : 'no']);
// SUB 0x08. Known to report 0 on units with years of history — suspected
// field-offset bug in the decode, so show it but do not trust it.
rows.push(['Lifetime events', idx._err ? err(idx._err)
: `${idx.lifetime_count} <span class="hint" style="display:inline">(unreliable — see CHANGELOG)</span>`]);
renderTable('diag-table', rows);
populateDeviceTab();
if (st) { st.textContent = ''; st.className = ''; }
}
async function diagStopMonitoring() {
const btn = document.getElementById('diag-stop-btn');
btn.disabled = true; diagResult('diag-stop-result', 'Sending…');
try {
await fetchJson('/device/monitor/stop', { method: 'POST' });
diagResult('diag-stop-result', 'Stop acknowledged — recording halted.', 'ok');
refreshDiagnostics();
} catch (e) {
diagResult('diag-stop-result', `Failed: ${e.message}`, 'error');
}
btn.disabled = false;
}
async function diagDisableAch() {
const btn = document.getElementById('diag-ach-btn');
btn.disabled = true; diagResult('diag-ach-result', 'Writing call-home config…');
try {
const r = await fetchJson('/device/rescue?erase=false', { method: 'POST' });
const steps = (r.steps || []).map(s => s.step).join(' → ') || 'done';
diagResult('diag-ach-result', `ACH disabled (${steps}). Events untouched.`, 'ok');
} catch (e) {
diagResult('diag-ach-result', `Failed: ${e.message}`, 'error');
}
btn.disabled = false;
}
async function diagEraseEvents() {
const serial = (unitInfo && unitInfo.serial) || 'this unit';
if (!confirm(`Permanently erase ALL events on ${serial}?\n\nThis cannot be undone.`)) return;
const btn = document.getElementById('diag-erase-btn');
btn.disabled = true; diagResult('diag-erase-result', 'Erasing…');
try {
await fetchJson('/device/events/erase', { method: 'POST' });
diagResult('diag-erase-result', 'Events erased — chain reset to 0x01110000.', 'ok');
document.getElementById('diag-erase-confirm').value = '';
eventList = []; eventsLoaded = false;
setEventButtonsEnabled(); populateEventChips();
refreshDiagnostics();
} catch (e) {
diagResult('diag-erase-result', `Failed: ${e.message}`, 'error');
}
diagCheckEraseConfirm();
}
async function diagSlowDrip() {
const btn = document.getElementById('diag-drip-btn');
btn.disabled = true;
diagResult('diag-drip-result', 'Holding a session for 120 s…');
try {
const r = await fetchJson('/device/stop_monitoring_slow_drip?duration_s=120&interval_s=3',
{ method: 'POST' });
const good = (r.bytes_received || 0) > 0;
diagResult('diag-drip-result',
`drips ${r.drips_sent} · held ${r.duration_s}s · bytes back ${r.bytes_received}` +
(r.send_error ? ` · ${r.send_error}` : '') +
(good ? ' → device responded' : ' → no response; the modem may not be bridging'),
good ? 'ok' : 'error');
} catch (e) {
diagResult('diag-drip-result', `Failed: ${e.message}`, 'error');
}
btn.disabled = false;
}
async function diagBlindStop() {
const btn = document.getElementById('diag-blind-btn');
btn.disabled = true; diagResult('diag-blind-result', 'Sending…');
try {
const r = await fetchJson('/device/stop_monitoring_blind', { method: 'POST' });
diagResult('diag-blind-result',
`Sent ${r.bytes_sent ?? '?'} bytes, no response read (fire-and-forget).`, 'ok');
} catch (e) {
diagResult('diag-blind-result', `Failed: ${e.message}`, 'error');
}
btn.disabled = false;
}
// ── Config form ────────────────────────────────────────────────────────────────
function populateConfigFromDeviceInfo() {
if (!unitInfo) return;
Binary file not shown.
-54
View File
@@ -1,54 +0,0 @@
"""Event timestamp decode — waveform trigger/stop vs histogram window start.
The Blastware footer holds two timestamps: ts1 = footer[2:10], ts2 = footer[10:18].
Their meaning depends on record type:
* Waveform: ts1 is the monitoring-SESSION start (e.g. 06:00 for a unit that
arms at 06:00 daily — shared across every event that day), and ts2 is THIS
event's recording STOP. read_blastware_file used to stamp events with ts1 →
every waveform showed the session start (~4.5 h off). Binary-only, the best
estimate is ts2 (the stop); the exact trigger BW displays (= ts2 - record
duration) comes from the paired report's event_datetime, since the binary
STRT record-time byte is a misparsed record-type marker.
* Histogram: ts1/ts2 are the ~24 h window [start, stop]; the event time is the
window start = ts1 (unchanged).
"""
import datetime
from pathlib import Path
from minimateplus.event_file_io import read_blastware_file, apply_report_to_event
from minimateplus.bw_ascii_report import BwAsciiReport
from minimateplus.models import Event
FIX = Path(__file__).parent / "fixtures"
WAVEFORM = FIX / "fft-oracle-2026-09-14" / "N844LQHB.ZT0W" # footer ts2 = 2026-08-25 10:33:32
HISTOGRAM = FIX / "ts-fix" / "K441LKZU.C30H" # window start 2026-05-10 19:04:50
def _tuple(ts):
return (ts.year, ts.month, ts.day, ts.hour, ts.minute, ts.second)
def test_waveform_timestamp_is_exact_trigger_from_binary():
ev = read_blastware_file(WAVEFORM)
# The EXACT Blastware trigger, from the binary alone: ts2 (stop 10:33:32)
# minus the config record time (3.0 s) = 10:33:29 — NOT the 06:00:13
# monitoring-session start the old decode used.
assert _tuple(ev.timestamp) == (2026, 8, 25, 10, 33, 29), _tuple(ev.timestamp)
def test_histogram_timestamp_is_window_start_unchanged():
ev = read_blastware_file(HISTOGRAM)
# Histogram event time = the window start (ts1); must NOT get the waveform
# ts2 treatment (that would land ~24 h off).
assert _tuple(ev.timestamp) == (2026, 5, 10, 19, 4, 50), _tuple(ev.timestamp)
def test_report_event_datetime_is_authoritative_over_binary():
# The binary already yields the exact trigger, but a paired report stays
# authoritative (e.g. if the unit clock had drifted) — applying it wins.
ev = read_blastware_file(WAVEFORM)
assert _tuple(ev.timestamp) == (2026, 8, 25, 10, 33, 29) # exact, from binary
apply_report_to_event(ev, BwAsciiReport(
event_datetime=datetime.datetime(2026, 8, 25, 10, 35, 0)))
assert _tuple(ev.timestamp) == (2026, 8, 25, 10, 35, 0) # report wins
-493
View File
@@ -1,493 +0,0 @@
"""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"
)
# ── Real 11.0BD bytes (UM20147, captured over USB 2026-09-30) ─────────────────
#
# The Thor firmware line, which was pure inference until this capture. These
# are the two responses where BD differs from CB. Captured with
# `mm_client_check.py --capture` and read out of the resulting pair with
# scratch/mm_frame_parse.py, so the data sections are verbatim; only the frame
# wrapper is rebuilt, and the framing is independently verified 251/251.
POLL_BD = bytes.fromhex( # 59 B -- flags 0x03, and the SHORTER model string
"30000000000000000000000000005649"
"6e7374616e74656c000600c3f04a00e4"
"194f0052024d4d2f495345452f530058"
"f406001f60c0755760c075"
)
MS_BD_MONITORING = bytes.fromhex( # 59 B -- FOUR BYTES LONGER than CB's 55
"3000000000000000000000000e1e0907"
"ea600e00310000000000000000000000"
"00000000000000000000000000015e00"
"e4e1c000e291c00fa00004"
)
_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.
✅ CONFIRMED on real hardware 2026-09-30: UM20147 (11.0BD) read back
model="MM/ISEE/S" over USB. The frame below is still synthesised because
no BD capture is in the repo, but the string it asserts is the real one.
"""
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()
# ── The Thor firmware line, on real bytes ─────────────────────────────────────
def test_bd_constants_have_the_lengths_the_parser_reported():
assert len(POLL_BD) == 59
assert len(MS_BD_MONITORING) == 59, "CB is 55; BD is four bytes longer"
assert MS_BD_MONITORING[0] == 0x30, "low-byte length 48 = 59 - 11"
assert MS_IDLE[0] == 0x2C, "the CB equivalent declares 44"
def test_connect_on_a_thor_line_unit():
"""UM20147, real POLL bytes. flags 0x03 is ETX, so it arrives as `10 03`."""
mm, _ = client([
frame(0xA4, POLL_BD, flags=FLAGS_THOR), frame(0xEA, SERIAL),
frame(0xB6, STATE_MONITORING), frame(0xBE, setup_response("test2.MMB")),
])
info = mm.connect()
assert info.firmware_line == "thor"
assert info.model == "MM/ISEE/S", "the BD model string is shorter than CB's"
assert info.manufacturer == "Instantel"
assert info.monitoring is True
def test_poll_content_3_is_not_a_constant():
"""⚠ 0x50 on UM12947, 0x56 on UM20147 -- it VARIES between units.
This is why the manufacturer is read at the fixed offset content[4] and not
by scanning: content[3] is printable in both cases ("P" and "V"), so a
printable-run scan would return "PInstantel" on one unit and "VInstantel" on
the other. Whatever the byte is, it is not a stable marker to anchor on.
"""
assert _content(POLL)[3] == 0x50
assert _content(POLL_BD)[3] == 0x56
assert chr(_content(POLL_BD)[3]) == "V"
def test_get_state_on_a_thor_line_unit():
"""⚠ THE test for the forward-offset decision. Real UM20147 bytes.
The 0x1C block is four bytes longer here, and the extras are TRAILING, so
offsets measured from the start of content are unmoved. Series III reads
battery and memory from the END of this block; test_series_iii_offsets_...
below shows what that produces.
"""
mm, _ = client([frame(0xE3, MS_BD_MONITORING, flags=FLAGS_THOR)])
st = mm.get_state()
assert st.monitoring is True
assert st.device_time == datetime.datetime(2026, 9, 30, 14, 0, 49)
assert st.battery_volts == 3.50
assert st.memory_total_bytes == 15_000_000
assert st.memory_free_bytes == 14_848_448
def test_series_iii_from_the_end_offsets_give_577_volts_on_a_bd_unit():
"""The exact number the protocol reference warned about, now demonstrated.
577.92 V is not a plausible battery reading for anything, which is what
makes it a free self-check -- bridges/mm_client_check.py watches for it.
"""
assert int.from_bytes(MS_BD_MONITORING[-10:-8], "big") / 100 == 577.92
def test_the_four_extra_bd_bytes_are_not_all_zero():
"""⚠ The reference records them as `0f a0 00 00`; UM20147 sent `0f a0 00 04`.
Only the first two bytes look fixed. Nothing reads them, but a future
decoder must not treat the last one as padding.
"""
assert _content(MS_BD_MONITORING)[44:] == bytes.fromhex("0fa00004")
assert _content(MS_IDLE)[44:] == b"", "the CB block has no such tail"
-343
View File
@@ -1,343 +0,0 @@
"""Event-chain tests for the Micromate (series-4) client.
The load-bearing test here replays THOR's captured six-event download session
through `MicromateClient.iter_events()` + `get_event()` and asserts **every byte
we put on the wire matches what THOR put on the wire** — 99 frames — while also
decoding all six events and cross-checking each waveform's peak vector sum
against the one the device computed itself.
That capture is gitignored, so those tests skip on a fresh clone; the offline
tests below use embedded real response bytes and cover the same logic.
"""
from __future__ import annotations
import datetime
import os
import sys
from pathlib import Path
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from micromate.client import CONTENT, DecodeMismatch, MicromateClient
from micromate.framing import ACK, DLE, ETX, STX, checksum, stuff
from micromate.models import MicromateEventRef
from micromate.protocol import ProtocolError
from test_micromate_client import ScriptedTransport, SERIAL, frame
# The real 0x0C record from UM20147 (11.0BD), captured over USB 2026-09-30.
# A histogram: content[11] = 0x08.
RECORD_BD = bytes.fromhex(
"d200000000055d4a8100001e0907eab30d1b21000000084c6f636174696f6e00"
"0000000000000000000000000074657374320000000000000000000000000000"
"0000000000000000000000000000000000000000000000554d32303134370000"
"003fcb3bde00000000053f000f5472616e00003e698cdb000300005665727400"
"003fc76e65000300004c6f6e6700003e567c21000300004d69630000003956b9"
"7c00050000000000000000000000000000000000000000000000000000000000"
"0000000000000000000000000000000000000000000000000000000000"
)
def chain_entry(key: bytes, size: int) -> bytes:
"""A 1E/1F response data section: 11-byte prefix then key + size."""
return (bytes([0x08]) + bytes(7) + bytes([0xFE]) + bytes(2)
+ key + size.to_bytes(4, "big"))
def client(responses):
t = ScriptedTransport(responses)
return MicromateClient(t, recv_timeout=0.5), t
# ── The chain walk ────────────────────────────────────────────────────────────
def test_chain_walk_arms_before_every_entry():
"""⚠ THOR sends 0x93 before EVERY 1E/1F, and this mirrors that."""
responses = [
frame(0xEA, SERIAL), # serial()
frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes.fromhex("055d4a82"), 11032)),
frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)), # sentinel
]
mm, t = client(responses)
refs = mm.list_events(with_records=False)
subs = [w[5] for w in t.written]
assert subs == [0x15, 0x93, 0x1E, 0x93, 0x1F, 0x93, 0x1F]
assert [r.key_hex for r in refs] == ["055d4a81", "055d4a82"]
assert [r.size for r in refs] == [4076, 11032]
def test_an_all_zero_key_ends_the_chain_and_is_not_an_error():
mm, _ = client([
frame(0xEA, SERIAL),
frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes(4), 0)),
])
assert mm.list_events(with_records=False) == []
def test_refs_carry_the_serial_because_a_key_alone_is_ambiguous():
"""⚠ UM12947 and UM20147 BOTH have an event 055d4a81.
A store keyed on the event key alone treats one unit's event as a duplicate
of the other's, and nothing raises. `uid` is the safe identifier.
"""
mm, _ = client([
frame(0xEA, SERIAL),
frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
])
(ref,) = mm.list_events(with_records=False)
assert ref.serial == "UM12947"
assert ref.uid == "UM12947:055d4a81"
def test_a_cursor_that_never_advances_raises_rather_than_hanging():
from micromate import client as C
responses = [frame(0xEA, SERIAL)]
entry = chain_entry(bytes.fromhex("055d4a81"), 4076)
responses += [frame(0x6C, bytes(11)), frame(0xE1, entry)] # the 1E read
for _ in range(C._MAX_EVENTS + 2): # then 1F forever
responses += [frame(0x6C, bytes(11)), frame(0xE0, entry)]
mm, _ = client(responses)
with pytest.raises(ProtocolError, match="not advancing"):
mm.list_events(with_records=False)
def test_a_truncated_chain_entry_raises():
mm, _ = client([
frame(0xEA, SERIAL), frame(0x6C, bytes(11)), frame(0xE1, bytes(14)),
])
with pytest.raises(ProtocolError, match="need 8"):
mm.list_events(with_records=False)
def test_iter_events_does_not_read_ahead():
"""It must yield at the cursor position, one arm/advance per event."""
mm, t = client([
frame(0xEA, SERIAL),
frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
])
it = mm.iter_events(with_records=False)
first = next(it)
assert [w[5] for w in t.written] == [0x15, 0x93, 0x1E], "no read-ahead"
assert first.key_hex == "055d4a81"
assert list(it) == []
# ── The 0x0C record ───────────────────────────────────────────────────────────
def test_record_decode_on_real_bd_bytes():
mm, _ = client([
frame(0xEA, SERIAL),
frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4796)),
frame(0xF3, RECORD_BD),
frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
])
(ref,) = mm.list_events()
assert ref.record_type == "histogram" # content[11] = 0x08
assert ref.is_histogram is True
assert ref.suffix == ".IDFH"
assert ref.timestamp == datetime.datetime(2026, 9, 30, 13, 27, 33)
assert ref.sensor_location == "Location"
assert ref.setup == "test2"
assert ref.serial == "UM20147", "the record's own serial wins over the cache"
assert ref.peak_vector_sum_ips == pytest.approx(1.587765, abs=1e-5)
assert ref.peaks_ips["Tran"] == pytest.approx(0.228076, abs=1e-5)
assert ref.peaks_ips["Vert"] == pytest.approx(1.558056, abs=1e-5)
assert ref.peaks_ips["Long"] == pytest.approx(0.209458, abs=1e-5)
def test_the_pvs_is_not_reconstructible_from_the_reported_peaks():
"""⚠ Why the PVS field matters: it cannot be recomputed from the peaks.
It is the PER-SAMPLE peak vector sum. `sqrt(Σpeak²)` is only an upper
bound, because the channel maxima do not occur at the same instant, and
`max(T,V,L)` is a lower bound. On THIS event the two happen to be within
0.05%, which is exactly the coincidence that made the field look like
`sqrt(Σpeak²)` on first inspection. Six other events separated them.
"""
import math
mm, _ = client([
frame(0xEA, SERIAL),
frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4796)),
frame(0xF3, RECORD_BD),
frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
])
(ref,) = mm.list_events()
p = ref.peaks_ips
upper = math.sqrt(p["Tran"] ** 2 + p["Vert"] ** 2 + p["Long"] ** 2)
lower = max(p["Tran"], p["Vert"], p["Long"])
assert lower < ref.peak_vector_sum_ips < upper
def test_filename_matches_thors_convention():
ref = MicromateEventRef(index=0, key=bytes.fromhex("055d4a82"), size=11032,
serial="UM12947", record_type="waveform",
timestamp=datetime.datetime(2026, 9, 23, 16, 33, 19))
assert ref.filename == "UM12947_20260923163319.IDFW"
ref.record_type = "histogram"
assert ref.filename == "UM12947_20260923163319.IDFH"
def test_filename_is_none_without_a_type_rather_than_guessing():
"""⚠ Guessing would file a histogram as a waveform, and read_idf_file()
dispatches on exactly that suffix."""
ref = MicromateEventRef(index=0, key=bytes(4), size=1, serial="UM12947",
timestamp=datetime.datetime(2026, 1, 1))
assert ref.record_type is None
assert ref.suffix is None
assert ref.filename is None
def test_an_unknown_record_type_warns_and_leaves_it_none(caplog):
bad = bytearray(RECORD_BD)
bad[CONTENT + 11] = 0x99
mm, _ = client([
frame(0xEA, SERIAL),
frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 10)),
frame(0xF3, bytes(bad)),
frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
])
with caplog.at_level("WARNING"):
(ref,) = mm.list_events()
assert ref.record_type is None
assert "unknown record type 0x99" in caplog.text
def test_get_event_refuses_without_a_record_type():
mm, _ = client([])
ref = MicromateEventRef(index=0, key=bytes.fromhex("055d4a81"), size=4076)
with pytest.raises(ValueError, match="record_type is unknown"):
mm.get_event(ref)
# ── Against the real captured session ─────────────────────────────────────────
_CAPTURES = (
Path(__file__).resolve().parents[1]
/ "bridges" / "captures" / "9-24-26 - micromate2"
)
_DOWNLOAD = "raw_bw_20260925_011403_Download_events_then_delete_1_event.bin"
def _destuffed(blob: bytes, is_req: bool):
i, n = 0, len(blob)
while i < n:
if is_req:
if not (blob[i] == ACK and i + 1 < n and blob[i + 1] == STX):
i += 1
continue
j = i + 2
else:
if blob[i] != STX:
i += 1
continue
j = i + 1
out = bytearray()
while j < n:
if blob[j] == DLE and j + 1 < n:
out.append(blob[j + 1])
j += 2
continue
if blob[j] == ETX:
break
out.append(blob[j])
j += 1
if len(out) >= 6:
yield blob[i:j + 1], bytes(out[:-1])
i = j + 1
@pytest.mark.skipif(
not (_CAPTURES / _DOWNLOAD).is_file(),
reason="capture is gitignored; present only on a dev box",
)
def test_replaying_thors_session_reproduces_every_byte_and_decodes_every_event():
"""The whole point of step 4.
Feed THOR's own responses to `iter_events()` + `get_event()`, and assert:
* every request byte we emit matches THOR's, in order
* all six events decode
* each waveform's decoded PVS matches the device's stored float
"""
reqs = list(_destuffed((_CAPTURES / _DOWNLOAD).read_bytes(), True))
rsps = list(_destuffed(
(_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False))
# THOR's session opens with commands our client does not send (POLL, 0x1C,
# 0x06 …) and ends with a delete. Take the contiguous run from the first
# 0x93 to the last 0x5A -- that is the event walk, and it is what we mirror.
subs = [p[2] for _, p in reqs]
lo = subs.index(0x93)
hi = len(subs) - 1 - subs[::-1].index(0x5A)
want_wire = [w for w, _ in reqs[lo:hi + 1]]
replay = [bytes([STX]) + stuff(p + bytes([checksum(p)])) + bytes([ETX])
for _, p in rsps[lo:hi + 1]]
# ⚠ THOR never reads the chain sentinel in this capture -- it downloaded
# exactly six events and stopped, so it knew the count in advance. It read
# `SUB 0x06` (storage range) at frame 7, BEFORE the walk, and that response
# begins `00 00 00 06` -- the event count. Our generator walks until the
# sentinel instead, so append one and compare only the overlapping frames.
replay += [frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0))]
# serial() would emit a 0x15 that is not in this slice, so prime the cache.
t = ScriptedTransport(replay)
mm = MicromateClient(t, recv_timeout=1.0)
mm._serial = "UM12947"
decoded, checked = 0, 0
for ref in mm.iter_events():
result = mm.get_event(ref) # verify=True by default
decoded += 1
assert ref.serial == "UM12947"
assert ref.filename and ref.filename.endswith(ref.suffix)
if not ref.is_histogram:
err = mm.decode_error(ref, result)
assert err is not None
assert abs(err) < 1e-4, f"{ref.uid}: PVS off by {100 * err:+.4f}%"
checked += 1
assert decoded == 6, "four waveforms and two histograms"
assert checked == 4
# Our trailing sentinel read is two frames THOR did not send; everything
# up to it must match byte for byte, in order.
assert t.written[:len(want_wire)] == want_wire, (
f"we emitted {len(t.written)} frames, THOR emitted {len(want_wire)}"
)
assert len(t.written) == len(want_wire) + 2, "only the sentinel read is extra"
@pytest.mark.skipif(
not (_CAPTURES / _DOWNLOAD).is_file(),
reason="capture is gitignored; present only on a dev box",
)
def test_a_corrupted_stored_peak_is_caught_by_the_self_check():
"""Prove the verify path actually fires -- otherwise it is decoration."""
reqs = list(_destuffed((_CAPTURES / _DOWNLOAD).read_bytes(), True))
rsps = list(_destuffed(
(_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False))
subs = [p[2] for _, p in reqs]
lo = subs.index(0x93)
hi = len(subs) - 1 - subs[::-1].index(0x5A)
replay = [bytes([STX]) + stuff(p + bytes([checksum(p)])) + bytes([ETX])
for _, p in rsps[lo:hi + 1]]
t = ScriptedTransport(replay)
mm = MicromateClient(t, recv_timeout=1.0)
mm._serial = "UM12947"
for ref in mm.iter_events():
if ref.is_histogram:
mm.download_event(ref) # keep the replay in step
continue
ref.peak_vector_sum_ips *= 1.5 # as a bad decode would look
with pytest.raises(DecodeMismatch, match="decode is suspect"):
mm.get_event(ref)
return
pytest.fail("no waveform event found in the capture")
-413
View File
@@ -1,413 +0,0 @@
"""Framing tests for the Micromate (series-4) live protocol.
Every constant below is a **real frame**, lifted from
``bridges/captures/9-24-26 - micromate2/`` (UM12947, firmware 11.0CB) or from
``scratch/fake_unit.py``, which preserves a POLL probe reply. Frames are
embedded as hex rather than read from disk because both ``bridges/captures/``
and ``tests/fixtures/`` are gitignored -- these tests must pass on a fresh
clone.
The few synthesised frames are marked ``SYNTH_`` and each says what it stands
in for and why a captured frame was not available.
Two of these tests exist because the first draft of
``docs/micromate_client_spec.md`` got the rule wrong, and both wrong rules
fail quietly -- a frame the unit ignores, or a checksum that reads as bad:
* ``test_builder_matches_thor_byte_for_byte`` -- the escape set. Escaping
only 0x10 (the series-3 rule) reproduces 161 of Thor's 218 read frames.
* ``test_checksum_is_plain_sum8_over_destuffed_payload`` -- the checksum.
The DLE-aware variant disagrees with the wire on 55 of 251 responses.
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from micromate.framing import (
ACK,
DLE,
ETX,
FLAGS_BLASTWARE,
FLAGS_THOR,
STX,
MicromateFrame,
MicromateFrameParser,
build_request,
checksum,
stuff,
unstuff,
)
# ── Captured response frames ──────────────────────────────────────────────────
# POLL probe reply, 19 B on the wire -- the shortest valid frame there is.
# Preserved in scratch/fake_unit.py, captured from UM12947 on 2026-09-24.
# Its payload[8:10] is 0x0030, the data length POLL then asks for.
RSP_POLL_PROBE = bytes.fromhex("0200c5a400000000000030000000000000009903")
# SUB 0x49 -> 0xB6, the cheap state read. Carries a literal 0x02, escaped.
RSP_STATE = bytes.fromhex("0200c5b6000005000000000000000000001002e8000b007503")
# SUB 0x48 -> 0xB7, a path-addressed file read. offset_hi 0x04 arrives escaped,
# so page_hi is only correct if the parser destuffs before indexing.
RSP_FILE_READ = bytes.fromhex("0200c5b70000100400000000000000010000000000018203")
# An 0x5A download chunk, 138 B on the wire. THE checksum case: its payload
# holds literal 0x10 bytes, so plain SUM8 (0xC1, correct) and the DLE-aware
# variant (0x91) disagree. Also holds a literal 0x41, which is NOT escaped.
RSP_CHUNK_WITH_DLE = bytes.fromhex(
"0200c5a500007000003400000000000000e3fd1f10020f0f0e2e1e1fd4f2d4c3d2f00f000"
"e003fe03fe200d6e2d12d2e1c4e101011d101f3d0f3e0f2efe23d2b2e3c0d3e0e2c4101e0"
"d4d3e21f1d311e202fe2e2f1e3f100210d201002ee2e22d2f2f0f30fd3f0101000f0111f1"
"ff01e011010011f0f1002b4d200e0f0302c4010020001fffedb2dc103"
)
# ── Captured request frames (Thor -> unit) ────────────────────────────────────
REQ_POLL = bytes.fromhex("41021010005b000030000000000000000000009b03")
REQ_SERIAL = bytes.fromhex("41021010001500000a000000000000000000002f03")
REQ_STATE = bytes.fromhex("41021010004900ffff000000000000000000005703")
REQ_COMPLIANCE = bytes.fromhex("41021010001a00ffff000000000000000000002803")
REQ_ARM_EVENT = bytes.fromhex("41021010009300ffff00000000000000000000a103")
# ⚠ The two frames that break a 0x10-only escaper.
# Scheduler enable: params[7] = 0x03, on the wire as `10 03`.
REQ_SCHED_ON = bytes.fromhex("41021010004700ffff00000000000000100300005803")
# Bulk download, first chunk of event 055d4a81: offset 0x0400 puts a literal
# 0x04 in offset_hi, on the wire as `10 04`. EVERY download frame needs this.
REQ_DOWNLOAD_CHUNK0 = bytes.fromhex(
"41021010005a00100400055d4a810000000000009b03"
)
# A later chunk of the same event: params[2:4] = 0x1000, doubled to `10 10`.
REQ_DOWNLOAD_CHUNK4 = bytes.fromhex(
"41021010005a0010040000001010000000000000007e03"
)
# ── Stuffing ──────────────────────────────────────────────────────────────────
def test_escape_set_is_exactly_four_bytes():
"""0x02, 0x03, 0x04 and 0x10 -- and nothing else.
ACK (0x41) in particular is NOT escaped; assuming it was reproduced only
196 of 251 captured responses.
"""
assert stuff(bytes([0x02, 0x03, 0x04, 0x10])) == bytes(
[DLE, 0x02, DLE, 0x03, DLE, 0x04, DLE, 0x10]
)
for b in (0x00, 0x01, 0x05, 0x41, 0xC5, 0xFF):
assert stuff(bytes([b])) == bytes([b]), f"0x{b:02x} must not be escaped"
def test_unstuff_is_uniform_with_no_inner_frame_carve_out():
"""`10 XX` -> `XX` for any XX -- the series-3 DLE+ETX exception is absent."""
assert unstuff(bytes.fromhex("1003")) == b"\x03"
assert unstuff(bytes.fromhex("1010")) == b"\x10"
assert unstuff(bytes.fromhex("001002ff")) == bytes.fromhex("0002ff")
def test_stuff_unstuff_round_trips_over_every_byte_value():
data = bytes(range(256))
assert unstuff(stuff(data)) == data
def test_a_trailing_dle_is_held_not_dropped():
"""A DLE as the last byte of a chunk must not consume nothing and vanish."""
assert unstuff(b"\xff\x10") == b"\xff\x10"
# ── Checksum ──────────────────────────────────────────────────────────────────
def test_checksum_is_plain_sum8_over_destuffed_payload():
"""⚠ Plain SUM8 -- do NOT exclude 0x10 bytes.
RSP_CHUNK_WITH_DLE is a real frame whose payload holds literal 0x10 bytes.
The wire says 0xC1; plain SUM8 gives 0xC1 and the DLE-aware variant used by
series-3 `5A`/write frames gives 0x91. 55 of 251 captured responses
disagree the same way.
"""
payload = unstuff(RSP_CHUNK_WITH_DLE[1:-1])[:-1]
chk_on_wire = unstuff(RSP_CHUNK_WITH_DLE[1:-1])[-1]
assert DLE in payload, "this frame is only interesting if it holds a 0x10"
assert checksum(payload) == chk_on_wire == 0xC1
assert (sum(b for b in payload if b != DLE) & 0xFF) == 0x91 # the wrong rule
def test_every_captured_frame_validates():
parser = MicromateFrameParser()
frames = parser.feed(
RSP_POLL_PROBE + RSP_STATE + RSP_FILE_READ + RSP_CHUNK_WITH_DLE
)
assert len(frames) == 4
assert all(f.checksum_valid for f in frames)
# ── Request builder ───────────────────────────────────────────────────────────
@pytest.mark.parametrize(
"wire, sub, offset, params",
[
(REQ_POLL, 0x5B, 0x0030, bytes(10)),
(REQ_SERIAL, 0x15, 0x000A, bytes(10)),
(REQ_STATE, 0x49, 0xFFFF, bytes(10)),
(REQ_COMPLIANCE, 0x1A, 0xFFFF, bytes(10)),
(REQ_ARM_EVENT, 0x93, 0xFFFF, bytes(10)),
(REQ_SCHED_ON, 0x47, 0xFFFF, bytes.fromhex("00000000000000030000")),
(REQ_DOWNLOAD_CHUNK0, 0x5A, 0x0400, bytes.fromhex("055d4a81000000000000")),
(REQ_DOWNLOAD_CHUNK4, 0x5A, 0x0400, bytes.fromhex("00001000000000000000")),
],
ids="poll serial state compliance arm sched_on dl_chunk0 dl_chunk4".split(),
)
def test_builder_matches_thor_byte_for_byte(wire, sub, offset, params):
assert build_request(sub, offset, params) == wire
def test_a_0x10_in_params_needs_no_special_handling():
"""The spec's one open question. Thor sends it; the wire doubles it."""
frame = build_request(0x5A, 0x0400, bytes.fromhex("00001000000000000000"))
assert bytes([DLE, DLE]) in frame
assert frame == REQ_DOWNLOAD_CHUNK4
def test_builder_rejects_malformed_arguments():
with pytest.raises(ValueError):
build_request(0x5B, 0, bytes(9))
with pytest.raises(ValueError):
build_request(0x5B, 0x10000)
with pytest.raises(ValueError):
build_request(0x100)
# ── Parsing ───────────────────────────────────────────────────────────────────
def test_poll_probe_reply_fields():
(f,) = MicromateFrameParser().feed(RSP_POLL_PROBE)
assert f.sub == 0xA4
assert f.request_sub == 0x5B
assert f.flags == FLAGS_BLASTWARE
assert f.firmware_line == "blastware"
assert f.checksum_valid
assert f.probe_length == 0x0030
def test_probe_length_is_a_uint16_not_a_byte():
"""⚠ Read as data[3] alone, SUB 0x1A's 0x082C (2092) reads as 44 -- 47x low.
SYNTHESISED: no probe response survives in the captures on disk (the
9-24-26 session uses single-step reads at offset 0xFFFF throughout, so it
contains no probes at all). The field position is taken from the captured
POLL probe reply above, which does exercise it for real.
"""
payload = bytes([0x00, FLAGS_BLASTWARE, 0xE5, 0x00, 0x00]) + bytes(
[0x00, 0x00, 0x00, 0x08, 0x2C]
)
synth = bytes([STX]) + stuff(payload + bytes([checksum(payload)])) + bytes([ETX])
(f,) = MicromateFrameParser().feed(synth)
assert f.probe_length == 0x082C == 2092
assert f.data[3] == 0x08, "the high byte is where a byte-wide read loses 2048"
def test_escaped_bytes_land_in_the_right_field():
"""RSP_FILE_READ's first data byte is 0x04, which arrives as `10 04`.
Without destuffing it reads as 0x10 and every field after it is one byte
late -- the failure mode that put `SUB 0x02` in the log as `SUB_10` for an
afternoon.
"""
(f,) = MicromateFrameParser().feed(RSP_FILE_READ)
assert f.sub == 0xB7
assert f.request_sub == 0x48
assert (f.page_hi, f.page_lo) == (0x00, 0x00)
assert f.data[0] == 0x04
assert len(f.data) == 15, "one byte shorter than the wire suggests"
assert f.checksum_valid
# The real 11.0BD POLL data section, UM20147 over USB 2026-09-30. This
# replaces a synthesised frame -- the Thor firmware line was inference-only
# until this capture.
_POLL_BD_DATA = bytes.fromhex(
"30000000000000000000000000005649"
"6e7374616e74656c000600c3f04a00e4"
"194f0052024d4d2f495345452f530058"
"f406001f60c0755760c075"
)
def test_thor_firmware_line_survives_destuffing():
"""⚠ flags = 0x03 is ETX, so it arrives as `10 03`.
A parser that does not destuff ends the frame at byte 2 on half the fleet.
The data section is REAL (UM20147, 11.0BD); the frame wrapper is rebuilt,
which is sound because the framing is verified 251/251 elsewhere in this
file. scratch/mm_frame_parse.py reported payload=64 for this frame, and
the assertion below pins that, so the reconstruction is checked rather than
assumed.
"""
payload = bytes([0x00, FLAGS_THOR, 0xA4, 0x00, 0x00]) + _POLL_BD_DATA
assert len(payload) == 64, "the parser reported payload=64 for this frame"
wire = bytes([STX]) + stuff(payload + bytes([checksum(payload)])) + bytes([ETX])
assert bytes([DLE, ETX]) in wire, "flags 0x03 must be escaped on the wire"
(f,) = MicromateFrameParser().feed(wire)
assert f.flags == FLAGS_THOR
assert f.firmware_line == "thor"
assert f.sub == 0xA4
assert f.request_sub == 0x5B
assert f.checksum_valid
assert b"MM/ISEE/S\x00" in f.data, "the shorter BD model string"
def test_an_escaped_checksum_byte_is_read_correctly():
"""SYNTHESISED, but the behaviour is real: three captured responses have a
checksum of 0x02/0x03/0x04 and all three escape it on the wire. The
shortest is 1,070 B (an 0x5A chunk in
raw_s3_20260925_011403_Download_events_then_delete_1_event.bin, chk = 0x03),
too long to embed for one byte's worth of assertion.
"""
payload = bytes([0x00, FLAGS_BLASTWARE, 0xA4, 0x00, 0x00, 0x03])
assert checksum(payload) == 0x03 + FLAGS_BLASTWARE + 0xA4 & 0xFF
body = payload + bytes([checksum(payload)])
synth = bytes([STX]) + stuff(body) + bytes([ETX])
(f,) = MicromateFrameParser().feed(synth)
assert f.chk_byte == checksum(payload)
assert f.checksum_valid
def test_a_corrupted_checksum_still_yields_a_frame():
"""Flag it, do not swallow it -- a dropped frame looks like a dead unit."""
broken = bytearray(RSP_POLL_PROBE)
broken[-2] ^= 0xFF
(f,) = MicromateFrameParser().feed(bytes(broken))
assert f.sub == 0xA4
assert not f.checksum_valid
def test_a_truncated_frame_yields_nothing():
parser = MicromateFrameParser()
assert parser.feed(RSP_POLL_PROBE[:-1]) == []
assert parser.frames == []
assert parser.bytes_fed == len(RSP_POLL_PROBE) - 1
def test_a_frame_too_short_to_hold_a_header_is_rejected():
assert MicromateFrameParser().feed(bytes([STX, 0x00, 0xC5, 0xA4, ETX])) == []
def test_request_frames_are_not_mistaken_for_responses():
"""Feeding a bidirectional capture must yield only the unit's side."""
parser = MicromateFrameParser()
frames = parser.feed(REQ_POLL + RSP_POLL_PROBE + REQ_SERIAL)
assert len(frames) == 1
assert frames[0].request_sub == 0x5B
def test_leading_noise_is_discarded():
"""Cold-boot banners and the RV55's RING/CONNECT chatter precede frames."""
noise = b"\r\nRING\r\n\r\nCONNECT\r\n" + bytes([ACK])
(f,) = MicromateFrameParser().feed(noise + RSP_POLL_PROBE)
assert f.sub == 0xA4
assert f.checksum_valid
def test_frames_split_across_feeds_reassemble():
"""The transport hands over whatever the socket returned, DLE pairs and all."""
whole = RSP_STATE + RSP_CHUNK_WITH_DLE
for cut in (1, 2, 5, 17, 24, 25, 40, len(RSP_STATE), len(whole) - 1):
parser = MicromateFrameParser()
got = parser.feed(whole[:cut]) + parser.feed(whole[cut:])
assert len(got) == 2, f"split at {cut} lost a frame"
assert all(f.checksum_valid for f in got), f"split at {cut} broke a checksum"
def test_reset_clears_partial_state():
parser = MicromateFrameParser()
parser.feed(RSP_POLL_PROBE[:6])
parser.reset()
assert parser.bytes_fed == 0
(f,) = parser.feed(RSP_POLL_PROBE)
assert f.checksum_valid
def test_firmware_line_of_an_unknown_flags_byte():
f = MicromateFrame(sub=0xA4, flags=0x99, page_hi=0, page_lo=0,
data=b"", checksum_valid=True)
assert f.firmware_line == "unknown"
assert f.probe_length is None
# ── Whole-session round trip ──────────────────────────────────────────────────
_CAPTURES = (
Path(__file__).resolve().parents[1]
/ "bridges" / "captures" / "9-24-26 - micromate2"
)
@pytest.mark.skipif(
not _CAPTURES.is_dir(),
reason="capture directory is gitignored; present only on a dev box",
)
def test_whole_captured_sessions_parse_with_no_bad_checksums():
"""Belt-and-braces against the real bytes when they happen to be here.
scratch/mm_frame_parse.py reported zero bad checksums on these sessions
only because it accepts a frame matching *either* checksum rule. This
asserts the single rule holds across all of them.
"""
total = 0
for path in sorted(_CAPTURES.rglob("raw_s3_*.bin")):
parser = MicromateFrameParser()
frames = parser.feed(path.read_bytes())
assert frames, f"{path.name}: no frames parsed"
bad = [f for f in frames if not f.checksum_valid]
assert not bad, f"{path.name}: {len(bad)} bad checksums"
total += len(frames)
assert total == 251, f"expected 251 response frames across the corpus, got {total}"
@pytest.mark.skipif(
not _CAPTURES.is_dir(),
reason="capture directory is gitignored; present only on a dev box",
)
def test_builder_reproduces_every_captured_read_frame():
"""218/218. This is the test that would have caught the escape-set error."""
checked = 0
for path in sorted(_CAPTURES.rglob("raw_bw_*.bin")):
blob = path.read_bytes()
i = 0
while i < len(blob):
if not (blob[i] == ACK and i + 1 < len(blob) and blob[i + 1] == STX):
i += 1
continue
j = i + 2
body = bytearray()
while j < len(blob):
if blob[j] == DLE and j + 1 < len(blob):
body.append(blob[j + 1])
j += 2
continue
if blob[j] == ETX:
break
body.append(blob[j])
j += 1
payload = bytes(body[:-1])
if len(payload) == 16: # a read frame; writes carry a data section
sub = payload[2]
offset = (payload[4] << 8) | payload[5]
assert build_request(sub, offset, payload[6:16]) == blob[i:j + 1], (
f"{path.name} @0x{i:04x} SUB=0x{sub:02x} offset=0x{offset:04x}"
)
checked += 1
i = j + 1
assert checked == 218, f"expected 218 read frames, checked {checked}"
-402
View File
@@ -1,402 +0,0 @@
"""Protocol-layer tests for the Micromate (series-4) live client.
The load-bearing assertion in here is not "our parser understands the device" —
it is **"the bytes we put on the wire are the bytes THOR puts on the wire."**
Every request constant below is lifted from
``bridges/captures/9-24-26 - micromate2/`` (UM12947, firmware 11.0CB), so a
passing test means a real unit has already answered exactly that frame.
Responses are replayed through a scripted transport. No hardware, no network.
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from micromate import protocol as P
from micromate.framing import ETX, STX, checksum, stuff
from micromate.protocol import (
ACK_DATA_LEN,
ChecksumError,
MicromateProtocol,
ShortRead,
UnexpectedResponse,
chunk_params,
key_lo_params,
key_params,
token_params,
)
FLAGS_CB = 0xC5
# ── Test doubles ──────────────────────────────────────────────────────────────
class ScriptedTransport:
"""Hands back queued responses; records every byte written."""
def __init__(self, responses: list[bytes] | None = None) -> None:
self.queue = list(responses or [])
self.written: list[bytes] = []
self._connected = True
# BaseTransport surface actually used by MicromateProtocol
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, page: int = 0) -> bytes:
"""Build a response frame the way a unit would."""
payload = bytes([0x00, flags, rsp_sub, (page >> 8) & 0xFF, page & 0xFF]) + data
return bytes([STX]) + stuff(payload + bytes([checksum(payload)])) + bytes([ETX])
def ack(rsp_sub: int) -> bytes:
return frame(rsp_sub, bytes(ACK_DATA_LEN))
def proto(responses: list[bytes], **kw) -> tuple[MicromateProtocol, ScriptedTransport]:
t = ScriptedTransport(responses)
return MicromateProtocol(t, recv_timeout=0.5, **kw), t
# ── Captured THOR request frames ──────────────────────────────────────────────
REQ = {
"poll": bytes.fromhex("41021010005b000030000000000000000000009b03"),
"serial": bytes.fromhex("41021010001500000a000000000000000000002f03"),
"state": bytes.fromhex("41021010004900ffff000000000000000000005703"),
"compliance": bytes.fromhex("41021010001a00ffff000000000000000000002803"),
"arm": bytes.fromhex("41021010009300ffff00000000000000000000a103"),
"setup_first": bytes.fromhex("41021010003f00ffff000000000000000000004d03"),
"setup_next": bytes.fromhex("41021010004000ffff000000000000000000004e03"),
}
# The complete 0x5A sequence for event 055d4a81 (4,076 bytes → 4 chunks), as
# THOR sent it. Chunk 1's params hold a literal 0x04 and chunk 3's offset is
# the exact remainder.
REQ_CHUNKS_4A81 = [
bytes.fromhex("41021010005a00100400055d4a810000000000009b03"),
bytes.fromhex("41021010005a0010040000001004000000000000007203"),
bytes.fromhex("41021010005a00100400000008000000000000007603"),
bytes.fromhex("41021010005a001003ec00000c000000000000006503"),
]
SIZE_4A81 = 4076
# ── Params builders ───────────────────────────────────────────────────────────
def test_event_token_sits_at_params_7():
"""⚠ THOR sends 0xFE here; the protocol reference documents all-zero params.
That reference entry describes our own browse probing, not THOR's.
"""
assert token_params() == bytes.fromhex("00000000000000fe0000")
def test_event_record_takes_the_full_key_at_params_4():
assert key_params(bytes.fromhex("055d4a81")) == bytes.fromhex("00000000055d4a810000")
def test_monitor_log_takes_only_the_low_half_of_the_key():
"""⚠ Inferred from one key value -- see key_lo_params' docstring."""
assert key_lo_params(bytes.fromhex("055d4a81")) == bytes.fromhex("0000000000004a810000")
def test_chunk_params_switch_from_key_to_byte_offset():
key = bytes.fromhex("055d4a81")
assert chunk_params(key, 0) == bytes.fromhex("055d4a81000000000000")
assert chunk_params(key, 1024) == bytes.fromhex("00000400000000000000")
assert chunk_params(key, 13312) == bytes.fromhex("00003400000000000000")
@pytest.mark.parametrize("bad", [b"", b"\x01\x02\x03", b"\x01\x02\x03\x04\x05"])
def test_params_builders_reject_a_wrong_length_key(bad):
for fn in (key_params, key_lo_params):
with pytest.raises(ValueError):
fn(bad)
# ── Each read emits the frame THOR emits ──────────────────────────────────────
@pytest.mark.parametrize(
"name, method, rsp_sub, data_len",
[
("poll", "poll", 0xA4, 59),
("serial", "read_serial", 0xEA, 21),
("state", "read_state", 0xB6, 16),
("compliance", "read_compliance_config", 0xE5, 2103),
("setup_first", "read_first_setup", 0xC0, 266),
("setup_next", "read_next_setup", 0xBF, 266),
],
)
def test_reads_match_thors_wire_bytes(name, method, rsp_sub, data_len):
p, t = proto([frame(rsp_sub, bytes(data_len))])
getattr(p, method)()
assert t.written == [REQ[name]]
def test_arm_event_matches_thors_wire_bytes():
p, t = proto([ack(0x6C)])
p.arm_event()
assert t.written == [REQ["arm"]]
def test_poll_is_the_only_read_with_a_non_ffff_offset_besides_serial():
"""Reads are single-step at 0xFFFF; POLL and SERIAL are the exceptions."""
assert set(P._OFFSETS) == {P.SUB_POLL, P.SUB_SERIAL}
assert P._OFFSETS[P.SUB_POLL] == 0x0030
assert P._OFFSETS[P.SUB_SERIAL] == 0x000A
# ── The chunk walk ────────────────────────────────────────────────────────────
def test_download_reproduces_thors_chunk_sequence_byte_for_byte():
"""The whole point of step 2. Four chunks, 4,076 bytes, THOR's exact frames."""
payload = bytes(range(256)) * 16 # 4096 B, we use the first 4076
payload = payload[:SIZE_4A81]
responses = []
for i in range(4):
want = min(P.CHUNK_SIZE, SIZE_4A81 - i * P.CHUNK_SIZE)
body = payload[i * P.CHUNK_SIZE: i * P.CHUNK_SIZE + want]
responses.append(frame(0xA5, bytes(11) + body, page=want // 256))
p, t = proto(responses)
got = p.read_event_file(bytes.fromhex("055d4a81"), SIZE_4A81)
assert t.written == REQ_CHUNKS_4A81
assert got == payload
assert len(got) == SIZE_4A81
@pytest.mark.parametrize(
"size, n_chunks, last_offset",
[
(4076, 4, 0x03EC), (11032, 11, 0x0318), (11502, 12, 0x00EE),
(13424, 14, 0x0070), (8746, 9, 0x022A), (6092, 6, 0x03CC),
(1024, 1, 0x0400), (1, 1, 0x0001), (1025, 2, 0x0001),
],
)
def test_chunk_count_and_final_offset(size, n_chunks, last_offset):
"""The first six rows are the six bench events, with THOR's real offsets."""
responses = []
for i in range(n_chunks):
want = min(P.CHUNK_SIZE, size - i * P.CHUNK_SIZE)
responses.append(frame(0xA5, bytes(11) + bytes(want)))
p, t = proto(responses)
p.read_event_file(bytes.fromhex("055d4a81"), size)
assert len(t.written) == n_chunks
# offset is payload[4:5] of the request; recover it from the built frame
final = t.written[-1]
assert final[7:9] in (
bytes([last_offset >> 8, last_offset & 0xFF]),
# a 0x02/0x03/0x04/0x10 high byte arrives escaped, shifting the pair
bytes([0x10, last_offset >> 8]),
)
def test_a_short_chunk_raises_rather_than_truncating():
"""A silently short event is the failure mode this codebase keeps hitting."""
p, _ = proto([frame(0xA5, bytes(11) + bytes(900))]) # asked for 1024
with pytest.raises(ShortRead, match="asked for 1024 B, got 900"):
p.read_event_file(bytes.fromhex("055d4a81"), 1024)
def test_a_chunk_too_short_to_hold_its_header_raises():
p, _ = proto([frame(0xA5, bytes(4))])
with pytest.raises(ShortRead, match="too short to hold a chunk header"):
p.read_event_file(bytes.fromhex("055d4a81"), 1024)
def test_download_rejects_a_nonsense_size():
p, _ = proto([])
with pytest.raises(ValueError):
p.read_event_file(bytes.fromhex("055d4a81"), 0)
# ── The monitor-log walk ──────────────────────────────────────────────────────
def test_monitor_log_walk_ends_on_a_short_response():
"""⚠ Not a keyed read -- the same request repeated, device-side cursor.
Eight records then an 11-byte ack, which is what the capture shows.
"""
key = bytes.fromhex("055d4a81")
responses = [frame(0xF5, bytes(297)) for _ in range(8)] + [ack(0xF5)]
p, t = proto(responses)
records = []
while (rec := p.read_monitor_log_next(key)) is not None:
records.append(rec)
assert len(records) == 8
assert len(t.written) == 9
assert len(set(t.written)) == 1, "every request in the walk is identical"
# ── Error handling ────────────────────────────────────────────────────────────
def test_a_bad_checksum_raises_by_default():
"""⚠ Deliberately stricter than the Series III sibling.
That one logs and continues because its parser cannot always tell an
inner-frame delimiter from a checksum byte. The Micromate rule is exact on
251/251 captured frames, so a mismatch here means something real.
"""
bad = bytearray(frame(0xA4, bytes(59)))
bad[-2] ^= 0xFF
p, _ = proto([bytes(bad)])
with pytest.raises(ChecksumError, match="checksum mismatch"):
p.poll()
def test_a_bad_checksum_can_be_downgraded_for_field_diagnosis():
bad = bytearray(frame(0xA4, bytes(59)))
bad[-2] ^= 0xFF
p, _ = proto([bytes(bad)], strict_checksums=False)
assert p.poll().sub == 0xA4
def test_the_wrong_response_sub_raises():
p, _ = proto([frame(0xE0, bytes(19))]) # 0xE0 answers 0x1F, not 0x1E
with pytest.raises(UnexpectedResponse, match="expected SUB 0xE1"):
p.read_event_first()
def test_a_timeout_reports_how_many_bytes_arrived():
"""Separates "nothing came back" from "bytes arrived but never framed".
Those have completely different causes -- and on a Micromate the second one
is the signature of a modem forwarding a session it should not be.
"""
p, _ = proto([])
with pytest.raises(P.TimeoutError, match="0 bytes were received"):
p.poll()
unframed = b"\x02\x00\xc5\xa4garbage-no-terminator"
p2, _ = proto([unframed])
with pytest.raises(P.TimeoutError, match=f"{len(unframed)} bytes were received"):
p2.poll()
def test_a_leftover_frame_is_discarded_rather_than_answered_with():
"""If a read returns two frames, the extra must not answer the NEXT request.
Every exchange resets the parser before sending, so anything already
buffered is treated as stale. Delivering it would be the worse failure:
`expected_sub` happens to catch a mismatched SUB, but a same-SUB leftover
would sail through and return data for the wrong key.
"""
# Both frames arrive while answering arm_event(); the 0xE1 is left over.
p, _ = proto([ack(0x6C) + frame(0xE1, b"\xaa" * 19)])
p.arm_event()
# The next request gets no bytes of its own, so it must time out rather
# than hand back the stale 0xE1.
with pytest.raises(P.TimeoutError):
p.read_event_first()
# ── Against the real capture, when it happens to be present ───────────────────
_CAPTURES = (
Path(__file__).resolve().parents[1]
/ "bridges" / "captures" / "9-24-26 - micromate2"
)
_DOWNLOAD = "raw_bw_20260925_011403_Download_events_then_delete_1_event.bin"
@pytest.mark.skipif(
not (_CAPTURES / _DOWNLOAD).is_file(),
reason="capture is gitignored; present only on a dev box",
)
def test_every_captured_download_frame_is_one_we_would_have_sent():
"""Replay the real session: for each event, assert our chunk walk emits
exactly the frames THOR emitted -- all 50-odd of them, six events."""
from micromate.framing import ACK, DLE
def destuffed_frames(blob: bytes, is_req: bool):
i, n = 0, len(blob)
while i < n:
if is_req:
if not (blob[i] == ACK and i + 1 < n and blob[i + 1] == STX):
i += 1
continue
j = i + 2
else:
if blob[i] != STX:
i += 1
continue
j = i + 1
out = bytearray()
while j < n:
if blob[j] == DLE and j + 1 < n:
out.append(blob[j + 1])
j += 2
continue
if blob[j] == ETX:
break
out.append(blob[j])
j += 1
if len(out) >= 6:
yield blob[i:j + 1], bytes(out[:-1])
i = j + 1
bw = list(destuffed_frames((_CAPTURES / _DOWNLOAD).read_bytes(), True))
s3 = list(
destuffed_frames(
(_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False
)
)
# Group THOR's 0x5A frames per event, taking each event's key+size from the
# 1E/1F that preceded them.
events, cur = [], None
for (wire, req), (_, rsp) in zip(bw, s3):
sub, data = req[2], rsp[5:]
if sub in (0x1E, 0x1F) and len(data) >= 19:
cur = {"key": data[11:15], "size": int.from_bytes(data[15:19], "big"),
"reqs": [], "rsps": []}
if cur["size"]:
events.append(cur)
elif sub == 0x5A and cur is not None:
cur["reqs"].append(wire)
cur["rsps"].append(rsp)
# The capture walks the chain twice (it deletes an event on the second
# pass), so some 1E/1F hits carry a size but no download behind them.
events = [e for e in events if e["reqs"]]
assert len(events) == 6, f"expected 6 downloaded events, found {len(events)}"
total = 0
for e in events:
p, t = proto([bytes([STX]) + stuff(r + bytes([checksum(r)])) + bytes([ETX])
for r in e["rsps"]])
got = p.read_event_file(e["key"], e["size"])
assert t.written == e["reqs"], (
f"event {e['key'].hex()}: our {len(t.written)} frames differ from "
f"THOR's {len(e['reqs'])}"
)
assert len(got) == e["size"]
total += len(e["reqs"])
assert total == 56, f"expected 56 download frames across the 6 events, saw {total}"