11.9. 주변장치로 동작하기¶

가장 일반적인 카메라 측 BLE 패턴은 peripheral 로 동작하는 것입니다. 즉, 작은 GATT 데이터베이스를 게시하고, 자신의 존재를 광고하며, 휴대폰이나 동반 장치로부터의 연결을 수락하고, 반대편에 있는 누구에게든 값을 스트리밍하는 것입니다.

11.9.1. GATT 데이터베이스 구축하기¶

주변장치가 시작 시 가장 먼저 하는 일은 – 무선 송수신기를 켜기도 전에 – 노출하려는 데이터베이스를 구축하고, 각 서비스와 특성에 대한 객체를 생성한 다음, 이를 전부 등록하는 것입니다:

import aioble
import bluetooth

ENV_SERVICE = bluetooth.UUID(0x181A)              # Environmental Sensing
TEMP_UUID = bluetooth.UUID(0x2A6E)                # Temperature
HUMID_UUID = bluetooth.UUID(0x2A6F)               # Humidity

env = aioble.Service(ENV_SERVICE)
temp_char = aioble.Characteristic(
    env, TEMP_UUID,
    read=True, notify=True, initial=b"\\x00\\x00",
)
humid_char = aioble.Characteristic(
    env, HUMID_UUID,
    read=True, notify=True, initial=b"\\x00\\x00",
)

aioble.register_services(env)

각 aioble.Characteristic 는 서비스를 첫 번째 인수로 전달하여 생성하는 것만으로 해당 서비스에 연결됩니다. 불리언 키워드 인수(read, write, write_no_response, notify, indicate)는 클라이언트가 수행할 수 있는 GATT 작업을 선택합니다. False (기본값)를 전달하면 해당 속성 비트가 설정되지 않습니다.

aioble.register_services() 는 조립된 트리를 GATT 서버에 커밋합니다. 이 함수는 어떤 aioble.advertise() 가 시작되기 전에 한 번 호출되어야 합니다. 다시 호출하면 이전 데이터베이스가 교체됩니다.

11.9.2. 광고하기¶

데이터베이스가 준비되면, 광고는 연결을 기다리는 하나의 코루틴 호출입니다:

async def serve_one():
    connection = await aioble.advertise(
        interval_us=250000,
        name="openmv-env",
        services=[ENV_SERVICE],
        appearance=0x0540,           # Generic Sensor
    )

키워드 인수는 광고 페이로드 필드에 직접 매핑됩니다. name 은 로컬 이름 필드입니다. services 는 장치가 호스팅하는 서비스 UUID 목록입니다(휴대폰 측 스캐너가 이를 기준으로 필터링할 수 있습니다). appearance 는 표준 16비트 appearance 값 중 하나로, central이 적절한 아이콘을 표시할 수 있게 해주는 힌트입니다. 제조사별 데이터는 manufacturer=(company_id, data_bytes) 를 통해 전달됩니다.

몇 가지 덜 일반적인 키워드가 나머지 광고 플래그 공간을 다룹니다:

connectable=False – 브로드캐스트 전용 모드(연결을 절대 수락하지 않음)입니다. 비콘 스타일 페이로드에 적합한 선택입니다.
limited_disc=True – general discoverable 대신 limited discoverable 플래그를 사용합니다. 일부 운영체제는 페어링 UI에서 이 둘을 다르게 취급합니다.
adv_data / resp_data – 애플리케이션이 레이아웃을 완전히 제어해야 하는 경우의 원시 바이트입니다.
timeout_ms – 정해진 시간이 지난 후 포기합니다. 기본값은 무한히 광고하는 것입니다.

central이 연결되면 aioble.advertise() 는 결과로 생성된 aioble.DeviceConnection 을 반환합니다. 이 시점에 주변장치는 광고를 중단합니다.

11.9.3. 한 클라이언트에 서비스 제공하기¶

주변장치의 메인 루프는 일반적으로 다음과 같습니다:

async def serve():
    while True:
        connection = await aioble.advertise(
            interval_us=250000,
            name="openmv-env",
            services=[ENV_SERVICE],
        )
        print("connected:", connection.device.addr_hex())
        async with connection:
            await connection.disconnected()
        print("disconnected; advertising again")

asyncio.run(serve())

async with connection 은 연결 해제 정리를 자동으로 처리합니다. disconnected() 는 어느 한쪽이 연결을 종료할 때까지 일시 중단되는 코루틴입니다. 이는 central이 사라질 때까지 주변장치가 계속 서비스를 제공하다가, 다음 라운드를 위해 다시 광고로 돌아가는 깔끔한 방법입니다.

11.9.4. 특성 업데이트하기¶

주변장치는 aioble.Characteristic.write() 로 로컬 GATT 데이터베이스를 업데이트합니다:

temp_char.write(b"\\x9a\\x09")              # 24.58 deg C as sint16, 0.01 units

이는 다음에 어떤 클라이언트가 read 할 때 반환되는 값을 변경합니다. 하지만 그 자체로는 새 값을 푸시하지 않습니다. 구독 중인 클라이언트는 클라이언트가 폴링하거나 주변장치가 명시적 알림을 보내기 전까지는 아무것도 보지 못합니다.

푸시 측은 동일한 호출의 단일 키워드입니다:

temp_char.write(temp_bytes, send_update=True)

send_update=True 는 이 특성을 구독한 모든 클라이언트에게 알림(또는 표시)을 보냅니다. 대부분의 센서 스타일 코드는 연결별 태스크 안에서 센서를 읽고 약 1초마다 send_update=True 로 값을 쓰는 루프로 동작합니다:

async def stream_temperature(connection):
    while connection.is_connected():
        temp_char.write(encode_temperature(read_sensor()), send_update=True)
        await asyncio.sleep(1)

async def serve():
    while True:
        connection = await aioble.advertise(
            interval_us=250000,
            name="openmv-env",
            services=[ENV_SERVICE],
        )
        async with connection:
            asyncio.create_task(stream_temperature(connection))
            await connection.disconnected()

전체 구독 집합이 아니라 특정 클라이언트 하나에게 알림을 보내고 싶다면(예: 해당 클라이언트의 명령에 대한 연결 전용 응답), aioble.Characteristic.notify() 와 indicate() 가 DeviceConnection 인수와 선택적 페이로드를 받습니다.

11.9.5. 쓰기 수신하기¶

반대 방향 – 클라이언트가 특성에 쓰는 것 – 은 특성을 write=True 또는 write_no_response=True 로 생성했을 때 사용할 수 있습니다. 주변장치는 aioble.Characteristic.written() 으로 다음 쓰기를 기다립니다:

cmd_char = aioble.Characteristic(env, CMD_UUID, write=True, capture=True)

async def handle_commands():
    while True:
        connection, data = await cmd_char.written()
        print("command from", connection.device.addr_hex(), "=", data)

capture=True 없이 written() 은 쓰기를 수행한 연결만 반환합니다. 새 값은 특성의 백킹 버퍼에 있으며 애플리케이션은 read() 로 가져옵니다. 애플리케이션이 첫 번째 값을 읽기 전에 두 번째 쓰기가 도착하면, 두 번째 값이 버퍼에서 첫 번째 값을 덮어쓰고 원래 값은 손실됩니다. written() 은 여전히 애플리케이션을 깨우지만, 쓰기 한 번마다가 아니라 “새로운 것이 있다”는 사실당 한 번만 깨웁니다.

capture=True 키워드가 이를 해결합니다. 들어오는 각 쓰기는 모듈 전역 큐에 추가되고, written() 은 모든 개별 쓰기에 대해 (connection, data) 튜플을 반환합니다. 애플리케이션 루프는 각각을 도착 순서대로 정확히 한 번씩 보게 됩니다. 두 가지 실용적인 결과가 있습니다:

큐는 크기가 제한되어 있으며 장치의 capture가 활성화된 모든 특성에 걸쳐 공유됩니다. 연속된 쓰기의 짧은 버스트는 허용되지만, 지속적인 오버런(애플리케이션이 비우는 속도보다 쓰기가 더 빠르게 도착하는 경우)은 큐에 들어 있는 가장 오래된 항목을 조용히 버립니다. 또한 한 특성에서 발생하는 버스트성 트래픽이 다른 특성의 대기 항목을 밀어낼 수 있습니다.
모든 값이 중요한 명령 스타일 쓰기에는 capture=True 를 선택하세요. 최신 값만이 관심 대상인 상태 스타일 특성에는 끄세요.

클라이언트의 읽기를 정적인 값이 아니라 요청 시 실행되는 코드로 응답해야 한다면 on_read() 를 재정의하세요. 이 메서드는 읽기가 들어올 때 동기적으로 호출됩니다. 읽기를 허용하려면 0 을 반환하고(write() 의 현재 값이 전송됨), 거부하려면 0이 아닌 ATT 오류 코드를 반환하세요:

import time

_ATT_ERR_READ_NOT_PERMITTED = const(0x02)
_MIN_READ_INTERVAL_MS = const(1000)            # at most once per second

class TempChar(aioble.Characteristic):
    _last_read_ms = 0

    def on_read(self, connection):
        now = time.ticks_ms()
        if time.ticks_diff(now, self._last_read_ms) < _MIN_READ_INTERVAL_MS:
            return _ATT_ERR_READ_NOT_PERMITTED
        self._last_read_ms = now
        self.write(encode_temperature(read_sensor()))
        return 0

temp_char = TempChar(env, TEMP_UUID, read=True)

콜백은 GATT 스택이 읽기를 처리하기 직전에 센서를 샘플링하고 특성의 값을 업데이트하므로, 클라이언트는 항상 최신 데이터를 봅니다. 속도 제한은 클라이언트가 센서를 샘플링 가능한 속도보다 빠르게 두드리는 것을 막습니다. 1초 쿨다운 안에 발생하는 모든 읽기는 오래된 값 대신 Read Not Permitted ATT 오류로 반환됩니다.

11.9.5.1. 더 큰 백킹 버퍼 – BufferedCharacteristic¶

일반 Characteristic 의 백킹 버퍼는 폭이 20바이트입니다 – 기본 23바이트 MTU에서의 실질적 한계입니다. 일반 특성에 그보다 많은 값을 쓰는 클라이언트는 값이 잘립니다. 더 큰 수신 값이나 애플리케이션 루프가 나중에 따라잡을 연속 쓰기를 큐잉하려면, 특성을 BufferedCharacteristic 로 선언하고 버퍼 크기를 미리 선택하세요:

blob = aioble.BufferedCharacteristic(
    service, BLOB_UUID,
    max_len=512, append=True,
    write=True, capture=True,
)

async def receive_blob():
    while True:
        connection, chunk = await blob.written()
        handle_chunk(connection, chunk)

일반 Characteristic 과 구별되는 두 가지 조절 항목이 있습니다:

max_len 은 백킹 버퍼의 크기(바이트)입니다. 클라이언트가 수행할 것으로 예상되는 가장 큰 단일 쓰기에 맞게 선택하세요(MTU 협상 후).
append=True 는 순차적 쓰기를 덮어쓰지 않고 버퍼에 추가하게 만듭니다 – 여러 번의 쓰기에 걸쳐 도착하는 값(펌웨어 업데이트 청크, 로그 줄)을 수신하는 데 유용합니다. append=False 이면 버퍼는 일반 특성처럼 동작하되 폭만 더 넓습니다.

다른 모든 생성자 플래그(read, write, notify, indicate, capture, initial)는 기본 특성으로 변경 없이 전달됩니다.

11.9.6. 표준 서비스와 SIG가 할당한 UUID¶

할당된 번호 UUID(배터리 서비스의 경우 0x180F, 환경 센싱의 경우 0x181A, 심박수의 경우 0x180D 등)를 사용하면 휴대폰의 일반 Bluetooth 메뉴나 모든 서드파티 스캐너 앱이 커스텀 클라이언트 코드 없이도 장치의 용도를 식별할 수 있습니다. 각 표준 특성 내부의 바이트 레이아웃 또한 사양에 의해 고정되어 있습니다 – 배터리 레벨(0x2A19)은 0..100 범위의 단일 바이트이고, 온도(0x2A6E)는 0.01 도-C 단위의 리틀 엔디안 sint16입니다. 표준 서비스에 맞지 않는 애플리케이션의 경우, 128비트 UUID를 한 번 생성하여 장치의 서비스와 특성 전반에 걸쳐 사용하세요.

커스텀 UUID만 게시하는 주변장치도 여전히 괜찮습니다 – 다만 그 UUID를 아는 커스텀 클라이언트 앱이 필요할 뿐입니다.

참고

BLE 값은 어디서나 리틀 엔디안 입니다 – GATT 사양, 모든 표준 특성, 모든 광고 필드가 그렇습니다. 멀티 바이트 정수는 하위 바이트가 먼저 전송됩니다. struct 형식 문자열의 < 접두사가 인코딩/디코딩에 필요한 것입니다("<h", "<H", "<I", …). 리틀 엔디안 MCU에서 기본 네이티브 바이트 순서를 사용해도 지금은 우연히 동작하지만, < 를 명시적으로 적는 것이 안전한 습관입니다.

11.9.7. 그 모든 것의 배후에 있는 무선 송수신기¶

무선 송수신기는 첫 번째 aioble 코루틴이 그것을 건드리는 순간 켜집니다. central이 연결되기 전까지 주변장치는 짧은 광고 버스트와 절전 상태를 오가며 시간을 보냅니다. 연결 후에는 협상된 연결 간격을 따릅니다. 주변장치는 광고마다 약간의 전력 비용을 지불하므로, aioble.advertise() 의 interval_us 선택은 검색 지연 시간과 배터리 수명을 절충하기 위해 주변장치가 가진 가장 직접적인 조절 수단입니다.