1
"""Shared setup for archived experiments; never flashes or contacts the SFU."""
2
import json
3
import os
4
from pathlib import Path
5
import subprocess
6
import sys
7
 
8
ROOT = Path(__file__).resolve().parents[1]
9
PINS = json.loads((Path(__file__).parent / "dependencies.json").read_text())
10
sys.path.insert(0, str(ROOT / "scripts"))
11
from project import idf_environment, sdk_root, verify_checkout
12
from config import read_env
13
 
14
 
15
def build_experiment(source, name, components, signaling_url=None, *, rust=False):
16
    sdk = sdk_root()
17
    verify_checkout(sdk / "esp-idf", PINS["esp-idf"])
18
    for component in components:
19
        verify_checkout(ROOT / ".tools/vendor" / component, PINS[component])
20
    if signaling_url:
21
        if not signaling_url.startswith("https://") or len(signaling_url) > 180:
22
            raise SystemExit("The signaling URL must use HTTPS and be at most 180 characters")
23
    if rust:
24
        version = subprocess.check_output(
25
            ["rustup", "run", "esp-radio", "rustc", "--version"], text=True)
26
        if f"({PINS['rust-release']})" not in version:
27
            raise SystemExit("Use the archived Xtensa Rust release from archive/dependencies.json")
28
 
29
    build = sdk / f"archive-{name}-build"
30
    private = build / "private"
31
    private.mkdir(parents=True, exist_ok=True)
32
    private.chmod(0o700)
33
    values = read_env(ROOT / ".credential.env")
34
    ssid = values.get("WIFI_SSID", "").encode()
35
    password = values.get("WIFI_PASSWORD", "").encode()
36
    if not 1 <= len(ssid) <= 32 or len(password) > 64:
37
        raise SystemExit("Check WIFI_SSID and WIFI_PASSWORD in .credential.env")
38
 
39
    def literal(value):
40
        return '"' + ''.join(f"\\x{byte:02x}" for byte in value) + '"'
41
 
42
    os.umask(0o077)
43
    header = private / "wifi_private.h"
44
    header.write_text("#pragma once\n" +
45
                      f"#define WIFI_SSID {literal(ssid)}\n" +
46
                      f"#define WIFI_PASSWORD {literal(password)}\n")
47
    header.chmod(0o600)
48
    token = ""
49
    if signaling_url:
50
        token = read_env(ROOT / "worker/.dev.vars").get("DEVICE_TOKEN", "")
51
        if not 32 <= len(token) <= 128:
52
            raise SystemExit("Run make secrets before building a streaming experiment")
53
    header = private / "signaling_private.h"
54
    header.write_text("#pragma once\n" +
55
                      f"#define SIGNALING_AUTONOMOUS {int(signaling_url is not None)}\n" +
56
                      f"#define SIGNALING_URL {literal((signaling_url or '').rstrip('/').encode())}\n" +
57
                      f"#define SIGNALING_DEVICE_TOKEN {literal(token.encode())}\n")
58
    header.chmod(0o600)
59
    python, env = idf_environment()
60
    return subprocess.run(
61
        [str(python), str(sdk / "esp-idf/tools/idf.py"), "-B", str(build), "build"],
62
        cwd=source, env=env).returncode