from pathlib import Path
import os
import json
import base64
import hashlib
import time
import threading
import secrets
import html
import httpx
from fastapi import FastAPI, Form, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
import uvicorn
from updater import start_update_manager
VERSION = "1.0.1"
print(
"HOMECORE SUPERVISOR: token "
+ ("available" if os.getenv("SUPERVISOR_TOKEN") else "missing"),
flush=True,
)
DATA = Path("/config")
DATA.mkdir(parents=True, exist_ok=True)
STATE = DATA / "state.json"
KEY = DATA / "controller_ed25519.pem"
OPTIONS = Path("/data/options.json")
PENDING = {}
def options():
try:
return json.loads(OPTIONS.read_text())
except Exception:
return {"cloud_api": "https://api.homecore.home-core.co.uk"}
def ensure_key():
if KEY.exists():
return serialization.load_pem_private_key(KEY.read_bytes(), password=None)
k = Ed25519PrivateKey.generate()
KEY.write_bytes(
k.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
)
return k
KEYOBJ = ensure_key()
def pub_pem():
return KEYOBJ.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
).decode()
def fingerprint():
raw = KEYOBJ.public_key().public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
)
return hashlib.sha256(raw).hexdigest()
def load_state():
try:
return json.loads(STATE.read_text())
except Exception:
return {}
def save_state(s):
STATE.write_text(json.dumps(s, indent=2))
def supervisor_get(path):
token = os.getenv("SUPERVISOR_TOKEN")
if not token:
return None
try:
r = httpx.get(
"http://supervisor" + path,
headers={"Authorization": "Bearer " + token},
timeout=5,
)
r.raise_for_status()
return r.json()
except Exception:
return None
def ha_version():
info = supervisor_get("/core/info") or {}
if isinstance(info, dict):
return ((info.get("data") or {}).get("version") or "")
return ""
def cloud_detail(response):
try:
data = response.json()
if isinstance(data, dict):
return data.get("detail") or str(data)
except Exception:
pass
return f"HomeCore Cloud returned HTTP {response.status_code}"
def controller_session():
s = load_state()
if not s.get("controller_id"):
return None
api = options()["cloud_api"].rstrip("/")
with httpx.Client(timeout=15) as c:
ch = c.get(f"{api}/api/v1/controllers/{s['controller_id']}/challenge")
ch.raise_for_status()
challenge = ch.json()["challenge"]
sig = base64.b64encode(
KEYOBJ.sign(challenge.encode())
).decode()
r = c.post(
f"{api}/api/v1/controllers/{s['controller_id']}/session",
json={
"challenge": challenge,
"signature_b64": sig,
},
)
r.raise_for_status()
return r.json()["controller_token"]
def heartbeat_loop():
while True:
try:
s = load_state()
if s.get("controller_id"):
token = controller_session()
api = options()["cloud_api"].rstrip("/")
httpx.post(
f"{api}/api/v1/controllers/{s['controller_id']}/heartbeat",
headers={"Authorization": "Bearer " + token},
json={
"software_version": VERSION,
"ha_version": ha_version(),
},
timeout=10,
)
except Exception:
pass
time.sleep(60)
# ===== HomeCore 1.0 product layer =====
from pathlib import Path
from datetime import datetime, timedelta, timezone
from urllib.parse import quote
import json
import os
import time
import secrets
import threading
import uuid
import httpx
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse, Response
CORE_API = "http://supervisor/core/api"
DEVICE_DOMAINS = {
"light", "climate", "water_heater", "camera", "cover",
"alarm_control_panel", "media_player", "switch", "lock",
"event", "binary_sensor", "fan", "scene", "script",
"automation", "button", "remote", "vacuum",
}
CATEGORY_LABELS = {
"lighting": "Lighting",
"heating": "Heating",
"cctv": "CCTV",
"blinds": "Blinds",
"alarm": "Alarm",
"doorbell": "Intercom",
"entertainment": "Entertainment",
"security": "Security",
"switch": "Switches",
"fan": "Fans",
"scene": "Scenes",
"automation": "Automations",
"other": "Other",
}
SESSIONS = {}
LOCK = threading.RLock()
APP_HTML = '\n\n\n\n\nHomeCore\n\n\n\n
HomeCore
Loading home…
\n\n
HomeCore
Settings
\n
\n
◉
HomeCore Intercom
Someone is at the door
\n'
def build_router(version, data_dir, load_state, options, ha_version):
router = APIRouter()
product_path = data_dir / "homecore_product.json"
activity_path = data_dir / "homecore_activity.json"
def read_json(path, default):
try:
return json.loads(path.read_text())
except Exception:
return default
def atomic_json(path, value):
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
os.replace(tmp, path)
def default_product():
return {
"schema": 1,
"home_name": "",
"theme": "system",
"rooms": [],
"devices": {},
}
def load_product():
current = read_json(product_path, default_product())
base = default_product()
if isinstance(current, dict):
base.update(current)
if not isinstance(base.get("rooms"), list):
base["rooms"] = []
if not isinstance(base.get("devices"), dict):
base["devices"] = {}
return base
def save_product(value):
atomic_json(product_path, value)
def activity(event_type, detail=None, actor="local"):
with LOCK:
rows = read_json(activity_path, [])
if not isinstance(rows, list):
rows = []
rows.insert(
0,
{
"id": uuid.uuid4().hex,
"type": event_type,
"detail": detail or {},
"actor": actor,
"created_at": datetime.now(timezone.utc).isoformat(),
},
)
atomic_json(activity_path, rows[:500])
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 ha_request(method, path, payload=None, params=None, timeout=20):
r = httpx.request(
method,
CORE_API + path,
headers=supervisor_headers(),
json=payload,
params=params,
timeout=timeout,
)
r.raise_for_status()
if not r.content:
return None
return r.json()
def ha_binary(path, timeout=30):
r = httpx.get(
CORE_API + path,
headers=supervisor_headers(),
timeout=timeout,
)
r.raise_for_status()
return r
def ingress_context(request):
user_id = request.headers.get("x-remote-user-id")
if not user_id:
return None
return {
"mode": "ingress",
"owner": True,
"permissions": ["*"],
"user": {
"id": user_id,
"display_name": request.headers.get(
"x-remote-user-display-name",
request.headers.get("x-remote-user-name", "Home Assistant"),
),
},
}
def local_session(request):
sid = request.cookies.get("hc_session") or request.headers.get("x-homecore-session")
if not sid:
auth = request.headers.get("authorization", "")
if auth.lower().startswith("bearer "):
sid = auth.split(" ", 1)[1]
value = SESSIONS.get(sid or "")
if not value:
return None
if value["expires"] < time.time():
SESSIONS.pop(sid, None)
return None
return {"sid": sid, **value, "mode": "homecore"}
def auth_context(request, required=True):
local = local_session(request)
if local:
prop = local.get("property") or {}
return {
"mode": "homecore",
"owner": bool(prop.get("owner")),
"permissions": list(prop.get("permissions") or []),
"user": local.get("user") or {},
"session": local,
}
ingress = ingress_context(request)
if ingress:
return ingress
if required:
raise HTTPException(401, "HomeCore sign in required")
return None
def cloud_headers(ctx):
session = (ctx or {}).get("session") or {}
token = session.get("cloud_token")
if not token:
raise HTTPException(
401,
"Connect a HomeCore account to manage cloud access",
)
return {"Authorization": "Bearer " + token}
def permission_words(category):
return {
"lighting": ["light", "lighting"],
"heating": ["heat", "heating", "climate"],
"cctv": ["camera", "cctv"],
"blinds": ["cover", "blind", "blinds"],
"alarm": ["alarm", "security"],
"doorbell": ["doorbell", "intercom"],
"entertainment": ["media", "entertainment", "tv"],
"security": ["lock", "security"],
"switch": ["switch"],
"fan": ["fan"],
"scene": ["scene"],
"automation": ["automation"],
}.get(category, [category])
def can_control(ctx, category):
if ctx.get("owner") or ctx.get("mode") == "ingress":
return True
perms = [str(x).lower() for x in ctx.get("permissions", [])]
if "*" in perms:
return True
if any(
p in {
"control", "control_all", "control_devices",
"devices.control", "home.control",
}
for p in perms
):
return True
words = permission_words(category)
return any(
any(w in p for w in words)
and ("control" in p or "manage" in p or p in words)
for p in perms
)
def can_view(ctx, category):
if can_control(ctx, category):
return True
if ctx.get("owner") or ctx.get("mode") == "ingress":
return True
perms = [str(x).lower() for x in ctx.get("permissions", [])]
if any(
p in {"view", "view_all", "view_devices", "home.view"}
for p in perms
):
return True
words = permission_words(category)
return any(
any(w in p for w in words)
and ("view" in p or "read" in p)
for p in perms
) or bool(perms)
def category_for(entity_id, attrs):
domain = entity_id.split(".", 1)[0]
device_class = str(attrs.get("device_class") or "").lower()
if domain == "light":
return "lighting"
if domain in ("climate", "water_heater"):
return "heating"
if domain == "camera":
return "cctv"
if domain == "cover":
return "blinds"
if domain == "alarm_control_panel":
return "alarm"
if domain == "media_player":
return "entertainment"
if domain == "lock":
return "security"
if domain == "switch":
return "switch"
if domain == "fan":
return "fan"
if domain == "scene":
return "scene"
if domain in ("script", "automation"):
return "automation"
if domain == "event" and device_class == "doorbell":
return "doorbell"
if domain == "binary_sensor":
label = (
entity_id + " " + str(attrs.get("friendly_name") or "")
).lower()
if device_class == "doorbell" or "doorbell" in label:
return "doorbell"
if domain == "button" and "doorbell" in entity_id.lower():
return "doorbell"
if domain == "remote":
return "entertainment"
return "other"
def capabilities(entity_id, attrs):
domain = entity_id.split(".", 1)[0]
if domain == "light":
caps = ["power"]
if "brightness" in attrs or attrs.get("supported_color_modes"):
caps.append("brightness")
modes = attrs.get("supported_color_modes") or []
if any(x in modes for x in ("rgb", "rgbw", "rgbww", "hs", "xy")):
caps.append("color")
if any(x in modes for x in ("color_temp", "color_temp_kelvin")):
caps.append("color_temperature")
return caps
if domain in ("climate", "water_heater"):
return ["temperature", "mode"]
if domain == "cover":
caps = ["open", "close", "stop"]
if attrs.get("current_position") is not None:
caps.append("position")
return caps
if domain == "alarm_control_panel":
return ["arm_home", "arm_away", "arm_night", "disarm"]
if domain == "media_player":
caps = ["power", "play_pause", "volume"]
if attrs.get("source_list"):
caps.append("source")
return caps
if domain == "switch":
return ["power"]
if domain == "lock":
return ["lock", "unlock"]
if domain == "fan":
return ["power", "percentage"]
if domain == "scene":
return ["activate"]
if domain == "script":
return ["run"]
if domain == "automation":
return ["enable", "trigger"]
if domain == "button":
return ["press"]
if domain == "remote":
return ["power"]
if domain == "vacuum":
return ["start", "pause", "stop", "return_home"]
if domain == "camera":
return ["view"]
return []
def normalize(raw, product):
entity_id = str(raw.get("entity_id") or "")
if "." not in entity_id:
return None
domain = entity_id.split(".", 1)[0]
if domain not in DEVICE_DOMAINS:
return None
attrs = raw.get("attributes") or {}
category = category_for(entity_id, attrs)
if category == "other" and domain in ("binary_sensor", "event"):
return None
override = (product.get("devices") or {}).get(entity_id, {})
if override.get("hidden"):
return None
name = (
override.get("name")
or attrs.get("friendly_name")
or entity_id.split(".", 1)[1].replace("_", " ").title()
)
item = {
"id": entity_id,
"entity_id": entity_id,
"domain": domain,
"category": category,
"category_label": CATEGORY_LABELS.get(category, "Other"),
"name": name,
"state": raw.get("state"),
"room_id": override.get("room_id") or "",
"favorite": bool(override.get("favorite")),
"available": raw.get("state") not in ("unavailable", "unknown"),
"last_changed": raw.get("last_changed"),
"last_updated": raw.get("last_updated"),
"capabilities": capabilities(entity_id, attrs),
"attributes": {},
}
for key in (
"brightness", "rgb_color", "color_temp_kelvin",
"current_temperature", "temperature", "target_temp_high",
"target_temp_low", "hvac_action", "hvac_modes",
"current_position", "current_tilt_position", "volume_level",
"is_volume_muted", "media_title", "media_artist", "source",
"source_list", "app_name", "friendly_name", "device_class",
"event_type", "event_types", "percentage", "preset_mode",
"preset_modes",
):
if key in attrs:
item["attributes"][key] = attrs[key]
if domain == "camera":
item["camera_url"] = (
"/api/v1/camera/" + quote(entity_id, safe="")
)
return item
def get_devices(ctx=None):
product = load_product()
states = ha_request("GET", "/states") or []
out = []
for raw in states:
item = normalize(raw, product)
if not item:
continue
if ctx and not can_view(ctx, item["category"]):
continue
out.append(item)
out.sort(
key=lambda x: (
x["room_id"],
x["category"],
x["name"].lower(),
)
)
return out
def overview(devices):
lighting = [x for x in devices if x["category"] == "lighting"]
heating = [x for x in devices if x["category"] == "heating"]
media = [x for x in devices if x["category"] == "entertainment"]
alarms = [x for x in devices if x["category"] == "alarm"]
locks = [x for x in devices if x["category"] == "security"]
cameras = [x for x in devices if x["category"] == "cctv"]
temps = []
for item in heating:
value = item["attributes"].get("current_temperature")
if isinstance(value, (int, float)):
temps.append(float(value))
return {
"lights_on": sum(1 for x in lighting if x["state"] == "on"),
"lights_total": len(lighting),
"average_temperature": (
round(sum(temps) / len(temps), 1) if temps else None
),
"media_active": sum(
1
for x in media
if x["state"] in ("on", "playing", "paused")
),
"cameras": len(cameras),
"alarm_state": alarms[0]["state"] if alarms else None,
"unlocked": sum(
1 for x in locks if x["state"] == "unlocked"
),
}
def find_device(entity_id, ctx):
for item in get_devices(ctx):
if item["id"] == entity_id:
return item
raise HTTPException(404, "Home Assistant entity not found")
def safe_service(domain, command, body):
mapping = {
"light": {
"turn_on": "turn_on",
"turn_off": "turn_off",
"toggle": "toggle",
},
"climate": {
"set_temperature": "set_temperature",
"set_mode": "set_hvac_mode",
"turn_on": "turn_on",
"turn_off": "turn_off",
},
"water_heater": {
"set_temperature": "set_temperature",
"set_mode": "set_operation_mode",
"turn_on": "turn_on",
"turn_off": "turn_off",
},
"cover": {
"open": "open_cover",
"close": "close_cover",
"stop": "stop_cover",
"set_position": "set_cover_position",
},
"alarm_control_panel": {
"arm_home": "alarm_arm_home",
"arm_away": "alarm_arm_away",
"arm_night": "alarm_arm_night",
"disarm": "alarm_disarm",
},
"media_player": {
"turn_on": "turn_on",
"turn_off": "turn_off",
"play_pause": "media_play_pause",
"play": "media_play",
"pause": "media_pause",
"next": "media_next_track",
"previous": "media_previous_track",
"volume": "volume_set",
"mute": "volume_mute",
"source": "select_source",
},
"switch": {
"turn_on": "turn_on",
"turn_off": "turn_off",
"toggle": "toggle",
},
"lock": {"lock": "lock", "unlock": "unlock"},
"fan": {
"turn_on": "turn_on",
"turn_off": "turn_off",
"toggle": "toggle",
"percentage": "set_percentage",
},
"scene": {"activate": "turn_on"},
"script": {"run": "turn_on"},
"automation": {
"enable": "turn_on",
"disable": "turn_off",
"trigger": "trigger",
},
"button": {"press": "press"},
"remote": {
"turn_on": "turn_on",
"turn_off": "turn_off",
},
"vacuum": {
"start": "start",
"pause": "pause",
"stop": "stop",
"return_home": "return_to_base",
},
}
service = mapping.get(domain, {}).get(command)
if not service:
raise HTTPException(
400,
f"Unsupported {domain} command: {command}",
)
data = {"entity_id": body["entity_id"]}
allowed = {
"brightness_pct", "rgb_color", "color_temp_kelvin",
"temperature", "hvac_mode", "operation_mode", "position",
"code", "volume_level", "is_volume_muted", "source",
"percentage", "preset_mode", "skip_condition",
}
aliases = {
"mode": "hvac_mode",
"cover_position": "position",
"volume": "volume_level",
}
for key, value in body.items():
target = aliases.get(key, key)
if target in allowed:
data[target] = value
return service, data
@router.post("/api/v1/session/login")
async def session_login(request: Request):
body = await request.json()
email = str(body.get("email") or "").strip().lower()
password = str(body.get("password") or "")
if not email or not password:
raise HTTPException(400, "Email and password are required")
api = options()["cloud_api"].rstrip("/")
with httpx.Client(timeout=20) as client:
response = client.post(
api + "/api/v1/auth/login",
json={"email": email, "password": password},
)
if response.status_code >= 400:
raise HTTPException(
401,
"Incorrect HomeCore email address or password",
)
login = response.json()
token = login["access_token"]
headers = {"Authorization": "Bearer " + token}
props = client.get(
api + "/api/v1/properties",
headers=headers,
)
if props.status_code >= 400:
raise HTTPException(
props.status_code,
"HomeCore Cloud rejected property access",
)
property_id = load_state().get("property_id")
prop = next(
(x for x in props.json() if x.get("id") == property_id),
None,
)
if not prop:
raise HTTPException(
403,
"This HomeCore account does not have access to this home",
)
sid = secrets.token_urlsafe(32)
SESSIONS[sid] = {
"cloud_token": token,
"user": login.get("user") or {"email": email},
"property": prop,
"expires": time.time() + (7 * 24 * 3600),
}
response = JSONResponse(
{
"ok": True,
"user": login.get("user"),
"owner": bool(prop.get("owner")),
"permissions": prop.get("permissions") or [],
}
)
response.set_cookie(
"hc_session",
sid,
httponly=True,
samesite="lax",
max_age=7 * 24 * 3600,
secure=False,
)
return response
@router.post("/api/v1/session/logout")
def session_logout(request: Request):
session = local_session(request)
if session:
SESSIONS.pop(session["sid"], None)
response = JSONResponse({"ok": True})
response.delete_cookie("hc_session")
return response
@router.get("/api/v1/home")
def home(request: Request):
ctx = auth_context(request)
product = load_product()
devices = get_devices(ctx)
state = load_state()
home_name = (
product.get("home_name")
or state.get("property_name")
or "HomeCore Home"
)
rooms = sorted(
product.get("rooms", []),
key=lambda x: (
x.get("order", 999),
x.get("name", "").lower(),
),
)
return {
"version": version,
"ha_version": ha_version(),
"controller_id": state.get("controller_id"),
"property_id": state.get("property_id"),
"home_name": home_name,
"theme": product.get("theme", "system"),
"rooms": rooms,
"devices": devices,
"overview": overview(devices),
"activity": read_json(activity_path, [])[:60],
"user": ctx.get("user"),
"owner": bool(ctx.get("owner")),
"permissions": ctx.get("permissions", []),
}
@router.get("/api/v1/devices")
def devices(request: Request):
ctx = auth_context(request)
return {"devices": get_devices(ctx)}
@router.get("/api/v1/devices/{entity_id:path}")
def device(entity_id: str, request: Request):
ctx = auth_context(request)
return find_device(entity_id, ctx)
@router.post("/api/v1/devices/{entity_id:path}/command")
async def device_command(entity_id: str, request: Request):
ctx = auth_context(request)
item = find_device(entity_id, ctx)
if not can_control(ctx, item["category"]):
raise HTTPException(
403,
"Your HomeCore permissions do not allow this control",
)
body = await request.json()
command = str(body.pop("command", "")).strip()
body["entity_id"] = entity_id
service, data = safe_service(
item["domain"],
command,
body,
)
result = ha_request(
"POST",
f"/services/{item['domain']}/{service}",
payload=data,
)
actor = (
(ctx.get("user") or {}).get("email")
or (ctx.get("user") or {}).get("display_name")
or "local"
)
activity(
"device.command",
{
"entity_id": entity_id,
"name": item["name"],
"category": item["category"],
"command": command,
},
actor=actor,
)
return {"ok": True, "changed": result or []}
@router.post("/api/v1/rooms/{room_id}/command")
async def room_command(room_id: str, request: Request):
ctx = auth_context(request)
body = await request.json()
category = str(body.get("category") or "")
command = str(body.get("command") or "")
targets = [
x for x in get_devices(ctx)
if x["room_id"] == room_id and x["category"] == category
]
if not can_control(ctx, category):
raise HTTPException(
403,
"Your HomeCore permissions do not allow this control",
)
count = 0
for item in targets:
try:
service, data = safe_service(
item["domain"],
command,
{"entity_id": item["id"]},
)
ha_request(
"POST",
f"/services/{item['domain']}/{service}",
payload=data,
)
count += 1
except Exception:
pass
activity(
"room.command",
{
"room_id": room_id,
"category": category,
"command": command,
"count": count,
},
)
return {"ok": True, "count": count}
@router.get("/api/v1/history/{entity_id:path}")
def history(entity_id: str, request: Request, hours: int = 24):
ctx = auth_context(request)
find_device(entity_id, ctx)
hours = max(1, min(hours, 168))
start = datetime.now(timezone.utc) - timedelta(hours=hours)
raw = ha_request(
"GET",
"/history/period/" + quote(
start.isoformat(),
safe=":TZ+-",
),
params={
"filter_entity_id": entity_id,
"minimal_response": "",
"no_attributes": "",
"significant_changes_only": "",
},
timeout=30,
) or []
rows = (
raw[0]
if raw and isinstance(raw[0], list)
else []
)
return {
"entity_id": entity_id,
"hours": hours,
"points": [
{
"state": x.get("state"),
"last_changed": x.get("last_changed"),
}
for x in rows[-500:]
],
}
@router.get("/api/v1/activity")
def local_activity(request: Request):
auth_context(request)
return {
"activity": read_json(activity_path, [])[:200]
}
@router.get("/api/v1/camera/{entity_id:path}")
def camera(entity_id: str, request: Request):
ctx = auth_context(request)
item = find_device(entity_id, ctx)
if item["category"] != "cctv":
raise HTTPException(400, "Entity is not a camera")
if not can_view(ctx, "cctv"):
raise HTTPException(403, "Camera permission required")
response = ha_binary(
"/camera_proxy/" + quote(entity_id, safe="")
)
return Response(
content=response.content,
media_type=response.headers.get(
"content-type",
"image/jpeg",
),
headers={"Cache-Control": "no-store"},
)
def require_owner(ctx):
if not ctx.get("owner") and ctx.get("mode") != "ingress":
raise HTTPException(403, "Owner permission required")
@router.post("/api/v1/settings")
async def settings(request: Request):
ctx = auth_context(request)
require_owner(ctx)
body = await request.json()
product = load_product()
if "home_name" in body:
name = str(body.get("home_name") or "").strip()
if len(name) > 80:
raise HTTPException(400, "Home name is too long")
product["home_name"] = name
if body.get("theme") in ("system", "dark", "light"):
product["theme"] = body["theme"]
save_product(product)
activity(
"settings.updated",
{"home_name": product.get("home_name")},
)
return {
"ok": True,
"home_name": product.get("home_name"),
"theme": product.get("theme"),
}
@router.post("/api/v1/rooms")
async def create_room(request: Request):
ctx = auth_context(request)
require_owner(ctx)
body = await request.json()
name = str(body.get("name") or "").strip()
if not name:
raise HTTPException(400, "Room name is required")
if len(name) > 60:
raise HTTPException(400, "Room name is too long")
product = load_product()
room = {
"id": "room_" + uuid.uuid4().hex[:12],
"name": name,
"icon": str(body.get("icon") or "⌂")[:8],
"order": len(product["rooms"]),
}
product["rooms"].append(room)
save_product(product)
activity(
"room.created",
{"room_id": room["id"], "name": room["name"]},
)
return room
@router.put("/api/v1/rooms/{room_id}")
async def update_room(room_id: str, request: Request):
ctx = auth_context(request)
require_owner(ctx)
body = await request.json()
product = load_product()
room = next(
(
x
for x in product["rooms"]
if x.get("id") == room_id
),
None,
)
if not room:
raise HTTPException(404, "Room not found")
if "name" in body:
name = str(body.get("name") or "").strip()
if not name:
raise HTTPException(
400,
"Room name is required",
)
room["name"] = name[:60]
if "icon" in body:
room["icon"] = str(
body.get("icon") or "⌂"
)[:8]
if "order" in body:
room["order"] = int(body["order"])
save_product(product)
activity(
"room.updated",
{"room_id": room_id, "name": room["name"]},
)
return room
@router.delete("/api/v1/rooms/{room_id}")
def delete_room(room_id: str, request: Request):
ctx = auth_context(request)
require_owner(ctx)
product = load_product()
before = len(product["rooms"])
product["rooms"] = [
x
for x in product["rooms"]
if x.get("id") != room_id
]
if len(product["rooms"]) == before:
raise HTTPException(404, "Room not found")
for settings in product["devices"].values():
if settings.get("room_id") == room_id:
settings["room_id"] = ""
save_product(product)
activity("room.deleted", {"room_id": room_id})
return {"ok": True}
@router.post("/api/v1/device-settings")
async def device_settings(request: Request):
ctx = auth_context(request)
require_owner(ctx)
body = await request.json()
entity_id = str(
body.get("entity_id") or ""
).strip()
if not entity_id or "." not in entity_id:
raise HTTPException(
400,
"Valid entity_id is required",
)
product = load_product()
value = dict(
product["devices"].get(entity_id) or {}
)
if "room_id" in body:
room_id = str(body.get("room_id") or "")
if room_id and not any(
x.get("id") == room_id
for x in product["rooms"]
):
raise HTTPException(
400,
"Room does not exist",
)
value["room_id"] = room_id
if "name" in body:
value["name"] = str(
body.get("name") or ""
).strip()[:80]
if "favorite" in body:
value["favorite"] = bool(
body["favorite"]
)
if "hidden" in body:
value["hidden"] = bool(
body["hidden"]
)
product["devices"][entity_id] = value
save_product(product)
activity(
"device.settings",
{"entity_id": entity_id, **value},
)
return {
"ok": True,
"entity_id": entity_id,
"settings": value,
}
@router.get("/api/v1/cloud/access")
def cloud_access(request: Request):
ctx = auth_context(request)
headers = cloud_headers(ctx)
api = options()["cloud_api"].rstrip("/")
property_id = load_state().get("property_id")
with httpx.Client(timeout=20) as client:
props = client.get(
api + "/api/v1/properties",
headers=headers,
)
props.raise_for_status()
prop = next(
(
x
for x in props.json()
if x.get("id") == property_id
),
None,
)
if not prop:
raise HTTPException(
403,
"Account no longer has access to this home",
)
members = []
if prop.get("owner"):
result = client.get(
f"{api}/api/v1/properties/{property_id}/members",
headers=headers,
)
if result.status_code < 400:
members = result.json()
user = ctx.get("user") or {}
return {
"owner": bool(prop.get("owner")),
"email": user.get("email", ""),
"permissions": prop.get("permissions") or [],
"members": members,
}
@router.get("/api/v1/cloud/permission-options")
def cloud_permission_options(request: Request):
ctx = auth_context(request)
headers = cloud_headers(ctx)
api = options()["cloud_api"].rstrip("/")
response = httpx.get(
api + "/api/v1/permissions",
headers=headers,
timeout=15,
)
response.raise_for_status()
return response.json()
@router.post("/api/v1/cloud/share")
async def cloud_share(request: Request):
ctx = auth_context(request)
headers = cloud_headers(ctx)
body = await request.json()
api = options()["cloud_api"].rstrip("/")
property_id = load_state().get("property_id")
response = httpx.post(
f"{api}/api/v1/properties/{property_id}/share",
headers=headers,
json={
"email": str(
body.get("email") or ""
).strip().lower(),
"permissions": list(
body.get("permissions") or []
),
},
timeout=20,
)
if response.status_code >= 400:
try:
detail = response.json().get("detail")
except Exception:
detail = response.text
raise HTTPException(
response.status_code,
detail or "HomeCore sharing failed",
)
activity(
"sharing.updated",
{
"email": body.get("email"),
"permissions": body.get("permissions"),
},
)
return response.json()
@router.delete("/api/v1/cloud/members/{user_id}")
def cloud_revoke(user_id: str, request: Request):
ctx = auth_context(request)
headers = cloud_headers(ctx)
api = options()["cloud_api"].rstrip("/")
property_id = load_state().get("property_id")
response = httpx.delete(
f"{api}/api/v1/properties/{property_id}/members/{user_id}",
headers=headers,
timeout=20,
)
if response.status_code >= 400:
try:
detail = response.json().get("detail")
except Exception:
detail = response.text
raise HTTPException(
response.status_code,
detail or "HomeCore access removal failed",
)
activity(
"sharing.revoked",
{"user_id": user_id},
)
return response.json()
return router
# ===== End HomeCore 1.0 product layer =====
app = FastAPI(
title="HomeCore Controller",
version=VERSION,
)
app.add_middleware(
CORSMiddleware,
allow_origins=[
"https://homecore.home-core.co.uk",
"https://www.home-core.co.uk",
],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
allow_headers=["*"],
)
app.include_router(
build_router(
version=VERSION,
data_dir=DATA,
load_state=load_state,
options=options,
ha_version=ha_version,
)
)
STYLE = '''
'''
def page(body):
return HTMLResponse(
STYLE +
'
' +
body +
'
'
)
def error_page(message):
return page(
'
HomeCore
'
'
Unable to continue
'
f'
{html.escape(message)}
'
'
No controller binding has been changed.
'
)
@app.get("/", response_class=HTMLResponse)
def setup_page():
s = load_state()
if s.get("controller_id"):
return HTMLResponse(APP_HTML)
return page(
'
HomeCore
'
'
Set up this controller
'
'
Sign in with the customer\'s HomeCore Owner account. '
'You will then choose one of the homes already created in HomeCore.
'
''
'
'
'Homes must be created in the HomeCore customer portal before a controller can be connected.'
'
'
)
@app.post("/", response_class=HTMLResponse)
def ingress_form_submit(
stage: str = Form(...),
email: str = Form(""),
password: str = Form(""),
nonce: str = Form(""),
property_id: str = Form(""),
):
"""
All commissioning forms submit to the ingress root.
This deliberately avoids nested browser paths because Home Assistant
Ingress owns the external URL prefix.
"""
if stage == "select-property":
return select_property(
email=email,
password=password,
)
if stage == "bind":
return bind(
nonce=nonce,
property_id=property_id,
)
return error_page(
"Unknown HomeCore setup step."
)
@app.post("/select-property", response_class=HTMLResponse)
def select_property(
email: str = Form(...),
password: str = Form(...),
):
api = options()["cloud_api"].rstrip("/")
try:
with httpx.Client(timeout=20) as c:
r = c.post(
api + "/api/v1/auth/login",
json={
"email": email,
"password": password,
},
)
if r.status_code >= 400:
return error_page("Incorrect HomeCore email address or password.")
token = r.json()["access_token"]
headers = {
"Authorization": "Bearer " + token
}
r = c.get(
api + "/api/v1/properties",
headers=headers,
)
if r.status_code >= 400:
return error_page(cloud_detail(r))
owned = [
p for p in r.json()
if p.get("owner") is True
]
except Exception as e:
return error_page(
"The controller could not contact HomeCore Cloud. " + str(e)
)
if not owned:
return error_page(
"This Owner account does not have a HomeCore home yet. "
"Create the home at homecore.home-core.co.uk, then return here."
)
nonce = secrets.token_urlsafe(32)
PENDING[nonce] = {
"token": token,
"expires": time.time() + 300,
"properties": {p["id"]: p for p in owned},
}
choices = "".join(
f''
for p in owned
)
return page(
'
HomeCore
'
'
Select the customer home
'
'
Only homes owned by this HomeCore account can be connected.
'
''
'
'
'After connection, this controller cannot be claimed by another HomeCore account.'
'
'
)
@app.post("/bind", response_class=HTMLResponse)
def bind(
nonce: str = Form(...),
property_id: str = Form(...),
):
if load_state().get("controller_id"):
raise HTTPException(
409,
"Controller already bound",
)
pending = PENDING.get(nonce)
if not pending:
return error_page(
"The setup session has expired. Sign in again."
)
if pending["expires"] < time.time():
PENDING.pop(nonce, None)
return error_page(
"The setup session has expired. Sign in again."
)
if property_id not in pending["properties"]:
return error_page(
"This HomeCore account does not own the selected property."
)
token = pending["token"]
api = options()["cloud_api"].rstrip("/")
headers = {
"Authorization": "Bearer " + token
}
payload = {
"property_id": property_id,
"public_key": pub_pem(),
"fingerprint": fingerprint(),
"software_version": VERSION,
"ha_version": ha_version(),
}
try:
with httpx.Client(timeout=20) as c:
r = c.post(
api + "/api/v1/controllers/bind",
headers=headers,
json=payload,
)
if r.status_code >= 400:
return error_page(cloud_detail(r))
out = r.json()
except Exception as e:
return error_page(
"The controller could not contact HomeCore Cloud. " + str(e)
)
prop = pending["properties"][property_id]
out["property_name"] = prop.get("name", "")
PENDING.pop(nonce, None)
save_state(out)
return page(
'
HomeCore
'
'
Controller connected
'
'
This Home Assistant Green is now permanently bound to the selected HomeCore home.
'
'
The customer can now sign into HomeCore using their normal HomeCore account.