Michel2/pybricks-api
1
0
Fork 0
mirror of https://github.com/pybricks/pybricks-api.git synced 2026-07-27 19:57:02 +00:00
Code Issues Releases Activity

pybricks.hubs: Move .ble to BLERadio.

Browse Source 
Follows upstream firmware change.

Update tests and examples too.
This commit is contained in:
Laurens Valk
2026-05-29 11:12:50 +02:00
parent 7900f03574
commit a2ba8e4dcd
20 changed files with 203 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/common/extensions/requirements-static.py
+14 -8
View File
@@ -22,10 +22,17 @@ FEATURES_LARGE = FEATURES_MEDIUM | set()
# Features per hub.
HUB_FEATURES = {
"movehub": {"movehub"} | FEATURES_SMALL,
"cityhub": {"cityhub"} | FEATURES_MEDIUM,
"technichub": {"technichub", "gyro", "xbox-controller"} | FEATURES_MEDIUM,
"primehub": {"primehub", "inventorhub", "light-matrix", "gyro", "xbox-controller"}
"movehub": {"movehub", "ble"} | FEATURES_SMALL,
"cityhub": {"cityhub", "ble"} | FEATURES_MEDIUM,
"technichub": {"technichub", "gyro", "xbox-controller", "ble"} | FEATURES_MEDIUM,
"primehub": {
"primehub",
"inventorhub",
"light-matrix",
"gyro",
"xbox-controller",
"ble",
}
| FEATURES_LARGE,
"inventorhub": {
"primehub",
@@ -33,9 +40,10 @@ HUB_FEATURES = {
"light-matrix",
"gyro",
"xbox-controller",
"ble",
}
| FEATURES_LARGE,
"essentialhub": {"essentialhub", "gyro", "xbox-controller"} | FEATURES_LARGE,
"essentialhub": {"essentialhub", "gyro", "xbox-controller", "ble"} | FEATURES_LARGE,
}
@@ -94,9 +102,7 @@ class PybricksRequirementsStaticDirective(Directive):
</tbody>
</table>
</div>
""".format(
compat_row
)
""".format(compat_row)
# Return the node.
node = nodes.raw("", html, format="html")
doc/main/hubs/cityhub.rst
-31
View File
@@ -26,20 +26,6 @@ City Hub
.. automethod:: pybricks.hubs::CityHub.light.animate
.. rubric:: Using connectionless Bluetooth messaging
.. blockimg:: pybricks_blockBleBroadcast_CityHub
.. automethod:: pybricks.hubs::CityHub.ble.broadcast
.. blockimg:: pybricks_blockBleObserve_CityHub
.. automethod:: pybricks.hubs::CityHub.ble.observe
.. automethod:: pybricks.hubs::CityHub.ble.signal_strength
.. automethod:: pybricks.hubs::CityHub.ble.version
.. rubric:: Using the battery
.. blockimg:: pybricks_blockBatteryMeasure_CityHub_battery.voltage
@@ -104,23 +90,6 @@ Creating light animations
.. literalinclude::
../../../examples/pup/hub_common/build/light_animate_cityhub.py
Bluetooth examples
------------------
Broadcasting data to other hubs
*******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_broadcast_cityhub.py
Observing data from other hubs
******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_observe_cityhub.py
Button and system examples
----------------------------------
doc/main/hubs/essentialhub.rst
-31
View File
@@ -96,20 +96,6 @@ Essential Hub
.. automethod:: pybricks.hubs::EssentialHub.imu.settings
.. rubric:: Using connectionless Bluetooth messaging
.. blockimg:: pybricks_blockBleBroadcast_EssentialHub
.. automethod:: pybricks.hubs::EssentialHub.ble.broadcast
.. blockimg:: pybricks_blockBleObserve_EssentialHub
.. automethod:: pybricks.hubs::EssentialHub.ble.observe
.. automethod:: pybricks.hubs::EssentialHub.ble.signal_strength
.. automethod:: pybricks.hubs::EssentialHub.ble.version
.. rubric:: Using the battery
.. blockimg:: pybricks_blockBatteryMeasure_EssentialHub_battery.voltage
@@ -204,23 +190,6 @@ Reading acceleration and angular velocity on one axis
.. literalinclude::
../../../examples/pup/hub_common/build/imu_read_scalar_essentialhub.py
Bluetooth examples
------------------
Broadcasting data to other hubs
*******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_broadcast_essentialhub.py
Observing data from other hubs
******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_observe_essentialhub.py
System examples
----------------------------------
doc/main/hubs/movehub.rst
-31
View File
@@ -49,20 +49,6 @@ Move Hub
Changed acceleration units from m/s² to mm/s².
.. rubric:: Using connectionless Bluetooth messaging
.. blockimg:: pybricks_blockBleBroadcast_PrimeHub
.. automethod:: pybricks.hubs::PrimeHub.ble.broadcast
.. blockimg:: pybricks_blockBleObserve_PrimeHub
.. automethod:: pybricks.hubs::PrimeHub.ble.observe
.. automethod:: pybricks.hubs::MoveHub.ble.signal_strength
.. automethod:: pybricks.hubs::MoveHub.ble.version
.. rubric:: Using the battery
.. blockimg:: pybricks_blockBatteryMeasure_MoveHub_battery.voltage
@@ -129,23 +115,6 @@ Reading acceleration
.. literalinclude::
../../../examples/pup/hub_movehub/imu_read_acceleration.py
Bluetooth examples
------------------
Broadcasting data to other hubs
*******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_broadcast_movehub.py
Observing data from other hubs
******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_observe_movehub.py
Button and system examples
----------------------------------
doc/main/hubs/primehub.rst
-31
View File
@@ -150,20 +150,6 @@ Prime Hub / Inventor Hub
.. automethod:: pybricks.hubs::PrimeHub.speaker.play_notes
.. rubric:: Using connectionless Bluetooth messaging
.. blockimg:: pybricks_blockBleBroadcast_PrimeHub
.. automethod:: pybricks.hubs::PrimeHub.ble.broadcast
.. blockimg:: pybricks_blockBleObserve_PrimeHub
.. automethod:: pybricks.hubs::PrimeHub.ble.observe
.. automethod:: pybricks.hubs::PrimeHub.ble.signal_strength
.. automethod:: pybricks.hubs::PrimeHub.ble.version
.. rubric:: Using the battery
.. blockimg:: pybricks_blockBatteryMeasure_PrimeHub_battery.voltage
@@ -329,23 +315,6 @@ Reading acceleration and angular velocity on one axis
.. literalinclude::
../../../examples/pup/hub_common/build/imu_read_scalar_primehub.py
Bluetooth examples
------------------
Broadcasting data to other hubs
*******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_broadcast_primehub.py
Observing data from other hubs
******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_observe_primehub.py
System examples
----------------------------------
doc/main/hubs/technichub.rst
-31
View File
@@ -84,20 +84,6 @@ Technic Hub
.. automethod:: pybricks.hubs::TechnicHub.imu.settings
.. rubric:: Using connectionless Bluetooth messaging
.. blockimg:: pybricks_blockBleBroadcast_TechnicHub
.. automethod:: pybricks.hubs::TechnicHub.ble.broadcast
.. blockimg:: pybricks_blockBleObserve_TechnicHub
.. automethod:: pybricks.hubs::TechnicHub.ble.observe
.. automethod:: pybricks.hubs::TechnicHub.ble.signal_strength
.. automethod:: pybricks.hubs::TechnicHub.ble.version
.. rubric:: Using the battery
.. blockimg:: pybricks_blockBatteryMeasure_TechnicHub_battery.voltage
@@ -195,23 +181,6 @@ Reading acceleration and angular velocity on one axis
.. literalinclude::
../../../examples/pup/hub_common/build/imu_read_scalar_technichub.py
Bluetooth examples
------------------
Broadcasting data to other hubs
*******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_broadcast_technichub.py
Observing data from other hubs
******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_observe_technichub.py
Button and system examples
----------------------------------
doc/main/index.rst
+1
View File
@@ -60,6 +60,7 @@ above to reveal this menu.
parameters/index
tools/index
robotics
messaging/index
signaltypes
.. toctree::
doc/main/messaging/index.rst
+33
View File
@@ -0,0 +1,33 @@
:mod:`messaging <pybricks.messaging>` -- Send and receive messages
==================================================================
.. automodule:: pybricks.messaging
:no-members:
.. pybricks-requirements:: ble
.. autoclass:: pybricks.messaging.BLERadio
:no-members:
.. automethod:: pybricks.messaging.BLERadio.broadcast
.. automethod:: pybricks.messaging.BLERadio.observe
.. automethod:: pybricks.messaging.BLERadio.signal_strength
.. automethod:: pybricks.messaging.BLERadio.version
BLERadio examples
------------------
Broadcasting data to other hubs
*******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_broadcast_cityhub.py
Observing data from other hubs
******************************
.. literalinclude::
../../../examples/pup/hub_common/build/ble_observe_cityhub.py
examples/pup/hub_common/ble_broadcast.py → examples/pup/ble_radio/ble_broadcast.py
+3 -4
View File
@@ -1,11 +1,10 @@
# ThisHub = MoveHub CityHub TechnicHub PrimeHub EssentialHub
from pybricks.hubs import ThisHub
from pybricks.pupdevices import Motor
from pybricks.parameters import Port
from pybricks.tools import wait
from pybricks.messaging import BLERadio
# Initialize the hub.
hub = ThisHub(broadcast_channel=1)
radio = BLERadio(broadcast_channel=1)
# Initialize the motors.
left_motor = Motor(Port.A)
@@ -18,7 +17,7 @@ while True:
# Set the broadcast data and start broadcasting if not already doing so.
data = (left_angle, right_angle)
hub.ble.broadcast(data)
radio.broadcast(data)
# Broadcasts are only sent every 100 milliseconds, so there is no reason
# to call the broadcast() method more often than that.
examples/pup/hub_common/ble_observe.py → examples/pup/ble_radio/ble_observe.py
+7 -13
View File
@@ -1,11 +1,10 @@
# ThisHub = MoveHub CityHub TechnicHub PrimeHub EssentialHub
from pybricks.hubs import ThisHub
from pybricks.pupdevices import Motor
from pybricks.parameters import Color, Port
from pybricks.parameters import Port
from pybricks.tools import wait
from pybricks.messaging import BLERadio
# Initialize the hub.
hub = ThisHub(observe_channels=[1])
radio = BLERadio(observe_channels=[1])
# Initialize the motors.
left_motor = Motor(Port.A)
@@ -14,17 +13,12 @@ right_motor = Motor(Port.B)
while True:
# Receive broadcast from the other hub.
data = hub.ble.observe(1)
data = radio.observe(1)
if data is None:
# No data has been received in the last 1 second.
hub.light.on(Color.RED)
else:
if data is not None:
# Data was received and is less that one second old.
hub.light.on(Color.GREEN)
# *data* contains the same values in the same order
# that were passed to hub.ble.broadcast() on the
# It contains the same values in the same order
# that were passed to radio.broadcast() on the
# other hub.
left_angle, right_angle = data
jedi/tests/test_complete_city_hub.py
-13
View File
@@ -32,7 +32,6 @@ def test_hub_dot():
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"battery",
"ble",
"buttons",
"light",
"system",
@@ -81,15 +80,3 @@ def test_hub_dot_system_dot():
"shutdown",
"storage",
]
def test_hub_dot_ble_dot():
line = "hub.ble."
code = _create_snippet(line)
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"broadcast",
"observe",
"signal_strength",
"version",
]
jedi/tests/test_complete_essential_hub.py
-13
View File
@@ -32,7 +32,6 @@ def test_hub_dot():
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"battery",
"ble",
"buttons",
"charger",
"imu",
@@ -113,15 +112,3 @@ def test_hub_dot_system_dot():
"shutdown",
"storage",
]
def test_hub_dot_ble_dot():
line = "hub.ble."
code = _create_snippet(line)
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"broadcast",
"observe",
"signal_strength",
"version",
]
jedi/tests/test_complete_inventor_hub.py
-2
View File
@@ -5,7 +5,6 @@
Tests for correct code completion of the InventorHub class.
"""
import json
from pybricks_jedi import complete, CompletionItem
@@ -39,5 +38,4 @@ from test_complete_prime_hub import ( # noqa F401
test_hub_dot_light_dot,
test_hub_dot_speaker_dot,
test_hub_dot_system_dot,
test_hub_dot_ble_dot,
)
jedi/tests/test_complete_move_hub.py
-13
View File
@@ -32,7 +32,6 @@ def test_hub_dot():
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"battery",
"ble",
"buttons",
"imu",
"light",
@@ -93,15 +92,3 @@ def test_hub_dot_system_dot():
"shutdown",
"storage",
]
def test_hub_dot_ble_dot():
line = "hub.ble."
code = _create_snippet(line)
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"broadcast",
"observe",
"signal_strength",
"version",
]
jedi/tests/test_complete_prime_hub.py
-13
View File
@@ -32,7 +32,6 @@ def test_hub_dot():
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"battery",
"ble",
"buttons",
"charger",
"display",
@@ -142,15 +141,3 @@ def test_hub_dot_system_dot():
"shutdown",
"storage",
]
def test_hub_dot_ble_dot():
line = "hub.ble."
code = _create_snippet(line)
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"broadcast",
"observe",
"signal_strength",
"version",
]
jedi/tests/test_complete_technic_hub.py
-13
View File
@@ -32,7 +32,6 @@ def test_hub_dot():
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"battery",
"ble",
"buttons",
"imu",
"light",
@@ -101,15 +100,3 @@ def test_hub_dot_system_dot():
"shutdown",
"storage",
]
def test_hub_dot_ble_dot():
line = "hub.ble."
code = _create_snippet(line)
completions: list[CompletionItem] = json.loads(complete(code, 3, len(line) + 1))
assert [c["insertText"] for c in completions] == [
"broadcast",
"observe",
"signal_strength",
"version",
]
jedi/tests/test_get_signature.py
+3 -15
View File
@@ -5,7 +5,6 @@
Tests for correct signatures of the pupdevices.Motor class.
"""
from itertools import zip_longest
import json
@@ -88,20 +87,15 @@ CONSTRUCTOR_PARAMS = [
"MoveHub",
[
[
"broadcast_channel: Optional[int]=None",
"observe_channels: Sequence[int]=[]",
"top_side: Axis=Axis.Z",
"front_side: Axis=Axis.X",
]
],
),
pytest.param(
"pybricks.hubs",
"CityHub",
[
[
"broadcast_channel: Optional[int]=None",
"observe_channels: Sequence[int]=[]",
]
],
[[]],
),
pytest.param(
"pybricks.hubs",
@@ -110,8 +104,6 @@ CONSTRUCTOR_PARAMS = [
[
"top_side: Axis=Axis.Z",
"front_side: Axis=Axis.X",
"broadcast_channel: Optional[int]=None",
"observe_channels: Sequence[int]=[]",
]
],
),
@@ -122,8 +114,6 @@ CONSTRUCTOR_PARAMS = [
[
"top_side: Axis=Axis.Z",
"front_side: Axis=Axis.X",
"broadcast_channel: Optional[int]=None",
"observe_channels: Sequence[int]=[]",
]
],
),
@@ -134,8 +124,6 @@ CONSTRUCTOR_PARAMS = [
[
"top_side: Axis=Axis.Z",
"front_side: Axis=Axis.X",
"broadcast_channel: Optional[int]=None",
"observe_channels: Sequence[int]=[]",
]
],
),
src/pybricks/_common.py
-103
View File
@@ -1439,106 +1439,3 @@ class AmbientColorSensor(CommonColorSensor):
Measured color. The color is described by a hue (0--359), a
saturation (0--100), and a brightness value (0--100).
"""
class BLE:
"""
Bluetooth Low Energy.
.. versionadded:: 3.3
"""
@overload
def broadcast(
self, data: Iterable[Union[bool, int, float, str, bytes]]
) -> MaybeAwaitable:
...
@overload
def broadcast(self, data: Union[bool, int, float, str, bytes]) -> MaybeAwaitable:
...
def broadcast(
self,
data: Union[
Iterable[Union[bool, int, float, str, bytes]], bool, int, float, str, bytes
],
) -> MaybeAwaitable:
"""broadcast(data)
Starts broadcasting the given data on
the ``broadcast_channel`` you selected when initializing the hub.
Data may be of type ``int``, ``float``, ``str``, ``bytes``,
``True``, or ``False``. It can also be a list or tuple of these.
Choose ``None`` to stop broadcasting. This helps improve performance
when you don't need the broadcast feature, especially when observing
at the same time.
The total data size is quite limited (26 bytes). ``True`` and
``False`` take 1 byte each. ``float`` takes 5 bytes. ``int`` takes 2 to
5 bytes depending on how big the number is. ``str`` and ``bytes`` take
the number of bytes in the object plus one extra byte.
When multitasking, only one task can broadcast at a time. To broadcast
information from multiple tasks (or block stacks), you could use a
dedicated separate task that broadcast new values when one or more
variables change.
Args:
data: The value or values to be broadcast.
.. versionadded:: 3.3
"""
def observe(
self, channel: int
) -> Optional[
Union[
Tuple[Union[bool, int, float, str, bytes], ...],
Union[bool, int, float, str, bytes],
]
]:
"""observe(channel) -> bool | int | float | str | bytes | tuple | None
Retrieves the last observed data for a given channel.
Receiving data is more reliable when the hub is not connected
to a computer or other devices at the same time.
Args:
channel (int): The channel to observe (0 to 255).
Returns:
The received data in the same format as it was sent, or ``None``
if no recent data is available.
.. versionadded:: 3.3
"""
def signal_strength(self, channel: int) -> int:
"""signal_strength(channel) -> int: dBm
Gets the average signal strength in dBm for the given channel.
This indicates how near the broadcasting device is. Nearby devices
may have a signal strength around -40 dBm, while far away devices
might have a signal strength around -70 dBm.
Args:
channel (int): The channel number (0 to 255).
Returns:
The signal strength or ``-128`` if there is no recent observed data.
.. versionadded:: 3.3
"""
def version(self) -> str:
"""version() -> str
Gets the firmware version from the Bluetooth chip.
.. versionadded:: 3.3
"""
src/pybricks/hubs.py
+8 -76
View File
@@ -3,8 +3,6 @@
"""LEGO® Programmable Hubs."""
from typing import Sequence, Optional
from . import _common
from .ev3dev import _speaker
from .media.ev3dev import Image as _Image
@@ -41,30 +39,19 @@ class MoveHub:
imu = _common.SimpleAccelerometer()
system = _common.System()
buttons = _common.Keypad([_Button.CENTER])
ble = _common.BLE()
def __init__(
self,
broadcast_channel: Optional[int] = None,
observe_channels: Sequence[int] = [],
top_side: Axis = Axis.Z,
front_side: Axis = Axis.X,
):
"""MoveHub(top_side=Axis.Z, front_side=Axis.X, broadcast_channel=None, observe_channels=[])
"""MoveHub(top_side=Axis.Z, front_side=Axis.X)
Arguments:
top_side (Axis): The axis that passes through the *top side* of
the hub.
front_side (Axis): The axis that passes through the *front side* of
the hub.
broadcast_channel:
Channel number (0 to 255) used to broadcast data.
Choose ``None`` when not using broadcasting.
observe_channels:
A list of channels to listen to when ``hub.ble.observe()`` is
called. Listening to more channels requires more memory.
Default is an empty list (no channels).
.. versionchanged:: 3.3
Added *broadcast_channel* and *observe_channels* arguments.
"""
@@ -77,25 +64,9 @@ class CityHub:
light = _common.ColorLight()
system = _common.System()
buttons = _common.Keypad([_Button.CENTER])
ble = _common.BLE()
def __init__(
self, broadcast_channel: Optional[int] = None, observe_channels: Sequence[int] = []
):
"""CityHub(broadcast_channel=None, observe_channels=[])
Arguments:
broadcast_channel:
Channel number (0 to 255) used to broadcast data.
Choose ``None`` when not using broadcasting.
observe_channels:
A list of channels to listen to when ``hub.ble.observe()`` is
called. Listening to more channels requires more memory.
Default is an empty list (no channels).
.. versionchanged:: 3.3
Added *broadcast_channel* and *observe_channels* arguments.
"""
def __init__(self):
"""CityHub()"""
class TechnicHub:
@@ -108,16 +79,13 @@ class TechnicHub:
imu = _common.IMU()
system = _common.System()
buttons = _common.Keypad([_Button.CENTER])
ble = _common.BLE()
def __init__(
self,
top_side: Axis = Axis.Z,
front_side: Axis = Axis.X,
broadcast_channel: Optional[int] = None,
observe_channels: Sequence[int] = [],
):
"""TechnicHub(top_side=Axis.Z, front_side=Axis.X, broadcast_channel=None, observe_channels=[])
"""TechnicHub(top_side=Axis.Z, front_side=Axis.X)
Initializes the hub. Optionally, specify how the hub is
:ref:`placed in your design <robotframe>` by saying in which
@@ -129,16 +97,6 @@ class TechnicHub:
the hub.
front_side (Axis): The axis that passes through the *front side* of
the hub.
broadcast_channel:
Channel number (0 to 255) used to broadcast data.
Choose ``None`` when not using broadcasting.
observe_channels:
A list of channels to listen to when ``hub.ble.observe()`` is
called. Listening to more channels requires more memory.
Default is an empty list (no channels).
.. versionchanged:: 3.3
Added *broadcast_channel* and *observe_channels* arguments.
"""
@@ -153,16 +111,13 @@ class EssentialHub:
light = _common.ColorLight()
imu = _common.IMU()
system = _common.System()
ble = _common.BLE()
def __init__(
self,
top_side: Axis = Axis.Z,
front_side: Axis = Axis.X,
broadcast_channel: Optional[int] = None,
observe_channels: Sequence[int] = [],
):
"""EssentialHub(top_side=Axis.Z, front_side=Axis.X, broadcast_channel=None, observe_channels=[])
"""EssentialHub(top_side=Axis.Z, front_side=Axis.X)
Initializes the hub. Optionally, specify how the hub is
:ref:`placed in your design <robotframe>` by saying in which
@@ -174,16 +129,6 @@ class EssentialHub:
the hub.
front_side (Axis): The axis that passes through the *front side* of
the hub.
broadcast_channel:
Channel number (0 to 255) used to broadcast data.
Choose ``None`` when not using broadcasting.
observe_channels:
A list of channels to listen to when ``hub.ble.observe()`` is
called. Listening to more channels requires more memory.
Default is an empty list (no channels).
.. versionchanged:: 3.3
Added *broadcast_channel* and *observe_channels* arguments.
"""
pass
@@ -208,16 +153,13 @@ class PrimeHub:
speaker = _common.Speaker()
imu = _common.IMU()
system = _common.System()
ble = _common.BLE()
def __init__(
self,
top_side: Axis = Axis.Z,
front_side: Axis = Axis.X,
broadcast_channel: Optional[int] = None,
observe_channels: Sequence[int] = [],
):
"""PrimeHub(top_side=Axis.Z, front_side=Axis.X, broadcast_channel=None, observe_channels=[])
"""PrimeHub(top_side=Axis.Z, front_side=Axis.X)
Initializes the hub. Optionally, specify how the hub is
:ref:`placed in your design <robotframe>` by saying in which
@@ -229,16 +171,6 @@ class PrimeHub:
the hub.
front_side (Axis): The axis that passes through the *front side* of
the hub.
broadcast_channel:
Channel number (0 to 255) used to broadcast data.
Choose ``None`` when not using broadcasting.
observe_channels:
A list of channels to listen to when ``hub.ble.observe()`` is
called. Listening to more channels requires more memory.
Default is an empty list (no channels).
.. versionchanged:: 3.3
Added *broadcast_channel* and *observe_channels* arguments.
"""
src/pybricks/messaging.py
+134 -9
View File
@@ -1,29 +1,154 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2018-2020 The Pybricks Authors
# Copyright (c) 2018-2026 The Pybricks Authors
"""
Classes to exchange messages between EV3 bricks.
Classes to send and receive messages from another device.
"""
from __future__ import annotations
from typing import abstractmethod, TypeVar, Optional, Callable, Generic
from typing import (
abstractmethod,
TypeVar,
Callable,
Generic,
Union,
Iterable,
overload,
Optional,
Tuple,
Sequence,
TYPE_CHECKING,
)
if TYPE_CHECKING:
from ._common import (
MaybeAwaitable,
)
T = TypeVar("T")
class BLERadio:
"""
Send and receive messages without a connection using Bluetooth Low Energy.
.. versionadded:: 4.0
This used to be part of each hub class.
"""
def __init__(
self,
broadcast_channel: Optional[int] = None,
observe_channels: Sequence[int] = [],
):
"""BLERadio(broadcast_channel=None, observe_channels=[])
Arguments:
broadcast_channel:
Channel number (0 to 255) used to broadcast data.
Choose ``None`` when not using broadcasting.
observe_channels:
A list of channels to listen to when ``hub.ble.observe()`` is
called. Listening to more channels requires more memory.
Default is an empty list (no channels).
"""
@overload
def broadcast(self, data: None) -> MaybeAwaitable: ...
@overload
def broadcast(
self, data: Iterable[Union[bool, int, float, str, bytes]]
) -> MaybeAwaitable: ...
@overload
def broadcast(
self, data: Union[bool, int, float, str, bytes]
) -> MaybeAwaitable: ...
def broadcast(self, data: object) -> MaybeAwaitable:
"""broadcast(data)
Starts broadcasting the given data on the previously selected
``broadcast_channel``.
Data may be of type ``int``, ``float``, ``str``, ``bytes``,
``True``, or ``False``. It can also be a list or tuple of these.
Choose ``None`` to stop broadcasting. This helps improve performance
when you don't need the broadcast feature, especially when observing
at the same time.
The total data size is quite limited (26 bytes). ``True`` and
``False`` take 1 byte each. ``float`` takes 5 bytes. ``int`` takes 2 to
5 bytes depending on how big the number is. ``str`` and ``bytes`` take
the number of bytes in the object plus one extra byte.
When multitasking, only one task can broadcast at a time. To broadcast
information from multiple tasks (or block stacks), you could use a
dedicated separate task that broadcast new values when one or more
variables change.
Args:
data: The value or values to be broadcast.
"""
def observe(self, channel: int) -> Optional[
Union[
Tuple[Union[bool, int, float, str, bytes], ...],
Union[bool, int, float, str, bytes],
]
]:
"""observe(channel) -> bool | int | float | str | bytes | tuple | None
Retrieves the last observed data for a given channel.
Receiving data is more reliable when the hub is not connected
to a computer or other devices at the same time.
Args:
channel (int): The channel to observe (0 to 255).
Returns:
The received data in the same format as it was sent, or ``None``
if no recent data is available.
"""
def signal_strength(self, channel: int) -> int:
"""signal_strength(channel) -> int: dBm
Gets the average signal strength in dBm for the given channel.
This indicates how near the broadcasting device is. Nearby devices
may have a signal strength around -40 dBm, while far away devices
might have a signal strength around -70 dBm.
Args:
channel (int): The channel number (0 to 255).
Returns:
The signal strength or ``-128`` if there is no recent observed data.
"""
def version(self) -> str:
"""version() -> str
Gets the firmware version from the Bluetooth chip.
"""
class Connection:
@abstractmethod
def read_from_mailbox(self, name: str) -> bytes:
...
def read_from_mailbox(self, name: str) -> bytes: ...
@abstractmethod
def send_to_mailbox(self, name: str, data: bytes) -> None:
...
def send_to_mailbox(self, name: str, data: bytes) -> None: ...
@abstractmethod
def wait_for_mailbox_update(self, name: str) -> None:
...
def wait_for_mailbox_update(self, name: str) -> None: ...
class Mailbox(Generic[T]):