Add Rachio Irrigation API Example

examples/wifi/expanded/requests_wifi_rachio_irrigation.py

@@ -0,0 +1,224 @@
+# SPDX-FileCopyrightText: 2024 DJDevon3
+# SPDX-License-Identifier: MIT
+# Coded for Circuit Python 9.x
+"""Rachio Irrigation Timer API Example"""
+
+import os
+import time
+
+import adafruit_connection_manager
+import wifi
+
+import adafruit_requests
+
+# Rachio API Key required (comes with purchase of a device)
+# API is rate limited to 1700 calls per day.
+# https://support.rachio.com/en_us/public-api-documentation-S1UydL1Fv
+# https://rachio.readme.io/reference/getting-started
+RACHIO_KEY = os.getenv("RACHIO_APIKEY")
+
+# Get WiFi details, ensure these are setup in settings.toml
+ssid = os.getenv("CIRCUITPY_WIFI_SSID")
+password = os.getenv("CIRCUITPY_WIFI_PASSWORD")
+
+# API Polling Rate
+# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour
+SLEEP_TIME = 900
+
+# Set debug to True for full JSON response.
+# WARNING: absolutely shows extremely sensitive personal information & credentials
+# Including your real name, latitude, longitude, account id, mac address, etc...
+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)
+
+RACHIO_HEADER = {"Authorization": " Bearer " + RACHIO_KEY}
+RACHIO_SOURCE = "https://api.rach.io/1/public/person/info/"
+RACHIO_PERSON_SOURCE = "https://api.rach.io/1/public/person/"
+
+
+def obfuscating_asterix(obfuscate_object, direction, characters=2):
+    """
+    Obfuscates a string with asterisks except for a specified number of characters.
+    param object: str The string to obfuscate with asterisks
+    param direction: str Option either 'prepend', 'append', or 'all' direction
+    param characters: int The number of characters to keep unobfuscated (default is 2)
+    """
+    object_len = len(obfuscate_object)
+    if direction not in {"prepend", "append", "all"}:
+        raise ValueError("Invalid direction. Use 'prepend', 'append', or 'all'.")
+    if characters >= object_len and direction != "all":
+        # If characters greater than or equal to string length,
+        # return the original string as it can't be obfuscated.
+        return obfuscate_object
+    asterix_replace = "*" * object_len
+    if direction == "append":
+        asterix_final = obfuscate_object[:characters] + "*" * (object_len - characters)
+    elif direction == "prepend":
+        asterix_final = "*" * (object_len - characters) + obfuscate_object[-characters:]
+    elif direction == "all":
+        # Replace all characters with asterisks
+        asterix_final = asterix_replace
+
+    return asterix_final
+
+
+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"
+
+
+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}"
+    )
+
+
+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 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"]
+        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("\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)