Sunday, 15 March 2015

Raspberry Pi Powered Coffee Need Predictor Using Python and API Data

I like coffee.  A lot.  Too much perhaps.  Hence a Geek project that includes coffee is extremely exciting for me!

In a queue in a coffee shop a couple of weeks ago, the Barista recognised my colleague and asked him if he wanted "his usual" coffee.  After commenting on how cool this was we continued to discuss what if a new Barista had started?  How could the coffee shop use technology to provide the same great service?  So we discussed:

  • Perhaps the coffee shop could allow you to register your photo and your preferred coffee online and then use facial recognition as you join the queue to spot you and tell the Barista what your preference is.

...but them, as Geek conversations go, we evolved this to be:

  • Facial recognition plus access to personal API data.  This API data will tell the coffee shop how tired you're likely to be and thus they can recommend the coffee you need rather than the coffee you prefer.

For some reasons the person serving behind the counter wasn't up for this as an idea so I decided to build it myself...

So overall this is based upon the premise that caffeine is a stimulant and the more tired you are the more caffeine you need.  Sounds plausible to me.

First to think about what information is readily available about how tired I might be.  Based upon the Geek work I've done recently I have:

  • Fitbit data based upon the FitBit API.  This will tell me how many steps I did the day before, how many floors I climbed and how well I slept the previous night.
  • Strava data from my Garmin Forerunner that I've accessed using the Strava API.
  • General exercise data I've logged using my "Jerks" system.

The big idea was that I could write a Python script on my Raspberry Pi to:

  1. Gather data from these sources.
  2. Compare the results against pre-defined thresholds.
  3. Accumulate coffee points if the thresholds are breached.
  4. Map the resulting number of points to a coffee (more points is a coffee with more caffeine) and email the result to myself.

There's a full code listing at the bottom but here's a quick summary of the key points.

Fitbit Data
Using Python Fitbit (see my last post on this which tells you how to authenticate and access the API using Python objects).

To get steps and floors I created a Fitbit object and then did:
fitbit_activities = authd_client._COLLECTION_RESOURCE('activities',date=MyDate)

Where MyDate is yesterdays date in the format YYYY-MM-DD.  This is formed using the GetADate() function (see full code listing below for this).  GetADate() takes a parameter to define if it's todays date, yesterdays etc. and then does a date subtraction to provide the required date.

The response is a JSON structure from which you can extract the step count and the number of floors.  JSON parsing is a bit hit 'n' miss for me but through trial and error I found out:
Steps = fitbit_activities['summary']['steps'])
Floors = fitbit_activities['summary']['floors']

This can then be compared with thresholds to see if a "coffee point" is accumulated.  Specifically:
  • More than 10,000 means a coffee point is accumulated.
  • More than 25 floors means a coffee point is accumulated.
(With the floor and steps thresholds being nicely defined as constants).

I got the sleep data by doing:
fitbit_sleep = authd_client._COLLECTION_RESOURCE('sleep')

(No date specified as I wanted today's sleep value).  The specific value wanted can then be extracted from the JSON using:
Sleep = fitbit_sleep['summary']['totalMinutesAsleep']

Then the threshold comparison can be done again:
  • Less than 420 minutes (7 hours) means a point accumulated.
As an aside, the sleep element of the API looks epic.  More investigation is required but it looks like a record is logged for every minute that the Fitbit thinks you're asleep.  More Geek investigation is required I think...

Strava Data
Strava data is easy to access using a direct API plus an access token you get when you register as a developer.  (See here for a refresher on this).

The semi-tricky bit was specifying a date parameter to avoid extracting all data from the API and having to parse it yourself to pick out all action since the day before.  With the Strava API you can specify "before" and "after" parameters in Unix time to limit what is provided by the API.  To do this I:
  1. Used the GetADate function described above to get yesterday's date.
  2. Turned this into Unix time using a function I wrote called ConvertToUnix()
You can then get the API date using methods from the built-in urllib2 module:
StravaText = urllib2.urlopen('' + StravaToken + '&after=' + TheUnixTime).read()

A coffee point is accumulated simply by virtue of having done one or more Strava activities since yesterday.  These can be counted by simply doing:
StravaCount = StravaText.count('tcx')

Where the string 'tcx' appears once and once only for each Strava activity (it's the filename I uploaded from Garmin with the exercise data in it).

Jerks Data
For the Jerks system I created, I log data in a mysql database on my Raspberry Pi.  Hence accessing data is simply a question of running an SQL query.  First create a database object:
db = MySQLdb.connect(dbHost,dbDatabase,dbUsername,dbPassword)

Then get yesterdays date (same method as above) and form the query:
JerksQuery =  'SELECT * FROM ' + dbMainTable + ' WHERE tdate = "' + MyDate + '";'
Then counting how many Jerks activities I did is a simple case of assessing curs.rowcount.  More than 5 Jerks (my average for the last couple of weeks) means a coffee point is accumulated.

Coffee Point to Coffee Translation
So that just leaves mapping the number of coffee points accumulated to an actual coffee.  This is done in the ChooseACoffee() function and this is the "complex" logic:
  • 0 points means a nice cup of tea.
  • 1 point means a small coffee.
  • 2 points means a medium coffee.
  • 3 points means a large coffee.
  • 4 points means a large coffee and a single espresso.
  • 5 points means a large coffee and a double espresso.
With the coffee in question (Latte, Americano or Cappuccino) selected by creating a random integer from 1 to 3 using the randint function.

Finishing it Off
So having chosen the coffee the script then writes a HTML email (re-using code from my Jerks summary project) then sends it (again re-using the Jerks summary code).

The resulting email is sent automatically as the script is run using a cron job and looks like this:

There's much room for improvement here:
  • The overall algorithm is far too simplistic.  For example, if I had 5 minutes sleep or 6 hours 59 minutes sleep this still only equates to 1 coffee point.  Equally a Strava (which could be pretty hard) is given the same weighting as a jerk (which could be relatively easy). 
  • The data structures I use are really lame.  I use multiple lists rather then using a single data structure.  I need to investigate this more.  Comments please on this.
  • The email is a bit boring.  I experimented with adding images to the HTML content of the email but it didn't work.  I think I need to use MIME multipart structures for this but didn't get round to doing it.
  • There's loads of code there; far too much.  I need to streamline it a bit.  Perhaps write it in Java...

The Code
It's all below.  Of course I've blanked out all the keys and passwords!  You'll need to get these yourself.  Happy drinking?

#V1 = First attempt
import fitbit
from datetime import date, datetime, timedelta
from time import time, mktime
import smtplib
import MySQLdb
import urllib2
from random import randint

#MIME multipart stuff
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.MIMEImage import MIMEImage

#Date manipulation related constants
DateToday = 'DateToday'
DateYesterday = 'DateYesterday'
DateFirstDayOfWeek = 'DateFirstDayOfWeek'
DateFirstDayOfMonth = 'DateFirstDayOfMonth'
DateFirstDayOfYear = 'DateFirstDayOfYear'

#Constants - For Fitbit API
CLIENT_KEY = 'lovelovemedo'
CLIENT_SECRET = 'youknowiloveyou'
USER_KEY = 'illalwaysbetrue'
#USER_KEY = 'sopleasepleaseplease'
USER_SECRET = 'lovemedo'

#Constants - Fitbit thresholds
StepThreshold = 10000   #Daily target
FloorThreshold = 25     #Rough average for when I'm not running
SleepThreshold = 420    #7 hours sleep

#Constants - For Strava
StravaToken = 'itsbeenaharddaysnight'
StravaThreshold = 0

#Constants - For Jerks
dbMainTable = 'iamtheeggman'
dbHost = 'theyaretheeggmen'
dbDatabase = 'iamthewalrus'
dbUsername = 'googoogajoob'
dbPassword = 'googoogajoob'

#Constants - Jerks thresholds
JerksThreshold = 5

#Email Constants  
smtpserver = ''
AUTHREQUIRED = 1 # if you need to use SMTP AUTH set to 1
smtpuser = ''  # for SMTP AUTH, set SMTP username here
smtppass = 'andhetoldmeofhislife'  # for SMTP AUTH, set SMTP password here
RECIPIENTS = 'in.a.yellow.submarine'

#Choose what coffee to drink.  Takes a points count as an input
#and decides what coffee to drink.  Randomises to choose a different coffee type
def ChooseACoffee(CoffPoints):
  #First choose a coffee type
  CoffeeVar = randint(1,3)
  if (CoffeeVar == 1):
    CoffeeType = 'Latte'
  elif (CoffeeVar == 2):
    CoffeeType = 'Cappucino'
  elif (CoffeeVar == 3):
    CoffeeType = 'Americano'

  #Now choose a size based upon the points value sent as a parameter
  if (CoffPoints == 0):
    CoffeeString = 'You need a nice cup of tea.'
  elif (CoffPoints == 1):
    CoffeeString = 'You need a small ' + CoffeeType + '.'
  elif (CoffPoints == 2):
    CoffeeString = 'You need a medium ' + CoffeeType + '.'
  elif (CoffPoints == 3):
    CoffeeString = 'You need a large ' + CoffeeType + '.'
  elif (CoffPoints == 4):
    CoffeeString = 'You need a large ' + CoffeeType + ' and a single espresso.'
  elif (CoffPoints == 5):
    CoffeeString = 'You need a large ' + CoffeeType + ' and a double espresso.'

  return CoffeeString

#Gets a initial timestampt for emails
def GetDateTime():
  #Get the time, format it and return it
  now =
  return str( + ' ' + str(now.strftime("%H:%M:%S"))

#Send an email
def SendEmail(TheSubject,TheMessage):
  #Form the message
  #msg = "To:%s\nFrom:%s\nSubject: %s\n\n%s" % (RECIPIENTS, SENDER, TheSubject, TheMessage)
  #Start forming the MIME Multipart message
  msg = MIMEMultipart('alternative')
  msg['Subject'] = TheSubject
  msg['From'] = SENDER
  msg['To'] = RECIPIENTS

  # Record the MIME types of both parts - text/plain and text/html.
  #part1 = MIMEText(text, 'plain')
  part2 = MIMEText(TheMessage, 'html')
  # Attach parts into message container.
  # According to RFC 2046, the last part of a multipart message, in this case
  # the HTML message, is best and preferred.
  #print msg

  #Do the stuff to send the message  
  server = smtplib.SMTP(smtpserver,587)
  server.sendmail(SENDER, [RECIPIENTS], msg.as_string())

#Creates a string with a HTML table and a heading based upon parameters sent
def CreateHTMLTable(InTitle,CatList,AcList,ThreshList,PtList):
    #Start with the heading
    OutString = '<H1>' + InTitle + '</H1>\r\n'
    #Start the table
    OutString = OutString + '<table border="1">\n\r'
    #See if there is anything to write
    for i in range(0,len(AcList)):
      #Add the table opening tag

      #Add the list elements as table rows
      OutString = OutString + '<tr>\r\n'
      #Add the table elements
      OutString = OutString + '<td>' + CatList[i] + '</td>\r\n'
      OutString = OutString + '<td>' + AcList[i] + '</td>\r\n'
      OutString = OutString + '<td>' + ThreshList[i] + '</td>\r\n'
      OutString = OutString + '<td>' + PtList[i] + '</td>\r\n'

      #Close the table row
      OutString = OutString + '</tr>\r\n'    

    #Close the table tag      
    OutString = OutString + '</table>\r\n'
    #Return the result
    return OutString
    return 'Error creating table'
#Returns dates in different formats for different uses based upon the parameter passed
#Returns a date based upon the parameter supplied
def GetADate(DateType):
    if (DateType == DateToday):    #Just get and return todays date
      #Get the time, format it and return it
      now =
      return now.strftime("%Y-%m-%d")
    elif (DateType == DateYesterday):
      now =
      TheDateYesterday = now - timedelta(days=1)
      return TheDateYesterday.strftime("%Y-%m-%d")
    elif (DateType == DateFirstDayOfWeek):   #The first day of the current week.  Sunday is 0. Monday is 1 etc.  We want to know how many days from Monday it is
      #Find what day of the week it is
      now =
      DayOfWeek = int(now.strftime("%w"))   #Get the number of the day of the week
      print 'Day of week ->>' + str(DayOfWeek)
      #See what to subtract.  Sunday is a special case
      if (DayOfWeek == 0):
        DayDelta = 6         #Monday was always 6 days ago on a Sunday!
        DayDelta = DayOfWeek - 1
      print 'Day delta ->>' + str(DayDelta)
      DateOfMonday = now - timedelta(days=DayDelta)
      print 'Monday was ->>' + str(DateOfMonday)
      return DateOfMonday.strftime("%Y-%m-%d")
    elif (DateType == DateFirstDayOfMonth):
      now =
      DayOfMonth =  int(now.strftime("%d"))
      DayDelta = DayOfMonth - 1
      FirstDateOfMonth = now - timedelta(days=DayDelta)
      return FirstDateOfMonth.strftime("%Y-%m-%d")
    elif (DateType == DateFirstDayOfYear):
      now =
      DayOfYear =  int(now.strftime("%j"))
      DayDelta = DayOfYear - 1
      FirstDateOfYear = now - timedelta(days=DayDelta)
      return FirstDateOfYear.strftime("%Y-%m-%d")

#Returns a Unix time based upon the date sent
def ConvertToUnix(InDate):
  #What we'll be sent is a text string in the format yyyy-mm-dd.  Need to change it to a proper date variable
  DateVal = datetime.strptime(InDate,"%Y-%m-%d")
  #Now turn into Unix time
  UnixTime = mktime(DateVal.timetuple())
  return int(UnixTime)

#Main body of code

#Start a list that we'll use to track all the data
CategoryList = []     #Contains a string showing the category
ActualsList = []      #Contains a string showing actuals
ThresholdList = []    #Contains a string showing thresholds
PointsList = []       #Contains a string showing resulting points

#Add the headings for the resulting table
PointsList.append('Coffee Points')

#First we do Fitbit
#Create an object to access the Fitbit API
authd_client = fitbit.Fitbit(CLIENT_KEY, CLIENT_SECRET, resource_owner_key=USER_KEY, resource_owner_secret=USER_SECRET)

#Get yesterdays activities - First get a date in format yyyy-mm-dd
#Now access the activities component of the API
MyDate = GetADate(DateYesterday)
fitbit_activities = authd_client._COLLECTION_RESOURCE('activities',date=MyDate)
print 'Activities:'
print fitbit_activities
print '~~~~~~~~~~~~~~~~~~~~'
print 'Steps: ' + str(fitbit_activities['summary']['steps'])
print 'Floors: ' + str(fitbit_activities['summary']['floors'])

#Add the various items to the lists.  Start with Steps
CategoryList.append('Fitbit Steps')
if (fitbit_activities['summary']['steps'] > StepThreshold):
  CoffeePoints = 1
  CoffeePoints = 0

#Add the various items to the lists.  Do Floors
CategoryList.append('Fitbit Floors')
if (fitbit_activities['summary']['floors'] > FloorThreshold):
  CoffeePoints = CoffeePoints + 1

#Now for sleep
fitbit_sleep = authd_client._COLLECTION_RESOURCE('sleep')    #No date as it's today's data we want
print '~~~~~~~~~~~~~~~~~~~~~'
print 'Sleep: ' + str(fitbit_sleep['summary']['totalMinutesAsleep'])

#Add the various items to the lists.  Now do sleep
CategoryList.append('Fitbit sleep')
if (fitbit_sleep['summary']['totalMinutesAsleep'] < SleepThreshold):
  CoffeePoints = CoffeePoints + 1

#Second we do Strava
#We want to see if we've done any Strava since first thing yesterday.  So after 00:00:00 yesterday.
#First we get yesterdays date, then we turn into Unix timestamp which can be used as Strava API "after" parameter
MyDate = GetADate(DateYesterday)
TheUnixTime = str(ConvertToUnix(MyDate))

#Access the Strava API using a URL
StravaText = urllib2.urlopen('' + StravaToken + '&after=' + TheUnixTime).read()
print StravaText
#See what we got
StravaCount = StravaText.count('tcx')
if (StravaCount > StravaThreshold):
  print 'You did a Strava!'
  print 'You did no Strava :-('

#Add to the output lists
if (StravaCount > StravaThreshold):
  CoffeePoints = CoffeePoints + 1

#Now we do Jerks - First connect to the database
db = MySQLdb.connect(dbHost,dbDatabase,dbUsername,dbPassword)

#Get yesterdays date for the SQL query
MyDate = GetADate(DateYesterday)
JerksQuery =  'SELECT * FROM ' + dbMainTable + ' WHERE tdate = "' + MyDate + '";'
print JerksQuery
curs.execute (JerksQuery)
print "Jerks count: " + str(curs.rowcount)

#Add to the output lists
CategoryList.append('Jerks count')
if (curs.rowcount > JerksThreshold):
  CoffeePoints = CoffeePoints + 1

for i in range(0,len(CategoryList)):
  print CategoryList[i] + '-' + ActualsList[i] + '-' + ThresholdList[i] + '-' + PointsList[i]

print '~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~'
print 'Total coffee points: ' + str(CoffeePoints)

#Get the email subject.  This is the coffee type inferred from the amountof coffee points
EmailSubject = ChooseACoffee(CoffeePoints)
print EmailSubject

#Start forming the HTML
EmailHTML = '<HTML>\r\n<HEAD></HEAD>\r\n<BODY>\r\n<H1>What Coffee do You Need?</H1>\r\n'

#Add the coffee image
#EmailHTML = EmailHTML + '<img src="/coffeecup.jpg" alt="coffee" width="256" height="256">'

#Add the table for the HTML
EmailHTML = EmailHTML + CreateHTMLTable(EmailSubject,CategoryList,ActualsList,ThresholdList,PointsList)

#End the HTML
EmailHTML = EmailHTML + '</BODY>\r\n</HTML>'

#Send the email
SendEmail(EmailSubject + ' ' + GetDateTime(),EmailHTML)

No comments:

Post a Comment