394 lines
12 KiB
Python
394 lines
12 KiB
Python
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:
|
|
"""
|
|
Manual, per-controller update manager.
|
|
|
|
The manager never asks for a global/latest release. It polls only for a
|
|
command addressed to this controller. A command exists only after an
|
|
administrator presses Install Update for that controller in HomeCore Admin.
|
|
Failed updates are not retried automatically.
|
|
"""
|
|
|
|
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, request_id="", target_version="", detail=""):
|
|
key = (status, request_id, target_version, detail)
|
|
if key == self.last_report:
|
|
return
|
|
self.last_report = key
|
|
|
|
print(
|
|
f"HOMECORE UPDATE: {status} "
|
|
f"request={request_id or '-'} "
|
|
f"current={self.current_version} "
|
|
f"target={target_version or '-'} "
|
|
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={
|
|
"request_id": request_id,
|
|
"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 update_command(self):
|
|
state = self.load_controller_state()
|
|
controller_id = state.get("controller_id")
|
|
if not controller_id:
|
|
return None
|
|
|
|
token = self.controller_session_fn()
|
|
r = httpx.get(
|
|
self.cloud_api()
|
|
+ f"/api/v1/controllers/{controller_id}/update-command",
|
|
headers={"Authorization": "Bearer " + token},
|
|
timeout=15,
|
|
)
|
|
r.raise_for_status()
|
|
body = r.json()
|
|
return body.get("command") if isinstance(body, dict) else None
|
|
|
|
def mark_completed_if_required(self):
|
|
st = load_update_state()
|
|
request_id = str(st.get("request_id", ""))
|
|
target = str(st.get("target_version", ""))
|
|
|
|
if (
|
|
request_id
|
|
and target
|
|
and target == self.current_version
|
|
and st.get("status") == "installing"
|
|
):
|
|
self.report(
|
|
"completed",
|
|
request_id,
|
|
target,
|
|
"HomeCore Controller update completed successfully",
|
|
)
|
|
save_update_state(
|
|
{
|
|
**st,
|
|
"status": "completed",
|
|
"completed_at": int(time.time()),
|
|
}
|
|
)
|
|
|
|
def check_once(self):
|
|
command = self.update_command()
|
|
if not command:
|
|
return
|
|
|
|
request_id = str(command.get("request_id", "")).strip()
|
|
target = str(command.get("target_version", "")).strip()
|
|
if not request_id or not target:
|
|
raise RuntimeError("HomeCore Cloud returned an incomplete update command")
|
|
|
|
previous = load_update_state()
|
|
if (
|
|
str(previous.get("request_id", "")) == request_id
|
|
and previous.get("status") == "installing"
|
|
):
|
|
# Supervisor already accepted this exact request. Wait for the app
|
|
# restart rather than ever issuing the update a second time.
|
|
return
|
|
|
|
if target == self.current_version:
|
|
self.report(
|
|
"completed",
|
|
request_id,
|
|
target,
|
|
"Controller already has the requested HomeCore version",
|
|
)
|
|
save_update_state(
|
|
{
|
|
"status": "completed",
|
|
"request_id": request_id,
|
|
"target_version": target,
|
|
"completed_at": int(time.time()),
|
|
}
|
|
)
|
|
return
|
|
|
|
if version_tuple(target) < version_tuple(self.current_version):
|
|
raise RuntimeError(
|
|
"Automatic downgrade is not supported; controller has "
|
|
+ self.current_version
|
|
+ " and request targets "
|
|
+ target
|
|
)
|
|
|
|
self.report(
|
|
"preparing",
|
|
request_id,
|
|
target,
|
|
"Manual HomeCore Admin update request received",
|
|
)
|
|
|
|
supervisor_request("POST", "/store/reload")
|
|
time.sleep(3)
|
|
|
|
app = find_homecore_app()
|
|
slug = str(app.get("slug", ""))
|
|
# Home Assistant Supervisor reports:
|
|
# version = installed app version
|
|
# version_latest = newest version offered by the repository
|
|
#
|
|
# Always compare the Admin target against version_latest first.
|
|
repository_version = str(
|
|
app.get("version_latest")
|
|
or app.get("version_available")
|
|
or app.get("version")
|
|
or ""
|
|
)
|
|
|
|
print(
|
|
"HOMECORE UPDATE: repository check "
|
|
f"installed={app.get('version') or '-'} "
|
|
f"latest={app.get('version_latest') or '-'} "
|
|
f"target={target}",
|
|
flush=True,
|
|
)
|
|
|
|
installed_field = app.get("installed")
|
|
installed_version = (
|
|
installed_field
|
|
if isinstance(installed_field, str)
|
|
else self.current_version
|
|
)
|
|
|
|
if not slug:
|
|
raise RuntimeError("HomeCore App slug could not be determined")
|
|
|
|
if repository_version != target:
|
|
self.report(
|
|
"waiting_repository",
|
|
request_id,
|
|
target,
|
|
"Home Assistant repository currently offers "
|
|
+ (repository_version or "no version"),
|
|
)
|
|
save_update_state(
|
|
{
|
|
"status": "waiting_repository",
|
|
"request_id": request_id,
|
|
"target_version": target,
|
|
"repository_version": repository_version,
|
|
"checked_at": int(time.time()),
|
|
}
|
|
)
|
|
return
|
|
|
|
supervisor_request("GET", f"/store/addons/{slug}/availability")
|
|
|
|
state = {
|
|
"status": "installing",
|
|
"request_id": request_id,
|
|
"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",
|
|
request_id,
|
|
target,
|
|
"Backup requested; Supervisor update starting",
|
|
)
|
|
|
|
supervisor_request(
|
|
"POST",
|
|
f"/store/addons/{slug}/update",
|
|
{
|
|
"backup": True,
|
|
"background": True,
|
|
},
|
|
)
|
|
|
|
print(
|
|
"HOMECORE UPDATE: Supervisor accepted manual 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()
|
|
request_id = str(st.get("request_id", ""))
|
|
target = str(st.get("target_version", ""))
|
|
|
|
# If the failure happened before state was written, attempt to
|
|
# recover the active command once for accurate reporting.
|
|
if not request_id:
|
|
try:
|
|
cmd = self.update_command() or {}
|
|
request_id = str(cmd.get("request_id", ""))
|
|
target = str(cmd.get("target_version", ""))
|
|
except Exception:
|
|
pass
|
|
|
|
failed_state = {
|
|
**st,
|
|
"status": "failed",
|
|
"request_id": request_id,
|
|
"target_version": target,
|
|
"detail": str(exc),
|
|
"failed_at": int(time.time()),
|
|
}
|
|
save_update_state(failed_state)
|
|
self.report("failed", request_id, target, str(exc))
|
|
|
|
# No automatic retry. The cloud suppresses a failed request;
|
|
# Admin must explicitly create a new request by pressing Install.
|
|
|
|
time.sleep(30)
|
|
|
|
|
|
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()
|