#!/usr/bin/env python3 import os import sys import time import requests from datetime import datetime, timedelta # --- CONFIGURATION --- API_KEY = "nle_bcf8844884488448844884488448778bf5182e57d69709cabd" DEVICE_ID = "39c1cddf-2222-2222-2222-848484848484" SERIAL = "02AAAS3344556677GW" LOG_FILE_PATH = "hvac_history.log" HARDWARE_FAN_DELAY_SECS = 77 TIMEOUT_SECS = 10 MAX_RETRIES = 3 # Base URLs STATUS_URL = f"https://nolongerevil.com/api/v1/thermostat/{DEVICE_ID}/status" TEMP_CONTROL_URL = f"https://nolongerevil.com/api/v1/thermostat/{DEVICE_ID}/temperature" FAN_CONTROL_URL = f"https://nolongerevil.com/api/v1/thermostat/{DEVICE_ID}/fan" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } # Settable globally, but will auto-promote to True if a fault occurs DEBUG_MODE = "--debug" in sys.argv def log_debug(method, url, payload=None, response=None, exception=None): """Verbose structural dump highlighting full request-response payloads.""" print("\n" + "="*60) print(f"🔍 DEBUG ACTIVE | HTTP {method} -> {url}") print(f" Headers: {headers}") if payload: print(f" Sent Payload: {payload}") if response is not None: print(f" Status Code: {response.status_code}") print(f" Reason: {response.reason}") try: print(f" Full Response: {response.text}") except Exception: print(" Full Response: [Unreadable text payload]") if exception: print(f" 💥 Network Fault Context: {exception}") print("="*60 + "\n") def trigger_fault_protection(method, url, payload=None, response=None, exception=None): """Enforces immediate runtime escalation into verbose debug mode upon error.""" global DEBUG_MODE if not DEBUG_MODE: print("\n🚨 UNEXPECTED FAULT DETECTED! Automatically promoting session to global debug level...") DEBUG_MODE = True log_debug(method, url, payload, response, exception) def get_thermostat_state(): """Queries and returns target metrics. Promotes to debug on failure.""" try: response = requests.get(STATUS_URL, headers=headers, timeout=TIMEOUT_SECS) if DEBUG_MODE: log_debug("GET", STATUS_URL, response=response) if response.status_code == 200: data = response.json() state_data = data.get("state", {}) # Temperature parsing shared_bucket = state_data.get(f"shared.{SERIAL}", {}).get("value", {}) current_temp = shared_bucket.get("current_temperature") setpoint = shared_bucket.get("target_temperature") is_cooling = shared_bucket.get("hvac_ac_state", False) hvac_state = "cooling" if is_cooling else "idle" # Fan duration tracking parameter parsing device_bucket = state_data.get(f"device.{SERIAL}", {}).get("value", {}) humidity = device_bucket.get("current_humidity") active_fan_timer = device_bucket.get("fan_timer_duration", 0) return current_temp, humidity, setpoint, hvac_state, active_fan_timer else: trigger_fault_protection("GET", STATUS_URL, response=response) return None, None, None, None, None except Exception as e: trigger_fault_protection("GET", STATUS_URL, exception=e) return None, None, None, None, None def set_thermostat_temperature(target_value, mode_setting): """Sends temp shift command and handles internal retries via verification loops.""" payload = {"value": round(target_value, 2), "mode": mode_setting, "scale": "C"} for attempt in range(1, MAX_RETRIES + 1): try: print(f"Sending temperature adjustments (Attempt {attempt}/{MAX_RETRIES})...") response = requests.post(TEMP_CONTROL_URL, headers=headers, json=payload, timeout=TIMEOUT_SECS) if DEBUG_MODE: log_debug("POST", TEMP_CONTROL_URL, payload=payload, response=response) if response.status_code == 200 and response.json().get("success") is True: # --- READ BACK VERIFICATION --- time.sleep(2) # Allow cloud processing window to settle _, _, verify_setpoint, _, _ = get_thermostat_state() if verify_setpoint is not None and abs(verify_setpoint - target_value) <= 0.1: print(f"✅ Read-back verified: Thermostat target matches {target_value}°C") return True else: print(f"⚠️ Read-back mismatch! Expected {target_value}°C, got {verify_setpoint}°C.") else: trigger_fault_protection("POST", TEMP_CONTROL_URL, payload=payload, response=response) except Exception as e: trigger_fault_protection("POST", TEMP_CONTROL_URL, payload=payload, exception=e) time.sleep(3) return False def set_fan_duration(minutes_int): """Sends fan execution frame and checks read-back state timer properties.""" payload = {"duration": int(minutes_int)} for attempt in range(1, MAX_RETRIES + 1): try: print(f"Sending fan timing command frame (Attempt {attempt}/{MAX_RETRIES})...") response = requests.post(FAN_CONTROL_URL, headers=headers, json=payload, timeout=TIMEOUT_SECS) if DEBUG_MODE: log_debug("POST", FAN_CONTROL_URL, payload=payload, response=response) if response.status_code == 200 and response.json().get("success") is True: # --- READ BACK VERIFICATION --- time.sleep(3) _, _, _, _, current_fan_timer = get_thermostat_state() # Verify that the fan_timer_duration parameter updated to greater than zero if current_fan_timer is not None and current_fan_timer > 0: print(f"✅ Read-back verified: Fan control line active ({current_fan_timer}s remaining).") return True else: print("⚠️ Read-back mismatch! Thermostat fan tracking timer reads empty/idle.") else: trigger_fault_protection("POST", FAN_CONTROL_URL, payload=payload, response=response) except Exception as e: trigger_fault_protection("POST", FAN_CONTROL_URL, payload=payload, exception=e) time.sleep(4) return False def append_to_spreadsheet_log(requested_mode, current_temp, humidity, current_setpoint, current_status, new_setpoint, new_status): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") headers_line = ( "Timestamp\tRequestedMode\tCurrentObservedTemp(C)\tCurrentObservedHumidity(%)\t" "CurrentObservedSetpoint(C)\tCurrentObservedSystemStatus\tNewSetpoint(C)\tNewSystemStatus\n" ) file_exists = os.path.exists(LOG_FILE_PATH) try: with open(LOG_FILE_PATH, "a") as f: if not file_exists: f.write(headers_line) log_line = f"{timestamp}\t{requested_mode}\t{current_temp}\t{humidity}\t{current_setpoint}\t{current_status}\t{new_setpoint}\t{new_status}\n" f.write(log_line) except IOError as e: print(f"⚠️ Warning: Failed to write to spreadsheet history file: {e}") # --- MODE DETECTION --- script_name = os.path.basename(sys.argv[0]) if "chillout" in script_name: mode = "chillout" elif "unchill" in script_name: mode = "unchill" else: print(f"❌ Unknown execution name '{script_name}'. Use 'chillout' or 'unchill'.") sys.exit(1) # --- EXECUTION --- print(f"Starting execution in [{mode.upper()}] mode...") if DEBUG_MODE: print("🔔 Debugging logging explicitly requested via command line parameter.") print("Retrieving initial thermostat metrics via NLE API...") initial_temp, initial_humidity, initial_setpoint, initial_hvac_state, _ = get_thermostat_state() if initial_temp is None: print("Could not verify device telemetry baseline. Aborting.") sys.exit(1) print(f"Initial metrics: Ambient={initial_temp}°C | Humidity={initial_humidity}% | Setpoint={initial_setpoint}°C | State={initial_hvac_state}") # ===================================================================== # CHILLOUT MODE # ===================================================================== if mode == "chillout": target_trigger = initial_temp - 3.0 print(f"Dropping target setpoint down to {target_trigger}°C to force trigger AC...") set_thermostat_temperature(target_trigger, mode_setting="cool") print("Monitoring HVAC activation state (waiting for AC to start)...") while True: time.sleep(15) _, _, _, loops_state, _ = get_thermostat_state() if loops_state is None: continue print(f"Current HVAC activity: {loops_state}") if loops_state == "cooling": print("❄️ AC engagement detected!") break final_target = initial_temp - 0.5 print(f"Re-adjusting final target threshold to: {final_target}°C") set_thermostat_temperature(final_target, mode_setting="cool") _, _, final_setpoint, final_state, _ = get_thermostat_state() append_to_spreadsheet_log("chillout", initial_temp, initial_humidity, initial_setpoint, initial_hvac_state, final_setpoint, final_state) # ===================================================================== # UNCHILL MODE # ===================================================================== elif mode == "unchill": target_trigger = initial_temp + 3.0 print(f"Raising target setpoint up to {target_trigger}°C to force-stop AC...") set_thermostat_temperature(target_trigger, mode_setting="cool") print("Waiting for AC to spin down to idle...") while True: time.sleep(15) _, _, _, loops_state, _ = get_thermostat_state() if loops_state is None: continue print(f"Current HVAC activity: {loops_state}") if loops_state == "idle": print("🛑 AC call has been dropped.") break _, _, final_setpoint, final_state, _ = get_thermostat_state() append_to_spreadsheet_log("unchill", initial_temp, initial_humidity, initial_setpoint, initial_hvac_state, final_setpoint, final_state) keep_fan = input("Do you want to keep the fan running? (y/n) [default: n]: ").strip().lower() if keep_fan == 'y': user_input = input("For how many minutes of absolute physical runtime? [default: 20]: ").strip() requested_minutes = int(user_input) if user_input.isdigit() else 20 total_requested_secs = requested_minutes * 60 api_target_seconds = total_requested_secs - HARDWARE_FAN_DELAY_SECS api_minutes_integer = max(1, round(api_target_seconds / 60.0)) print(f"🌀 Adjusting API command to {api_minutes_integer} whole minutes to compensate for mechanical tail lag...") now = datetime.now() actual_runtime_seconds = (api_minutes_integer * 60) + HARDWARE_FAN_DELAY_SECS expected_fan_stop_time = now + timedelta(seconds=actual_runtime_seconds) if set_fan_duration(api_minutes_integer): print(f"✅ Calling for fan.") print(f" Thermostat signal call initiated at: {now.strftime('%I:%M:%S %p')}") print(f" Physical fan expected to stop at: {expected_fan_stop_time.strftime('%I:%M:%S %p')}") else: print("❌ Verification loop failed. The thermostat rejected the fan duration command.") else: print("No longer calling for fan.") print("Process completed successfully.")