file3
unknown
python
a year ago
7.7 kB
3
Indexable
Never
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()