Pybricks
Code works perfectly on pybricksIDE, but fails when compiled and uploaded via PlatformIO and script/pybricksdev.exe #2818
Unanswered
toyotafj40
asked this question in
Q&A
Replies: 3 comments 1 reply
|
With a fresh mind this morning, I continued testing and found the test code also fails when uploaded via the Pybricks IDE. **And now for a curveball! ** Whilst testing, I also tried the identical code on the Pybricks Beta v3.1.0-beta.4 with firmware v4.1.0b3 and it doesn't run at all. In fact, the newest beta was problematic as soon as it was introduced. Anyway, I made some improvements to the code to show a gray LED when the Broadcast is active, but still can't get reliable transmission when in standalone mode. Expected output is correct when gatt service/debug is running (Single session) Typical bad output when running in standalone mode ( 4x sessions) |
0 replies
|
I did a simple measurement here with a sniffer on a pico2w. broadcast_rate_test.py"""
broadcast_rate_test.py
Runs on the Pybricks City Hub.
Minimal test program for comparing the Pybricks broadcast transmit rate
- when the hub is BLE-connected to a PC running Pybricks Code
versus
- standalone (started without a host connection).
Broadcasts (counter, host_connected_ble)
so the data shows if connected.
Holds each value to be transmitted for HOLD_S seconds, long enough
that ble_sniffer.py sees several repeats of the same payload before it changes.
"""
from pybricks.hubs import ThisHub
from pybricks.messaging import BLERadio
from pybricks.tools import wait
CHANNEL = 0x66
HOLD_S = 5 # seconds to hold each value before changing it
hub = ThisHub()
radio = BLERadio(broadcast_channel=CHANNEL)
counter = 0
while True:
connected = hub.system.info()["host_connected_ble"]
radio.broadcast((counter, connected))
wait(HOLD_S * 1000)
counter += 1ble_sniffer.py>"""
ble_sniffer.py
Minimal passive BLE scanner for Pico 2 W (MicroPython) that detects
Pybricks broadcast advertisements, LEGO manufacturer data, company ID
0x0397, and prints the decoded values.
Requires ble_pybricks.py (Kees Smit) in the same directory on the board.
"""
import time
import bluetooth
from micropython import const
from ble_pybricks import check_lego_manufacturer_id, get_list_of_values
_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)
_ADV_TYPE_MANUFACTURER_DATA = const(0xFF)
# RSSI is mostly noise, so drop the report
SHOW_RSSI = False
# Burst tracking: repeats of the same (channel, values) payload get counted
# so the underlying advertisement repeat rate is visible without manually
# eyeballing timestamps.
_burst_key = None
_burst_first_ts = 0
_burst_last_ts = 0
_burst_count = 0
# Per-connection-state aggregates, keyed by the "host_connected_ble" flag
# (True/False) if the payload carries one as its 2nd value (see
# broadcast_rate_test.py), else keyed by None ("unknown"). This is what
# produces the final summary - the thing to actually compare across a
# tethered run and a standalone run.
_stats = {}
def _record_packet(conn, elapsed, is_repeat, gap):
global _stats
s = _stats.get(conn)
if s is None:
s = {"count": 0, "first_ts": elapsed, "last_ts": elapsed,
"gap_sum": 0, "gap_count": 0, "min_gap": None, "max_gap": None}
_stats[conn] = s
s["count"] += 1
s["last_ts"] = elapsed
if is_repeat:
s["gap_sum"] += gap
s["gap_count"] += 1
if s["min_gap"] is None or gap < s["min_gap"]:
s["min_gap"] = gap
if s["max_gap"] is None or gap > s["max_gap"]:
s["max_gap"] = gap
def _flush_burst():
global _burst_count
if _burst_count > 1:
span = time.ticks_diff(_burst_last_ts, _burst_first_ts)
avg_gap = span / (_burst_count - 1)
print(" -- burst: {} packets, span {}ms, avg gap {:.0f}ms".format(
_burst_count, span, avg_gap))
elif _burst_count == 1:
print(" -- single packet, no repeat seen")
_burst_count = 0
def _print_final_stats():
print()
print("=== summary (grouped by host_connected_ble) ===")
if not _stats:
print("no Pybricks broadcasts received")
return
labels = {True: "connected", False: "standalone", None: "unknown"}
for conn in (True, False, None):
s = _stats.get(conn)
if not s:
continue
duration = time.ticks_diff(s["last_ts"], s["first_ts"])
rate = (s["count"] / (duration / 1000)) if duration > 0 else float("nan")
line = "{:>10}: {:4d} packets over {:6d}ms ({:.2f} pkt/s)".format(
labels[conn], s["count"], duration, rate)
if s["gap_count"]:
avg_gap = s["gap_sum"] / s["gap_count"]
line += " repeat gap avg {:.0f}ms (min {}, max {})".format(
avg_gap, s["min_gap"], s["max_gap"])
else:
line += " no repeats observed"
print(line)
def _find_adv_structure(payload, adv_type):
"""Return the raw AD structure (length byte + type byte + data) for
adv_type, or None. Keeping the length/type header intact matches what
ble_pybricks.py expects (it indexes adv_data[0]=length, [1]=type,
[2:4]=company id)."""
i = 0
n = len(payload)
while i + 1 < n:
length = payload[i]
if length == 0:
break
if payload[i + 1] == adv_type:
return bytes(payload[i:i + 1 + length])
i += 1 + length
return None
def _bt_irq(event, data):
if event == _IRQ_SCAN_RESULT:
addr_type, addr, adv_type, rssi, adv_data = data
manu = _find_adv_structure(adv_data, _ADV_TYPE_MANUFACTURER_DATA)
if manu and check_lego_manufacturer_id(manu):
channel = manu[4] # Pybricks broadcast channel, prefix to the data
elapsed = time.ticks_diff(time.ticks_ms(), _start_ms)
try:
values = get_list_of_values(manu)
except Exception as e:
print("decode error:", e)
return
global _burst_key, _burst_first_ts, _burst_last_ts, _burst_count
key = (channel, repr(values))
if key == _burst_key:
gap = time.ticks_diff(elapsed, _burst_last_ts)
is_repeat = True
else:
_flush_burst()
_burst_key = key
_burst_first_ts = elapsed
_burst_count = 0
gap = 0
is_repeat = False
_burst_count += 1
_burst_last_ts = elapsed
# values == [counter, host_connected_ble] when broadcast_rate_test.py
# is the sender; anything else (wrong length, plain int, etc.)
# falls into the "unknown" bucket rather than raising.
conn = values[1] if isinstance(values, list) and len(values) >= 2 else None
_record_packet(conn, elapsed, is_repeat, gap)
if SHOW_RSSI:
print("{:5d} Channel {} (RSSI {:4d}): {}".format(elapsed, channel, rssi, values))
else:
print("{:5d} Channel {}: {}".format(elapsed, channel, values))
elif event == _IRQ_SCAN_DONE:
# duration_ms=0 below means continuous, so this normally won't
# fire, but restart defensively if it ever does.
ble.gap_scan(0, 30000, 30000, False)
ble = bluetooth.BLE()
ble.active(True)
ble.irq(_bt_irq)
print("Scanning for Pybricks broadcasts (Ctrl-C to stop)...")
_start_ms = time.ticks_ms()
# duration_ms=0 -> scan forever, interval_us=30000, window_us=30000,
# active_scan=False -> passive (no scan requests, matches Pybricks observe)
ble.gap_scan(0, 30000, 30000, False)
# gap_scan() is non-blocking: it schedules background scanning and returns
# immediately. Without something keeping the script alive here, execution
# reaches the end of the file and the run tool (mpremote/Thonny) resets the
# board, killing the scan before it produces anything.
try:
while True:
time.sleep_ms(500)
except KeyboardInterrupt:
print("stopping scan")
_flush_burst()
ble.gap_scan(None)
ble.active(False)
_print_final_stats()Two tests show no real difference between connected and standalone: Could you test with v4.1.b2 frmware, see v4.1.b2 artifacts |
1 reply
|
You missed the important 'RECEIVE at the same time', as TRANSMIT, as per My own testing last year confirmed sending OR receiving ONLY, as a separate function for the BLERadio worked fine and as per your 36/37ms results above. My current implementation operates in SINGLE mode, default as receive only, then switch to TX for 200mS with NO RECEIVE, then back to receive only. Each slave hub is also polled every 30 seconds for battery voltage and current along with occasional other commands which will send replies. During that response time the 'remote button presses' need to be buffered for 200mS (slave hub response period), which degrades the user experience if fast button presses occur. I was hoping to implement a true full-duplex scenario, where commands are always received, especially when responses are being sent. The Full-Duplex requirement became necessary when implementing a 'motor ramp' command, which as an example, slows down a train from 50% speed to 20% speed over 5 seconds. With multiple hubs in the train, it is preferred to have all hubs in sync when ramonig down the motor speed over 5 seconds. Due to BLERadio Tx latency and measured send to RF transmit times between 33mS and 200mS, knowing when to apply the motor ramp command to the Master Hub is proving difficult. If the Master Hub could listen as a slave, then a synchronised motor ramp command has a better chance of occurring. An alternate approach is to sync all hubs and then execute 'ramp' commands at set times, but syncing all hubs is a challenge in itself, given all the BLE latency. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
PlatformIO is my preferred app for code development but lately I have been tripped up by some weird behavior after the code is uploaded to the hub, in my case a CityHub.
For example, this code is intended to broadcast bursts of packets each second, but stopping the broadcast after ideally 4 packets are sent. But given there is no 'counter' option for packets sent, I opted for 200mS as the send duration,
Watching the packets on a sniffer and then counting them reveals:
44 packets sent over 8 seconds (session) and all is perfect!
(Column 1 is milliseconds)
I then run the hub in standalone mode (just powered off and on with no BLE/gatt) and the packet count drops significantly
after the first session is sent... only 15 packets get sent and some packets are totally lost (04 is missing)
The exact same code works perfectly with or without gatt on the hub when uploaded via PybricksIDE
Here is the code:
Not sure if this is something that can be fixed or whether I will just need to cut the code in PlatformIO and then Upload in Pybricks.
Trouble is I just wasted a day trying to get the code to work / read through lots of forum posts to find something new and then finally tried the code on the Pybricks IDE and everything started working!
This code was actually part of test code I am writing to prove the ability of the BLE Rx to receive the BLE Tx.
Previously my code (uploaded via PlatformIO) caused all sorts of issues in the BLE comms and sporadic packet count when sending a reply. Allowing 100mS of BLE burst/broadcast would often not send anything, so that time is now 300mS and occasionally only one packet gets through or occasionally packet loss.
Tomorrow I will retest the original code with Pybricks direct upload...
Also, do you have any example code of a hub receiving its own transmission and possibly counting the packets sent?
Regards
Peter
All reactions