healthypi_dumper.py

import serial
import struct
import sys
import json
from serial.tools.list_ports import comports


header = bytes([0x0A, 0xFA])
footer = bytes([0x00, 0x0b])

if __name__ == '__main__':
    if len(sys.argv) > 1:
        port = sys.argv[1]
    else:
        ports = [p.device for p in comports()]
        if len(ports) > 1:
            print((
                "Multiple serial devices %s. Pass the "
                "desired one as a command line argument") % (', '.join(ports),
                ), file=sys.stderr
            )
            sys.exit(1)
        if len(ports) == 0:
            print("No serial device connected", file=sys.stderr)
            sys.exit(1)
        port = ports[0]

    data = serial.Serial(port)

    def read_literal(string):
        i = 0
        while i < len(string):
            c = data.read(1)[0]
            if string[i] == c:
                i += 1
            elif string[0] == c:
                i = 1
            else:
                i = 0

    while True:
        read_literal(header)

        data_len, = struct.unpack("<H", data.read(2))

        marker = data.read(1)[0]
        assert marker == 2, marker

        packet = data.read(data_len)

        if data_len != 20:
            print(
                "Unexpected packet! %d bytes %r" % (data_len, packet),
                file=sys.stderr)
        else:
            parsed = struct.unpack(
                "<HHIIHBBBBBB",
                packet
            )

            fields = [
                "ecg", "resp", "sp02_ir", "sp02_red", "temp",
                "global_respiration", "global_spo2", "global_heartrate",
                "bp_value_sys", "bp_value_dia",
                "leadstatus"
            ]

            assert len(fields) == len(parsed)

            print(json.dumps(dict(zip(fields, parsed))))

        next2 = data.read(2)
        if next2 != footer:
            print("Uh oh. Was expecting footer but got", next2, file=sys.stderr)