Merge branch 'FitbitIntegration'

.gitignore

@@ -133,6 +133,7 @@ dmypy.json
 withings.json
 wahoo.json
 strava.json
+fitbit.json
 .vscode/settings.json
 HealthSync.code-workspace
 healtsync/pyvenv.cfg

env.md

@@ -11,3 +11,7 @@ wahoo_redirect_uri = 'https://www.yoururl.com'
 
 strava_athlete_id = 'sssss'
 strava_client_secret = 'cccccccccc'
+
+fitbit_client_id = 'cccccc'
+fitbit_client_secret = 'yyyyyyyyyyyyyyyyyyyyyyy'
+fitbit_token_url = 'https://api.fitbit.com/oauth2/token'

fitbit_api.py

@@ -0,0 +1,215 @@
+"""
+    FitBit API
+    https://dev.fitbit.com/build/reference/web-api/troubleshooting-guide/oauth2-tutorial/
+"""
+
+import json
+import logging
+import os
+import sys
+from datetime import datetime
+import base64
+import webbrowser
+import secrets
+import hashlib
+from dotenv import load_dotenv
+import requests
+
+load_dotenv()
+
+# Strava API access
+fitbit_client_id = os.getenv("fitbit_client_id")
+fitbit_client_secret = os.getenv("fitbit_client_secret")
+fitbit_token_url = os.getenv("fitbit_token_url")
+fitbit_cfg = "fitbit.json"
+
+# Fitbit URLs to use for the API call
+dateparam = datetime.now().strftime("%Y-%m-%d")
+urlDict = {
+    "activities": "https://api.fitbit.com/1/user/-/activities/date/" + dateparam + ".json",
+    "weight": "https://api.fitbit.com/1/user/-/body/log/weight/date/" + dateparam + ".json",
+    "set_weight": "https://api.fitbit.com/1/user/-/body/log/weight.json",
+    "set_bodyfat": "https://api.fitbit.com/1/user/-/body/log/fat.json",
+    "water": "https://api.fitbit.com/1/user/-/foods/log/date/" + dateparam + ".json",
+    "sleep": "https://api.fitbit.com/1.2/user/-/sleep/date/" + dateparam + ".json",
+    "user": "https://api.fitbit.com/1/user/-/profile.json",
+    "heartrate": "https://api.fitbit.com/1/user/-/activities/heart/date/" + dateparam + "/1d.json",
+}
+
+# # def main():
+#     """Main function for testing"""
+#     # fitbit_set_weight(user_weight, user_fat)
+#     # fitbit_introspect(fitbit_access_token)
+
+def fitbit_read_user():
+    """Read user info info from fitbit"""
+    fitbit_access_token = (
+        fitbit_refresh(json.load(open(fitbit_cfg, encoding="utf8")))
+        if os.path.isfile(fitbit_cfg)
+        else fitbit_authenticate()
+    )
+
+    headers = {
+        "Authorization": f"Bearer {fitbit_access_token}",
+        "accept": "application/json",
+    }
+
+    # urlDict["user"] = https://api.fitbit.com/1/user/-/profile.json
+    response = requests.get(urlDict["user"], headers=headers, timeout=10)
+    print("User info:", response.json()["user"]["fullName"])
+    logging.info("User info")
+    logging.info("User weight")
+
+    # urlDict["weight"] = https://api.fitbit.com/1/user/-/body/log/weight/date/
+    response = requests.get(urlDict["weight"], headers=headers, timeout=10)
+    return fitbit_access_token
+
+
+def fitbit_set_weight(user_weight,user_fat):
+    """Write user weight to fitbit"""
+
+    ''' Get tokens '''
+    fitbit_access_token = fitbit_read_user()
+    today = datetime.now().strftime("%Y-%m-%d")
+
+    ''' Now writing body weight to fitbit '''
+    dataweight = {"weight": user_weight, "date": today}
+    content_length_weight = len(str(dataweight))
+
+    headers = {
+        "Authorization": f"Bearer {fitbit_access_token}",
+        "accept": "application/json",
+        "content-length": str(content_length_weight),
+    }
+    response = requests.post(
+        urlDict["set_weight"], data=dataweight, headers=headers, timeout=10
+    )
+    if response.status_code != 201:
+            print("There was an error writing to Fitbit API:")
+            logging.info("There was an error writing to Fitbit API:")
+            logging.info(response.json())
+            return False
+
+    print(f"Succesful writing weight {user_weight} to FitBit API")
+    logging.info("Succesful writing weight %s to FitBit API", user_weight)
+
+    ''' Now writing body fat to fitbit '''
+    datafat = {"fat": user_fat, "date": today}
+    content_length_fat = len(str(datafat))
+
+    headers = {
+        "Authorization": f"Bearer {fitbit_access_token}",
+        "accept": "application/json",
+        "content-length": str(content_length_weight),
+    }
+    response = requests.post(
+        urlDict["set_bodyfat"], data=datafat, headers=headers, timeout=10
+    )
+
+    if response.status_code != 201:
+            print("There was an error writing fat to Fitbit API:")
+            logging.info("There was an error writing fat to Fitbit API:")
+            logging.info(response.json())
+            return False
+
+    print(f"Succesful writing weight {user_fat} to FitBit API")
+    logging.info("Succesful writing weight %s to FitBit API", user_fat)
+    return True
+
+
+
+def fitbit_introspect(fitbit_access_token):
+    """check token permissions for problem solving """
+    url = "https://api.fitbit.com/1.1/oauth2/introspect"
+    headers = {
+        "Authorization": f"Bearer {fitbit_access_token}",
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+    data = f"token={fitbit_access_token}"
+    response = requests.post(url, data=data, headers=headers, timeout=10)
+    print(response.json())
+
+
+def fitbit_refresh(token):
+    """Refresh fitbit token"""
+    url = "https://api.fitbit.com/oauth2/token"
+    encoded_cred = f"{fitbit_client_id}:{fitbit_client_secret}"
+    basic_token = base64.b64encode(bytes(encoded_cred, "utf8")).decode("utf-8")
+
+    headers = {
+        "Authorization": f"Basic {basic_token}",
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+
+    params = {
+        "grant_type": "refresh_token",
+        "refresh_token": token["refresh_token"],
+    }
+
+    response = requests.post(
+        url,
+        params=params,
+        headers=headers,
+        timeout=10,
+    )
+    output = response.json()
+
+    if response.status_code == 200:
+        json.dump(output, open(fitbit_cfg, "w", encoding="utf8"))
+        return output["access_token"]
+
+    print("Authenication failed")
+    logging.info("Authenication failed")
+    logging.info(output)
+    sys.exit()
+
+
+def fitbit_authenticate():
+    """Get a new Token"""
+    code_verifier = secrets.token_urlsafe(96)
+    hashed = hashlib.sha256(code_verifier.encode("ascii")).digest()
+    encoded = base64.urlsafe_b64encode(hashed)
+    code_challenge = encoded.decode("ascii")[:-1]
+
+    encoded_cred = f"{fitbit_client_id}:{fitbit_client_secret}"
+    basic_token = base64.b64encode(bytes(encoded_cred, "utf8")).decode("utf-8")
+
+    url = "https://www.fitbit.com/oauth2/authorize"
+    params = {
+        "response_type": "code",
+        "client_id": fitbit_client_id,
+        "code_challenge": code_challenge,
+        "code_challenge_method": "S256",
+        "scope": "activity heartrate profile sleep weight",
+    }
+
+    response = requests.get(url, params=params, timeout=10)
+
+    webbrowser.open(response.url, new=2)
+    fitbit_code = input(
+        "Copy the authorization code located between the code parameter name and the string #_=_     : "
+    )
+
+    params = {
+        "client_id": fitbit_client_id,
+        "code": fitbit_code,
+        "code_verifier": code_verifier,
+        "grant_type": "authorization_code",
+    }
+    headers = {
+        "Authorization": f"Basic {basic_token}",
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+
+    response = requests.post(
+        fitbit_token_url, params=params, headers=headers, timeout=10
+    )
+    output = response.json()
+    logging.info(response.json())
+    if response.status_code == 200:
+        json.dump(output, open(fitbit_cfg, "w", encoding="utf8"))
+        return output["access_token"]
+
+    print("Authenication failed")
+    logging.info(output)
+    sys.exit()

healthsync.py

@@ -1,15 +1,14 @@
 """
     config section
 """
-
-# from genericpath import isfile
 import sys
 import argparse
 import logging
 import wahoo_api
 import strava_api
 import intervals_api
 import withings_api
+import fitbit_api
 import send_telegram
 
 logging.basicConfig(filename="weightsync.log", encoding="utf-8", level=logging.INFO)
@@ -37,7 +36,7 @@ def main():
 
     if args.weight is None:
         logging.info("No weight on commandline, try to retrieve from withings.")
-        user_weight = withings_api.get_withings_user_weight()
+        user_weight, user_fat = withings_api.get_withings_user_weight()
         if user_weight is not None:
             withings_read_ok = True
         else:
@@ -63,9 +62,14 @@ def main():
     # Write the weight to Strava
     strava_send_ok = strava_api.set_strava_weight(user_weight)
 
+    # Write weight and fat to fitbit
+    fitbit_send_ok = fitbit_api.fitbit_set_weight(user_weight, user_fat)
+
+
+
     telegram_text = send_telegram.create_body_text(
-        user_weight, withings_read_ok, wahoo_send_ok, intervals_send_ok,
-        strava_send_ok
+        user_weight, withings_read_ok, wahoo_send_ok, intervals_send_ok, 
+        strava_send_ok, fitbit_send_ok
     )
     send_telegram.send_telegram_message(telegram_text)
 

send_telegram.py

@@ -34,7 +34,7 @@ def send_telegram_message(message):
 
 
 def create_body_text(
-    user_weight, withings_read_ok, wahoo_send_ok, intervals_send_ok, strava_send_ok
+    user_weight, withings_read_ok, wahoo_send_ok, intervals_send_ok, strava_send_ok, fitbit_send_ok
 ):
     """Create a body text for the telegram message"""
     today = date.today().strftime("%d-%m-%Y")
@@ -47,4 +47,5 @@ def create_body_text(
     text += "Data sent to Wahoo " + str(wahoo_send_ok) + "\n"
     text += "Data sent to Intervals " + str(intervals_send_ok) + "\n"
     text += "Data sent to Strava " + str(strava_send_ok) + "\n"
+    text += "Data sent to Fitbit " + str(fitbit_send_ok) + "\n"
     return text

withings_api.py

@@ -27,7 +27,8 @@ def get_withings_user_weight():
     """Read user weight from withings"""
     withings_data = get_withings_data()
     user_weight = withings_data["weight"]
-    return user_weight
+    user_fat = withings_data["bodyFat"]
+    return user_weight, user_fat
 
 
 def get_withings_data():