https://xtech.nikkei.com/atcl/nxt/column/18/02584/のなかの
https://xtech.nikkei.com/atcl/nxt/column/18/02584/091900003/
にある以下の題目について学習した
「超小型ラズパイ「Pico W」でBluetooth通信、PCで情報を受け取ってみよう」
ーーーーーーーーーーーーーーーーーー
ble_test_led.py :: changeをうけとるとbuiltin ledがトグル操作される
-------------- code is below (main.py とも言える) in picow -------------------
from machine import Pin
import bluetooth
from ble_simple_peripheral import BLESimplePeripheral
#ble_simple_peripheralがadverting.pyを起動する仕様
BLE_NAME = "picow"
ble = bluetooth.BLE()
sp = BLESimplePeripheral(ble, name=BLE_NAME )
led = Pin("LED", Pin.OUT)
state_led = 0
def on_rx( data ):
print( f"Received Data: {data}" )
global state_led
if( data == b'change' ):
# androi app :: serial bluetooth terminal use ble4! ただし\r\nもはいって
# 送られるので上記のchangeはchagen\r\nにする必要がある(settingめんど）
if( state_led == 0 ):
state_led = 1
led.value( state_led )
sp.send( "LED ON" )
else:
state_led = 0
led.value( state_led )
sp.send( "LED OFF" )
while True:
if( sp.is_connected() ):
sp.on_write( on_rx )
-----android appを使わん場合は--sudo ble_control_led.py -------------
from bluepy import btle
import time
picow_address = '28:cd:c1:0a:20:f4' # picowにいれたble_mac.pyで調べる！
picow_uuid = '6e400001-b5a3-f393-e0a9-e50e24dcca9e'
data_characteristic_uuid = '6e400002-b5a3-f393-e0a9-e50e24dcca9e'
#上記２個のUUIDはble_simple_peripheral.pyで定義されている
timeout = 5 # ５秒以上、通信がないとdisconnect（picowはadvertisingとなる)
class MyDelegate( btle.DefaultDelegate ):
def __init__(self):
btle.DefaultDelegate.__init__(self)
def handleNotification(self, cHandle, data):
print("Receiverd Data:", data.decode( "utf-8" ) )
bledev = btle.Peripheral( picow_address )
bledev.setDelegate( MyDelegate() )
service = bledev.getServiceByUUID( picow_uuid )
data_characteristic = service.getCharacteristics( data_characteristic_uuid )[0]
send_data = b'change'
data_characteristic.write( send_data, withResponse=True )
# 上記２行がperipheralにデータ送る部分
st_time = time.time()
while True:
if( bledev.waitForNotifications( 1.0 ) ): #なにか受け取れば無条件にループ
continue
if( time.time() > st_time + timeout ): #一定時間受取なければループ脱出
break
bledev.disconnect() #ここまでくるとadvertinsingはじまる
ーーーーーーble_simple_peripheral.py in picow ーーーーーーーーーーーー
# This example demonstrates a UART periperhal.
# This example demonstrates the low-level bluetooth module. For most
# applications, we recommend using the higher-level aioble library which takes
# care of all IRQ handling and connection management. See
# https://github.com/micropython/micropython-lib/tree/master/micropython/bluetooth/aioble
import bluetooth
import random
import struct
import time
from ble_advertising import advertising_payload
from micropython import const
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_GATTS_WRITE = const(3)
_FLAG_READ = const(0x0002)
_FLAG_WRITE_NO_RESPONSE = const(0x0004)
_FLAG_WRITE = const(0x0008)
_FLAG_NOTIFY = const(0x0010)
_UART_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
_UART_TX = (
bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E"),
_FLAG_READ | _FLAG_NOTIFY,
)
_UART_RX = (
bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E"),
_FLAG_WRITE | _FLAG_WRITE_NO_RESPONSE,
)
_UART_SERVICE = (
_UART_UUID,
(_UART_TX, _UART_RX),
)
class BLESimplePeripheral:
def __init__(self, ble, name="mpy-uart"):
self._ble = ble
self._ble.active(True)
self._ble.irq(self._irq)
((self._handle_tx, self._handle_rx),) = self._ble.gatts_register_services((_UART_SERVICE,))
self._connections = set()
self._write_callback = None
self._payload = advertising_payload(name=name, services=[_UART_UUID])
self._advertise()
def _irq(self, event, data):
# Track connections so we can send notifications.
if event == _IRQ_CENTRAL_CONNECT:
conn_handle, _, _ = data
print("New connection", conn_handle)
self._connections.add(conn_handle)
elif event == _IRQ_CENTRAL_DISCONNECT:
conn_handle, _, _ = data
print("Disconnected", conn_handle)
self._connections.remove(conn_handle)
# Start advertising again to allow a new connection.
self._advertise()
elif event == _IRQ_GATTS_WRITE:
conn_handle, value_handle = data
value = self._ble.gatts_read(value_handle)
if value_handle == self._handle_rx and self._write_callback:
self._write_callback(value)
def send(self, data):
for conn_handle in self._connections:
self._ble.gatts_notify(conn_handle, self._handle_tx, data)
def is_connected(self):
return len(self._connections) > 0
def _advertise(self, interval_us=500000):
print("Starting advertising")
self._ble.gap_advertise(interval_us, adv_data=self._payload)
def on_write(self, callback):
self._write_callback = callback
def demo():
ble = bluetooth.BLE()
p = BLESimplePeripheral(ble)
def on_rx(v):
print("RX", v)
p.on_write(on_rx)
i = 0
while True:
if p.is_connected():
# Short burst of queued notifications.
for _ in range(3):
data = str(i) + "_"
print("TX", data)
p.send(data)
i += 1
time.sleep_ms(100)
if __name__ == "__main__":
demo()
ーーーーble_advertising.py in picow ーーーーー
# Helpers for generating BLE advertising payloads.
# A more fully-featured (and easier to use) version of this is implemented in
# aioble. This code is provided just as a basic example. See
# https://github.com/micropython/micropython-lib/tree/master/micropython/bluetooth/aioble
from micropython import const
import struct
import bluetooth
# Advertising payloads are repeated packets of the following form:
# 1 byte data length (N + 1)
# 1 byte type (see constants below)
# N bytes type-specific data
_ADV_TYPE_FLAGS = const(0x01)
_ADV_TYPE_NAME = const(0x09)
_ADV_TYPE_UUID16_COMPLETE = const(0x3)
_ADV_TYPE_UUID32_COMPLETE = const(0x5)
_ADV_TYPE_UUID128_COMPLETE = const(0x7)
_ADV_TYPE_UUID16_MORE = const(0x2)
_ADV_TYPE_UUID32_MORE = const(0x4)
_ADV_TYPE_UUID128_MORE = const(0x6)
_ADV_TYPE_APPEARANCE = const(0x19)
_ADV_MAX_PAYLOAD = const(31)
# Generate a payload to be passed to gap_advertise(adv_data=...).
def advertising_payload(limited_disc=False, br_edr=False, name=None, services=None, appearance=0):
payload = bytearray()
def _append(adv_type, value):
nonlocal payload
payload += struct.pack("BB", len(value) + 1, adv_type) + value
_append(
_ADV_TYPE_FLAGS,
struct.pack("B", (0x01 if limited_disc else 0x02) + (0x18 if br_edr else 0x04)),
)
if name:
_append(_ADV_TYPE_NAME, name)
if services:
for uuid in services:
b = bytes(uuid)
if len(b) == 2:
_append(_ADV_TYPE_UUID16_COMPLETE, b)
elif len(b) == 4:
_append(_ADV_TYPE_UUID32_COMPLETE, b)
elif len(b) == 16:
_append(_ADV_TYPE_UUID128_COMPLETE, b)
# See org.bluetooth.characteristic.gap.appearance.xml
if appearance:
_append(_ADV_TYPE_APPEARANCE, struct.pack("<h", appearance))
if len(payload) > _ADV_MAX_PAYLOAD:
raise ValueError("advertising payload too large")
return payload
def decode_field(payload, adv_type):
i = 0
result = []
while i + 1 < len(payload):
if payload[i + 1] == adv_type:
result.append(payload[i + 2 : i + payload[i] + 1])
i += 1 + payload[i]
return result
def decode_name(payload):
n = decode_field(payload, _ADV_TYPE_NAME)
return str(n[0], "utf-8") if n else ""
def decode_services(payload):
services = []
for u in decode_field(payload, _ADV_TYPE_UUID16_COMPLETE):
services.append(bluetooth.UUID(struct.unpack("<h", u)[0]))
for u in decode_field(payload, _ADV_TYPE_UUID32_COMPLETE):
services.append(bluetooth.UUID(struct.unpack("<d", u)[0]))
for u in decode_field(payload, _ADV_TYPE_UUID128_COMPLETE):
services.append(bluetooth.UUID(u))
return services
def demo():
payload = advertising_payload(
name="micropython",
services=[bluetooth.UUID(0x181A), bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")],
)
print(payload)
print(decode_name(payload))
print(decode_services(payload))
if __name__ == "__main__":
demo()
0 件のコメント:
コメントを投稿