【体重自動記録ツール】体重計のBLE情報を取得してみる #1

pi@raspberrypi:~ $ cat /etc/os-release 
PRETTY_NAME="Raspbian GNU/Linux 10 (buster)"
NAME="Raspbian GNU/Linux"
VERSION_ID="10"
VERSION="10 (buster)"
VERSION_CODENAME=buster
ID=raspbian
ID_LIKE=debian
HOME_URL="http://www.raspbian.org/"
SUPPORT_URL="http://www.raspbian.org/RaspbianForums"
BUG_REPORT_URL="http://www.raspbian.org/RaspbianBugs

pi@raspberrypi:~ $ python3 -V
Python 3.7.3

pi@raspberrypi:~ $ sudo apt-get install python-dev libbluetooth3-dev
pi@raspberrypi:~ $ sudo apt-get install libglib2.0 libboost-python-dev libboost-thread-dev

pi@raspberrypi:~ $ sudo pip3 install -r requirements.txt

requirements.txt
-------------------------------------------------------
bleak==0.12.1

#!/usr/bin/env python3

import sys
import asyncio
import requests
import warnings
from datetime import datetime
from logging import Logger, getLogger, StreamHandler, DEBUG
from typing_extensions import Final
from bleak import BleakClient
from bleak import _logger as logger

WEIGHT_MEASUREMENT:Final[str] = "00002a9d-0000-1000-8000-00805f9b34fb"

warnings.simplefilter('ignore', FutureWarning)

class BleHandler():
    """ble handler.
    """
    SLEEP_TIME:Final[int] = 2
    TIME_OUT:Final[int] = 20

    def __init__(self, address:str, debug:bool=False) -> None:
        self.address:str = address
        self._logger:Logger = getLogger()

        if debug:
            l = getLogger("asyncio")
            l.setLevel(DEBUG)
            h = StreamHandler(sys.stdout)
            h.setLevel(DEBUG)
            l.addHandler(h)
            logger.addHandler(h)

    async def __call__(self) -> None:
        """when call, connect to target address device.
        """
        self._logger.info(f'connect to device({self.address}) start.')
        await self.connect_to_device()
    
    def _calculate_weight(self, data) -> float:
        """calculate weight value from bytearray.
        Args:
            data ([type]): [description]
        Returns:
            float: weight(format : xxx.xx)
        """
        return (data[1] + data[2] * 256) * 0.005

    def _weight_measurement_notification_handler(self, sender, data:bytearray):
        """Weight measurement notification handler."""
        weight:float = (data[1] + data[2] * 256) * 0.005
        self._logger.info(f'weight measurement notification handler: {data.hex()} / {weight} kg')

    def _disconnect_callback(self, client: BleakClient):
        """disconnect callback. only logging message."""
        self._logger.info(f'Client with address {client.address} got disconnected. try to reconnect.')

    async def connect_to_device(self):
        while True:
            self._logger.info("Waiting connect to device.")
            try:
                async with BleakClient(self.address, timeout=self.TIME_OUT, disconnected_callback=self._disconnect_callback) as client:

                    if await client.is_connected():
                        self._logger.info("Connect to device successfuly.")
                        await client.start_notify(
                            WEIGHT_MEASUREMENT, self._weight_measurement_notification_handler
                        )

                        while True:
                            if not await client.is_connected():
                                self._logger.debug("Device disconnected.")
                                break
                            await asyncio.sleep(self.SLEEP_TIME)
                            self._logger.debug("wait for message arraived from device.")
                    else:
                        self._logger.debug("Device disconnected.")
            except Exception as e:
                self._logger.error(f"Exception when connect: {e}")
            
            await asyncio.sleep(self.SLEEP_TIME)

#!/usr/bin/env python3

import sys
import asyncio
from logging import getLogger, StreamHandler, Formatter, DEBUG, INFO
from typing_extensions import Final
from blehandler import BleHandler
from typing import List
from bleak import discover

TARGET_ADDRESS:Final[str] = "<your-device-address>"

logger = getLogger()
logger.setLevel(INFO)
_streamHandler = StreamHandler(sys.stdout)
_streamHandler.setLevel(INFO)
_streamHandler.setFormatter(Formatter('[%(asctime)s][%(levelname)s] %(message)s'))
logger.addHandler(_streamHandler)

async def main():
    while True:
        logger.info('start device scan...')
        tasks = []
        devices:List = await discover()
        for device in devices:
            if device.address == TARGET_ADDRESS:
                handler = BleHandler(TARGET_ADDRESS, debug=True)
                tasks.append(asyncio.ensure_future(handler()))
                logger.info(f'found target device : {device}. discover process end.')
        
        if len(tasks) > 0:
            [await task for task in tasks]
        else:
            logger.info('target device not found. rescan after sleep 5s.')
        await asyncio.sleep(5)

asyncio.run(main())

pi@raspberrypi:~ $ sudo python3 weight_measurement.py

[2021-09-20 23:04:58,447][INFO] start device scan...
[2021-09-20 23:05:03,609][INFO] found target device : xx:xx:xx:xx:xx:xx: QN-Scale. discover process end.
[2021-09-20 23:05:03,610][INFO] connect to device(xx:xx:xx:xx:xx:xx) start.
[2021-09-20 23:05:03,610][INFO] Waiting connect to device.
[2021-09-20 23:05:04,852][INFO] Client with address xx:xx:xx:xx:xx:xx got disconnected. try to reconnect.
[2021-09-20 23:05:13,394][INFO] Waiting connect to device.
[2021-09-20 23:05:15,040][INFO] Connect to device successfuly.
[2021-09-20 23:05:27,596][INFO] weight measurement notification handler: 024209e107010e003023 / 11.85 kg
[2021-09-20 23:05:35,798][INFO] Client with address xx:xx:xx:xx:xx:xx got disconnected. try to reconnect.
[2021-09-20 23:05:39,145][INFO] Waiting connect to device.