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

v0.10.0 — monitor log entry support (SUB 0x0A partial records)

Browse Source 
Add full decode pipeline for 0x2C partial records from the device's event
list, representing continuous monitoring intervals where no threshold was
crossed.  These records appear interleaved with full triggered events in the
browse walk and were previously ignored.

minimateplus/models.py
- Add MonitorLogEntry dataclass: key, start_time, stop_time, serial,
  geo_threshold_ips, raw_header, duration_seconds property

minimateplus/protocol.py
- read_waveform_header() now returns (data_rsp.data, length) — full payload
  including the record-type byte at position 0 — instead of the sliced header.
  Callers that need the old slice use raw_data[11:11+length] as before.

minimateplus/client.py
- Add _decode_0a_partial_header(): auto-detects 9-byte (sub_code=0x10) vs
  10-byte (sub_code=0x03) timestamp format, handles 1-byte inter-timestamp
  gap, extracts serial via BE anchor and geo threshold via Geo: anchor.
- Add get_monitor_log_entries(skip_keys=None): browse walk (1E → 0A → 1F),
  decodes partial records, skips full records and already-seen keys.

minimateplus/__init__.py
- Export MonitorLogEntry

bridges/ach_server.py
- After get_events(), call get_monitor_log_entries(skip_keys=seen_keys) and
  save new entries to monitor_log.json in the session directory.
- Add _monitor_log_entry_to_dict() helper.
- Include monitor log keys in downloaded_keys for state persistence.

CLAUDE.md / CHANGELOG.md
- Document 0x2C partial record layout (timestamp format, ASCII metadata
  region, 1-byte gap edge case) confirmed from 4-11-26 MITM capture.
- Version bump to v0.10.0; update What's next.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Brian Harrison
2026-04-11 02:59:40 -04:00
committed by serversdown
parent 6acb419ebd
commit b241da970d
7 changed files with 525 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
minimateplus/client.py
+209
View File
@@ -28,6 +28,7 @@ Example (TCP / modem):
from __future__ import annotations
import datetime
import logging
import struct
from typing import Optional
@@ -37,6 +38,7 @@ from .models import (
ComplianceConfig,
DeviceInfo,
Event,
MonitorLogEntry,
MonitorStatus,
PeakValues,
ProjectInfo,
@@ -300,6 +302,96 @@ class MiniMateClient:
log.info("list_event_keys: %d key(s): %s", len(keys), keys)
return keys
def get_monitor_log_entries(
self,
skip_keys: Optional[set] = None,
) -> list[MonitorLogEntry]:
"""
Collect all monitor log entries (partial records, type 0x2C) from the
device using the browse-mode 1E → 0A → 1F walk.
This is the fast path for monitor log data. No 0C or 5A commands are
issued — all available monitor log information is in the 0x0A response
header alone.
Full triggered events (0x0A response type 0x46) are silently skipped.
Only partial records (type 0x2C) are returned as MonitorLogEntry objects.
Confirmed from 4-11-26 MITM capture: Blastware's ACH mode performs a
full browse walk (Phase 3: 0x0A + 1F × all records) AFTER the triggered-
event download phase. The partial records encountered in this walk are
the monitor log entries.
Args:
skip_keys: optional set of 8-hex key strings to skip (already seen).
Keys in this set still advance the walk (0A + 1F) but are
not decoded or returned.
Returns:
List of MonitorLogEntry objects in device storage order.
Raises:
ProtocolError: on unrecoverable communication failure.
"""
proto = self._require_proto()
try:
key4, data8 = proto.read_event_first()
except ProtocolError as exc:
log.warning("get_monitor_log_entries: 1E failed: %s -- returning []", exc)
return []
if data8[4:8] == b"\x00\x00\x00\x00":
log.info("get_monitor_log_entries: device is empty")
return []
entries: list[MonitorLogEntry] = []
idx = 0
while data8[4:8] != b"\x00\x00\x00\x00":
cur_key = key4
key_hex = cur_key.hex()
try:
raw_data, rec_len = proto.read_waveform_header(cur_key)
except ProtocolError as exc:
log.warning(
"get_monitor_log_entries: 0A failed for key=%s: %s -- stopping",
key_hex, exc,
)
break
# Only decode partial records (0x2C); full records (0x46) are silently skipped.
if rec_len < 0x40 and raw_data and (not skip_keys or key_hex not in skip_keys):
entry = _decode_0a_partial_header(raw_data, idx, cur_key)
if entry is not None:
entries.append(entry)
log.debug(
"get_monitor_log_entries: [%d] key=%s %s → %s",
idx, key_hex, entry.start_time, entry.stop_time,
)
else:
log.debug(
"get_monitor_log_entries: [%d] key=%s type=0x%02X %s",
idx, key_hex, rec_len,
"skip (already seen)" if skip_keys and key_hex in skip_keys else "skip (full record)",
)
try:
key4, data8 = proto.advance_event(browse=True)
except ProtocolError as exc:
log.warning(
"get_monitor_log_entries: 1F failed after %d record(s): %s -- stopping",
idx, exc,
)
break
idx += 1
log.info(
"get_monitor_log_entries: walked %d record(s), found %d monitor log entry(s)",
idx, len(entries),
)
return entries
def delete_all_events(self) -> None:
"""
Erase all stored events from the device memory.
@@ -1856,6 +1948,123 @@ def _find_first_string(data: bytes, start: int, end: int, min_len: int) -> Optio
def _decode_0a_partial_header(raw_data: bytes, index: int, key4: bytes) -> Optional[MonitorLogEntry]:
"""
Decode a SUB 0x0A response for a partial (monitor log) record into a
MonitorLogEntry.
Called when read_waveform_header() returns rec_len < 0x40 (i.e. 0x2C = 44).
raw_data is the complete data_rsp.data from the protocol layer.
Layout of raw_data:
[0] = 0x2C (partial record type)
[1:5] = 0x00 × 4
[5:9] = event key (big-endian)
[9:11] = 0x00 × 2
[11:] = timestamp_start + timestamp_stop + sep + serial + geo_string
Timestamp format detection (auto):
raw_data[11] == 0x10 → 10-byte sub_code=0x03 continuous format
raw_data[12] == 0x10 → 9-byte sub_code=0x10 single-shot format
Both timestamps use the same format (detected from the first byte).
A 1-byte gap can appear between ts1 and ts2 for certain timestamps
(observed empirically when both timestamps share the same minute:second).
The parser handles this by trying ts2 immediately after ts1, then with
a 1-byte skip if that fails.
Returns:
MonitorLogEntry if decoding succeeds, None on error.
"""
if len(raw_data) < 20 or raw_data[0] != 0x2C:
return None
key_hex = key4.hex()
def try_ts9(b: bytes):
"""9-byte sub_code=0x10 format. Returns datetime or None."""
if len(b) < 9 or b[1] != 0x10:
return None
day = b[0]; month = b[2]; year = (b[3] << 8) | b[4]
hr = b[6]; mn = b[7]; sec = b[8]
if not (1 <= day <= 31 and 1 <= month <= 12 and 2000 <= year <= 2050
and hr <= 23 and mn <= 59 and sec <= 59):
return None
try:
return datetime.datetime(year, month, day, hr, mn, sec)
except ValueError:
return None
def try_ts10(b: bytes):
"""10-byte sub_code=0x03 format. Returns datetime or None."""
if len(b) < 10 or b[0] != 0x10 or b[2] != 0x10:
return None
day = b[1]; month = b[3]; year = (b[4] << 8) | b[5]
hr = b[7]; mn = b[8]; sec = b[9]
if not (1 <= day <= 31 and 1 <= month <= 12 and 2000 <= year <= 2050
and hr <= 23 and mn <= 59 and sec <= 59):
return None
try:
return datetime.datetime(year, month, day, hr, mn, sec)
except ValueError:
return None
ts_offset = 11
if len(raw_data) <= ts_offset:
return MonitorLogEntry(index=index, key=key_hex, raw_header=raw_data)
# Detect timestamp format.
if raw_data[ts_offset] == 0x10:
ts_size = 10
try_ts = try_ts10
else:
ts_size = 9
try_ts = try_ts9
# Parse ts1.
ts1 = try_ts(raw_data[ts_offset:ts_offset + ts_size])
ts1_end = ts_offset + ts_size
# Parse ts2 immediately after ts1, then with 1-byte skip if needed.
ts2 = try_ts(raw_data[ts1_end:ts1_end + ts_size])
if ts2 is None:
ts2 = try_ts(raw_data[ts1_end + 1:ts1_end + 1 + ts_size])
# Extract serial and geo threshold from "BE11529\0" and "Geo: X.XXX in/s\0".
serial: Optional[str] = None
geo_ips: Optional[float] = None
serial_pos = raw_data.find(b"BE")
if serial_pos >= 0:
# Read null-terminated serial starting at serial_pos.
null_pos = raw_data.find(b"\x00", serial_pos)
if null_pos > serial_pos:
serial = raw_data[serial_pos:null_pos].decode("ascii", errors="replace")
# Geo string follows the null byte.
geo_start = (null_pos + 1) if null_pos > serial_pos else serial_pos + 7
geo_bytes = raw_data[geo_start:]
# "Geo: X.XXX in/s\0" — extract float after "Geo: ".
geo_str_pos = geo_bytes.find(b"Geo: ")
if geo_str_pos >= 0:
geo_val_bytes = geo_bytes[geo_str_pos + 5:] # after "Geo: "
geo_val_end = geo_val_bytes.find(b" ") # before " in/s"
if geo_val_end > 0:
try:
geo_ips = float(geo_val_bytes[:geo_val_end].decode("ascii"))
except ValueError:
pass
return MonitorLogEntry(
index=index,
key=key_hex,
start_time=ts1,
stop_time=ts2,
serial=serial,
geo_threshold_ips=geo_ips,
raw_header=raw_data,
)
def _decode_monitor_status(data: bytes) -> MonitorStatus:
"""
Decode SUB 0x1C response payload into a MonitorStatus object.