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.

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 Rachio example with Rate Limit endpoints, fixed 1700 daily requests #196

Merged
merged 1 commit into from
Jun 30, 2024
Merged
Changes from all commits
Commits
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
262 changes: 144 additions & 118 deletions examples/wifi/expanded/requests_wifi_rachio_irrigation.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
# https://support.rachio.com/en_us/public-api-documentation-S1UydL1Fv
# https://rachio.readme.io/reference/getting-started
RACHIO_KEY = os.getenv("RACHIO_APIKEY")
RACHIO_PERSONID = os.getenv("RACHIO_PERSONID")

# Get WiFi details, ensure these are setup in settings.toml
ssid = os.getenv("CIRCUITPY_WIFI_SSID")
Expand Down Expand Up @@ -100,125 +101,150 @@ def _format_datetime(datetime):
print("Retrying in 10 seconds")
print("✅ Wifi!")

try:
print(" | Attempting to GET Rachio Authorization")
# RETREIVE PERSONID AND PASTE IT TO SETTINGS.TOML
if RACHIO_PERSONID is None or RACHIO_PERSONID == "":
try:
with requests.get(
url=RACHIO_SOURCE, headers=RACHIO_HEADER
) as rachio_response:
rachio_json = rachio_response.json()
except ConnectionError as e:
print("Connection Error:", e)
print("Retrying in 10 seconds")
print(" | ✅ Authorized")

rachio_id = rachio_json["id"]
if DEBUG:
print(" | | Person ID: ", rachio_id)
print(" | | This ID will be used for subsequent calls")
print("\nFull API GET URL: ", RACHIO_SOURCE)
print(rachio_json)

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

try:
print(" | Attempting to GET Rachio JSON")
try:
with requests.get(
url=RACHIO_PERSON_SOURCE + rachio_id, headers=RACHIO_HEADER
) as rachio_response:
rachio_json = rachio_response.json()
except ConnectionError as e:
print("Connection Error:", e)
print("Retrying in 10 seconds")
print(" | ✅ Rachio JSON")

rachio_id = rachio_json["id"]
rachio_id_ast = obfuscating_asterix(rachio_id, "append", 3)
print(" | | UserID: ", rachio_id_ast)

rachio_username = rachio_json["username"]
rachio_username_ast = obfuscating_asterix(rachio_username, "append", 3)
print(" | | Username: ", rachio_username_ast)

rachio_name = rachio_json["fullName"]
rachio_name_ast = obfuscating_asterix(rachio_name, "append", 3)
print(" | | Full Name: ", rachio_name_ast)

rachio_deleted = rachio_json["deleted"]
if not rachio_deleted:
print(" | | Account Status: Active")
else:
print(" | | Account Status?: Deleted!")

rachio_createdate = rachio_json["createDate"]
rachio_timezone_offset = rachio_json["devices"][0]["utcOffset"]
# Rachio Unix time is in milliseconds, convert to seconds
rachio_createdate_seconds = rachio_createdate // 1000
rachio_timezone_offset_seconds = rachio_timezone_offset // 1000
# Apply timezone offset in seconds
local_unix_time = rachio_createdate_seconds + rachio_timezone_offset_seconds
if DEBUG:
print(f" | | Unix Registration Date: {rachio_createdate}")
print(f" | | Unix Timezone Offset: {rachio_timezone_offset}")
current_struct_time = time.localtime(local_unix_time)
final_timestamp = "{}".format(_format_datetime(current_struct_time))
print(f" | | Registration Date: {final_timestamp}")

rachio_devices = rachio_json["devices"][0]["name"]
print(" | | Device: ", rachio_devices)

rachio_model = rachio_json["devices"][0]["model"]
print(" | | | Model: ", rachio_model)

rachio_serial = rachio_json["devices"][0]["serialNumber"]
rachio_serial_ast = obfuscating_asterix(rachio_serial, "append")
print(" | | | Serial Number: ", rachio_serial_ast)

rachio_mac = rachio_json["devices"][0]["macAddress"]
rachio_mac_ast = obfuscating_asterix(rachio_mac, "append")
print(" | | | MAC Address: ", rachio_mac_ast)

rachio_status = rachio_json["devices"][0]["status"]
print(" | | | Device Status: ", rachio_status)

rachio_timezone = rachio_json["devices"][0]["timeZone"]
print(" | | | Time Zone: ", rachio_timezone)

# Latitude & Longtitude are used for smart watering & rain delays
rachio_latitude = str(rachio_json["devices"][0]["latitude"])
rachio_lat_ast = obfuscating_asterix(rachio_latitude, "all")
print(" | | | Latitude: ", rachio_lat_ast)

rachio_longitude = str(rachio_json["devices"][0]["longitude"])
rachio_long_ast = obfuscating_asterix(rachio_longitude, "all")
print(" | | | Longitude: ", rachio_long_ast)

rachio_rainsensor = rachio_json["devices"][0]["rainSensorTripped"]
print(" | | | Rain Sensor: ", rachio_rainsensor)

rachio_zone0 = rachio_json["devices"][0]["zones"][0]["name"]
rachio_zone1 = rachio_json["devices"][0]["zones"][1]["name"]
rachio_zone2 = rachio_json["devices"][0]["zones"][2]["name"]
rachio_zone3 = rachio_json["devices"][0]["zones"][3]["name"]
zones = f"{rachio_zone0}, {rachio_zone1}, {rachio_zone2}, {rachio_zone3}"
print(f" | | | Zones: {zones}")

if DEBUG:
print(f"\nFull API GET URL: {RACHIO_PERSON_SOURCE+rachio_id}")
print(rachio_json)

print(" | Attempting to GET Rachio Authorization")
try:
with requests.get(
url=RACHIO_SOURCE, headers=RACHIO_HEADER
) as rachio_response:
rachio_json = rachio_response.json()
except ConnectionError as e:
print("Connection Error:", e)
print("Retrying in 10 seconds")
print(" | ✅ Authorized")

rachio_id = rachio_json["id"]
print("\nADD THIS 🔑 TO YOUR SETTINGS.TOML FILE!")
print(f'RACHIO_PERSONID = "{rachio_id}"')

if DEBUG:
print("\nFull API GET URL: ", RACHIO_SOURCE)
print(rachio_json)

except (ValueError, RuntimeError) as e:
print(f"Failed to GET data: {e}")
time.sleep(60)
break
print(
"\nThis script can only continue when a proper APIKey & PersonID is used."
)
print("\nFinished!")
print(f"Board Uptime: {time_calc(time.monotonic())}")
print(f"Next Update: {time_calc(SLEEP_TIME)}")
print("===============================")
time.sleep(SLEEP_TIME)

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

time.sleep(SLEEP_TIME)
# Main Script
if RACHIO_PERSONID is not None and RACHIO_PERSONID != "":
try:
print(" | Attempting to GET Rachio JSON")
try:
with requests.get(
url=RACHIO_PERSON_SOURCE + RACHIO_PERSONID, headers=RACHIO_HEADER
) as rachio_response:
rachio_json = rachio_response.json()
except ConnectionError as e:
print("Connection Error:", e)
print("Retrying in 10 seconds")
print(" | ✅ Rachio JSON")
response_headers = rachio_response.headers
if DEBUG:
print(f"Response Headers: {response_headers}")
call_limit = int(response_headers["x-ratelimit-limit"])
calls_remaining = int(response_headers["x-ratelimit-remaining"])
calls_made_today = call_limit - calls_remaining

print(" | | Headers:")
print(f" | | | Date: {response_headers['date']}")
print(f" | | | Maximum Daily Requests: {call_limit}")
print(f" | | | Today's Requests: {calls_made_today}")
print(f" | | | Remaining Requests: {calls_remaining}")
print(f" | | | Limit Reset: {response_headers['x-ratelimit-reset']}")
print(f" | | | Content Type: {response_headers['content-type']}")

rachio_id = rachio_json["id"]
rachio_id_ast = obfuscating_asterix(rachio_id, "append", 3)
print(" | | PersonID: ", rachio_id_ast)

rachio_username = rachio_json["username"]
rachio_username_ast = obfuscating_asterix(rachio_username, "append", 3)
print(" | | Username: ", rachio_username_ast)

rachio_name = rachio_json["fullName"]
rachio_name_ast = obfuscating_asterix(rachio_name, "append", 3)
print(" | | Full Name: ", rachio_name_ast)

rachio_deleted = rachio_json["deleted"]
if not rachio_deleted:
print(" | | Account Status: Active")
else:
print(" | | Account Status?: Deleted!")

rachio_createdate = rachio_json["createDate"]
rachio_timezone_offset = rachio_json["devices"][0]["utcOffset"]
# Rachio Unix time is in milliseconds, convert to seconds
rachio_createdate_seconds = rachio_createdate // 1000
rachio_timezone_offset_seconds = rachio_timezone_offset // 1000
# Apply timezone offset in seconds
local_unix_time = rachio_createdate_seconds + rachio_timezone_offset_seconds
if DEBUG:
print(f" | | Unix Registration Date: {rachio_createdate}")
print(f" | | Unix Timezone Offset: {rachio_timezone_offset}")
current_struct_time = time.localtime(local_unix_time)
final_timestamp = "{}".format(_format_datetime(current_struct_time))
print(f" | | Registration Date: {final_timestamp}")

rachio_devices = rachio_json["devices"][0]["name"]
print(" | | Device: ", rachio_devices)

rachio_model = rachio_json["devices"][0]["model"]
print(" | | | Model: ", rachio_model)

rachio_serial = rachio_json["devices"][0]["serialNumber"]
rachio_serial_ast = obfuscating_asterix(rachio_serial, "append")
print(" | | | Serial Number: ", rachio_serial_ast)

rachio_mac = rachio_json["devices"][0]["macAddress"]
rachio_mac_ast = obfuscating_asterix(rachio_mac, "append")
print(" | | | MAC Address: ", rachio_mac_ast)

rachio_status = rachio_json["devices"][0]["status"]
print(" | | | Device Status: ", rachio_status)

rachio_timezone = rachio_json["devices"][0]["timeZone"]
print(" | | | Time Zone: ", rachio_timezone)

# Latitude & Longtitude are used for smart watering & rain delays
rachio_latitude = str(rachio_json["devices"][0]["latitude"])
rachio_lat_ast = obfuscating_asterix(rachio_latitude, "all")
print(" | | | Latitude: ", rachio_lat_ast)

rachio_longitude = str(rachio_json["devices"][0]["longitude"])
rachio_long_ast = obfuscating_asterix(rachio_longitude, "all")
print(" | | | Longitude: ", rachio_long_ast)

rachio_rainsensor = rachio_json["devices"][0]["rainSensorTripped"]
print(" | | | Rain Sensor: ", rachio_rainsensor)

rachio_zone0 = rachio_json["devices"][0]["zones"][0]["name"]
rachio_zone1 = rachio_json["devices"][0]["zones"][1]["name"]
rachio_zone2 = rachio_json["devices"][0]["zones"][2]["name"]
rachio_zone3 = rachio_json["devices"][0]["zones"][3]["name"]
zones = f"{rachio_zone0}, {rachio_zone1}, {rachio_zone2}, {rachio_zone3}"
print(f" | | | Zones: {zones}")

if DEBUG:
print(f"\nFull API GET URL: {RACHIO_PERSON_SOURCE+rachio_id}")
print(rachio_json)

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(f"Failed to get data, retrying\n {e}")
time.sleep(60)
break

time.sleep(SLEEP_TIME)
Loading