fitbit.py

#!/usr/bin/env python
# coding: utf-8

# In[27]:


import logging
import json
import requests
import uuid
from datetime import datetime
from urllib.parse import urlencode, quote, urlparse, parse_qs
from http.server import HTTPServer, BaseHTTPRequestHandler

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("basic_server")



OAUTH_CONFIG = {**dict(
    client_id="<from .credentials.json>",
    client_secret="<from .credentials.json>",
    authorize_uri="https://www.fitbit.com/oauth2/authorize",
    token_uri="https://api.fitbit.com/oauth2/token",
    redirect_uri="http://localhost:8189/auth_code"
), **json.loads(open(".credentials.json").read())}


# In[2]:


class FitbitAuth: 
    CREDENTIALS = None
    
    @staticmethod
    def initiate(r):
        FitbitAuth.STATE = uuid.uuid4()
        params = dict(
            response_type="code",
            client_id=OAUTH_CONFIG["client_id"],
            redirect_uri=OAUTH_CONFIG["redirect_uri"],
            scope="activity heartrate location nutrition profile settings sleep social weight",
            state=FitbitAuth.STATE,
            expires_in=604800,
        )
        auth_url = f"https://www.fitbit.com/oauth2/authorize?{urlencode(params, quote_via=quote)}"
        r.send_response(302)
        r.send_header("Location", auth_url)
        r.end_headers()
        r.wfile.write(f'Redirecting to {auth_url}'.encode("utf-8"))

    @staticmethod
    def token(r):
        response = requests.post(
            OAUTH_CONFIG["token_uri"],
            auth=requests.auth.HTTPBasicAuth(OAUTH_CONFIG["client_id"], OAUTH_CONFIG["client_secret"]),
            data=dict(
                code=parse_qs(urlparse(r.path).query)["code"][0],
                grant_type="authorization_code",
                client_id=OAUTH_CONFIG["client_id"],
                state=FitbitAuth.STATE,
                redirect_uri=OAUTH_CONFIG["redirect_uri"]
            )
        )
        if response.status_code != 200:
            log.info("Response %s", response.status_code)
            log.info("Response %s", response.content)
        else:
            log.info("Authentication successful")
            FitbitAuth.CREDENTIALS = response.json()
        r.send_response(200)
        r.wfile.write(b'OK')

class BasicHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        log.debug("Client %s", self.client_address)
        log.debug("Request %s", self.requestline)
        log.debug("Path %s", self.path)
        log.debug("Headers %s", self.headers)

        if self.path == "/":
            FitbitAuth.initiate(self)
        elif self.path.startswith("/auth_code"):
            FitbitAuth.token(self)
        else:
            self.send_response(200, message="OK")
            self.wfile.write(b'OK')
        return
    
    def do_POST(self):
        self.do_GET()

httpd = HTTPServer(('localhost', 8189), BasicHandler)
httpd.handle_request()
httpd.handle_request()
httpd.server_close()
FitbitAuth.CREDENTIALS


# In[3]:


auth_header = {"Authorization": f'{FitbitAuth.CREDENTIALS["token_type"]} {FitbitAuth.CREDENTIALS["access_token"]}'}
response = requests.get(
    f"https://api.fitbit.com/1/user/-/activities/heart/date/{datetime.now().strftime('%Y-%m-%d')}/1m.json",
    headers=auth_header,
)
print(response.status_code)
print(response.json())


# In[4]:


date = datetime.now().strftime("%Y-%m-%d")
response = requests.get(
    f"https://api.fitbit.com/1/user/-/activities/heart/date/{date}/1d/1sec.json",
    headers=auth_header,
)
print(response.status_code)
print(response.json())
data = response.json()


# In[19]:


import pandas as pd
df = pd.DataFrame(data['activities-heart-intraday']['dataset']).rename({"value": "heartrate"}, axis=1)
df['datetime'] = pd.to_datetime(df.apply(lambda r: f"{date} {r['time']}", axis=1))
df = df.drop("time", axis=1).set_index("datetime")
df = df[ (df.index > pd.to_datetime(f"{date} 7:30")) & (df.index < pd.to_datetime(f"{date} 8:20"))]


# In[20]:


df.plot(
    figsize=(16,10),
    grid=True,
    yticks=range(70, 210, 10)
)


# In[21]:


import seaborn as sns
sns.relplot(data=df, kind='line', height=6, aspect=2)


# In[23]:


import plotly.express as py

py.line(df, y="heartrate")


# In[26]:


from bokeh.models import ColumnDataSource, DatetimeTickFormatter
from bokeh.plotting import figure, output_notebook, show
import itertools
output_notebook()

p = figure(plot_width=900, plot_height=300)
p.line(source=ColumnDataSource(df), x="datetime", y="heartrate", line_width=2)
p.xaxis.formatter=DatetimeTickFormatter(**dict(zip(("minutes", "hourmin", "hours"), itertools.repeat('%I:%M %P'))))

show(p)


# In[ ]:

pip install requests pandas seaborn plotly bokeh