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

Update YouTube API Example with Connection Manager #174

Merged
merged 2 commits into from
Apr 1, 2024
Merged

Update YouTube API Example with Connection Manager #174

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1def0b6
Update YouTube API Example with Connection Manager
DJDevon3 Mar 24, 2024
fdc0da7
added developer dashboard url comment
DJDevon3 Mar 24, 2024
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
166 changes: 82 additions & 84 deletions examples/wifi/expanded/requests_wifi_api_youtube.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,123 +1,121 @@
# SPDX-FileCopyrightText: 2022 DJDevon3 for Adafruit Industries
# SPDX-FileCopyrightText: 2024 DJDevon3
# SPDX-License-Identifier: MIT
# Coded for Circuit Python 8.0
"""DJDevon3 Adafruit Feather ESP32-S2 YouTube_API_Example"""
import gc
import json
# Coded for Circuit Python 8.2.x
"""YouTube API Subscriber Count Example"""
# pylint: disable=import-error

import os
import ssl
import time

import socketpool
import adafruit_connection_manager
import wifi

import adafruit_requests

# Ensure these are uncommented and in secrets.py or .env
# "YT_username": "Your YouTube Username",
# "YT_token" : "Your long API developer token",

# Initialize WiFi Pool (There can be only 1 pool & top of script)
pool = socketpool.SocketPool(wifi.radio)
# 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)

# 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.
# WARNING: Will show credentials
DEBUG = False

# Ensure these are uncommented and in settings.toml
# YOUTUBE_USERNAME = "Your YouTube Username",
# YOUTUBE_TOKEN = "Your long API developer token",

# Get WiFi details, ensure these are setup in settings.toml
ssid = os.getenv("CIRCUITPY_WIFI_SSID")
password = os.getenv("CIRCUITPY_WIFI_PASSWORD")
yt_username = os.getenv("YT_username")
yt_token = os.getenv("YT_token")


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"
# Requires YouTube/Google API key
# https://console.cloud.google.com/apis/dashboard
YT_USERNAME = os.getenv("YOUTUBE_USERNAME")
YT_TOKEN = os.getenv("YOUTUBE_TOKEN")


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:
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"


# https://youtube.googleapis.com/youtube/v3/channels?part=statistics&forUsername=[YOUR_USERNAME]&key=[YOUR_API_KEY]
YT_SOURCE = (
"https://youtube.googleapis.com/youtube/v3/channels?"
+ "part=statistics"
+ "&forUsername="
+ yt_username
YOUTUBE_SOURCE = (
"https://youtube.googleapis.com/youtube/v3/channels?part=statistics&forUsername="
+ str(YT_USERNAME)
+ "&key="
+ yt_token
+ str(YT_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(ssid, password)
except ConnectionError as e:
print("Connection Error:", e)
print("Retrying in 10 seconds")
time.sleep(10)
gc.collect()
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:
print("Attempting to GET YouTube Stats!") # ----------------------------------
debug_request = False # Set true to see full request
if debug_request:
print("Full API GET URL: ", YT_SOURCE)
print("===============================")
print(" | Attempting to GET YouTube JSON...")
try:
response = requests.get(YT_SOURCE).json()
youtube_response = requests.get(url=YOUTUBE_SOURCE)
youtube_json = youtube_response.json()
except ConnectionError as e:
print("Connection Error:", e)
print("Retrying in 10 seconds")
print(" | ✅ YouTube JSON!")

if DEBUG:
print(f" | Full API GET URL: {YOUTUBE_SOURCE}")
print(f" | Full API Dump: {youtube_json}")

# Print Full JSON to Serial
debug_response = False # Set true to see full response
if debug_response:
dump_object = json.dumps(response)
print("JSON Dump: ", dump_object)
# Key:Value RESPONSES
if "pageInfo" in youtube_json:
totalResults = youtube_json["pageInfo"]["totalResults"]
print(f" | | Matching Results: {totalResults}")

# Print to Serial
yt_debug_keys = True # Set to True to print Serial data
if yt_debug_keys:
print("Matching Results: ", response["pageInfo"]["totalResults"])
if "items" in youtube_json:
YT_request_kind = youtube_json["items"][0]["kind"]
print(f" | | Request Kind: {YT_request_kind}")

YT_request_kind = response["items"][0]["kind"]
print("Request Kind: ", YT_request_kind)
YT_channel_id = youtube_json["items"][0]["id"]
print(f" | | Channel ID: {YT_channel_id}")

YT_response_kind = response["kind"]
print("Response Kind: ", YT_response_kind)
YT_videoCount = youtube_json["items"][0]["statistics"]["videoCount"]
print(f" | | Videos: {YT_videoCount}")

YT_channel_id = response["items"][0]["id"]
print("Channel ID: ", YT_channel_id)
YT_viewCount = youtube_json["items"][0]["statistics"]["viewCount"]
print(f" | | Views: {YT_viewCount}")

YT_videoCount = response["items"][0]["statistics"]["videoCount"]
print("Videos: ", YT_videoCount)
YT_subsCount = youtube_json["items"][0]["statistics"]["subscriberCount"]
print(f" | | Subscribers: {YT_subsCount}")

YT_viewCount = response["items"][0]["statistics"]["viewCount"]
print("Views: ", YT_viewCount)
if "kind" in youtube_json:
YT_response_kind = youtube_json["kind"]
print(f" | | Response Kind: {YT_response_kind}")

YT_subsCount = response["items"][0]["statistics"]["subscriberCount"]
print("Subscribers: ", YT_subsCount)
print("Monotonic: ", time.monotonic())
youtube_response.close()
print("✂️ Disconnected from YouTube API")

print("\nFinished!")
print("Next Update in %s %s" % (int(sleep_int), sleep_time_conversion))
print(f"Board Uptime: {time_calc(time.monotonic())}")
print(f"Next Update: {time_calc(SLEEP_TIME)}")
print("===============================")
gc.collect()

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