homecoreadmin/homecore-apps
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HomeCore Controller v0.1.5 manual per-controller updates

Browse Source
This commit is contained in:
HomeCore 2026-09-16 12:31:39 +00:00
parent 5d35543444
commit 9778115297
4 changed files with 135 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
homecore_controller/CHANGELOG.md
View File

@ -1,5 +1,14 @@
# HomeCore Controller Changelog
## 0.1.5
- Changed HomeCore updates to manual per-controller deployment from HomeCore Admin.
- Removed global automatic release polling.
- Controllers now accept only update commands explicitly addressed to their controller ID.
- Failed updates do not retry automatically.
- Home Assistant backup remains requested before each controller update.
- Preserves the 0.1.4 Home Assistant Ingress commissioning fix.
## 0.1.4
- Fixed Home Assistant Ingress commissioning form routing.

2
homecore_controller/app/main.py
View File

@ -16,7 +16,7 @@ from cryptography.hazmat.primitives import serialization
import uvicorn
from updater import start_update_manager
VERSION = "0.1.4"
VERSION = "0.1.5"
DATA = Path("/config")
DATA.mkdir(parents=True, exist_ok=True)

277
homecore_controller/app/updater.py
View File

@ -9,7 +9,6 @@ import httpx
UPDATE_STATE = Path("/config/homecore_update.json")
SUPERVISOR = "http://supervisor"
@ -26,19 +25,13 @@ def load_update_state():
def save_update_state(data):
UPDATE_STATE.write_text(
json.dumps(data, indent=2)
)
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"
)
raise RuntimeError("SUPERVISOR_TOKEN is not available")
return {
"Authorization": "Bearer " + token,
"Content-Type": "application/json",
@ -53,7 +46,6 @@ def supervisor_request(method, path, payload=None, timeout=90):
json=payload,
timeout=timeout,
)
r.raise_for_status()
try:
@ -63,10 +55,7 @@ def supervisor_request(method, path, payload=None, timeout=90):
if isinstance(body, dict):
if body.get("result") == "error":
raise RuntimeError(
body.get("message") or "Supervisor error"
)
raise RuntimeError(body.get("message") or "Supervisor error")
if "data" in body:
return body["data"]
@ -74,10 +63,7 @@ def supervisor_request(method, path, payload=None, timeout=90):
def find_homecore_app():
data = supervisor_request(
"GET",
"/store/addons",
)
data = supervisor_request("GET", "/store/addons")
if isinstance(data, dict):
apps = data.get("addons", [])
@ -85,11 +71,9 @@ def find_homecore_app():
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"
@ -102,15 +86,19 @@ def find_homecore_app():
"HomeCore Controller was not found in the Home Assistant App store"
)
installed = [
x for x in candidates
if x.get("installed")
]
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,
@ -123,56 +111,39 @@ class HomeCoreUpdateManager:
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,
)
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} "
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
},
headers={"Authorization": "Bearer " + token},
json={
"request_id": request_id,
"status": status,
"target_version": target_version,
"current_version": self.current_version,
@ -180,159 +151,159 @@ class HomeCoreUpdateManager:
},
timeout=15,
).raise_for_status()
except Exception as exc:
print(
"HOMECORE UPDATE: status report failed:",
exc,
flush=True,
)
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:
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()
return r.json()
body = r.json()
return body.get("command") if isinstance(body, dict) else None
def mark_completed_if_required(self):
st = load_update_state()
target = st.get("target_version", "")
request_id = str(st.get("request_id", ""))
target = str(st.get("target_version", ""))
if (
target
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",
"target_version": target,
"completed_at": int(time.time()),
}
)
def check_once(self):
desired = self.desired_release()
if not desired:
command = self.update_command()
if not command:
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
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 (
previous.get("target_version") == target
and previous.get("status") in (
"installing",
"failed",
)
str(previous.get("request_id", "")) == request_id
and previous.get("status") == "installing"
):
attempted = int(
previous.get("attempted_at", 0)
)
# Supervisor already accepted this exact request. Wait for the app
# restart rather than ever issuing the update a second time.
return
# Avoid update loops.
if time.time() - attempted < 600:
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(
"checking",
"preparing",
request_id,
target,
"Approved HomeCore release detected",
)
supervisor_request(
"POST",
"/store/reload",
"Manual HomeCore Admin update request received",
)
supervisor_request("POST", "/store/reload")
time.sleep(3)
app = find_homecore_app()
slug = app.get("slug")
slug = str(app.get("slug", ""))
repository_version = str(
app.get("version")
or app.get("version_latest")
or app.get("version_available")
or ""
)
installed_version = str(
app.get("installed")
or self.current_version
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"
)
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",
)
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",
)
@ -347,46 +318,46 @@ class HomeCoreUpdateManager:
)
print(
"HOMECORE UPDATE: Supervisor accepted update job",
"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", ""))
target = 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
save_update_state(
{
**st,
"status": "failed",
"target_version": target,
"detail": str(exc),
"attempted_at": int(time.time()),
}
)
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))
self.report(
"failed",
target,
str(exc),
)
# No automatic retry. The cloud suppresses a failed request;
# Admin must explicitly create a new request by pressing Install.
time.sleep(60)
time.sleep(30)
def start_update_manager(

2
homecore_controller/config.yaml
View File

@ -1,5 +1,5 @@
name: HomeCore Controller
version: "0.1.4"
version: "0.1.5"
slug: homecore_controller
description: HomeCore local controller agent for Home Assistant Green.
url: https://homecore.home-core.co.uk