seismo-relay/docs/parse_0x1c_response.py

#!/usr/bin/env python3
"""
Parse SUB 0x1C (monitoring status) response frames.

SUB 0x1C returns device monitoring status with different payload sizes depending on state:
  - IDLE (not monitoring): 58 bytes with full details
  - MONITORING (actively streaming): 12 bytes condensed format
"""

import struct
from dataclasses import dataclass
from typing import Optional


@dataclass
class MonitoringStatus:
    """Parsed SUB 0x1C response fields."""

    monitor_mode: int  # 0x2c = OFF, 0x00 = ON
    day: int           # 1–31
    hour: int          # 0–23
    month: int         # 1–12
    year: int          # 2000–2100
    minute: int        # 0–59 (uncertain encoding)
    second: int        # 0–59 (uncertain encoding)
    battery_voltage_v: float  # Volts (6–8V typical)
    memory_total_kb: float    # Kilobytes
    memory_free_kb: float     # Kilobytes
    raw_payload: bytes

    def __str__(self) -> str:
        mode_str = "OFF" if self.monitor_mode == 0x2c else "ON"
        date_str = f"{self.year:04d}-{self.month:02d}-{self.day:02d}"
        time_str = f"{self.hour:02d}:{self.minute:02d}:{self.second:02d}"
        return (
            f"MonitoringStatus(\n"
            f"  mode={mode_str} (0x{self.monitor_mode:02x})\n"
            f"  datetime={date_str} {time_str}\n"
            f"  battery={self.battery_voltage_v:.2f}V\n"
            f"  memory=total {self.memory_total_kb:.1f} KB, "
            f"free {self.memory_free_kb:.1f} KB\n"
            f")"
        )


def parse_0x1c_response(data: bytes) -> Optional[MonitoringStatus]:
    """
    Parse a SUB 0x1C response payload (after S3 header removed).

    Args:
        data: Destuffed payload bytes (without the 5-byte S3 header)

    Returns:
        MonitoringStatus object, or None if parse fails
    """

    if len(data) < 39:
        # Minimum size for idle response
        print(f"[!] Payload too short: {len(data)} bytes (need >=39)")
        return None

    try:
        monitor_mode = data[0x00]

        day = data[0x0d]
        hour = data[0x0e]
        month = data[0x0f]
        year = struct.unpack('>H', data[0x10:0x12])[0]
        minute = data[0x12]
        second = data[0x13]

        # Battery voltage: uint16 BE, divide by 100
        # At offset [2f:31]
        voltage_raw = struct.unpack('>H', data[0x2f:0x31])[0]
        battery_voltage_v = voltage_raw / 100.0

        # Memory total: uint32 BE, in bytes
        # At offset [31:35]
        memory_total_bytes = struct.unpack('>I', data[0x31:0x35])[0]
        memory_total_kb = memory_total_bytes / 1024.0

        # Memory free: uint32 BE, in bytes
        # At offset [35:39]
        memory_free_bytes = struct.unpack('>I', data[0x35:0x39])[0]
        memory_free_kb = memory_free_bytes / 1024.0

        return MonitoringStatus(
            monitor_mode=monitor_mode,
            day=day,
            hour=hour,
            month=month,
            year=year,
            minute=minute,
            second=second,
            battery_voltage_v=battery_voltage_v,
            memory_total_kb=memory_total_kb,
            memory_free_kb=memory_free_kb,
            raw_payload=data
        )

    except (struct.error, IndexError) as e:
        print(f"[!] Parse error: {e}")
        return None


def hex_dump(data: bytes, offset: int = 0) -> str:
    """Pretty-print hex dump of binary data."""
    lines = []
    for i in range(0, len(data), 16):
        chunk = data[i:i+16]
        hex_str = ' '.join(f'{b:02x}' for b in chunk)
        ascii_str = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {offset+i:04x}: {hex_str:<48} {ascii_str}")
    return '\n'.join(lines)


if __name__ == '__main__':
    import sys

    if len(sys.argv) < 2:
        print("Usage: parse_0x1c_response.py <hex_string_or_file>")
        print()
        print("Example (hex string):")
        print("  python3 parse_0x1c_response.py 2c00000000000000000000000008100407ea00013b2d...")
        print()
        print("Example (from capture file, idle frame):")
        print("  Idle response (58 bytes):")
        idle_hex = (
            "2c00000000000000000000000008100407ea00013b2d000000000000"
            "010107cb00060000010107cb0015000000001002a8000efff2000e9e52ef"
        )
        status = parse_0x1c_response(bytes.fromhex(idle_hex))
        print(hex_dump(bytes.fromhex(idle_hex)))
        print()
        if status:
            print(status)

        sys.exit(0)

    # Parse input
    input_str = sys.argv[1]

    try:
        payload = bytes.fromhex(input_str)
    except ValueError:
        print(f"[!] Invalid hex string: {input_str}")
        sys.exit(1)

    print(f"Parsing {len(payload)} bytes:")
    print(hex_dump(payload))
    print()

    status = parse_0x1c_response(payload)
    if status:
        print(status)
    else:
        print("[!] Failed to parse")
        sys.exit(1)