B4J Question [PyBridge] How to use BLE with Bleak

[rwblinn]

Have started to explore the new PyBridge and how to use as a B4J client of an ESP32 running as BLE server using the B4R library rBLEServer.
Created a simple Python script using Bleak.
This script could run as an external process with jShell and handle bidirectional communication, but that is probably the old way since the new PyBridge has been developed.

Questions
How to use this kind of communication with PyBridge? or is the PyBridge not designed for this?
How to handle sending data or receiving data from the BLEServer?
Any guidance appreciated.

import asyncio
import argparse
from bleak import BleakScanner, BleakClient
import struct

ESP32_NAME = "BLEServer"
ESP32_MAC = "30:C9:22:D1:80:2E"
SERVICE_UUID = "6E400001-B5A3-F393-E0A9-E50E24DCCA9E"  # UART Service UUID
RX_CHARACTERISTIC_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"  # RX Notify & Write Characteristic

async def scan_and_parse():
    def callback(device, advertisement_data):
        if device.name == ESP32_NAME or device.address == ESP32_MAC:
            service_data = advertisement_data.service_data.get(SERVICE_UUID)
            if service_data and len(service_data) >= 4:
                int1, int2 = struct.unpack("hh", service_data[:4])
                print(f"Received: {int1}, {int2}")
   
    scanner = BleakScanner(callback)
    await scanner.start()
    await asyncio.sleep(10)  # Scan for 10 seconds
    await scanner.stop()

async def connect():
    client = BleakClient(ESP32_MAC)
    await client.connect()
    if client.is_connected:
        print("Connected to ESP32")
    return client

async def write(client, command: bytes, commanddata: bytes):
    if client.is_connected:
        await client.write_gatt_char(RX_CHARACTERISTIC_UUID, command + commanddata)
        print(f"Wrote command: {command.hex()} data: {commanddata.hex()}")

async def main(command: bytes, commanddata: bytes):
    await scan_and_parse()
    client = await connect()
    await write(client, command, commanddata)
    await client.disconnect()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Send command to ESP32 BLE server")
    parser.add_argument("-c", "--command", required=True, type=lambda x: bytes.fromhex(x), help="Command in hex format (e.g., 01)")
    parser.add_argument("-d", "--commanddata", required=True, type=lambda x: bytes.fromhex(x), help="Command data in hex format (e.g., 02)")
    args = parser.parse_args()
   
    asyncio.run(main(args.command, args.commanddata))

 

[teddybear]

Questions
How to use this kind of communication with PyBridge? or is the PyBridge not designed for this?
How to handle sending data or receiving data from the BLEServer?
Any guidance appreciated.

I don't think it can do that for you. see [PyBridge] The very basics, you need to handle bidirectional communication. but

Note that the flow is generally unidirectional - B4J sends commands to Python. The commands are also unidirectional. The allows the commands to be queued, and B4J doesn't wait for the commands to actually be executed. It is very important in terms of performance.

 

[Erel]

Note that PyBridge does support raising events.

And, just yesterday I've ordered a BLE adapter to test Bleak...

Once it arrives, and will hopefully work, I will publish the code.

 

[Daestrum]

Bleak appears to work ok - I made a tiny test prog
Python code

import asyncio
from bleak import BleakScanner
async def main():
    devices = await BleakScanner.discover()
    print("found devices {}".format(len(devices)))
    for d in devices:
        print(d)

B4J(PyBridge) code

Sub run_scanner() As ResumableSub
    wait for (py.RunCodeAwait("main",Array(),File.ReadString(File.DirAssets,"my_scanner.py"))) Complete (res As PyWrapper)
    Return True
End Sub

Result ( on lappy)

Waiting for debugger to connect...
Program started.
Server is listening on port: 51703
Python path: d:/python.3.12.8/python.exe
connected
starting PyBridge v0.50
watchdog set to 30 seconds
Connecting to port: 51703

found devices 12
.........................................: [LG] webOS TV NANO916PA
.........................................: [TV] Samsung 7 Series (43)
.........................................: lime
.........................................: None
.........................................: None
...
.........................................: [AV] Samsung Soundbar MS650

 

[Mariano Ismael Castro]

It works fine, I also started doing some tests.
I still need to understand how events work in PyBridge, but this is exciting.
I am attaching my progress, keep in mind that I am still learning and therefore the code may not be correct.

 

[Erel]

Event example:

Private Sub Py_Tick (Args As Map)
    Dim time As String = Args.Get("time")
    Log(time)
End Sub

Private Sub TimerExample
    Dim Code As String = $"
import asyncio
from datetime import datetime
async def TimerExample ():
    while True:
        await asyncio.sleep(1)
        bridge_instance.raise_event("tick", {"time": str(datetime.now())})
"$
    Wait For (Py.RunCodeAwait("TimerExample", Array(), Code)) Complete (Result As PyWrapper)
End Sub

It is important to use async methods, and not block the Python thread.

 

[Daestrum]

So the event name is the bit after Py_ in the sub name.

Sub Py_Check(Args as Map)

Would be raised by

bridge_instance.raise_event("check", {"something": str(someValue)})

 

[Erel]

Yes. Note that you can include more entries in the map and they don't need to be strings. They do need to be types that are supported by B4XSerializator.

 

[Erel]

Updated example that uses a custom converter that converts datetime instances to ticks, during serializaton:

Private Sub Py_Tick (Args As Map)
    Dim time As Long = Args.Get("time")
    Log(DateTime.Time(time))
End Sub

Private Sub TimerExample
    Dim Code As String = $"
import asyncio
from datetime import datetime
bridge_instance.comm.serializator.converters[datetime] = lambda x: int(x.timestamp() * 1000)
async def TimerExample ():
    while True:
        await asyncio.sleep(1)
        bridge_instance.raise_event("tick", {"time": datetime.now()})
"$
    Wait For (Py.RunCodeAwait("TimerExample", Array(), Code)) Complete (Result As PyWrapper)
End Sub

serializator.converters is a dict of types as keys and functions as values. Note that the converters are only applied when data is sent from the Python process.

There is also an option to add custom types, I will show it in the future.

 

[rwblinn]

Thanks for the information. Very good. Have made progress with Python script using Bleak ...
Unidirectional communication = BLE server to B4J application.
Connect to the BLE server, listen to advertisements, update B4J app UI = OK
Test project attached including Python script.

Issue
When using boolean in Python script, got error. Workaround convert boolean to str. Example:

bridge_instance.raise_event("notification", {"connected": str(client.is_connected), "message": msg})

at anywheresoftware.b4a.randomaccessfile.AsyncStreams$AIN.readNumberOfBytes(AsyncStreams.java:330)
at anywheresoftware.b4a.randomaccessfile.AsyncStreams$AIN.run(AsyncStreams.java:254)

Next Step
Enhance with bidirectional communication = Inter-Process Communication (IPC).
B4J application to not only listen to advertisements, but also sends data to the Python script (use the function write to handle commands).
1. TCP socket approach: In the log there is a message "Server is listening on port: 58441", so could sent using TCP to that port
2. File-based : store commands to a file and let the Python script process (check for file regularly)
Any Thoughts?

Screenshot

Log for Info

Server is listening on port: 58785
Python path: C:\Prog\B4J\Libraries\Python\python\python.exe
Call B4XPages.GetManager.LogEvents = True to enable logging B4XPages events.
connected
starting PyBridge v0.50
watchdog set to 30 seconds
Connecting to port: 58785
[RunBLEListener] Started ...
Python version: 3.12.8 (tags/v3.12.8:2dc476b, Dec  3 2024, 19:30:04) [MSC v.1942 64 bit (AMD64)]
[Py_Notification] Received {connected=True, message=MAC address 30:C9:22:D1:80:2E}
key: connected, value: True
key: message, value: MAC address 30:C9:22:D1:80:2E
[Py_Notification] BLE Connected True
Connected to device 30:C9:22:D1:80:2E
[connect][bridge_instance.raise_event] connected=True
[Py_Notification] Received {temperature=1610, humidity=6900}
key: temperature, value: 1610
key: humidity, value: 6900