构建dongle版的microbit插件 about codelab_adapter_extensions HOT 18 OPEN

dongle.bled_write('6e400003-b5a3-f393-e0a9-e50e24dcca9e', [0x0a]) # 往tx写换行符

'''
EIM: Everything Is Message
'''
import time
import threading

from codelab_adapter import settings
from codelab_adapter.core_extension import Extension
from codelab_adapter.utils import threaded
from codelab_adapter.dongle import Dongle
import time
import pygatt
import tenacity
import binascii
import base64


class BleMicrobit(Extension):
    def __init__(self):
        name = type(self).__name__  # class name
        super().__init__(name)
        self.TOPIC = "eim"

    def read_cb(self, handle, value):
        message = base64.b64encode(value)

        self.publish({
            "payload": {
                "encoding": "base64",
                "message": str(message)[2:-1]  #处理字符
            },
            "method": "characteristicDidChange",
            "topic": self.TOPIC,
        })

    @threaded
    def message_monitor(self):
        while self._running:
            read_message = self.read()
            topic = read_message.get("topic")
            if topic == self.TOPIC:

                self.logger.info("消息:{}".format(read_message))

                method = read_message.get("method")
                data = read_message.get("payload")

                if method == 'discover':
                    try:
                        self.services = data.get('filters')[
                            0].get('services')[0]
                    except Exception as e:
                        self.publish({
                            "messageID": read_message.get("messageID"),
                            "error": str(e),
                            "topic": self.TOPIC,
                        })
                    else:
                        self.publish({
                            "messageID": read_message.get("messageID"),
                            "result": "",
                            "topic": self.TOPIC,
                        })

                    try:
                        self.logger.info("开始扫描BLE")
                        devices = self.dongle.bled_scan()
                    except Exception as e:
                        self.logger.info("BLE扫描错误:{}".format(e))
                    else:
                        for dev in devices:
                            try:
                                complete_list_16_bit_service_class_uuids = dev.get('packet_data').get(
                                    'connectable_advertisement_packet').get('complete_list_16-bit_service_class_uuids')
                            except Exception as e:
                                pass
                            else:
                                if isinstance(complete_list_16_bit_service_class_uuids, bytearray):
                                    if int.from_bytes(complete_list_16_bit_service_class_uuids, byteorder='little') == self.services:
                                        self.publish({
                                            "payload": {
                                                "name": dev.get('name'),
                                                "rssi": dev.get('rssi'),
                                                "peripheralId": dev.get('address')
                                            },
                                            "method": "didDiscoverPeripheral",
                                            "topic": self.TOPIC,
                                        })

                elif method == 'connect':
                    self.peripheralId = data.get("peripheralId")

                    try:
                        self.logger.info("开始连接")
                        self.dongle.bled_connect(self.peripheralId)
                    except Exception as e:
                        self.logger.info("连接错误:{}".format(str(e)))

                        self.publish({
                            "messageID": read_message.get("messageID"),
                            "error": str(e),
                            "topic": self.TOPIC
                        })

                        self.ble_connected = False
                    else:
                        self.publish({
                            "messageID": read_message.get("messageID"),
                            "result": "",
                            "topic": self.TOPIC
                        })

                elif method == 'read':
                    characteristicId = data.get("characteristicId")

                    self.dongle.bled_subscribe(
                        characteristicId, callback=self.read_cb, indication=True)  # 订阅传感器数据

                    self.publish(
                        {
                            "messageID": read_message.get("messageID"),
                            "result": {
                                "encoding": "base64",
                                "message": ""
                            },
                            "topic": self.TOPIC
                        }
                    )

                elif method == 'write':
                    characteristicId = data.get("characteristicId")
                    text = data.get("message").encode('ascii')
                    self.logger.info("收到的消息为:{}".format(text))
                    try:
                        self.dongle.bled_write(
                            characteristicId, text)
                    except Exception as e:
                        self.logger.info("写消息错误:{}".format(e))
                    else:
                        self.publish(
                            {
                                "messageID": read_message.get("messageID"),
                                "topic": self.TOPIC,
                                "result": 6
                            }
                        )

    def run(self):

        self.dongle = Dongle()

        try:
            self.logger.info("启动BLE")

            self.dongle.bled_start()
            self.message_monitor()

            while True:
                time.sleep(1)

        except Exception as e:
            self.logger.info("BLE启动出错:{}".format(e))

        finally:
            self.dongle.bled_stop()


export = BleMicrobit

RetryError[<Future at 0x2ee1130 state=finished raised BGAPIError>]

dongle.bled_connect('E1:4C:E0:01:43:18') #这是我Microbit的Mac地址
dongle.bled_subscribe('6e400002-b5a3-f393-e0a9-e50e24dcca9e', callback=read_cb)   #订阅rx

dongle.bled_write('6e400003-b5a3-f393-e0a9-e50e24dcca9e',0x0a) # 往tx写换行符

import time
import pygatt
import tenacity
import binascii


# pygatt api doc : https://github.com/peplin/pygatt/blob/master/pygatt/device.py
class Dongle(object):
    def __init__(self):
        self.adapter = pygatt.BGAPIBackend()
        self.device = None
        self.is_running = True

    @tenacity.retry(stop=tenacity.stop_after_attempt(3))
    def bled_start(self):
        self.adapter.start()

    def bled_stop(self):
        self.adapter.stop()
        self.is_running = False

    def bled_scan(self):
        devices = self.adapter.scan()
        return devices

    def bled_rssi(self):
        return self.device.get_rssi()

    @tenacity.retry(stop=tenacity.stop_after_attempt(3))
    def bled_connect(self, mac_addr):
        self.device = self.adapter.connect(
            mac_addr, address_type=pygatt.BLEAddressType.random)
        print('connect to {}'.format(mac_addr))
        print("rssi   : " + str(self.device.get_rssi()))

        # print("chars  : " + str(self.device.discover_characteristics()))
        # for i in self.device.discover_characteristics():
        #     print(i)

        time.sleep(1)

    def bled_subscribe(self, uuid, callback=None, indication=False):
        try:
            self.device.subscribe(uuid, callback=callback, indication=indication)
        except Exception as e:
            print(e)

    def bled_write(self, uuid, data,length = 1):
        if isinstance(data,int):
            data = data.to_bytes(length, byteorder='little')
        print("写入：", bytearray(data))
        self.device.char_write(uuid, bytearray(data))

    def bled_read(self, uuid):
        data = self.device.char_read(uuid)
        return data

    @staticmethod
    def _ascii_to_hex(rawhex):
        return binascii.unhexlify(rawhex)


if __name__ == '__main__':

    def read_cb(handle, value):

        a = value
        b = [bytes([a[i]])+ bytes([a[i+1]]) for i in range(0,len(a),2)]
        c = [int.from_bytes(i, byteorder='little', signed=True) for i in b]
        print(c)

        # bytes_str = binascii.hexlify(value)
        # print(bytes_str)

    dongle = Dongle()
    try:
        dongle.bled_start()

        devices = dongle.bled_scan()
        for dev in devices:
            print(dev.get('address'), dev.get('rssi'), dev.get('name'))

        target_mac_address = 'DA:19:8A:61:AD:95'

        microbit_uuid = {
        'button_A' : 'e95dda90-251d-470a-a062-fa1922dfa9a8', #按钮A数据
        'button_B' : 'e95dda91-251d-470a-a062-fa1922dfa9a8', #按钮B数据
        'accelerometer' : 'e95dca4b-251d-470a-a062-fa1922dfa9a8', #加速度数据
        'Temperature' : 'e95d9250-251d-470a-a062-fa1922dfa9a8', #温度计数据
        'magnetometer' : 'e95dfb11-251d-470a-a062-fa1922dfa9a8', #磁力计数据
        'led_char' : 'e95d93ee-251d-470a-a062-fa1922dfa9a8', #led文字
        'led_matrix' : 'e95d7b77-251d-470a-a062-fa1922dfa9a8', #led矩阵
        'pin_date' : 'e95d8d00-251d-470a-a062-fa1922dfa9a8',  # pin读写服务
        'pinMode_config' : 'e95d5899-251d-470a-a062-fa1922dfa9a8',  # pin引脚配置
        'ADC_config' : 'e95db9fe-251d-470a-a062-fa1922dfa9a8',  # pinAD配置
        'PWM_config' : 'e95dd822-251d-470a-a062-fa1922dfa9a8',  # PWM设置
        'tx': '6e400003-b5a3-f393-e0a9-e50e24dcca9e',  # tx设置
        'rx': '6e400002-b5a3-f393-e0a9-e50e24dcca9e',  # rx设置
        }

        dongle.bled_connect(target_mac_address)  # 这是我Microbit的Mac地址

        dongle.bled_write(microbit_uuid['led_matrix'], [17, 27, 27, 27, 27]) #矩阵图标的显示方式
        
        dongle.bled_subscribe(microbit_uuid['accelerometer'], callback=read_cb,indication=False)  # 订阅传感器数据

        while True:
            time.sleep(1)
    finally:
        dongle.bled_stop()