auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-04-16 00:25:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2025-02-05 16:23:47 -05:00
parent 9756572c93
commit efae307b3d
6 changed files with 27 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
examples/run_gatt_with_adapters.py
View File

@@ -25,7 +25,7 @@ import struct
import sys
from typing import Any, List, Union
from bumble.device import Connection, Device, Peer
from bumble.device import Device, Peer
from bumble import transport
from bumble import gatt
from bumble import hci
@@ -82,19 +82,19 @@ async def client(device: Device, address: hci.Address) -> None:
for index in range(1, 9):
characteristics.append(
service.get_characteristics_by_uuid(
CHARACTERISTIC_UUID_BASE + f"{index:02X}"
core.UUID(CHARACTERISTIC_UUID_BASE + f"{index:02X}")
)[0]
)
# Read all characteristics as raw bytes.
for characteristic in characteristics:
value = await characteristic.read_value()
print(f"### {characteristic} = {value} ({value.hex()})")
print(f"### {characteristic} = {value!r} ({value.hex()})")
# Static characteristic with a bytes value.
c1 = characteristics[0]
c1_value = await c1.read_value()
print(f"@@@ C1 {c1} value = {c1_value} (type={type(c1_value)})")
print(f"@@@ C1 {c1} value = {c1_value!r} (type={type(c1_value)})")
await c1.write_value("happy π day".encode("utf-8"))
# Static characteristic with a string value.
@@ -136,7 +136,7 @@ async def client(device: Device, address: hci.Address) -> None:
# Dynamic characteristic with a bytes value.
c7 = characteristics[6]
c7_value = await c7.read_value()
print(f"@@@ C7 {c7} value = {c7_value} (type={type(c7_value)})")
print(f"@@@ C7 {c7} value = {c7_value!r} (type={type(c7_value)})")
await c7.write_value(bytes.fromhex("01020304"))
# Dynamic characteristic with a string value.

14
examples/run_vcp_renderer.py
View File

@@ -174,16 +174,10 @@ async def main() -> None:
ws = websocket
async for message in websocket:
volume_state = json.loads(message)
vcs.volume_state_bytes = bytes(
[
volume_state['volume_setting'],
volume_state['muted'],
volume_state['change_counter'],
]
)
await device.notify_subscribers(
vcs.volume_state, vcs.volume_state_bytes
)
vcs.volume_setting = volume_state['volume_setting']
vcs.muted = volume_state['muted']
vcs.change_counter = volume_state['change_counter']
await device.notify_subscribers(vcs.volume_state)
ws = None
await websockets.serve(serve, 'localhost', 8989)