FitBit and Continuous Glucose Monitoring to detect heart rate & glucose abnormalities during sleep.
Authors: Vadim Astakhov ; Eva Khmelinskaya
Abstract
This blog demonstrates the use of wearable devices like FitBit with Google Cloud Platform to create a monitoring pipeline for people in some vulnerable groups. People with heart disease, diabetes and the elderly often need help with inexpensive monitoring devices at home to notify caregivers or family members in case some anomaly behavior has been detected.
This solution can be used to demonstrate how cloud technologies can be integrated with mobile devices and adapted for healthcare applications. (20 min demo can be found here)
I don’t speak for my employer. This is not official Google work. Any errors that remain are mine, of course.
Problem Statement and use cases
Recently Fitbit’s ECG app got FDA nod to track heart rhythm irregularities and with new FitBit ECG app and FitBit Charge 5 customers can assess their heart for atrial fibrillation (AFib) and share results with the doctor.
In this blog, we are exploring a few cases where a FitBit might be used to notify user about some abnormalities with heart signals.
Use Case 1 (Cardio): One is to leverage Wearables and Artificial Intelligence technology to detect signs of cardiac arrest while someone is asleep.
American Heart Association reported that nearly half a million people in America die every year when their hearts stop beating during a cardiac arrest and sudden cardiac death (SCD). SCD is unexpected death caused by loss of heart function (SCA — sudden cardiac arrest) is the largest cause of natural death in the United States, causing about 325,000 adult deaths each year.
Sudden cardiac death, occurs during the nocturnal period at an annual rate of ~36000 in the United States. As only 12% of all cardiovascular deaths occur at night, it would appear that sleep is a time of relative protection against cardiovascular death. However, that heightened susceptibility to sudden death coincides with periods of increased ischemia and arrhythmias, and concentrated during the 90 min of REM. Thus, the relative risk for sudden death during REM sleep may be as high as 1.2 times that of wakefulness.
REM sleep could initiate coronary plaque disruption, with death delayed until wakefulness. Clinical reports have provided evidence that REM sleep and dreams play an even greater role in precipitating myocardial infarction and sudden death in people already afflicted with coronary disease.
Use case 2 (Diabetes): another category of users which might be able to benefit from using FitBit during a sleep is people with diabetes. FitBit heart monitoring can be used in conjunction with constant Glucose Monitoring devices.
It was reported that dramatic drop of glucose level during sleep which might lead to glycemic coma often produce an increase in the heart rate. Thus listening to the heart while monitoring glucose level while sleeping could potentially be lifesaving.
Use Case 3 (Sleep Apnea): Clinical reports have provided evidence that REM sleep and dreams play a greater role in people with respiratory disorders ranging from snoring to central and obstructive sleep apnea. Researchers at University of Washington start building applications to utilize existing smartphones or smart speakers to listen to breathing. While they claim that chance of detecting agonic breathing is 97%, considering various sounds that can interfere, it seems that direct measuring of heart pulse might be a more robust way to detect heart abnormal activity. Also as we suggested in previous use cases, both approaches can be used in conjunction to increase sensitivity and reduce rate of false alarm.
In this blog, we demonstrate the use of FitBit and Google AI platform to measure heart pulse and make emergency calls to a caregiver or family member if abnormal activity is detected.
Solution
This diagram demonstrates the proposed solution (Github)
1. Authenticate your-self
You will have to go to the FitBit portal and create an account for your FitBit device.
Then, you will need to authenticate yourself on the Fitbit Developer Portal and create an “App” on it. It is a relatively simple process, through the public service at https://exain.com/fitbit, with the tutorial on YouTube on how to retrieve your “Client ID” .
Note: This public service saves the heart rate data in a database. It means that it will have your heart rate data. However, it does not ask for or collect any user information. So despite the fact that public service has heart rate data, it does not link it with an individual. For production deployment the entire code of the service is available on GitHub -https://github.com/technotablet/fitbit and can be deployed on cloud personal account.
2. Create a cloud function to pool Heart Rate data.
Cloud function will leverage Heart Rate API calls to pool heart rate data. (Code for the cloud function provided in Appendix A, B). And use Google Cloud Scheduler to schedule this cloud function invocation every 1 min.
Note: You can visualize your Heart Rate Chart at https://exain.com/fitbit or deploy your own environment using the code from github.
3. Constant Glucose Monitoring Use Case — Augment CGM with FitBit
Use of FitBit with Constant Glucose Monitoring devices during sleep for people with diabetes might be another important use case. Parents of diabetic kids often wake up several times during the night to check the child’s glucose level. Adding an additional alarm system to cross validate CGM devices might provide additional levels of safety in case of device malfunctioning or some anomaly drop in glucose level.
If glucose level drops significantly it often leads to an increase in the heart rate. Detecting anomalous heart increase during sleep might be used as additional confirmation for Constant Glucose Monitoring device.
Detecting anomaly behavior in both time series (glucose level and heart rate) can potentially help with sensitivity and specificity for the health alarm system.
In this solution we use artificial Dexcom Web and Rest API to create a simple cloud function, but the same approach can be used for other CGM devices.
After registering and creating your app, you can use sandbox data for testing and leverage a few examples provided by Dexcom to create a cloud function to pool glucose data time series and plot them for analysis.
from pydexcom import Dexcom
dexcom = Dexcom(“username”, “password”) # add ous=True if outside of US
bg = dexcom.get_current_glucose_reading()
bg.value
bg.mmol_l
bg.trend
bg.trend_description
bg.trend_arrow
Bg.time
4. Leverage DataFlow to move data to Bigquery
Google provides a streaming template to move data from PubSub to Bigquery. This documentation provides step by step instructions to configure the template and start moving Heart Rate data from PubSub to BigQuery.
If we want to make our solution more secure and prevent accidental leaking of sensitive data then we can leverage more advanced solutions that will have to leverage services like DLP to enhance security (code in Github).
5. Leverage BigQuery ARIMA Anomaly Detection
Data propagated to Bigquery will be used to create a model for anomaly detection by leveraging ARIMA_PLUS anomaly detection
Example (Create Model):
CREATE OR REPLACE MODEL `my-project.cardiology.fitbit_sleep06_arima_plus_seond_0` OPTIONS(MODEL_TYPE=’ARIMA_PLUS’, TIME_SERIES_TIMESTAMP_COL=’tm’, TIME_SERIES_DATA_COL=’Heart_Rate’) AS SELECT timestamp_seconds(_0*60) tm, Heart_Rate FROM `covid-19–271622.cardiology.cardio_test1` where Date___Time < ‘2021–06–25 06:00:00 UTC’ order by _0
Example (Detect Anomalies in historical data)
SELECT timestamp_seconds(_0*60) timestmp, Heart_Rate FROM `my-project.cardiology.cardio_test1_anomaly` where timestamp_seconds(_0*60) >= ‘2021–08–23 03:23:00 UTC’ and timestamp_seconds(_0*60) <= ‘2021–08–23 07:15:00 UTC’ order by timestmp
=======================================
Proposed solution with cloud function allows quick incorporation of various anomaly detection rules. For example, If low (<70 mg/dL) glucose level detected and Heart rate high (>250–300) then an urgent call triggered for a caregiver.
6. Create a cloud function to detect anomalies and initiate alarm calls.
Last component of the solution is another cloud function (example of Python in Appendix C) which will be triggered every minute to inspect heart rate for the last few minutes and detect if there are any anomalies.
This function will make a call to Bigquery to run statement
SELECT * FROM ML.DETECT_ANOMALIES(MODEL `mydataset.my_arima_plus_model`, STRUCT(0.8 AS anomaly_prob_threshold))
If anomaly detected then the cloud function will make an emergent call to a pre-configured number using one of the partner solutions (Twillo or Plivo) as described in the next section.
Proposed solution with cloud function allows quick incorporation of various anomaly detection rules. For example, If low (<70 mg/dL) glucose level detected and Heart rate high (>250–300) then urgent call triggered for a caregiver.
7. Make a phone call from Google Cloud function (Python)
The easiest way to orchestrate outbound phone calls is to use one of two python libraries for partners like Twillo or Plivo.
Plivo lets you program outbound calls in a few lines of code. You will have to create an account and find your Auth ID and Auth Token in the Account section at the top of your Plivo Console. Then you can turn to PHLO, Plivo’s visual workflow design studio, to set up the workflow for an outbound call.
Note: Appendix E provides a python code for your cloud function to invoke Plivo workflow and make an outbound call.
Twillo as a Google partner and can be used as a part of comprehensive Dialog Flow. You will have to create an account to Copy the Account SID and Auth Token from the Twilio Console and paste them into your application’s code:
The Twilio trial account allows you to dial and receive phone calls to your own validated phone number. Thus I can set it up to get an emergency call from my family member to which I am providing caregiving.
To handle calls from any phone number then you need to upgrade your account (hit the upgrade button on the top navigation bar).
Appendix D: provides a python code for your cloud function to invoke Twillo workflow and make an outbound call.
8. Summary and Future Development
This blog demonstrates the use of Google Cloud Platform with wearable devices like FitBit to create a monitoring pipeline for people in some vulnerable groups. People with heart disease, diabetes and the elderly often need help with inexpensive monitoring devices at home to notify caregivers or family members in case some anomaly behavior has been detected.
This solution can be used to demonstrate how cloud technologies can be integrated with mobile devices and adapted for healthcare applications.
We are constantly adding new functionality to this solution and planning to enhance it with Sleep Sensing from Nest Hub. Sleep Sensing from Nest Hub provides an ability to track person’s breathing pattern and environmental sounds which can help to increase specificity and reduce false alarm rate
Appendix A: Cloud Function to pool FitBit Data
(Github)
import requests
import pandas as pd
import datetime
import numpy as np
import seaborn as sns
import base64
import json
import os
from google.cloud import pubsub_v1
def fitbit_call(request):
# Instantiates a Pub/Sub client
publisher = pubsub_v1.PublisherClient()
PROJECT_ID = “my-project”
access_token = “ey..vpY”
day_range_length = 5
# start_data provide a window for historical activity of the heart
# end_data is current data and should be automatically set by scheduler
start_date = “2021–08–19”
end_date = “2021–08–23”
start = datetime.datetime.strptime(start_date, “%Y-%m-%d”)
end = datetime.datetime.strptime(end_date, “%Y-%m-%d”)
date_array = (start + datetime.timedelta(days=x) for x in range(0, (end-start).days))
day_list = []
for date_object in date_array:
day_list.append(date_object.strftime(“%Y-%m-%d”))
print(“day range : “,day_list)
df_all = pd.DataFrame()
header = {‘Authorization’: ‘Bearer {}’.format(access_token)}
for single_day in day_list:
response = requests.get(“https://api.fitbit.com/1/user/-/activities/heart/date/"+ single_day +”/1d/1min/time/00:00/23:59.json”, headers=header).json()
try:
df = pd.DataFrame(response[‘activities-heart-intraday’][‘dataset’])
date = pd.Timestamp(single_day).strftime(‘%Y-%m-%d’)
df = df.set_index(pd.to_datetime(date + ‘ ‘ + df[‘time’].astype(str)))
#print(df)
df_all = df_all.append(df, sort=True)
except KeyError as e:
print(“No data for the given date”, date)
del df_all[‘time’]
df_all[df_all.size -1:df_all.size]
print(df_all[df_all.size -1:df_all.size])
summary_df = (df_all[‘value’].resample(‘5s’).mean())
summary_df[summary_df.size -1:summary_df.size]
print(summary_df[summary_df.size -1:summary_df.size])
topic_name = “cgm-fitbit”
message = df_all[df_all.size -1:df_all.size].to_string()
topic_path = publisher.topic_path(PROJECT_ID, topic_name)
message_json = json.dumps({
‘data’: {‘message’: message},
})
message_bytes = message_json.encode(‘utf-8’)
# Publishes a message
publish_future = publisher.publish(topic_path, data=message_bytes)
publish_future.result() # Verify the publish succeeded
Appendix B: requirements.txt
# Function dependencies, for example:
# package>=version
pandas
datetime
numpy
seaborn
google-cloud-pubsub
Appendix C: Cloud Function to call BigQuery
import time
from google.protobuf.timestamp_pb2 import Timestamp
from google.cloud import bigquery_datatransfer_v1
def runQuery (parent, requested_run_time):
client = bigquery_datatransfer_v1.DataTransferServiceClient()
projectid = ‘[enter your projectId here]’ # Enter your projectID here
transferid = ‘[enter your transferId here]’ # Enter your transferId here
parent = client.project_transfer_config_path(projectid, transferid)
start_time = bigquery_datatransfer_v1.types.Timestamp(seconds=int(time.time() + 10))
response = client.start_manual_transfer_runs(parent, requested_run_time=start_time)
print(response)
# do not forget to put google-cloud-bigquery-datatransfer==1 in the requirements.txt
Appendix D: Twillo outbound phone call
Example 1:
# Download the helper library from https://www.twilio.com/docs/python/install
import os
from twilio.rest import Client
# Find your Account SID and Auth Token at twilio.com/console
# and set the environment variables. See http://twil.io/secure
account_sid = os.environ[‘TWILIO_ACCOUNT_SID’]
auth_token = os.environ[‘TWILIO_AUTH_TOKEN’]
client = Client(account_sid, auth_token)
call = client.calls.create(
twiml=’<Response><Say>Ahoy, World!</Say></Response>’,
to=’+14155551212',
from_=’+15017122661'
)
print(call.sid)
Example 2:
from twilio.rest import TwilioRestClient
# Twilio phone number goes here. Grab one at https://twilio.com/try-twilio
# and use the E.164 format, for example: “+12025551234”
TWILIO_PHONE_NUMBER = “”
# list of one or more phone numbers to dial, in “+19732644210” format
DIAL_NUMBERS = [“”,]
# URL location of TwiML instructions for how to handle the phone call
TWIML_INSTRUCTIONS_URL = \
“http://static.fullstackpython.com/phone-calls-python.xml"
# replace the placeholder values with your Account SID and Auth Token
# found on the Twilio Console: https://www.twilio.com/console
client = TwilioRestClient(“ACxxxxxxxxxx”, “yyyyyyyyyy”)
def dial_numbers(numbers_list):
“””Dials one or more phone numbers from a Twilio phone number.”””
for number in numbers_list:
print(“Dialing “ + number)
# set the method to “GET” from default POST because Amazon S3 only
# serves GET requests on files. Typically POST would be used for apps
client.calls.create(to=number, from_=TWILIO_PHONE_NUMBER,
url=TWIML_INSTRUCTIONS_URL, method=”GET”)
if __name__ == “__main__”:
dial_numbers(DIAL_NUMBERS)
Appendix E. Pvilo outbound call
import plivo
auth_id = ‘Your AUTH ID’
auth_token = ‘Your AUTH Token’
phlo_id = ‘Your PHLO ID’ # https://console.plivo.com/phlo/list/
phlo_client = plivo.phlo.RestClient(auth_id=auth_id, auth_token=auth_token)
phlo = phlo_client.phlo.get(phlo_id)
response = phlo.run()
print str(response)
