Adds a Water Sensor to the Home Assistant Fitbit Integration

readme.md

Fitbit Water Tracker for Home Assistant

Always forgetting to stay hydrated? Wish you could track your water intake along with all your other Fitbit data on Home Assistant? Well, we've got you covered with our "Fitbit Water Tracker" AppDaemon script!

What It Does:

• This nifty little script fetches and logs your daily water consumption data from your Fitbit account using the Fitbit API.

• You can now keep an eye on your hydration levels right from your Home Assistant dashboard. It's like having a personal water butler reminding you to drink more water!

Why You Need It:

• It's a bit puzzling why Fitbit and Home Assistant don't already track water consumption together. But don't worry, we've filled this thirst gap in the smart home world with our app!

• Sure, you know how many steps you've taken and your sleep patterns, but now you can complete the picture with how much H2O you're guzzling. Hydration heroes unite!

How It Works:

• The app automatically fetches your daily water log from the Fitbit API every 30 minutes. Stay up-to-date on your hydration game without lifting a finger!

• Feeling extra thirsty? No worries! You can manually trigger the script by toggling the "input_boolean.trigger_script" entity. Your water data is just a click away!

Requirements:

To use the "Fitbit Water Tracker" app, you'll need to set the following Fitbit API credentials in your Home Assistant:

• access_token: Your Fitbit API access token. Don't have it? Follow the Fitbit API OAuth2 Tutorial to get one.

• refresh_token: Your Fitbit API refresh token. Obtain it during the OAuth2 setup following the tutorial above.

• client_id: Your Fitbit API client ID. You'll need to create an app and get this from your Fitbit developer account.

• client_secret: Your Fitbit API client secret. Also obtained during the app creation in your Fitbit developer account.

• user_id: Your Fitbit user ID. It's part of your Fitbit API credentials.

Note: Just a little secret between us – make sure to keep your credentials secure and don't share them with anyone! Also, when declaring the script in apps.yaml, don't forget to add client_id and client_secret along with other parameters.

Here's an example of how to declare the script in apps.yaml:

fitbit_water_tracker:
  module: FitbitWaterLog
  class: FitbitWaterLog
  client_id: YOUR_FITBIT_CLIENT_ID
  client_secret: YOUR_FITBIT_CLIENT_SECRET

Stay hydrated, fit, and fabulous with the Fitbit Water Tracker for Home Assistant. Let's quench that thirst for data, one refreshing sip at a time!

Keywords: Fitbit, Water, Home Assistant, hydration, AppDaemon, API integration, stay hydrated, smart home, thirst gap.

FitbitWaterLog.py

import appdaemon.plugins.hass.hassapi as hass
import requests
from datetime import datetime, timedelta

class FitbitWaterLog(hass.Hass):
    """
    AppDaemon app to fetch and log the daily water consumption from Fitbit API.
    """

    def initialize(self):
        """
        Initialize the app and schedule periodic fetching of water log data.
        Also, listen for manual triggering of the script through an input_boolean.
        """
        # Define the interval to run the script (every 30 minutes)
        interval = 30 * 60
        self.run_every(self.fetch_water_log, datetime.now(), interval)

        # Listen for the input_boolean.trigger_script state change to trigger the script manually
        self.listen_state(self.trigger_script, "input_boolean.trigger_script")

    def fetch_water_log(self, kwargs):
        """
        Fetch the user's water log data for the current day from the Fitbit API.

        Parameters:
            kwargs (dict): A dictionary of additional arguments. Not used in this function.

        Returns:
            None
        """
        self.log("fetch_water_log function executed at: " + str(datetime.now()))

        # Get the user-defined Fitbit API credentials from input_text objects
        access_token = self.get_state("input_text.fitbit_access_token")
        refresh_token = self.get_state("input_text.fitbit_refresh_token")
        user_id = self.get_state("input_text.fitbit_user_id")

        # Get today's date in the format yyyy-MM-dd
        today = datetime.now().strftime("%Y-%m-%d")

        # Make the API call to retrieve water log for today
        url = f"https://api.fitbit.com/1/user/{user_id}/foods/log/water/date/{today}.json"
        headers = {"Authorization": f"Bearer {access_token}"}
        response = requests.get(url, headers=headers)

        # Check if the response is unauthorized (status code 401)
        if response.status_code == 401:
            # Renew the access token using the refresh token
            new_access_token = self.renew_access_token(refresh_token)

            if new_access_token is not None:
                # Retry the API call with the new access token
                headers["Authorization"] = f"Bearer {new_access_token}"
                response = requests.get(url, headers=headers)
            else:
                self.log("Failed to renew access token.")
                return

        if response.status_code == 200:
            # Successfully retrieved the water log
            water_log = response.json()
            self.log(f"Water log for {today}: {water_log}")

            # Calculate and save the total amount of water consumed
            total_water_consumed = sum(entry["amount"] for entry in water_log["water"])
            self.set_state("input_number.water_consumed", state=total_water_consumed)
            self.log(f"Total water consumed for {today}: {total_water_consumed} ml")
        else:
            # Failed to retrieve the water log
            self.log(f"Failed to fetch water log. Status code: {response.status_code}")

    def renew_access_token(self, refresh_token):
        """
        Renew the Fitbit API access token using the provided refresh token.

        Parameters:
            refresh_token (str): The current refresh token used to obtain a new access token.

        Returns:
            str or None: The new access token if successfully renewed, otherwise returns None.
        """
        # Replace CLIENT_ID and CLIENT_SECRET with your actual Fitbit OAuth credentials
        client_id = self.args.get("client_id")
        client_secret = self.args.get("client_secret")

        url = "https://api.fitbit.com/oauth2/token"
        payload = {
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": client_id,
            "client_secret": client_secret
        }

        response = requests.post(url, headers=payload)
        if response.status_code == 200:
            data = response.json()
            new_access_token = data.get("access_token")
            new_refresh_token = data.get("refresh_token")  # The API may return a new refresh token

            # Save the new access token to input_text.fitbit_access_token
            self.set_state("input_text.fitbit_access_token", state=new_access_token)

            # If the API returned a new refresh token, save it to input_text.fitbit_refresh_token
            if new_refresh_token:
                self.set_state("input_text.fitbit_refresh_token", state=new_refresh_token)

            self.log("SUCCESS: Access token renewed successfully.")
            return new_access_token
        else:
            self.log(f"Failed to renew access token. Status code: {response.status_code}")
            self.log(response.json())
    
    def trigger_script(self, entity, attribute, old, new, kwargs):
        """
        Manually trigger the script to fetch and log the water consumption.

        Parameters:
            entity (str): The entity ID of the input_boolean.
            attribute (str): The attribute that changed. Not used in this function.
            old (str): The previous state of the input_boolean. Not used in this function.
            new (str): The new state of the input_boolean.
            kwargs (dict): A dictionary of additional arguments. Not used in this function.

        Returns:
            None
        """
        # Trigger the script manually using input_boolean.trigger_script
        if new == "on":
            self.fetch_water_log(None)
            self.log("Manually triggered the script.")
            self.turn_off("input_boolean.trigger_script")