diff --git a/python/README.md b/python/README.md new file mode 100644 index 0000000..fe5ba4e --- /dev/null +++ b/python/README.md @@ -0,0 +1,94 @@ +# rockblock9704-tools: the RockBLOCK 9704 from a computer + +Two commands and a test suite, in one Python package: + +- `rb9704`, a command line for a RockBLOCK 9704 plugged into USB and for + the Cloudloop account behind it, for proving the modem, the account and + the delivery path with no firmware in the loop; +- `rb9704-fake`, a fake modem that speaks JSPR on a pty or a serial port, + for testing anything that drives a 9704 without one; +- `tests/`, which run Ground Control's own Python binding, Zephyr on + `native_sim`, and Zephyr on a board against the fake + (`../docs/TESTING.md`). + +`../docs/CLOUDLOOP.md` is the account, the provisioning and the order to +test things in; this is the tools. + +## Install + +```sh +python/install.sh # python/.venv, this package, rockblock9704, pytest +python/install.sh --udev # ... and keep ModemManager off the port (sudo) +``` + +Or by hand: `pip install -e 'python[modem,test]'`. Python 3.10 to 3.14, +which is where `rockblock9704` has wheels (Linux x86_64 and aarch64, +macOS, Windows). Without the `modem` extra, the fake modem, `jspr`, +`provision` and the cloud verbs work and the library verbs do not. + +## Verbs + +| Verb | Costs | What it does | +| --- | --- | --- | +| `ports` | | USB serial ports, the FT234XD (`0403:6015`) first | +| `info [--json]` | | hardware and firmware, serial, IMEI, ICCID, SIM, temperature, signal | +| `signal [--watch S] [--every S]` | | signal bars 0 to 5, once or for S seconds with a summary | +| `provision [--timeout S]` | | the setup `begin` does, then signal and provisioning every 30 s until topics appear | +| `jspr [--setup] [LINE ...]` | | raw JSPR: the lines given, or a prompt; every answer and unsolicited line printed | +| `send --text/--hex/--file [--topic N]` | airtime | one MO, blocking, through the library | +| `listen [--out DIR] [--for S]` | | MT as it arrives, each written to `DIR/MT-.bin` | +| `session --text/--hex/--file [--quiet S]` | airtime | one MO, then MT until S seconds pass with nothing new | +| `cloud ping` | | the token works | +| `cloud things` | | every Thing on the account, with its Certus subscriber id | +| `cloud mo [--thing ID] [--hours H]` | | message records for a Thing over the last H hours | +| `cloud msg ID` | | one message record and its Lingo form | +| `cloud mt --text/--hex/--file [--thing ID]` | airtime | queue an MT on `IMT_TOPIC_RAW` | +| `cloud status ID` | | an MT's delivery status | + +The verbs that cost airtime print the size and ask; `--yes` skips the +question and is the only way to send from a script. + +Options that go with any verb, before or after it: + +- `-p PORT`: the serial port. Otherwise `$RB9704_PORT`, otherwise the one + 9704 bridge attached (`0403:6015`); two attached is an error. +- `--log FILE`: append one JSON line per event (`{"t": ..., "ev": ...}`, + UTC), for the bench log. + +The cloud verbs read the token from `$CLOUDLOOP_TOKEN` or +`~/.config/rb9704/cloudloop.token`, and the Thing from `--thing` or +`$CLOUDLOOP_THING`. + +## The fake modem + +```sh +rb9704-fake # a pty; prints READY /dev/pts/N +rb9704-fake --port /dev/ttyUSB0 # a real port, e.g. an adapter wired to a board +rb9704-fake --topics 244,313 # the provisioning it reports +``` + +It answers everything Ground Control's library asks (`../docs/JSPR.md`), +takes MO messages segment by segment and checks their CRC, and on a +command on stdin sends an MT, a signal change, a refusal or a failed MO. +The commands and the events it prints are in `rb9704tools/fakemodem.py`'s +docstring. Run it in a process of its own beside the library's Python +binding, which holds the GIL through every blocking call. + +## How it is built + +- The modem verbs (`info`, `signal`, `send`, `listen`, `session`) use + Ground Control's `rockblock9704`, a binding of the C library this + repository builds into Zephyr. `listen` and `session` use its + callback API, polled every 10 ms as the library asks, with the + callbacks only recording and the loop doing the work. +- `provision` and `jspr` speak JSPR over pyserial, because the library + asks for the provisioning only inside a send. `Jspr.setup` repeats + `rbBegin`'s sequence: an active API version, the internal SIM, the + state set active. JSPR lines are ` ` ending in CR; + 299 is unsolicited. +- The cloud verbs are plain `urllib` POSTs to `api.cloudloop.com`, every + argument in the query string as the API's Swagger + () describes it. + +Only one program can hold the port: `jspr` and the library verbs cannot +run at once. diff --git a/python/install.sh b/python/install.sh new file mode 100755 index 0000000..a54bf2e --- /dev/null +++ b/python/install.sh @@ -0,0 +1,104 @@ +#!/usr/bin/env bash +# +# install.sh - set up the RockBLOCK 9704 host tools. +# +# Makes python/.venv, installs this package into it with Ground Control's +# rockblock9704 binding and pytest (`pip install -e .[modem,test]`), checks +# the imports, and checks the two things that stop a USB serial port +# opening on Linux: the dialout group, and ModemManager probing the port. +# +# Usage: +# python/install.sh the venv and the checks, no root +# python/install.sh --udev ... and a udev rule (sudo) that keeps +# ModemManager off the 9704 and names it +# /dev/rockblock9704 +# python/install.sh --fresh delete the venv and start again +# +# Then python/.venv/bin/rb9704, python/.venv/bin/rb9704-fake, and +# python/.venv/bin/pytest (docs/TESTING.md). +# +# Knobs: +# PYTHON the interpreter for the venv, default python3 (3.10 to 3.14; +# rockblock9704 ships wheels for those and nothing else) +set -e + +HERE="$(cd "$(dirname "$0")" && pwd)" +VENV="$HERE/.venv" +PY="${PYTHON:-python3}" +UDEV="" +FRESH="" + +for arg in "$@"; do + case "$arg" in + --udev) UDEV=1 ;; + --fresh) FRESH=1 ;; + -h|--help) sed -n '2,/^set -e/p' "$0" | sed 's/^# \{0,1\}//; /^set -e/d'; exit 0 ;; + *) echo "!! unknown option $arg"; exit 1 ;; + esac +done + +command -v "$PY" >/dev/null 2>&1 || { echo "!! $PY not found"; exit 1; } +"$PY" -c 'import sys; sys.exit(not ((3, 10) <= sys.version_info[:2] <= (3, 14)))' || { + echo "!! $("$PY" --version) is outside 3.10 to 3.14, where rockblock9704 has wheels." + echo " Set PYTHON to another interpreter." + exit 1 +} +"$PY" -c 'import venv, ensurepip' 2>/dev/null || { + echo "!! $PY has no venv module. On Debian and Ubuntu: sudo apt install python3-venv" + exit 1 +} + +if [ -n "$FRESH" ] && [ -d "$VENV" ]; then + echo "== removing $VENV" + rm -rf "$VENV" +fi +if [ ! -x "$VENV/bin/python" ]; then + echo "== making $VENV with $("$PY" --version)" + "$PY" -m venv "$VENV" +fi +echo "== installing rockblock9704-tools with the binding and pytest" +"$VENV/bin/python" -m pip install --quiet --upgrade pip +"$VENV/bin/python" -m pip install --quiet -e "$HERE[modem,test]" +"$VENV/bin/python" - <<'EOF' +import importlib.metadata as md +import rockblock9704, serial +print(f" rockblock9704 {md.version('rockblock9704')}, pyserial {serial.__version__}") +EOF + +echo "== checks" +if [ "$(uname)" = Linux ]; then + if id -nG | tr ' ' '\n' | grep -qx dialout; then + echo " ok: $(id -un) is in dialout" + else + echo "!! $(id -un) is not in dialout, so /dev/ttyUSB* will not open." + echo " sudo usermod -aG dialout $(id -un), then log out and in again." + fi + if systemctl is-active --quiet ModemManager 2>/dev/null; then + if [ -f /etc/udev/rules.d/99-rockblock9704.rules ]; then + echo " ok: ModemManager runs, and the udev rule keeps it off the 9704" + elif [ -z "$UDEV" ]; then + echo "!! ModemManager is running and may open the 9704's port for a" + echo " few seconds after it is plugged in. Run this with --udev to stop" + echo " it, or wait 30 s after plugging in before the first command." + fi + fi +fi + +if [ -n "$UDEV" ]; then + RULE=/etc/udev/rules.d/99-rockblock9704.rules + echo "== writing $RULE (sudo)" + sudo tee "$RULE" >/dev/null <<'EOF' +# RockBLOCK 9704: FTDI FT234XD on the USB-C port. Keep ModemManager from +# probing it, and give it a stable name. Written by rockblock9704-zephyr's +# python/install.sh. +SUBSYSTEM=="tty", ATTRS{idVendor}=="0403", ATTRS{idProduct}=="6015", ENV{ID_MM_DEVICE_IGNORE}="1", SYMLINK+="rockblock9704", MODE="0660", GROUP="dialout" +EOF + sudo udevadm control --reload-rules + sudo udevadm trigger --subsystem-match=tty + echo " done; unplug and replug the 9704 if it was attached" + echo " (this matches every FT234XD, not only the 9704's)" +fi + +echo +echo "Run it as $VENV/bin/rb9704, or put $VENV/bin on the PATH." +echo "First command: rb9704 ports" diff --git a/python/pyproject.toml b/python/pyproject.toml new file mode 100644 index 0000000..d47465d --- /dev/null +++ b/python/pyproject.toml @@ -0,0 +1,29 @@ +[build-system] +requires = ["setuptools>=64"] +build-backend = "setuptools.build_meta" + +[project] +name = "rockblock9704-tools" +version = "0.1.0" +description = "Host tools for the RockBLOCK 9704: a command line for the modem and Cloudloop, and a fake modem for testing" +readme = "README.md" +license = { text = "MIT" } +requires-python = ">=3.10" +dependencies = ["pyserial>=3.5"] + +[project.optional-dependencies] +# Ground Control's binding of the C library, for the verbs that use it. +# Pinned with the C library in ../west.yml (v1.0.1; 1.0.2 is the first +# Python release after it). +modem = ["rockblock9704==1.0.2"] +test = ["pytest>=7"] + +[project.scripts] +rb9704 = "rb9704tools.cli:main" +rb9704-fake = "rb9704tools.fakemodem:main" + +[tool.setuptools] +packages = ["rb9704tools"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/python/rb9704tools/__init__.py b/python/rb9704tools/__init__.py new file mode 100644 index 0000000..190daf3 --- /dev/null +++ b/python/rb9704tools/__init__.py @@ -0,0 +1,3 @@ +"""Host tools for the RockBLOCK 9704.""" + +__version__ = "0.1.0" diff --git a/python/rb9704tools/cli.py b/python/rb9704tools/cli.py new file mode 100644 index 0000000..4bc270b --- /dev/null +++ b/python/rb9704tools/cli.py @@ -0,0 +1,644 @@ +""" +rb9704 - drive a RockBLOCK 9704 from the host over its USB-C port. + + rb9704 ports + rb9704 [-p PORT] info + rb9704 [-p PORT] signal [--watch S] [--every S] + rb9704 [-p PORT] provision [--timeout S] + rb9704 [-p PORT] jspr [LINE ...] + rb9704 [-p PORT] send (--text T | --hex H | --file F) [--topic N] [--yes] + rb9704 [-p PORT] listen [--out DIR] [--for S] + rb9704 [-p PORT] session (--text T | --hex H | --file F) [--quiet S] [--yes] + rb9704 cloud ping | things | mo | msg ID | mt ... | status ID + +The modem side uses Ground Control's rockblock9704 package, a binding of +the same C library this repository builds into Zephyr, so what works here +is the library the firmware runs. `provision` and `jspr` speak +JSPR directly over pyserial instead, because the library only asks for +the provisioning when a message is sent, and asking that way is free. + +`send` and `session` cost airtime and ask before sending; --yes skips the +question. Everything else on the modem side is local serial traffic and +costs nothing. `cloud mt` costs airtime too. + +The port is -p, or $RB9704_PORT, or the one FTDI FT234XD attached (the +9704's USB-C bridge, 0403:6015). --log FILE appends every event as a JSON +line, for the bench log. + +The cloud verbs talk to the Cloudloop Data API (api.cloudloop.com). The +token is $CLOUDLOOP_TOKEN or the first line of ~/.config/rb9704/cloudloop.token; +it never goes in a repository. docs/CLOUDLOOP.md says how to get one. +""" +import argparse +import base64 +import datetime as dt +import json +import os +import sys +import time +import urllib.parse +import urllib.request + +FTDI_VID, FT234X_PID = 0x0403, 0x6015 +BAUD = 230400 +RAW_TOPIC = 244 +CLOUDLOOP = "https://api.cloudloop.com" +TOKEN_FILE = os.path.expanduser("~/.config/rb9704/cloudloop.token") + +LOG = None + + +def now(): + return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" + + +def ev(kind, **kw): + """Print an event and append it to --log as one JSON line.""" + rec = {"t": now(), "ev": kind, **kw} + shown = " ".join(f"{k}={v}" for k, v in kw.items()) + print(f"{rec['t']} {kind:<10} {shown}", flush=True) + if LOG: + with open(LOG, "a") as f: + f.write(json.dumps(rec) + "\n") + + +def die(msg, code=1): + print(f"!! {msg}", file=sys.stderr) + sys.exit(code) + + +# ---------------------------------------------------------------- ports + +def find_ports(): + """Every serial port, with the ones that look like a 9704 first.""" + try: + from serial.tools import list_ports + except ImportError: + die("pyserial is missing; run host/rockblock/install.sh") + out = [] + for p in list_ports.comports(): + is_rb = p.vid == FTDI_VID and p.pid == FT234X_PID + out.append((is_rb, p)) + out.sort(key=lambda x: (not x[0], x[1].device)) + return out + + +def pick_port(arg): + if arg: + return arg + if os.environ.get("RB9704_PORT"): + return os.environ["RB9704_PORT"] + rbs = [p for is_rb, p in find_ports() if is_rb] + if len(rbs) == 1: + return rbs[0].device + if not rbs: + die("no FTDI FT234XD attached; plug the 9704's USB-C in, or give -p PORT " + "(rb9704.py ports lists them)") + die("more than one FT234XD attached (" + ", ".join(p.device for p in rbs) + + "); give -p PORT") + + +def cmd_ports(a): + # the motherboard's ttyS ports have no USB ids and are never the 9704 + ports = [x for x in find_ports() if a.all or x[1].vid is not None] + if not ports: + print("no serial ports") + return + for is_rb, p in ports: + ids = f"{p.vid:04x}:{p.pid:04x}" if p.vid is not None else "-" + mark = "9704?" if is_rb else "" + print(f"{p.device:<16} {ids:<10} {mark:<6} {p.serial_number or '':<12} {p.description}") + if os.path.isdir("/dev/serial/by-id"): + for n in sorted(os.listdir("/dev/serial/by-id")): + if "FT234X" in n: + print(f" by-id: /dev/serial/by-id/{n}") + + +# ---------------------------------------------------------------- library + +def open_rb(port): + try: + from rockblock9704 import RockBlock9704 + except ImportError: + die("rockblock9704 is missing; run host/rockblock/install.sh") + rb = RockBlock9704() + if not rb.begin(port): + die(f"begin({port}) failed: wrong port, modem not booted (green RDY " + "LED), or another program holds the port (ModemManager, a " + "terminal). docs/CLOUDLOOP.md, \"When begin fails\"") + return rb + + +def cmd_info(a): + port = pick_port(a.port) + rb = open_rb(port) + try: + info = { + "port": port, + "hardware": rb.get_hardware_version(), + "firmware": rb.get_firmware_version(), + "serial": rb.get_serial_number(), + "imei": rb.get_imei(), + "iccid": rb.get_iccid(), + "card_present": rb.get_card_present(), + "sim_connected": rb.get_sim_connected(), + "board_temp_c": rb.get_board_temp(), + "signal": rb.get_signal(), + } + finally: + rb.end() + if a.json: + print(json.dumps(info, indent=2)) + else: + for k, v in info.items(): + print(f"{k:<14} {v}") + if LOG: + ev("info", **info) + + +def cmd_signal(a): + port = pick_port(a.port) + rb = open_rb(port) + try: + if not a.watch: + ev("signal", bars=rb.get_signal()) + return + end = time.monotonic() + a.watch + seen = {} + while time.monotonic() < end: + bars = rb.get_signal() + seen[bars] = seen.get(bars, 0) + 1 + ev("signal", bars=bars) + time.sleep(a.every) + total = sum(seen.values()) + ev("summary", samples=total, + **{f"bars{k}": f"{100 * v / total:.0f}%" for k, v in sorted(seen.items())}) + except KeyboardInterrupt: + pass + finally: + rb.end() + + +def payload(a): + if a.text is not None: + return a.text.encode() + if a.hex is not None: + return bytes.fromhex(a.hex) + with open(a.file, "rb") as f: + return f.read() + + +def confirm(data, topic, yes): + print(f"AIRTIME: {len(data)} bytes to topic {topic}. This costs money.", flush=True) + if yes: + return + if not sys.stdin.isatty(): + die("not a terminal; pass --yes to send") + if input("send? [y/N] ").strip().lower() not in ("y", "yes"): + die("not sent", 0) + + +def cmd_send(a): + data = payload(a) + port = pick_port(a.port) + confirm(data, a.topic, a.yes) + rb = open_rb(port) + try: + ev("signal", bars=rb.get_signal()) + t0 = time.monotonic() + ok = rb.send_message(data, topic=None if a.topic == RAW_TOPIC else a.topic, + timeout=a.timeout) + secs = time.monotonic() - t0 + ev("mo", ok=ok, bytes=len(data), topic=a.topic, secs=f"{secs:.1f}") + if not ok and secs < 2: + # the library checks provisioning before it queues anything, and + # gives up at once if the topic is not there: nothing was sent + print(f" refused before transmitting: the modem holds no " + f"provisioning for topic {a.topic}. Run provision under sky.") + finally: + rb.end() + sys.exit(0 if ok else 2) + + +class Async: + """The library's callback API, polled from one loop. The callbacks only + record; all the work is done in the loop, between polls.""" + + def __init__(self, port): + self.rb = None + self.mo_done = None + self.mo_id = None + self.mt_ready = 0 + self.bars = None + self.port = port + + def start(self): + from rockblock9704 import RockBlock9704 + rb = RockBlock9704() + rb.set_message_provisioning_callback(self._prov) + rb.set_mo_message_started_callback(self._mo_started) + rb.set_mo_message_complete_callback(self._mo) + rb.set_mt_message_complete_callback(self._mt) + rb.set_constellation_state_callback(self._sig) + if not rb.begin(self.port): + die(f"begin({self.port}) failed") + time.sleep(0.1) # the library asks for 100 ms before the first queue + self.rb = rb + return rb + + def _prov(self, p): + ev("prov", set=p["provisioningSet"], topics=",".join( + f"{t['topicId']}:{t['topicName']}" for t in p["provisioning"])) + + def _mo_started(self, mid): + self.mo_id = mid + ev("mo_start", id=mid) + + def _mo(self, mid, status): + self.mo_done = status == 1 + ev("mo", id=mid, ok=self.mo_done) + + def _mt(self, mid, status): + ev("mt_done", id=mid, ok=status == 1) + if status == 1: + self.mt_ready += 1 + + def _sig(self, s): + if s["signalBars"] != self.bars: + self.bars = s["signalBars"] + ev("signal", bars=self.bars, level=s["signalLevel"], + visible=s["constellationVisible"]) + + def drain(self, out): + """Collect every completed MT. Returns how many.""" + n = 0 + while self.mt_ready: + msg = self.rb.receive_message_async() + if msg is None: + break + self.rb.acknowledge_receive_head_async() + self.mt_ready -= 1 + n += 1 + name = None + if out: + os.makedirs(out, exist_ok=True) + name = os.path.join(out, dt.datetime.now(dt.timezone.utc) + .strftime("MT-%Y%m%dT%H%M%S.%fZ.bin")) + with open(name, "wb") as f: + f.write(msg) + ev("mt", bytes=len(msg), head=msg[:24].hex(), file=name or "-") + return n + + +def cmd_listen(a): + s = Async(pick_port(a.port)) + rb = s.start() + end = time.monotonic() + a.for_ if a.for_ else None + got = 0 + ev("listen", port=s.port, out=a.out or "-") + try: + while end is None or time.monotonic() < end: + rb.poll() + got += s.drain(a.out) + time.sleep(0.01) + except KeyboardInterrupt: + pass + finally: + rb.end() + ev("summary", mt=got) + + +def cmd_session(a): + """One MO, then listen until a.quiet seconds pass with no new MT. This is + the 9704 session designed in docs/PLAN.md, "The session", run from the + host so the quiet window can be chosen from what is seen.""" + data = payload(a) + port = pick_port(a.port) + confirm(data, a.topic, a.yes) + s = Async(port) + rb = s.start() + got = 0 + try: + if not rb.send_message_async(data, a.topic): + die("the modem would not queue the message (not provisioned for " + f"topic {a.topic}? run provision)") + ev("mo_queue", bytes=len(data), topic=a.topic) + t0 = time.monotonic() + while s.mo_done is None: + rb.poll() + got += s.drain(a.out) + if time.monotonic() - t0 > a.timeout: + if s.mo_id is not None: + rb.cancel_message(a.topic, s.mo_id) + ev("mo", ok=False, why="timeout", secs=a.timeout) + break + time.sleep(0.01) + ev("mo_secs", secs=f"{time.monotonic() - t0:.1f}") + last = time.monotonic() + while time.monotonic() - last < a.quiet: + rb.poll() + n = s.drain(a.out) + if n: + got += n + last = time.monotonic() + time.sleep(0.01) + except KeyboardInterrupt: + pass + finally: + rb.end() + ev("summary", mo_ok=s.mo_done, mt=got, quiet=a.quiet) + sys.exit(0 if s.mo_done else 2) + + +# ---------------------------------------------------------------- raw JSPR + +class Jspr: + """JSPR by hand: one line out, ` ` lines back, all + ending in CR. 299 is unsolicited. The same setup as the library's + rbBegin: an active API version, the internal SIM, state active.""" + + def __init__(self, port): + import serial + self.s = serial.Serial(port, BAUD, timeout=0.1) + self.buf = b"" + self.s.reset_input_buffer() + + def close(self): + self.s.close() + + def lines(self, secs): + end = time.monotonic() + secs + while time.monotonic() < end: + self.buf += self.s.read(4096) + while b"\r" in self.buf: + line, self.buf = self.buf.split(b"\r", 1) + # bootInfo can arrive behind a DC1; the library strips it too + line = line.lstrip(b"\x11\n ").decode(errors="replace") + if len(line) >= 4: + yield parse(line) + + def cmd(self, line, target=None, secs=3.0, show=False): + """Send a line, return the first response for its target. Anything + else that arrives meanwhile is shown as it goes past.""" + target = target or line.split()[1] + self.s.write((line + "\r").encode()) + for code, tgt, js, raw in self.lines(secs): + if show or tgt != target or code == 299: + print(f" < {raw}") + if tgt == target and code != 299: + return code, js + return None, None + + def setup(self): + code, js = self.cmd("GET apiVersion {}") + if code != 200: + die(f"GET apiVersion answered {code}: is this a 9704 at {BAUD}?") + if not js.get("active_version"): + v = js["supported_versions"][0] + self.cmd("PUT apiVersion " + json.dumps({"active_version": v})) + code, js = self.cmd("GET simConfig {}") + if js.get("interface") != "internal": + self.cmd('PUT simConfig {"interface": "internal"}') + code, js = self.cmd("GET operationalState {}") + st = js.get("state") + if st != "active": + if st != "inactive": + self.cmd('PUT operationalState {"state": "inactive"}') + self.cmd('PUT operationalState {"state": "active"}') + + +def parse(line): + code, _, rest = line.partition(" ") + tgt, _, js = rest.partition(" ") + try: + return int(code), tgt, json.loads(js) if js else {}, line + except ValueError: + return 0, "", {}, line + + +def cmd_provision(a): + port = pick_port(a.port) + j = Jspr(port) + t0 = time.monotonic() + try: + j.setup() + while True: + code, sig = j.cmd("GET constellationState {}") + code, prov = j.cmd("GET messageProvisioning {}") + topics = (prov or {}).get("provisioning") or [] + ev("prov", code=code, set=bool(topics), + bars=(sig or {}).get("signal_bars"), + visible=(sig or {}).get("constellation_visible"), + topics=",".join(f"{t.get('topic_id')}:{t.get('topic_name')}" + for t in topics) or "-") + if topics: + print(json.dumps(prov, indent=2)) + return + if time.monotonic() - t0 > a.timeout: + die(f"not provisioned after {a.timeout} s; a clear sky view " + "for 10 to 30 minutes is what it needs", 2) + for _ in j.lines(a.every): + pass + except KeyboardInterrupt: + pass + finally: + j.close() + + +def cmd_jspr(a): + """Lines from the arguments, or typed, one at a time. Responses and + anything unsolicited are printed as they arrive.""" + j = Jspr(pick_port(a.port)) + try: + if a.setup: + j.setup() + if a.line: + for line in a.line: + print(f" > {line}") + j.cmd(line, show=True, secs=a.wait) + return + print("JSPR at %d. A line such as: GET hwInfo {} (ctrl-d to quit)" % BAUD) + while True: + try: + line = input("> ").strip() + except EOFError: + return + if line: + j.cmd(line, show=True, secs=a.wait) + else: + for _, _, _, raw in j.lines(1.0): + print(f" < {raw}") + except KeyboardInterrupt: + pass + finally: + j.close() + + +# ---------------------------------------------------------------- Cloudloop + +def token(): + t = os.environ.get("CLOUDLOOP_TOKEN") + if not t and os.path.exists(TOKEN_FILE): + with open(TOKEN_FILE) as f: + t = f.readline().strip() + if not t: + die(f"no Cloudloop token: set CLOUDLOOP_TOKEN or write {TOKEN_FILE} " + "(docs/CLOUDLOOP.md, \"The API token\")") + return t + + +def api(path, **params): + """Every Cloudloop call is a POST with its arguments, the token among + them, in the query string.""" + q = urllib.parse.urlencode({"token": token(), **params}) + req = urllib.request.Request(f"{CLOUDLOOP}{path}?{q}", data=b"", method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + body = json.load(r) + for k in ("DispatcherException", "DispatcherValidationException"): + if k in body: + die(f"{path}: {json.dumps(body[k])}") + return body + + +def thing(a): + t = a.thing or os.environ.get("CLOUDLOOP_THING") + if not t: + die("which thing? --thing ID or $CLOUDLOOP_THING (cloud things lists them)") + return t + + +def stamp(d): + return d.strftime("%Y-%m-%d %H:%M:%S") + + +def cmd_cloud(a): + if a.verb == "ping": + print(json.dumps(api("/Platform/Ping"))) + elif a.verb == "things": + for t in api("/Data/GetThings").get("things", []): + print(f"{t.get('id')} certus={t.get('subscriberCertus') or '-'} " + f"sbd={t.get('subscriberSbd') or '-'} {t.get('name', '')}") + elif a.verb == "mo": + to = dt.datetime.now(dt.timezone.utc) + frm = to - dt.timedelta(hours=a.hours) + recs = api("/Data/GetMessageRecordsForThing", thing=thing(a), + **{"from": stamp(frm), "to": stamp(to)}).get("messageRecords", []) + for r in recs: + if a.json: + print(json.dumps(r)) + else: + print(f"{r.get('id')} {r.get('receivedAt', r.get('createdAt', ''))} " + f"{r.get('type')} {r.get('direction', '')} " + f"{(r.get('snippet') or '')[:40]}") + print(f"{len(recs)} record(s) in the last {a.hours} h", file=sys.stderr) + elif a.verb == "msg": + print(json.dumps(api("/Data/GetMessageRecord", messageRecord=a.id), indent=2)) + print(json.dumps(api("/Data/GetLingoMo", messageRecord=a.id), indent=2)) + elif a.verb == "status": + print(json.dumps(api("/Data/GetMtDeliveryStatus", message=a.id))) + elif a.verb == "mt": + data = payload(a) + token() + confirm(data, a.topic, a.yes) + r = api("/Data/DoSendImtMessage", thing=thing(a), topic=a.topic, + ringStyle=a.ring, message=base64.b64encode(data).decode()) + ev("cloud_mt", bytes=len(data), topic=a.topic, result=json.dumps(r)) + + +# ---------------------------------------------------------------- main + +def add_payload(p, required=True): + g = p.add_mutually_exclusive_group(required=required) + g.add_argument("--text", help="the message as UTF-8 text") + g.add_argument("--hex", help="the message as hex bytes") + g.add_argument("--file", help="the message is this file's bytes") + p.add_argument("--yes", action="store_true", help="send without asking") + + +def main(): + global LOG + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0].strip(), + formatter_class=argparse.RawDescriptionHelpFormatter) + # -p and --log go before or after the verb; SUPPRESS keeps a verb's + # default from hiding a value given before it + common = argparse.ArgumentParser(add_help=False) + common.add_argument("-p", "--port", default=argparse.SUPPRESS, + help="serial port (default: $RB9704_PORT or the FT234XD)") + common.add_argument("--log", default=argparse.SUPPRESS, + help="append every event to this file as JSON lines") + ap.add_argument("-p", "--port", help=argparse.SUPPRESS) + ap.add_argument("--log", help=argparse.SUPPRESS) + sub = ap.add_subparsers(dest="cmd", required=True) + real_add = sub.add_parser + sub.add_parser = lambda *x, **k: real_add(*x, parents=[common], **k) + + p = sub.add_parser("ports", help="list USB serial ports, the 9704 first") + p.add_argument("--all", action="store_true", help="the ttyS ports too") + p.set_defaults(fn=cmd_ports) + + p = sub.add_parser("info", help="identity, SIM and signal (free)") + p.add_argument("--json", action="store_true") + p.set_defaults(fn=cmd_info) + + p = sub.add_parser("signal", help="signal bars 0-5 (free)") + p.add_argument("--watch", type=float, default=0, help="keep reading for S seconds") + p.add_argument("--every", type=float, default=5, help="seconds between reads") + p.set_defaults(fn=cmd_signal) + + p = sub.add_parser("provision", help="wait until the modem holds its topics (free)") + p.add_argument("--timeout", type=float, default=3600) + p.add_argument("--every", type=float, default=30) + p.set_defaults(fn=cmd_provision) + + p = sub.add_parser("jspr", help="raw JSPR lines, typed or given") + p.add_argument("line", nargs="*") + p.add_argument("--setup", action="store_true", + help="set the API version, SIM and state first, as begin does") + p.add_argument("--wait", type=float, default=3, help="seconds to wait for each answer") + p.set_defaults(fn=cmd_jspr) + + p = sub.add_parser("send", help="one MO message, blocking (COSTS AIRTIME)") + add_payload(p) + p.add_argument("--topic", type=int, default=RAW_TOPIC) + p.add_argument("--timeout", type=int, default=120) + p.set_defaults(fn=cmd_send) + + p = sub.add_parser("listen", help="collect MT messages as they arrive (free)") + p.add_argument("--out", help="write each MT here as MT-.bin") + p.add_argument("--for", dest="for_", type=float, default=0, help="stop after S seconds") + p.set_defaults(fn=cmd_listen) + + p = sub.add_parser("session", help="one MO, then MT until a quiet window (COSTS AIRTIME)") + add_payload(p) + p.add_argument("--topic", type=int, default=RAW_TOPIC) + p.add_argument("--timeout", type=int, default=180, help="give up on the MO after S seconds") + p.add_argument("--quiet", type=float, default=60, help="end after S seconds with no new MT") + p.add_argument("--out", help="write each MT here as MT-.bin") + p.set_defaults(fn=cmd_session) + + p = sub.add_parser("cloud", help="the Cloudloop Data API") + cs = p.add_subparsers(dest="verb", required=True) + cs.add_parser("ping") + cs.add_parser("things") + q = cs.add_parser("mo", help="message records for a thing") + q.add_argument("--thing") + q.add_argument("--hours", type=float, default=24) + q.add_argument("--json", action="store_true") + q = cs.add_parser("msg", help="one message record and its Lingo form") + q.add_argument("id") + q = cs.add_parser("status", help="delivery status of an MT") + q.add_argument("id") + q = cs.add_parser("mt", help="queue an MT to the modem (COSTS AIRTIME)") + q.add_argument("--thing") + q.add_argument("--topic", default="IMT_TOPIC_RAW") + q.add_argument("--ring", default="RING_STYLE_NORMAL") + add_payload(q) + p.set_defaults(fn=cmd_cloud) + + a = ap.parse_args() + LOG = a.log + a.fn(a) + + +if __name__ == "__main__": + main() diff --git a/python/rb9704tools/fakemodem.py b/python/rb9704tools/fakemodem.py new file mode 100644 index 0000000..290ac98 --- /dev/null +++ b/python/rb9704tools/fakemodem.py @@ -0,0 +1,381 @@ +""" +rb9704-fake - a fake RockBLOCK 9704: JSPR over a serial port, no satellites. + + rb9704-fake make a pty, print its path, serve it + rb9704-fake --port /dev/ttyUSB0 serve a real serial port instead (a + USB-serial adapter wired to a board's + modem pins) + +It answers what Ground Control's library asks (docs/JSPR.md is the +protocol, taken from the library's source), takes MO messages segment by +segment and checks their CRC, and sends MT messages, signal and +provisioning when told to. One fake serves three kinds of test: the +library's own Python binding on the host, Zephyr on native_sim, and Zephyr +on a board through a USB-serial adapter. + +Control is one command per line on stdin, events one per line on stdout: + + mt send an MT + mthex send an MT, bytes as hex + signal [hidden] send a constellationState, 0 to 5 bars + topics [,...] the provisioning from now on (empty: none) + mo-fail the next MO ends with this final_mo_status + mo-refuse the next PUT messageOriginate is refused (406) + segment ask for MO segments of n bytes (default 1446) + state the operationalState from now on + quit + + READY + RX every request from the host, as received + TX every line sent to the host + MO a whole MO with a good CRC + MOBAD a whole MO whose CRC did not match + MT an MT sent + +Run it in a process of its own beside Ground Control's Python binding: the +binding holds the GIL through every blocking call, so a fake in a thread of +the same process would freeze while the library waits on it. +""" + +import argparse +import base64 +import json +import os +import select +import sys +import threading +import time +import tty + +MO_SEGMENT_MAX = 1446 +MT_SEGMENT_MAX = 1083 # the library cuts MT data at 1446 base64 characters + + +def crc16_xmodem(data): + crc = 0 + for b in data: + crc ^= b << 8 + for _ in range(8): + crc = ((crc << 1) ^ 0x1021) if crc & 0x8000 else (crc << 1) + crc &= 0xFFFF + return crc + + +def with_crc(data): + c = crc16_xmodem(data) + return bytes(data) + bytes([c >> 8, c & 0xFF]) + + +class FakeModem: + """The modem's half of JSPR. Feed it bytes from the host with + `feed()`; it writes replies through `write`, a function taking bytes. + `events` is called with one string per event. Thread-safe: the control + methods may be called from another thread than the one feeding.""" + + def __init__(self, write, events=print, topics=(244,)): + self.write = write + self.events = events + self.lock = threading.RLock() + self.buf = b"" + self.topics = list(topics) + self.state = "active" + self.sim_interface = "internal" + self.api_active = True + self.bars, self.visible = 5, True + self.next_id = 1 + self.segment = MO_SEGMENT_MAX + self.mo = None # the MO being received + self.mo_fail = None # final status for the next MO + self.mo_refuse = False + self.imei = "300434030000000" + self.serial = "123456" + self.iccid = "8988169712345678901" + + # --- the wire ------------------------------------------------------- + + def send(self, code, target, obj): + line = "%d %s %s" % (code, target, json.dumps(obj, separators=(",", ":"))) + self.events("TX " + line) + self.write((line + "\r").encode()) + + def feed(self, data): + with self.lock: + self.buf += data + while True: + i = self.buf.find(b"\r") + if i < 0: + # The library ends lines with CR; tolerate LF from a person. + i = self.buf.find(b"\n") + if i < 0: + return + line, self.buf = self.buf[:i], self.buf[i + 1:] + line = line.strip(b"\r\n").decode(errors="replace") + if line: + self.request(line) + + def request(self, line): + self.events("RX " + line) + parts = line.split(" ", 2) + if len(parts) < 2: + return + verb, target = parts[0], parts[1] + if verb not in ("GET", "PUT"): + return # boot chatter from whatever is on the other end of the wire + try: + body = json.loads(parts[2]) if len(parts) > 2 else {} + except ValueError: + self.send(407, target, {}) + return + handler = getattr(self, "on_%s_%s" % (verb.lower(), target), None) + if handler is None: + self.send(404, target, {}) + return + handler(body) + + # --- requests ------------------------------------------------------- + + def on_get_apiVersion(self, body): + v = {"major": 1, "minor": 0, "patch": 0} + reply = {"supported_versions": [v]} + if self.api_active: + reply["active_version"] = v + self.send(200, "apiVersion", reply) + + def on_put_apiVersion(self, body): + self.api_active = True + self.send(200, "apiVersion", {"active_version": body.get("active_version", {})}) + + def on_get_simConfig(self, body): + self.send(200, "simConfig", {"interface": self.sim_interface}) + + def on_put_simConfig(self, body): + self.sim_interface = body.get("interface", "internal") + self.send(200, "simConfig", {"interface": self.sim_interface}) + self.send(299, "simStatus", self.sim_status()) + + def sim_status(self): + return {"card_present": True, "sim_connected": True, "iccid": self.iccid} + + def on_get_simStatus(self, body): + self.send(200, "simStatus", self.sim_status()) + + def on_get_operationalState(self, body): + self.send(200, "operationalState", {"state": self.state, "reason": 0}) + + def on_put_operationalState(self, body): + self.state = body.get("state", self.state) + self.send(200, "operationalState", {"state": self.state}) + self.send(299, "operationalState", {"state": self.state, "reason": 0}) + + def on_put_serviceConfig(self, body): + self.send(200, "serviceConfig", {}) + + def on_get_hwInfo(self, body): + self.send(200, "hwInfo", {"hw_version": "1.0", "serial_number": self.serial, + "imei": self.imei, "board_temp": 27}) + + def on_get_firmware(self, body): + self.send(200, "firmware", {"slot": body.get("slot", "primary"), "validity": 1, + "version": {"major": 1, "minor": 2, "patch": 3, + "build_info": "fake"}, + "hash": "0" * 64}) + + def constellation(self): + return {"constellation_visible": self.visible, "signal_bars": self.bars, + "signal_level": -100 - (5 - self.bars) * 5} + + def on_get_constellationState(self, body): + self.send(200, "constellationState", self.constellation()) + + def on_get_messageProvisioning(self, body): + self.send(200, "messageProvisioning", {"provisioning": [ + {"topic_id": t, "topic_name": "RAW" if t == 244 else "TOPIC%d" % t, + "priority": "Low"} for t in self.topics]}) + + def on_put_messageOriginate(self, body): + topic, length = body.get("topic_id", 0), body.get("message_length", 0) + if self.mo_refuse or topic not in self.topics or length < 3: + self.mo_refuse = False + self.send(406, "messageOriginate", {}) + return + mid = self.take_id() + self.mo = {"topic": topic, "id": mid, "len": length, "data": bytearray(length), + "next": 0} + self.send(200, "messageOriginate", {"topic_id": topic, + "request_reference": body.get("request_reference", 1), + "message_id": mid, + "message_response": "message_accepted"}) + self.ask_segment() + + def ask_segment(self): + mo = self.mo + n = min(self.segment, mo["len"] - mo["next"]) + self.send(299, "messageOriginateSegment", {"topic_id": mo["topic"], "message_id": mo["id"], + "segment_length": n, + "segment_start": mo["next"]}) + + def on_put_messageOriginateSegment(self, body): + mo = self.mo + if mo is None or body.get("message_id") != mo["id"]: + self.send(405, "messageOriginateSegment", {"message_id": body.get("message_id", 0)}) + return + start, n = body.get("segment_start", 0), body.get("segment_length", 0) + data = base64.b64decode(body.get("data", "")) + mo["data"][start:start + len(data)] = data + mo["next"] = start + n + self.send(200, "messageOriginateSegment", {"topic_id": mo["topic"], "message_id": mo["id"]}) + if mo["next"] < mo["len"]: + self.ask_segment() + return + whole, payload = bytes(mo["data"]), bytes(mo["data"][:-2]) + good = with_crc(payload) == whole + self.events("%s %d %d %s" % ("MO" if good else "MOBAD", mo["topic"], mo["id"], + payload.hex())) + status = self.mo_fail or ("mo_ack_received" if good else "crc_error_in_transfer") + self.mo_fail = None + self.mo = None + self.send(299, "messageOriginateStatus", {"topic_id": mo["topic"], "message_id": mo["id"], + "final_mo_status": status}) + + def on_put_messageOriginateStatus(self, body): + # Cancel: the library does not read the reply, only the status after. + self.send(200, "messageOriginateStatus", {}) + if self.mo is not None and body.get("action") == "cancel": + mo, self.mo = self.mo, None + self.send(299, "messageOriginateStatus", {"topic_id": mo["topic"], + "message_id": mo["id"], + "final_mo_status": + "message_cancelled_pre_transit"}) + + # --- what the network does -------------------------------------------- + + def take_id(self): + mid = self.next_id + self.next_id = self.next_id % 255 + 1 + return mid + + def send_mt(self, topic, payload): + with self.lock: + whole = with_crc(payload) + mid = self.take_id() + self.send(299, "messageTerminate", {"topic_id": topic, "message_id": mid, + "message_length_max": len(whole)}) + for start in range(0, len(whole), MT_SEGMENT_MAX): + seg = whole[start:start + MT_SEGMENT_MAX] + self.send(299, "messageTerminateSegment", { + "topic_id": topic, "message_id": mid, "segment_length": len(seg), + "segment_start": start, "data": base64.b64encode(seg).decode()}) + self.send(299, "messageTerminateStatus", {"topic_id": topic, "message_id": mid, + "final_mt_status": "complete"}) + self.events("MT %d %d" % (topic, mid)) + + def set_signal(self, bars, visible=True): + with self.lock: + self.bars, self.visible = bars, visible + self.send(299, "constellationState", self.constellation()) + + def control(self, line): + """One stdin command. Returns False on quit.""" + w = line.split(" ", 2) + cmd = w[0] if w else "" + if cmd == "mt": + self.send_mt(int(w[1]), w[2].encode() if len(w) > 2 else b"") + elif cmd == "mthex": + self.send_mt(int(w[1]), bytes.fromhex(w[2])) + elif cmd == "signal": + self.set_signal(int(w[1]), not (len(w) > 2 and w[2] == "hidden")) + elif cmd == "topics": + with self.lock: + self.topics = [int(t) for t in w[1].split(",")] if len(w) > 1 and w[1] else [] + elif cmd == "mo-fail": + self.mo_fail = w[1] + elif cmd == "mo-refuse": + self.mo_refuse = True + elif cmd == "segment": + self.segment = max(1, min(MO_SEGMENT_MAX, int(w[1]))) + elif cmd == "state": + self.state = w[1] + elif cmd == "quit": + return False + elif cmd: + self.events("ERR unknown command " + cmd) + return True + + +# --- serving a port ----------------------------------------------------------- + + +def open_pty(): + master, slave = os.openpty() + tty.setraw(slave) + path = os.ttyname(slave) + # Keep the slave open ourselves so the master never sees EIO while the + # host has it closed (the library opens, closes and opens it again). + return master, slave, path + + +def serve(fd, modem, stop): + while not stop.is_set(): + r, _, _ = select.select([fd], [], [], 0.1) + if not r: + continue + try: + data = os.read(fd, 4096) + except OSError: + time.sleep(0.05) # EIO: no reader on the other side yet + continue + if data: + modem.feed(data) + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0].strip()) + ap.add_argument("--port", help="a real serial port instead of a pty") + ap.add_argument("--baud", type=int, default=230400) + ap.add_argument("--topics", default="244", help="provisioned topics, comma separated") + ap.add_argument("--quiet", action="store_true", help="no RX/TX lines on stdout") + ap.add_argument("--stay", action="store_true", + help="keep serving when stdin closes, until interrupted") + a = ap.parse_args(argv) + + out_lock = threading.Lock() + + def events(s): + if a.quiet and (s.startswith("RX ") or s.startswith("TX ")): + return + with out_lock: + sys.stdout.write(s + "\n") + sys.stdout.flush() + + if a.port: + import serial + port = serial.Serial(a.port, a.baud, timeout=0) + fd, path, keep = port.fileno(), a.port, port + else: + fd, keep, path = open_pty() + + def write(b): + os.write(fd, b) + + topics = [int(t) for t in a.topics.split(",") if t] + modem = FakeModem(write, events, topics) + stop = threading.Event() + t = threading.Thread(target=serve, args=(fd, modem, stop), daemon=True) + t.start() + events("READY " + path) + try: + for line in sys.stdin: + if not modem.control(line.strip()): + break + else: + while a.stay: + time.sleep(3600) + except KeyboardInterrupt: + pass + stop.set() + del keep + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python/tests/conftest.py b/python/tests/conftest.py new file mode 100644 index 0000000..4bb0478 --- /dev/null +++ b/python/tests/conftest.py @@ -0,0 +1,224 @@ +""" +The test harness: a fake modem in a process of its own, and a console to +drive whatever is under test. + + pytest the fake and the CLI, and the + library's Python binding if + rockblock9704 is installed + pytest --zephyr-exe build/zephyr/zephyr.exe + ... and Zephyr on native_sim + pytest --board-console /dev/ttyACM0 --board-modem /dev/ttyUSB0 + ... and Zephyr on a board: its + shell on one port, its modem + UART wired to a USB-serial + adapter on the other, where the + fake modem answers + pytest --real-modem /dev/ttyUSB0 [--zephyr-exe ...] + ... and the free checks against + a real modem: nothing is sent + +The native_sim and board suites are the same tests (test_zephyr.py) run +through a different console. +""" + +import os +import queue +import re +import subprocess +import sys +import threading +import time + +import pytest + +HERE = os.path.dirname(os.path.abspath(__file__)) +ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") + + +def pytest_addoption(parser): + g = parser.getgroup("rb9704") + g.addoption("--zephyr-exe", default=os.environ.get("RB9704_ZEPHYR_EXE"), + help="the shell sample built for native_sim (zephyr.exe)") + g.addoption("--board-console", default=os.environ.get("RB9704_BOARD_CONSOLE"), + help="a board's shell, with the shell sample flashed") + g.addoption("--board-modem", default=os.environ.get("RB9704_BOARD_MODEM"), + help="the host serial port wired to the board's modem UART") + g.addoption("--board-baud", type=int, default=115200) + g.addoption("--real-modem", default=os.environ.get("RB9704_REAL_MODEM"), + help="a real RockBLOCK 9704's port, for the free checks in test_real.py") + + +class Lines: + """Lines from a stream, collected on a thread, waited on by regex.""" + + def __init__(self, stream): + self.q = queue.Queue() + self.seen = [] + threading.Thread(target=self._read, args=(stream,), daemon=True).start() + + def _read(self, stream): + for line in stream: + self.q.put(ANSI.sub("", line).rstrip("\r\n")) + + def expect(self, pattern, timeout=10.0): + rx = re.compile(pattern) + end = time.monotonic() + timeout + while True: + for line in self.seen: + m = rx.search(line) + if m: + self.seen = self.seen[self.seen.index(line) + 1:] + return m + left = end - time.monotonic() + if left <= 0: + raise AssertionError("no line matching %r in %.0f s; last lines:\n%s" + % (pattern, timeout, "\n".join(self.seen[-15:]))) + try: + self.seen.append(self.q.get(timeout=min(left, 0.2))) + except queue.Empty: + pass + + def clear(self): + while not self.q.empty(): + self.seen.append(self.q.get()) + self.seen = [] + + +class Fake: + """rb9704-fake in a subprocess, on a pty or on a real port.""" + + def __init__(self, port=None, topics="244"): + args = [sys.executable, "-m", "rb9704tools.fakemodem", "--quiet", "--topics", topics] + if port: + args += ["--port", port] + self.p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + text=True, bufsize=1, + cwd=os.path.dirname(HERE)) + self.out = Lines(self.p.stdout) + self.port = self.out.expect(r"^READY (\S+)").group(1) + + def cmd(self, line): + self.p.stdin.write(line + "\n") + self.p.stdin.flush() + + def close(self): + try: + self.cmd("quit") + self.p.wait(timeout=3) + except Exception: + self.p.kill() + + +@pytest.fixture +def fake(): + f = Fake() + yield f + f.close() + + +class NativeConsole: + """The shell sample on native_sim, its modem UART on the fake's pty.""" + + def __init__(self, exe, modem_port): + self.p = subprocess.Popen([exe, "-uart_stdinout", "-modem_uart_port=" + modem_port], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, text=True, bufsize=1) + self.out = Lines(self.p.stdout) + # Up when the shell answers; the boot log line can come before the + # shell's log backend is there to print it. + for _ in range(20): + self.sh("rb9704 status") + try: + self.out.expect(r"powered", timeout=0.5) + return + except AssertionError: + pass + self.p.kill() + raise AssertionError("native_sim's shell did not answer") + + def sh(self, line): + self.out.clear() + self.p.stdin.write(line + "\n") + self.p.stdin.flush() + + def close(self): + self.p.kill() + + +class SerialConsole: + """The shell sample on a board, over its console port.""" + + def __init__(self, port, baud): + import serial + self.s = serial.Serial(port, baud, timeout=0.1, write_timeout=2) + self.out = Lines(self._lines()) + self.sh("") + self.sh("rb9704 end") # a clean start whatever the board was doing + + def _lines(self): + buf = b"" + while True: + try: + buf += self.s.read(512) + except Exception: + return + while b"\n" in buf: + line, buf = buf.split(b"\n", 1) + yield line.decode(errors="replace") + + def sh(self, line): + self.out.clear() + data = (line + "\r").encode() + # A shell's receive buffer is small; typed in one burst a long line + # loses characters. A person types slower than this. + for i in range(0, len(data), 16): + self.s.write(data[i:i + 16]) + time.sleep(0.01) + time.sleep(0.2) + + def close(self): + self.s.close() + + +def _targets(config): + t = [] + if config.getoption("--zephyr-exe"): + t.append("native_sim") + if config.getoption("--board-console"): + t.append("board") + return t + + +def pytest_generate_tests(metafunc): + if "zephyr" in metafunc.fixturenames: + targets = _targets(metafunc.config) + metafunc.parametrize("zephyr", targets or ["none"], indirect=True) + + +@pytest.fixture +def zephyr(request): + """(console, fake): Zephyr under test and the fake modem it talks to.""" + cfg = request.config + if request.param == "none": + pytest.skip("no Zephyr target: --zephyr-exe or --board-console") + topics = getattr(request, "node").get_closest_marker("topics") + topics = topics.args[0] if topics else "244" + if request.param == "board" and not cfg.getoption("--board-modem"): + pytest.skip("--board-console needs --board-modem") + f = Fake(port=cfg.getoption("--board-modem") if request.param == "board" else None, + topics=topics) + try: + if request.param == "native_sim": + c = NativeConsole(cfg.getoption("--zephyr-exe"), f.port) + else: + c = SerialConsole(cfg.getoption("--board-console"), cfg.getoption("--board-baud")) + except Exception: + f.close() + raise + yield c, f + c.close() + f.close() + + +def pytest_configure(config): + config.addinivalue_line("markers", "topics(ids): the fake modem's provisioning") diff --git a/python/tests/test_cli.py b/python/tests/test_cli.py new file mode 100644 index 0000000..82eb137 --- /dev/null +++ b/python/tests/test_cli.py @@ -0,0 +1,46 @@ +"""The rb9704 command line against the fake modem.""" + +import subprocess +import sys + +import pytest + + +def run(*args, timeout=30): + return subprocess.run([sys.executable, "-m", "rb9704tools.cli", *args], capture_output=True, + text=True, timeout=timeout) + + +def test_jspr_raw(fake): + r = run("-p", fake.port, "jspr", "--setup", "GET hwInfo {}") + assert r.returncode == 0, r.stderr + assert "300434030000000" in r.stdout + + +def test_provision_sees_topics(fake): + r = run("-p", fake.port, "provision", "--timeout", "10") + assert r.returncode == 0, r.stdout + r.stderr + assert "244" in r.stdout + + +def test_info_through_the_library(fake): + pytest.importorskip("rockblock9704") + r = run("-p", fake.port, "info") + assert r.returncode == 0, r.stderr + assert "300434030000000" in r.stdout + + +def test_send_asks_first(fake): + pytest.importorskip("rockblock9704") + r = subprocess.run([sys.executable, "-m", "rb9704tools.cli", "-p", fake.port, "send", + "--text", "x"], input="n\n", capture_output=True, text=True, timeout=30) + assert "airtime" in (r.stdout + r.stderr).lower() + assert not any(l.startswith("MO ") for l in fake.out.seen) + + +def test_send_with_yes(fake): + pytest.importorskip("rockblock9704") + r = run("-p", fake.port, "send", "--text", "cli-hello", "--yes") + assert r.returncode == 0, r.stdout + r.stderr + m = fake.out.expect(r"^MO 244 \d+ (\w+)") + assert bytes.fromhex(m.group(1)) == b"cli-hello" diff --git a/python/tests/test_fakemodem.py b/python/tests/test_fakemodem.py new file mode 100644 index 0000000..9a30d75 --- /dev/null +++ b/python/tests/test_fakemodem.py @@ -0,0 +1,106 @@ +"""The fake modem on its own, in this process: no serial port, no library.""" + +import base64 +import json + +from rb9704tools.fakemodem import FakeModem, crc16_xmodem, with_crc + + +class Wire: + def __init__(self, topics=(244,)): + self.out = [] + self.events = [] + self.m = FakeModem(self.out.append, self.events.append, topics) + + def put(self, line): + self.m.feed((line + "\r").encode()) + + def lines(self): + got = [b.decode().rstrip("\r") for b in self.out] + self.out.clear() + return [(int(l.split(" ", 2)[0]), l.split(" ", 2)[1], json.loads(l.split(" ", 2)[2])) + for l in got] + + +def test_crc_is_xmodem(): + assert crc16_xmodem(b"123456789") == 0x31C3 + assert with_crc(b"hello") == bytes.fromhex("68656c6c6fc362") + + +def test_begin_sequence(): + w = Wire() + w.put("GET apiVersion {}") + w.put("GET simConfig {}") + w.put("GET operationalState {}") + got = w.lines() + assert [(c, t) for c, t, _ in got] == [(200, "apiVersion"), (200, "simConfig"), + (200, "operationalState")] + assert "active_version" in got[0][2] + assert got[1][2]["interface"] == "internal" + assert got[2][2]["state"] == "active" + + +def test_put_sim_config_answers_then_reports_status(): + w = Wire() + w.put('PUT simConfig {"interface": "internal"}') + assert [(c, t) for c, t, _ in w.lines()] == [(200, "simConfig"), (299, "simStatus")] + + +def test_mo_in_segments_with_crc(): + w = Wire() + w.m.segment = 4 + payload = b"hello world" + whole = with_crc(payload) + w.put('PUT messageOriginate {"topic_id":244, "message_length":%d, "request_reference":1}' + % len(whole)) + got = w.lines() + assert got[0][:2] == (200, "messageOriginate") + mid = got[0][2]["message_id"] + asks = [got[1]] + while True: + code, target, body = asks[-1] + if target == "messageOriginateStatus": + break + assert (code, target) == (299, "messageOriginateSegment") + s, n = body["segment_start"], body["segment_length"] + w.put('PUT messageOriginateSegment {"topic_id":244, "message_id":%d, "segment_length":%d,' + ' "segment_start":%d, "data":"%s"}' + % (mid, n, s, base64.b64encode(whole[s:s + n]).decode())) + asks = [l for l in w.lines() if l[0] == 299] + assert asks[-1][2]["final_mo_status"] == "mo_ack_received" + assert "MO 244 %d %s" % (mid, payload.hex()) in w.events + + +def test_mo_with_a_bad_crc(): + w = Wire() + bad = b"abc\x00\x00" + w.put('PUT messageOriginate {"topic_id":244, "message_length":5, "request_reference":1}') + mid = w.lines()[0][2]["message_id"] + w.put('PUT messageOriginateSegment {"topic_id":244, "message_id":%d, "segment_length":5,' + ' "segment_start":0, "data":"%s"}' % (mid, base64.b64encode(bad).decode())) + status = [b for c, t, b in w.lines() if t == "messageOriginateStatus"] + assert status[0]["final_mo_status"] == "crc_error_in_transfer" + assert any(e.startswith("MOBAD 244") for e in w.events) + + +def test_unprovisioned_topic_is_refused(): + w = Wire(topics=()) + w.put('PUT messageOriginate {"topic_id":244, "message_length":7, "request_reference":1}') + assert w.lines()[0][:2] == (406, "messageOriginate") + + +def test_mt_segments_fit_the_library(): + w = Wire() + w.m.send_mt(244, b"y" * 3000) + got = w.lines() + segs = [b for c, t, b in got if t == "messageTerminateSegment"] + assert all(len(s["data"]) <= 1446 for s in segs) + data = b"".join(base64.b64decode(s["data"]) for s in segs) + assert data == with_crc(b"y" * 3000) + assert got[-1][2]["final_mt_status"] == "complete" + + +def test_unknown_target_is_404(): + w = Wire() + w.put("GET nonsense {}") + assert w.lines()[0][:2] == (404, "nonsense") diff --git a/python/tests/test_library.py b/python/tests/test_library.py new file mode 100644 index 0000000..189b68e --- /dev/null +++ b/python/tests/test_library.py @@ -0,0 +1,56 @@ +""" +Ground Control's own Python binding (the rockblock9704 package, the same C +library this repository builds into Zephyr) against the fake modem on a pty. +The fake runs in a process of its own: the binding holds the GIL through +every blocking call. +""" + +import time + +import pytest + +rb = pytest.importorskip("rockblock9704") + + +@pytest.fixture +def modem(fake): + m = rb.RockBlock9704() + assert m.begin(fake.port) + yield m + m.end() + + +def poll_until(m, cond, secs=10): + end = time.monotonic() + secs + while time.monotonic() < end: + m.poll() + if cond(): + return True + time.sleep(0.01) + return False + + +def test_identity(modem): + assert modem.get_imei() == "300434030000000" + assert modem.get_iccid() == "8988169712345678901" + assert modem.get_firmware_version() == "v1.2.3" + assert modem.get_signal() == 5 + + +def test_send_async(modem, fake): + done = [] + modem.set_mo_message_complete_callback(lambda mid, status: done.append(status)) + assert modem.send_message_async(b"from-python", 244) + assert poll_until(modem, lambda: done) + assert done == [1] + m = fake.out.expect(r"^MO 244 \d+ (\w+)") + assert bytes.fromhex(m.group(1)) == b"from-python" + + +def test_receive_async(modem, fake): + got = [] + modem.set_mt_message_complete_callback(lambda mid, status: got.append(status)) + fake.cmd("mt 244 to-python") + assert poll_until(modem, lambda: got) + assert modem.receive_message_async() == b"to-python" + assert modem.acknowledge_receive_head_async() diff --git a/python/tests/test_real.py b/python/tests/test_real.py new file mode 100644 index 0000000..f6d9507 --- /dev/null +++ b/python/tests/test_real.py @@ -0,0 +1,49 @@ +""" +A real RockBLOCK 9704, with --real-modem PORT. Free checks only: identity +and signal, never a message, so nothing here costs airtime. Indoors the +signal is usually 0, which passes. +""" + +import re + +import pytest + +from conftest import NativeConsole + + +@pytest.fixture +def real(request): + port = request.config.getoption("--real-modem") + if not port: + pytest.skip("no --real-modem") + return port + + +def test_library_identity(real): + rb = pytest.importorskip("rockblock9704") + m = rb.RockBlock9704() + assert m.begin(real) + try: + assert re.fullmatch(r"\d{15}", m.get_imei() or "") + assert m.get_card_present() + assert 0 <= m.get_signal() <= 5 + finally: + m.end() + + +def test_zephyr_native_sim_identity(real, request): + exe = request.config.getoption("--zephyr-exe") + if not exe: + pytest.skip("no --zephyr-exe") + c = NativeConsole(exe, real) + try: + c.sh("rb9704 begin") + c.out.expect(r"begin: ok", timeout=20) + c.sh("rb9704 info") + c.out.expect(r"imei\s+\d{15}", timeout=20) + c.sh("rb9704 signal") + c.out.expect(r"signal: [0-5] bars", timeout=20) + c.sh("rb9704 end") + c.out.expect(r"end: ok") + finally: + c.close() diff --git a/python/tests/test_zephyr.py b/python/tests/test_zephyr.py new file mode 100644 index 0000000..12f5b4f --- /dev/null +++ b/python/tests/test_zephyr.py @@ -0,0 +1,125 @@ +""" +Zephyr, through the shell sample, against the fake modem: on native_sim +with --zephyr-exe, on a board with --board-console and --board-modem. +""" + +import zlib + +import pytest + + +def begin(z): + c, f = z + c.sh("rb9704 begin") + c.out.expect(r"begin: ok") + + +def test_begin_and_info(zephyr): + c, f = zephyr + begin(zephyr) + c.sh("rb9704 info") + c.out.expect(r"imei\s+300434030000000") + c.out.expect(r"firmware\s+v1\.2\.3") + c.out.expect(r"sim\s+card present, connected") + + +def test_begin_twice_is_already(zephyr): + c, f = zephyr + begin(zephyr) + c.sh("rb9704 begin") + c.out.expect(r"begin: -\d+ EALREADY") + + +def test_signal(zephyr): + c, f = zephyr + begin(zephyr) + f.cmd("signal 3") + c.out.expect(r"signal 3 bars") + c.sh("rb9704 signal") + c.out.expect(r"signal: 3 bars") + + +def test_send_reaches_the_network(zephyr): + c, f = zephyr + begin(zephyr) + c.sh("rb9704 send 244 hello-network") + c.out.expect(r"send: ok") + m = f.out.expect(r"^MO 244 (\d+) (\w+)") + assert bytes.fromhex(m.group(2)) == b"hello-network" + c.out.expect(r"mo done topic 244 id %s result 0" % m.group(1)) + + +def test_send_in_many_segments(zephyr): + c, f = zephyr + begin(zephyr) + f.cmd("segment 16") + text = "".join(chr(ord("a") + i % 26) for i in range(150)) + c.sh("rb9704 send 244 " + text) + m = f.out.expect(r"^MO 244 \d+ (\w+)") + assert bytes.fromhex(m.group(1)).decode() == text + c.out.expect(r"result 0") + + +def test_send_failure_reaches_the_callback(zephyr): + c, f = zephyr + begin(zephyr) + f.cmd("mo-fail network_error") + c.sh("rb9704 send 244 doomed") + c.out.expect(r"mo done topic 244 id \d+ result -\d+") + + +def test_send_refused_by_the_modem(zephyr): + c, f = zephyr + begin(zephyr) + f.cmd("mo-refuse") + c.sh("rb9704 send 244 refused") + c.out.expect(r"send: -\d+ EIO") + + +def test_topic_below_64_is_refused_here(zephyr): + c, f = zephyr + begin(zephyr) + c.sh("rb9704 send 10 nope") + c.out.expect(r"send: -\d+ EINVAL") + + +@pytest.mark.topics("") +def test_provisioning_that_arrives_later_is_seen(zephyr): + """The library keeps the first answer for good; the module asks again.""" + c, f = zephyr + begin(zephyr) + c.sh("rb9704 send 244 too-early") + # EACCES when the library asks first; EIO when it had 244 cached from an + # earlier test on the same boot and the modem refused the message. + c.out.expect(r"send: -\d+ (EACCES|EIO)") + f.cmd("topics 244") + c.sh("rb9704 send 244 now-provisioned") + c.out.expect(r"send: ok") + f.out.expect(r"^MO 244 ") + + +def test_message_from_the_network(zephyr): + c, f = zephyr + begin(zephyr) + f.cmd("mt 244 hello-buoy") + c.out.expect(r"mt topic 244 id \d+ len 10 crc %08x: hello-buoy$" % zlib.crc32(b"hello-buoy")) + c.sh("rb9704 status") + c.out.expect(r"mt\s+1 received") + + +def test_long_message_from_the_network(zephyr): + c, f = zephyr + begin(zephyr) + body = "".join(chr(ord("a") + i % 26) for i in range(1500)) # two MT segments + f.cmd("mt 244 " + body) + c.out.expect(r"mt topic 244 id \d+ len 1500 crc %08x: %s$" + % (zlib.crc32(body.encode()), body[:64])) + + +def test_end(zephyr): + c, f = zephyr + begin(zephyr) + c.sh("rb9704 end") + c.out.expect(r"end: ok") + c.sh("rb9704 send 244 after-end") + c.out.expect(r"send: -\d+ ENOTCONN")