Sh3d/rockblock9704-zephyr
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Python: the rb9704 command line, a fake modem, and the tests

Browse Source 
rb9704 drives a 9704 on USB and the Cloudloop account behind it, moved
from WII6's host/rockblock with its project names taken out. rb9704-fake
speaks JSPR on a pty or a serial port: everything the library asks, MO in
segments with the CRC checked, MT, signal, provisioning, refusals and
failures on command.

The tests run against the fake at every level: the fake itself, the
command line, Ground Control's Python binding, the shell sample on
native_sim, and the sample on a board (test_zephyr.py, one suite through
two consoles). test_real.py makes free checks against a real modem.
This commit is contained in:
scottp committed 2026-10-09 12:48:16 +11:00
1 parent dbed9bbc88
commit 917522d1f4
12 files changed
+1861
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

python/README.md
+94
View File
@@ -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-<utc>.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 `<code> <target> <json>` 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
(<https://api.cloudloop.com/swagger/index.html>) describes it.
Only one program can hold the port: `jspr` and the library verbs cannot
run at once.
python/install.sh
Executable
+104
View File
@@ -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"
python/pyproject.toml
+29
View File
@@ -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"]
python/rb9704tools/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Host tools for the RockBLOCK 9704."""
__version__ = "0.1.0"
python/rb9704tools/cli.py
+644
View File
@@ -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, `<code> <target> <json>` 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-<utc>.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-<utc>.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()
python/rb9704tools/fakemodem.py
+381
View File
@@ -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 <topic> <text> send an MT
mthex <topic> <hex> send an MT, bytes as hex
signal <bars> [hidden] send a constellationState, 0 to 5 bars
topics <id>[,<id>...] the provisioning from now on (empty: none)
mo-fail <status> the next MO ends with this final_mo_status
mo-refuse the next PUT messageOriginate is refused (406)
segment <n> ask for MO segments of n bytes (default 1446)
state <state> the operationalState from now on
quit
READY <port>
RX <line> every request from the host, as received
TX <line> every line sent to the host
MO <topic> <id> <hex> a whole MO with a good CRC
MOBAD <topic> <id> <hex> a whole MO whose CRC did not match
MT <topic> <id> 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())
python/tests/conftest.py
+224
View File
@@ -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")
python/tests/test_cli.py
+46
View File
@@ -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"
python/tests/test_fakemodem.py
+106
View File
@@ -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")
python/tests/test_library.py
+56
View File
@@ -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()
python/tests/test_real.py
+49
View File
@@ -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()
python/tests/test_zephyr.py
+125
View File
@@ -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")