Biostrap/EVO Full Suite

CONTRIBUTORS.md

@@ -9,3 +9,4 @@
 - Jack Hung (jjhung66@stanford.edu)
 - Saarth Shah (sashah@ucsd.edu)
 - Suvan Kumar (kumarsuvan0@gmail.com)
+- Hee Jung Choi (cheejung@stanford.edu)

gitignore

@@ -0,0 +1,3 @@
+.DS_Store
+.project
+*.pyc

pyproject.toml

@@ -46,6 +46,7 @@ fbm = "^0.3.0"
 beautifulsoup4 = "^4.12.2"
 myfitnesspal = "^2.0.1"
 polyline = "^2.0.0"
+jupyter = "^1.0.0"
 
 [tool.poetry.group.dev.dependencies]
 bandit = "^1.7.1"
@@ -157,4 +158,4 @@ branch = true
 
 [coverage.report]
 fail_under = 50
-show_missing = true
+show_missing = true

tests/devices/biostrap/test_evo.py

@@ -0,0 +1,153 @@
+from datetime import datetime, timedelta
+
+import pytest
+
+import wearipedia
+
+data_formats = {
+    "bpm": (int, float),
+    "brpm": (int, float),
+    "hrv": (int, float),
+    "resting_bpm": (int, float),
+    "resting_hrv": (int, float),
+    "spo2": (int, float),
+    "rest_cals": (int, float),
+    "work_cals": (int, float),
+    "active_cals": (int, float),
+    "step_cals": (int, float),
+    "total_cals": (int, float),
+    "steps": (int, float),
+    "distance": (int, float),
+}
+
+
+@pytest.mark.parametrize("real", [True, False])
+def test_evo(real):
+    start_synthetic = datetime(2023, 6, 5)
+    end_synthetic = datetime(2023, 6, 20, 23, 59, 59)
+
+    device = wearipedia.get_device(
+        "biostrap/evo",
+        start_date=datetime.strftime(start_synthetic, "%Y-%m-%d"),
+        end_date=datetime.strftime(end_synthetic, "%Y-%m-%d"),
+    )
+    if real:
+        wearipedia._authenticate_device("evo", device)
+
+    for data_type, data_format in data_formats.items():
+        data = device.get_data(data_type)
+
+        # Checks specific to date-keyed data
+        if data_type in [
+            "rest_cals",
+            "work_cals",
+            "active_cals",
+            "step_cals",
+            "total_cals",
+        ]:
+            dates = list(data.keys())
+
+            # Check dates are consecutive and within the range
+            expected_date = start_synthetic
+            for date_str in dates:
+                date = datetime.strptime(date_str, "%Y-%m-%d")
+                assert (
+                    date == expected_date
+                ), f"Expected date {expected_date}, but got {date}"
+                expected_date += timedelta(days=1)
+
+        # Checks specific to datetime-keyed data
+        elif data_type in ["steps", "distance"]:
+
+            datetimes = [
+                datetime.strptime(dt_str, "%Y-%m-%d %H:%M:%S") for dt_str in data.keys()
+            ]
+
+            # Check that datetimes are within the specified range.
+            for dt in datetimes:
+                assert (
+                    start_synthetic <= dt <= end_synthetic
+                ), f"Datetime {dt} out of range"
+
+            # Check that the datetimes are sequential and increase by a minute.
+            datetimes.sort()
+            for i in range(1, len(datetimes)):
+                expected_dt = datetimes[i - 1] + timedelta(minutes=1)
+                assert (
+                    datetimes[i] == expected_dt
+                ), f"Expected datetime {expected_dt}, but got {datetimes[i]}"
+        else:
+            datetimes = [key[0] for key in data.keys()]
+
+            # Check datetimes are within the range
+            for datetime_str in datetimes:
+                dt = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S")
+                assert (
+                    start_synthetic <= dt <= end_synthetic
+                ), f"Datetime {dt} out of range"
+
+        # Check data values are of expected type
+        for value in data.values():
+            assert isinstance(
+                value, data_format
+            ), f"{data_type} data {value} is not a {data_format.__name__}"
+
+
+# from datetime import datetime
+
+# import pytest
+
+# import wearipedia
+
+
+# @pytest.mark.parametrize("real", [True, False])
+# def test_evo(real):
+#     start_synthetic = datetime(2023, 6, 5)
+#     end_synthetic = datetime(2023, 6, 20)
+
+#     device = wearipedia.get_device(
+#         "biostrap/evo",
+#         start_date=datetime.strftime(start_synthetic, "%Y-%m-%d"),
+#         end_date=datetime.strftime(end_synthetic, "%Y-%m-%d"),
+#     )
+#     if real:
+#         wearipedia._authenticate_device("evo", device)
+
+#     steps = device.get_data("steps")
+#     dates = list(steps.keys())  # extracting dates from steps data
+#     bpm = device.get_data("bpm")
+#     brpm = device.get_data("brpm")
+#     spo2 = device.get_data("spo2")
+
+#     assert len(dates) == len(steps) == (end_synthetic - start_synthetic).days + 1, (
+#         f"Expected dates and steps data to be the same length and to match the number of days between"
+#         f" {start_synthetic} and {end_synthetic}, but got {len(dates)}, {len(steps)}"
+#     )
+
+#     # first make sure that the dates are correct
+#     for date_1, date_2 in zip(dates[:-1], dates[1:]):
+#         assert (
+#             datetime.strptime(date_2, "%Y-%m-%d")
+#             - datetime.strptime(date_1, "%Y-%m-%d")
+#         ).days == 1, f"Dates are not consecutive: {date_1}, {date_2}"
+
+#     assert dates[0] == start_synthetic.strftime(
+#         "%Y-%m-%d"
+#     ), f"First date {dates[0]} is not equal to {start_synthetic.strftime('%Y-%m-%d')}"
+#     assert dates[-1] == end_synthetic.strftime(
+#         "%Y-%m-%d"
+#     ), f"Last date {dates[-1]} is not equal to {end_synthetic.strftime('%Y-%m-%d')}"
+
+#     # Now make sure that the steps are correct.
+#     for step in steps.values():
+#         assert isinstance(step, int), f"Step data {step} is not an integer"
+
+#     # Now make sure that the bpm, brpm and spo2 are correct.
+#     for hr in bpm.values():
+#         assert isinstance(hr, float), f"BPM data {hr} is not a float"
+
+#     for br in brpm.values():
+#         assert isinstance(br, float), f"BRPM data {br} is not a float"
+
+#     for s in spo2.values():
+#         assert isinstance(s, float), f"SPO2 data {s} is not a float"

wearipedia/__init__.py

@@ -61,6 +61,7 @@ def get_all_device_names():
 
     return [
         "apple/healthkit",
+        "biostrap/evo",
         "cronometer/cronometer",
         "whoop/whoop_4",
         "withings/scanwatch",

wearipedia/devices/__init__.py

@@ -1,4 +1,5 @@
 from .apple import *
+from .biostrap import *
 from .cronometer import *
 from .dexcom import *
 from .dreem import *

wearipedia/devices/biostrap/__init__.py

@@ -0,0 +1 @@
+from .evo import *

wearipedia/devices/biostrap/evo.py

@@ -0,0 +1,170 @@
+import os
+import pickle
+import webbrowser
+from datetime import datetime
+from urllib.parse import urlencode
+
+import requests
+
+from ...utils import seed_everything
+from ..device import BaseDevice
+from .evo_fetch import *
+from .evo_gen import *
+
+class_name = "EVO"
+
+
+class EVO(BaseDevice):
+    def __init__(self, seed=0, start_date="2023-06-05", end_date="2023-06-20"):
+        params = {
+            "seed": seed,
+            "start_date": str(start_date),
+            "end_date": str(end_date),
+        }
+
+        self._initialize_device_params(
+            [
+                "activities",
+                "bpm",
+                "brpm",
+                "hrv",
+                "resting_bpm",
+                "resting_hrv",
+                "spo2",
+                "rest_cals",
+                "work_cals",
+                "active_cals",
+                "step_cals",
+                "total_cals",
+                "sleep_session",
+                "sleep_detail",
+                "steps",
+                "distance",
+            ],
+            params,
+            {
+                "seed": 0,
+                "synthetic_start_date": "2023-06-05",
+                "synthetic_end_date": "2023-06-20",
+            },
+        )
+
+    def _default_params(self):
+        return {
+            "start_date": self.init_params["synthetic_start_date"],
+            "end_date": self.init_params["synthetic_end_date"],
+        }
+
+    def _get_real(self, data_type, params):
+        return fetch_real_data(
+            self.access_token, params["start_date"], params["end_date"], data_type
+        )
+
+    def _filter_synthetic(self, data, data_type, params):
+        start_date = params["start_date"]
+        end_date = params["end_date"]
+
+        start_datetime = f"{start_date} 00:00:00"
+        end_datetime = f"{end_date} 23:59:59"
+
+        # For data types that are stored with a date string as a key
+        if data_type in [
+            "rest_cals",
+            "work_cals",
+            "active_cals",
+            "step_cals",
+            "total_cals",
+            "sleep_session",
+            "sleep_detail",
+        ]:
+            return {
+                date: value
+                for date, value in data.items()
+                if start_date <= date <= end_date
+            }
+
+        # For data types that are stored with a datetime string as a key in a tuple
+        elif data_type in [
+            "bpm",
+            "brpm",
+            "spo2",
+            "activities",
+            "hrv",
+            "resting_bpm",
+            "resting_hrv",
+        ]:
+            return {
+                key: value
+                for key, value in data.items()
+                if start_datetime <= key[0] <= end_datetime
+            }
+
+        # For steps and distance that use datetime strings as keys (assuming distance also has datetime as per steps)
+        elif data_type in ["steps", "distance"]:
+            return {
+                datetime: value
+                for datetime, value in data.items()
+                if start_datetime <= datetime <= end_datetime
+            }
+
+        return data
+
+    def _gen_synthetic(self):
+        # generate random data according to seed
+        seed_everything(self.init_params["seed"])
+        # and based on start and end dates
+        (
+            self.activities,
+            self.bpm,
+            self.brpm,
+            self.hrv,
+            self.resting_bpm,
+            self.resting_hrv,
+            self.spo2,
+            self.rest_cals,
+            self.work_cals,
+            self.active_cals,
+            self.step_cals,
+            self.total_cals,
+            self.sleep_session,
+            self.sleep_detail,
+            self.steps,
+            self.distance,
+        ) = create_syn_data(
+            self.init_params["synthetic_start_date"],
+            self.init_params["synthetic_end_date"],
+        )
+
+    # We get the access token to make requests to the Biostrap API
+    def _authenticate(self, auth_creds):
+        self.client_id = auth_creds["client_id"]
+        self.client_secret = auth_creds["client_secret"]
+        redirect_uri = "https://127.0.0.1:8080"
+        token_url = "https://auth.biostrap.com/token"
+
+        # Generate the authorization URL and open it in the default web browser
+        params = {
+            "client_id": self.client_id,
+            "response_type": "code",
+            "redirect_uri": redirect_uri,
+        }
+        authorization_url = f"https://auth.biostrap.com/authorize?{urlencode(params)}"
+        webbrowser.open(authorization_url)
+
+        # Get the authorization response URL from the command line (the method Jack and I talked about didn't work)
+        authorization_response = input("Enter the full callback URL: ")
+
+        code = authorization_response.split("code=")[1].split("&")[0]
+
+        # Now we can request the access token
+        data = {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": redirect_uri,
+        }
+        auth = (self.client_id, self.client_secret)
+        response = requests.post(token_url, auth=auth, data=data)
+
+        # Get the access token from the response
+        token_json = response.json()
+        self.access_token = token_json["access_token"]