rich/MeshCore
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Document emergency power controls and record three-board wake tests

Browse Source
This commit is contained in:
mikecarper committed 2026-10-28 03:40:25 -07:00
1 parent c2ae465ece
commit e86710eda2
9 files changed
+1449 -5
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.github/workflows/run-unit-tests.yml
+1
View File
@@ -598,6 +598,7 @@ jobs:
python3 -B test/test_emergency_power.py -v
python3 -B test/test_emergency_power_runtime.py -v
python3 -B test/test_emergency_power_ota.py -v
python3 -B test/test_emergency_power_bench.py -v
python3 -B test/test_emergency_sleep_backends.py -v
python3 -B test/test_battery_calibration.py -v
python3 -B test/test_battery_calibration_store.py -v
docs/cli_commands.md
+6 -2
View File
@@ -3141,6 +3141,9 @@ radio sleeps and normal mesh, sensor and screen work is suspended. The CPU
uses timed sleep with brief watchdog, clock, voltage-protection and command
maintenance wakes. Active firmware updates and unfinished critical operations
defer sleep. Existing hard low-voltage protection remains in force.
On ESP32, an active native USB connection uses short CPU idle waits to keep
the console usable while the radio sleeps. Timed light sleep resumes after
USB detaches and its disconnect grace expires.
```text
get power.emergency
@@ -3161,8 +3164,9 @@ up to one ten-second maintenance interval while the node is sleeping.
`get power.emergency` reports the mode, battery estimate, desired `window`,
actual radio ownership, seconds until the next awake window, and remaining
hold. With an unset clock it reports `clock=uptime` and follows the same
15/45-minute cycle from boot until a valid UTC clock is available. Queued
hold. With an implausible clock timestamp it reports `clock=uptime` and follows
the same 15/45-minute cycle from boot. Keep the clock set to UTC: a plausible
timestamp alone does not prove it is synchronized. Queued
future traffic does not keep a critically low node awake indefinitely.
This schedule uses resumable timed sleep, preserving identity and settings.
docs/test-results/emergency_power_2026-10-28.md
+116
View File
@@ -0,0 +1,116 @@
<!-- meshcore-hosted-doc-link:start -->
<p class="meshcore-hosted-doc-link"><a href="https://mikecarper.github.io/MeshCore/test-results/emergency_power_2026-10-28/">View this page on MeshCore Docs</a>.</p>
<!-- meshcore-hosted-doc-link:end -->
# Emergency battery schedule, 28 October 2026
Native checks and **nine packaged firmware builds pass**, using
[`f4f945324`](https://github.com/mikecarper/MeshCore/commit/f4f9453248d351bd91be269ca0a58a6edb51882e),
plus the ESP32 USB recovery fix in
[`be6d636c7`](https://github.com/mikecarper/MeshCore/commit/be6d636c7).
SolarXiao 33S, RAK3401 and Heltec V4.3 passed physical sleep, override, automatic
wake and reboot-persistence trials. Current savings and operation with a real
battery at 5% were not measured.
## Behavior
Supported infrastructure images enter automatic emergency mode at **5% or
less**, and recover at **7% or more**, using a voltage-derived battery estimate
sampled every 30 seconds. Invalid readings neither activate the mode nor clear
an existing low-battery state. Existing hard low-voltage shutdown still applies.
Normal work runs during the first **15 minutes of each UTC hour**. Outside that
window, LoRa sleeps and normal forwarding, sensor and display work pauses.
With an implausible clock timestamp, the same cycle follows monotonic uptime
from boot. A plausible timestamp still needs synchronization to UTC.
Brief maintenance wakes, at intervals of **10 seconds or less**, service commands,
clock and watchdogs. Active updates and unfinished critical work defer sleep.
```text
get power.emergency
set power.emergency on
set power.awake 30
set power.awake 0
set power.emergency auto
```
`on` persists manual emergency mode; `auto` restores the default automatic
policy. Entry allows five seconds for the reply. `power.awake` grants a runtime
override of 0–10080 minutes; zero cancels it and reboot clears it. LoRa commands
require an awake window. Local USB can wait for the next maintenance wake.
Status separates the desired `window` from actual radio ownership.
nRF52 retains state through timed RTOS delay and idle sleep; ESP32 uses
timer-woken light sleep. Neither path uses a rebooting deep-sleep cycle.
ESP32 keeps an attached native USB console usable with short CPU idle waits
instead of peripheral-stopping light sleep; detached operation uses timed light sleep.
Wireless services restore their previous live subset and preserve manual
wireless-off settings. RAK3401 places its amplifier in standby and restores it
before radio wake, without cutting the shared supply rail.
Forwarding defaults move from 10/20% to **15/25%**. The exact old pair migrates
from a v1 record; other pairs, enabled controls and voltage endpoints survive.
A v2 record preserves an explicitly selected 10/20 pair. Companion and profiles
that omit battery-forwarding controls retain their existing feature scope.
## Checks
**85 native tests across eleven suites pass**: policy/CLI 7, runtime 13, platform
backends 3, OTA work 1, battery forwarding 20, build-role policy 8, wireless 4,
logging/sleep 3, sleep deadlines 3, CPU clock 9, and bench harness 14. The OTA test compiles
three configurations and executes the actual guards for all three roles.
Coverage includes rollover, failed persistence, reply holds, active transfers,
bounded serving holds, wireless restoration failures and prior Tracker ownership.
| Packaged firmware | Capability checks | Available / required RAM, bytes |
| --- | ---: | ---: |
| SolarXiao 33S Repeater Full | 11/11 | 151,348 / 82,784 |
| SolarXiao 33S Repeater Reduced | 9/9 | 152,528 / 80,736 |
| RAK3401 Repeater Full | 14/14 | 80,124 / 75,920 |
| RAK3401 Repeater Reduced | 12/12 | 82,648 / 73,872 |
| Heltec V4 Full MQTT Repeater | 16/16 | 240,112 / 238,732 |
| Heltec V4 Full MQTT Repeater, USB fix | 16/16 | 240,112 / 238,732 |
| SolarXiao 33S Room Server Full | 7/7 | 157,108 / 67,328 |
| SolarXiao 33S Room Server Reduced | 5/5 | 158,264 / 65,280 |
| Heltec V4 Sensor Full | 9/9 | 271,872 / 150,592 |
Memory admission and packaged update-path checks pass. These are linked capacity
estimates, not measurements of peak runtime heap. PlatformIO ran sequentially.
The [safe machine-readable report](emergency_power_2026-10-28.safe.json) records
ELF, package and manifest hashes. Package hashes were independently compared
with every corresponding memory-manifest entry.
## Physical trials
Each board entered manual radio sleep, woke for a real one-minute override,
returned to sleep when it expired, and retained manual mode after reboot.
Two accelerated hourly boundaries then produced queued adverts on an
independent RAW receiver before any further USB write to the target.
| Target | Independent receiver | Passive capture intervals |
| --- | --- | --- |
| SolarXiao 33S | RAK3401 | 31.586 / 31.587 s |
| RAK3401 | SolarXiao 33S | 31.586 / 31.583 s |
| Heltec V4.3 OLED | RAK3401 | 21.730 / 21.677 s |
The clock was set near the hour boundary; these intervals are evidence of
autonomous RF resumption, not a measurement of the CPU's exact wake latency.
The V4's initial build lost its USB console during off-window light sleep.
The corrected build kept USB responsive while RF slept and passed the whole
trial. WiFi/NTP was temporarily stopped so it could not reset the test clock.
All identities, existing settings and present clocks were checked after
cleanup. SolarXiao retains the tested Full Repeater in automatic mode with no
awake hold. RAK3401 has its original unified Repeater; V4 has its original Full
Companion and verified original NVS/filesystem. The Pi collector is active
with its original configuration and serial ownership.
## Limits
Trials used manual entry and USB power. They do not qualify real 5% battery
entry, current draw, or a continuous 45-minute sleep interval. V4 exercised
the USB-held CPU idle path; detached light-sleep current remains unmeasured.
Solar LTC4162 telemetry was unavailable, and the self ADC is not carrier state
of charge. Room and Sensor hooks have native/build coverage, not physical
trials. Sustained RF delivery and maximum runtime load need separate evidence.
docs/test-results/emergency_power_2026-10-28.safe.json
+513
View File
@@ -0,0 +1,513 @@
{
"schema_version": 1,
"date": "2026-10-28",
"source_commit": "f4f9453248d351bd91be269ca0a58a6edb51882e",
"implementation_commits": [
"0c7dec9f1",
"f4f945324",
"5cbff1946",
"be6d636c7"
],
"scope": "Native policy/runtime checks, selected production packaging, and three restored hardware trials; current draw and real low-battery entry remain unmeasured.",
"policy": {
"automatic_enter_percent_max": 5,
"automatic_release_percent_min": 7,
"sample_interval_ms": 30000,
"utc_awake_window_seconds": [
0,
900
],
"hour_seconds": 3600,
"fallback_clock": "boot-relative monotonic uptime",
"reply_hold_ms": 5000,
"maximum_cpu_sleep_slice_ms": 10000,
"maximum_awake_override_minutes": 10080,
"default_mode": "auto",
"manual_mode_persisted": true,
"awake_override_persisted": false,
"hard_low_voltage_protection_preserved": true,
"platform_sleep": {
"nrf52": "Retained RTOS timed delay with idle sleep and watchdog service; no SYSTEMOFF/reboot.",
"esp32": "Timed light sleep when detached; native USB uses bounded RTOS idle waits and a reserved post-wake USB service interval."
},
"forwarding_defaults": {
"repeat_below_percent": 15,
"flood_below_percent": 25,
"v1_migration": "Only the exact legacy 10/20 pair migrates to 15/25; enabled bits/endpoints and other pairs are preserved.",
"v2_custom_10_20_preserved": true
},
"commands": [
"get power.emergency",
"set power.emergency on",
"set power.emergency auto",
"set power.awake <0..10080 minutes>"
]
},
"native_tests": {
"status": "passed",
"total_test_methods": 85,
"suites": {
"test_emergency_power.py": 7,
"test_emergency_power_runtime.py": 13,
"test_emergency_sleep_backends.py": 3,
"test_emergency_power_ota.py": 1,
"test_battery_forwarding.py": 20,
"test_client_acl_role.py": 8,
"test_wireless_control.py": 4,
"test_logging_sleep_contract.py": 3,
"test_power_saving_deadlines.py": 3,
"test_emergency_power_bench.py": 14,
"test_esp32_cpu_clock_board.py": 9
},
"ota_role_test_compile_configurations": 3,
"qualification": "Native fakes validate state and call ordering; they do not measure physical current or battery state."
},
"builds": [
{
"artifact_target": "solarxiao_33S_repeater-full-ota",
"pio_environment": "solarxiao_33S_repeater",
"platform": "NRF52_PLATFORM",
"mcu": "nrf52840",
"passed": true,
"capability_checks": {
"passed": 11,
"total": 11
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "e70e0dc40d7b57b026ac79dc2bbaf5ceb1517e8ec2f92a0ae65acb71125a1634",
"available_internal_bytes": 151348,
"largest_internal_region_bytes": 151348,
"required_heap_bytes": 82784,
"required_contiguous_bytes": 19168
},
"memory_manifest_sha256": "5c9b97e0f5643fe4c48e446d373d78278fb3ce1446522f3166c1eb352a189bf4",
"capabilities_manifest_sha256": "f487331b66faaff19771d713468ff9b02f77a60cadf18d104527d3fa32332ab3",
"artifact_files": {
"solarxiao_33S_repeater-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.uf2": {
"sha256": "8baab7d62feed91eb7382f2e38304bcedabff15f92dfca889cde1d33c879359e",
"size_bytes": 1255424
},
"solarxiao_33S_repeater-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.zip": {
"sha256": "da62778bc5add51749deb9e36e5fbe781c9e277d404a3117ca655b0ab57cb6a0",
"size_bytes": 628363
}
}
},
{
"artifact_target": "solarxiao_33S_repeater-reduced-ota",
"pio_environment": "solarxiao_33S_repeater",
"platform": "NRF52_PLATFORM",
"mcu": "nrf52840",
"passed": true,
"capability_checks": {
"passed": 9,
"total": 9
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "b3568203b8550be591a0937355c4764bd3de5cdc7007be27cd2bf33a33ef8e83",
"available_internal_bytes": 152528,
"largest_internal_region_bytes": 152528,
"required_heap_bytes": 80736,
"required_contiguous_bytes": 19168
},
"memory_manifest_sha256": "b5cf910705fff6ad5333f65a9c7e282f9757ce53a1ad5d26385c70a9ff525ba5",
"capabilities_manifest_sha256": "f1148c7fbc1448bde7a6b244ac179f34c0cb47d366736eed7a1f94e3a1dbf07b",
"artifact_files": {
"solarxiao_33S_repeater-reduced-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.uf2": {
"sha256": "1f7dc7d181704b975f4ac598551ea8c8b6cebf89b301826bfb551821e33f37bd",
"size_bytes": 1181696
},
"solarxiao_33S_repeater-reduced-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.zip": {
"sha256": "fc8345308fee06c0059b8f0791ded9c04a8d3a5fdb9212fde64961fa32bb4e0d",
"size_bytes": 591611
}
}
},
{
"artifact_target": "RAK_3401_repeater_lora_ota_no_external_sensors-full-ota",
"pio_environment": "RAK_3401_repeater",
"platform": "NRF52_PLATFORM",
"mcu": "nrf52840",
"passed": true,
"capability_checks": {
"passed": 14,
"total": 14
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "8e4d72940ae57e8c4aaf4c1e14efdd42a26c9f1c717bef2dfae19ddc2217507b",
"available_internal_bytes": 80124,
"largest_internal_region_bytes": 80124,
"required_heap_bytes": 75920,
"required_contiguous_bytes": 19168
},
"memory_manifest_sha256": "fdc17070df586a0b8a5341c4934bb6697b9dfad5d12943383166211b0e442f23",
"capabilities_manifest_sha256": "fb661625a4851dfb53b889e524aec4ee4c604a4eff675f9513a47c65c32ea768",
"artifact_files": {
"RAK_3401_repeater_lora_ota_no_external_sensors-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.uf2": {
"sha256": "498fbfb12cb2abd0969456752bf39adbaa82134c53e63be3d6fa594be04cb593",
"size_bytes": 1255936
},
"RAK_3401_repeater_lora_ota_no_external_sensors-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.zip": {
"sha256": "3d979e38b38d1b4af77d68b21ca67b7a2d9cf3f2cf3ed4e35051b354dafb4807",
"size_bytes": 628603
}
}
},
{
"artifact_target": "RAK_3401_repeater_lora_ota_no_external_sensors-reduced-ota",
"pio_environment": "RAK_3401_repeater",
"platform": "NRF52_PLATFORM",
"mcu": "nrf52840",
"passed": true,
"capability_checks": {
"passed": 12,
"total": 12
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "d2862c5939afcd4401bbe8fdb31f980ee9a48e92f89b45e6e840bb3a990a5213",
"available_internal_bytes": 82648,
"largest_internal_region_bytes": 82648,
"required_heap_bytes": 73872,
"required_contiguous_bytes": 19168
},
"memory_manifest_sha256": "923ec558f2a9e4c236f543c0c31121bd601ffbaa753268d20ab52725daea37b1",
"capabilities_manifest_sha256": "4e4a91cba3cecab699ddcdb215d2b020f766fd1843d45e19244cd6dfa432ca1f",
"artifact_files": {
"RAK_3401_repeater_lora_ota_no_external_sensors-reduced-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.uf2": {
"sha256": "eac7a79110c130dfde00b771cd58dd2fd791c559b8b96df2f0369c99d8a6d742",
"size_bytes": 1137152
},
"RAK_3401_repeater_lora_ota_no_external_sensors-reduced-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.zip": {
"sha256": "81f5c2285a6407346cdcdcbf33c090ece3590ee1f497817aea07f5b1da098b1c",
"size_bytes": 569367
}
}
},
{
"artifact_target": "heltec_v4_repeater_observer_mqtt",
"pio_environment": "heltec_v4_repeater_observer_mqtt",
"platform": "ESP32_PLATFORM",
"mcu": "esp32s3",
"passed": true,
"capability_checks": {
"passed": 16,
"total": 16
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "fe7c84f61b24ed87b3451b9d7d62e5df38dd2382a7456f3b3b838901d064b13d",
"available_internal_bytes": 240112,
"largest_internal_region_bytes": 207344,
"required_heap_bytes": 238732,
"required_contiguous_bytes": 19168
},
"memory_manifest_sha256": "d8a38eceb3563314ac8659466d57e0ba225173fcbb89ed469d26f025c0e4a4eb",
"capabilities_manifest_sha256": "26c650a6e3086486a35fc7e6404f31d74e56930ab2497d36de5dff3b126f6956",
"artifact_files": {
"heltec_v4_repeater_observer_mqtt-full-usb-wifi-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.bin": {
"sha256": "e4814e3a0304be78ec26f3ec54514dced662c6a851b49508e37c2096f33b6e88",
"size_bytes": 2176680
},
"heltec_v4_repeater_observer_mqtt-full-usb-wifi-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532-merged.bin": {
"sha256": "958f0aae6cb8d87b3fcd9fe04616988b5e48b7a6ec2a521252b06333ffce8cad",
"size_bytes": 2242216
}
}
},
{
"artifact_target": "solarxiao_33S_room_server_lora_ota-full-ota",
"pio_environment": "solarxiao_33S_room_server_lora_ota",
"platform": "NRF52_PLATFORM",
"mcu": "nrf52840",
"passed": true,
"capability_checks": {
"passed": 7,
"total": 7
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "e6d9ae0909d1bb9d34cd434e0fe50ce7752672cb5b9e4651e86580c6938a831f",
"available_internal_bytes": 157108,
"largest_internal_region_bytes": 157108,
"required_heap_bytes": 67328,
"required_contiguous_bytes": 10256
},
"memory_manifest_sha256": "398652d19d46ea3011404a91cd7cd9b4d5916ae26732f239f5054fee362d9b47",
"capabilities_manifest_sha256": "e6e68d818c297f2c3dd3732c94c92314ade500f10f4384fb13ffac132ebd4398",
"artifact_files": {
"solarxiao_33S_room_server_lora_ota-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.uf2": {
"sha256": "dda7c6d03e4a7a507ea174cde1a6e7ef478e783637b2cd5144d63c670c439780",
"size_bytes": 1095680
},
"solarxiao_33S_room_server_lora_ota-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.zip": {
"sha256": "f12af1af797d502963d23c34f44c3934d3791dee2c5f2e6ac910fa6ec102063a",
"size_bytes": 548491
}
}
},
{
"artifact_target": "solarxiao_33S_room_server_lora_ota-reduced-ota",
"pio_environment": "solarxiao_33S_room_server_lora_ota",
"platform": "NRF52_PLATFORM",
"mcu": "nrf52840",
"passed": true,
"capability_checks": {
"passed": 5,
"total": 5
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "05d808a0d74a76bc0e3b14d4d54b75e059e36193f366601605b3c78bece1cbc4",
"available_internal_bytes": 158264,
"largest_internal_region_bytes": 158264,
"required_heap_bytes": 65280,
"required_contiguous_bytes": 10256
},
"memory_manifest_sha256": "1b27404c085b6dc2b6363b4e6a47ef2cfc6afc11ee32c5f1a9bf6bf22a2bf884",
"capabilities_manifest_sha256": "b64e5229c28253c2df916dbe2ac1eba65076594ca640d0b51c10d58385920412",
"artifact_files": {
"solarxiao_33S_room_server_lora_ota-reduced-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.uf2": {
"sha256": "a56a7cfae82bb28f0579206a0c19d1f75e04ff366510a01c7fb978af0a045e11",
"size_bytes": 1022976
},
"solarxiao_33S_room_server_lora_ota-reduced-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.zip": {
"sha256": "b93ba11be67c9e74fb0dab4b21310fed65f47d856652a7b037b139f01cf22867",
"size_bytes": 512115
}
}
},
{
"artifact_target": "heltec_v4_sensor",
"pio_environment": "heltec_v4_sensor",
"platform": "ESP32_PLATFORM",
"mcu": "esp32s3",
"passed": true,
"capability_checks": {
"passed": 9,
"total": 9
},
"ota_update_verified": true,
"memory": {
"passed": true,
"elf_sha256": "f8956d176b5a15df8818f0955779934340d735b7f2f8a30044dd1ddb0d574ab5",
"available_internal_bytes": 271872,
"largest_internal_region_bytes": 239104,
"required_heap_bytes": 150592,
"required_contiguous_bytes": 18464
},
"memory_manifest_sha256": "fa55caceb0882d2db0c8fca5bf12ffa5b84289d6347da99521e132646b063069",
"capabilities_manifest_sha256": "1b713a6e1073731e76cc360563087cea10f9385d69aec3992bb166aaf1498f5c",
"artifact_files": {
"heltec_v4_sensor-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532.bin": {
"sha256": "b16e69620277fa98079bcd9516a6f9d4254a04a07914dc8951d293c10dab3fb8",
"size_bytes": 1541288
},
"heltec_v4_sensor-full-ota-v1.18.0.2-halo-keymind-cascade-dev-f4f94532-merged.bin": {
"sha256": "3fe8fbe319cf6d2ba389f9fd088fdbee04d7195fafa3220d4c14a7bb87ab8a1e",
"size_bytes": 1606824
}
}
}
],
"additional_builds": {
"heltec_v4_native_usb_fix": {
"source_commit": "be6d636c79413fbadf7f23a1fc97cd9f92b76ba6",
"artifact_target": "heltec_v4_repeater_observer_mqtt",
"pio_environment": "heltec_v4_repeater_observer_mqtt",
"platform": "ESP32_PLATFORM",
"mcu": "esp32s3",
"passed": true,
"capability_checks": {
"passed": 16,
"total": 16
},
"memory": {
"passed": true,
"elf_sha256": "4c74d111ae0c535a69657ba5807f5dcc2b2e5b4b9aabdaa371f6d3a3ddb33d42",
"available_internal_bytes": 240112,
"largest_internal_region_bytes": 207344,
"required_heap_bytes": 238732,
"required_contiguous_bytes": 19168
},
"memory_manifest_sha256": "d2663a83298dbd4daa12e675a8737e8948fc16b5cdda71a9ded57f3c528e9820",
"capabilities_manifest_sha256": "afb6fc8e8f883f2b03e436461aef3c4acedbeaddef7ba55b0b42215439566c62",
"artifact_files": {
"heltec_v4_repeater_observer_mqtt-full-usb-wifi-ota-v1.18.0.2-halo-keymind-cascade-dev-be6d636c.bin": {
"sha256": "145afa4085b37f9c2eec0e5ba39c1aad045c46dfd0c476bc2ed1977d1dd0213a",
"size_bytes": 2177160
},
"heltec_v4_repeater_observer_mqtt-full-usb-wifi-ota-v1.18.0.2-halo-keymind-cascade-dev-be6d636c-merged.bin": {
"sha256": "a78920426a9f7edff7620425cca813e76df8dde228022573c744823bbbbee065",
"size_bytes": 2242696
}
}
}
},
"physical_validation": {
"status": "passed_with_documented_limits",
"schema_version": 1,
"bench_profile": {
"frequency_mhz": 909.5,
"bandwidth_khz": 500,
"spreading_factor": 5,
"coding_rate": 5,
"configured_tx_dbm": -9
},
"boards": [
{
"board": "SolarXiao 33S",
"tested_version": "v1.18.0.2-halo-keymind-cascade-dev-f4f94532 (Build: 28-Oct-2026)",
"installed_version": "v1.18.0.2-halo-keymind-cascade-dev-f4f94532 (Build: 28-Oct-2026)",
"trial_passed": true,
"manual_radio_sleep_verified": true,
"independent_witness_board": "RAK3401",
"baseline_rf_advert_received": true,
"awake_lease": {
"requested_seconds": 60,
"real_time_expiry_verified": true,
"clock_not_accelerated_during_lease": true
},
"accelerated_hourly_wake_cycles": [
{
"no_target_usb_writes_seconds": 31.586,
"independent_raw_capture_before_status_query": true,
"expected_hour_boundary_timestamp": true,
"radio_awake_after_capture": true
},
{
"no_target_usb_writes_seconds": 31.587,
"independent_raw_capture_before_status_query": true,
"expected_hour_boundary_timestamp": true,
"radio_awake_after_capture": true
}
],
"manual_mode_survived_reboot": true,
"cleanup_passed": true,
"final_state": {
"role": "Repeater",
"identity_equal": true,
"existing_settings_equal": true,
"present_clock_confirmed": true,
"emergency_cli_supported": true,
"emergency_mode": "auto",
"active": "off",
"radio": "awake",
"awake_hold_seconds": 0
}
},
{
"board": "RAK3401",
"tested_version": "v1.18.0.2-halo-keymind-cascade-dev-f4f94532 (Build: 28-Oct-2026)",
"installed_version": "v1.18.0.1-halo-keymind-cascade-dev-4c0f2acf (Build: 23-Oct-2026)",
"trial_passed": true,
"manual_radio_sleep_verified": true,
"independent_witness_board": "SolarXiao 33S",
"baseline_rf_advert_received": true,
"awake_lease": {
"requested_seconds": 60,
"real_time_expiry_verified": true,
"clock_not_accelerated_during_lease": true
},
"accelerated_hourly_wake_cycles": [
{
"no_target_usb_writes_seconds": 31.586,
"independent_raw_capture_before_status_query": true,
"expected_hour_boundary_timestamp": true,
"radio_awake_after_capture": true
},
{
"no_target_usb_writes_seconds": 31.583,
"independent_raw_capture_before_status_query": true,
"expected_hour_boundary_timestamp": true,
"radio_awake_after_capture": true
}
],
"manual_mode_survived_reboot": true,
"cleanup_passed": true,
"final_state": {
"role": "Repeater",
"identity_equal": true,
"existing_settings_equal": true,
"present_clock_confirmed": true,
"emergency_cli_supported": false,
"auto_and_zero_awake_hold_confirmed_before_original_application_restore": true
},
"original_unified_application_restored": true
},
{
"board": "Heltec V4.3 OLED",
"tested_version": "v1.18.0.2-halo-keymind-cascade-dev-observer-be6d636c (Build: 28-Oct-2026)",
"installed_version": "Companion v1.18.0.2-kiss-test-ec4aba8a (protocol 14, build 27-Oct-2026)",
"trial_passed": true,
"manual_radio_sleep_verified": true,
"independent_witness_board": "RAK3401",
"baseline_rf_advert_received": true,
"awake_lease": {
"requested_seconds": 60,
"real_time_expiry_verified": true,
"clock_not_accelerated_during_lease": true
},
"accelerated_hourly_wake_cycles": [
{
"no_target_usb_writes_seconds": 21.73,
"independent_raw_capture_before_status_query": true,
"expected_hour_boundary_timestamp": true,
"radio_awake_after_capture": true
},
{
"no_target_usb_writes_seconds": 21.677,
"independent_raw_capture_before_status_query": true,
"expected_hour_boundary_timestamp": true,
"radio_awake_after_capture": true
}
],
"manual_mode_survived_reboot": true,
"cleanup_passed": true,
"final_state": {
"role": "Full Companion",
"identity_equal": true,
"existing_settings_equal": true,
"present_clock_confirmed": true,
"emergency_cli_supported": false,
"auto_and_zero_awake_hold_confirmed_before_original_application_restore": true
},
"usb_console_usable_while_radio_asleep": true,
"usb_held_cpu_idle_fallback": true,
"wireless_runtime_subset_restored": true,
"original_application_and_data_readbacks_verified": true,
"earlier_usb_sleep_regression_reproduced_and_fixed": true
}
],
"collector": {
"active": true,
"unit_configuration_equal": true,
"serial_owned_by_collector": true
},
"bench_native_regressions": {
"tests": 14,
"passed": true
},
"measurement_limits": [
"All physical emergency tests used manual entry while USB powered; the actual 5% battery trigger and charging hysteresis were not physically exercised.",
"No current draw or power-saving percentage was measured.",
"Hourly boundaries used accelerated isolated RTC values; this was not a real 45-minute sleep-duration test.",
"V4 used the USB-held CPU idle fallback while RF/wireless slept; detached USB light-sleep current was not measured.",
"Physical tests used Repeater firmware; Room and Sensor hooks were validated separately by native tests and builds."
],
"current_draw_measured": false,
"real_five_percent_battery_qualified": false,
"continuous_45_minute_sleep_qualified": false,
"bench_source_sha256": "2c4799cb239f9d092d5cf46c0366408aa95f766f786f56af4b913e07b636418f"
}
}
docs/test-results/index.md
+1
View File
@@ -77,6 +77,7 @@ Completed validation reports, experiments, and test guides are collected here. O
## Build, storage, memory, and checklists
- [Emergency battery schedule: native and firmware build checks, 2026-10-28](emergency_power_2026-10-28.md)
- [RAK4631 Full/Reduced RAM repair and RAK3401 Room Server radio/sleep qualification, 2026-10-22](rak_release_fixes_2026-10-22.md)
- [Shared battery ADC calibration validation, 2026-10-22](battery_calibration_2026-10-22.md)
- [Heltec V2 BLE RAM reserve repair, 2026-10-22](heltec_v2_ble_memory_2026-10-22.md)
test/test_emergency_power_bench.py
+248
View File
@@ -0,0 +1,248 @@
"""Host harness guards; these do not claim hardware or battery qualification."""
from contextlib import contextmanager
import importlib.util
import json
from pathlib import Path
import tempfile
import threading
import time
import types
import unittest
from unittest import mock
PATH = Path(__file__).resolve().parents[1] / "tools/hil/emergency_power_bench.py"
SPEC = importlib.util.spec_from_file_location("emergency_power_bench", PATH)
bench = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(bench)
def reply(command, value):
return command + "\n -> > " + value + "\n"
class EmergencyBench(unittest.TestCase):
def test_status_requires_every_field_and_seconds(self):
text = "mode=manual active=on battery=5% clock=rtc window=sleep radio=sleep wake=2699s hold=0s"
state = bench.status(text)
self.assertEqual(state["wake"], 2699)
self.assertEqual(state["hold"], 0)
self.assertEqual(state["mode"], "manual")
for bad in (text.replace("2699s", "2699m"), text.replace(" radio=sleep", ""),
text.replace("mode=manual", "mode=unknown")):
with self.assertRaises(RuntimeError):
bench.status(bad)
def test_raw_advert_uses_actual_route_path_identity_and_timestamp(self):
identity = bytes(range(32))
epoch = 1793000000
payload = identity + epoch.to_bytes(4, "little") + bytes(64) + b"node"
for route in range(4):
for width in (1, 2, 3):
raw = bytes((16 | route,)) + (bytes(4) if route in (0, 3) else b"")
raw += bytes((((width - 1) << 6) | 2,)) + b"x" * (width * 2) + payload
self.assertEqual(bench.advert_timestamp("2026 RAW: " + raw.hex(), identity), epoch)
self.assertEqual(bench.advert_timestamp("R" + raw.hex(), identity), epoch)
self.assertIsNone(bench.advert_timestamp("TX RAW? " + raw.hex(), identity))
self.assertIsNone(bench.advert_timestamp("R" + raw.hex(), bytes(32)))
self.assertIsNone(bench.advert_timestamp("R" + raw[:10].hex(), identity))
for header, path in ((16, 192), (16, 127), (20, 0), (80, 0)):
self.assertIsNone(bench.advert_timestamp("R" + (bytes((header, path)) + payload).hex(), identity))
def test_reply_and_identity_require_complete_exact_key(self):
self.assertEqual(bench.public_key(reply("get public.key", "12" * 32)), b"\x12" * 32)
for value in ("12" * 31, "12" * 33, "x" * 64, "pending"):
with self.assertRaises(RuntimeError):
bench.public_key(reply("get public.key", value))
with self.assertRaises(RuntimeError):
bench.reply_value("get radio\n> partial")
def test_rxps_restore_preserves_level_or_manual_timings(self):
self.assertEqual(bench.rxps_restore("off,level=8,preamble=16,rx=39657,sleep=15975"), "set radio.rxps off")
self.assertEqual(bench.rxps_restore("on,level=8,preamble=16,rx=39657,sleep=15975"), "set radio.rxps level 8 preamble 16")
self.assertEqual(bench.rxps_restore("on,level=0,preamble=0,rx=65625,sleep=60000"), "set radio.rxps 65625 60000")
for bad in ("unknown setting", "on", "on,level=8,preamble=16,rx=1"):
with self.assertRaises(RuntimeError):
bench.rxps_restore(bad)
def test_clock_uses_real_minute_precision_cli(self):
epoch = bench.clock_epoch(reply("clock", "12:34 - 28/10/2026 UTC"))
self.assertEqual(epoch % 3600, 34 * 60)
with self.assertRaises(RuntimeError):
bench.clock_epoch(reply("clock", "OK"))
def test_passive_capture_matches_key_and_minimum_epoch_without_command(self):
identity = bytes(range(32))
witness = types.SimpleNamespace(condition=threading.Condition(), failed=False)
raw = bytes((17, 0)) + identity + (200).to_bytes(4, "little") + bytes(64)
witness.lines = [(8, "RAW: " + raw.hex()), (12, "RAW: " + raw.hex())]
observed = bench.await_advert(witness, identity, after=10, minimum_epoch=200, timeout=.1)
self.assertEqual(observed, {"received_monotonic": 12, "advert_epoch": 200})
with self.assertRaises(RuntimeError):
bench.await_advert(witness, identity, after=10, minimum_epoch=201, timeout=.01)
with self.assertRaises(RuntimeError):
bench.await_advert(witness, identity, after=10, minimum_epoch=100,
maximum_epoch=150, timeout=.01)
def test_preexisting_manual_mode_is_not_changed_during_failed_preflight(self):
calls = []
collector_events = []
class FakeConsole:
def __init__(self, serial_number, path):
self.serial_number = serial_number
def command(self, command, **kwargs):
calls.append(command)
values = {"get public.key": "12" * 32, "board": "SolarXiao 33S" if self.serial_number.startswith("9E") else "RAK 3401",
"ver": "verified build", "get power.emergency": "mode=manual active=on battery=unknown clock=rtc window=sleep radio=sleep wake=200s hold=0s"}
return reply(command, values[command])
def close(self):
pass
@contextmanager
def collector():
collector_events.append("pause")
try:
yield
finally:
collector_events.append("restore")
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
backup = root / "baseline.json"
backup.write_text(json.dumps({"usb_serial": bench.TARGETS["solarxiao33s"][0],
"queries": {"get public.key": reply("get public.key", "12" * 32)}}))
args = types.SimpleNamespace(run_dir=root / "run", identity_backup=backup,
target="solarxiao33s", witness=None, expected_version="verified build")
with mock.patch.object(bench, "Console", FakeConsole), mock.patch.object(bench, "paused_collector", collector), \
mock.patch.object(bench, "snapshot", return_value={}), mock.patch("builtins.print"):
with self.assertRaisesRegex(RuntimeError, "automatic mode"):
bench.run_trial(args)
self.assertFalse(any(command.startswith(("set ", "time ", "clkreboot")) for command in calls))
self.assertEqual(collector_events, ["pause", "restore"])
report = json.loads((args.run_dir / "report.json").read_text())
self.assertFalse(report["passed"])
self.assertEqual(report["cleanup"], "complete")
def test_default_plan_opens_no_port_or_service(self):
arguments = ["bench", "--target", "rak3401", "--expected-version", "verified", "--identity-backup", "unused.json", "--run-dir", "unused"]
with mock.patch("sys.argv", arguments), mock.patch.object(bench, "run_trial") as run, mock.patch("builtins.print"):
bench.main()
run.assert_not_called()
def test_console_initialization_delimits_stale_input_without_reset_pulse(self):
writes = []
class Port:
def __init__(self, **kwargs):
self.dtr = self.rts = None
def open(self):
self_values = (self.dtr, self.rts, self.port)
if self_values != (True, False, "/dev/ttyACM10"):
raise AssertionError(self_values)
def write(self, value):
writes.append(value)
def flush(self):
pass
def close(self):
pass
serial_module = types.SimpleNamespace(Serial=Port)
ports = types.SimpleNamespace(comports=lambda: [types.SimpleNamespace(
serial_number="9EBCE8853076EE86", interface="TinyUSB Serial", device="/dev/ttyACM10")])
with tempfile.TemporaryDirectory() as directory, \
mock.patch.dict("sys.modules", {"serial": serial_module, "serial.tools": types.SimpleNamespace(list_ports=ports)}), \
mock.patch.object(bench.subprocess, "run", return_value=types.SimpleNamespace(returncode=1, stdout=b"", stderr=b"")), \
mock.patch.object(bench.threading, "Thread"), mock.patch.object(bench.time, "sleep"):
console = bench.Console("9EBCE8853076EE86", Path(directory) / "private.log")
console.close()
self.assertEqual(writes, [b"\r"])
def test_snapshot_preserves_disabled_radio2_ownership_annotation(self):
values = {"radio": "910.5250244,62.5,7,5", "tx": "-6", "repeat": "on", "tempradio": "off",
"radio.rxps.config": "off,level=0,preamble=0,rx=65625,sleep=60000",
"usb.logging": "on", "usb.logging.always": "off", "radioat": "-none-",
"tempradioat": "-none-", "radioat2": "slots:", "tempradioat2": "slots:"}
console = types.SimpleNamespace(command=lambda command, **kwargs: reply(command, values[command[4:]]))
for disabled in ("off", "off (manual)", "off (automatic allowed)"):
values["radio2"] = disabled
self.assertEqual(bench.snapshot(console)["radio2"], disabled)
values["radio2"] = "911.150,250,7,5,rxtx,32"
with self.assertRaisesRegex(RuntimeError, "inactive"):
bench.snapshot(console)
values["radio2"] = "off"
for key in ("radioat", "tempradioat", "radioat2", "tempradioat2"):
previous = values[key]
values[key] = "1:911.5,125,7,5@1900000000"
with self.assertRaisesRegex(RuntimeError, "empty future"):
bench.snapshot(console)
values[key] = previous
def test_radio_admission_never_confuses_reboot_required_with_live_retune(self):
calls = []
def command(text):
calls.append(text)
return reply(text, "OK - reboot to apply")
console = types.SimpleNamespace(command=command)
with mock.patch.object(bench.time, "sleep") as sleep:
with self.assertRaisesRegex(RuntimeError, "saved getter"):
bench.apply_radio(console, bench.BENCH_RADIO)
self.assertEqual(calls, ["set radio " + bench.BENCH_RADIO])
sleep.assert_not_called()
def test_rf_witness_cannot_share_an_original_main_frequency(self):
target = {"radio": "910.5250244,62.5,7,5"}
bench.require_distinct_original_frequencies(target, {"radio": "917,62.5,7,5"})
with self.assertRaisesRegex(RuntimeError, "different original"):
bench.require_distinct_original_frequencies(target, {"radio": "910.525,125,8,5"})
def test_wireless_snapshot_distinguishes_pending_and_failure(self):
console = types.SimpleNamespace(command=lambda command: reply(command,
"wifi=on, bluetooth=unavailable, espnow=off (pending)"))
self.assertEqual(bench.wireless_snapshot(console),
{"wifi": "on", "bluetooth": "unavailable", "espnow": "off", "pending": True})
console.command = lambda command: reply(command,
"wifi=off, bluetooth=unavailable, espnow=off (failed: retry command)")
with self.assertRaisesRegex(RuntimeError, "failed transition"):
bench.wireless_snapshot(console)
def test_network_clock_hold_restores_only_original_runtime_subset(self):
calls = []
live = {"wifi": "on", "bluetooth": "unavailable", "espnow": "off"}
refused = [False]
def command(text, **kwargs):
calls.append(text)
if text == "get 2.4ghz":
return reply(text, ", ".join(key + "=" + value for key, value in live.items()))
if text == "get power.emergency":
return reply(text, "mode=manual active=on battery=100% clock=rtc window=awake radio=awake wake=0s hold=60s")
if text == "set wifi off force":
if not refused[0]:
refused[0] = True
return reply(text, "Error: emergency sleep; use set power.awake <minutes> first")
live["wifi"] = "off"
elif text == "set wifi on":
live["wifi"] = "on"
elif text not in ("set power.awake 0", "set power.awake 1"):
raise AssertionError(text)
return reply(text, "OK")
console = types.SimpleNamespace(command=command)
original = bench.wireless_snapshot(console)
with mock.patch.object(bench.time, "sleep"):
bench.hold_network_clock(console)
self.assertEqual(live["wifi"], "off")
bench.restore_wireless(console, original)
self.assertEqual(bench.wireless_snapshot(console), original)
self.assertEqual(calls.count("set wifi off force"), 2)
self.assertEqual(calls.count("set wifi on"), 1)
self.assertFalse(any("ssid" in value or "pwd" in value or "password" in value for value in calls))
if __name__ == "__main__":
unittest.main()
test/test_emergency_sleep_backends.py
+9
View File
@@ -23,6 +23,7 @@ HARNESS = r'''
static std::vector<std::string> events;
static bool radio_test = false, interrupt_context = false, wireless_ready = true;
static bool usb_held = false;
static bool attach_on_delay = false;
static unsigned primask = 0, basepri = 0, faultmask = 0;
static unsigned __get_IPSR() { return interrupt_context; }
static unsigned __get_PRIMASK() { return primask; }
@@ -40,6 +41,7 @@ static void vTaskDelay(TickType_t ticks) {
assert(ticks != 0);
delayed_ticks = ticks;
events.push_back("delay");
if (attach_on_delay) usb_held = true; // Model the USB SOF hook observing a new host.
}
static unsigned nrf_feeds = 0;
static constexpr uint32_t WDT_RR_RR_Reload = 0x6e524635;
@@ -112,6 +114,7 @@ struct ESP32Board : Base {
static void reset() {
events.clear(); radio_test = interrupt_context = false; wireless_ready = true;
usb_held = false;
attach_on_delay = false;
primask = basepri = faultmask = 0; nrf_nvic_state.__cr_flag = 0;
scheduler = taskSCHEDULER_RUNNING;
tick_ms = 1; delayed_ticks = nrf_feeds = 0;
@@ -293,6 +296,12 @@ int main() {
reset(); usb_held = true; wireless_ready = false;
assert(esp.emergencySleepMilliseconds(10000));
assert(delayed_ticks == 100 && !sleep_calls); // Existing host still bypasses physical USB sleep.
reset(); attach_on_delay = true;
assert(esp.emergencySleepMilliseconds(10000) && usb_held);
assert(sleep_calls == 1 && clear_calls == 1);
events.clear();
assert(esp.emergencySleepMilliseconds(10000));
assert(events == std::vector<std::string>{"delay"} && sleep_calls == 1);
}
'''
self.run_code(code, ("-DMESH_ESP32_USB_CONSOLE_COOPERATIVE=1",))
test/test_esp32_cpu_clock_board.py
+5 -3
View File
@@ -186,9 +186,11 @@ int main(){
clock_service=source.index("mesh::esp32::serviceCpuFrequencyPolicy(",loop_start)
# Emergency maintenance also services UART restoration before its
# early return. Keep this fixture anchored to the clock policy.
start=source.rindex(" board.servicePowerTestControls();",loop_start,clock_service)
end=source.index("#endif",clock_service)
service=source[start:end]
quiet_service=source.rindex(" board.servicePowerTestControls();",loop_start,clock_service)
# Execute the two real calls without importing unrelated enclosing
# feature guards (early emergency placement is tested separately).
service=source[quiet_service:source.index(";",quiet_service)+1]+"\n"
service+=source[clock_service:source.index(";",clock_service)+1]+"\n"
tests="void role_service(){\n"+service+"}\n"+r'''
int main(){
prefs.cpu_freq_mhz=160;role_service();assert(actual_mhz==160&&board.quiet_services==1);
tools/hil/emergency_power_bench.py
+550
View File
@@ -0,0 +1,550 @@
#!/usr/bin/env python3
"""Guarded USB emergency-power trials with an independent RAW radio witness.
Run on the Mercerwood Pi after separately flashing a verified application.
The default is a plan only. --execute permits CLI changes and temporarily
stops only mctomqtt.service, restoring it in finally. No flashing, erasing,
power cycling, credential changes, or firmware downloads occur here.
Private transcripts remain in a newly created0700 run directory.
"""
import argparse
from contextlib import contextmanager
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import threading
import time
TARGETS = {
"solarxiao33s": ("9EBCE8853076EE86", "SolarXiao 33S"),
"rak4631": ("9AB3B64C641BA927", "RAK4631"),
"heltec_v43": ("44:1B:F6:69:C9:C0", "Heltec V4"),
"rak3401": ("0B81C9C68D8D01B4", "RAK3401"),
}
WITNESSES = {"rak3401": "0B81C9C68D8D01B4", "rak4631": "9AB3B64C641BA927",
"solarxiao33s": "9EBCE8853076EE86"}
BENCH_RADIO = "909.5,500,5,5"
STATUS_RE = re.compile(
r"mode=(manual|auto) active=(on|off) battery=(unknown|\d+%) "
r"clock=(rtc|uptime) window=(awake|sleep) radio=(awake|sleep) "
r"wake=(\d+)s hold=(\d+)s\b")
def private_json(path, value):
with os.fdopen(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as stream:
json.dump(value, stream, indent=2)
stream.write("\n")
def status(reply):
match = STATUS_RE.search(reply)
if not match:
raise RuntimeError("Incomplete emergency-power status")
return dict(zip(("mode", "active", "battery", "clock", "window", "radio", "wake", "hold"),
(*match.groups()[:6], int(match[7]), int(match[8]))))
def reply_value(reply):
match = re.search(r"(?m)^ -> (.*)\r?$", reply)
if not match:
raise RuntimeError("Missing completed infrastructure CLI reply")
return match[1].removeprefix("> ").strip()
def public_key(reply):
value = reply_value(reply)
if not re.fullmatch(r"[0-9A-Fa-f]{64}", value):
raise RuntimeError("Cannot verify the complete public identity")
return bytes.fromhex(value)
def advert_timestamp(line, identity):
"""Parse the actual Packet wire layout; never match TX/detail log lines."""
match = re.search(r"(?:^|\s)RAW: ([0-9A-Fa-f]+)\s*$", line)
if not match:
match = re.fullmatch(r"R([0-9A-Fa-f]+)", line.strip())
if not match or len(match[1]) % 2:
return None
raw = bytes.fromhex(match[1])
if len(raw) < 2 or raw[0] >> 6 or (raw[0] >> 2) & 15 != 4:
return None
index = 5 if (raw[0] & 3) in (0, 3) else 1
if len(raw) <= index:
return None
count, width = raw[index] & 63, (raw[index] >> 6) + 1
if width == 4 or count * width > 64:
return None
index += 1 + count * width
if len(raw) < index + 100 or raw[index:index + 32] != identity:
return None
return int.from_bytes(raw[index + 32:index + 36], "little")
def rxps_restore(value):
match = re.fullmatch(r"(on|off),level=(\d+),preamble=(\d+),rx=(\d+),sleep=(\d+)", value)
if not match:
raise RuntimeError("Cannot safely preserve RX power-saving configuration")
enabled, level, preamble, rx, sleep = match.groups()
if enabled == "off":
return "set radio.rxps off"
if int(level):
return f"set radio.rxps level {level} preamble {preamble}"
return f"set radio.rxps {rx} {sleep}"
def clock_epoch(reply):
value = reply_value(reply)
try:
return int(datetime.strptime(value, "%H:%M - %d/%m/%Y UTC").replace(tzinfo=timezone.utc).timestamp())
except ValueError:
raise RuntimeError("Unrecognized current clock response") from None
class Console:
"""Single reader records RAW traffic even while no USB writes are made."""
def __init__(self, serial_number, transcript):
import serial
from serial.tools import list_ports
matches = [p for p in list_ports.comports()
if (p.serial_number or "").upper() == serial_number.upper()
and p.interface != "MeshCore Logging"]
if len(matches) != 1:
raise RuntimeError("Missing or ambiguous exact application USB identity")
guard = subprocess.run(["sudo", "-n", "fuser", matches[0].device],
capture_output=True, timeout=5)
if guard.returncode != 1 or guard.stdout or guard.stderr:
raise RuntimeError("Serial ownership could not be proved idle")
self.serial_number = serial_number
self.port = serial.Serial(port=None, baudrate=115200, timeout=.03, exclusive=True, dsrdtr=True)
self.port.dtr, self.port.rts = True, False
self.port.port = matches[0].device
self.port.open()
self.port.rts, self.port.dtr = False, True
self.log = os.fdopen(os.open(transcript, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb", buffering=0)
self.condition = threading.Condition()
self.lines = []
self.running = True
self.failed = False
self.last_write = 0
self.thread = threading.Thread(target=self._read, daemon=True)
self.thread.start()
# Boot banners and a prior host's partial input can share a line.
# End that input before the first exact-echo transaction; never replay
# a timed-out command or infer a reply from an unframed startup tail.
self.port.write(b"\r")
self.port.flush()
time.sleep(.3)
def _read(self):
pending = bytearray()
try:
while self.running:
chunk = self.port.read(min(self.port.in_waiting or 1, 4096))
if not chunk:
continue
self.log.write(chunk)
pending.extend(chunk)
if len(pending) > 65536:
raise RuntimeError("Unterminated serial line exceeds capture bound")
while b"\n" in pending:
line, _, pending = pending.partition(b"\n")
with self.condition:
self.lines.append((time.monotonic(), line.decode(errors="replace").rstrip("\r")))
self.condition.notify_all()
except Exception:
with self.condition:
self.failed = True
self.condition.notify_all()
def command(self, command, timeout=25, allow_error=False):
if not command or any(c in command for c in "\r\n\0") or self.failed:
raise RuntimeError("Unsafe command or desynchronized CLI session")
with self.condition:
first = len(self.lines)
self.port.write((command + "\r").encode("ascii"))
self.port.flush()
self.last_write = time.monotonic()
deadline = self.last_write + timeout
while time.monotonic() < deadline and not self.failed:
text = "\n".join(line for _, line in self.lines[first:]) + "\n"
echo = re.search(r"(?:^|\n)(?:> )?" + re.escape(command) + r"\n", text)
if echo:
done = re.search(r"(?m)^ -> .*\n", text[echo.end():])
if done:
reply = text[echo.start():echo.end() + done.end()]
if not allow_error and re.search(r"\b(?:Error|ERROR|ERR|Unknown)\b|(?m:^ -> \?\?:)", reply):
raise RuntimeError(f"CLI rejected {command.split()[0]} request; see private transcript")
return reply
self.condition.wait(min(.1, max(0, deadline - time.monotonic())))
self.failed = True
raise RuntimeError("CLI response incomplete; session will not be reused")
def reboot(self, command, transcript, expected_key, after_identity=None):
if command not in ("reboot", "clkreboot") or self.failed:
raise RuntimeError("Unverified reboot request")
self.port.write((command + "\r").encode("ascii"))
self.port.flush()
self.close()
time.sleep(3)
deadline = time.monotonic() + 40
while True:
try:
candidate = Console(self.serial_number, transcript)
break
except (RuntimeError, OSError):
if time.monotonic() > deadline:
raise
time.sleep(1)
try:
if public_key(candidate.command("get public.key")) != expected_key:
raise RuntimeError("Identity changed after reboot")
if after_identity:
after_identity(candidate)
except BaseException:
candidate.close()
raise
return candidate
def close(self):
self.running = False
self.thread.join(timeout=2)
self.port.close()
self.log.close()
@contextmanager
def paused_collector():
"""Restore the existing unit state without reading its credentials."""
unit = "/etc/systemd/system/mctomqtt.service"
digest = hashlib.sha256(Path(unit).read_bytes()).digest()
active = subprocess.run(["systemctl", "is-active", "--quiet", "mctomqtt.service"]).returncode == 0
try:
if active:
subprocess.run(["sudo", "-n", "systemctl", "stop", "mctomqtt.service"], check=True, timeout=30)
yield
finally:
if hashlib.sha256(Path(unit).read_bytes()).digest() != digest:
raise RuntimeError("Collector unit changed during bench run; inspect before restoring")
if active:
subprocess.run(["sudo", "-n", "systemctl", "start", "mctomqtt.service"], check=True, timeout=30)
subprocess.run(["systemctl", "is-active", "--quiet", "mctomqtt.service"], check=True, timeout=15)
def snapshot(console):
keys = ("radio", "tx", "repeat", "radio2", "tempradio", "radio.rxps.config",
"usb.logging", "usb.logging.always", "radioat", "tempradioat", "radioat2", "tempradioat2")
values = {key: reply_value(console.command("get " + key, allow_error=True)) for key in keys}
for key in ("radioat", "tempradioat", "radioat2", "tempradioat2"):
if values[key] != ("-none-" if key in ("radioat", "tempradioat") else "slots:"):
raise RuntimeError("Bench requires confirmed empty future radio schedules")
if not re.fullmatch(r"\d+(?:\.\d+)?,\d+(?:\.\d+)?,\d+,\d+", values["radio"]):
raise RuntimeError("Unrecognized primary radio configuration")
if values["tempradio"] != "off" or values["radio2"] not in ("off", "off (manual)", "off (automatic allowed)"):
raise RuntimeError("Bench requires inactive temporary and secondary radio profiles")
if values["repeat"] not in ("on", "off") or not re.fullmatch(r"-?\d+", values["tx"]):
raise RuntimeError("Cannot safely restore TX/repeater configuration")
values["rxps_restore"] = rxps_restore(values["radio.rxps.config"])
return values
def require_distinct_original_frequencies(target, witness):
# A baseline RF success must not be two failed retunes remaining together
# on a shared original channel. These guarded bench pairs start apart.
if abs(float(target["radio"].split(",")[0]) - float(witness["radio"].split(",")[0])) < .001:
raise RuntimeError("Independent witness must start on a different original main frequency")
def apply_radio(console, value):
admission = reply_value(console.command("set radio " + value))
if "reboot" in admission.lower() or "applying in 2s" not in admission:
raise RuntimeError("Live radio application not confirmed; a saved getter cannot prove an RF retune")
time.sleep(4) # CLI admits primary radio changes with a2-second reply grace.
actual = tuple(map(float, reply_value(console.command("get radio")).split(",")))
wanted = tuple(map(float, value.split(",")))
if any(abs(a - b) > .001 for a, b in zip(actual, wanted)):
raise RuntimeError("Saved primary radio differs after applying the bench profile")
def isolate(console, saved, witness=False):
console.command("set radio.rxps off")
console.command("set repeat off")
console.command("set tx -9")
apply_radio(console, BENCH_RADIO)
if witness:
console.command("set usb.logging on")
if saved["usb.logging.always"] in ("on", "off"):
console.command("set usb.logging.always on")
def restore(console, saved):
apply_radio(console, saved["radio"])
console.command("set tx " + saved["tx"])
console.command(saved["rxps_restore"])
console.command("set repeat " + saved["repeat"])
for key in ("usb.logging.always", "usb.logging"):
if saved[key] in ("on", "off"):
console.command("set " + key + " " + saved[key])
current = snapshot(console)
for key in ("radio", "tx", "repeat", "radio2", "radio.rxps.config", "usb.logging", "usb.logging.always"):
if key == "radio":
if any(abs(float(a) - float(b)) > .001 for a, b in zip(current[key].split(","), saved[key].split(","))):
raise RuntimeError("Original radio was not restored")
elif current[key] != saved[key]:
raise RuntimeError("Original " + key + " was not restored")
def await_status(console, expected, timeout=30):
deadline = time.monotonic() + timeout
while True:
observed = status(console.command("get power.emergency"))
if all(observed[key] == value for key, value in expected.items()):
return observed
if time.monotonic() > deadline:
raise RuntimeError("Emergency status did not reach " + repr(expected))
time.sleep(.5)
def wireless_snapshot(console):
value = reply_value(console.command("get 2.4ghz"))
match = re.fullmatch(r"wifi=(on|off|unavailable), bluetooth=(on|off|unavailable), "
r"espnow=(on|off|unavailable)( \(pending\))?", value)
if not match:
raise RuntimeError("Wireless runtime state is unavailable or reports a failed transition")
return dict(zip(("wifi", "bluetooth", "espnow", "pending"),
(*match.groups()[:3], bool(match[4]))))
def hold_network_clock(console):
"""Keep NTP from replacing accelerated RTC values; credentials stay saved."""
console.command("set power.awake 1")
await_status(console, {"radio": "awake", "window": "awake"})
deadline = time.monotonic() + 35
while True:
value = reply_value(console.command("set wifi off force", allow_error=True))
if value.startswith("OK"):
break
if not value.startswith("Error: emergency sleep") or time.monotonic() > deadline:
raise RuntimeError("WiFi runtime clock isolation was rejected")
time.sleep(.1) # Wait for the completed awake request to resume its backend.
while True:
live = wireless_snapshot(console)
if live["wifi"] == "off" and not live["pending"]:
break
if time.monotonic() > deadline:
raise RuntimeError("WiFi/NTP did not stop for accelerated clock testing")
time.sleep(.1)
console.command("set power.awake 0")
def restore_wireless(console, original):
for service in ("espnow", "wifi"):
live = wireless_snapshot(console)
if original[service] != live[service]:
if original[service] == "unavailable":
raise RuntimeError("Wireless availability changed during the test")
console.command("set " + service + " " + original[service]
+ (" force" if original[service] == "off" else ""))
deadline = time.monotonic() + 35
while True:
live = wireless_snapshot(console)
if live[service] == original[service] and not live["pending"]:
break
if time.monotonic() > deadline:
raise RuntimeError("Original wireless runtime subset was not restored")
time.sleep(.1)
if wireless_snapshot(console) != original:
raise RuntimeError("Original wireless runtime subset differs after restoration")
def await_advert(witness, identity, after, minimum_epoch, timeout=20, maximum_epoch=None):
deadline = time.monotonic() + timeout
with witness.condition:
while True:
for stamp, line in witness.lines:
epoch = advert_timestamp(line, identity)
if (stamp >= after and epoch is not None and epoch >= minimum_epoch
and (maximum_epoch is None or epoch < maximum_epoch)):
return {"received_monotonic": stamp, "advert_epoch": epoch}
if witness.failed or time.monotonic() > deadline:
raise RuntimeError("Independent witness did not receive the target advertisement")
witness.condition.wait(.1)
def run_trial(args):
args.run_dir.mkdir(mode=0o700, parents=False, exist_ok=False)
original = json.loads(args.identity_backup.read_text())
serial_number, board_label = TARGETS[args.target]
if original.get("usb_serial", "").upper() != serial_number.upper():
raise RuntimeError("Private identity backup is for a different USB board")
original_key = public_key(original["queries"]["get public.key"])
report = {"board": board_label, "usb_serial": serial_number,
"real_minute_lease": False, "accelerated_hourly_cycles": [],
"manual_mode_survived_reboot": False, "cleanup": "not needed",
"limitations": ["USB-powered manual entry; battery5% threshold/current savings not measured",
"Hourly boundaries accelerated with RTC; no real45-minute sleep duration claim",
"Application restoration is a separate guarded flashing step"]}
target = witness = None
target_saved = witness_saved = None
wireless_before = None
clock_guard = hold_network_clock if args.target == "heltec_v43" else None
if clock_guard:
report["limitations"].append("USB-held CPU idle fallback; detached USB light-sleep current not measured")
target_changed = witness_changed = False
changed_clock = False
try:
with paused_collector():
try:
target = Console(serial_number, args.run_dir / "target.private.log")
witness_name = args.witness or ("solarxiao33s" if args.target == "rak3401" else "rak3401")
if WITNESSES[witness_name] == serial_number:
raise RuntimeError("Target and independent witness must be different boards")
witness = Console(WITNESSES[witness_name], args.run_dir / "witness.private.log")
if public_key(target.command("get public.key")) != original_key:
raise RuntimeError("Target public identity differs from its private baseline backup")
actual_board = reply_value(target.command("board"))
if re.sub(r"[^a-z0-9]", "", board_label.lower()) not in re.sub(r"[^a-z0-9]", "", actual_board.lower()):
raise RuntimeError("Unexpected target board label")
version = reply_value(target.command("ver"))
if version != args.expected_version:
raise RuntimeError("Running firmware differs from the explicitly selected verified build")
witness_board = re.sub(r"[^a-z0-9]", "", reply_value(witness.command("board")).lower())
if re.sub(r"[^a-z0-9]", "", TARGETS[witness_name][1].lower()) not in witness_board:
raise RuntimeError("Wrong independent witness board")
report["firmware"] = version
target_saved, witness_saved = snapshot(target), snapshot(witness)
if clock_guard:
wireless_before = wireless_snapshot(target)
if wireless_before["pending"] or wireless_before["wifi"] == "unavailable":
raise RuntimeError("WiFi must expose a settled reversible runtime state")
private_json(args.run_dir / "settings.private.json", {"target": target_saved, "witness": witness_saved,
"wireless": wireless_before})
initial = status(target.command("get power.emergency"))
if initial["mode"] != "auto" or initial["hold"]:
raise RuntimeError("Target must start with automatic mode and no awake lease")
if witness_saved["usb.logging.always"] not in ("on", "off"):
raise RuntimeError("Witness needs the current logging override and live radio CLI")
require_distinct_original_frequencies(target_saved, witness_saved)
witness_changed = True
isolate(witness, witness_saved, witness=True)
target_changed = True
isolate(target, target_saved)
report["cleanup"] = "required"
target.command("advert.zerohop")
report["baseline_advert"] = await_advert(witness, original_key, target.last_write, 0)
target.command("set power.awake 0")
target.command("set power.emergency auto")
if clock_guard:
clock_guard(target)
# Past test clocks avoid future advert timestamps in the witness's
# neighbour cache. RF stays isolated until present time is proven.
future = ((int(time.time()) // 3600) - 48) * 3600
changed_clock = True
target = target.reboot("clkreboot", args.run_dir / "target.bench-clock.private.log", original_key, clock_guard)
target.command(f"time {future + 1800}")
target.command("set power.emergency on")
report["manual_entry"] = await_status(target, {"mode": "manual", "active": "on", "radio": "sleep", "window": "sleep", "hold": 0})
started = time.monotonic()
target.command("set power.awake 1")
awake = await_status(target, {"radio": "awake", "window": "awake"})
if not 30 <= awake["hold"] <= 60:
raise RuntimeError("The one-minute awake lease was not observed")
target.command("advert.zerohop")
report["lease_advert"] = await_advert(witness, original_key, target.last_write, future,
maximum_epoch=future + 3600)
time.sleep(max(0, started + 65 - time.monotonic()))
await_status(target, {"radio": "sleep", "window": "sleep", "hold": 0})
report["real_minute_lease"] = True
for cycle in range(2):
cycle_hour = future + cycle * 3600
if cycle:
target.command(f"time {cycle_hour + 1800}")
await_status(target, {"radio": "sleep", "window": "sleep"})
# Defer signing/dispatch until the mesh loop runs at timer wake.
target.command("advert.zerohop")
target.command(f"time {cycle_hour + 3580}")
last_write = target.last_write
received = await_advert(witness, original_key, last_write, cycle_hour + 3600, timeout=45,
maximum_epoch=cycle_hour + 3660)
if target.last_write != last_write:
raise RuntimeError("Target USB write occurred during the passive timer-wake interval")
received["no_usb_write_seconds"] = time.monotonic() - last_write
received["status_after_passive_capture"] = await_status(target, {"radio": "awake", "window": "awake"})
report["accelerated_hourly_cycles"].append(received)
target.command(f"time {future + 2 * 3600 + 1800}")
await_status(target, {"mode": "manual", "radio": "sleep"})
target = target.reboot("reboot", args.run_dir / "target.after-reboot.private.log", original_key, clock_guard)
report["after_reboot"] = await_status(target, {"mode": "manual", "active": "on", "hold": 0})
target.command(f"time {future + 3 * 3600 + 1800}")
report["after_reboot_sleep"] = await_status(target, {"mode": "manual", "radio": "sleep", "window": "sleep"})
report["manual_mode_survived_reboot"] = True
report["passed"] = True
finally:
cleanup_errors = []
if target and target_changed:
try:
target.command("set power.awake 0")
target.command("set power.emergency auto")
await_status(target, {"mode": "auto", "radio": "awake"})
if changed_clock:
target = target.reboot("clkreboot", args.run_dir / "target.clock-restored.private.log", original_key, clock_guard)
target.command(f"time {int(time.time())}")
current_time = clock_epoch(target.command("clock"))
if abs(current_time - time.time()) > 90:
raise RuntimeError("Present clock not confirmed; retaining isolated RF")
restore(target, target_saved)
report["final_power"] = status(target.command("get power.emergency"))
if wireless_before:
restore_wireless(target, wireless_before)
report["wireless_runtime_restored"] = True
except Exception as error:
cleanup_errors.append("Target cleanup: " + str(error))
if witness and witness_changed:
try:
witness_key = public_key(witness.command("get public.key"))
if abs(clock_epoch(witness.command("clock")) - time.time()) > 90:
witness = witness.reboot("clkreboot", args.run_dir / "witness.clock-restored.private.log", witness_key)
witness.command(f"time {int(time.time())}")
if abs(clock_epoch(witness.command("clock")) - time.time()) > 90:
raise RuntimeError("Witness present clock not confirmed; retaining isolated RF")
restore(witness, witness_saved)
except Exception as error:
cleanup_errors.append("Witness cleanup: " + str(error))
for console in (target, witness):
if console:
console.close()
report["cleanup"] = "complete" if not cleanup_errors else "NEEDS MANUAL RECOVERY"
if cleanup_errors:
report["cleanup_errors"] = cleanup_errors
raise RuntimeError("Bench cleanup incomplete; inspect the private report before further tests")
except BaseException as error:
report["passed"] = False
report["error"] = str(error)
raise
finally:
report["finished_utc"] = datetime.now(timezone.utc).isoformat()
private_json(args.run_dir / "report.json", report)
print(json.dumps({key: report.get(key) for key in ("board", "passed", "cleanup", "real_minute_lease", "manual_mode_survived_reboot")}))
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--target", choices=TARGETS, required=True)
parser.add_argument("--witness", choices=WITNESSES, help="Default RAK3401, or qualified SolarXiao when testing the3401")
parser.add_argument("--expected-version", required=True, help="Exact ver response from the verified new build")
parser.add_argument("--identity-backup", type=Path, required=True, help="Private preflash JSON for the selected USB identity")
parser.add_argument("--run-dir", type=Path, required=True, help="New private run directory (must not already exist)")
parser.add_argument("--execute", action="store_true", help="Permit the documented CLI changes and collector pause")
args = parser.parse_args()
if not args.execute:
print("Plan only: isolated909.5/BW500/SF5/CR5, manual sleep, real60s lease, two accelerated hourly timer wakes, reboot persistence, finally restore clock/RF/collector. No hardware touched.")
return
run_trial(args)
if __name__ == "__main__":
main()