Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Update Twitch API Example with Connection Manager #170

Merged
merged 1 commit into from
Mar 18, 2024
Merged

Update Twitch API Example with Connection Manager #170

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 99 additions & 83 deletions examples/wifi/expanded/requests_wifi_api_twitch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,44 +1,49 @@
# SPDX-FileCopyrightText: 2023 DJDevon3
# SPDX-FileCopyrightText: 2024 DJDevon3
# SPDX-License-Identifier: MIT
# Coded for Circuit Python 8.2.x
# Twitch_API_Example
"""Twitch API Example"""
# pylint: disable=import-error

import os
import ssl
import time

import socketpool
import adafruit_connection_manager
import wifi

import adafruit_requests

# 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 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",
# Ensure these are in settings.toml
# TWITCH_CLIENT_ID = "Your Developer APP ID Here"
# TWITCH_CLIENT_SECRET = "APP ID secret here"
# TWITCH_USER_ID = "Your Twitch UserID here"

# Get WiFi details, ensure these are setup in settings.toml
ssid = os.getenv("CIRCUITPY_WIFI_SSID")
password = os.getenv("CIRCUITPY_WIFI_PASSWORD")
twitch_client_id = os.getenv("Twitch_ClientID")
twitch_client_secret = os.getenv("Twitch_Client_Secret")
TWITCH_CID = os.getenv("TWITCH_CLIENT_ID")
TWITCH_CS = os.getenv("TWITCH_CLIENT_SECRET")
# For finding your Twitch User ID
# https://www.streamweasels.com/tools/convert-twitch-username-to-user-id/
twitch_user_id = os.getenv("Twitch_UserID") # User ID you want endpoints from
TWITCH_UID = os.getenv("TWITCH_USER_ID")

# Time between API refreshes
# API Polling Rate
# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour
sleep_time = 900
SLEEP_TIME = 900

# Set DEBUG to True for full JSON response.
# STREAMER WARNING: Credentials will be viewable
DEBUG = False

# Initalize Wifi, Socket Pool, Request Session
pool = adafruit_connection_manager.get_radio_socketpool(wifi.radio)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(wifi.radio)
requests = adafruit_requests.Session(pool, ssl_context)


# Converts seconds to minutes/hours/days
def time_calc(input_time):
"""Converts seconds to minutes/hours/days"""
if input_time < 60:
return f"{input_time:.0f} seconds"
if input_time < 3600:
Expand All @@ -48,42 +53,44 @@ def time_calc(input_time):
return f"{input_time / 60 / 60 / 24:.1f} days"


def _format_datetime(datetime):
"""F-String formatted struct time conversion"""
return (
f"{datetime.tm_mon:02}/"
+ f"{datetime.tm_mday:02}/"
+ f"{datetime.tm_year:02} "
+ f"{datetime.tm_hour:02}:"
+ f"{datetime.tm_min:02}:"
+ f"{datetime.tm_sec:02}"
)


# First we use Client ID & Client Secret to create a token with POST
# No user interaction is required for this type of scope (implicit grant flow)
twitch_0auth_header = {"Content-Type": "application/x-www-form-urlencoded"}
TWITCH_0AUTH_TOKEN = "https://id.twitch.tv/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.connected:
try:
wifi.radio.connect(ssid, password)
except ConnectionError as e:
print("Connection Error:", e)
print("Retrying in 10 seconds")
time.sleep(10)
print("Connected!\n")

while True:
# Connect to Wi-Fi
print("\nConnecting to WiFi...")
while not wifi.radio.ipv4_address:
try:
wifi.radio.connect(ssid, password)
except ConnectionError as e:
print("❌ Connection Error:", e)
print("Retrying in 10 seconds")
print("✅ Wifi!")

try:
# ----------------------------- POST FOR BEARER TOKEN -----------------------
print(
"Attempting Bearer Token Request!"
) # ---------------------------------------
# Print Request to Serial
debug_bearer_request = (
False # STREAMER WARNING: your client secret will be viewable
)
if debug_bearer_request:
print("Full API GET URL: ", TWITCH_0AUTH_TOKEN)
print("===============================")
# ------------- POST FOR BEARER TOKEN -----------------
print(" | Attempting Bearer Token Request!")
if DEBUG:
print(f"Full API GET URL: {TWITCH_0AUTH_TOKEN}")
twitch_0auth_data = (
"&client_id="
+ twitch_client_id
+ TWITCH_CID
+ "&client_secret="
+ twitch_client_secret
+ TWITCH_CS
+ "&grant_type=client_credentials"
)

Expand All @@ -95,70 +102,79 @@ def time_calc(input_time):
twitch_0auth_json = twitch_0auth_response.json()
twitch_access_token = twitch_0auth_json["access_token"]
except ConnectionError as e:
print("Connection Error:", e)
print(f"Connection Error: {e}")
print("Retrying in 10 seconds")
print(" | 🔑 Token Authorized!")

# Print Response to Serial
debug_bearer_response = (
False # STREAMER WARNING: your client secret will be viewable
)
if debug_bearer_response:
print("JSON Dump: ", twitch_0auth_json)
print("Header: ", twitch_0auth_header)
print("Access Token: ", twitch_access_token)
# STREAMER WARNING: your client secret will be viewable
if DEBUG:
print(f"JSON Dump: {twitch_0auth_json}")
print(f"Header: {twitch_0auth_header}")
print(f"Access Token: {twitch_access_token}")
twitch_token_type = twitch_0auth_json["token_type"]
print("Token Type: ", twitch_token_type)
print(f"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))
print(f" | Token Expires in: {time_calc(twitch_token_expiration)}")

# ----------------------------- GET DATA -------------------------------------
# ----------------------------- GET DATA --------------------
# Bearer token is refreshed every time script runs :)
# Twitch sets token expiration to about 64 days
# Helix is the name of the current Twitch API
# Now that we have POST bearer token we can do a GET for data
# ----------------------------------------------------------------------------
# -----------------------------------------------------------
twitch_header = {
"Authorization": "Bearer " + twitch_access_token + "",
"Client-Id": "" + twitch_client_id + "",
"Client-Id": "" + TWITCH_CID + "",
}
TWITCH_FOLLOWERS_SOURCE = (
"https://api.twitch.tv/helix/channels"
+ "/followers?"
+ "broadcaster_id="
+ twitch_user_id
+ TWITCH_UID
)
print(
"\nAttempting to GET TWITCH Stats!"
) # ------------------------------------------------
print("===============================")
twitch_followers_response = requests.get(
print(" | Attempting to GET Twitch JSON!")
twitch_response = requests.get(
url=TWITCH_FOLLOWERS_SOURCE, headers=twitch_header
)
try:
twitch_followers_json = twitch_followers_response.json()
twitch_json = twitch_response.json()
except ConnectionError as e:
print("Connection Error:", e)
print(f"Connection Error: {e}")
print("Retrying in 10 seconds")

# Print Response to Serial
debug_bearer_response = (
False # STREAMER WARNING: your bearer token will be viewable
)
if debug_bearer_response:
print("Full API GET URL: ", TWITCH_FOLLOWERS_SOURCE)
print("Header: ", twitch_header)
print("JSON Full Response: ", twitch_followers_json)

twitch_followers = twitch_followers_json["total"]
print("Followers: ", twitch_followers)
print("Finished!")
print("Next Update in: ", time_calc(sleep_time))
if DEBUG:
print(f" | Full API GET URL: {TWITCH_FOLLOWERS_SOURCE}")
print(f" | Header: {twitch_header}")
print(f" | JSON Full Response: {twitch_json}")

if "status" in twitch_json:
twitch_error_status = twitch_json["status"]
print(f"❌ Status: {twitch_error_status}")

if "error" in twitch_json:
twitch_error = twitch_json["error"]
print(f"❌ Error: {twitch_error}")

if "message" in twitch_json:
twitch_error_msg = twitch_json["message"]
print(f"❌ Message: {twitch_error_msg}")

if "total" in twitch_json:
print(" | ✅ Twitch JSON!")
twitch_followers = twitch_json["total"]
print(f" | | Followers: {twitch_followers}")

twitch_response.close()
print("✂️ Disconnected from Twitch API")

print("\nFinished!")
print(f"Board Uptime: {time_calc(time.monotonic())}")
print(f"Next Update: {time_calc(SLEEP_TIME)}")
print("===============================")

except (ValueError, RuntimeError) as e:
print("Failed to get data, retrying\n", e)
print(f"Failed to get data, retrying\n {e}")
time.sleep(60)
continue
time.sleep(sleep_time)
break
time.sleep(SLEEP_TIME)
Loading