diff --git a/examples/requests_adafruit_discord_active_online.py b/examples/requests_adafruit_discord_active_online.py index 0175593..48a201c 100644 --- a/examples/requests_adafruit_discord_active_online.py +++ b/examples/requests_adafruit_discord_active_online.py @@ -1,8 +1,11 @@ -# SPDX-FileCopyrightText: 2022 DJDevon3 for Adafruit Industries +# SPDX-FileCopyrightText: 2023 DJDevon3 # SPDX-License-Identifier: MIT -# Coded for Circuit Python 8.0 -"""DJDevon3 Adafruit Feather ESP32-S2 Adafruit_Discord_Active_Users_Example""" +""" +Coded for Circuit Python 8.2.3 +requests_adafruit_discord_active_online +""" import gc +import os import time import ssl import json @@ -10,34 +13,38 @@ import socketpool import adafruit_requests -# No user or token required, 100% JSON web scrape from SHIELDS.IO +# Public API. No user or token required +# JSON web scrape from SHIELDS.IO # Adafruit uses Shields.IO to see online users # Initialize WiFi Pool (There can be only 1 pool & top of script) pool = socketpool.SocketPool(wifi.radio) -# Time between API refreshes -# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour +# Time in seconds between updates (polling) +# 600 = 10 mins, 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 -try: - from secrets import secrets -except ImportError: - print("Secrets File Import Error") - raise +# this example uses settings.toml for credentials +ssid = os.getenv("WIFI_SSID") +appw = os.getenv("WIFI_PASSWORD") + + +# Converts seconds to minutes/hours/days +def time_calc(input_time): + if input_time < 60: + sleep_int = input_time + time_output = f"{sleep_int:.0f} seconds" + elif 60 <= input_time < 3600: + sleep_int = input_time / 60 + time_output = f"{sleep_int:.0f} minutes" + elif 3600 <= input_time < 86400: + sleep_int = input_time / 60 / 60 + time_output = f"{sleep_int:.0f} hours" + else: + sleep_int = input_time / 60 / 60 / 24 + time_output = f"{sleep_int:.1f} days" + return time_output -if sleep_time < 60: - sleep_time_conversion = "seconds" - sleep_int = sleep_time -elif 60 <= sleep_time < 3600: - sleep_int = sleep_time / 60 - sleep_time_conversion = "minutes" -elif 3600 <= sleep_time < 86400: - sleep_int = sleep_time / 60 / 60 - sleep_time_conversion = "hours" -else: - sleep_int = sleep_time / 60 / 60 / 24 - sleep_time_conversion = "days" # Originally attempted to use SVG. Found JSON exists with same filename. # https://img.shields.io/discord/327254708534116352.svg @@ -49,7 +56,7 @@ requests = adafruit_requests.Session(pool, ssl.create_default_context()) while not wifi.radio.ipv4_address: try: - wifi.radio.connect(secrets["ssid"], secrets["password"]) + wifi.radio.connect(ssid, appw) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") @@ -93,7 +100,7 @@ print("Monotonic: ", time.monotonic()) print("\nFinished!") - print("Next Update in %s %s" % (int(sleep_int), sleep_time_conversion)) + print("Next Update: ", time_calc(sleep_time)) print("===============================") gc.collect() diff --git a/examples/requests_api_fitbit.py b/examples/requests_api_fitbit.py new file mode 100644 index 0000000..6494237 --- /dev/null +++ b/examples/requests_api_fitbit.py @@ -0,0 +1,312 @@ +# SPDX-FileCopyrightText: 2023 DJDevon3 +# SPDX-License-Identifier: MIT +# Coded for Circuit Python 8.2 + +import os +import time +import ssl +import wifi +import socketpool +import microcontroller +import adafruit_requests + +# Initialize WiFi Pool (There can be only 1 pool & top of script) +pool = socketpool.SocketPool(wifi.radio) + +# STREAMER WARNING: private data will be viewable while debug True +debug = False # Set True for full debug view + +# Can use to confirm first instance of NVM is correct refresh token +top_nvm = microcontroller.nvm[0:64].decode() +if debug: + print(f"Top NVM: {top_nvm}") # NVM before settings.toml loaded + +# --- Fitbit Developer Account & oAuth App Required: --- +# Required: Google Login (Fitbit owned by Google) & Fitbit Device +# Step 1: Create a personal app here: https://dev.fitbit.com +# Step 2: Use their Tutorial to get the Token and first Refresh Token +# Fitbit's Tutorial Step 4 is as far as you need to go. +# https://dev.fitbit.com/build/reference/web-api/troubleshooting-guide/oauth2-tutorial/ + +# Ensure these are in settings.toml +# Fitbit_ClientID = "YourAppClientID" +# Fitbit_Token = "Long 256 character string (SHA-256)" +# Fitbit_First_Refresh_Token = "64 character string" +# Fitbit_UserID = "UserID authorizing the ClientID" + +Fitbit_ClientID = os.getenv("Fitbit_ClientID") +Fitbit_Token = os.getenv("Fitbit_Token") +Fitbit_First_Refresh_Token = os.getenv( + "Fitbit_First_Refresh_Token" +) # overides nvm first run only +Fitbit_UserID = os.getenv("Fitbit_UserID") + +wifi_ssid = os.getenv("CIRCUITPY_WIFI_SSID") +wifi_pw = os.getenv("CIRCUITPY_WIFI_PASSWORD") + +# Time between API refreshes +# 300 = 5 mins, 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour +sleep_time = 900 + + +# Converts seconds in minutes/hours/days +def time_calc(input_time): + if input_time < 60: + sleep_int = input_time + time_output = f"{sleep_int:.0f} seconds" + elif 60 <= input_time < 3600: + sleep_int = input_time / 60 + time_output = f"{sleep_int:.0f} minutes" + elif 3600 <= input_time < 86400: + sleep_int = input_time / 60 / 60 + time_output = f"{sleep_int:.1f} hours" + else: + sleep_int = input_time / 60 / 60 / 24 + time_output = f"{sleep_int:.1f} days" + return time_output + + +# Authenticates Client ID & SHA-256 Token to POST +fitbit_oauth_header = {"Content-Type": "application/x-www-form-urlencoded"} +fitbit_oauth_token = "https://api.fitbit.com/oauth2/token" + +# Connect to Wi-Fi +print("\n===============================") +print("Connecting to WiFi...") +requests = adafruit_requests.Session(pool, ssl.create_default_context()) +while not wifi.radio.ipv4_address: + try: + wifi.radio.connect(wifi_ssid, wifi_pw) + except ConnectionError as e: + print("Connection Error:", e) + print("Retrying in 10 seconds") + time.sleep(10) +print("Connected!\n") + +# First run uses settings.toml token +Refresh_Token = Fitbit_First_Refresh_Token + +if debug: + print(f"Top NVM Again (just to make sure): {top_nvm}") + print(f"Settings.toml Initial Refresh Token: {Fitbit_First_Refresh_Token}") + +latest_15_avg = "Latest 15 Minute Averages" +while True: + # Use Settings.toml refresh token on first run + if top_nvm != Fitbit_First_Refresh_Token: + Refresh_Token = microcontroller.nvm[0:64].decode() + if debug: + # NVM 64 should match Current Refresh Token + print(f"NVM 64: {microcontroller.nvm[0:64].decode()}") + print(f"Current Refresh_Token: {Refresh_Token}") + else: + if debug: + # First run use settings.toml refresh token instead + print(f"Initial_Refresh_Token: {Refresh_Token}") + + try: + if debug: + print("\n-----Token Refresh POST Attempt -------") + fitbit_oauth_refresh_token = ( + "&grant_type=refresh_token" + + "&client_id=" + + str(Fitbit_ClientID) + + "&refresh_token=" + + str(Refresh_Token) + ) + + # ----------------------------- POST FOR REFRESH TOKEN ----------------------- + if debug: + print( + f"FULL REFRESH TOKEN POST:{fitbit_oauth_token}" + + f"{fitbit_oauth_refresh_token}" + ) + print(f"Current Refresh Token: {Refresh_Token}") + # TOKEN REFRESH POST + fitbit_oauth_refresh_POST = requests.post( + url=fitbit_oauth_token, + data=fitbit_oauth_refresh_token, + headers=fitbit_oauth_header, + ) + try: + fitbit_refresh_oauth_json = fitbit_oauth_refresh_POST.json() + + fitbit_new_token = fitbit_refresh_oauth_json["access_token"] + if debug: + print("Your Private SHA-256 Token: ", fitbit_new_token) + fitbit_access_token = fitbit_new_token # NEW FULL TOKEN + + # If current token valid will respond + fitbit_new_refesh_token = fitbit_refresh_oauth_json["refresh_token"] + Refresh_Token = fitbit_new_refesh_token + fitbit_token_expiration = fitbit_refresh_oauth_json["expires_in"] + fitbit_scope = fitbit_refresh_oauth_json["scope"] + fitbit_token_type = fitbit_refresh_oauth_json["token_type"] + fitbit_user_id = fitbit_refresh_oauth_json["user_id"] + if debug: + print("Next Refresh Token: ", Refresh_Token) + + # Store Next Token into NVM + try: + nvmtoken = b"" + fitbit_new_refesh_token + microcontroller.nvm[0:64] = nvmtoken + if debug: + print(f"Next Token for NVM: {nvmtoken.decode()}") + print("Next token written to NVM Successfully!") + except OSError as e: + print("OS Error:", e) + continue + + if debug: + # Extraneous token data for debugging + print("Token Expires in: ", time_calc(fitbit_token_expiration)) + print("Scope: ", fitbit_scope) + print("Token Type: ", fitbit_token_type) + print("UserID: ", fitbit_user_id) + + except KeyError as e: + print("Key Error:", e) + print("Expired token, invalid permission, or (key:value) pair error.") + time.sleep(300) + continue + + # ----------------------------- GET DATA ------------------------------------- + # POST should respond with current & next refresh token we can GET for data + # 64-bit Refresh tokens will "keep alive" SHA-256 token indefinitely + # Fitbit main SHA-256 token expires in 8 hours unless refreshed! + # ---------------------------------------------------------------------------- + detail_level = "1min" # Supported: 1sec | 1min | 5min | 15min + requested_date = "today" # Date format yyyy-MM-dd or today + fitbit_header = { + "Authorization": "Bearer " + fitbit_access_token + "", + "Client-Id": "" + Fitbit_ClientID + "", + } + # Heart Intraday Scope + FITBIT_SOURCE = ( + "https://api.fitbit.com/1/user/" + + Fitbit_UserID + + "/activities/heart/date/today" + + "/1d/" + + detail_level + + ".json" + ) + + print("\nAttempting to GET FITBIT Stats!") + print("===============================") + fitbit_get_response = requests.get(url=FITBIT_SOURCE, headers=fitbit_header) + try: + fitbit_json = fitbit_get_response.json() + intraday_response = fitbit_json["activities-heart-intraday"]["dataset"] + except ConnectionError as e: + print("Connection Error:", e) + print("Retrying in 10 seconds") + + if debug: + print(f"Full API GET URL: {FITBIT_SOURCE}") + print(f"Header: {fitbit_header}") + # print(f"JSON Full Response: {fitbit_json}") + # print(f"Intraday Full Response: {intraday_response}") + + try: + # Fitbit's sync to your mobile device & server every 15 minutes in chunks. + # Pointless to poll their API faster than 15 minute intervals. + activities_heart_value = fitbit_json["activities-heart-intraday"]["dataset"] + response_length = len(activities_heart_value) + if response_length >= 15: + activities_timestamp = fitbit_json["activities-heart"][0]["dateTime"] + print(f"Fitbit Date: {activities_timestamp}") + activities_latest_heart_time = fitbit_json["activities-heart-intraday"][ + "dataset" + ][response_length - 1]["time"] + print(f"Fitbit Time: {activities_latest_heart_time[0:-3]}") + print(f"Today's Logged Pulses : {response_length}") + + # Each 1min heart rate is a 60 second average + activities_latest_heart_value0 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 1]["value"] + activities_latest_heart_value1 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 2]["value"] + activities_latest_heart_value2 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 3]["value"] + activities_latest_heart_value3 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 4]["value"] + activities_latest_heart_value4 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 5]["value"] + activities_latest_heart_value5 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 6]["value"] + activities_latest_heart_value6 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 7]["value"] + activities_latest_heart_value7 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 8]["value"] + activities_latest_heart_value8 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 9]["value"] + activities_latest_heart_value9 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 10]["value"] + activities_latest_heart_value10 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 11]["value"] + activities_latest_heart_value11 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 12]["value"] + activities_latest_heart_value12 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 13]["value"] + activities_latest_heart_value13 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 14]["value"] + activities_latest_heart_value14 = fitbit_json[ + "activities-heart-intraday" + ]["dataset"][response_length - 15]["value"] + latest_15_avg = "Latest 15 Minute Averages" + print( + f"{latest_15_avg}" + + f"{activities_latest_heart_value14}," + + f"{activities_latest_heart_value13}," + + f"{activities_latest_heart_value12}," + + f"{activities_latest_heart_value11}," + + f"{activities_latest_heart_value10}," + + f"{activities_latest_heart_value9}," + + f"{activities_latest_heart_value8}," + + f"{activities_latest_heart_value7}," + + f"{activities_latest_heart_value6}," + + f"{activities_latest_heart_value5}," + + f"{activities_latest_heart_value4}," + + f"{activities_latest_heart_value3}," + + f"{activities_latest_heart_value2}," + + f"{activities_latest_heart_value1}," + + f"{activities_latest_heart_value0}" + ) + else: + print("Waiting for latest 15 values sync...") + print("Not enough values for today to display yet.") + print("No display from midnight to 00:15") + + except KeyError as keyerror: + print(f"Key Error: {keyerror}") + print( + "Too Many Requests, Expired token," + + "invalid permission," + + "or (key:value) pair error." + ) + continue + + print("Board Uptime: ", time_calc(time.monotonic())) # Board Up-Time seconds + print("\nFinished!") + print("Next Update in: ", time_calc(sleep_time)) + print("===============================") + + except (ValueError, RuntimeError) as e: + print("Failed to get data, retrying\n", e) + time.sleep(60) + continue + time.sleep(sleep_time) diff --git a/examples/requests_api_twitch.py b/examples/requests_api_twitch.py index e4b0dbb..396712f 100644 --- a/examples/requests_api_twitch.py +++ b/examples/requests_api_twitch.py @@ -1,57 +1,49 @@ -# SPDX-FileCopyrightText: 2022 DJDevon3 for Adafruit Industries +# SPDX-FileCopyrightText: 2023 DJDevon3 # SPDX-License-Identifier: MIT -# Coded for Circuit Python 8.0 -"""DJDevon3 Adafruit Feather ESP32-S2 Twitch_API_Example""" -import gc +# Coded for Circuit Python 8.2.x +# Twitch_API_Example + +import os import time import ssl import wifi import socketpool import adafruit_requests -# Twitch Developer Account & 0Auth App Required: +# Initialize WiFi Pool (There can be only 1 pool & top of script) +pool = socketpool.SocketPool(wifi.radio) + +# Twitch Developer Account & oauth App Required: # Visit https://dev.twitch.tv/console to create an app -# Ensure Twitch_ClientID & Twitch_Client_Secret are in secrets.py or .env +# Ensure these are in secrets.py or settings.toml # "Twitch_ClientID": "Your Developer APP ID Here", # "Twitch_Client_Secret": "APP ID secret here", +# "Twitch_UserID": "Your Twitch UserID here", +# Use settings.toml for credentials +ssid = os.getenv("CIRCUITPY_WIFI_SSID") +appw = os.getenv("CIRCUITPY_WIFI_PASSWORD") +twitch_client_id = os.getenv("Twitch_ClientID") +twitch_client_secret = os.getenv("Twitch_Client_Secret") # For finding your Twitch User ID # https://www.streamweasels.com/tools/convert-twitch-username-to-user-id/ -Twitch_UserID = "0000000" # Set User ID you want endpoints from - -# Initialize WiFi Pool (There can be only 1 pool & top of script) -pool = socketpool.SocketPool(wifi.radio) +twitch_user_id = os.getenv("Twitch_UserID") # User ID you want endpoints from # Time between API refreshes # 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour sleep_time = 900 -try: - from secrets import secrets -except ImportError: - print("Secrets File Import Error") - raise - -# Converts seconds in minutes/hours/days +# Converts seconds to minutes/hours/days def time_calc(input_time): if input_time < 60: - sleep_int = input_time - time_output = f"{sleep_int:.0f} seconds" - elif 60 <= input_time < 3600: - sleep_int = input_time / 60 - time_output = f"{sleep_int:.0f} minutes" - elif 3600 <= input_time < 86400: - sleep_int = input_time / 60 / 60 - time_output = f"{sleep_int:.0f} hours" - elif 86400 <= input_time < 432000: - sleep_int = input_time / 60 / 60 / 24 - time_output = f"{sleep_int:.1f} days" - else: # if > 5 days convert float to int & display whole days - sleep_int = input_time / 60 / 60 / 24 - time_output = f"{sleep_int:.0f} days" - return time_output + return f"{input_time:.0f} seconds" + if input_time < 3600: + return f"{input_time / 60:.0f} minutes" + if input_time < 86400: + return f"{input_time / 60 / 60:.0f} hours" + return f"{input_time / 60 / 60 / 24:.1f} days" # First we use Client ID & Client Secret to create a token with POST @@ -63,21 +55,20 @@ def time_calc(input_time): print("\n===============================") print("Connecting to WiFi...") requests = adafruit_requests.Session(pool, ssl.create_default_context()) -while not wifi.radio.ipv4_address: +while not wifi.radio.connected: try: - wifi.radio.connect(secrets["ssid"], secrets["password"]) + wifi.radio.connect(ssid, appw) except ConnectionError as e: print("Connection Error:", e) print("Retrying in 10 seconds") time.sleep(10) - gc.collect() print("Connected!\n") while True: try: # ----------------------------- POST FOR BEARER TOKEN ----------------------- print( - "\nAttempting to GENERATE Twitch Bearer Token!" + "Attempting Bearer Token Request!" ) # --------------------------------------- # Print Request to Serial debug_bearer_request = ( @@ -88,9 +79,9 @@ def time_calc(input_time): print("===============================") twitch_0auth_data = ( "&client_id=" - + secrets["Twitch_ClientID"] + + twitch_client_id + "&client_secret=" - + secrets["Twitch_Client_Secret"] + + twitch_client_secret + "&grant_type=client_credentials" ) @@ -113,12 +104,12 @@ def time_calc(input_time): print("JSON Dump: ", twitch_0auth_json) print("Header: ", twitch_0auth_header) print("Access Token: ", twitch_access_token) + twitch_token_type = twitch_0auth_json["token_type"] + print("Token Type: ", twitch_token_type) + print("Board Uptime: ", time_calc(time.monotonic())) twitch_token_expiration = twitch_0auth_json["expires_in"] print("Token Expires in: ", time_calc(twitch_token_expiration)) - twitch_token_type = twitch_0auth_json["token_type"] - print("Token Type: ", twitch_token_type) - print("Monotonic: ", time.monotonic()) # ----------------------------- GET DATA ------------------------------------- # Bearer token is refreshed every time script runs :) @@ -128,14 +119,13 @@ def time_calc(input_time): # ---------------------------------------------------------------------------- twitch_header = { "Authorization": "Bearer " + twitch_access_token + "", - "Client-Id": "" + secrets["Twitch_ClientID"] + "", + "Client-Id": "" + twitch_client_id + "", } TWITCH_FOLLOWERS_SOURCE = ( - "https://api.twitch.tv/helix/users" - + "/follows?" - + "to_id=" - + Twitch_UserID - + "&first=1" + "https://api.twitch.tv/helix/channels" + + "/followers?" + + "broadcaster_id=" + + twitch_user_id ) print( "\nAttempting to GET TWITCH Stats!" @@ -159,16 +149,11 @@ def time_calc(input_time): print("Header: ", twitch_header) print("JSON Full Response: ", twitch_followers_json) - twitch_username = twitch_followers_json["data"][0]["to_name"] - print("Username: ", twitch_username) twitch_followers = twitch_followers_json["total"] print("Followers: ", twitch_followers) - print("Monotonic: ", time.monotonic()) # Board Up-Time seconds - - print("\nFinished!") + print("Finished!") print("Next Update in: ", time_calc(sleep_time)) print("===============================") - gc.collect() except (ValueError, RuntimeError) as e: print("Failed to get data, retrying\n", e)