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")
RACHIO_PERSONID = os.getenv("RACHIO_PERSONID")

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

    # RETREIVE PERSONID AND PASTE IT TO SETTINGS.TOML
    if RACHIO_PERSONID is None or RACHIO_PERSONID == "":
        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"]
            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("===============================")
        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 = f"{_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)