from pathlib import Path import json import os import re import threading import time import httpx UPDATE_STATE = Path("/config/homecore_update.json") SUPERVISOR = "http://supervisor" def version_tuple(value): parts = re.findall(r"\d+", str(value)) return tuple(int(x) for x in parts[:4]) def load_update_state(): try: return json.loads(UPDATE_STATE.read_text()) except Exception: return {} def save_update_state(data): UPDATE_STATE.write_text( json.dumps(data, indent=2) ) def supervisor_headers(): token = os.getenv("SUPERVISOR_TOKEN") if not token: raise RuntimeError( "SUPERVISOR_TOKEN is not available" ) return { "Authorization": "Bearer " + token, "Content-Type": "application/json", } def supervisor_request(method, path, payload=None, timeout=90): r = httpx.request( method, SUPERVISOR + path, headers=supervisor_headers(), json=payload, timeout=timeout, ) r.raise_for_status() try: body = r.json() except Exception: return {} if isinstance(body, dict): if body.get("result") == "error": raise RuntimeError( body.get("message") or "Supervisor error" ) if "data" in body: return body["data"] return body def find_homecore_app(): data = supervisor_request( "GET", "/store/addons", ) if isinstance(data, dict): apps = data.get("addons", []) else: apps = data or [] candidates = [] for app in apps: slug = str(app.get("slug", "")) name = str(app.get("name", "")) if ( name == "HomeCore Controller" or slug == "homecore_controller" or slug.endswith("_homecore_controller") ): candidates.append(app) if not candidates: raise RuntimeError( "HomeCore Controller was not found in the Home Assistant App store" ) installed = [ x for x in candidates if x.get("installed") ] return (installed or candidates)[0] class HomeCoreUpdateManager: def __init__( self, current_version, cloud_api_fn, load_controller_state, controller_session_fn, ): self.current_version = current_version self.cloud_api_fn = cloud_api_fn self.load_controller_state = load_controller_state self.controller_session_fn = controller_session_fn self.last_report = None def cloud_api(self): return self.cloud_api_fn()["cloud_api"].rstrip("/") def report( self, status, target_version="", detail="", ): key = ( status, target_version, detail, ) if key == self.last_report: return self.last_report = key print( f"HOMECORE UPDATE: {status} " f"current={self.current_version} " f"target={target_version} " f"{detail}", flush=True, ) state = self.load_controller_state() controller_id = state.get("controller_id") if not controller_id: return try: token = self.controller_session_fn() httpx.post( self.cloud_api() + f"/api/v1/controllers/{controller_id}/update-status", headers={ "Authorization": "Bearer " + token }, json={ "status": status, "target_version": target_version, "current_version": self.current_version, "detail": detail, }, timeout=15, ).raise_for_status() except Exception as exc: print( "HOMECORE UPDATE: status report failed:", exc, flush=True, ) def desired_release(self): r = httpx.get( self.cloud_api() + "/api/v1/releases/latest", params={ "channel": "stable", }, timeout=15, ) if r.status_code == 404: return None r.raise_for_status() return r.json() def mark_completed_if_required(self): st = load_update_state() target = st.get("target_version", "") if ( target and target == self.current_version and st.get("status") == "installing" ): self.report( "completed", target, "HomeCore Controller update completed successfully", ) save_update_state( { "status": "completed", "target_version": target, "completed_at": int(time.time()), } ) def check_once(self): desired = self.desired_release() if not desired: return target = str( desired.get("version", "") ).strip() if not target: return if version_tuple(target) <= version_tuple( self.current_version ): self.report( "up_to_date", target, "Controller matches the approved HomeCore release", ) return previous = load_update_state() if ( previous.get("target_version") == target and previous.get("status") in ( "installing", "failed", ) ): attempted = int( previous.get("attempted_at", 0) ) # Avoid update loops. if time.time() - attempted < 600: return self.report( "checking", target, "Approved HomeCore release detected", ) supervisor_request( "POST", "/store/reload", ) time.sleep(3) app = find_homecore_app() slug = app.get("slug") repository_version = str( app.get("version") or app.get("version_latest") or "" ) installed_version = str( app.get("installed") or self.current_version ) if not slug: raise RuntimeError( "HomeCore App slug could not be determined" ) if repository_version != target: self.report( "waiting_repository", target, "Home Assistant repository currently offers " + (repository_version or "no version"), ) return supervisor_request( "GET", f"/store/addons/{slug}/availability", ) state = { "status": "installing", "target_version": target, "installed_version": installed_version, "repository_version": repository_version, "slug": slug, "attempted_at": int(time.time()), } save_update_state(state) self.report( "installing", target, "Backup requested; Supervisor update starting", ) supervisor_request( "POST", f"/store/addons/{slug}/update", { "backup": True, "background": True, }, ) print( "HOMECORE UPDATE: Supervisor accepted update job", flush=True, ) def run(self): self.mark_completed_if_required() while True: try: self.check_once() except Exception as exc: st = load_update_state() target = st.get( "target_version", "", ) save_update_state( { **st, "status": "failed", "target_version": target, "detail": str(exc), "attempted_at": int(time.time()), } ) self.report( "failed", target, str(exc), ) time.sleep(60) def start_update_manager( current_version, cloud_api_fn, load_controller_state, controller_session_fn, ): manager = HomeCoreUpdateManager( current_version=current_version, cloud_api_fn=cloud_api_fn, load_controller_state=load_controller_state, controller_session_fn=controller_session_fn, ) threading.Thread( target=manager.run, daemon=True, name="homecore-update-manager", ).start()