imrishabh18/pedometer

companion/read_steps.py

"""Read the Stride GATT contract after the TI peripheral integration is flashed."""
import argparse
import asyncio
import json
import struct

SERVICE = "438d0001-0d15-4f45-9b3a-3a4c7942e201"
SNAPSHOT = "438d0002-0d15-4f45-9b3a-3a4c7942e201"

def decode(data):
    if len(data) != 14:
        raise ValueError(f"Expected 14 bytes, received {len(data)}")
    version, flags, steps, millivolts, soc, reserved = struct.unpack("<BBQHBB", data)
    if version != 1 or reserved != 0 or (soc > 100 and soc != 255):
        raise ValueError("Unsupported or malformed Stride snapshot")
    return dict(steps=steps, battery_mV=millivolts,
                battery_percent=None if soc == 255 else soc, fault_flags=flags)

async def main(seconds):
    from bleak import BleakClient, BleakScanner
    device = await BleakScanner.find_device_by_filter(
        lambda device, advertisement: SERVICE in
        [uuid.lower() for uuid in advertisement.service_uuids], timeout=15.0,
        service_uuids=[SERVICE])
    if device is None:
        raise RuntimeError("Stride not found. Flash the integrated peripheral and enable Bluetooth.")
    def report(sender, data):
        try:
            print(json.dumps(decode(data)), flush=True)
        except ValueError as error:
            print(json.dumps({"error": str(error)}), flush=True)
    async with BleakClient(device) as client:
        report(None, await client.read_gatt_char(SNAPSHOT))
        await client.start_notify(SNAPSHOT, report)
        await asyncio.sleep(seconds)
        await client.stop_notify(SNAPSHOT)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--seconds", type=float, default=120)
    args = parser.parse_args()
    if args.seconds <= 0:
        parser.error("--seconds must be positive")
    try:
        asyncio.run(main(args.seconds))
    except (RuntimeError, OSError) as error:
        parser.exit(1, f"{error}\n")