SMS Nobreak Serial Script

read_sms_ups.py

#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.10"
# dependencies = ["pyserial"]
# ///
from __future__ import annotations

import argparse
import json
import os
import sys
import time
from dataclasses import dataclass, asdict
from glob import glob

import serial


CMD_STATUS = bytes.fromhex("51ffffffffb30d")
CMD_NAME = bytes.fromhex("49ffffffffbb0d")
CMD_INFO = bytes.fromhex("46ffffffffbe0d")


@dataclass
class Reading:
    port: str
    baudrate: int
    command: str
    raw: str = ""
    text: str = ""
    lastinput_vac: float | None = None
    input_vac: float | None = None
    output_vac: float | None = None
    output_power: float | None = None
    output_hz: float | None = None
    battery_level: float | None = None
    temperature_c: float | None = None
    bitfield: str = ""
    ups_ok: bool | None = None
    beep_on: bool | None = None
    shutdown_active: bool | None = None
    test_active: bool | None = None
    boost: bool | None = None
    bypass: bool | None = None
    battery_low: bool | None = None
    battery_in_use: bool | None = None


def open_serial(port: str, baudrate: int, timeout: float) -> serial.Serial:
    return serial.Serial(
        port=port,
        baudrate=baudrate,
        parity=serial.PARITY_NONE,
        stopbits=serial.STOPBITS_ONE,
        bytesize=serial.EIGHTBITS,
        timeout=timeout,
    )


def send_and_read(ser: serial.Serial, payload: bytes, wait: float = 0.1, read_window: float = 2.0) -> bytes:
    ser.reset_input_buffer()
    ser.reset_output_buffer()
    for b in payload:
        ser.write(bytes([b]))
        time.sleep(0.1)
    ser.flush()
    time.sleep(wait)
    chunks: list[bytes] = []
    deadline = time.time() + read_window
    while time.time() < deadline:
        chunk = ser.read(32)
        if chunk:
            chunks.append(chunk)
        elif chunks:
            break
    return b"".join(chunks)


def parse_name(raw: bytes) -> str:
    return raw.decode("ascii", errors="ignore").strip("\r\n\x00")


def parse_info(raw: bytes) -> str:
    return raw.decode("ascii", errors="ignore").strip("\r\n\x00")


def split_status(raw_hex: str) -> list[str] | None:
    if not raw_hex:
        return None
    r = raw_hex.lower()
    if not r.startswith("3d"):
        return None
    parts = [r[0:2], r[2:6], r[6:10], r[10:14], r[14:18], r[18:22], r[22:26], r[26:30], r[30:32], r[32:34], r[34:36]]
    return parts


def parse_status(raw: bytes) -> dict:
    raw_hex = raw.hex()
    parts = split_status(raw_hex)
    if not parts:
        return {"raw_hex": raw_hex}

    def u16(i: int) -> int:
        return int(parts[i], 16)

    bitfield = f"{u16(8):08b}"
    return {
        "raw_hex": raw_hex,
        "lastinput_vac": u16(1) / 10,
        "input_vac": u16(2) / 10,
        "output_vac": u16(3) / 10,
        "output_power": u16(4) / 10,
        "output_hz": u16(5) / 10,
        "battery_level": u16(6) / 10,
        "temperature_c": u16(7) / 10,
        "bitfield": bitfield,
        "beep_on": bitfield[7] == "1",
        "shutdown_active": bitfield[6] == "1",
        "test_active": bitfield[5] == "1",
        "ups_ok": bitfield[4] == "1",
        "boost": bitfield[3] == "1",
        "bypass": bitfield[2] == "1",
        "battery_low": bitfield[1] == "1",
        "battery_in_use": bitfield[0] == "1",
    }


def format_status(parsed: dict) -> str:
    if "raw_hex" in parsed and len(parsed) == 1:
        return f"raw_hex={parsed['raw_hex']}"
    order = [
        "lastinput_vac",
        "input_vac",
        "output_vac",
        "output_power",
        "output_hz",
        "battery_level",
        "temperature_c",
        "ups_ok",
        "battery_in_use",
        "battery_low",
        "bypass",
        "boost",
        "test_active",
        "shutdown_active",
        "beep_on",
        "bitfield",
        "raw_hex",
    ]
    return "\n".join(f"{k}={parsed.get(k)}" for k in order)


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="Safe SMS UPS helper")
    p.add_argument("--port", default=os.environ.get("SMSUPS_PORT", "/dev/ttyUSB0"), help="Serial port, e.g. /dev/ttyUSB0")
    p.add_argument("--baudrate", type=int, default=2400, help="Serial baudrate")
    p.add_argument("--timeout", type=float, default=1.0, help="Per-read timeout in seconds")
    p.add_argument("--json", action="store_true", help="Force JSON output")
    p.add_argument("--list-ports", action="store_true", help="List common serial ports and exit")

    sub = p.add_subparsers(dest="cmd")
    sub.add_parser("name", help="Read UPS name")
    sub.add_parser("info", help="Read UPS model/info")
    sub.add_parser("status", help="Read UPS status")
    send = sub.add_parser("send", help="Send a custom hex payload")
    send.add_argument("hex", help="Hex payload, e.g. 4dffffffffb70d")

    return p


def main() -> None:
    args = build_parser().parse_args()

    if args.list_ports:
        ports = sorted(set(glob("/dev/ttyUSB*") + glob("/dev/ttyACM*") + glob("/dev/ttyS*") + glob("/dev/cu.*")))
        print("\n".join(ports) if ports else "(no serial ports found)")
        return

    cmd = args.cmd or "status"
    reading = Reading(port=args.port, baudrate=args.baudrate, command=cmd)

    try:
        with open_serial(args.port, args.baudrate, args.timeout) as ser:
            if cmd == "name":
                raw = send_and_read(ser, CMD_NAME)
                if not raw:
                    print("erro: sem resposta do nobreak", file=sys.stderr)
                    raise SystemExit(2)
                print(parse_name(raw))
                return

            if cmd == "info":
                raw = send_and_read(ser, CMD_INFO)
                if not raw:
                    print("erro: sem resposta do nobreak", file=sys.stderr)
                    raise SystemExit(2)
                reading.raw = raw.hex()
                reading.text = parse_info(raw)
                if args.json:
                    print(json.dumps(asdict(reading), ensure_ascii=False, indent=2))
                else:
                    print(reading.text)
                return

            if cmd == "status":
                raw = send_and_read(ser, CMD_STATUS)
                if not raw:
                    print("erro: sem resposta do nobreak", file=sys.stderr)
                    raise SystemExit(2)
                parsed = parse_status(raw)
                if len(parsed) == 1:
                    print(f"erro: resposta fora do formato esperado: {parsed['raw_hex']}", file=sys.stderr)
                    raise SystemExit(3)
                reading.raw = parsed.get("raw_hex", "")
                reading.lastinput_vac = parsed.get("lastinput_vac")
                reading.input_vac = parsed.get("input_vac")
                reading.output_vac = parsed.get("output_vac")
                reading.output_power = parsed.get("output_power")
                reading.output_hz = parsed.get("output_hz")
                reading.battery_level = parsed.get("battery_level")
                reading.temperature_c = parsed.get("temperature_c")
                reading.bitfield = parsed.get("bitfield", "")
                reading.ups_ok = parsed.get("ups_ok")
                reading.beep_on = parsed.get("beep_on")
                reading.shutdown_active = parsed.get("shutdown_active")
                reading.test_active = parsed.get("test_active")
                reading.boost = parsed.get("boost")
                reading.bypass = parsed.get("bypass")
                reading.battery_low = parsed.get("battery_low")
                reading.battery_in_use = parsed.get("battery_in_use")
                if args.json:
                    print(json.dumps(asdict(reading), ensure_ascii=False, indent=2))
                else:
                    print(format_status(parsed))
                return

            if cmd == "send":
                try:
                    payload = bytes.fromhex(args.hex)
                except ValueError:
                    print("erro: hex inválido", file=sys.stderr)
                    raise SystemExit(4)
                raw = send_and_read(ser, payload, read_window=0.5)
                reading.raw = raw.hex()
                if args.json:
                    print(json.dumps(asdict(reading), ensure_ascii=False, indent=2))
                else:
                    print(f"sent={args.hex.lower()} raw={raw.hex()}")
                return
    except serial.SerialException as e:
        print(f"erro serial: {e}", file=sys.stderr)
        raise SystemExit(1)


if __name__ == "__main__":
    main()