Add heartrate monitor examples for central and peripheral

[chadrockey]

As promised. :)

Take a look at these, I'd love to update them and learn more about how you intended the library to be used and serve as "the defacto" examples for BLE.

For better or worse, a pair of scripts like this seems to accompany most Bluetooth implementations. Now you've got them too.

[ukBaz]

Thanks for taking the time to do this. I will take a longer look at this the weekend.

As a quick initial quick initial look I ran the linting checks on the code and there were a few issues raised by it:

(venv) pi@SensePi:~/project/python-bluezero $ pycodestyle examples
examples/heartrate_monitor_central.py:5:16: E222 multiple spaces after operator
examples/heartrate_monitor_central.py:7:18: E222 multiple spaces after operator
examples/heartrate_monitor_central.py:9:1: E302 expected 2 blank lines, found 1
examples/heartrate_monitor_central.py:9:80: E501 line too long (85 > 79 characters)
examples/heartrate_monitor_central.py:10:80: E501 line too long (85 > 79 characters)
examples/heartrate_monitor_central.py:16:1: W293 blank line contains whitespace
examples/heartrate_monitor_central.py:30:1: E302 expected 2 blank lines, found 1
examples/heartrate_monitor_central.py:34:1: W293 blank line contains whitespace
examples/heartrate_monitor_central.py:42:1: W293 blank line contains whitespace
examples/heartrate_monitor_central.py:45:1: E302 expected 2 blank lines, found 1
examples/heartrate_monitor_central.py:48:80: E501 line too long (84 > 79 characters)
examples/heartrate_monitor_central.py:52:80: E501 line too long (98 > 79 characters)
examples/heartrate_monitor_central.py:60:5: E265 block comment should start with '# '
examples/heartrate_monitor_central.py:76:5: E266 too many leading '#' for block comment
examples/heartrate_monitor_central.py:76:5: E303 too many blank lines (2)
examples/heartrate_monitor_central.py:79:80: E501 line too long (87 > 79 characters)
examples/heartrate_monitor_central.py:82:5: E265 block comment should start with '# '
examples/heartrate_monitor_central.py:87:1: W293 blank line contains whitespace
examples/heartrate_monitor_central.py:99:1: E303 too many blank lines (3)
examples/heartrate_monitor_peripheral.py:12:16: E222 multiple spaces after operator
examples/heartrate_monitor_peripheral.py:14:18: E222 multiple spaces after operator
examples/heartrate_monitor_peripheral.py:18:1: E302 expected 2 blank lines, found 0
examples/heartrate_monitor_peripheral.py:36:1: E302 expected 2 blank lines, found 1
examples/heartrate_monitor_peripheral.py:43:1: E302 expected 2 blank lines, found 1
examples/heartrate_monitor_peripheral.py:70:1: E302 expected 2 blank lines, found 1
examples/heartrate_monitor_peripheral.py:74:1: E302 expected 2 blank lines, found 1
examples/heartrate_monitor_peripheral.py:81:41: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:82:41: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:88:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:89:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:89:61: E261 at least two spaces before inline comment
examples/heartrate_monitor_peripheral.py:89:80: E501 line too long (126 > 79 characters)
examples/heartrate_monitor_peripheral.py:90:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:91:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:92:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:93:36: E124 closing bracket does not match visual indentation
examples/heartrate_monitor_peripheral.py:96:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:97:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:98:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:99:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:100:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:101:36: E124 closing bracket does not match visual indentation
examples/heartrate_monitor_peripheral.py:104:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:105:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:106:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:107:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:108:36: E127 continuation line over-indented for visual indent
examples/heartrate_monitor_peripheral.py:109:36: E124 closing bracket does not match visual indentation
examples/heartrate_monitor_peripheral.py:117:55: W292 no newline at end of file

[ukBaz]

I've started running the code. Overall it looks good and worth pursuing. There are a few issues that need to be resolved first. I'll add them to this ticket as I find them.

[ukBaz]

It looks like the intention was to have the heart rate increment every 1 second and then reset to 60 when it got above 180. That doesn't happen. It appears you have heartrate and heart_rate. https://github.com/chadrockey/python-bluezero/blob/0351ce3a8b5104e10b825a2df47d490fd5be3f9b/examples/heartrate_monitor_peripheral.py#L32-L34

[ukBaz]

heartrate_monitor_peripheral.py

Heart Rate Control Point characteristic is intended to reset the value of the Energy Expended field in the Heart Rate Measurement characteristic to 0. However that value isn't being tracked by heartrate_monitor_peripheral.py. Would it be easy to add this functionality? I notice from the Heart Rate Service specifiation that the Heart Rate Control Point characteristic is only mandatory if the Energy Expended feature is supported in the Heart Rate Measurement characteristic. Currently you are not supporting it so removing the control point could be an alternative.

[ukBaz]

For testing heart rate heartrate_monitor_central.py I've used the following app: https://play.google.com/store/apps/details?id=io.github.webbluetoothcg.bletestperipheral

[ukBaz]

Central

Printing out the location of the sensor as 2 might be a bit cryptic for many.

I think you could create an enum class like:

class SensorLocation(IntEnum):
    OTHER = 0
    CHEST = 1
    WRIST = 2
    FINGER = 3
    HAND = 4
    EAR_LOBE = 5
    FOOT = 6

The change the print to:

    sensor_location = SensorLocation(
        int.from_bytes(location_char.value, 'little')
    )
    print(f"The location of the heart rate sensor is: {sensor_location!r}")

This would give the following output which I think might be a little more helpful.

The location of the heart rate sensor is: <SensorLocation.WRIST: 2>

I have also used Python f-strings for formatting the output rather than the % format.

And used int.from_bytes to convert the characteristic value to an integer from bytes.

[ukBaz]

Central

The Heart Rate Control Point characteristic worked correctly when I uncommented it.

The only issue was you were trying to send a 0 which is reserved.

When I changed it to send a 1 it worked as expected.

[ukBaz]

central

The test app was sending Heart Rate and Energy Expended For a Heart Rate = 66 and Energy Expended = 555

Raw: [dbus.Byte(8), dbus.Byte(66), dbus.Byte(43), dbus.Byte(2)]

Would be good if the central supported it but not essential

[ukBaz]

A suggestion of how on_new_heart_rate_measurement function could be redone to handle flags, HR format and energy expended value.

def get_hr_flags(data):
    flags = []
    for bit in range(5):
        flags.append(bool(data >> bit & 1))
    return flags

def on_new_heart_rate_measurement(iface, changed_props, invalidated_props):
    value = bytes(changed_props.get('Value', []))

    if not value:
        return
    byte_loc = 0
    (hr_value_format,
     sensor_contact_detected,
     sensor_contact_supported,
     energy_expended_present,
     rr_intervals_present) = get_hr_flags(value[byte_loc])
    byte_loc += 1
    print("\nHeart Rate Notification:")
    print(f"\tFlags:")
    print(f"\t\t{hr_value_format=}")
    print(f"\t\t{sensor_contact_detected=}")
    print(f"\t\t{sensor_contact_supported=}")
    print(f"\t\t{energy_expended_present=}")
    print(f"\t\t{rr_intervals_present=}")
    if hr_value_format:
        hr, = struct.unpack_from("<H", value, byte_loc)
        byte_loc += 2
    else:
        hr, = struct.unpack_from("<B", value, byte_loc)
        byte_loc += 1
    print(f"\tHeart Rate: {hr}")
    if energy_expended_present:
        ee, = struct.unpack_from("<H", value, byte_loc)
        byte_loc += 2
        print(f"\tEnergy Expended: {ee}")

[chadrockey]

I’m not sure I want to tackle energy or the control point to be actually implemented. Maybe we make it a stub in the peripheral and support it in the central? Very few other examples go that far and it’s conceptually too complex and gets in the way of demonstrating simple uses for the library. The core purpose is to show someone a minimal script that uses scan, connect, read, write, and notify.

I’ll work to address the comments in the next week. Good catch on the bitshift instead of struct. You can tell I’m a C++ firmware dev at heart. No struct available for me at that level.

As for the global, how does Python handle function static variables? I made it a global as I wanted the value to increment each call. If the global really bothers you and there’s no other static functionality, we could use random to generate a number instead of incrementing. If this were real code, we’d obviously use a class, but I would prefer these examples to be functions only. I feel like the best tutorials let newer programmers make whatever mistakes and unmaintainable code so that they get a fast victory. If they change the UUIDs at the top of the file, they might be connecting to and reading their BLE device in a few minutes. Sometimes the easiest to read code isn’t the easiest to maintain.

Can we add a third file for enums and definitions that support and explain the standard? I’d like the two main scripts to be pretty light. In hardware, when we have to implement a spec, we usually copy the flags into a header. If someone is super interested in the spec, they can go into the third file, but otherwise we will replace the magic numbers with ENUMs and human readable context.

Python also does not have any great bitwise manipulation so I think our best compromise is to put ENUM like flags there and then use bitwise operations. I don’t think there’s a way for someone to create useful BLE applications without understanding binary AND and binary OR.

So for example, the flags will be created something like:

flags = ???.FORMAT_UINT8 | ???.SENSOR_CONACT_SUPPORTED | etc

where the enum is:

class ???:
       FORMAT_UINT8 = 0b00000000
       FORMAT_UINT16 = 0b00000001
       SENSOR_CONTACT_SUPPORTED = 0b00000010

And so on

It’s a common usage pattern to OR flags together to create options like this.

The central can also use this for the format check such as:

if flags & ???.FORMAT_UINT16:
     # process as uint16
else:
     # process as uint8

It doesn’t quite work for uint8 because that’s an absence of a definition, so we may have it be the assumption that it’s uint8 once I double check the standard to make sure the flag is uint16 format.

[ukBaz]

1. Energy and Control point support Yep, understand about not wanting to do Energy Expended and Control Point. I don't think it needs to be a full implementation but I understand the balance between demonstrating the functionality and overwhelming the learner.
2. Use of global That is my bad. I didn't spot that it was the being updated. Having the global does simplify things. I agree there is a balance to be had between showing best practice and obfuscating the point you are trying to demonstrate.
3. Third file of enums I am not sure I like the idea of having a third file with enums in. I don't think there is enough to justify it and will look out of place in the examples directory.

My thought would be to put all of the non-Bluetooth functionality in a class as (in my mind) that simplifies things a lot. I have quickly cobbled something together something for how I would structure the peripheral. Not saying it is better, just offering it as an alternative data point that will hopefully lead to a better way.

"""Example of how to create a Peripheral device/GATT Server"""
# Standard modules
from dataclasses import dataclass
from enum import IntEnum
import logging
import struct

# Bluezero modules
from bluezero import async_tools
from bluezero import adapter
from bluezero import peripheral

HRM_SRV = '0000180D-0000-1000-8000-00805f9b34fb'
HR_MSRMT_UUID = '00002a37-0000-1000-8000-00805f9b34fb'
BODY_SNSR_LOC_UUID = '00002a38-0000-1000-8000-00805f9b34fb'
HR_CTRL_PT_UUID = '00002a39-0000-1000-8000-00805f9b34fb'

class MeasurementFlags(IntEnum):
    FORMAT_UINT16 = 0x01
    SENSOR_CONTACT_DETECTED = 0x02
    SENSOR_CONTACT_SUPPORTED = 0x04
    ENERGY_EXPENDED_PRESENT = 0x08
    RR_INTERVALS_PRESENT = 0x10

class SensorLocation(IntEnum):
    OTHER = 0
    CHEST = 1
    WRIST = 2
    FINGER = 3
    HAND = 4
    EAR_LOBE = 5
    FOOT = 6

@dataclass
class Monitor:
    flags: int = (MeasurementFlags.SENSOR_CONTACT_SUPPORTED |
                  MeasurementFlags.SENSOR_CONTACT_DETECTED |
                  MeasurementFlags.ENERGY_EXPENDED_PRESENT)
    location: SensorLocation = SensorLocation.CHEST
    heart_rate: int = 60
    energy_expended: int = 808

    def read_measurement(self):
        """
        Example read callback. Value returned needs to a list of bytes/integers
        in little endian format.

        // Send one byte of flags
        // As well as an 8 bit measurement

        :return: list of uint8 values
        """
        self.heart_rate += 1
        # Average weight for women
        weight = 70
        # Simplified formula in Keytel paper (Metric Units) for women
        self.energy_expended += int(5 - 20.4022 + 0.4472 * self.heart_rate -
                                    0.1263 * weight)
        if self.heart_rate > 180:
            self.heart_rate = 60
        print("Sending Values:")
        print(f"\tFlags {self.flags:08b}")
        print(f"\tHeart Rate: {self.heart_rate}")
        print(f"\tEnergy Expended: {self.energy_expended} KJ")
        return struct.pack('<BBH',
                           self.flags,
                           self.heart_rate,
                           self.energy_expended)

    def read_sensor_location(self):
        print(f"Sending sensor location of {self.location}")
        return struct.pack('<B', self.location)

    def update_value(self, characteristic):
        """
        Example of callback to send notifications

        :param characteristic:
        :return: boolean to indicate if timer should continue
        """
        # read/calculate new value.
        new_value = self.read_measurement()
        # Causes characteristic to be updated and send notification
        characteristic.set_value(new_value)
        # Return True to continue notifying. Return a False will stop notifications
        # Getting the value from the characteristic of if it is notifying
        return characteristic.is_notifying

    def notify_callback(self, notifying, characteristic):
        """
        Noitificaton callback example. In this case used to start a timer event
        which calls the update callback ever 2 seconds

        :param notifying: boolean for start or stop of notifications
        :param characteristic: The python object for this characteristic
        """
        if notifying:
            async_tools.add_timer_seconds(1, self.update_value, characteristic)

    def write_control_point(self, value, options):
        control_point,  = struct.unpack('<B', bytes(value))
        if control_point == 1:
            print(f"Received control point value of: {control_point}")
            print(f"Resetting Energy Expended to 0 from {self.energy_expended}")
            self.energy_expended = 0

def ble_heart_rate(adapter_address, monitor):
    """Creation of peripheral"""
    logger = logging.getLogger('localGATT')
    logger.setLevel(logging.DEBUG)

    # Create peripheral
    hr_monitor = peripheral.Peripheral(adapter_address,
                                       local_name='Heart Rate Monitor',
                                       appearance=0x0340)
    # Add service
    hr_monitor.add_service(srv_id=1, uuid=HRM_SRV, primary=True)

    # Add characteristics
    hr_monitor.add_characteristic(srv_id=1, chr_id=1, uuid=HR_MSRMT_UUID,
                                  value=[], notifying=False,
                                  # HR Service spec is notify only
                                  # Including read for tutorial
                                  flags=['read', 'notify'],
                                  read_callback=monitor.read_measurement,
                                  write_callback=None,
                                  notify_callback=monitor.notify_callback
                                  )

    hr_monitor.add_characteristic(srv_id=1, chr_id=2, uuid=BODY_SNSR_LOC_UUID,
                                  value=[], notifying=False,
                                  flags=['read'],
                                  read_callback=monitor.read_sensor_location,
                                  write_callback=None,
                                  notify_callback=None
                                  )

    hr_monitor.add_characteristic(srv_id=1, chr_id=3, uuid=HR_CTRL_PT_UUID,
                                  value=[], notifying=False,
                                  flags=['write'],
                                  read_callback=None,
                                  write_callback=monitor.write_control_point,
                                  notify_callback=None
                                  )

    # Publish peripheral and start event loop
    hr_monitor.publish()

def main(adapter_address):
    monitor = Monitor()
    ble_heart_rate(adapter_address, monitor)

if __name__ == '__main__':
    # Get the default adapter address and pass it to main
    main(list(adapter.Adapter.available())[0].address)

[chadrockey]

Comments addressed. A few big notes:

I decided that energy expended was important because we really need to have an example of central's write value support, so I left energy expended in. Please check the center correctly parses your app's output. If prints a warning but otherwise is fine if a hardware heartrate adapter does not support the characteristic.

I also would prefer that this remains in script format rather than using classes. You already have a lot of great examples of classes, but nothing that necessarily steps through using central. I also believe that newer programmers or hardware hackers would appreciate a script format where they can copy/paste this and start modifying it for their one off programs.

I backed off from having a third shared data class file, as I've submitted here is likely the easiest to digest despite the code duplication. Usually one side is an existing device or you're writing them in multiple languages, so code duplication is the expectation. This is also the cleanest to copy/paste into your own file and modify, but still use some of the best practices we've demonstrated.

The original motivation was to create a 'Hello World'-like central usage script that can be used as a starting point for applications. If these scripts are modified then they serve as a known working point that would demonstrate intended usage patterns and prevent issues like #370

Things to verify as I only have one compatible device (laptop is still using Ubuntu 18.04 with bluez < 5.50):

• [x] I couldn't verify that central prints the ENUM format for body sensor location, please check that if you can
• [x] I also couldn't easily verify that central prints out the correct format for energy_expended

[ukBaz]

Good work! Thanks @chadrockey