From 9368287c12f0f6b4f80e3f5445394c8cf63e199e Mon Sep 17 00:00:00 2001 From: ltadeu6 Date: Wed, 1 Apr 2026 14:55:16 -0300 Subject: [PATCH] Improve AC widget responsiveness --- waybar/air_control.py | 189 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 163 insertions(+), 26 deletions(-) diff --git a/waybar/air_control.py b/waybar/air_control.py index decedc1..e20a103 100755 --- a/waybar/air_control.py +++ b/waybar/air_control.py @@ -1,9 +1,12 @@ #!/run/current-system/sw/bin/python3 -import requests import sys import os import json +import time +import subprocess +import urllib.request +import urllib.error HA_URL = "http://localhost:8123" ENTITY = "climate.ar" @@ -12,6 +15,10 @@ TOKEN_FILE = os.path.expanduser("~/.config/secrets/ha_token") STEP = 1 MIN_TEMP = 18 MAX_TEMP = 30 +PENDING_FILE = os.path.expanduser("~/.cache/waybar_air_pending.json") +SEND_DELAY = 3 +PENDING_TTL = 15 +CONFIRM_TTL = 20 def get_token(): @@ -23,52 +30,146 @@ HEADERS = {"Authorization": f"Bearer {get_token()}", "Content-Type": "applicatio def get_state(): - r = requests.get(f"{HA_URL}/api/states/{ENTITY}", headers=HEADERS) - return r.json() + return request_json("GET", f"/api/states/{ENTITY}") def set_temperature(temp): - requests.post( - f"{HA_URL}/api/services/climate/set_temperature", - headers=HEADERS, - json={"entity_id": ENTITY, "temperature": temp}, + request_json( + "POST", + "/api/services/climate/set_temperature", + {"entity_id": ENTITY, "temperature": temp}, ) def set_mode(mode): - requests.post( - f"{HA_URL}/api/services/climate/set_hvac_mode", - headers=HEADERS, - json={"entity_id": ENTITY, "hvac_mode": mode}, + request_json( + "POST", + "/api/services/climate/set_hvac_mode", + {"entity_id": ENTITY, "hvac_mode": mode}, ) +def request_json(method, path, payload=None): + url = f"{HA_URL}{path}" + data = None + headers = dict(HEADERS) + if payload is not None: + data = json.dumps(payload).encode("utf-8") + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=10) as resp: + body = resp.read() + except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, ValueError): + return None + if not body: + return None + try: + return json.loads(body) + except json.JSONDecodeError: + return None + + +def read_pending(): + try: + with open(PENDING_FILE) as f: + data = json.load(f) + temp = data.get("temperature") + ts = data.get("ts") + sent = data.get("sent", False) + sent_ts = data.get("sent_ts") + if isinstance(temp, (int, float)) and isinstance(ts, (int, float)): + sent_ts_val = None + if isinstance(sent_ts, (int, float)): + sent_ts_val = float(sent_ts) + return { + "temperature": float(temp), + "ts": float(ts), + "sent": bool(sent), + "sent_ts": sent_ts_val, + } + except FileNotFoundError: + return None + except (OSError, ValueError, json.JSONDecodeError): + return None + return None + + +def write_pending(temp, sent=False, sent_ts=None, ts=None): + os.makedirs(os.path.dirname(PENDING_FILE), exist_ok=True) + if ts is None: + ts = time.time() + payload = {"temperature": float(temp), "ts": float(ts), "sent": bool(sent)} + if sent_ts is not None: + payload["sent_ts"] = float(sent_ts) + with open(PENDING_FILE, "w") as f: + json.dump(payload, f) + + +def clear_pending(): + try: + os.remove(PENDING_FILE) + except FileNotFoundError: + pass + except OSError: + pass + + def display(): state = get_state() + if not state: + print(json.dumps({"text": "-- - ", "class": "off"})) + return mode = state["state"] attrs = state["attributes"] - current = attrs["current_temperature"] + current = attrs.get("current_temperature") target = attrs.get("temperature") + pending = read_pending() + if pending is not None: + pending_temp = pending["temperature"] + pending_ts = pending["ts"] + now = time.time() + show_pending = now - pending_ts <= PENDING_TTL + if pending["sent"] and pending.get("sent_ts") is not None: + show_pending = show_pending or (now - pending["sent_ts"] <= CONFIRM_TTL) + if target == pending_temp: + clear_pending() + elif show_pending: + target = pending_temp + else: + clear_pending() + + if current is None: + current = 0 if mode == "off": - print(json.dumps({"text": f"{current:.0f} - ", "class": "off"})) + print( + json.dumps( + { + "text": f" {current:.0f} - ", + "class": "off", + } + ) + ) return - if mode == "cool": - icon = "" - css_class = "cool" - elif mode == "heat": - icon = "" - css_class = "heat" - else: - icon = "" - css_class = "other" + mode_map = { + "cool": ("", "cool"), + "heat": ("", "heat"), + "dry": ("", "dry"), + "fan_only": ("󰈐", "fan"), + } + icon, css_class = mode_map.get(mode, ("", "other")) + if target is None: + target_text = f"{current:.0f}" + else: + target_text = f"{target:.0f}" print( json.dumps( { - "text": f"{icon} {target:.0f} - ", + "text": f"{icon} {target_text} - ", "class": css_class, } ) @@ -77,11 +178,45 @@ def display(): def change(delta): state = get_state() - attrs = state["attributes"] - current_target = attrs["temperature"] + if state: + attrs = state["attributes"] + current_target = attrs.get("temperature") + if current_target is None: + current_target = attrs.get("current_temperature") + else: + current_target = None + pending = read_pending() + if pending is not None: + current_target = pending["temperature"] + if current_target is None: + return new_temp = max(MIN_TEMP, min(MAX_TEMP, current_target + delta)) - set_temperature(new_temp) + write_pending(new_temp, sent=False) + subprocess.Popen( + [sys.executable, os.path.abspath(__file__), "flush"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def flush_pending(): + pending = read_pending() + if pending is None: + return + ts = pending["ts"] + remaining = SEND_DELAY - (time.time() - ts) + if remaining > 0: + time.sleep(remaining) + pending = read_pending() + if pending is None: + return + temp2 = pending["temperature"] + ts2 = pending["ts"] + if ts2 != ts: + return + set_temperature(temp2) + write_pending(temp2, sent=True, sent_ts=time.time(), ts=ts2) def main(): @@ -101,6 +236,8 @@ def main(): change(STEP) elif cmd == "down": change(-STEP) + elif cmd == "flush": + flush_pending() if __name__ == "__main__":