1
#!/usr/bin/env python3
2
"""Stream filtered USB device logs. --check waits for a successful HTTPS heartbeat."""
3
import argparse
4
import json
5
import termios
6
import time
7
 
8
from config import read_env
9
from project import ROOT
10
from serial_device import device_port
11
 
12
VISIBLE_LOG_MARKERS = (
13
    "HTTPS ",
14
    "ON AIR:",
15
    "Music validated:",
16
    "Live spectrum ready:",
17
    "Crypto ",
18
    "Crypto:",
19
    "crypto security tests passed",
20
    "str0m initialized",
21
)
22
 
23
 
24
def open_port():
25
    import serial
26
    port = serial.Serial(port=None, baudrate=115200, timeout=0.1, exclusive=True)
27
    port.dtr = True
28
    port.rts = True
29
    port.port = device_port()
30
    port.open()
31
    attrs = termios.tcgetattr(port.fileno())
32
    attrs[2] &= ~termios.HUPCL
33
    termios.tcsetattr(port.fileno(), termios.TCSANOW, attrs)
34
    return port
35
 
36
 
37
def main(argv=None):
38
    import serial
39
    parser = argparse.ArgumentParser(description=__doc__)
40
    parser.add_argument(
41
        "--reset", action="store_true", help="Reboot the board before monitoring"
42
    )
43
    parser.add_argument(
44
        "--check",
45
        action="store_true",
46
        help="Require a successful heartbeat (and startup banner with --reset); nonzero on failure",
47
    )
48
    parser.add_argument(
49
        "--seconds",
50
        type=float,
51
        help="Stop after this duration (check defaults to 90 seconds)",
52
    )
53
    args = parser.parse_args(argv)
54
    if args.seconds is not None and args.seconds <= 0:
55
        parser.error("--seconds must be positive")
56
    seconds = args.seconds if args.seconds is not None else (90 if args.check else None)
57
    secrets = read_env(ROOT / ".credential.env")
58
    out = ROOT / "artifacts/radio"
59
    out.mkdir(parents=True, exist_ok=True, mode=0o700)
60
    prefix = "autonomous-check" if args.check else "autonomous-monitor"
61
    path = out / (prefix + ".log")
62
    path.touch(mode=0o600, exist_ok=True)
63
    path.chmod(0o600)
64
    port = open_port()
65
    if args.reset:
66
        port.write(b'{"cmd":"restart_device"}\n')
67
        port.flush()
68
    pending = bytearray()
69
    result = {
70
        "reset_requested": args.reset,
71
        "usb_signaling_commands_sent": 0,
72
        "on_air_banner_captured": False,
73
        "https_stages": [],
74
        "successful_heartbeats": 0,
75
    }
76
    deadline = time.monotonic() + seconds if seconds is not None else None
77
    try:
78
        with path.open("w") as log:
79
            while deadline is None or time.monotonic() < deadline:
80
                try:
81
                    pending.extend(port.read(8192))
82
                except (serial.SerialException, OSError):
83
                    port.close()
84
                    time.sleep(0.3)
85
                    try:
86
                        port = open_port()
87
                    except (serial.SerialException, OSError, RuntimeError):
88
                        continue
89
                    continue
90
                while b"\n" in pending:
91
                    raw, _, pending = pending.partition(b"\n")
92
                    line = raw.decode(errors="replace").strip()
93
                    for value in secrets.values():
94
                        if value:
95
                            line = line.replace(value, "[redacted]")
96
                    log.write(line + "\n")
97
                    log.flush()
98
                    # Print selected diagnostics; save all redacted lines in the private log.
99
                    if any(marker in line for marker in VISIBLE_LOG_MARKERS):
100
                        print(line[:400], flush=True)
101
                    if "ON AIR: autonomous" in line:
102
                        result["on_air_banner_captured"] = True
103
                    for stage in ("start", "channels", "ready"):
104
                        if (
105
                            f"HTTPS {stage}: status=200" in line
106
                            and stage not in result["https_stages"]
107
                        ):
108
                            result["https_stages"].append(stage)
109
                    if "HTTPS heartbeat: status=200" in line:
110
                        result["successful_heartbeats"] += 1
111
                if (
112
                    args.check
113
                    and result["successful_heartbeats"]
114
                    and (not args.reset or result["on_air_banner_captured"])
115
                ):
116
                    break
117
    except KeyboardInterrupt:
118
        pass
119
    finally:
120
        port.close()
121
    if args.check:
122
        result["passed"] = bool(result["successful_heartbeats"]) and (
123
            not args.reset or result["on_air_banner_captured"]
124
        )
125
        summary = out / (prefix + ".json")
126
        summary.touch(mode=0o600, exist_ok=True)
127
        summary.chmod(0o600)
128
        summary.write_text(json.dumps(result, indent=2) + "\n")
129
        print(json.dumps(result, indent=2), flush=True)
130
        return 0 if result["passed"] else 1
131
    return 0
132
 
133
 
134
if __name__ == "__main__":
135
    raise SystemExit(main())