[Python] BLE 통신 하기 with bleak (MacOS) - 2) 데이터 받기

2023.05.24 - [🐍 | Python/파이썬 초간단 프로젝트] - [Python] BLE 통신 하기 with bleak (MacOS) - 1) 탐색, 연결

 

지난 포스트에서는 bleak 모듈을 이용해 주변 기기를 탐색하고 연결하는 것을 해보았다.

 

이번에는 현재 Kit의 서비스 정보를 알아보고 그를 활용하여 kit로부터 데이터를 받아보자!

 

1. 연결한 장치의 서비스

1) 연결된 서비스 확인

import asyncio
from bleak import BleakClient, BleakScanner

async  def  main ():
    devices = await BleakScanner.discover()
    for idx, device in enumerate(devices):
        print(f'{idx}\t\t',device)
    select = int(input())
    address = devices[select].address

    async  with BleakClient(address) as client:
        try:
            print("Connecting...", end='')
            await client.connect()
            print('Connected!!')
            
            # 서비스 받기
            services = await client.get_services()
            print("Services:", type(services))
            
            for service in services:
	            print(service)


            
        except Exception as e:
            print('error: ', e, end='')
        finally:
            print('Start disconnect')
            await client.disconnect()
            print('disconnected')

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
print('Done!')

 

수행 결과 중 일부

 

선택한 장치와 정상 연결 후 get_services()를 통해 서비스 정보를 반환받는다.

 

 

2) chracteristic 정보 얻기

import asyncio
from bleak import BleakClient, BleakScanner

async  def  main ():
    devices = await BleakScanner.discover()
    for idx, device in enumerate(devices):
        print(f'{idx}\t\t',device)
    select = int(input())
    address = devices[select].address

    async  with BleakClient(address) as client:
        try:
            print("Connecting...", end='')
            await client.connect()
            print('Connected!!')
            
            # 서비스 받기
            services = await client.get_services()
            print("Services:", type(services))
            
            for service in services:
                print(service)

                print(service.uuid)
                print("Characteristic List:")

                for characteristic in service.characteristics:
                    print("\n\t", characteristic)
                    print("\t", characteristic.uuid)
                    print("\t", characteristic.description)
                    print("\t", characteristic.properties)

            
        except Exception as e:
            print('error: ', e, end='')
        finally:
            print('Start disconnect')
            await client.disconnect()
            print('disconnected')

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
print('Done!')

 

수행 결과 중 일부

 

serive 내부에 들어있는 정보들에 대해서 반환해보면 위와 같다.

uuid, 설명, 속성 들을 알 수 있다.

 

이 중 chracteristic UUID와 속성이 다음 단계를 위해 필요한 정보들이다.

 

위의 정보를 통해 알 수 있는 것은

6e400003-b5a3-f393-e0a9-e50e24dcca9e 이라는 UUID를 가지고 있는 캐릭터리스틱이

notify, write 기능을 수행할 수 있다 라는 것이다.

 

(( 주요 속성 ))
- read : 읽기 (장치에 있는 데이터를)
- write : 쓰기 (장치에게 데이터를) & 응답 기다리기
- notify : 클라이언트 요청이 없어도 장치에서 데이터를 보내기 & 응답 기다리지 않음
... (indicate, write-without-response 등)

 

 

3) 데이터 보내기 (notify)

필자의 프로젝트에서 필요한 속성인 notify를 통해

장치에서 클라이언트에게 데이터를 보내도록 해보자!

import asyncio
from bleak import BleakClient, BleakScanner


def notify_callback(sender: int, data: bytearray):
    print('sender: ', sender, 'data: ', data)
    print('\tint list: ', [b for b in data])

async  def  main ():
    devices = await BleakScanner.discover()
    for idx, device in enumerate(devices):
        print(f'{idx}\t\t',device)
    select = int(input())
    address = devices[select].address

    async  with BleakClient(address) as client:
        try:
            print("Connecting...", end='')
            await client.connect()
            print('Connected!!')
            
            # 서비스 받기
            services = await client.get_services()
            print("Services:", type(services))
            
            for service in services:
                print(service)
                for characteristic in service.characteristics:
                    if 'notify' in characteristic.properties:
                      notity_charcteristic_uuid = characteristic.uuid
                      print("\n\t", characteristic)
                      print('Activate notify...')
                      await client.start_notify(characteristic, notify_callback)

            
        except Exception as e:
            print('error: ', e, end='')
        
        if client.is_connected:
            await asyncio.sleep(2.0) #20초 동안 스트림
            print('try to deactivate notify.')
            await client.stop_notify(notity_charcteristic_uuid)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
print('Done!')

 

수행 결과 중 일부

 

위와 같이 장치로부터 데이터를 받아올 수 있다.

 

 

 

 

추후에 다른 속성들을 다르는 글도 포스팅해볼 예정이다!