file3
unknown
python
3 years ago
7.7 kB
9
Indexable
from datetime import date, timedelta # TODO(alaurens): just you import datetime
import hashlib
from api.fitbit_client import FitbitApiClient # TODO(alaurens): just import fitbit_client
import pymongo
class FitbitMongoClient:
"""A class to import sleep and heart rate data from Fitbit API into a MongoDB database."""
def __init__(
# TODO(alaurens): Here you should be adding types
self,
connection_string: str,
database: str,
collection, # : ??? what type is this,
fitbit_client_id,
fitbit_client_secret,
):
"""Initialize the FitbitMongoClient with the given parameters.
# TODO(alaurens): Here you should put the arguments like this
Args:
connection_string: MongoDB connection string
database: MongoDB database name
collection: MongoDB collection name
fitbit_client_id: Fitbit API client ID
fitbit_client_secret: Fitbit API client secret
"""
try:
# Connect to MongoDB
self.mongo_client = pymongo.MongoClient(connection_string)
self.db = self.mongo_client[database]
self.collection = self.db[collection]
# Initialize Fitbit API client
self.fitbit_api_client = FitbitApiClient(
fitbit_client_id, fitbit_client_secret
)
except Exception as e:
# If any error occurs, set objects to None and raise an exception
# TODO(alaurens): Why do you need to raise to None here?
self.mongo_client = None
self.db = None
self.collection = None
self.fitbit_api_client = None
raise Exception(e)
def import_sleep_data_for_daterange(self, startTime=None, endTime=None):
# TODO(alaurens): This is a public api you should have both typing and
# also a docstring
sleep_data = self.fitbit_api_client.get_sleep_data_for_data_range(
startTime, endTime
)
user_id = hashlib.sha256(
self.fitbit_api_client.USER_ID.encode('utf-8')
).hexdigest()
# TODO(alaurens): These comments don't always bring a lot of value # Iterate through sleep data
# TODO(alaurens): Where do you get this schema? It would be good to either
# build a dataclass to use it or some proto for example?
for item in sleep_data['sleep']:
# Create document for each data record
document = {
'id': user_id,
'type': 'sleep',
'dateOfSleep': item['dateOfSleep'],
'metrics': {
'startTime': item['startTime'],
'endTime': item['endTime'],
'duration': item['duration'],
'efficiency': item['efficiency'],
'minutesAsleep': item['minutesAsleep'],
'minutesAwake': item['minutesAwake'],
'minutesToFallAsleep': item['minutesToFallAsleep'],
'timeInBed': item['timeInBed'],
},
'summary': item['levels']['summary'],
'data': item['levels']['data'],
}
# Insert document into MongoDB
self.collection.insert_one(document)
return
# TODO(alaurens): For any date and time data use the datetime library
# TODO(alaurens): For the detail level you should use an enum here and not a str, prone too error and hard to read
def import_heart_data_for_daterange(
self,
startTime: Optional[str] = None,
endTime: Optional[str] = None,
detail_level: DetailLevel = fitbit_client.DetailLevel.ONE_MINUTE
):
"""The function imports heart rate data from the Fitbit API for a given date range and saves it to a MongoDB collection.
# TODO(alaurens): You don't need to add the type here if you add it to the function signature
# TODO(alaurens): Make sure you are consistent in the way you define arguments and that you indent properly
Args:
startTime (str): Start date for data import in yyyy-MM-dd format. Defaults
to None.
endTime (str): End date for data import in yyyy-MM-dd format. Defaults to
None.
detail_level (str): Detail level for heart rate data. Must be one of "1sec",
"1min", or "15min". Defaults to "1min".
Returns:
bool: Returns True if data was successfully imported and saved to the
collection.
"""
# Retrieve heart rate data from Fitbit API for specified date range and detail level
# TODO(alaurens): If the fitbit API expect a string for all of these you can easily make some convertions
multiple_heart_data = (
self.fitbit_api_client.get_heart_rate_data_for_data_range(
startTime, endTime, detail_level.value
)
)
# Hash user ID to maintain anonymity
user_id = hashlib.sha256(
self.fitbit_api_client.USER_ID.encode('utf-8')
).hexdigest()
# Iterate through each heart rate data point
for heart_data in multiple_heart_data:
# Extract relevant data and create a document to be inserted into the database
# TODO(alaurens): Don't have magic numbers in your code, define constants at the top
document = {
'id': user_id,
'type': 'heart',
'date': heart_data['activities-heart'][0]['dateTime'], # what is 0 here?
'heartRateZones': heart_data['activities-heart'][0]['value'][
'heartRateZones'
],
'heartIntraday': heart_data['activities-heart-intraday']['dataset'],
}
# TODO(alaurens): This comment again doesn't really add much.
# Check if resting heart rate data is present and add to document if available
if 'restingHeartRate' in heart_data['activities-heart'][0]['value']:
document['restingHeartrate'] = heart_data['activities-heart'][0][
'value'
]['restingHeartRate']
# Insert document into database collection
self.collection.insert_one(document)
# Return True to indicate successful data import
return True
def import_hrv_data_for_daterange(self, startTime=None, endTime=None):
"""The function imports Heart Rate Variability (HRV) from the Fitbit API for a given date range and saves it to a MongoDB collection.
Args:
startTime (str): Start date for data import in yyyy-MM-dd format. Defaults
to None.
endTime (str): End date for data import in yyyy-MM-dd format. Defaults to
None.
Returns:
bool: Returns True if data was successfully imported and saved to the
collection.
"""
# Retrieve heart rate data from Fitbit API for specified date range and detail level
multiple_hrv_data = self.fitbit_api_client.get_hrv_data_for_data_range(
startTime, endTime
)
# Hash user ID to maintain anonymity
user_id = hashlib.sha256(
self.fitbit_api_client.USER_ID.encode('utf-8')
).hexdigest()
# Iterate through each heart rate data point
for hrv in multiple_hrv_data['hrv']:
# Extract relevant data and create a document to be inserted into the database
document = {
'id': user_id,
'type': 'hrv',
'date': hrv['dateTime'],
'dailyRmssd': hrv['value']['dailyRmssd'],
'deepRmssd': hrv['value']['deepRmssd'],
}
# Insert document into database collection
self.collection.insert_one(document)
# Return True to indicate successful data import
return True
# EXAMPLE CODE
# client = FitbitMongoClient(
# connection_string = "mongodb://localhost:27017/",
# database="local",
# collection="fitbit",
# fitbit_client_id= "23QRJ6",
# fitbit_client_secret= "abb49f0cdfcfd2605f02fcae11dda3b4",
# )
# startTime = date(year = 2023, month = 4, day = 22)
# endTime = date.today()
# client.import_sleep_data_for_daterange()
# client.import_heart_data_for_daterange()
# client.import_hrv_data_for_daterange()
Editor is loading...