feat: Refactor HR data processing and visualization with new HrPacket model

app/model/__init__.py

@@ -138,3 +138,76 @@ class AlgoReport(BaseModel):
             accel_z=accel_z,
             data=data,
         )
+
+
+class HrConfidence(IntEnum):
+    """Equivalent to max::HR_CONFIDENCE"""
+
+    ZERO = 0
+    LOW = 1
+    MEDIUM = 2
+    HIGH = 3
+
+
+def hr_confidence_to_num(hr_confidence: HrConfidence) -> float:
+    if hr_confidence == HrConfidence.ZERO:
+        return 0
+    elif hr_confidence == HrConfidence.LOW:
+        return 25
+    elif hr_confidence == HrConfidence.MEDIUM:
+        return 62.5
+    elif hr_confidence == HrConfidence.HIGH:
+        return 100
+    else:
+        raise ValueError(f"Invalid HR confidence: {hr_confidence}")
+
+
+@dataclass
+class HrStatusFlags:
+    # 2 bits
+    hr_confidence: HrConfidence
+    # 1 bit
+    is_active: bool
+    # 1 bit
+    is_on_skin: bool
+    # 4 bits
+    battery_level: int
+
+    @staticmethod
+    def unmarshal(data: bytes) -> "HrStatusFlags":
+        val = data[0]
+        return HrStatusFlags(
+            hr_confidence=HrConfidence(val & 0b11),
+            is_active=(val & 0b100) != 0,
+            is_on_skin=(val & 0b1000) != 0,
+            battery_level=val >> 4,
+        )
+
+
+@dataclass
+class HrPacket:
+    # 8 bits
+    status: HrStatusFlags
+    # 8 bits
+    id: int
+    # 8 bits
+    hr: int
+    # 8 bits (as `n`) + n x 24 bits
+    raw_data: list[int]
+
+    @staticmethod
+    def unmarshal(data: bytes) -> "HrPacket":
+        status = HrStatusFlags.unmarshal(data[0:1])
+        id = data[1]
+        hr = data[2]
+        raw_data_count = data[3]
+        raw_data_payload = data[4:]
+        if len(raw_data_payload) != (expected_raw_data_len := raw_data_count * 3):
+            raise ValueError(
+                f"Invalid raw data payload length: {len(raw_data_payload)}, expected {expected_raw_data_len}"
+            )
+        raw_data = [
+            int.from_bytes(raw_data_payload[i : i + 3], "little")
+            for i in range(0, expected_raw_data_len, 3)
+        ]
+        return HrPacket(status, id, hr, raw_data)

ble_forward.py

@@ -6,11 +6,11 @@ from typing import Final, Optional
 from loguru import logger
 from anyio import create_udp_socket, create_connected_udp_socket
 
-DEVICE_NAME: Final[str] = "MAX-BAND"
+DEVICE_NAME: Final[str] = "MAX-HUB"
 UDP_SERVER_HOST: Final[str] = "localhost"
 UDP_SERVER_PORT: Final[int] = 50_000
 BLE_HR_SERVICE_UUID: Final[str] = "180D"
-BLE_HR_CHARACTERISTIC_RAW_UUID: Final[str] = "c4f5233d-430a-4ca1-bb60-1d896c10e807"
+BLE_HR_CHARACTERISTIC_RAW_UUID: Final[str] = "ff241160-8a02-4626-b499-b1572d2b5a29"
 
 
 async def main():

main.py

@@ -27,7 +27,7 @@ from pydantic import BaseModel, computed_field
 from datetime import datetime, timedelta
 import awkward as ak
 from awkward import Array as AwkwardArray, Record as AwkwardRecord
-from app.model import AlgoReport
+from app.model import AlgoReport, HrPacket, hr_confidence_to_num
 from app.utils import Instant
 from collections import deque
 from dataclasses import dataclass
@@ -36,7 +36,7 @@ from dataclasses import dataclass
 class AppHistory(TypedDict):
     timescape: deque[datetime]
     hr_data: deque[float]
-    hr_conf: deque[int]  # in %
+    hr_conf: deque[float]  # in %
     accel_x_data: deque[int]
     accel_y_data: deque[int]
     accel_z_data: deque[int]
@@ -153,38 +153,28 @@ def main():
             message = state["message_queue"].receive_nowait()
         except anyio.WouldBlock:
             continue
-        report = AlgoReport.unmarshal(message)
+        try:
-        if state["refresh_inst"].mut_every_ms(500):
+            packet = HrPacket.unmarshal(message)
-            md_placeholder.markdown(
+        except ValueError as e:
-                f"""
+            logger.error(f"bad packet: {e}")
-            - HR: {report.data.hr_f}bpm
+            continue
-            - HR CONF: {report.data.hr_conf}%
+
-            - ACTIVITY: {report.data.activity_class.name}
-            - SCD: {report.data.scd_contact_state.name}
-                        """
-            )
         with placeholder.container():
-            history["timescape"].append(datetime.now())
+            history["hr_data"].append(packet.hr)
-            history["hr_data"].append(report.data.hr_f)
+            history["hr_conf"].append(hr_confidence_to_num(packet.status.hr_confidence))
-            history["hr_conf"].append(report.data.hr_conf)
+            history["pd_data"].extend(packet.raw_data)
-            history["accel_x_data"].append(report.accel_x)
+            fig_hr, fig_pd = st.tabs(["Heart Rate", "PD"])
-            history["accel_y_data"].append(report.accel_y)
-            history["accel_z_data"].append(report.accel_z)
-            history["pd_data"].append(report.led_2)
-            fig_hr, fig_accel, fig_pd = st.tabs(["Heart Rate", "Accelerometer", "PD"])
 
             with fig_hr:
                 st.plotly_chart(
                     go.Figure(
                         data=[
                             go.Scatter(
-                                x=list(history["timescape"]),
                                 y=list(history["hr_data"]),
                                 mode="lines",
                                 name="HR",
                             ),
                             go.Scatter(
-                                x=list(history["timescape"]),
                                 y=list(history["hr_conf"]),
                                 mode="lines",
                                 name="HR Confidence",
@@ -192,37 +182,11 @@ def main():
                         ]
                     )
                 )
-            with fig_accel:
-                st.plotly_chart(
-                    go.Figure(
-                        data=[
-                            go.Scatter(
-                                x=list(history["timescape"]),
-                                y=list(history["accel_x_data"]),
-                                mode="lines",
-                                name="x",
-                            ),
-                            go.Scatter(
-                                x=list(history["timescape"]),
-                                y=list(history["accel_y_data"]),
-                                mode="lines",
-                                name="y",
-                            ),
-                            go.Scatter(
-                                x=list(history["timescape"]),
-                                y=list(history["accel_z_data"]),
-                                mode="lines",
-                                name="z",
-                            ),
-                        ]
-                    )
-                )
             with fig_pd:
                 st.plotly_chart(
                     go.Figure(
                         data=[
                             go.Scatter(
-                                x=list(history["timescape"]),
                                 y=list(history["pd_data"]),
                                 mode="lines",
                                 name="PD",