Heart Rate Monitor

#include <simpleble/SimpleBLE.h>

#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <iostream>
#include <mutex>
#include <thread>

const std::string HR_SERVICE = "0000180d-0000-1000-8000-00805f9b34fb";
const std::string HR_MEASUREMENT = "00002a37-0000-1000-8000-00805f9b34fb";

struct HeartRateMeasurement {
    uint16_t bpm = 0;
    bool contact_supported = false;
    bool contact_detected = false;
};

HeartRateMeasurement parse_heart_rate(const SimpleBLE::ByteArray& payload) {
    HeartRateMeasurement hr;
    if (payload.size() < 2) return hr;

    uint8_t flags = static_cast<uint8_t>(payload[0]);

    if (flags & 0x01) {
        // 16-bit heart rate value, little-endian.
        hr.bpm = static_cast<uint8_t>(payload[1]) | (static_cast<uint8_t>(payload[2]) << 8);
    } else {
        // 8-bit heart rate value.
        hr.bpm = static_cast<uint8_t>(payload[1]);
    }

    hr.contact_supported = (flags & 0x04) != 0;
    hr.contact_detected = (flags & 0x02) != 0;
    return hr;
}

int main() {
    auto adapters = SimpleBLE::Adapter::get_adapters();
    if (adapters.empty()) {
        std::cerr << "No Bluetooth adapters found." << std::endl;
        return 1;
    }
    auto adapter = adapters.front();

    // 1. Scan and keep the first device advertising the Heart Rate service.
    SimpleBLE::Peripheral monitor;
    adapter.set_callback_on_scan_found([&](SimpleBLE::Peripheral peripheral) {
        for (auto& service : peripheral.services()) {
            if (service.uuid() == HR_SERVICE && !monitor.initialized()) {
                monitor = peripheral;
                std::cout << "Found monitor: " << peripheral.identifier() << std::endl;
            }
        }
    });
    adapter.scan_for(5000);

    if (!monitor.initialized()) {
        std::cerr << "No heart rate monitor found." << std::endl;
        return 1;
    }

    // 2. Watch for disconnects. Only signal from the callback; act elsewhere.
    std::mutex mutex;
    std::condition_variable cv;
    bool disconnected = false;
    monitor.set_callback_on_disconnected([&]() {
        std::lock_guard<std::mutex> lock(mutex);
        disconnected = true;
        cv.notify_all();
    });

    // 3. Connect and subscribe.
    monitor.connect();
    monitor.notify(HR_SERVICE, HR_MEASUREMENT, [](SimpleBLE::ByteArray payload) {
        HeartRateMeasurement hr = parse_heart_rate(payload);
        std::cout << "Heart rate: " << hr.bpm << " bpm";
        if (hr.contact_supported && !hr.contact_detected) {
            std::cout << " (no skin contact)";
        }
        std::cout << std::endl;
    });

    // 4. Stream for 30 seconds or until the device disconnects.
    {
        std::unique_lock<std::mutex> lock(mutex);
        cv.wait_for(lock, std::chrono::seconds(30), [&]() { return disconnected; });
    }

    if (disconnected) {
        std::cout << "Device disconnected." << std::endl;
    } else {
        monitor.unsubscribe(HR_SERVICE, HR_MEASUREMENT);
        monitor.disconnect();
    }

    return 0;
}

import threading

import simplepyble

HR_SERVICE = "0000180d-0000-1000-8000-00805f9b34fb"
HR_MEASUREMENT = "00002a37-0000-1000-8000-00805f9b34fb"


def parse_heart_rate(payload: bytes) -> dict:
    flags = payload[0]

    if flags & 0x01:
        # 16-bit heart rate value, little-endian.
        bpm = int.from_bytes(payload[1:3], "little")
        offset = 3
    else:
        # 8-bit heart rate value.
        bpm = payload[1]
        offset = 2

    result = {
        "bpm": bpm,
        "contact_supported": bool(flags & 0x04),
        "contact_detected": bool(flags & 0x02),
    }

    if flags & 0x08:  # Energy Expended present.
        result["energy_kj"] = int.from_bytes(payload[offset : offset + 2], "little")
        offset += 2

    if flags & 0x10:  # RR intervals present (1/1024 s units).
        rr = []
        while offset + 1 < len(payload):
            rr.append(int.from_bytes(payload[offset : offset + 2], "little") / 1024.0)
            offset += 2
        result["rr_intervals_s"] = rr

    return result


def main():
    adapter = simplepyble.Adapter.get_adapters()[0]

    # 1. Scan for a device advertising the Heart Rate service.
    adapter.scan_for(5000)
    monitor = None
    for peripheral in adapter.scan_get_results():
        service_uuids = [service.uuid() for service in peripheral.services()]
        if HR_SERVICE in service_uuids:
            monitor = peripheral
            print(f"Found monitor: {peripheral.identifier()}")
            break

    if monitor is None:
        print("No heart rate monitor found.")
        return

    # 2. Watch for disconnects. Only signal from the callback; act elsewhere.
    disconnected = threading.Event()
    monitor.set_callback_on_disconnected(disconnected.set)

    # 3. Connect and subscribe.
    monitor.connect()

    def on_measurement(payload):
        hr = parse_heart_rate(bytes(payload))
        contact = "" if hr.get("contact_detected", True) else " (no skin contact)"
        print(f"Heart rate: {hr['bpm']} bpm{contact}")

    monitor.notify(HR_SERVICE, HR_MEASUREMENT, on_measurement)

    # 4. Stream for 30 seconds or until the device disconnects.
    if disconnected.wait(timeout=30):
        print("Device disconnected.")
    else:
        monitor.unsubscribe(HR_SERVICE, HR_MEASUREMENT)
        monitor.disconnect()


if __name__ == "__main__":
    main()