Adolfo Reyna
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""CircuitPython BLE UART utility.
|
|
|
|
Wraps the adafruit_ble library to advertise Nordic UART Service and scan for nearby devices.
|
|
"""
|
|
|
|
import time
|
|
from adafruit_ble import BLERadio
|
|
from adafruit_ble.services.nordic import UARTService
|
|
from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
|
|
|
|
class BLEUART:
|
|
def __init__(self, ble=None, name="ESP32-S3-RLCD"):
|
|
self.ble = BLERadio()
|
|
self.ble.name = name
|
|
self.uart = UARTService()
|
|
self.advertisement = ProvideServicesAdvertisement(self.uart)
|
|
self.rx_callback = None
|
|
self._is_advertising = False
|
|
self._start_advertise()
|
|
|
|
def _start_advertise(self):
|
|
if not self.ble.connected and not self._is_advertising:
|
|
try:
|
|
self.ble.start_advertising(self.advertisement)
|
|
self._is_advertising = True
|
|
print(f"BLE advertising started as '{self.ble.name}'")
|
|
except Exception as e:
|
|
print(f"Failed to start BLE advertising: {e}")
|
|
|
|
def on_rx(self, callback):
|
|
self.rx_callback = callback
|
|
return callback
|
|
|
|
def update(self):
|
|
"""Polls the UART stream for incoming data and triggers callbacks."""
|
|
if self.ble.connected:
|
|
self._is_advertising = False
|
|
try:
|
|
if self.uart.in_waiting:
|
|
data = self.uart.read(self.uart.in_waiting)
|
|
if self.rx_callback and data:
|
|
try:
|
|
decoded = data.decode("utf-8").strip()
|
|
self.rx_callback(decoded)
|
|
except:
|
|
self.rx_callback(data)
|
|
except Exception as e:
|
|
print(f"Error reading BLE RX: {e}")
|
|
else:
|
|
if not self._is_advertising:
|
|
self._start_advertise()
|
|
|
|
def write(self, data):
|
|
if not self.ble.connected:
|
|
return False
|
|
if isinstance(data, str):
|
|
data = data.encode("utf-8")
|
|
try:
|
|
self.uart.write(data)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error sending BLE data: {e}")
|
|
return False
|
|
|
|
def is_connected(self):
|
|
return self.ble.connected
|
|
|
|
def scan(self, duration_ms=3000):
|
|
"""Scans for nearby BLE devices."""
|
|
duration_s = duration_ms / 1000.0
|
|
results = {}
|
|
was_advertising = self._is_advertising
|
|
if was_advertising:
|
|
try:
|
|
self.ble.stop_advertising()
|
|
except:
|
|
pass
|
|
self._is_advertising = False
|
|
|
|
try:
|
|
print("Starting BLE scan...")
|
|
for adv in self.ble.start_scan(timeout=duration_s):
|
|
# Clean up address representation (mac address)
|
|
mac = str(adv.address).replace("Address(", "").replace(")", "").strip()
|
|
name = adv.complete_name or ""
|
|
results[mac] = {"rssi": adv.rssi, "name": name}
|
|
self.ble.stop_scan()
|
|
except Exception as e:
|
|
print(f"BLE scan error: {e}")
|
|
|
|
if was_advertising:
|
|
self._start_advertise()
|
|
return results
|
|
|
|
def close(self):
|
|
try:
|
|
self.ble.stop_advertising()
|
|
except:
|
|
pass
|
|
self._is_advertising = False
|
|
print("BLE closed.")
|