Darian Johnson
Published © GPL3+

Amazon Machine Learning to Recommend Best Workout Time

Use your Fitbit history to provide a recommendation on the best time of day to workout.

IntermediateFull instructions provided20 hours2,239
Amazon Machine Learning to Recommend Best Workout Time

Things used in this project

Hardware components

Fitbit
×1
Withings Scale
Note - I obtain weight through the Fitbit platorm/apis. I integrated (via api) my scale to log data to my Fitbit account.
×1

Software apps and online services

AWS Lambda
Amazon Web Services AWS Lambda
AWS ML
Amazon Web Services AWS ML
AWS SNS
Amazon Web Services AWS SNS
AWS DynamoDB
Amazon Web Services AWS DynamoDB
AWS S3
Amazon Web Services AWS S3

Story

Read more

Schematics

Architecture and VUI Diagrams

The following slide shows the AWS architecture used to build the app. It also contains a VUI diagram

Code

Get Best Workout Time - Interface with Alexa

Python
This is the lambda function that communicates with the Alexa skill
"""
Darian Johnson
http://darianbjohnson.com
This application uses historical data from your Fitbit to recommend the best time to workout.
1) a User accessed the application and asks what for best time to work out (for either today or tomorrow) 
2) The application pulls the following details from the users fitbit account - Weight, Sleep, Calories Burned the Previous Day
3) This information is passed to an AWS Machine Learning Model to determine at what time of day you are most likely to have the highest number of 'very active minutes'

Notes:
1) I assumed that very active minutes is the best way to determine the effectiveness of a workout.
2) The user must link their fitbit account in order to get a recommendation
3) The application uses a machine learning model - based on the users previous work-out history - to create a recommendation
4) the model is created with two additional lambda functions - one function created the datasource and is kicked off by an SNS call in this function
   The second lamdba function runs every 30 minutes to take follow-up steps to create the model
5) The creation of the model takes between 30-90 mins. The user is informed of progress if s/he attempts to access the application while the model is running
6) The user is also notified if the model fails to be created

You can find out more about this application at http://darianbjohnson.com/workout-recommendation
"""

from __future__ import print_function

import json
import urllib
import base64
import urllib2
import boto3
import time
import datetime

from datetime import date
from datetime import timedelta

#set up variables for AWS services calls
ml = boto3.client('machinelearning')
lamb = boto3.client('lambda')
dynamodb = boto3.client('dynamodb')
sns = boto3.client('sns')

def lambda_handler(event, context):
    """ Route the incoming request based on type (LaunchRequest, IntentRequest,
    etc.) The JSON body of the request is provided in the event parameter.
    """
    print("event.session.application.applicationId=" +
           event['session']['application']['applicationId'])

    """
    Uncomment this if statement and populate with your skill's application ID to
    prevent someone else from configuring a skill that sends requests to this
    function.
    """
    # if (event['session']['application']['applicationId'] !=
    #         "amzn1.echo-sdk-ams.app.49ae8765-f94c-4bca-8dc3-1894a93b4f9c"):
    #     raise ValueError("Invalid Application ID")

    if event['session']['new']:
        on_session_started({'requestId': event['request']['requestId']},
                           event['session'])
    
    #this checks to see if the user has lined their fitbit account   
    if 'accessToken' in event['session']['user']:
        print("exists")
    else:
        print("does not exist")
        return on_NoLink()
    
    #routes based on the request
    if event['request']['type'] == "LaunchRequest":
        return on_launch(event['request'], event['session'])
    elif event['request']['type'] == "IntentRequest":
        return on_intent(event['request'], event['session'])
    elif event['request']['type'] == "SessionEndedRequest":
        return on_session_ended(event['request'], event['session'])
        

def on_session_started(session_started_request, session):
    """ Called when the session starts """
        
    print("on_session_started requestId=" + session_started_request['requestId']
          + ", sessionId=" + session['sessionId'])


def on_launch(launch_request, session):
    """ Called when the user launches the skill without specifying what they
    want
    """

    print("on_launch requestId=" + launch_request['requestId'] +
          ", sessionId=" + session['sessionId'])
    # Dispatch to your skill's launch
    return get_welcome_response()

#routes based on the intent
def on_intent(intent_request, session):
    """ Called when the user specifies an intent for this skill """

    print("on_intent requestId=" + intent_request['requestId'] +
          ", sessionId=" + session['sessionId'])

    intent = intent_request['intent']
    intent_name = intent_request['intent']['name']

    # Dispatch to your skill's intent handlers
    if intent_name == "WhatTimeShouldIWorkOut":
        return getWorkoutTime(intent, session)
    elif intent_name == "WhatTimeShouldIWorkOut_noDay":
        return getWorkoutDay()
    elif intent_name == "AMAZON.HelpIntent":
        return get_welcome_response()
    elif intent_name == "AMAZON.CancelIntent" or intent_name == "AMAZON.StopIntent" or intent_name == "Finished":
        return handle_session_end_request()
    else:
        return misunderstood()

#routes based on the intent
def on_session_ended(session_ended_request, session):
    """ Called when the user ends the session.
    Is not called when the skill returns should_end_session=true
    """
    print("on_session_ended requestId=" + session_ended_request['requestId'] +
          ", sessionId=" + session['sessionId'])
    # add cleanup logic here
    #delete_endpoint(sMLModelId) - we will not need to delete the endpoint, as we have decided to leave the up all the time


# --------------- Functions that control the skill's behavior ------------------

#This is called if the app is started without a secific intent to get the workout recommendation
def get_welcome_response():
    """ If we wanted to initialize the session to have some attributes we could
    add those here
    """
    session_attributes = {}
    card_title = "Welcome"
    speech_output = "Welcome to the Work Out Recommendation App. " \
                    "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow."
    reprompt_text = "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow."
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))

#This is called if the skill is not linked to a Fitbit account
def on_NoLink():
    """ If we wanted to initialize the session to have some attributes we could
    add those here
    """
    session_attributes = {}
    card_title = "Link Accounts"
    speech_output = "You must have a Fitbit account to use this skill. Please use the Alexa app to link your Amazon account with your Fitbit Account"
    reprompt_text = "Use the Alexa app to link your Amazon account with your Fitbit Account."
    should_end_session = True
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))
 
#This is called if the Machine Learning Model needs to be created or is in the process of being created
def on_Running_Model(card_title, message):
    """ If we wanted to initialize the session to have some attributes we could
    add those here
    """
    session_attributes = {}
    card_title = card_title
    speech_output = message
    # If the user either does not reply to the welcome message or says something
    # that is not understood, they will be prompted again with this text.
    reprompt_text = message
    should_end_session = True
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))
        
#This is called if the request is not understood       
def misunderstood():
    """ If we wanted to initialize the session to have some attributes we could
    add those here
    """
    session_attributes = {}
    card_title = "I Do Not Understand" 
    speech_output = "I did not understand your request. " \
                    "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow."
    # If the user either does not reply to the welcome message or says something
    # that is not understood, they will be prompted again with this text.
    reprompt_text = "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow."
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))

#This is called to end the session 
def handle_session_end_request():
    card_title = "Session Ended"
    speech_output = "Thank you for trying the Work Out Recommendation App. " \
                    "Have a nice day! "
    # Setting this to true ends the session and exits the skill.
    should_end_session = True
    return build_response({}, build_speechlet_response(
        card_title, speech_output, None, should_end_session))


#This is the main function that retrieves values from the users Fitbit account and interacts with the machine learning model to recommend the best workout time 
def getWorkoutTime(intent, session):
    card_title = "Workout Recommendation"
    speech_output = "Based on your schedule, the best time to work out is " 
    reprompt_text = "no Value"
    current_date = time.strftime("%Y-%m-%d")

    session_attributes = {}
    should_end_session = False
    
    Fitbit_Access_Token= session['user']['accessToken']
    
    #This either sets or retrieves session variables
    if session.get('attributes', {}) and "fitbit_user_id" in session.get('attributes', {}):
        weight = session['attributes']['weight']
        fitbit_user_id = session['attributes']['fitbit_user_id']
        sMLModelId = session['attributes']['Model_Id']
        sMLModelType = session['attributes']['Model_Type']
        
        endpoint = create_endpoint(sMLModelId)
        
    else:
        #calls the Fitbit function to retrieve user id and current weight
        profile = get_profile(Fitbit_Access_Token)
        parsed_profile = json.loads(profile)
        weight = parsed_profile['weight']
        fitbit_user_id = parsed_profile['fitbit_user_id']
        
        #this block routes the user to either proceed or exit, based on the readiness of the machine learning model
        if check_if_model_exists(fitbit_user_id) == "Pending":
            return on_Running_Model("Running Model", "Your Fitbit historical data is being analyzed to provide the best workout recommendations. Please check back in an hour.")
        elif check_if_model_exists(fitbit_user_id) == "Failed":
            return on_Running_Model("Running Model", "For some reason, the system was not able to complete the analysis. Our engineers are looking into a solution.")
        elif check_if_model_exists(fitbit_user_id) == "Small":
            return on_Running_Model("Running Model", "You do not have enough Fitbt workout history to perform custom analysis. Please try again after you have used your Fitbit to log approximately 30 workouts.")
        elif check_if_model_exists(fitbit_user_id) == "Run Model":
            #if the model has not been created, then a message is sent via SNS to a second lambda function that creates the required machine learning model
            sns_message = json.dumps({'Fitbit_User_Id':fitbit_user_id,'Fitbit_Access_Token':Fitbit_Access_Token,'weight':weight })
            response = sns.publish(
                TopicArn='arn:aws:sns:us-east-1:039057814095:Start_ML_Model',
                Message=sns_message)
            return on_Running_Model("Running Model", "Your Fitbit historical data needs to be analyzed to provide the best workout recommendations. This process will take about an hour.") 
        
        model_attr = get_ml_model(fitbit_user_id)
        parsed_model_attr = json.loads(model_attr)
        sMLModelId = parsed_model_attr['Model_Id']
        sMLModelType = parsed_model_attr['Model_Type']
        
        session_attributes = {'weight':weight, 'fitbit_user_id': fitbit_user_id, 'Model_Id':sMLModelId, 'Model_Type':sMLModelType}

        endpoint = create_endpoint(sMLModelId)
    
    #this looks at the slot to determine if the request is for today or tomorrow 
    if 'Day' in intent['slots']:
        daytoschedule = intent['slots']['Day']['value']
        if daytoschedule == "today":
            today = date.today()
            DayofWeek = today.strftime("%A")
            Spoken_value = ['this morning', 'mid-day today', 'this afternoon', 'this evening']
        elif daytoschedule == "tomorrow":
            tomorrow = date.today() + timedelta(days=1)
            DayofWeek = tomorrow.strftime("%A")
            Spoken_value = ['tomorrow morning', 'mid-day tomorrow', 'tomorrow afternoon', 'tomorrow evening']
        
        #this block retrieves sleep and calories from the Fitbit. These values are passed to the machine learning model to retrieve a recommendation
        #I have left placeholders for calendar details. Right now, it is not integrated as the skill can only link to one 3rd party account
        #I will build a seperate authetication on my website to support adding calendar functionality
        Sleep_Mins = get_sleep(Fitbit_Access_Token, daytoschedule)
        Previous_Day_Active_Calorie_Output = get_calories(Fitbit_Access_Token, daytoschedule) 
        Meeting_Tentative_Mins = 0
        Meeting_Organizer_Mins = 0
        Meetng_Attendee_Mins = 0
        OutOfOffice_Mins = 0
        
        #this recommendation will look at 4 times of day to determine the most optimal time to workout
        timeofday = ['Morning', 'Mid-Day', 'Afternoon', 'Evening']

        #this block loops through the timeofday dictionary and saves the value with the highest prediction 
        OldPredictValue = -1000
        Best_Time_Of_Day = "any time of day"
        i=0
        for times in timeofday:
            print(DayofWeek + ":" + times + ":" + str(get_projectedvalue(sMLModelId, endpoint, DayofWeek, times, str(Sleep_Mins), str(Meeting_Tentative_Mins),\
            str(Meeting_Organizer_Mins), str(Meetng_Attendee_Mins), str(OutOfOffice_Mins), str(weight), str(Previous_Day_Active_Calorie_Output))))
            PredictValue = get_projectedvalue(sMLModelId, endpoint, DayofWeek, times, str(Sleep_Mins), str(Meeting_Tentative_Mins),\
            str(Meeting_Organizer_Mins), str(Meetng_Attendee_Mins), str(OutOfOffice_Mins), str(weight), str(Previous_Day_Active_Calorie_Output))
            if PredictValue > OldPredictValue:
                Best_Time_Of_Day = Spoken_value[i]
                OldPredictValue = PredictValue
                
            i = i+1
            
            
        speech_output = "Based on your previous activity, the best time for you to work out is " + Best_Time_Of_Day + "."
        reprompt_text = speech_output
        
        print("The best time of day to work out is " + Best_Time_Of_Day)
   
        
    else:
        speech_output = "I did not understand your request. " \
                    "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow."
        reprompt_text = "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow."
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))

#this is called if the slot - today or tommorrow - is not included in the request.
#as an enhancement, I could ask if the user assumes today, and if yes, then run the mdoel for today
def getWorkoutDay():
    """ If we wanted to initialize the session to have some attributes we could
    add those here
    """
    session_attributes = {}
    card_title = "What Day should I check?" 
    speech_output = "You did not tell me what day to check. " \
                    "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow"
    reprompt_text = "Please ask me, what is the best time to work out today or what is the best time to work out tomorrow."
    should_end_session = False
    return build_response(session_attributes, build_speechlet_response(
        card_title, speech_output, reprompt_text, should_end_session))


# --------------- Helpers that build all of the responses ----------------------


def build_speechlet_response(title, output, reprompt_text, should_end_session):
    return {
        'outputSpeech': {
            'type': 'PlainText',
            'text': output
        },
        'card': {
            'type': 'Simple',
            #'title': 'SessionSpeechlet - ' + title,
            #'content': 'SessionSpeechlet - ' + output
            'title': title,
            'content': output
        },
        'reprompt': {
            'outputSpeech': {
                'type': 'PlainText',
                'text': reprompt_text
            }
        },
        'shouldEndSession': should_end_session
    }


def build_response(session_attributes, speechlet_response):
    return {
        'version': '1.0',
        'sessionAttributes': session_attributes,
        'response': speechlet_response
    }


# --------------- Functions to access the users Fitbit account ----------------------    

#this returns the user id and the current weight
#user id is needed to determine the correct machine learning model
#weight is used as part of the prediction input
def get_profile(Fitbit_Access_Token):
    URL = 'https://api.fitbit.com/1/user/-/profile.json'
    req = urllib2.Request(URL)
    
    #Add the headers, first we base64 encode the client id and client secret with a : inbetween and create the authorisation header
    req.add_header('Authorization', 'Bearer ' + Fitbit_Access_Token)
    req.add_header('Content-Type', 'application/x-www-form-urlencoded')
    
    #Fire off the request
    
    try:
        response = urllib2.urlopen(req)
        FullResponse = response.read()
        #print ("Output >>> " + FullResponse)
        parsed_response = json.loads(FullResponse)
        weight = round((parsed_response['user']['weight'])*2.20462,1)
        user_id = parsed_response['user']['encodedId']
        return json.dumps({'weight':weight, 'fitbit_user_id':user_id})
    except urllib2.URLError as e:
        print (e.code)
        print (e.read())   
        
#this returns the number of minutes asleep
#if sleep is 0 or the request is for tomorrow, then we assume the user got 6 hours of sleep
def get_sleep(Fitbit_Access_Token, day):
    today = date.today()
    URL =  'https://api.fitbit.com/1/user/-/sleep/date/' + today.strftime("%Y-%m-%d") + '.json'
    req = urllib2.Request(URL)
    
    #Add the headers, first we base64 encode the client id and client secret with a : inbetween and create the authorisation header
    req.add_header('Authorization', 'Bearer ' + Fitbit_Access_Token)
    req.add_header('Content-Type', 'application/x-www-form-urlencoded')
    
    if day == "today":
        try:
            response = urllib2.urlopen(req)
            FullResponse = response.read()
            #print ("Output >>> " + FullResponse)
            parsed_response = json.loads(FullResponse)
            sleep = parsed_response['summary']['totalMinutesAsleep']
            if str(sleep) == str(0):
                sleep = 360 
            return sleep
        except urllib2.URLError as e:
            print (e.code)
            print (e.read()) 
            return 360

    elif day == "tomorrow":
            return 360

#this returns the number of active calories burned the previous day
def get_calories(Fitbit_Access_Token, day):
    today = date.today()
    yesterday = today - timedelta(days=1)

    if day == "today":
        URL =  'https://api.fitbit.com/1/user/-/activities/date/' + yesterday.strftime("%Y-%m-%d") + '.json'
    elif day == "tomorrow":
        URL =  'https://api.fitbit.com/1/user/-/activities/date/' + today.strftime("%Y-%m-%d") + '.json'
    
    req = urllib2.Request(URL)
    #Add the headers, first we base64 encode the client id and client secret with a : inbetween and create the authorisation header
    req.add_header('Authorization', 'Bearer ' + Fitbit_Access_Token)
    req.add_header('Content-Type', 'application/x-www-form-urlencoded')
    
    try:
        response = urllib2.urlopen(req)
        FullResponse = response.read()
        #print ("Output >>> " + FullResponse)
        parsed_response = json.loads(FullResponse)
        activities = parsed_response['summary']['activityCalories']
        return activities
    except urllib2.URLError as e:
        print (e.code)
        print (e.read()) 
        return 0

# --------------- Functions to access the machine learning model ---------------------- 

#creates the endpoint to access the machine model via API
def create_endpoint(sMLModelId):
    oCreateEndpoint = ml.create_realtime_endpoint(
        MLModelId=sMLModelId)
    return oCreateEndpoint['RealtimeEndpointInfo']['EndpointUrl']

#deletes the endpoint
def delete_endpoint(sMLModelId):
    oCreateEndpoint = ml.delete_realtime_endpoint(
        MLModelId=sMLModelId
        )

#passes in the values from the fitbit account and returns the predicted number of active minutes
#I have left in placeholder for calendar details, to integrate at a later point in time
def get_projectedvalue(sMLModelId, EndpointUrl, DayofWeek, TimeOfDay, Sleep_Mins, Meeting_Tentative_Mins, Meeting_Organizer_Mins, Meetng_Attendee_Mins, OutOfOffice_Mins, Weight, Previous_Day_Active_Calorie_Output):
    prediction = ml.predict(
        MLModelId=sMLModelId,
        Record={
            'DayofWeek': DayofWeek,
            'TimeofDay': TimeOfDay,
            'SleepMins': Sleep_Mins,
            'Weight': Weight,
            'PreviousDayCalories': Previous_Day_Active_Calorie_Output
        },
        PredictEndpoint=EndpointUrl)
    return prediction['Prediction']['predictedValue']
 
#queries a dynamodb table to determine the appropriate model for the active user  
def get_ml_model(fitbit_user_id):
    table = 'Workout_Recommendations'
    response = dynamodb.get_item(TableName = table,
        Key={
             'Fitbit_User_Id' : {
              "S" : fitbit_user_id
                }
            }
    )
    return json.dumps({'Model_Id':response['Item']['ML_Model_Id']['S'],'Model_Type': response['Item']['Model_Type']['S']})

#determines if the model exists and is ready for use    
def check_if_model_exists(fitbit_user_id):
    table = 'Workout_Recommendations'
    response = dynamodb.get_item(TableName = table,
        Key={
             'Fitbit_User_Id' : {
              "S" : fitbit_user_id
                }
            }
    )

    if 'Item' in response:
        if response['Item']['Model_Type']['S'] == 'PENDING':
            return "Pending"
        elif response['Item']['Model_Type']['S'] == 'FAILED':
            return "Failed"
        elif response['Item']['Model_Type']['S'] == 'SMALL':
            return "Small"
        else:
            return "Continue"
    else:
        return "Run Model"
    

Create ML Model Workout App

Python
This lambda function is triggered (via SNS, from another lambda function) when a user needs to have a machine learning model created
"""
Darian Johnson
http://darianbjohnson.com

This lambda funtion is part of the Best Time of Day to Workout Recommendation app

This script is fired when a user attepts to ask the app for a recommendation without having the requisite machine learning model created. 
1) Fitbit variables/auth token recieved via SNS
2) The application pulls up to 200 activity records from Fitbit
3) The data is saved to S3 and then used to create a Machine Learning Datasurce

A second function runs every 30 minutes to validate that the datasource was successfully created and to create an ML model

You can find out more about this application at http://darianbjohnson.com/workout-recommendation
"""

from __future__ import print_function

import json
import urllib
import base64
import urllib2
import boto3
import time
import datetime

import uuid

from datetime import date
from datetime import timedelta
from datetime import datetime

print('Loading function')

#setup AWS Services
ml = boto3.client('machinelearning')
dynamodb = boto3.client('dynamodb')
s3 = boto3.client('s3')


def lambda_handler(event, context):

    #get required variables from SNS message
    Fitbit_Variables = json.loads(event['Records'][0]['Sns']['Message'])
    Fitbit_Access_Token= Fitbit_Variables['Fitbit_Access_Token']
    Fitbit_User_Id= Fitbit_Variables['Fitbit_User_Id']
    Default_Weight = Fitbit_Variables['weight']
   
    print(Fitbit_Access_Token)
    
    #Check to ensure that the system is not attempting to create a new model when an existing model exists
    response = dynamodb.get_item(TableName = 'Workout_Recommendations',
        Key={
             'Fitbit_User_Id' : {
              "S" : Fitbit_User_Id
                }
            }
    )

    if 'Item' in response:
       return 'Complete - already in system'
 
    #Create a random ID for the DataSource
    DS_ID = Fitbit_User_Id + "-" + str(uuid.uuid4())
    
    #this block of code calls the function that retrieves Fitbit data. The fitbit api only returns 20 activities at a time, so pagination logic is included
    #we call the API until no more data is available, or 200 records are obtained
    #We get 1 years worth of Calorie Data and Sleep Data up front; we query this data once and re-use it to map to activity data
    
    calorie_array = get_caloriearray(Fitbit_Access_Token)
    sleep_array = get_sleeparray(Fitbit_Access_Token)
    count = 1
    data = []
    datasetcount = 0
    offset= 0
    while datasetcount < 200: 
        response = get_activities(Fitbit_Access_Token, data,offset, Default_Weight,calorie_array,sleep_array)
        data = response['dataset']
        datasetcount = len(response['dataset'])
        offset = response['offset']
        Default_Weight = response['Default_Weight']
        count = count + 1
        if offset == 0:
            break
    
    #Add this logic later to account for a small data size. For right now, we handle this logic with a "dataset fail" message in the program that evaluated the dataset
    #if datasetcount < 25:
    #    User_Details = {
    #        'Fitbit_User_Id': {'S':Fitbit_User_Id},
    #        'Datasource_Id': {'S':DS_ID},
    #        'Model_Type': {'S':'SMALL'}
    #    }
    #    response = dynamodb.put_item(TableName = 'Workout_Recommendations', Item = User_Details)
    #    return "Complete - Small"
    
    #The Body variable stores that data needed for the dataset
    body = 'ActivityName,DayofWeek,TimeofDay,Weight,PreviousDayCalories,VeryActiveMin,SleepMins\n'
    for i in data:
        body = body + i['Activity_Name'] + ',' + i['DayofWeek'] + ',' + i['TimeofDay'] + ',' + str(i['Weight']) + ',' + str(i['PreviousDayCalories']) + ',' + str(i['VeryActiveMin']) +  ',' + str(i['Sleep']) +'\n'
    
    #The Body data is stored as a CSV file in an S3 bucket
    bucket = 'darianbjohnson'
    key = "LearningData/" + Fitbit_User_Id + ".csv"
    
    response = s3.put_object(
        ACL='public-read',
        Body=body,
        Bucket=bucket,
        Key=key,
        ContentType = "text/csv"
        )
    
    #The S3 file is used to create a machine learning data source. Two data sources are created - one for evaliation, and one for training
    
    dataschema = json.dumps({'excludedAttributeNames': [], 'version': '1.0', 'dataFormat': 'CSV', 'dataFileContainsHeader': 'true', 'attributes': [{'attributeName': 'ActivityName', 'attributeType': 'CATEGORICAL'}, {'attributeName': 'DayofWeek', 'attributeType': 'CATEGORICAL'}, {'attributeName': 'TimeofDay', 'attributeType': 'CATEGORICAL'}, {'attributeName': 'Weight', 'attributeType': 'NUMERIC'}, {'attributeName': 'PreviousDayCalories', 'attributeType': 'NUMERIC'}, {'attributeName': 'VeryActiveMin', 'attributeType': 'NUMERIC'}, {'attributeName': 'SleepMins', 'attributeType': 'NUMERIC'}], 'targetAttributeName': 'VeryActiveMin'})
    datarearrangement = json.dumps({  "splitting": {    "percentBegin": 0,    "percentEnd": 80,    "strategy": "sequential",    "complement": 'false',    "strategyParams": {}  }})
    evaldatarearrangement = json.dumps({  "splitting": {    "percentBegin": 80,    "percentEnd": 100,    "strategy": "sequential",    "complement": 'false',    "strategyParams": {}  }})
    
    #training datasource
    response = ml.create_data_source_from_s3(
        DataSourceId= DS_ID,
        DataSourceName='DS-' + Fitbit_User_Id,
        DataSpec={ 
            'DataLocationS3': 's3://' + bucket + '/' + key,
            'DataSchema': dataschema,
            'DataRearrangement': datarearrangement
        } ,
        ComputeStatistics=True
    )
    
    #evaluation datasource - note: at this point in time, I do not use the eval datasource to evaulate the model. I plan to automatme this at a later point in time
    response = ml.create_data_source_from_s3(
        DataSourceId= 'EVAL-' + DS_ID,
        DataSourceName='EVAL-DS-' + Fitbit_User_Id,
        DataSpec={ 
            'DataLocationS3': 's3://' + bucket + '/' + key,
            'DataSchema': dataschema,
            'DataRearrangement': evaldatarearrangement
        } ,
        ComputeStatistics=True
    )
    
    print("Data Source in process of being built")
    
    #the DynamoDB table is updated with the datasource ID
    User_Details = {
        'Fitbit_User_Id': {'S':Fitbit_User_Id},
        'Datasource_Id': {'S':DS_ID},
        'Model_Type': {'S':'PENDING'}
    }
    response = dynamodb.put_item(TableName = 'Workout_Recommendations', Item = User_Details)
    
    return "Complete"
    #raise Exception('Something went wrong')
    
#----------------------- Functions to obtain and organize the Fitbit data ---------------------------

#This function retieves 20 activity records at a time
def get_activities(Fitbit_Access_Token, data, offset, Default_Weight, calorie_array,sleep_array):
    today = date.today()
    today.strftime("%Y-%m-%d")
    yesterday = today - timedelta(days=1)
    URL = 'https://api.fitbit.com/1/user/-/activities/list.json?beforeDate=' + today.strftime("%Y-%m-%d") + '&sort=desc&limit=20&offset=' + str(offset)
    req = urllib2.Request(URL)
    
    #Add the headers, first we base64 encode the client id and client secret with a : inbetween and create the authorisation header
    req.add_header('Authorization', 'Bearer ' + Fitbit_Access_Token)
    req.add_header('Content-Type', 'application/x-www-form-urlencoded')
    
    #Fire off the request
    
    try:
        response = urllib2.urlopen(req)
        FullResponse = response.read()
        #print ("Output >>> " + FullResponse)
        parsed_response = json.loads(FullResponse)
        populate_weight_array = True
        
        
        for i in parsed_response['activities']:
            Activity_Name = i['activityName']
            Very_Active_Minutes = i['activityLevel'][3]['minutes']
            Start_Time = datetime.strptime(i['startTime'][:10], '%Y-%m-%d')
            Cal_Search_Date = Start_Time - timedelta(days=1)
            StartHour =i['startTime'][11:13]
            
            #we need to get the weight values; we can only get up to 1mo of data; so we have to call this multiple time
            if populate_weight_array == True:
                populate_weight_array = False
                weight_array = get_weightarray(Fitbit_Access_Token,Start_Time)
            
            #we exclude "Walk" activities; all other activities are included
            #We call functions to get the specific values for Calories, Sleep and Weight
            if Activity_Name != 'Walk':
                PrevDayCalories = get_prevdaycalories(calorie_array, Cal_Search_Date.strftime("%Y-%m-%d"))
                Sleep = get_sleep(sleep_array, Start_Time.strftime("%Y-%m-%d"))
                Weight = get_weight(Default_Weight, weight_array, Start_Time.strftime("%Y-%m-%d"))
                Default_Weight = Weight
                d = {
                  'Activity_Name': Activity_Name,
                  'Date': Start_Time.strftime("%Y-%m-%d"),
                  'DayofWeek': Start_Time.strftime("%A"),
                  'TimeofDay':get_timeofday(int(StartHour)),
                  'Weight': Weight,
                  'Sleep': Sleep,
                  'PreviousDayCalories': PrevDayCalories,
                  'VeryActiveMin':Very_Active_Minutes 
                }
                data.append(d)
        
        #If more records can be obained, then there will be a pagination value; we use this to determine the record offset.
        if parsed_response['pagination']['next'] == "":
            offset=0
        else:
            offset=offset+20
        
        return({'dataset':data,'offset':offset,'Default_Weight':Default_Weight})
        
    except urllib2.URLError as e:
        print (e.code)
        print (e.read())   
        
#This function returns the timeofday bases on the activity start time      
def get_timeofday(starthour):
    if starthour < 11:
        return "Morning"
    elif starthour < 14:
        return "Mid-Day"
    elif starthour < 18:
        return "Afternoon"
    else:
        return "Evening"

#This returns the weight array containing 1 month of data     
def get_weightarray(Fitbit_Access_Token, startdate):
    URL = 'https://api.fitbit.com/1/user/-/body/log/weight/date/' + startdate.strftime("%Y-%m-%d") + '/1m.json'
    #URL = 'https://api.fitbit.com/1/user/-/activities/list.json?beforeDate=2016-01-28&sort=desc&limit=20&offset=0'
    req = urllib2.Request(URL)
    
    #Add the headers, first we base64 encode the client id and client secret with a : inbetween and create the authorisation header
    req.add_header('Authorization', 'Bearer ' + Fitbit_Access_Token)
    req.add_header('Content-Type', 'application/x-www-form-urlencoded')
    
    #Fire off the request
    
    try:
        response = urllib2.urlopen(req)
        FullResponse = response.read()
        #print ("Output >>> " + FullResponse)
        parsed_response = json.loads(FullResponse)
       
        return(parsed_response)
        
    except urllib2.URLError as e:
        print (e.code)
        print (e.read())   
    
#This returns the weight for a specific day; If there was not a weight recorded, we use the last known weight value 
def get_weight(default_weight, weight_array, start_date):
    for i in weight_array['weight']:
        if str(i['date']) == str(start_date):
            return round(i['weight']*2.20462,1)
            break

    return(default_weight)

#This returns the sleep array containing 1 year of data 
def get_sleeparray(Fitbit_Access_Token):
    URL = 'https://api.fitbit.com/1/user/-/sleep/minutesAsleep/date/today/1y.json'
    req = urllib2.Request(URL)
    
    #Add the headers, first we base64 encode the client id and client secret with a : inbetween and create the authorisation header
    req.add_header('Authorization', 'Bearer ' + Fitbit_Access_Token)
    req.add_header('Content-Type', 'application/x-www-form-urlencoded')
    
    #Fire off the request
    
    try:
        response = urllib2.urlopen(req)
        FullResponse = response.read()
        #print ("Output >>> " + FullResponse)
        parsed_response = json.loads(FullResponse)
       
        return(parsed_response)
        
    except urllib2.URLError as e:
        print (e.code)
        print (e.read())           

#This returns the sleep minutes for a specific day; If there was not a sleep count recorded, we use 360 mins (6 hrs)         
def get_sleep(sleep_array, search_date):
    for i in reversed(sleep_array['sleep-minutesAsleep']):
        if str(i['dateTime']) == str(search_date):
            if i['value'] == str(0):
                sleepval = 360
            else:
                sleepval = i['value']
            return sleepval
            break
    return(360)

#This returns the calorie array containing 1 year of data 
def get_caloriearray(Fitbit_Access_Token):
    URL = 'https://api.fitbit.com/1/user/-/activities/calories/date/today/1y.json'
    #URL = 'https://api.fitbit.com/1/user/-/activities/list.json?beforeDate=2016-01-28&sort=desc&limit=20&offset=0'
    req = urllib2.Request(URL)
    
    #Add the headers, first we base64 encode the client id and client secret with a : inbetween and create the authorisation header
    req.add_header('Authorization', 'Bearer ' + Fitbit_Access_Token)
    req.add_header('Content-Type', 'application/x-www-form-urlencoded')
    
    #Fire off the request
    
    try:
        response = urllib2.urlopen(req)
        FullResponse = response.read()
        #print ("Output >>> " + FullResponse)
        parsed_response = json.loads(FullResponse)
       
        return(parsed_response)
        
    except urllib2.URLError as e:
        print (e.code)
        print (e.read())           

#This returns the calories burned for a specific day; If there was not a calorie count recorded, we send 0)         
def get_prevdaycalories(calorie_array, search_date):
    for i in reversed(calorie_array['activities-calories']):
        if str(i['dateTime']) == str(search_date):
            return i['value']
            break
    return(0)    
        


    
    

Create ML Model Workout App 2

Python
This code runs every 30 mins to perform the final steps of creating a machine learning model.
"""
Darian Johnson
http://darianbjohnson.com

This lambda funtion is part of the Best Time of Day to Workout Recommendation app

This script runs everye 30 minutes to complete the set up of a machine learning model
1) If the datasource statue changes to complete, then the machine model is created
2) if the machine model is created, then the status table is updated to allow the Alexa app to use the model
3) Any failures are recorded and relay back to the user via Alexa


You can find out more about this application at http://darianbjohnson.com/workout-recommendation
"""
from __future__ import print_function

import json
import urllib
import base64
import urllib2
import boto3
import time
import datetime

import uuid

from datetime import date
from datetime import timedelta
from datetime import datetime

print('Loading function')

ml = boto3.client('machinelearning')
lamb = boto3.client('lambda')
dynamodb = boto3.client('dynamodb')
sns = boto3.client('sns')
s3 = boto3.client('s3')

def lambda_handler(event, context):
    
    table = 'Workout_Recommendations'
    response = dynamodb.scan(TableName = table)

    #Only take action if the statsus is Pending
    for i in response['Items']:
        if i['Model_Type']['S'] == "PENDING":
            
            #If there is an assigned machine learning model id, check to see if the model is ready for use
            if 'ML_Model_Id' in i:
                ml_ready = ml.get_ml_model(
                    MLModelId=i['ML_Model_Id']['S'],
                    Verbose=False
                )
                
                #If the model is ready, change the status and create a real time endpoint
                if ml_ready['Status'] == 'COMPLETED':
                    status = 'SIMPLE'
                    
                    response = ml.create_realtime_endpoint(
                        MLModelId=i['ML_Model_Id']['S']
                    )
                    
                #If the model is not ready, update the status   
                elif ml_ready['Status'] == 'FAILED':
                    status = 'FAILED'
                elif ml_ready['Status'] == 'INPROGRESS':
                    status = 'PENDING'
                elif ml_ready['Status'] == 'PENDING':
                    status = 'PENDING'
                else:
                    status = 'PENDING'
                
                Details = {
                    'Fitbit_User_Id': {'S':i['Fitbit_User_Id']['S']},
                    'Model_Type': {'S':status},
                    'ML_Model_Id':{'S':i['ML_Model_Id']['S']},
                    'Datasource_Id':{'S':i['Datasource_Id']['S']}
                }
 
            #If the model id does not exist then check to see if the datasource is ready, and if so, kick off the function to create the model     
            else:
                ds_ready = ml.get_data_source(
                    DataSourceId=i['Datasource_Id']['S'],
                    Verbose=False
                )
                
                #If the datasource is complete, then start the model creation
                if ds_ready['Status'] == 'COMPLETED':
                    status = 'PENDING'
                    
                    ML_ID = 'ML-' + i['Fitbit_User_Id']['S'] + '-' + str(uuid.uuid4())
                    
                    ml_start = ml.create_ml_model(
                        MLModelId=ML_ID,
                        MLModelName='ML-' + i['Fitbit_User_Id']['S'],
                        MLModelType='REGRESSION',
                        TrainingDataSourceId=i['Datasource_Id']['S']
                    )
                
                #If the datasource is no complete, then update status as appropriate   
                elif ds_ready['Status'] == 'INPROGRESS':
                    status = 'PENDING'
                elif ds_ready['Status'] == 'PENDING':
                    status = 'PENDING'
                elif ds_ready['Status'] == 'FAILED':
                    status = 'FAILED'
                elif ds_ready['Status'] == 'DELETED':
                    status = 'FAILED'
                else:
                    status = 'PENDING'

                # Set the fields to update the status table
                if ds_ready['Status'] == 'COMPLETED':
                    Details = {
                        'Fitbit_User_Id': {'S':i['Fitbit_User_Id']['S']},
                        'ML_Model_Id':{'S':ML_ID},
                        'Model_Type': {'S':status},
                        'Datasource_Id':{'S':i['Datasource_Id']['S']}
                    }
                else:
                    Details = {
                        'Fitbit_User_Id': {'S':i['Fitbit_User_Id']['S']},
                        'Model_Type': {'S':status},
                        'Datasource_Id':{'S':i['Datasource_Id']['S']}
                    }
                    
             # Update the table as required    
            response = dynamodb.put_item(TableName = table, Item = Details)
            
    return('Complete')

Credits

Darian Johnson

Darian Johnson

8 projects • 139 followers
Technologist. Music lover. Fitness enthusiast. Movie buff. Fan of sci-fi and comic books.

Comments