""" models.py β€” Plain-Python data models for the MiniMate Plus protocol library. All models are intentionally simple dataclasses with no protocol logic. They represent *decoded* device data β€” the client layer translates raw frame bytes into these objects, and the SFM API layer serialises them to JSON. Notes on certainty: Fields marked βœ… are confirmed from captured data. Fields marked πŸ”Ά are strongly inferred but not formally proven. Fields marked ❓ are present in the captured payload but not yet decoded. See docs/instantel_protocol_reference.md for full derivation details. """ from __future__ import annotations import datetime import struct from dataclasses import dataclass, field from typing import Optional # ── Timestamp ───────────────────────────────────────────────────────────────── @dataclass class Timestamp: """ Event timestamp decoded from the MiniMate Plus wire format. Two source formats exist: 1. 6-byte format (from event index / 1E header β€” not yet decoded in client): [flag:1] [year:2 BE] [unknown:1] [month:1] [day:1] Use Timestamp.from_bytes(). 2. 9-byte format (from Full Waveform Record / 0C, bytes 0–8) βœ… CONFIRMED: [day:1] [sub_code:1] [month:1] [year:2 BE] [unknown:1] [hour:1] [min:1] [sec:1] Use Timestamp.from_waveform_record(). Confirmed 2026-04-01 against Blastware event report (BE11529 thump event): raw bytes: 01 10 04 07 ea 00 00 1c 0c β†’ day=1, sub_code=0x10 (Waveform mode), month=4, year=2026, hour=0, minute=28, second=12 ← matches Blastware "00:28:12 April 1, 2026" The sub_code at byte[1] is the record-mode indicator: 0x10 β†’ Waveform (continuous / single-shot) βœ… other β†’ Histogram (code not yet captured ❓) The year 1995 is the device's factory-default RTC date β€” it appears whenever the battery has been disconnected. Treat 1995 as "clock not set". """ raw: bytes # raw bytes for round-tripping flag: int # byte 0 of 6-byte format, or sub_code from 9-byte format year: int # βœ… unknown_byte: int # separator byte (purpose unclear ❓) month: int # βœ… day: int # βœ… # Time fields β€” populated only from the 9-byte waveform-record format hour: Optional[int] = None # βœ… (waveform record format) minute: Optional[int] = None # βœ… (waveform record format) second: Optional[int] = None # βœ… (waveform record format) @classmethod def from_bytes(cls, data: bytes) -> "Timestamp": """ Decode a 6-byte timestamp (6-byte event-index format). Args: data: exactly 6 bytes from the device payload. Returns: Decoded Timestamp (no time fields). Raises: ValueError: if data is not exactly 6 bytes. """ if len(data) != 6: raise ValueError(f"Timestamp requires exactly 6 bytes, got {len(data)}") flag = data[0] year = struct.unpack_from(">H", data, 1)[0] unknown_byte = data[3] month = data[4] day = data[5] return cls( raw=bytes(data), flag=flag, year=year, unknown_byte=unknown_byte, month=month, day=day, ) @classmethod def from_waveform_record(cls, data: bytes) -> "Timestamp": """ Decode a 9-byte timestamp from the first bytes of a 210-byte waveform record (SUB 0C / Full Waveform Record response). Wire layout (βœ… CONFIRMED 2026-04-01 against Blastware event report): byte[0]: day (uint8) byte[1]: sub_code / mode flag (0x10 = Waveform single-shot) byte[2]: month (uint8) bytes[3–4]: year (big-endian uint16) byte[5]: unknown (0x00 in all observed samples) byte[6]: hour (uint8) byte[7]: minute (uint8) byte[8]: second (uint8) Used for sub_code=0x10 records only. For sub_code=0x03 (continuous mode) use from_continuous_record() β€” the layout is shifted by 1 byte. Args: data: at least 9 bytes; only the first 9 are consumed. Returns: Decoded Timestamp with hour/minute/second populated. Raises: ValueError: if data is fewer than 9 bytes. """ if len(data) < 9: raise ValueError( f"Waveform record timestamp requires at least 9 bytes, got {len(data)}" ) day = data[0] sub_code = data[1] # 0x10 = Waveform single-shot month = data[2] year = struct.unpack_from(">H", data, 3)[0] unknown_byte = data[5] hour = data[6] minute = data[7] second = data[8] return cls( raw=bytes(data[:9]), flag=sub_code, year=year, unknown_byte=unknown_byte, month=month, day=day, hour=hour, minute=minute, second=second, ) @classmethod def from_continuous_record(cls, data: bytes) -> "Timestamp": """ Decode a 10-byte timestamp from the first bytes of a sub_code=0x03 (Waveform Continuous) 210-byte record. Wire layout (βœ… CONFIRMED 2026-04-03 against Blastware event report, event recorded at 15:20:17 April 3 2026, raw: 10 03 10 04 07 ea 00 0f 14 11): byte[0]: unknown_a (0x10 observed β€” meaning TBD) byte[1]: day (uint8) byte[2]: unknown_b (0x10 observed β€” meaning TBD) bytes[3]: month (uint8) bytes[4–5]: year (big-endian uint16) byte[6]: unknown (0x00 in all observed samples) byte[7]: hour (uint8) byte[8]: minute (uint8) byte[9]: second (uint8) This is the sub_code=0x10 layout shifted forward by 1 byte, with two extra unknown bytes at [0] and [2]. The sub_code (0x03) itself is at byte[1] in the raw record, which also encodes the day β€” but the day value (3 = April 3rd) happens to differ from the sub_code (0x03) only in semantics; the byte is shared. Args: data: at least 10 bytes; only the first 10 are consumed. Returns: Decoded Timestamp with hour/minute/second populated. Raises: ValueError: if data is fewer than 10 bytes. """ if len(data) < 10: raise ValueError( f"Continuous record timestamp requires at least 10 bytes, got {len(data)}" ) unknown_a = data[0] # 0x10 observed; meaning unknown day = data[1] # doubles as the sub_code byte (0x03) β€” day=3 on Apr 3 unknown_b = data[2] # 0x10 observed; meaning unknown month = data[3] year = struct.unpack_from(">H", data, 4)[0] unknown_byte = data[6] hour = data[7] minute = data[8] second = data[9] return cls( raw=bytes(data[:10]), flag=unknown_a, year=year, unknown_byte=unknown_byte, month=month, day=day, hour=hour, minute=minute, second=second, ) @property def clock_set(self) -> bool: """False when year == 1995 (factory default / battery-lost state).""" return self.year != 1995 def __str__(self) -> str: if not self.clock_set: return f"CLOCK_NOT_SET ({self.year}-{self.month:02d}-{self.day:02d})" date_str = f"{self.year}-{self.month:02d}-{self.day:02d}" if self.hour is not None: return f"{date_str} {self.hour:02d}:{self.minute:02d}:{self.second:02d}" return date_str # ── Device identity ─────────────────────────────────────────────────────────── @dataclass class DeviceInfo: """ Combined device identity information gathered during the startup sequence. Populated from three response SUBs: - SUB EA (SERIAL_NUMBER_RESPONSE): serial, firmware_minor - SUB FE (FULL_CONFIG_RESPONSE): serial (repeat), firmware_version, dsp_version, manufacturer, model - SUB A4 (POLL_RESPONSE): manufacturer (repeat), model (repeat) All string fields are stripped of null padding before storage. """ # ── From SUB EA (SERIAL_NUMBER_RESPONSE) ───────────────────────────────── serial: str # e.g. "BE18189" βœ… firmware_minor: int # 0x11 = 17 for S337.17 βœ… serial_trail_0: Optional[int] = None # unit-specific byte β€” purpose unknown ❓ # ── From SUB FE (FULL_CONFIG_RESPONSE) ──────────────────────────────────── firmware_version: Optional[str] = None # e.g. "S337.17" βœ… dsp_version: Optional[str] = None # e.g. "10.72" βœ… manufacturer: Optional[str] = None # e.g. "Instantel" βœ… model: Optional[str] = None # e.g. "MiniMate Plus" βœ… # ── From SUB 1A (COMPLIANCE_CONFIG_RESPONSE) ────────────────────────────── compliance_config: Optional["ComplianceConfig"] = None # E5 response, read in connect() # ── From SUB 08 (EVENT_INDEX_RESPONSE) ──────────────────────────────────── event_count: Optional[int] = None # stored event count from F7 response πŸ”Ά def __str__(self) -> str: fw = self.firmware_version or f"?.{self.firmware_minor}" mdl = self.model or "MiniMate Plus" return f"{mdl} S/N:{self.serial} FW:{fw}" # ── Channel threshold / scaling ─────────────────────────────────────────────── @dataclass class ChannelConfig: """ Per-channel threshold and scaling values from SUB E5 / SUB 71. Floats are stored in the device in imperial units (in/s for geo channels, psi for MicL). Unit strings embedded in the payload confirm this. Certainty: βœ… CONFIRMED for trigger_level, alarm_level, unit strings. """ label: str # e.g. "Tran", "Vert", "Long", "MicL" βœ… trigger_level: float # in/s (geo) or psi (MicL) βœ… alarm_level: float # in/s (geo) or psi (MicL) βœ… max_range: float # hardware/firmware sensitivity constant (e.g. 6.206053) βœ… confirmed same on all units unit_label: str # e.g. "in./s" or "psi" βœ… # ── Peak values for one event ───────────────────────────────────────────────── @dataclass class PeakValues: """ Per-channel peak particle velocity / pressure for a single event, plus the scalar Peak Vector Sum. Extracted from the Full Waveform Record (SUB F3 / 0C response), stored as IEEE 754 big-endian floats in the device's native units (in/s / psi). Per-channel PPV location (βœ… CONFIRMED 2026-04-01): Found by searching for the 4-byte channel label string ("Tran", "Vert", "Long", "MicL") and reading the float at label_offset + 6. Peak Vector Sum (βœ… CONFIRMED 2026-04-01): Fixed offset 87 in the 210-byte record. = √(TranΒ² + VertΒ² + LongΒ²) at the sample instant of maximum combined geo motion. NOT the vector sum of the three per-channel peak values (those may occur at different times). Matches Blastware's "Peak Vector Sum" display exactly. """ tran: Optional[float] = None # Transverse PPV (in/s) βœ… vert: Optional[float] = None # Vertical PPV (in/s) βœ… long: Optional[float] = None # Longitudinal PPV (in/s) βœ… micl: Optional[float] = None # Air overpressure (psi) πŸ”Ά (units uncertain) peak_vector_sum: Optional[float] = None # Scalar geo PVS (in/s) βœ… # ── Project / operator metadata ─────────────────────────────────────────────── @dataclass class ProjectInfo: """ Operator-supplied project and location strings from the Full Waveform Record (SUB F3) and compliance config block (SUB E5 / SUB 71). All fields are optional β€” they may be blank if the operator did not fill them in through Blastware. """ setup_name: Optional[str] = None # "Standard Recording Setup" project: Optional[str] = None # project description client: Optional[str] = None # client name βœ… confirmed offset operator: Optional[str] = None # operator / user name sensor_location: Optional[str] = None # sensor location string notes: Optional[str] = None # extended notes # ── Compliance Config ────────────────────────────────────────────────────────── @dataclass class ComplianceConfig: """ Device compliance and recording configuration from SUB 1A (response E5). Contains device-wide settings like record time, trigger/alarm thresholds, and operator-supplied strings. This is read once during connect() and cached in DeviceInfo. All fields are optional β€” some may not be decoded yet or may be absent from the device configuration. """ raw: Optional[bytes] = None # full 2090-byte payload (for debugging) # Recording parameters (βœ… CONFIRMED from Β§7.6) recording_mode: Optional[int] = None # uint8: 0x00=Single Shot, 0x01=Continuous, # 0x03=Histogram, 0x04=Histogram+Continuous βœ… confirmed 2026-04-20 # Read (E5): data[anchor_pos - 8] (6-byte anchor) # Write (SUB 71): data[anchor_pos - 7] sample_rate: Optional[int] = None # sps (1024, 2048, 4096) histogram_interval_sec: Optional[int] = None # uint16 BE, seconds βœ… confirmed 2026-04-20 # anchor_pos - 4 (same offset in read & write) # Valid values: 2, 5, 15, 60, 300, 900 # Mode-gated: only active in Histogram/Histogram+Continuous record_time: Optional[float] = None # seconds (e.g. 3.0, 5.0, 8.0, 10.0) # Trigger/alarm levels (βœ… CONFIRMED per-channel at Β§7.6) # For now we store the first geo channel (Transverse) as representatives; # full per-channel data would require structured Channel objects. trigger_level_geo: Optional[float] = None # in/s (first geo channel) βœ… alarm_level_geo: Optional[float] = None # in/s (first geo channel) βœ… geo_adc_scale: Optional[float] = None # ADC-to-velocity scale factor (float32 at Tran+28) βœ… # = inverse sensitivity = 1/sensitivity (in/s per V) # Formula (Interface Handbook Β§4.5): Range = 1.61133 V Γ— scale_factor # β†’ 1.61133 Γ— 6.206053 = 10.000 in/s (Normal range) βœ… # Firmware uses: PPV (in/s) = ADC_voltage (V) Γ— 6.206053 # Identical on BE11529 and BE18189 β€” same Instantel geophone hardware. # NOT a user-configurable setting. Must NOT be written. geo_range: Optional[int] = None # range/sensitivity selector β€” CONFIRMED 2026-04-20 # 0x00 = Normal 10.000 in/s (standard gain) # 0x01 = Sensitive 1.250 in/s (high gain) # Offset: Tran+33 in both E5 read and SUB 71 write payloads # (same 2126-byte buffer is round-tripped; applied to Tran/Vert/Long) # Project/setup strings (sourced from E5 / SUB 71 write payload) # These are the FULL project metadata from compliance config, # complementing the sparse ProjectInfo found in the waveform record (SUB 0C). setup_name: Optional[str] = None # "Standard Recording Setup" project: Optional[str] = None # project description client: Optional[str] = None # client name operator: Optional[str] = None # operator / user name sensor_location: Optional[str] = None # sensor location string 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 class Event: """ A single seismic event record downloaded from the device. Populated progressively across several request/response pairs: 1. SUB 1E (EVENT_HEADER) β†’ index, timestamp, sample_rate 2. SUB 0C (FULL_WAVEFORM_RECORD) β†’ peak_values, project_info, record_type 3. SUB 5A (BULK_WAVEFORM_STREAM) β†’ raw_samples (downloaded on demand) Fields not yet retrieved are None. """ # ── Identity ────────────────────────────────────────────────────────────── index: int # 0-based event number on device # ── From EVENT_HEADER (SUB 1E) ──────────────────────────────────────────── timestamp: Optional[Timestamp] = None # 6-byte timestamp βœ… sample_rate: Optional[int] = None # samples/sec (e.g. 1024) πŸ”Ά # ── From FULL_WAVEFORM_RECORD (SUB F3) ─────────────────────────────────── peak_values: Optional[PeakValues] = None project_info: Optional[ProjectInfo] = None record_type: Optional[str] = None # e.g. "Histogram", "Waveform" πŸ”Ά # ── From BULK_WAVEFORM_STREAM (SUB 5A) ─────────────────────────────────── # Raw ADC samples keyed by channel label. Not fetched unless explicitly # requested (large data transfer β€” up to several MB per event). raw_samples: Optional[dict] = None # {"Tran": [...], "Vert": [...], ...} total_samples: Optional[int] = None # from STRT record: expected total sample-sets pretrig_samples: Optional[int] = None # from STRT record: pre-trigger sample count rectime_seconds: Optional[int] = None # from STRT record: record duration (seconds) # ── Debug / introspection ───────────────────────────────────────────────── # Raw 210-byte waveform record bytes, set when debug mode is active. # Exposed by the SFM server via ?debug=true so field layouts can be verified. _raw_record: Optional[bytes] = field(default=None, repr=False) # 4-byte waveform key used to request this event via SUB 5A. # Set by get_events(); required by download_waveform(). _waveform_key: Optional[bytes] = field(default=None, repr=False) # Raw A5 frames from the full bulk waveform download (full_waveform=True). # Populated by get_events() when full_waveform=True; used by write_n00(). _a5_frames: Optional[list] = field(default=None, repr=False) def __str__(self) -> str: ts = str(self.timestamp) if self.timestamp else "no timestamp" ppv = "" if self.peak_values: pv = self.peak_values parts = [] if pv.tran is not None: parts.append(f"T={pv.tran:.4f}") if pv.vert is not None: parts.append(f"V={pv.vert:.4f}") if pv.long is not None: parts.append(f"L={pv.long:.4f}") if pv.micl is not None: parts.append(f"M={pv.micl:.6f}") ppv = " [" + ", ".join(parts) + " in/s]" return f"Event#{self.index} {ts}{ppv}" # ── MonitorLogEntry ─────────────────────────────────────────────────────────── @dataclass class MonitorLogEntry: """ A monitor log entry decoded from a SUB 0x0A (WAVEFORM_HEADER) response whose first byte is 0x2C (partial record, recording mode = continuous monitoring without a triggered event). These are the "partial bins" that Blastware stores between triggered events. Each entry represents one monitoring interval β€” the span of time during which the unit was actively monitoring but no threshold crossing occurred. Confirmed from 4-11-26 MITM capture analysis (2026-04-11): Header layout (full response data[0:]): data[0] = 0x2C (partial record type / data length in probe response) data[1:5] = 0x00 Γ— 4 data[5:9] = event key (4 bytes, big-endian hex) data[9:11] = 0x00 Γ— 2 data[11:] = timestamp_start (9 or 10 bytes depending on recording mode) + timestamp_stop (same format) + separator (4–5 bytes, variable) + serial null-terminated (e.g. "BE11529\\0") + "Geo: X.XXX in/s\\0" (trigger threshold string) Timestamp format detection: data[11] == 0x10 β†’ 10-byte sub_code=0x03 (continuous) format data[12] == 0x10 β†’ 9-byte sub_code=0x10 (single-shot) format In contrast to Event (triggered records, type 0x46), MonitorLogEntry records do NOT have a waveform record (SUB 0x0C) or bulk waveform stream (SUB 5A). All available metadata is in the 0x0A header alone. """ index: int # 0-based position in device record list key: str # 8-hex event key (e.g. "01114290") βœ… start_time: Optional[datetime.datetime] = None # monitoring session start βœ… stop_time: Optional[datetime.datetime] = None # monitoring session stop βœ… serial: Optional[str] = None # device serial (e.g. "BE11529") βœ… geo_threshold_ips: Optional[float] = None # trigger level from "Geo: X.XXX in/s" βœ… # Raw bytes for debugging / future decoding raw_header: Optional[bytes] = field(default=None, repr=False) @property def duration_seconds(self) -> Optional[float]: """Duration of monitoring interval in seconds, or None if times unavailable.""" if self.start_time and self.stop_time: return (self.stop_time - self.start_time).total_seconds() return None def __str__(self) -> str: start = self.start_time.isoformat() if self.start_time else "?" stop = self.stop_time.isoformat() if self.stop_time else "?" dur = f" ({self.duration_seconds:.0f}s)" if self.duration_seconds is not None else "" return f"MonitorLog#{self.index} key={self.key} {start}β†’{stop}{dur}" # ── MonitorStatus ───────────────────────────────────────────────────────────── @dataclass class MonitorStatus: """ Current monitoring state decoded from SUB 0x1C response. Confirmed field locations from 4-8-26/2ndtry BW capture: battery_v : data[11 + 0x2F : 11 + 0x31] uint16 BE Γ· 100 e.g. 680 β†’ 6.80 V memory_total: data[11 + 0x31 : 11 + 0x35] uint32 BE bytes e.g. 983040 β†’ 960 KB memory_free : data[11 + 0x35 : 11 + 0x39] uint32 BE bytes (subset of total) is_monitoring: inferred from payload length β€” idle = 44 bytes, monitoring = 12 bytes """ is_monitoring: bool # True if unit is actively recording βœ… battery_v: Optional[float] = None # Battery voltage in volts βœ… memory_total: Optional[int] = None # Total flash memory in bytes βœ… memory_free: Optional[int] = None # Free flash memory in bytes βœ