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

feat(call-home): Implement Auto Call Home configuration management

Browse Source 
- Added `CallHomeConfig` model to represent the Auto Call Home settings.
- Introduced methods in `MiniMateClient` for reading (`get_call_home_config`) and writing (`set_call_home_config`) the call home configuration.
- Updated `MiniMateProtocol` with new commands for call home operations (SUB 0x2C for read, SUB 0x7E for write, and SUB 0x7F for confirm).
- Created API endpoints for retrieving and updating call home settings in the server.
- Enhanced the web interface with a new "Call Home" tab for user interaction with call home settings.
- Implemented JavaScript functions for reading and writing call home configurations from the web app.
This commit is contained in:
Brian Harrison
2026-04-20 18:23:48 -04:00
parent 7bdd7c92f2
commit 3fb24e1895
8 changed files with 1081 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
minimateplus/client.py
+245
View File
@@ -35,6 +35,7 @@ from typing import Optional
from .framing import S3Frame
from .models import (
CallHomeConfig,
ComplianceConfig,
DeviceInfo,
Event,
@@ -956,6 +957,93 @@ class MiniMateClient:
notes=notes,
)
# ── Call home config ──────────────────────────────────────────────────────
def get_call_home_config(self) -> CallHomeConfig:
"""
Read the auto call home (ACH) configuration from the device.
Sends SUB 0x2C (two-step read) and decodes the raw 125-byte payload
into a CallHomeConfig object.
Returns:
CallHomeConfig with all confirmed fields populated.
Raises:
RuntimeError: if not connected.
ProtocolError: on timeout or wrong response SUB.
"""
proto = self._require_proto()
raw = proto.read_call_home_config()
return _decode_call_home_config(raw)
def set_call_home_config(
self,
*,
auto_call_home_enabled: Optional[bool] = None,
after_event_recorded: Optional[bool] = None,
at_specified_times: Optional[bool] = None,
time1_enabled: Optional[bool] = None,
time1_hour: Optional[int] = None,
time1_min: Optional[int] = None,
time2_enabled: Optional[bool] = None,
time2_hour: Optional[int] = None,
time2_min: Optional[int] = None,
) -> None:
"""
Read the current call home config, apply any supplied changes, and
write the updated config back to the device.
Only non-None arguments are modified. All other bytes are round-tripped
verbatim from the device.
Configurable fields
-------------------
auto_call_home_enabled : bool — master enable for ACH
after_event_recorded : bool — call home after each triggered event
at_specified_times : bool — call home at scheduled times
time1_enabled : bool — enable time slot 1
time1_hour : int — hour for time slot 1 (0-23)
time1_min : int — minute for time slot 1 (0-59)
time2_enabled : bool — enable time slot 2
time2_hour : int — hour for time slot 2 (0-23)
time2_min : int — minute for time slot 2 (0-59)
Write sequence (confirmed from 4-20-26 call home settings captures):
SUB 0x2C (read, 2-step) → 125-byte raw payload
patch fields in-place
SUB 0x7E (write, 127-byte payload) → ack 0x81
SUB 0x7F (confirm) → ack 0x80
Raises:
RuntimeError: if not connected.
ProtocolError: if any read or write step fails.
"""
proto = self._require_proto()
# 1. Read current config
log.info("set_call_home_config: reading current config (SUB 0x2C)")
raw = proto.read_call_home_config()
# 2. Patch fields and build write payload
write_data = _encode_call_home_config(
raw,
auto_call_home_enabled=auto_call_home_enabled,
after_event_recorded=after_event_recorded,
at_specified_times=at_specified_times,
time1_enabled=time1_enabled,
time1_hour=time1_hour,
time1_min=time1_min,
time2_enabled=time2_enabled,
time2_hour=time2_hour,
time2_min=time2_min,
)
# 3. Write back
log.info("set_call_home_config: writing updated config (SUB 0x7E + 0x7F)")
proto.write_call_home_config(write_data)
log.info("set_call_home_config: complete")
def poll(self) -> None:
"""
Perform just the POLL startup handshake — no config reads.
@@ -2232,3 +2320,160 @@ def _decode_monitor_status(data: bytes) -> MonitorStatus:
memory_total=memory_total,
memory_free=memory_free,
)
def _decode_call_home_config(raw: bytes) -> CallHomeConfig:
"""
Decode the raw 125-byte call home config payload into a CallHomeConfig.
*raw* is data[11:] from the SUB 0xD3 data response frame.
Field offsets (confirmed from 4-20-26 captures, all 11 BW+S3 pairs):
[5] auto_call_home_enabled (0x00=off, 0x01=on)
[6:46] dial_string 40-byte null-padded ASCII
[87] after_event_recorded (0x01=on, 0x00=off)
[91] at_specified_times (0x01=on, 0x00=off)
[93] time1_enabled (0x01=on, 0x00=off)
[95] time2_enabled (0x01=on, 0x00=off)
[101] time1_hour uint8 decimal 0-23
[102] time1_min uint8 decimal 0-59
[105] time2_hour uint8 decimal 0-23
[106] time2_min uint8 decimal 0-59
[117:119] 10 03 = DLE-escaped num_retries=3 (logical value = 0x03)
[120] time_between_retries_sec (shifted +1 from logical by DLE prefix)
[122] wait_for_connection_sec
[124] warm_up_time_sec
The DLE-escaped ETX at raw[117:119] = b'\\x10\\x03' means the logical value
0x03 (3 retries) is stored there. The S3FrameParser keeps both bytes verbatim.
Subsequent fields are at logical_offset + 1 in the raw byte array.
"""
cfg = CallHomeConfig(raw=raw)
if len(raw) < 10:
return cfg
# Simple boolean and string fields — direct reads, no DLE complications
if len(raw) > 5:
cfg.auto_call_home_enabled = bool(raw[5])
if len(raw) >= 46:
ds = raw[6:46]
cfg.dial_string = ds.split(b"\x00", 1)[0].decode("ascii", errors="replace") or None
if len(raw) > 87:
cfg.after_event_recorded = bool(raw[87])
if len(raw) > 91:
cfg.at_specified_times = bool(raw[91])
if len(raw) > 93:
cfg.time1_enabled = bool(raw[93])
if len(raw) > 95:
cfg.time2_enabled = bool(raw[95])
if len(raw) > 102:
cfg.time1_hour = raw[101]
cfg.time1_min = raw[102]
if len(raw) > 106:
cfg.time2_hour = raw[105]
cfg.time2_min = raw[106]
# num_retries: raw[117]=0x10 (DLE prefix), raw[118]=0x03 (value)
# Subsequent fields shift by +1 from logical positions.
if len(raw) > 118 and raw[117] == 0x10:
cfg.num_retries = raw[118] # 0x03 = 3
if len(raw) > 120:
cfg.time_between_retries_sec = raw[120] # logical 119, shifted to 120
if len(raw) > 122:
cfg.wait_for_connection_sec = raw[122] # logical 121, shifted to 122
if len(raw) > 124:
cfg.warm_up_time_sec = raw[124] # logical 123, shifted to 124
elif len(raw) > 117:
# Fallback: no DLE prefix (num_retries is not 0x03)
cfg.num_retries = raw[117]
if len(raw) > 119:
cfg.time_between_retries_sec = raw[119]
if len(raw) > 121:
cfg.wait_for_connection_sec = raw[121]
if len(raw) > 123:
cfg.warm_up_time_sec = raw[123]
log.debug(
"_decode_call_home_config: enabled=%s dial=%r after_event=%s at_times=%s "
"t1=%s %02d:%02d t2=%s %02d:%02d retries=%s gap=%s wait=%s warmup=%s",
cfg.auto_call_home_enabled, cfg.dial_string,
cfg.after_event_recorded, cfg.at_specified_times,
cfg.time1_enabled, cfg.time1_hour or 0, cfg.time1_min or 0,
cfg.time2_enabled, cfg.time2_hour or 0, cfg.time2_min or 0,
cfg.num_retries, cfg.time_between_retries_sec,
cfg.wait_for_connection_sec, cfg.warm_up_time_sec,
)
return cfg
def _encode_call_home_config(
raw: bytes,
*,
auto_call_home_enabled: Optional[bool] = None,
after_event_recorded: Optional[bool] = None,
at_specified_times: Optional[bool] = None,
time1_enabled: Optional[bool] = None,
time1_hour: Optional[int] = None,
time1_min: Optional[int] = None,
time2_enabled: Optional[bool] = None,
time2_hour: Optional[int] = None,
time2_min: Optional[int] = None,
) -> bytes:
"""
Patch specific fields in the 125-byte raw call home payload and return
the 127-byte write payload (raw + b'\\x00\\x00' footer).
Only non-None arguments are modified. All other bytes including the
DLE-escaped \\x10\\x03 at [117:119] are preserved verbatim for round-trip.
The write payload footer (2 trailing zeros) matches Blastware's confirmed
write frame format from the 4-20-26 captures.
CAUTION: hour and minute values must not equal 0x03 (3) — such values would
require DLE-escaping that this encoder does not implement. Values 0x03 in
hour/minute slots are rejected with ValueError.
"""
if len(raw) < 107:
raise ValueError(
f"call home raw payload too short: {len(raw)} bytes (need ≥107)"
)
buf = bytearray(raw) # 125 bytes
def _set_bool(offset: int, value: Optional[bool]) -> None:
if value is not None:
buf[offset] = 0x01 if value else 0x00
def _set_uint8(offset: int, value: Optional[int], name: str) -> None:
if value is None:
return
if value == 0x03:
raise ValueError(
f"{name}={value} (0x03) requires DLE escaping — "
"not supported by this encoder; avoid using 3 for hour/minute fields"
)
buf[offset] = value & 0xFF
_set_bool(5, auto_call_home_enabled)
_set_bool(87, after_event_recorded)
_set_bool(91, at_specified_times)
_set_bool(93, time1_enabled)
_set_bool(95, time2_enabled)
_set_uint8(101, time1_hour, "time1_hour")
_set_uint8(102, time1_min, "time1_min")
_set_uint8(105, time2_hour, "time2_hour")
_set_uint8(106, time2_min, "time2_min")
# num_retries, time_between_retries_sec, wait_for_connection_sec, warm_up_time_sec
# are not writable via this encoder — they're preserved verbatim including the
# DLE-escaped 0x03 at [117:119].
log.debug(
"_encode_call_home_config: patched fields: "
"enabled=%s after_event=%s at_times=%s "
"t1=%s %s:%s t2=%s %s:%s",
auto_call_home_enabled, after_event_recorded, at_specified_times,
time1_enabled, time1_hour, time1_min,
time2_enabled, time2_hour, time2_min,
)
return bytes(buf) + b"\x00\x00" # append 2-byte footer (confirmed BW pattern)
minimateplus/models.py
+72
View File
@@ -378,6 +378,78 @@ class ComplianceConfig:
notes: Optional[str] = None # extended notes / additional info
# ── Call Home Config ──────────────────────────────────────────────────────────
@dataclass
class CallHomeConfig:
"""
Auto Call Home (ACH) configuration from SUB 0x2C (response 0xD3).
Read with a standard two-step protocol (probe offset=0x00, data offset=0x7C).
Written via SUB 0x7E (write, 127-byte payload) + SUB 0x7F (confirm).
Confirmed from 4-20-26 call home settings captures (11 BW + S3 capture pairs).
Raw payload layout (data[11:] from S3 response, 125 bytes):
[0] 0x00 header byte
[1] 0x7C = 124 inner length (= offset for SUB 0x7E write - 2)
[2] 0xDC constant
[3:5] 0x00 0x00 padding
[5] auto_call_home_enabled (0x00=off, 0x01=on) ✅
[6:46] dial_string 40-byte null-padded ASCII ✅
[46:87] auto_answer_raw AT command strings (not decoded) ✅ present
[87] after_event_recorded (0x01=on, 0x00=off) ✅
[91] at_specified_times (0x01=on, 0x00=off) ✅
[93] time1_enabled (0x01=on, 0x00=off) ✅
[95] time2_enabled (0x01=on, 0x00=off) ✅
[101] time1_hour uint8 decimal 0-23 ✅
[102] time1_min uint8 decimal 0-59 ✅
[105] time2_hour uint8 decimal 0-23 ✅
[106] time2_min uint8 decimal 0-59 ✅
[117] DLE prefix (0x10) ┐ DLE-escaped num_retries=3 (0x03)
[118] 0x03 ┘ device stores/returns 0x03 DLE-escaped ✅
[120] time_between_retries_sec uint8 (= 0x0F = 15 s default) ✅
[122] wait_for_connection_sec uint8 (= 0x3C = 60 s default) ✅
[124] warm_up_time_sec uint8 (= 0x3C = 60 s default) ✅
Write payload = raw 125 bytes + b'\\x00\\x00' (2 trailing zeros) = 127 bytes.
Offset for SUB 0x7E: data[1] + 2 = 0x7C + 2 = 0x7E = 126.
Note on DLE-escaped 0x03: The device's S3 response DLE-escapes ETX (0x03)
bytes as \\x10\\x03. The S3FrameParser preserves both bytes in frame.data.
Subsequent fields after offset 117 are therefore at raw_offset = logical+1.
The raw payload must be round-tripped verbatim in write; do NOT reapply DLE
destuffing or stripping.
"""
raw: Optional[bytes] = None # raw 125-byte read payload (for round-trip write)
# ── Main enable ──────────────────────────────────────────────────────────
auto_call_home_enabled: Optional[bool] = None # raw[5] ✅
# ── Dial string ──────────────────────────────────────────────────────────
dial_string: Optional[str] = None # raw[6:46] 40-byte null-padded ASCII ✅
# ── When to call ─────────────────────────────────────────────────────────
after_event_recorded: Optional[bool] = None # raw[87] ✅
at_specified_times: Optional[bool] = None # raw[91] ✅
# ── Time slot 1 ──────────────────────────────────────────────────────────
time1_enabled: Optional[bool] = None # raw[93] ✅
time1_hour: Optional[int] = None # raw[101] 0-23 ✅
time1_min: Optional[int] = None # raw[102] 0-59 ✅
# ── Time slot 2 ──────────────────────────────────────────────────────────
time2_enabled: Optional[bool] = None # raw[95] ✅
time2_hour: Optional[int] = None # raw[105] 0-23 ✅
time2_min: Optional[int] = None # raw[106] 0-59 ✅
# ── Retry / timeout settings (read-only; not writable via set_call_home_config) ──
num_retries: Optional[int] = None # raw[117:119]=10 03 → value 3 ✅
time_between_retries_sec: Optional[int] = None # raw[120] (shifted +1 by DLE) ✅
wait_for_connection_sec: Optional[int] = None # raw[122] ✅
warm_up_time_sec: Optional[int] = None # raw[124] ✅
# ── Event ─────────────────────────────────────────────────────────────────────
@dataclass
minimateplus/protocol.py
+89
View File
@@ -65,6 +65,7 @@ SUB_WAVEFORM_HEADER = 0x0A
SUB_WAVEFORM_RECORD = 0x0C
SUB_BULK_WAVEFORM = 0x5A
SUB_COMPLIANCE = 0x1A
SUB_CALL_HOME = 0x2C # Call home config read → response 0xD3 ✅
SUB_UNKNOWN_2E = 0x2E
# Write command SUBs (= Read SUB + 0x60, confirmed from BW captures 3-11-26)
@@ -78,6 +79,10 @@ SUB_WRITE_CONFIRM_C = 0x74 # Confirm C — sent after 69 ✅
SUB_TRIGGER_CONFIG_WRITE = 0x82 # Write trigger config (0x22 + 0x60) ✅
SUB_TRIGGER_CONFIRM = 0x83 # Confirm trigger write ✅
# Call home write SUBs (confirmed from 4-20-26 call home settings captures)
SUB_CALL_HOME_WRITE = 0x7E # Write call home config → response 0x81 ✅
SUB_CALL_HOME_CONFIRM = 0x7F # Confirm call home write → response 0x80 ✅
# Monitoring control SUBs (confirmed from 4-8-26/2ndtry BW TX capture)
SUB_START_MONITORING = 0x96 # Start monitoring → response 0x69 ✅
SUB_STOP_MONITORING = 0x97 # Stop monitoring → response 0x68 ✅
@@ -109,6 +114,7 @@ DATA_LENGTHS: dict[int, int] = {
# SUB_WAVEFORM_HEADER (0x0A) is VARIABLE — length read from probe response
# data[4]. Do NOT add it here; use read_waveform_header() instead. ✅
SUB_WAVEFORM_RECORD: 0xD2, # 210-byte waveform/histogram record ✅
SUB_CALL_HOME: 0x7C, # 124-byte call home config ✅ (confirmed 4-20-26)
SUB_UNKNOWN_2E: 0x1A, # 26 bytes, purpose TBD 🔶
0x09: 0xCA, # 202 bytes, purpose TBD 🔶
# SUB_COMPLIANCE (0x1A) uses a multi-step sequence with a 2090-byte total;
@@ -1087,6 +1093,89 @@ class MiniMateProtocol:
self._send(frame)
return self.recv_write_ack(expected_sub=rsp_sub)
# ── Call home config (SUBs 0x2C / 0x7E / 0x7F) ──────────────────────────
def read_call_home_config(self) -> bytes:
"""
Read the auto call home configuration (SUB 0x2C → response 0xD3).
Standard two-step read: probe (offset=0x00) then data (offset=0x7C=124).
Returns the raw 125-byte payload (data[11:] of the data response).
Confirmed from 4-20-26 call home settings capture:
- Probe response: data[4]=0x7C (confirms data length = 124)
- Data response: 136 bytes total (11-byte echo header + 125 bytes payload)
- Payload[0:3] = 0x00 0x7C 0xDC (header: zero, inner-length, constant)
- Payload[5] = auto_call_home_enabled
- Payload[6:46] = dial_string (40-byte null-padded ASCII "RADIO RING")
Returns:
Raw 125-byte call home config payload (data[11:]).
Suitable for round-trip write (append \\x00\\x00 → 127-byte write payload).
Raises:
ProtocolError: on timeout or wrong response SUB.
"""
rsp_sub = _expected_rsp_sub(SUB_CALL_HOME) # 0xFF - 0x2C = 0xD3
length = DATA_LENGTHS[SUB_CALL_HOME] # 0x7C = 124
log.debug("read_call_home_config: 0x2C probe")
self._send(build_bw_frame(SUB_CALL_HOME, 0))
self._recv_one(expected_sub=rsp_sub)
log.debug("read_call_home_config: 0x2C data request offset=0x%02X", length)
self._send(build_bw_frame(SUB_CALL_HOME, length))
data_rsp = self._recv_one(expected_sub=rsp_sub)
payload = data_rsp.data[11:]
log.debug("read_call_home_config: received %d payload bytes", len(payload))
return payload
def write_call_home_config(self, data: bytes) -> None:
"""
Write the auto call home configuration (SUB 0x7E → 0x7F confirm).
Write sequence (confirmed from 4-20-26 call home settings captures):
SUB 0x7E write 127-byte payload → device acks SUB 0x81
SUB 0x7F confirm (no data) → device acks SUB 0x80
The 127-byte write payload = 125-byte read payload + b'\\x00\\x00'.
The offset field = data[1] + 2 = 0x7C + 2 = 0x7E = 126.
Write frame format: build_bw_write_frame (minimal DLE stuffing — only
BW_CMD is doubled; all other bytes are RAW). The \\x10\\x03 sequence
within the payload is preserved as-is (device interprets DLE+ETX as the
literal value 0x03 per the inner-frame terminator convention).
Args:
data: 127-byte write payload (read payload + \\x00\\x00 footer).
Must start with [0x00][0x7C][...] (standard header).
Raises:
ValueError: if data is not exactly 127 bytes or lacks expected header.
ProtocolError: on timeout or wrong response SUB.
"""
if len(data) < 2:
raise ValueError(f"call home write payload must be at least 2 bytes, got {len(data)}")
rsp_sub_write = _expected_rsp_sub(SUB_CALL_HOME_WRITE) # 0xFF - 0x7E = 0x81
rsp_sub_confirm = _expected_rsp_sub(SUB_CALL_HOME_CONFIRM) # 0xFF - 0x7F = 0x80
# Offset formula: data[1] + 2 (same pattern as other single-chunk writes)
offset = data[1] + 2 # 0x7C + 2 = 0x7E = 126
frame = build_bw_write_frame(SUB_CALL_HOME_WRITE, data, offset=offset)
log.debug(
"write_call_home_config: %d bytes data[1]=0x%02X offset=0x%04X",
len(data), data[1], offset,
)
self._send(frame)
self.recv_write_ack(expected_sub=rsp_sub_write)
log.debug("write_call_home_config: write acked; sending confirm 0x7F")
confirm_frame = build_bw_write_frame(SUB_CALL_HOME_CONFIRM, b"")
self._send(confirm_frame)
self.recv_write_ack(expected_sub=rsp_sub_confirm)
log.debug("write_call_home_config: confirm acked — done")
# ── Monitoring ────────────────────────────────────────────────────────────
def read_monitor_status(self) -> S3Frame: