requests_wifi_rachio_irrigation.py

# 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)