Mirrors_Framework/PX4-Autopilot
1
0
Fork 0
mirror of https://github.com/PX4/PX4-Autopilot.git synced 2026-08-17 22:29:21 +08:00
Code Issues Actions 56 Packages Projects Releases Wiki Activity

refactor(bench): restructure suite into px4bench package with bench/ and sih/ split

Browse Source 
Turn the flat script pile into a standalone project so contributors can
navigate and extend it:

- px4bench/ shared library package: core primitives in __init__ (Reporter,
  connect, MavlinkShell, reboot/replug, viewer tee, mavlink status parsers,
  pymavlink add_message workaround) plus protocol modules params.py,
  missions.py, and ftp.py extracted from the tests. Zero helper duplication
  remains across scripts.
- bench/ holds the real-firmware tests (boot_health, reboot_loop,
  usb_replug, link_forwarding, param_stress, mission_stress, log_transfer);
  sih/ holds the simulated flight (flight_mission), making the
  simulation/no-simulation boundary explicit.
- pyproject.toml (px4bench 0.1.0, BSD-3-Clause, pymavlink/pyserial deps,
  pyulog extra) so pip install -e Tools/bench_test works; every script
  remains directly runnable without installation via a parent-dir path shim.
- README rewritten contributor-first: architecture, per-test justification
  tied to the v1.18 risk areas, why pymavlink over MAVSDK, how to add a
  test, baseline workflow.
- Consistent CLI surface (shared connection args; --report-dir replaces
  log_transfer's --outdir); decorative section banners removed.

All hardware-learned behavior is preserved exactly: param echo drain and
match-by-value, shell sentinel strip-all, the add_message workaround,
explicit param save before reboot, mission clear before upload, RTL in
MAV_FRAME_MISSION, and the post-flight ULog download.

Signed-off-by: Ramon Roche <mrpollo@gmail.com>
This commit is contained in:
Ramon Roche
2026-07-07 21:33:15 -07:00
parent 8d709a5e08
commit 9526efa4f6
15 changed files with 687 additions and 707 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Tools/bench_test/README.md
+215 -199
View File
@@ -1,252 +1,269 @@
# PX4 bench-test suite
# px4bench: PX4 bench-test suite
Semi-automated release qualification for PX4 on real NuttX hardware
(Pixhawk-class boards on a bench, connected over USB and optionally a
telemetry radio).
PX4 v1.18 merged a large TSAN/concurrency series (PR #27606 plus follow-ups
#27809 and #27813) that reworked mavlink locking, uORB callbacks, WorkQueue
lifetime, parameters, and dataman. CI builds NuttX firmware but never boots it.
The characteristic failure mode of this class of bug on NuttX is a silent hang
(for example a mutex that was previously zero-initialized and locked from the
wrong context), not a crash. This suite makes hangs visible on real
Pixhawk-class hardware on a bench: every operation has a timeout, and a timeout
is reported as FAIL naming exactly what stalled.
lifetime, parameters, and dataman. CI builds NuttX firmware but never boots
it. The characteristic failure mode of this class of bug on NuttX is a
silent hang (for example a mutex that was previously zero-initialized and
locked from the wrong context), not a crash. This suite makes hangs visible:
every operation has a timeout, and a timeout is reported as FAIL naming
exactly what stalled. A hang is the finding, not a nuisance.
The suite never arms the vehicle. Logging is triggered with `logger on` /
`logger off`. A hang is the finding, not a nuisance: when a check times out the
tool prints which operation stalled and exits nonzero rather than blocking.
The bench tests never arm the vehicle; logging is triggered with
`logger on` / `logger off`. The one test that flies does so in simulation on
the FMU (SIH, `SYS_HITL=2`) with `pwm_out_sim` in place of real outputs, and
it lives in its own directory so there is no ambiguity about which is which.
## Layout
```
Tools/bench_test/
README.md
pyproject.toml # pip install -e Tools/bench_test (optional)
run_bench_suite.py # orchestrator for the non-interactive bench tests
px4bench/ # shared library package
__init__.py # Reporter, connect, MavlinkShell, reboot/replug,
# viewer tee, mavlink-status parsers,
# pymavlink add_message workaround
params.py # param read/set/drain/echo, int32 union encoding
missions.py # mission items, upload/download/compare/clear
ftp.py # MAVFTP list/download, ULog magic, log root
bench/ # real-firmware bench tests, no simulation
boot_health.py
reboot_loop.py
usb_replug.py
link_forwarding.py
param_stress.py
mission_stress.py
log_transfer.py
sih/ # simulation-in-hardware (SYS_HITL=2) tests
flight_mission.py
```
Every script is directly executable from any working directory; no install
is required (each carries a path shim to find `px4bench/`). Installing the
library is optional and only needed if you want `import px4bench` from your
own scripts:
```
pip install -e Tools/bench_test # library only
pip install -e "Tools/bench_test[ulog]" # + pyulog for deep ULog verification
```
## Requirements
- Python 3.
- Required: `pymavlink`, `pyserial`.
- Optional: `pyulog` (better .ulg verification; magic-byte check is used as
fallback when absent).
- Python 3.8+.
- Required: `pymavlink`, `pyserial` (`pip3 install --user pymavlink pyserial`).
- Optional: `pyulog` (deep .ulg verification; a magic-byte check is the
fallback).
```
pip3 install --user pymavlink pyserial
pip3 install --user pyulog # optional
```
Finding the board on macOS: `ls /dev/tty.usbmodem*`. Close QGroundControl
first; it holds the serial port.
Finding the board on macOS:
```
ls /dev/tty.usbmodem*
```
Close QGroundControl before running anything. QGC holds the serial port and the
tools will not be able to open it.
Every script prints `[PASS]` / `[FAIL]` per check and exits nonzero on any
failure.
`CONNECTION` is a serial device (`/dev/tty.usbmodem*`) or `udp:IP:PORT` /
`tcp:IP:PORT`. Default baud is 57600. Every script prints `[PASS]` /
`[FAIL]` per check and exits nonzero on any failure.
## Quick start
Full suite over a single USB link:
```
./run_bench_suite.py /dev/tty.usbmodem*
# full non-interactive bench suite, single USB link
./run_bench_suite.py /dev/tty.usbmodem01
# with a telemetry radio as a second link (enables link_forwarding and
# alternating mission uploads)
./run_bench_suite.py /dev/tty.usbmodem01 /dev/tty.usbserial-RADIO
# operator-assisted USB re-enumeration test (not part of the suite)
./bench/usb_replug.py /dev/tty.usbmodem01
# simulated flight on the FMU (reconfigures the board, run separately)
./sih/flight_mission.py /dev/tty.usbmodem01
```
Full suite with a second link (telemetry radio); enables the dual-link tests:
## Why pymavlink
```
./run_bench_suite.py /dev/tty.usbmodem* /dev/tty.usbserial-RADIO
```
The operator-assisted USB re-enumeration test is not part of the automated
suite. Run it manually:
```
./usb_replug.py /dev/tty.usbmodem*
```
The bench tests exercise the raw MAVLink protocol surface on purpose: param
echo semantics (duplicate PARAM_VALUE broadcasts are part of what we
verify), mission handshake retransmits and per-seq item requests, MAVFTP
burst-read internals with stall detection by byte count, and the nsh shell
over SERIAL_CONTROL. MAVSDK abstracts exactly those layers away, and adds an
asyncio runtime plus the mavsdk_server gRPC binary as dependencies. The one
plausible candidate was flight orchestration in `sih/flight_mission.py`, but
that test also depends on the nsh shell (arming via `commander`, explicit
`param save`, enabling viewer streams) and on teeing raw frames to a viewer,
neither of which MAVSDK exposes. Verdict: pymavlink everywhere.
## Risk map
| Test | Subsystem exercised | What a failure looks like |
Repetition is the test: single-shot operations do not hit races, which is
why the stress tests loop.
| Test | Subsystem exercised | Why it exists / what failure looks like |
| --- | --- | --- |
| boot_health | WorkQueue lifetime + uORB callback rework | Task in ERROR state, a required work queue missing, or a topic not publishing |
| usb_replug | mavlink instance lifecycle across USB re-enumeration | Heartbeat does not return, mavlink instance count changes, or free RAM grows across cycles |
| dual_link_forwarding | reworked mavlink nested-send lock path (silently deadlockable on NuttX) | Silent traffic gap > 5s on either link, param download stalls, or a link is dead afterward |
| param_torture | parameter subsystem locking | A set/readback mismatch, or the param does not persist across reboot |
| mission_torture | dataman + new mission shared-state mutex | Upload/download mismatch, or an operation stalls |
| mavftp_log | logger + MAVFTP path | Log does not start/stop, download stalls, or the .ulg fails integrity check |
| reboot_loop | boot-time initialization ordering (locked zero-init mutex bugs bite at boot) | Board does not reconnect within the cycle timeout |
| bench/boot_health | WorkQueue lifetime + uORB callback rework | Task in ERROR state, required work queue missing, topic not publishing; baseline mode catches rate drift across firmware versions |
| bench/reboot_loop | boot-time initialization ordering | Locked zero-init mutex bugs bite at boot; a board that boots once but not every time. No reconnect within timeout = FAIL with elapsed time |
| bench/usb_replug | mavlink instance lifecycle across USB re-enumeration | Instance count climbing or free RAM growing per replug cycle = leak in link teardown |
| bench/link_forwarding | reworked mavlink nested-send lock path (silently deadlockable on NuttX) | Silent traffic gap > 5s on either link, a stalled param download, or a dead link after heavy simultaneous two-link use |
| bench/param_stress | parameter storage concurrency from the 2026-07-02 TSAN batch (DynamicSparseLayer races, AtomicTransaction) | Echo/readback mismatch under repetition, a download that stalls mid-list, or a save that does not survive reboot |
| bench/mission_stress | new mission shared-state mutex (#27813) + dataman TSAN fixes | Upload/download round-trip corruption or a handshake that stalls; with two links the mutex is hit from two channels |
| bench/log_transfer | logger + MAVFTP (FILE_TRANSFER_PROTOCOL replies are nested sends) | A burst-read download that stops mid-file, or a ULog that fails integrity checks |
| sih/flight_mission | commander/navigator/land-detector flight logic on real NuttX scheduling | Arming, takeoff, waypoint progression, RTL, land detection or auto-disarm failing per-phase timeouts |
`CONNECTION` is a serial device (`/dev/tty.usbmodem*`) or `udp:IP:PORT` /
`tcp:IP:PORT`. Default baud is 57600.
## Tests
## Bench tests (real firmware, no simulation)
### boot_health
Purpose: capture a snapshot of the running system after the WorkQueue and uORB
callback rework and check it is healthy.
Snapshot of a running board: `top once`, `work_queue status`, `perf`,
`uorb top -1`, `mavlink status`, `free`, `ver all` over the MAVLink shell,
saved into a timestamped report dir.
```
./boot_health.py CONNECTION [-b BAUD] [--report-dir DIR] [--require-wq LIST]
./bench/boot_health.py CONNECTION [-b BAUD] [--report-dir DIR] [--require-wq LIST]
./bench/boot_health.py --baseline OLDDIR NEWDIR [--tolerance PCT]
```
Captures `top once`, `work_queue status`, `perf`, `uorb top -1`,
`mavlink status`, `free`, and `ver all` over the MAVLink shell into a
timestamped report dir. PASS means no task is in ERROR state, the required work
queues are present (`wq:rate_ctrl`, `wq:hp_default`, `wq:lp_default`), and
topics are publishing. A FAIL names the failing check; if the shell itself
stalls, that stall is the finding. Attach the report dir when reporting.
Baseline diff of two report dirs (uorb rates within tolerance, work queue sets
equal):
```
./boot_health.py --baseline OLDDIR NEWDIR [--tolerance PCT]
```
### usb_replug (operator-assisted)
Purpose: exercise mavlink instance lifecycle across USB re-enumeration.
```
./usb_replug.py CONNECTION [-b BAUD] [--cycles M] [--heartbeat-timeout S] [--ram-tolerance BYTES]
```
Guides you through unplug/replug cycles (default 5). PASS means the heartbeat
returns after each replug, the mavlink instance count stays constant, and free
RAM does not grow. A FAIL names the cycle and the failing condition; a heartbeat
that never returns within `--heartbeat-timeout` is the finding.
### dual_link_forwarding (flagship, needs 2 links)
Purpose: exercise the reworked mavlink nested-send lock path that was silently
deadlockable on NuttX.
```
./dual_link_forwarding.py CONNECTION CONNECTION2 [-b BAUD] [--baudrate2 BAUD2] [--duration S]
```
Verifies heartbeats on both links, then runs 60s of sustained bidirectional
traffic (a silent gap > 5s on either link is FAIL), then a full param download
over the radio link WHILE a shell session runs over USB, then proves both links
are still alive afterward. PASS means every stage completed and both links
survived. A hang anywhere reports FAIL naming the stage instead of blocking.
### param_torture
Purpose: hammer the parameter subsystem locking.
```
./param_torture.py CONNECTION [-b BAUD] [--iterations N] [--param NAME] [--skip-reboot]
```
Full param download, then 50 set/readback iterations on `SDLOG_UTC_OFFSET` (a
harmless scratch param, restored afterward), then param save + reboot +
persistence verification. PASS means every readback matched, the value
persisted across reboot, and the original value was restored. A FAIL names the
iteration or the persistence step. Use `--skip-reboot` to avoid the reboot
stage.
### mission_torture
Purpose: exercise dataman and the new mission shared-state mutex.
```
./mission_torture.py CONNECTION [CONNECTION2] [-b BAUD] [--baudrate2 BAUD2] [--iterations N] [--items K]
```
220-waypoint mission upload/download/compare/clear, 10 iterations. With two
links it alternates uploads between the links. PASS means every download matched
the upload across all iterations. A FAIL names the iteration and the mismatch,
or the operation that stalled.
### mavftp_log
Purpose: exercise the logger and the MAVFTP path.
```
./mavftp_log.py CONNECTION [-b BAUD] [--log-duration S] [--outdir DIR]
```
Starts a short log via `logger on` / `logger off` (no arming), downloads the
newest .ulg via MAVFTP, and verifies ULog integrity (pyulog if installed, magic
bytes otherwise). PASS means the log was created, downloaded, and passed the
integrity check. A FAIL names the stage; a stalled MAVFTP transfer is the
finding.
PASS: no task in ERROR state, required work queues present
(`wq:rate_ctrl,wq:hp_default,wq:lp_default` by default), topics publishing.
Baseline mode diffs two report dirs offline: uORB rate drift beyond
tolerance, lost topics, or lost work queues are FAIL.
### reboot_loop
Purpose: exercise boot-time initialization ordering, where locked zero-init
mutex bugs bite.
```
./reboot_loop.py CONNECTION [-b BAUD] [--iterations N] [--cycle-timeout S]
./bench/reboot_loop.py CONNECTION [-b BAUD] [--iterations N] [--cycle-timeout S]
```
10 reboot/reconnect cycles via `MAV_CMD_PREFLIGHT_REBOOT_SHUTDOWN`. PASS means
the board came back within `--cycle-timeout` on every cycle. Any failure to
reconnect in time is FAIL with the elapsed time and the cycle number.
Reboots via `MAV_CMD_PREFLIGHT_REBOOT_SHUTDOWN` (default 10 cycles). Each
cycle must produce a heartbeat and a completed shell command within the
timeout; a board that never comes back aborts the loop with the elapsed time.
### run_bench_suite
Runs all non-interactive tests in sequence with a per-test watchdog.
### usb_replug (operator-assisted)
```
./run_bench_suite.py CONNECTION [CONNECTION2] [-b BAUD] [--skip NAMES] [--report-dir DIR] [--per-test-timeout S] [--stop-on-fail]
./bench/usb_replug.py CONNECTION [-b BAUD] [--cycles M] [--heartbeat-timeout S] [--ram-tolerance BYTES]
```
Default per-test watchdog is 900s; a test that exceeds it is killed and recorded
as FAIL `<test> hung`. `usb_replug` is operator-assisted and must be run
manually. Dual-link tests run only when a second connection is given.
Prints UNPLUG/REPLUG instructions and detects the device node vanishing and
returning programmatically (no keyboard needed mid-test). PASS per cycle:
heartbeat back within the timeout, mavlink instance count equal to baseline,
free RAM within budget.
### sih_flight.py
Scripted SIH flight on the FMU itself (SYS_HITL=2, physics on the autopilot):
switches to a SIH airframe, clears any stored mission (an identical re-upload
would match the stored mission CRC and keep finished=true), uploads takeoff +
3-waypoint square + RTL, arms via the nsh commander, and asserts arming,
takeoff, waypoint progression, landing, and auto-disarm with per-phase
timeouts. The flight ULog is downloaded into the report directory afterwards,
also on failure, since post-flight verification starts from the log. The
original SYS_AUTOSTART/SYS_HITL are restored with an explicit `param save`
(autosave races an immediate reboot) unless `--keep-config` is given.
### link_forwarding (needs 2 links)
```
./sih_flight.py CONNECTION [-b BAUD] [--airframe N] [--alt M] [--viewer] [--keep-config]
./bench/link_forwarding.py CONNECTION CONNECTION2 [-b BAUD] [--baudrate2 BAUD2] [--duration S]
```
Requires firmware built with `CONFIG_MODULES_SIMULATION_SIMULATOR_SIH=y`.
Real outputs are replaced by `pwm_out_sim` in HITL, so nothing is driven on
the output rails; still, fly it on a bench board with nothing connected.
Run standalone; it is intentionally not part of `run_bench_suite.py` because
it reboots the board into a different airframe twice.
Four phases: liveness on both links, sustained bidirectional traffic
(silent gap > 5s = FAIL), the nested-send hammer (full param download over
the radio while an nsh shell session runs over USB), and post-stress
liveness on both links. A hang in any phase is a FAIL naming the phase.
With `--viewer`, every received MAVLink frame is teed to UDP 19410 (one frame
per datagram, the SITL viewer framing) and the HIL_STATE_QUATERNION /
HIL_ACTUATOR_CONTROLS streams are enabled on the board, so a locally running
[Hawkeye](https://github.com/PX4/Hawkeye) (`hawkeye -udp 19410 -mc`) renders
the flight live.
### param_stress
```
./bench/param_stress.py CONNECTION [-b BAUD] [--iterations N] [--param NAME] [--skip-reboot]
```
Full param download, then a set/readback loop (default 50) on
`SDLOG_UTC_OFFSET` (harmless scratch param, always restored), then
`param save` + reboot + persistence verification.
### mission_stress
```
./bench/mission_stress.py CONNECTION [CONNECTION2] [-b BAUD] [--baudrate2 BAUD2] [--iterations N] [--items K]
```
220-item mission upload/download/item-by-item compare/clear/verify-cleared,
default 10 iterations, alternating links when two are given.
### log_transfer
```
./bench/log_transfer.py CONNECTION [-b BAUD] [--log-duration S] [--report-dir DIR]
```
`logger on`, record, `logger off` (no arming), then download the newest
.ulg via MAVFTP and verify it (pyulog when available, ULog magic bytes
otherwise). A stalled transfer reports the byte count it reached.
## SIH test (simulation on the FMU)
### flight_mission
```
./sih/flight_mission.py CONNECTION [-b BAUD] [--airframe N] [--alt M]
[--viewer] [--viewer-port PORT] [--keep-config]
[--board-dev DEV] [--report-dir DIR]
```
Switches the board to a SIH airframe (`SYS_AUTOSTART=1100`, `SYS_HITL=2`,
physics simulated on the FMU, `pwm_out_sim` in place of real outputs), flies
takeoff, a 3-waypoint square, and RTL as an auto mission with per-phase
timeouts (arming, airborne, waypoint progression, touchdown, auto-disarm),
downloads the flight ULog, and restores the original configuration even on
failure. `--viewer` tees every MAVLink frame to UDP so Hawkeye
(`hawkeye -udp 19410 -mc`) renders the flight live from the serial-connected
board.
This is deliberately separated from `bench/`: it reconfigures the board and
exercises simulated flight logic, while the bench tests exercise real
firmware paths without simulation.
## Baseline workflow
Use this to compare pre-upgrade and post-upgrade firmware. Each capture
creates a timestamped directory under `--report-dir` and prints its path
(`report dir: ...`); the diff takes those printed paths, not the base
directory:
Each `boot_health` capture creates a timestamped directory under
`--report-dir` and prints its path; the diff takes those printed paths:
1. On known-good firmware, capture a report:
`./boot_health.py CONNECTION` and note the printed report dir, for
example `bench_reports/20260706T164512Z_boot_health`.
1. On known-good firmware: `./bench/boot_health.py CONNECTION`, note the
printed report dir.
2. Flash the candidate firmware.
3. Capture again: `./boot_health.py CONNECTION` and note the new report dir.
4. Diff them:
`./boot_health.py --baseline <old report dir> <new report dir> [--tolerance PCT]`
3. Capture again, note the new report dir.
4. `./bench/boot_health.py --baseline <old dir> <new dir> [--tolerance PCT]`
The diff checks that uorb rates are within tolerance and the work queue sets are
equal. A missing work queue or a rate outside tolerance is FAIL.
## Adding a new test
Conventions the suite relies on; new tests should follow all of them:
- Use `px4bench.Reporter` for every check; end with
`sys.exit(report.finish())` so the exit code reflects the result.
- Use `px4bench.add_connection_args(parser)` so the CLI surface stays
uniform (`CONNECTION`, `--baudrate/-b`, `--connect-timeout`; add
`--report-dir` if the test writes artifacts).
- Every wait has a timeout. A timeout is a FAIL whose detail names what
stalled (command, phase, seq, byte count). Never block forever: a hung
test hides the hang it was supposed to expose.
- Restore any board state you change (params, airframe), in a `finally`,
even when the test failed.
- No arming in `bench/`. Anything that flies (simulated) goes in `sih/`.
- Reuse the library: connection and shell primitives from `px4bench`,
protocol helpers from `px4bench.params` / `.missions` / `.ftp`. If two
tests need the same helper, it moves into the library.
- Shebang, exec bit, and the parent-dir path shim so the script runs
without installation; add non-interactive tests to `run_bench_suite.py`.
Two behaviors are hardware-learned and must not be "simplified" away:
parameter echoes are drained before a set and matched by expected value
(PX4 emits multiple PARAM_VALUE per set), and the shell strips every
BENCHDONE sentinel from captured output (the second safety echo of the
previous command can arrive late).
## Safety
- Bench use only. Props off.
- The suite never arms the vehicle.
- `param_torture` and `reboot_loop` reboot the board multiple times. Expect the
link to drop and return.
- `param_torture` writes only to `SDLOG_UTC_OFFSET` and restores the original
value automatically.
- `bench/` never arms the vehicle. `sih/flight_mission.py` arms only in
simulation (`SYS_HITL=2`, `pwm_out_sim`); nothing is driven on the output
rails.
- `param_stress`, `reboot_loop`, and `flight_mission` reboot the board.
Expect the link to drop and return.
- Scratch params and airframe configuration are restored automatically,
including on failure.
## Troubleshooting
@@ -257,10 +274,9 @@ No heartbeat:
- Wrong baud on a radio link. Match the radio's configured baud with `-b` /
`--baudrate2`.
Shell stalls: if `boot_health` or a shell-based stage hangs, that stall is the
finding, not a tooling bug. Note which command it stalled on and keep the report
dir.
Shell stalls: if a shell-based stage hangs, that stall is the finding, not a
tooling bug. Note which command it stalled on and keep the report dir.
Telemetry radio bandwidth: dual-link traffic, param download, and mission
upload/download over a radio link are much slower than over USB. Allow longer
durations (`--duration`, `--per-test-timeout`) when a radio is in the path.
transfer over a radio are much slower than USB. Allow longer durations
(`--duration`, `--per-test-timeout`) when a radio is in the path.
Tools/bench_test/boot_health.py → Tools/bench_test/bench/boot_health.py
+2 -26
View File
@@ -24,7 +24,7 @@ import os
import re
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import px4bench
@@ -58,10 +58,6 @@ def save_output(report_dir, cmd, output):
return path
# ---------------------------------------------------------------------------
# Parsers
# ---------------------------------------------------------------------------
def top_has_error_task(output):
"""Return (has_error, list_of_offending_lines).
@@ -99,14 +95,6 @@ def parse_work_queues(output):
return names
def count_mavlink_instances(output):
"""Count MAVLink instances from `mavlink status`.
Each instance block begins with `instance #N:` (mavlink_main.cpp:535).
"""
return len(re.findall(r'instance\s*#\s*\d+', output))
def parse_uorb_top(output):
"""Parse `uorb top -1` output into {topic_key: rate}.
@@ -141,10 +129,6 @@ def parse_uorb_top(output):
return rates
# ---------------------------------------------------------------------------
# Capture mode
# ---------------------------------------------------------------------------
def run_capture(args, report):
report_dir = px4bench.make_report_dir(args.report_dir, 'boot_health')
report.info('report dir: {}'.format(report_dir))
@@ -202,7 +186,7 @@ def run_capture(args, report):
# mavlink instances: informational count
if 'mavlink status' in outputs:
n = count_mavlink_instances(outputs['mavlink status'])
n = px4bench.count_mavlink_instances(outputs['mavlink status'])
report.info('mavlink instances: {}'.format(n))
shell.close()
@@ -210,10 +194,6 @@ def run_capture(args, report):
return report_dir
# ---------------------------------------------------------------------------
# Baseline mode
# ---------------------------------------------------------------------------
def read_saved(report_dir, cmd):
"""Read a previously saved command output from a report dir, or ''."""
path = os.path.join(report_dir, sanitize_cmd(cmd) + '.txt')
@@ -290,10 +270,6 @@ def run_baseline(old_dir, new_dir, tolerance, report):
report.info('new work queues: {}'.format(', '.join(added)))
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description='PX4 boot-health snapshot and offline baseline diff.')
Tools/bench_test/dual_link_forwarding.py → Tools/bench_test/bench/link_forwarding.py
+8 -37
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
"""
Dual-link forwarding stress test for the mavlink nested-send lock path.
Dual-link forwarding stress test for the mavlink nested-send lock path
(link_forwarding).
This is the flagship bench test for qualifying PX4 v1.18 on real NuttX
hardware after the mavlink locking rework. The rework changed how the
@@ -26,11 +27,10 @@ import sys
import threading
import time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from px4bench import Reporter, MavlinkShell, add_connection_args, connect
from pymavlink import mavutil
from px4bench import (Reporter, MavlinkShell, add_connection_args, connect,
parse_mavlink_status, send_heartbeat)
HEARTBEAT_INTERVAL = 1.0 # GCS -> autopilot heartbeat cadence, seconds
@@ -42,11 +42,6 @@ PARAM_HARD_CAP = 120.0 # absolute cap on the param download, seconds
GLOBAL_BUDGET_EXTRA = 300.0 # wall-clock budget = duration + this
def send_heartbeat(mav):
mav.mav.heartbeat_send(mavutil.mavlink.MAV_TYPE_GCS,
mavutil.mavlink.MAV_AUTOPILOT_INVALID, 0, 0, 0)
class ParamDownloader(threading.Thread):
"""Full parameter download over one link, run in a daemon worker thread.
@@ -147,30 +142,6 @@ class LinkPump(threading.Thread):
pass
def parse_status_instances(text):
"""Parse `mavlink status` output into a list of per-instance dicts with
tx/rx B/s. Blocks start with a line containing 'instance #'.
"""
instances = []
cur = None
for line in text.splitlines():
stripped = line.strip()
if 'instance #' in stripped:
if cur is not None:
instances.append(cur)
cur = {'header': stripped, 'tx': None, 'rx': None}
continue
if cur is None:
continue
if stripped.startswith('tx:') and cur['tx'] is None:
cur['tx'] = stripped
elif stripped.startswith('rx:') and cur['rx'] is None:
cur['rx'] = stripped
if cur is not None:
instances.append(cur)
return instances
def phase1_liveness(report, mav1, mav2, global_deadline):
"""Independent heartbeat check on each link. FAIL names the dead link."""
report.info('Phase 1: liveness on both links')
@@ -317,7 +288,7 @@ def phase4_post_liveness(report, mav1, mav2, global_deadline):
if timed_out:
report.fail('phase4_shell_stall', '`mavlink status` over link1 stalled after stress')
return False
instances = parse_status_instances(out)
instances = parse_mavlink_status(out)
if not instances:
report.fail('phase4_status_parse', 'could not parse any instance block from mavlink status')
return False
@@ -341,11 +312,11 @@ def main():
args = parser.parse_args()
if args.connection2 is None:
print('error: dual_link_forwarding requires a second connection (CONNECTION2)',
print('error: link_forwarding requires a second connection (CONNECTION2)',
file=sys.stderr)
sys.exit(2)
report = Reporter('dual_link_forwarding')
report = Reporter('link_forwarding')
global_deadline = time.monotonic() + args.duration + GLOBAL_BUDGET_EXTRA
mav1 = None
Tools/bench_test/mavftp_log.py → Tools/bench_test/bench/log_transfer.py
+7 -90
View File
@@ -22,112 +22,29 @@ import os
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from px4bench import Reporter, MavlinkShell, add_connection_args, connect, make_report_dir
from px4bench import (Reporter, MavlinkShell, add_connection_args, connect,
make_report_dir, send_heartbeat)
from px4bench.ftp import LOG_ROOT, ULOG_MAGIC7, ftp_list, ftp_download
from pymavlink import mavutil
from pymavlink import mavftp
LOG_ROOT = '/fs/microsd/log'
# ULog file header magic: 'U' 'L' 'o' 'g' 0x01 0x12 0x35, then a file-version
# byte. See src/modules/logger/logger.cpp:2099-2106.
ULOG_MAGIC7 = bytes([0x55, 0x4C, 0x6F, 0x67, 0x01, 0x12, 0x35])
FTP_BUDGET = 300.0 # overall MAVFTP budget in seconds
SHELL_TIMEOUT = 10.0
MIN_LOG_SIZE = 1024 # a real ULog is more than 1 KB
def send_heartbeat(mav):
mav.mav.heartbeat_send(mavutil.mavlink.MAV_TYPE_GCS,
mavutil.mavlink.MAV_AUTOPILOT_INVALID, 0, 0, 0)
def ftp_list(ftp, path):
"""List a remote directory. cmd_list drives its own reply loop internally
and populates ftp.list_result. Returns the list of DirectoryEntry.
"""
ftp.cmd_list([path])
return list(ftp.list_result)
def ftp_download(mav, ftp, remote, local, report):
"""Download one remote file to a local path over MAVFTP, bounded by
FTP_BUDGET and reporting a stall by byte count.
cmd_get only sends OpenFileRO and returns; the transfer itself is driven
by process_ftp_reply, which loops recv_match -> __mavlink_packet (burst
read handling, temp-file writes) -> __idle_task until idle or timeout.
We use a completion callback so we know the burst read finished, and we
pump process_ftp_reply in bounded slices so a stalled burst becomes a
named FAIL instead of a hang. process_ftp_reply requires its timeout to
be greater than idle_detection_time (default 3.7s).
Returns (elapsed_seconds, None) on success, (None, error_string) on
failure, so the caller can branch on the error explicitly instead of
ever mixing a message string into arithmetic.
"""
done = {'ok': False}
def on_complete(fh):
# fh is a BytesIO on success, or None if the session was terminated
# (failure). cmd_get with a callback keeps the payload in memory, so we
# persist it here ourselves.
if fh is None:
return
try:
fh.seek(0)
with open(local, 'wb') as out:
out.write(fh.read())
done['ok'] = True
except Exception as e: # noqa: BLE001 - surface write failures to the caller
report.info('local write failed: {}'.format(e))
ftp.cmd_get([remote], callback=on_complete)
start = time.monotonic()
last_bytes = 0
last_progress = start
next_hb = 0.0
# process_ftp_reply timeout must exceed idle_detection_time; use a short
# slice and loop so we can enforce our own budget and stall detection.
slice_timeout = max(4.0, ftp.ftp_settings.idle_detection_time + 0.5)
while not done['ok']:
now = time.monotonic()
if now > next_hb:
send_heartbeat(mav)
next_hb = now + 1.0
if now - start > FTP_BUDGET:
cur = ftp.read_total if ftp.read_total else last_bytes
return None, 'MAVFTP transfer stalled at {} bytes (budget {:.0f}s)'.format(
cur, FTP_BUDGET)
# drive one slice of the FTP state machine
ftp.process_ftp_reply('OpenFileRO', timeout=slice_timeout)
cur = ftp.read_total
if cur > last_bytes:
last_bytes = cur
last_progress = now
elif now - last_progress > 30.0 and not done['ok']:
# no forward progress for 30s and callback never fired
return None, 'MAVFTP transfer stalled at {} bytes (no progress 30s)'.format(cur)
# if the session ended without completing, process_ftp_reply keeps
# returning immediately on idle; break out via the budget/stall checks
elapsed = time.monotonic() - start
return elapsed, None
def main():
parser = argparse.ArgumentParser(
description='Logger + MAVFTP download path test (v1.18 bench).')
add_connection_args(parser)
parser.add_argument('--log-duration', type=float, default=10,
help='seconds to log before stopping (default: %(default)s)')
parser.add_argument('--outdir', default='bench_reports',
parser.add_argument('--report-dir', default='bench_reports',
help='base directory for the downloaded log (default: %(default)s)')
args = parser.parse_args()
report = Reporter('mavftp_log')
report = Reporter('log_transfer')
mav = None
try:
@@ -211,7 +128,7 @@ def main():
newest.name, newest.size_b))
remote = '{}/{}'.format(log_dir, newest.name)
outdir = make_report_dir(args.outdir, 'mavftp_log')
outdir = make_report_dir(args.report_dir, 'log_transfer')
local = os.path.join(outdir, newest.name)
report.info('Downloading {} -> {}'.format(remote, local))
Tools/bench_test/bench/mission_stress.py
Executable
+127
View File
@@ -0,0 +1,127 @@
#!/usr/bin/env python3
"""
Mission upload/download/clear stress test for on-bench PX4 hardware.
v1.18 risk area: dataman and the mission shared-state path were reworked and
a brand-new mutex now guards mission shared state (#27813). A regression
there shows up as a SILENT HANG (a MISSION_REQUEST that never arrives, a
MISSION_ACK that never comes, a download that stalls mid-list) or as a
corrupted round-trip. Repetition is the test: single-shot operations do not
hit races, so this script repeatedly uploads a large mission, reads it back
and compares it item-by-item, then clears it, with a hard timeout on every
wait so a stall becomes a FAIL naming the stalled step and the seq reached.
With a second connection given, iterations alternate between the two links
(iteration parity picks the link) to exercise the new mission shared-state
mutex from two MAVLink channels.
Usage:
mission_stress.py CONNECTION [CONNECTION2] [-b BAUD]
[--baudrate2 BAUD2] [--iterations N] [--items K]
"""
import argparse
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import px4bench
from px4bench import missions
DEFAULT_ITERATIONS = 10
DEFAULT_ITEMS = 220 # <= smallest CONFIG_NUM_MISSION_ITMES_SUPPORTED (500)
DEFAULT_BAUD2 = 57600
def run_iteration(report, mav, iteration, items_k, link_label):
"""One full upload/download/compare/clear cycle. Returns True on full pass."""
expected = missions.generate_mission(iteration, items_k)
up_ok, up_dur, up_detail = missions.upload_mission(report, mav, expected, iteration)
if not report.check('iter{}_upload'.format(iteration), up_ok,
'[{}] {} ({:.1f}s)'.format(link_label, up_detail, up_dur)):
return False
downloaded, dl_dur, dl_detail = missions.download_mission(mav)
if downloaded is None:
report.fail('iter{}_download'.format(iteration),
'[{}] {} ({:.1f}s)'.format(link_label, dl_detail, dl_dur))
return False
cmp_ok, cmp_detail = missions.compare_mission(expected, downloaded)
if not report.check('iter{}_compare'.format(iteration), cmp_ok,
'[{}] {}'.format(link_label, cmp_detail)):
return False
clr_ok, clr_detail = missions.clear_mission(mav)
if not report.check('iter{}_clear'.format(iteration), clr_ok,
'[{}] {}'.format(link_label, clr_detail)):
return False
vfy_ok, vfy_detail = missions.verify_cleared(mav)
report.check('iter{}_verify_clear'.format(iteration), vfy_ok,
'[{}] {}'.format(link_label, vfy_detail))
if up_ok and downloaded is not None and cmp_ok and clr_ok and vfy_ok:
report.info('iter {} PASS [{}] upload {:.1f}s download {:.1f}s'.format(
iteration, link_label, up_dur, dl_dur))
return True
return False
def main():
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
px4bench.add_connection_args(parser, dual_link=True)
parser.add_argument('--baudrate2', type=int, default=DEFAULT_BAUD2,
help='baud rate for the second connection (default: %(default)s)')
parser.add_argument('--iterations', type=int, default=DEFAULT_ITERATIONS,
help='upload/download/clear iterations (default: %(default)s)')
parser.add_argument('--items', type=int, default=DEFAULT_ITEMS,
help='mission items per iteration (default: %(default)s)')
args = parser.parse_args()
report = px4bench.Reporter('mission_stress')
try:
mav1 = px4bench.connect(args.connection, baud=args.baudrate,
timeout=args.connect_timeout)
except (TimeoutError, OSError) as e:
report.fail('connect', 'link 1 {}: {}'.format(args.connection, e))
sys.exit(report.finish())
report.info('link 1 connected to system {} component {}'.format(
mav1.target_system, mav1.target_component))
mav2 = None
if args.connection2:
try:
mav2 = px4bench.connect(args.connection2, baud=args.baudrate2,
timeout=args.connect_timeout)
report.info('link 2 connected to system {} component {}'.format(
mav2.target_system, mav2.target_component))
except (TimeoutError, OSError) as e:
report.fail('connect', 'link 2 {}: {}'.format(args.connection2, e))
sys.exit(report.finish())
try:
for iteration in range(args.iterations):
if mav2 is not None and (iteration % 2 == 1):
mav, label = mav2, 'link2'
else:
mav, label = mav1, 'link1'
run_iteration(report, mav, iteration, args.items, label)
finally:
for m in (mav1, mav2):
try:
if m is not None:
m.close()
except Exception:
pass
sys.exit(report.finish())
if __name__ == '__main__':
main()
Tools/bench_test/param_torture.py → Tools/bench_test/bench/param_stress.py
+15 -130
View File
@@ -1,13 +1,14 @@
#!/usr/bin/env python3
"""
Parameter-subsystem torture test for on-bench PX4 hardware.
Parameter-subsystem stress test for on-bench PX4 hardware.
v1.18 risk area: the parameters backend received a concurrency and locking
rework. A regression there shows up as a SILENT HANG (a PARAM_VALUE that
never arrives, a full download that stalls halfway) or as a dropped write
that only surfaces after a reboot. This script hammers the parameter
protocol over MAVLink with hard timeouts on every wait, so a stall becomes
a named FAIL instead of a hung process.
rework (DynamicSparseLayer races, AtomicTransaction). A regression there
shows up as a SILENT HANG (a PARAM_VALUE that never arrives, a full download
that stalls halfway) or as a dropped write that only surfaces after a
reboot. Repetition is the test: single-shot operations do not hit races, so
this script hammers the parameter protocol over MAVLink with hard timeouts
on every wait, and a stall becomes a named FAIL instead of a hung process.
Three phases:
1. Full parameter download (PARAM_REQUEST_LIST), stall-detected.
@@ -19,20 +20,22 @@ timestamps, so it is safe to thrash on a bench. Its original value is
always restored while a connection is alive, even on failure.
Usage:
param_torture.py CONNECTION [-b BAUD] [--iterations N]
[--param NAME] [--skip-reboot]
param_stress.py CONNECTION [-b BAUD] [--iterations N]
[--param NAME] [--skip-reboot]
"""
import argparse
import os
import struct
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import px4bench
from px4bench import mavutil
from px4bench.params import (READ_TIMEOUT_S, SET_ECHO_TIMEOUT_S,
drain_param_values, param_float_to_int32,
param_id_str, read_param, set_param_int32,
wait_param_echo)
DEFAULT_PARAM = 'SDLOG_UTC_OFFSET'
DEFAULT_ITERATIONS = 50
@@ -40,116 +43,10 @@ DEFAULT_ITERATIONS = 50
# Persistence-phase marker value (within SDLOG_UTC_OFFSET min/max -1000..1000).
MARKER_VALUE = 777
MAV_PARAM_TYPE_INT32 = mavutil.mavlink.MAV_PARAM_TYPE_INT32
# Download stall thresholds.
DOWNLOAD_STALL_S = 10.0 # FAIL if no new param index for this long
DOWNLOAD_OVERALL_CAP_S = 180.0
# Per-message wait timeouts.
SET_ECHO_TIMEOUT_S = 5.0
READ_TIMEOUT_S = 5.0
# ---------------------------------------------------------------------------
# INT32 <-> float union (byte-wise) encoding.
#
# PX4 transports an INT32 parameter as the raw bit pattern of the int placed
# into the float param_value field. To send: pack the int as '<i', reinterpret
# those 4 bytes as '<f'. To receive: pack the received float as '<f',
# reinterpret as '<i'. This mirrors the union trick the firmware uses; the
# param float is never a numeric conversion of the int.
# ---------------------------------------------------------------------------
def int32_to_param_float(value):
"""Reinterpret an int32 bit pattern as the float carried in param_value."""
return struct.unpack('<f', struct.pack('<i', int(value)))[0]
def param_float_to_int32(value):
"""Reinterpret a param_value float's bit pattern as an int32."""
return struct.unpack('<i', struct.pack('<f', float(value)))[0]
def param_id_str(raw):
"""Normalize a PARAM_VALUE.param_id (bytes or str) to a plain string."""
if isinstance(raw, bytes):
raw = raw.decode('ascii', errors='replace')
return raw.rstrip('\x00')
# ---------------------------------------------------------------------------
# Phase helpers
# ---------------------------------------------------------------------------
def request_param_read(mav, name):
"""Send PARAM_REQUEST_READ by name (param_index = -1)."""
mav.mav.param_request_read_send(
mav.target_system, mav.target_component,
name.encode('ascii'), -1)
def read_param(mav, name, timeout=READ_TIMEOUT_S):
"""PARAM_REQUEST_READ then wait for the matching PARAM_VALUE.
Returns (int32_value, raw_float) or (None, None) on timeout.
"""
request_param_read(mav, name)
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
m = mav.recv_match(type='PARAM_VALUE', blocking=True,
timeout=max(0.1, deadline - time.monotonic()))
if m is None:
continue
if param_id_str(m.param_id) == name:
return param_float_to_int32(m.param_value), m.param_value
return None, None
def set_param_int32(mav, name, value):
"""PARAM_SET an int32 using the union encoding."""
mav.mav.param_set_send(
mav.target_system, mav.target_component,
name.encode('ascii'),
int32_to_param_float(value),
MAV_PARAM_TYPE_INT32)
def drain_param_values(mav):
"""Discard any queued PARAM_VALUE messages (stale echoes/broadcasts)."""
while mav.recv_match(type='PARAM_VALUE', blocking=False) is not None:
pass
def wait_param_echo(mav, name, expected, timeout=SET_ECHO_TIMEOUT_S):
"""Wait for a PARAM_VALUE for name carrying the expected int32 value.
PX4 can emit more than one PARAM_VALUE per set (handler reply plus the
changed-param announcement, times the number of mavlink instances), so
consuming a single message desyncs the harness by one echo forever.
Returns (matched, seen) where seen is every value observed for name; a
genuine wrong-echo firmware bug shows up as matched=False with the wrong
value(s) in seen.
"""
seen = []
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
m = mav.recv_match(type='PARAM_VALUE', blocking=True,
timeout=max(0.1, deadline - time.monotonic()))
if m is None:
continue
if param_id_str(m.param_id) != name:
continue
value = param_float_to_int32(m.param_value)
seen.append(value)
if value == expected:
return True, seen
return False, seen
# ---------------------------------------------------------------------------
# Phase 1: full download
# ---------------------------------------------------------------------------
def phase_full_download(report, mav):
"""PARAM_REQUEST_LIST; collect until param_count reached, stall-detected.
@@ -229,10 +126,6 @@ def phase_full_download(report, mav):
return by_name
# ---------------------------------------------------------------------------
# Phase 2: set / readback loop
# ---------------------------------------------------------------------------
def phase_set_readback(report, mav, param, iterations):
"""Loop set/readback with a value that changes every iteration.
@@ -278,10 +171,6 @@ def phase_set_readback(report, mav, param, iterations):
iterations, echo_failures, read_failures, mismatch_failures))
# ---------------------------------------------------------------------------
# Phase 3: persistence across reboot
# ---------------------------------------------------------------------------
def phase_persistence(report, mav, param, conn_str, baud, original_value):
"""Write a marker, save, reboot, verify it survived, then restore.
@@ -339,10 +228,6 @@ def phase_persistence(report, mav, param, conn_str, baud, original_value):
return mav
# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description=__doc__,
@@ -356,7 +241,7 @@ def main():
help='skip the reboot persistence phase')
args = parser.parse_args()
report = px4bench.Reporter('param_torture')
report = px4bench.Reporter('param_stress')
try:
mav = px4bench.connect(args.connection, baud=args.baudrate,
Tools/bench_test/reboot_loop.py → Tools/bench_test/bench/reboot_loop.py
+1 -1
View File
@@ -19,7 +19,7 @@ import argparse
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import px4bench
Tools/bench_test/usb_replug.py → Tools/bench_test/bench/usb_replug.py
+3 -11
View File
@@ -26,15 +26,10 @@ import os
import re
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import px4bench
def count_mavlink_instances(output):
"""Count MAVLink instances from `mavlink status` (instance #N blocks)."""
return len(re.findall(r'instance\s*#\s*\d+', output))
def parse_free_used(output):
"""Parse NuttX `free` output, return (total_bytes, used_bytes) or (None, None).
@@ -89,7 +84,7 @@ def read_baseline(shell, report):
report.fail('baseline mavlink status',
'shell command mavlink status stalled (no completion within 10s)')
return None
instances = count_mavlink_instances(ms_out)
instances = px4bench.count_mavlink_instances(ms_out)
free_out, timed_out = shell.run('free', timeout=10)
if timed_out:
@@ -131,7 +126,6 @@ def main():
device = args.connection
# ----- baseline -----
try:
mav = px4bench.connect(device, baud=args.baudrate, timeout=args.connect_timeout)
except (TimeoutError, OSError) as e:
@@ -157,7 +151,6 @@ def main():
last_used = baseline['used']
# ----- replug cycles -----
for i in range(1, args.cycles + 1):
print('-' * 60, flush=True)
print('Cycle {}/{}: UNPLUG the USB cable now.'.format(i, args.cycles), flush=True)
@@ -215,7 +208,7 @@ def main():
shell.close()
mav.close()
continue
instances = count_mavlink_instances(ms_out)
instances = px4bench.count_mavlink_instances(ms_out)
report.check('cycle {} instances'.format(i),
instances == baseline['instances'],
'instances={} (baseline={})'.format(instances, baseline['instances']))
@@ -241,7 +234,6 @@ def main():
used - last_used))
last_used = used
# ----- final summary -----
print('-' * 60, flush=True)
total_growth = last_used - baseline['used']
report.info('free used: first={} bytes, last={} bytes, net growth={:+d} bytes '
Tools/bench_test/px4bench.py → Tools/bench_test/px4bench/__init__.py
+50 -33
View File
@@ -1,10 +1,14 @@
#!/usr/bin/env python3
"""
Shared helpers for the PX4 bench-test suite (Tools/bench_test).
"""px4bench: shared library for the PX4 bench-test suite (Tools/bench_test).
Provides MAVLink connection setup, a non-interactive nsh shell over
SERIAL_CONTROL (reusing the pattern from Tools/mavlink_shell.py), reboot
and USB re-detection helpers, and PASS/FAIL reporting.
Core primitives every test builds on: MAVLink connection setup, PASS/FAIL
reporting, a non-interactive nsh shell over SERIAL_CONTROL (pattern from
Tools/mavlink_shell.py), reboot and USB re-detection helpers, a live viewer
tee, and small parsers for `mavlink status` output.
Protocol-area helpers live in submodules:
px4bench.params parameter read/set/echo with PX4's int32 union encoding
px4bench.missions mission item generation and the upload/download protocol
px4bench.ftp MAVFTP listing/download and ULog constants
Every operation takes a timeout: a hung board is a finding, not an excuse
for a hung test script.
@@ -53,10 +57,6 @@ USB_DEVICE_GLOB_DARWIN = '/dev/tty.usbmodem*'
USB_DEVICE_GLOB_LINUX = '/dev/serial/by-id/*PX4*'
# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------
class Reporter:
"""Collects named PASS/FAIL checks and prints a summary.
@@ -117,10 +117,6 @@ def make_report_dir(base='bench_reports', test_name=''):
return path
# ---------------------------------------------------------------------------
# Connection
# ---------------------------------------------------------------------------
def is_serial_device(conn_str):
return conn_str.startswith('/dev/') or conn_str.upper().startswith('COM')
@@ -135,8 +131,7 @@ def connect(conn_str, baud=DEFAULT_BAUD, timeout: float = 20, source_system=254)
source_component=mavutil.mavlink.MAV_COMP_ID_MISSIONPLANNER)
assert isinstance(mav, mavutil.mavfile), 'unexpected connection type for {}'.format(conn_str)
# announce ourselves so the autopilot streams to us
mav.mav.heartbeat_send(mavutil.mavlink.MAV_TYPE_GCS,
mavutil.mavlink.MAV_AUTOPILOT_INVALID, 0, 0, 0)
send_heartbeat(mav)
hb = mav.wait_heartbeat(timeout=int(timeout))
if hb is None:
mav.close()
@@ -149,6 +144,12 @@ def wait_heartbeat(mav, timeout: float = 10):
return mav.wait_heartbeat(timeout=int(timeout))
def send_heartbeat(mav):
"""Send one GCS heartbeat so the autopilot keeps streaming to us."""
mav.mav.heartbeat_send(mavutil.mavlink.MAV_TYPE_GCS,
mavutil.mavlink.MAV_AUTOPILOT_INVALID, 0, 0, 0)
def drain(mav, duration=0.5):
"""Read and discard pending messages for a short window."""
deadline = time.monotonic() + duration
@@ -157,9 +158,37 @@ def drain(mav, duration=0.5):
pass
# ---------------------------------------------------------------------------
# nsh shell over MAVLink SERIAL_CONTROL
# ---------------------------------------------------------------------------
def parse_mavlink_status(text):
"""Parse `mavlink status` shell output into per-instance dicts.
Each instance block begins with `instance #N:` and contains rate lines
`tx: X B/s` / `rx: X B/s` (src/modules/mavlink/mavlink_main.cpp). Returns
a list of {'header', 'tx', 'rx'} dicts; its length is the instance count.
"""
instances = []
cur = None
for line in text.splitlines():
stripped = line.strip()
if 'instance #' in stripped:
if cur is not None:
instances.append(cur)
cur = {'header': stripped, 'tx': None, 'rx': None}
continue
if cur is None:
continue
if stripped.startswith('tx:') and cur['tx'] is None:
cur['tx'] = stripped
elif stripped.startswith('rx:') and cur['rx'] is None:
cur['rx'] = stripped
if cur is not None:
instances.append(cur)
return instances
def count_mavlink_instances(text):
"""Count MAVLink instances in `mavlink status` output."""
return len(re.findall(r'instance\s*#\s*\d+', text))
class MavlinkShell:
"""Non-interactive nsh shell over SERIAL_CONTROL (dev 10).
@@ -194,8 +223,7 @@ class MavlinkShell:
"""Read pending SERIAL_CONTROL data into self.buf, keep heartbeats going."""
now = time.monotonic()
if now > self._next_heartbeat:
self.mav.mav.heartbeat_send(mavutil.mavlink.MAV_TYPE_GCS,
mavutil.mavlink.MAV_AUTOPILOT_INVALID, 0, 0, 0)
send_heartbeat(self.mav)
self._next_heartbeat = now + 1.0
m = self.mav.recv_match(condition='SERIAL_CONTROL.count!=0',
type='SERIAL_CONTROL', blocking=True, timeout=window)
@@ -255,10 +283,6 @@ class MavlinkShell:
pass
# ---------------------------------------------------------------------------
# Live viewer tee (Hawkeye)
# ---------------------------------------------------------------------------
def attach_viewer_tee(conn, host='127.0.0.1', port=19410):
"""Forward every received MAVLink frame to a UDP endpoint, one frame per
datagram (same framing the SITL viewer channel uses), so Hawkeye can
@@ -285,10 +309,6 @@ def attach_viewer_tee(conn, host='127.0.0.1', port=19410):
return conn
# ---------------------------------------------------------------------------
# Reboot / device replug helpers
# ---------------------------------------------------------------------------
def send_reboot(mav):
"""Request an autopilot reboot (MAV_CMD_PREFLIGHT_REBOOT_SHUTDOWN, param1=1)."""
mav.mav.command_long_send(
@@ -379,11 +399,8 @@ def reboot_and_reconnect(mav, conn_str, baud=DEFAULT_BAUD, timeout=60):
timeout, last_err))
# ---------------------------------------------------------------------------
# Common CLI plumbing
# ---------------------------------------------------------------------------
def add_connection_args(parser, dual_link=False):
"""Standard CLI surface shared by every test in the suite."""
parser.add_argument('connection',
help='MAVLink connection: serial device (/dev/tty.usbmodem01), '
'udp:IP:PORT, or tcp:IP:PORT')
Tools/bench_test/px4bench/ftp.py
+90
View File
@@ -0,0 +1,90 @@
"""MAVFTP helpers: directory listing, bounded file download, ULog constants.
Built on pymavlink.mavftp. cmd_get only sends OpenFileRO and returns; the
transfer itself is driven by process_ftp_reply, so ftp_download pumps it in
bounded slices with its own budget and stall detection: a transfer that
stops mid-file becomes an error naming the byte count, never a hang.
"""
import time
from . import send_heartbeat
LOG_ROOT = '/fs/microsd/log'
# ULog file header magic: 'U' 'L' 'o' 'g' 0x01 0x12 0x35, then a file-version
# byte. See src/modules/logger/logger.cpp:2099-2106.
ULOG_MAGIC7 = bytes([0x55, 0x4C, 0x6F, 0x67, 0x01, 0x12, 0x35])
FTP_BUDGET = 300.0 # overall MAVFTP budget in seconds
def ftp_list(ftp, path):
"""List a remote directory. cmd_list drives its own reply loop internally
and populates ftp.list_result. Returns the list of DirectoryEntry.
"""
ftp.cmd_list([path])
return list(ftp.list_result)
def ftp_download(mav, ftp, remote, local, report):
"""Download one remote file to a local path over MAVFTP, bounded by
FTP_BUDGET and reporting a stall by byte count.
cmd_get only sends OpenFileRO and returns; the transfer itself is driven
by process_ftp_reply, which loops recv_match -> __mavlink_packet (burst
read handling, temp-file writes) -> __idle_task until idle or timeout.
We use a completion callback so we know the burst read finished, and we
pump process_ftp_reply in bounded slices so a stalled burst becomes a
named FAIL instead of a hang. process_ftp_reply requires its timeout to
be greater than idle_detection_time (default 3.7s).
Returns (elapsed_seconds, None) on success, (None, error_string) on
failure, so the caller can branch on the error explicitly instead of
ever mixing a message string into arithmetic.
"""
done = {'ok': False}
def on_complete(fh):
# fh is a BytesIO on success, or None if the session was terminated
# (failure). cmd_get with a callback keeps the payload in memory, so we
# persist it here ourselves.
if fh is None:
return
try:
fh.seek(0)
with open(local, 'wb') as out:
out.write(fh.read())
done['ok'] = True
except Exception as e: # noqa: BLE001 - surface write failures to the caller
report.info('local write failed: {}'.format(e))
ftp.cmd_get([remote], callback=on_complete)
start = time.monotonic()
last_bytes = 0
last_progress = start
next_hb = 0.0
# process_ftp_reply timeout must exceed idle_detection_time; use a short
# slice and loop so we can enforce our own budget and stall detection.
slice_timeout = max(4.0, ftp.ftp_settings.idle_detection_time + 0.5)
while not done['ok']:
now = time.monotonic()
if now > next_hb:
send_heartbeat(mav)
next_hb = now + 1.0
if now - start > FTP_BUDGET:
cur = ftp.read_total if ftp.read_total else last_bytes
return None, 'MAVFTP transfer stalled at {} bytes (budget {:.0f}s)'.format(
cur, FTP_BUDGET)
# drive one slice of the FTP state machine
ftp.process_ftp_reply('OpenFileRO', timeout=slice_timeout)
cur = ftp.read_total
if cur > last_bytes:
last_bytes = cur
last_progress = now
elif now - last_progress > 30.0 and not done['ok']:
# no forward progress for 30s and callback never fired
return None, 'MAVFTP transfer stalled at {} bytes (no progress 30s)'.format(cur)
# if the session ended without completing, process_ftp_reply keeps
# returning immediately on idle; break out via the budget/stall checks
elapsed = time.monotonic() - start
return elapsed, None
Tools/bench_test/mission_torture.py → Tools/bench_test/px4bench/missions.py
Executable → Regular
+9 -149
View File
@@ -1,18 +1,4 @@
#!/usr/bin/env python3
"""
Mission upload/download/clear torture test for on-bench PX4 hardware.
v1.18 risk area: dataman and the mission shared-state path were reworked and
a brand-new mutex now guards mission shared state. A regression there shows
up as a SILENT HANG (a MISSION_REQUEST that never arrives, a MISSION_ACK that
never comes, a download that stalls mid-list) or as a corrupted round-trip.
This script repeatedly uploads a large mission, reads it back and compares it
item-by-item, then clears it, with a hard timeout on every wait so a stall
becomes a named FAIL naming the stalled step and the seq reached.
With a second connection given, iterations alternate between the two links
(iteration parity picks the link) to exercise the new mission shared-state
mutex from two MAVLink channels.
"""Mission protocol helpers: item generation, upload/download/compare/clear.
Mission protocol (verified in src/modules/mavlink/mavlink_mission.cpp):
upload: MISSION_COUNT -> per-item MISSION_REQUEST_INT / MISSION_REQUEST
@@ -21,25 +7,15 @@ Mission protocol (verified in src/modules/mavlink/mavlink_mission.cpp):
MISSION_REQUEST_INT -> we reply, then send MISSION_ACK(ACCEPTED)
clear: MISSION_CLEAR_ALL -> MISSION_ACK(ACCEPTED)
Usage:
mission_torture.py CONNECTION [CONNECTION2] [-b BAUD]
[--baudrate2 BAUD2] [--iterations N] [--items K]
The autopilot may request items with either MISSION_REQUEST_INT or the
deprecated MISSION_REQUEST; both are answered with MISSION_ITEM_INT, and
duplicate or out-of-order requests are answered as asked.
"""
import argparse
import math
import os
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import px4bench
from px4bench import mavutil
DEFAULT_ITERATIONS = 10
DEFAULT_ITEMS = 220 # <= smallest CONFIG_NUM_MISSION_ITMES_SUPPORTED (500)
DEFAULT_BAUD2 = 57600
from pymavlink import mavutil
MAV_CMD_NAV_WAYPOINT = mavutil.mavlink.MAV_CMD_NAV_WAYPOINT
MAV_FRAME = mavutil.mavlink.MAV_FRAME_GLOBAL_RELATIVE_ALT_INT
@@ -49,11 +25,14 @@ MAV_MISSION_ACCEPTED = mavutil.mavlink.MAV_MISSION_ACCEPTED
BASE_LAT = 47.397742
BASE_LON = 8.545594
# Per-message wait cap and per-iteration transaction deadline.
STEP_TIMEOUT_S = 5.0
TRANSACTION_DEADLINE_S = 60.0
COUNT_RETRANSMITS = 3
UPLOAD_TYPES = ['MISSION_REQUEST_INT', 'MISSION_REQUEST', 'MISSION_ACK']
DOWNLOAD_COUNT_TYPES = ['MISSION_COUNT', 'MISSION_ACK']
DOWNLOAD_ITEM_TYPES = ['MISSION_ITEM_INT', 'MISSION_ITEM', 'MISSION_ACK']
def transaction_deadline(num_items):
"""Transaction budget scaled for large missions on slow links.
@@ -64,15 +43,6 @@ def transaction_deadline(num_items):
"""
return max(TRANSACTION_DEADLINE_S, 0.5 * num_items)
# Messages we care about; anything else (PARAM_VALUE, HEARTBEAT, ...) is drained.
UPLOAD_TYPES = ['MISSION_REQUEST_INT', 'MISSION_REQUEST', 'MISSION_ACK']
DOWNLOAD_COUNT_TYPES = ['MISSION_COUNT', 'MISSION_ACK']
DOWNLOAD_ITEM_TYPES = ['MISSION_ITEM_INT', 'MISSION_ITEM', 'MISSION_ACK']
# ---------------------------------------------------------------------------
# Deterministic mission generation
# ---------------------------------------------------------------------------
class Item:
"""A single generated waypoint; comparison uses the fields we round-trip."""
@@ -103,10 +73,6 @@ def generate_mission(iteration, k):
return items
# ---------------------------------------------------------------------------
# Upload
# ---------------------------------------------------------------------------
def send_item(mav, item):
mav.mav.mission_item_int_send(
mav.target_system, mav.target_component,
@@ -170,10 +136,6 @@ def upload_mission(report, mav, items, iteration):
transaction_deadline(len(items)), seq_reached))
# ---------------------------------------------------------------------------
# Download + compare
# ---------------------------------------------------------------------------
def download_mission(mav):
"""Request and collect the mission. Returns (items_or_None, duration, detail)."""
start = time.monotonic()
@@ -182,7 +144,6 @@ def download_mission(mav):
mav.mav.mission_request_list_send(
mav.target_system, mav.target_component, MAV_MISSION_TYPE_MISSION)
# Wait for MISSION_COUNT.
count = None
while time.monotonic() < deadline:
m = mav.recv_match(type=DOWNLOAD_COUNT_TYPES, blocking=True, timeout=STEP_TIMEOUT_S)
@@ -255,10 +216,6 @@ def compare_mission(expected, downloaded):
return True, 'all {} items match'.format(len(expected))
# ---------------------------------------------------------------------------
# Clear
# ---------------------------------------------------------------------------
def clear_mission(mav):
"""MISSION_CLEAR_ALL -> MISSION_ACK(ACCEPTED). Returns (ok, detail)."""
deadline = time.monotonic() + TRANSACTION_DEADLINE_S
@@ -298,100 +255,3 @@ def verify_cleared(mav):
return True, 'empty mission (ACK) after clear'
return False, 'verify-clear got MISSION_ACK type {}'.format(m.type)
return False, 'verify-clear transaction deadline hit'
# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------
def run_iteration(report, mav, iteration, items_k, link_label):
"""One full upload/download/compare/clear cycle. Returns True on full pass."""
expected = generate_mission(iteration, items_k)
up_ok, up_dur, up_detail = upload_mission(report, mav, expected, iteration)
if not report.check('iter{}_upload'.format(iteration), up_ok,
'[{}] {} ({:.1f}s)'.format(link_label, up_detail, up_dur)):
return False
downloaded, dl_dur, dl_detail = download_mission(mav)
if downloaded is None:
report.fail('iter{}_download'.format(iteration),
'[{}] {} ({:.1f}s)'.format(link_label, dl_detail, dl_dur))
return False
cmp_ok, cmp_detail = compare_mission(expected, downloaded)
if not report.check('iter{}_compare'.format(iteration), cmp_ok,
'[{}] {}'.format(link_label, cmp_detail)):
return False
clr_ok, clr_detail = clear_mission(mav)
if not report.check('iter{}_clear'.format(iteration), clr_ok,
'[{}] {}'.format(link_label, clr_detail)):
return False
vfy_ok, vfy_detail = verify_cleared(mav)
report.check('iter{}_verify_clear'.format(iteration), vfy_ok,
'[{}] {}'.format(link_label, vfy_detail))
if up_ok and downloaded is not None and cmp_ok and clr_ok and vfy_ok:
report.info('iter {} PASS [{}] upload {:.1f}s download {:.1f}s'.format(
iteration, link_label, up_dur, dl_dur))
return True
return False
def main():
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
px4bench.add_connection_args(parser, dual_link=True)
parser.add_argument('--baudrate2', type=int, default=DEFAULT_BAUD2,
help='baud rate for the second connection (default: %(default)s)')
parser.add_argument('--iterations', type=int, default=DEFAULT_ITERATIONS,
help='upload/download/clear iterations (default: %(default)s)')
parser.add_argument('--items', type=int, default=DEFAULT_ITEMS,
help='mission items per iteration (default: %(default)s)')
args = parser.parse_args()
report = px4bench.Reporter('mission_torture')
try:
mav1 = px4bench.connect(args.connection, baud=args.baudrate,
timeout=args.connect_timeout)
except (TimeoutError, OSError) as e:
report.fail('connect', 'link 1 {}: {}'.format(args.connection, e))
sys.exit(report.finish())
report.info('link 1 connected to system {} component {}'.format(
mav1.target_system, mav1.target_component))
mav2 = None
if args.connection2:
try:
mav2 = px4bench.connect(args.connection2, baud=args.baudrate2,
timeout=args.connect_timeout)
report.info('link 2 connected to system {} component {}'.format(
mav2.target_system, mav2.target_component))
except (TimeoutError, OSError) as e:
report.fail('connect', 'link 2 {}: {}'.format(args.connection2, e))
sys.exit(report.finish())
try:
for iteration in range(args.iterations):
if mav2 is not None and (iteration % 2 == 1):
mav, label = mav2, 'link2'
else:
mav, label = mav1, 'link1'
run_iteration(report, mav, iteration, args.items, label)
finally:
for m in (mav1, mav2):
try:
if m is not None:
m.close()
except Exception:
pass
sys.exit(report.finish())
if __name__ == '__main__':
main()
Tools/bench_test/px4bench/params.py
+106
View File
@@ -0,0 +1,106 @@
"""Parameter protocol helpers: PX4 int32 union encoding, read/set/echo.
PX4 transports an INT32 parameter as the raw bit pattern of the int placed
into the float param_value field. To send: pack the int as '<i', reinterpret
those 4 bytes as '<f'. To receive: pack the received float as '<f',
reinterpret as '<i'. This mirrors the union trick the firmware uses; the
param float is never a numeric conversion of the int.
Echo handling is hardware-learned: PX4 can emit more than one PARAM_VALUE
per set (handler reply plus the changed-param announcement, times the number
of mavlink instances), so callers must drain stale PARAM_VALUE messages
before a set and then match the echo by expected value, never consume it
positionally.
"""
import struct
import time
from pymavlink import mavutil
MAV_PARAM_TYPE_INT32 = mavutil.mavlink.MAV_PARAM_TYPE_INT32
SET_ECHO_TIMEOUT_S = 5.0
READ_TIMEOUT_S = 5.0
def int32_to_param_float(value):
"""Reinterpret an int32 bit pattern as the float carried in param_value."""
return struct.unpack('<f', struct.pack('<i', int(value)))[0]
def param_float_to_int32(value):
"""Reinterpret a param_value float's bit pattern as an int32."""
return struct.unpack('<i', struct.pack('<f', float(value)))[0]
def param_id_str(raw):
"""Normalize a PARAM_VALUE.param_id (bytes or str) to a plain string."""
if isinstance(raw, bytes):
raw = raw.decode('ascii', errors='replace')
return raw.rstrip('\x00')
def request_param_read(mav, name):
"""Send PARAM_REQUEST_READ by name (param_index = -1)."""
mav.mav.param_request_read_send(
mav.target_system, mav.target_component,
name.encode('ascii'), -1)
def read_param(mav, name, timeout=READ_TIMEOUT_S):
"""PARAM_REQUEST_READ then wait for the matching PARAM_VALUE.
Returns (int32_value, raw_float) or (None, None) on timeout.
"""
request_param_read(mav, name)
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
m = mav.recv_match(type='PARAM_VALUE', blocking=True,
timeout=max(0.1, deadline - time.monotonic()))
if m is None:
continue
if param_id_str(m.param_id) == name:
return param_float_to_int32(m.param_value), m.param_value
return None, None
def set_param_int32(mav, name, value):
"""PARAM_SET an int32 using the union encoding."""
mav.mav.param_set_send(
mav.target_system, mav.target_component,
name.encode('ascii'),
int32_to_param_float(value),
MAV_PARAM_TYPE_INT32)
def drain_param_values(mav):
"""Discard any queued PARAM_VALUE messages (stale echoes/broadcasts)."""
while mav.recv_match(type='PARAM_VALUE', blocking=False) is not None:
pass
def wait_param_echo(mav, name, expected, timeout=SET_ECHO_TIMEOUT_S):
"""Wait for a PARAM_VALUE for name carrying the expected int32 value.
PX4 can emit more than one PARAM_VALUE per set (handler reply plus the
changed-param announcement, times the number of mavlink instances), so
consuming a single message desyncs the harness by one echo forever.
Returns (matched, seen) where seen is every value observed for name; a
genuine wrong-echo firmware bug shows up as matched=False with the wrong
value(s) in seen.
"""
seen = []
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
m = mav.recv_match(type='PARAM_VALUE', blocking=True,
timeout=max(0.1, deadline - time.monotonic()))
if m is None:
continue
if param_id_str(m.param_id) != name:
continue
value = param_float_to_int32(m.param_value)
seen.append(value)
if value == expected:
return True, seen
return False, seen
Tools/bench_test/pyproject.toml
+24
View File
@@ -0,0 +1,24 @@
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[project]
name = "px4bench"
version = "0.1.0"
description = "PX4 bench-test suite: hang-visible release qualification on real NuttX hardware"
readme = "README.md"
license = { text = "BSD-3-Clause" }
requires-python = ">=3.8"
dependencies = [
"pymavlink>=2.4.40",
"pyserial",
]
[project.optional-dependencies]
ulog = ["pyulog"]
[project.urls]
Homepage = "https://github.com/PX4/PX4-Autopilot"
[tool.setuptools]
packages = ["px4bench"]
Tools/bench_test/run_bench_suite.py
+15 -15
View File
@@ -22,22 +22,22 @@ import sys
import time
# Test order for the suite. dual_link_forwarding is inserted after
# mission_torture only when a second connection is provided.
# Test order for the suite. link_forwarding is inserted after
# mission_stress only when a second connection is provided.
BASE_SEQUENCE = [
'boot_health',
'param_torture',
'mission_torture',
'mavftp_log',
'param_stress',
'mission_stress',
'log_transfer',
'reboot_loop',
]
DUAL_LINK_TEST = 'dual_link_forwarding'
DUAL_LINK_AFTER = 'mission_torture'
DUAL_LINK_TEST = 'link_forwarding'
DUAL_LINK_AFTER = 'mission_stress'
def script_path(name):
here = os.path.dirname(os.path.abspath(__file__))
return os.path.join(here, name + '.py')
return os.path.join(here, 'bench', name + '.py')
def build_argv(name, args):
@@ -45,7 +45,7 @@ def build_argv(name, args):
argv = [sys.executable, script_path(name), args.connection]
# tests that take a second positional connection
if name in ('mission_torture', DUAL_LINK_TEST):
if name in ('mission_stress', DUAL_LINK_TEST):
if args.connection2:
argv.append(args.connection2)
@@ -53,10 +53,8 @@ def build_argv(name, args):
argv += ['-b', str(args.baudrate)]
# per-test extra flags
if name == 'boot_health':
if name in ('boot_health', 'log_transfer'):
argv += ['--report-dir', args.report_dir]
elif name == 'mavftp_log':
argv += ['--outdir', args.report_dir]
return argv
@@ -113,8 +111,8 @@ def main():
parser.add_argument('connection',
help='MAVLink connection: serial device, udp:IP:PORT, or tcp:IP:PORT')
parser.add_argument('connection2', nargs='?', default=None,
help='optional second connection; enables dual_link_forwarding '
'and is passed to mission_torture')
help='optional second connection; enables link_forwarding '
'and is passed to mission_stress')
parser.add_argument('--baudrate', '-b', type=int, default=57600,
help='serial baud rate (default: %(default)s)')
parser.add_argument('--skip', default='',
@@ -143,7 +141,9 @@ def main():
if skip:
print(' skipping : {}'.format(', '.join(sorted(skip))))
print(' NOTE: usb_replug is interactive and is not run by this suite; '
'run it manually.')
'run it manually (bench/usb_replug.py).')
print(' NOTE: the SIH simulated flight (sih/flight_mission.py) reconfigures '
'the board and is run separately.')
print()
results = [] # (name, status, duration, detail)
Tools/bench_test/sih_flight.py → Tools/bench_test/sih/flight_mission.py
+15 -16
View File
@@ -22,17 +22,16 @@ import os
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import px4bench
from px4bench import Reporter, MavlinkShell, add_connection_args, connect
from param_torture import (read_param, set_param_int32, wait_param_echo,
drain_param_values)
import mission_torture
from mission_torture import Item, BASE_LAT, BASE_LON
import mavftp_log
from mavftp_log import LOG_ROOT, ULOG_MAGIC7
from px4bench import ftp as bench_ftp
from px4bench import missions
from px4bench.ftp import LOG_ROOT, ULOG_MAGIC7
from px4bench.missions import BASE_LAT, BASE_LON, Item
from px4bench.params import (drain_param_values, read_param, set_param_int32,
wait_param_echo)
from pymavlink import mavutil
@@ -52,7 +51,7 @@ WP_OFFSET_DEG = 0.0005 # ~55 m legs
def build_flight_mission(alt):
"""Takeoff, 3-waypoint square leg, RTL. Reuses mission_torture's Item."""
"""Takeoff, 3-waypoint square leg, RTL. Reuses px4bench.missions.Item."""
home_lat = int(BASE_LAT * 1e7)
home_lon = int(BASE_LON * 1e7)
off = int(WP_OFFSET_DEG * 1e7)
@@ -207,13 +206,13 @@ def fly(report, mav, shell, alt, report_dir):
# matches the stored mission CRC and PX4 keeps the completed progress
# (mission-resume semantics), so the vehicle arms into finished=true
# and never takes off.
ok, detail = mission_torture.clear_mission(mav)
ok, detail = missions.clear_mission(mav)
report.check('mission_clear', ok, detail)
if not ok:
return False
items = build_flight_mission(alt)
ok, duration, detail = mission_torture.upload_mission(report, mav, items, 0)
ok, duration, detail = missions.upload_mission(report, mav, items, 0)
report.check('mission_upload', ok, detail or 'uploaded in {:.1f}s'.format(duration))
if not ok:
return False
@@ -315,20 +314,20 @@ def download_flight_log(report, mav, report_dir):
try:
ftp = mavftp.MAVFTP(mav, target_system=mav.target_system,
target_component=1)
dirs = [e.name for e in mavftp_log.ftp_list(ftp, LOG_ROOT)
dirs = [e.name for e in bench_ftp.ftp_list(ftp, LOG_ROOT)
if e.is_dir and not e.name.startswith('.')]
if not dirs:
report.fail('flight_log', 'no log directories on SD')
return
log_dir = '{}/{}'.format(LOG_ROOT, sorted(dirs)[-1])
ulogs = sorted(e.name for e in mavftp_log.ftp_list(ftp, log_dir)
ulogs = sorted(e.name for e in bench_ftp.ftp_list(ftp, log_dir)
if e.name.endswith('.ulg'))
if not ulogs:
report.fail('flight_log', 'no .ulg in {}'.format(log_dir))
return
remote = '{}/{}'.format(log_dir, ulogs[-1])
local = os.path.join(report_dir, ulogs[-1])
elapsed, err = mavftp_log.ftp_download(mav, ftp, remote, local, report)
elapsed, err = bench_ftp.ftp_download(mav, ftp, remote, local, report)
if err is not None:
report.fail('flight_log', 'download failed: {}'.format(err))
return
@@ -363,8 +362,8 @@ def main():
parser.add_argument('--report-dir', default='bench_reports')
args = parser.parse_args()
report = Reporter('sih_flight')
report_dir = px4bench.make_report_dir(args.report_dir, 'sih_flight')
report = Reporter('flight_mission')
report_dir = px4bench.make_report_dir(args.report_dir, 'flight_mission')
report.info('report dir: {}'.format(report_dir))
try: