1
#!/usr/bin/env python3
2
"""Read diagnostic output over native USB without toggling reset lines."""
3
import json
4
import os
5
from pathlib import Path
6
import sys
7
import re
8
import shlex
9
import time
10
import termios
11
import serial
12
 
13
root = Path(__file__).resolve().parents[2]
14
sys.path.insert(0, str(root / "scripts"))
15
from serial_device import device_port
16
os.umask(0o077)
17
secrets = []
18
for line in (root / ".credential.env").read_text().splitlines():
19
    match = re.match(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*)$", line)
20
    if match:
21
        value = " ".join(shlex.split(match.group(2), comments=True))
22
        if value:
23
            secrets.append(value)
24
port = serial.Serial(port=None, baudrate=115200, timeout=0.2, exclusive=True)
25
port.dtr = True
26
port.rts = True
27
find_port = device_port
28
for attempt in range(40):
29
    try:
30
        port.port = find_port()
31
        port.open()
32
        attrs = termios.tcgetattr(port.fileno())
33
        attrs[2] &= ~termios.HUPCL
34
        termios.tcsetattr(port.fileno(), termios.TCSANOW, attrs)
35
        break
36
    except (serial.SerialException, RuntimeError):
37
        time.sleep(0.25)
38
else:
39
    raise SystemExit("USB serial port did not reappear")
40
pending = bytearray()
41
deadline = time.monotonic() + 80
42
out = root / "artifacts/archive/sfu-bringup"
43
out.mkdir(parents=True, exist_ok=True, mode=0o700)
44
ready = False
45
with (out / "hardware-serial.log").open("a") as log:
46
    while time.monotonic() < deadline:
47
        try:
48
            pending.extend(port.read(4096))
49
        except serial.SerialException:
50
            print("USB disconnected during the diagnostic run", flush=True)
51
            break
52
        while b"\n" in pending:
53
            raw, _, remainder = pending.partition(b"\n")
54
            pending = bytearray(remainder)
55
            line = raw.decode(errors="replace").strip()
56
            for value in secrets:
57
                line = line.replace(value, "[redacted]")
58
            log.write(line + "\n")
59
            log.flush()
60
            print(line, flush=True)
61
            if line.startswith("PROBE "):
62
                try:
63
                    event = json.loads(line[6:])
64
                except ValueError:
65
                    continue
66
                if event.get("event") == "hardware_ready":
67
                    (out / "hardware-result.json").write_text(json.dumps(event, indent=2) + "\n")
68
                    ready = True
69
        if ready:
70
            break
71
port.close()
72
if not ready:
73
    raise SystemExit("No diagnostic summary received within 80 seconds")