1
#!/usr/bin/env python3
2
"""Follow echo-datachannels' SFU flow, with the S3 signaled over USB.
3
4
The SFU token stays on this computer. The S3 connects to the SFU over Wi-Fi.
5
The second WebRTC endpoint uses aiortc to validate the transport without a UI.
6
"""
7
import asyncio
8
import json
9
import os
10
from pathlib import Path
11
import sys
12
import re
13
import shlex
14
import time
15
import termios
16
from urllib.error import HTTPError
17
from urllib.request import Request, urlopen
18
 
19
from aiortc import RTCConfiguration, RTCPeerConnection, RTCSessionDescription
20
import serial
21
 
22
ROOT = Path(__file__).resolve().parents[2]
23
sys.path.insert(0, str(ROOT / "scripts"))
24
from serial_device import device_port
25
OUT = ROOT / "artifacts/archive/sfu-bringup"
26
 
27
 
28
 
29
class SFU:
30
    def __init__(self):
31
        os.umask(0o077)
32
        OUT.mkdir(parents=True, exist_ok=True, mode=0o700)
33
        values = {}
34
        for line in (ROOT / ".credential.env").read_text().splitlines():
35
            match = re.match(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*)$", line)
36
            if match:
37
                values[match.group(1)] = " ".join(shlex.split(match.group(2), comments=True))
38
        self.base = "https://rtc.live.cloudflare.com/v1/apps/" + values["REALTIME_APP_ID"]
39
        self.token = values["REALTIME_APP_TOKEN"]
40
 
41
    def request(self, path, body=None, method="POST"):
42
        request = Request(self.base + path,
43
            data=None if body is None else json.dumps(body).encode(), method=method,
44
            headers={"Authorization": "Bearer " + self.token,
45
                "Content-Type": "application/json", "User-Agent": "esp32-hardware-validation/0.1"})
46
        try:
47
            with urlopen(request, timeout=20) as response:
48
                result = json.load(response)
49
        except HTTPError as error:
50
            try:
51
                result = json.loads(error.read(2048))
52
            except (ValueError, UnicodeDecodeError):
53
                result = {}
54
            raise RuntimeError(f"SFU HTTP {error.code}: {result.get('errorCode', 'unknown_error')}") from None
55
        if result.get("errorCode"):
56
            raise RuntimeError(f"SFU: {result['errorCode']}")
57
        return result
58
 
59
    async def call(self, *args, **kwargs):
60
        return await asyncio.to_thread(self.request, *args, **kwargs)
61
 
62
    async def establish(self):
63
        session = await self.call("/sessions/new")
64
        sid = session["sessionId"]
65
        transport = await self.call(f"/sessions/{sid}/datachannels/establish", {
66
            "dataChannel": {"location": "remote", "dataChannelName": "server-events"}})
67
        if not transport.get("requiresImmediateRenegotiation"):
68
            raise RuntimeError("SFU transport did not request SDP negotiation")
69
        return sid, transport
70
 
71
 
72
class Device:
73
    def __init__(self):
74
        os.umask(0o077)
75
        OUT.mkdir(parents=True, exist_ok=True, mode=0o700)
76
        self.port = serial.Serial(port=None, baudrate=115200, timeout=0, write_timeout=5, exclusive=True)
77
        # Keep both lines asserted on open to avoid an intermediate USB-JTAG
78
        # reset state, and preserve them across close/reopen on Linux.
79
        self.port.dtr = True
80
        self.port.rts = True
81
        self.port.port = device_port()
82
        self.port.open()
83
        attrs = termios.tcgetattr(self.port.fileno())
84
        attrs[2] &= ~termios.HUPCL
85
        termios.tcsetattr(self.port.fileno(), termios.TCSANOW, attrs)
86
        self.pending = bytearray()
87
        self.events = []
88
        self.log = (OUT / "sfu-device.log").open("a")
89
 
90
    def send(self, **command):
91
        self.port.write(json.dumps(command).encode() + b"\n")
92
        self.port.flush()
93
 
94
    def poll(self):
95
        self.pending.extend(self.port.read(16384))
96
        while b"\n" in self.pending:
97
            raw, _, remainder = self.pending.partition(b"\n")
98
            self.pending = bytearray(remainder)
99
            line = raw.decode(errors="replace").strip()
100
            self.log.write(line + "\n")
101
            self.log.flush()
102
            if line.startswith("PROBE "):
103
                try:
104
                    event = json.loads(line[6:])
105
                except ValueError:
106
                    continue
107
                self.events.append(event)
108
                if event.get("event") in ("peer_state", "channel_open", "command_result"):
109
                    print("Device:", json.dumps(event), flush=True)
110
 
111
    async def wait(self, predicate, timeout=25):
112
        deadline = time.monotonic() + timeout
113
        while time.monotonic() < deadline:
114
            self.poll()
115
            for index, event in enumerate(self.events):
116
                if predicate(event):
117
                    return self.events.pop(index)
118
            await asyncio.sleep(0.02)
119
        raise TimeoutError("Timed out waiting for device event")
120
 
121
    async def handshake(self):
122
        # Native USB can discard a command during the final port transition
123
        # after flashing. Require a reply before starting a new SFU session.
124
        for attempt in range(3):
125
            self.send(cmd="ping")
126
            try:
127
                return await self.wait(lambda e: e.get("cmd") == "ping" and e.get("result") == 0,
128
                    timeout=3 if attempt < 2 else 25)
129
            except TimeoutError:
130
                if attempt == 2:
131
                    raise
132
 
133
    def close(self):
134
        self.port.close()
135
        self.log.close()
136
 
137
 
138
async def main():
139
    api = SFU()
140
    device = Device()
141
    host = RTCPeerConnection(RTCConfiguration(iceServers=[]))
142
    registered = []
143
    result = {"s3_connected": False, "host_connected": False, "telemetry_received": False,
144
        "command_received": False, "led_driver_acknowledged": False}
145
    try:
146
        await device.handshake()
147
        device.send(cmd="peer_init", initiator=True)
148
        opened = await device.wait(lambda e: e.get("cmd") == "peer_init")
149
        if opened["result"]:
150
            raise RuntimeError(f"esp_peer_open failed: {opened['result']}")
151
        offer = await device.wait(lambda e: e.get("event") == "sdp")
152
        (OUT / "s3-offer.sdp").write_text(offer["text"])
153
        session = await api.call("/sessions/new", {
154
            "sessionDescription": {"type": "offer", "sdp": offer["text"]}})
155
        source = session["sessionId"]
156
        (OUT / "sfu-answer.sdp").write_text(session["sessionDescription"]["sdp"])
157
        device.send(cmd="sdp", text=session["sessionDescription"]["sdp"])
158
        await device.wait(lambda e: e.get("event") == "peer_state" and e.get("state") == 9)
159
        result["s3_connected"] = True
160
        print("S3 ICE, DTLS and SCTP connected to Cloudflare", flush=True)
161
 
162
        sink, transport = await api.establish()
163
        # The SFU opens server-events in-band using DCEP. Only application
164
        # channels returned by datachannels/new are externally negotiated.
165
        await host.setRemoteDescription(RTCSessionDescription(**transport["sessionDescription"]))
166
        await host.setLocalDescription(await host.createAnswer())
167
        await api.call(f"/sessions/{sink}/renegotiate", {
168
            "sessionDescription": {"type": "answer", "sdp": host.localDescription.sdp}}, method="PUT")
169
        for _ in range(1000):
170
            device.poll()
171
            if host.connectionState == "connected": break
172
            if host.connectionState in ("failed", "closed"): raise RuntimeError("Host WebRTC connection failed")
173
            await asyncio.sleep(0.02)
174
        if host.connectionState != "connected": raise TimeoutError("Host connection timeout")
175
        result["host_connected"] = True
176
        local = await api.call(f"/sessions/{source}/datachannels/new", {
177
            "dataChannels": [{"location": "local", "dataChannelName": "robot", "ordered": True}]})
178
        local_channel = local["dataChannels"][0]
179
        if local_channel.get("errorCode"): raise RuntimeError(local_channel["errorCode"])
180
        registered.append((source, local_channel["id"]))
181
        device.send(cmd="create_channel", label="robot", ordered=True)
182
        created = await device.wait(lambda e: e.get("cmd") == "create_channel")
183
        if created["result"] != 0:
184
            raise RuntimeError(f"Local DataChannel creation failed: {created['result']}")
185
        result["source_stream_id"] = local_channel["id"]
186
        remote = await api.call(f"/sessions/{sink}/datachannels/new", {
187
            "dataChannels": [{"location": "remote", "sessionId": source, "dataChannelName": "robot",
188
                "ordered": True, "waitForAck": True, "canReply": True}]})
189
        remote_channel = remote["dataChannels"][0]
190
        if remote_channel.get("errorCode"): raise RuntimeError(remote_channel["errorCode"])
191
        registered.append((sink, remote_channel["id"]))
192
        channel = host.createDataChannel("robot", negotiated=True, id=remote_channel["id"], ordered=True)
193
        incoming = asyncio.Queue()
194
        channel.on("message", incoming.put_nowait)
195
        for _ in range(500):
196
            if channel.readyState == "open": break
197
            await asyncio.sleep(0.02)
198
        if channel.readyState != "open": raise TimeoutError("Host DataChannel timeout")
199
        channel.send("ack")
200
        await asyncio.sleep(0.2)
201
        device.send(cmd="send", stream_id=local_channel["id"], text='{"sequence":1,"value":42}')
202
        sent = await device.wait(lambda e: e.get("cmd") == "send")
203
        result["esp_send_result"] = sent["result"]
204
        try:
205
            message = await asyncio.wait_for(incoming.get(), timeout=5)
206
            result["telemetry_received"] = json.loads(message).get("value") == 42
207
        except asyncio.TimeoutError:
208
            print("No telemetry arrived; testing reverse traffic independently", flush=True)
209
        channel.send('{"led":[16,0,16],"command_id":1}')
210
        try:
211
            reply = await device.wait(lambda e: e.get("event") == "data_received", timeout=8)
212
            result["command_received"] = json.loads(reply.get("text", "{}" )).get("command_id") == 1
213
            led = await device.wait(lambda e: e.get("event") == "led" and e.get("r") == 16, timeout=2)
214
            result["led_driver_acknowledged"] = led.get("result") == 0
215
        except TimeoutError:
216
            pass
217
    except Exception as error:
218
        result["error"] = f"{type(error).__name__}: {error}"
219
    finally:
220
        for sid, channel_id in reversed(registered):
221
            try:
222
                await api.call(f"/sessions/{sid}/datachannels/close", {"dataChannels": [{"id": channel_id}]}, method="PUT")
223
            except Exception:
224
                result["cleanup_incomplete"] = True
225
        await host.close()
226
        try:
227
            device.send(cmd="peer_close")
228
            closed = await device.wait(lambda e: e.get("cmd") == "peer_close", timeout=5)
229
            result["peer_close_acknowledged"] = closed.get("result") == 0
230
        except Exception:
231
            result["peer_close_acknowledged"] = False
232
        device.close()
233
        (OUT / "sfu-hardware-result.json").write_text(json.dumps(result, indent=2) + "\n")
234
        print(json.dumps(result, indent=2), flush=True)
235
 
236
 
237
if __name__ == "__main__":
238
    asyncio.run(main())