S4 + esp32 (Micropython) - не получается выполнить pair

[aalexey07]

Версия модуля: 3.3.5
Модель бризера: S4
Версия прошивки (если известна): 02D2

Описание проблемы

Пишу модуль на микропитоне для esp32, планирую использовать ваши наработки по протоколу для формирования сообщений для бризера S4. Начал с сопряжения и не получается его выполнить. Помогите пожалуйста разобраться с логикой обмена сообщениями с бризером.

Порядок действий, которые приводят к проявлению проблемы

1. Использую модуль bluetooth (пробовал и с ubluetooth) микропитона https://docs.micropython.org/en/latest/library/bluetooth.html.
2. Перевожу бризер в режим сопряжения (светодиод на бризере мигает синим цветом).
3. На esp32 (скетч в конце тикета) отправляю команду ble.connect(1, ubinascii.unhexlify(b'f022a36c3a55')).
4. Получаю событие об успешном подключении к бризеру.

# _IRQ_PERIPHERAL_CONNECT:
# A successful gap_connect().

Connection_handle:0 addr_type:1 addr:b'f022a36c3a55'

5. Отправляю команду запроса статуса ble.send(ubinascii.unhexlify(b'8010003aa13232b788d5da1ec11843bbaa')).
6. Ответа на нее не вижу. Пока представляю себе этот процесс так, что бризер должен уведомить esp32, что положил в ее UUID сообщение и возникнет событие, которое успешно обрабатывается, когда проверял обмен сообщениями с телефоном:

# _IRQ_GATTS_WRITE:
# A client has written to this characteristic or descriptor.
buffer = self.ble.gatts_read(self.rx)
ble_msg = buffer.decode('UTF-8').strip()

7. Далее пробовал и делать дисконект, и не делать его ble.disconnect(). Пробовал отправлять как одну команду запроса статуса за один сеанс привязки, так и несколько команд за один сеанс. Для отладки пока команды формирую в отдельном файле кодом из вашего модуля с 8 рандомными байтами.
8. Бризер выходит из режима сопряжения.
9. Получаю событие о дисконнекте (либо получаю его, когда сам сделал дисконнект)

# _IRQ_PERIPHERAL_DISCONNECT:
# Connected peripheral has disconnected.

10. На esp32 отправляю команду ble.connect(1, ubinascii.unhexlify(b'f022a36c3a55')) и сразу же получаю событие о дисконнекте.
11. Делаю вывод, что сопряжение не удалось.

Вопросы:

1. Нужно ли во время сопряжения перед командой запроса состояния бризера отправлять отдельную команду сопряжения ble.pair()? Посмотрел ваш модуль и не увидел, что она отправляется.

Я ее пробовал отправлять, но модуль на микропитоне ее похоже не поддерживает, т.к. получаю в терминале:

Pairing...
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "main.py", line 168, in con2br
  File "main.py", line 150, in pair
AttributeError: 'BLE' object has no attribute 'gap_pair'

И в описании модуля написано, что на esp32 она не поддерживается, или же pair поддерживается, но не поддерживается bonding...

2. Корректную ли команду запроса состояния бризера я отправляю?
3. Как вы предполагаете, почему не получается сделать сопряжение?

Логи

Вставьте сюда debug-log, в момент проявления проблемы

Скетч для esp32:

from machine import Pin
from machine import Timer
from machine import soft_reset
from time import sleep_ms, sleep
# import ubluetooth
import bluetooth
import ubinascii

ble_msg = ""


class ESP32_BLE():
    def __init__(self, name):
        # Create internal objects for the onboard LED
        # blinking when no BLE device is connected
        # stable ON when connected
        self.led = Pin(2, Pin.OUT)
        self.timer1 = Timer(0)

        self.name = name
        # self.ble = ubluetooth.BLE()
        self.ble = bluetooth.BLE()
        self.ble.active(True)
        self.disconnected()
        self.ble.irq(self.ble_irq)
        self.register()
        self.advertiser()

    def connected(self):
        self.led.value(1)
        self.timer1.deinit()

    def disconnected(self):
        self.timer1.init(period=100, mode=Timer.PERIODIC, callback=lambda t: self.led.value(not self.led.value()))

    def ble_irq(self, event, data):
        global ble_msg

        if event == 1:  # _IRQ_CENTRAL_CONNECT:
            # A central has connected to this peripheral
            self.connected()

        elif event == 2:  # _IRQ_CENTRAL_DISCONNECT:
            # A central has disconnected from this peripheral.
            self.advertiser()
            self.disconnected()

        elif event == 3:  # _IRQ_GATTS_WRITE:
            # A client has written to this characteristic or descriptor.
            buffer = self.ble.gatts_read(self.rx)
            ble_msg = buffer.decode('UTF-8').strip()

        elif event == 5:  # _IRQ_SCAN_RESULT:
            # A single scan result.
            addr_type, addr, adv_type, rssi, adv_data = data
            print(
                'Addr_type:{} addr:{} adv_type:{} rssi:{} adv_data:{}'.format(addr_type, ubinascii.hexlify(addr),
                                                                              adv_type, rssi,
                                                                              ubinascii.hexlify(adv_data)))

        elif event == 6:    # _IRQ_SCAN_DONE:
            # Scan duration finished or manually stopped.
            print('Scan stopped (duration finished or manually)')

        elif event == 7:  # _IRQ_PERIPHERAL_CONNECT:
            # A successful gap_connect().
            print('Connected')
            conn_handle, addr_type, addr = data
            print('Connection_handle:{} addr_type:{} addr:{}'.format(conn_handle, addr_type, ubinascii.hexlify(addr)))

        elif event == 8:        # _IRQ_PERIPHERAL_DISCONNECT:
            # Connected peripheral has disconnected.
            conn_handle, addr_type, addr = data
            print('Disconnected')

        elif event == 28:       # _IRQ_ENCRYPTION_UPDATE:
            # The encryption state has changed (likely as a result of pairing or bonding).
            conn_handle, encrypted, authenticated, bonded, key_size = data
            print('After pairing: ', conn_handle, encrypted, authenticated, bonded, key_size)

    def register(self):
        # Nordic UART Service (NUS)
        NUS_UUID = '6E400001-B5A3-F393-E0A9-E50E24DCCA9E'
        RX_UUID = '6E400002-B5A3-F393-E0A9-E50E24DCCA9E'
        TX_UUID = '6E400003-B5A3-F393-E0A9-E50E24DCCA9E'

        # BLE_NUS = ubluetooth.UUID(NUS_UUID)
        # BLE_RX = (ubluetooth.UUID(RX_UUID), ubluetooth.FLAG_WRITE)
        # BLE_TX = (ubluetooth.UUID(TX_UUID), ubluetooth.FLAG_NOTIFY)

        BLE_NUS = bluetooth.UUID(NUS_UUID)
        BLE_RX = (bluetooth.UUID(RX_UUID), bluetooth.FLAG_WRITE)
        BLE_TX = (bluetooth.UUID(TX_UUID), bluetooth.FLAG_NOTIFY)

        BLE_UART = (BLE_NUS, (BLE_TX, BLE_RX,))
        SERVICES = (BLE_UART,)
        ((self.tx, self.rx,),) = self.ble.gatts_register_services(SERVICES)

    def send(self, data):
        self.ble.gatts_notify(0, self.tx, data + '\n')

    def advertiser(self):
        name = bytes(self.name, 'UTF-8')
        adv_data = bytearray('\x02\x01\x02') + bytearray((len(name) + 1, 0x09)) + name
        self.ble.gap_advertise(100, adv_data)
        print(adv_data)
        print("\r\n")
        # adv_data
        # raw: 0x02010209094553503332424C45
        # b'\x02\x01\x02\t\tESP32BLE'
        #
        # 0x02 - General discoverable mode
        # 0x01 - AD Type = 0x01
        # 0x02 - value = 0x02

        # https://jimmywongiot.com/2019/08/13/advertising-payload-format-on-ble/
        # https://docs.silabs.com/bluetooth/latest/general/adv-and-scanning/bluetooth-adv-data-basics

    def scan(self, duration):
        print('Started scanning')
        self.ble.gap_scan(duration)

    def connect(self, addr_type, mac, scan_duration=2000):
        # addr_type values indicate public or random addresses:
        # 0x00 - PUBLIC
        # 0x01 - RANDOM(either static, RPA, or NRPA, the type is encoded in the address itself)
        print('Connecting')
        x = 0
        while x < 1:
            try:
                self.ble.gap_connect(addr_type, mac, scan_duration)
                # ble.connect(1, ubinascii.unhexlify(b'f022a36c3a55')) breezer
                # ble.connect(1, ubinascii.unhexlify(b'a4c337077418')) iphone
                ble.send(ubinascii.unhexlify(b'8010003aa13232b788d5da1ec11843bbaa'))
                # ble.send(b'\x80\x10\x00:\xa122(?\xfc\xe7\xb9\x16\xd3\x9d\xbb\xaa')
                break
            except Exception:
                pass
            sleep(2)
            x += 1
        print('Failed to connect')

    def disconnect(self, conn_handle=0):
        self.ble.gap_disconnect(conn_handle)

    def pair(self):
        print('Pairing...')
        self.ble.gap_pair(0)


led = Pin(2, Pin.OUT)
but = Pin(0, Pin.IN)
ble = ESP32_BLE("ESP32BLE_2")


def buttons_irq(pin):
    led.value(not led.value())
    ble.send('LED state will be toggled.')
    print('LED state will be toggled.')


but.irq(trigger=Pin.IRQ_FALLING, handler=buttons_irq)

def con2br():
    ble.connect(1, ubinascii.unhexlify(b'f022a36c3a55'))
    # ble.pair()
    ble.send(ubinascii.unhexlify(b'8010003aa132325b1033dcac14955bbbaa'))
    ble.send(ubinascii.unhexlify(b'8010003aa13232810bd0980babd51abbaa'))
    # ble.send(ubinascii.unhexlify(b'8010003aa132325b1033dcac14955bbbaa'))
    # ble.disconnect()

# con2br()

while True:
    if ble_msg.lower() == 'read_led':
        print(ble_msg)
        ble_msg = ""
        print('LED is ON.' if led.value() else 'LED is OFF')
        sleep(10)
        ble.send('LED is ON.' if led.value() else 'LED is OFF')
    elif ble_msg.lower() == 'scan':
        print(ble_msg)
        ble_msg = ""
        ble.scan(30000)     # 10K ms == 10 s
    elif ble_msg.lower() == 'connect':
        print(ble_msg)
        ble_msg = ""
        # print(ubinascii.unhexlify('f022a36c3a55'))
        ble.connect(1, ubinascii.unhexlify(b'f022a36c3a55'))
    elif ble_msg.lower() == 'dis':
        print(ble_msg)
        ble_msg = ""
        #if ble.disconnect():
           # print('Disconnected')
        #else:
           # print('Failed to disconnect')
    elif ble_msg.lower() == 'req':
        print(ble_msg)
        ble_msg = ""
        ble.send(b'\x80\x10\x00:\xa122(?\xfc\xe7\xb9\x16\xd3\x9d\xbb\xaa')
    elif ble_msg != '':
        print(ble_msg)
    sleep_ms(100)

[IATkachenko]

Извините, пропустил это issue.
Посмотрите вот этот проект https://github.com/dentra/esphome-tion там уже все работает.

По поводу логики: вам нужно, на своей стороне, еще включить получение уведомлений.
В коде библиотеки это делает вот это место

tion_python/tion_btle/tion.py

Line 335 in 7c07708

async def _enable_notifications(self):

тогда при получении пакета (будет несколько пакетов, котоыре нужно собрать в сообщение перед дешифровкой) вы будете получать event (по крайней мере так это работает на python).

Нужно ли во время сопряжения перед командой запроса состояния бризера отправлять отдельную команду сопряжения ble.pair()?

Да BT pairing нужно сделать, чтобы не загонять бризер в режим сопряжения каждый раз. Как -- не знаю. В ESP не силен.
Но для этого запрашивать параметры, вроде как, не обязательно.

Посмотрел ваш модуль и не увидел, что она отправляется.

Отправляется

tion_python/tion_btle/tion.py

Line 480 in 7c07708

async def pair(self):

    @final
    async def pair(self):
        _LOGGER.debug("Pairing")
        await self._connect(need_notifications=False)
        _LOGGER.debug("Connected. BT pairing ...")
        try:
            await self._btle.pair()
...

Корректную ли команду запроса состояния бризера я отправляю?

В коде модуля команда выглядит так

tion_python/tion_btle/s4.py

Line 98 in 7c07708

def command_getStatus(self) -> bytearray:

    @property
    def command_getStatus(self) -> bytearray:
        return bytearray([TionLiteFamily.SINGLE_PACKET_ID, 0x10, 0x00, self.MAGIC_NUMBER, 0xa1] +
                         self.REQUEST_PARAMS +
                         self.random4 + self.random4 +
                         self.CRC
                         )

Вы отправляете 80 10 00 3a a1 32 32 b7 88 d5 da 1e c1 18 43 bb aa -- похоже на правду.

[aalexey07]

Игорь, спасибо за ответы. За время отпуска пришла идея и я смог немного продвинуться вперед. Pair получилось сделать AT командами. К бризеру теперь могу подключаться в любой момент, а не только во время включенного на бризере режима сопряжения. Могу получать список характеристик, читать некоторые.

Но когда пытаюсь записать данные в эту характеристику "98f00002-3788-83ea-453e-f52244709ddb" или прочитать из этой "98f00003-3788-83ea-453e-f52244709ddb" - получаю статус "insufficient authorization". Почитал issue по вашему проекту - возможно у меня не сохранились ключи, но пока не нашел, где их посмотреть на esp32.

Попробую разобраться с проектом, ссылку на который Вы прислали.

Я изучаю Python, поэтому интересно все-таки эту задачу решить на нем, заодно и потренироваться :)