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.3"
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", "sensor", "number", "select",
}
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",
"sensor": "Sensors",
"other": "Other",
}
HOMECORE_DEVICE_CATALOG = [
{
"id": "shelly",
"name": "Shelly",
"category": "Smart devices",
"description": "Relays, switches, dimmers, plugs, shutters, meters and supported Shelly devices.",
"capabilities": ["Lighting", "Switching", "Blinds", "Energy"],
"variants": [
{
"id": "shelly",
"name": "Shelly Wi-Fi / Ethernet",
"handlers": ["shelly"],
"note": "HomeCore discovers the Shelly on your network or lets you enter its address.",
},
{
"id": "shelly_bluetooth",
"name": "Shelly BLU / Bluetooth",
"handlers": ["bthome"],
"note": "For compatible Shelly BLU devices exposed through BTHome.",
},
{
"id": "shelly_zwave",
"name": "Shelly Wave / Z-Wave",
"handlers": ["zwave_js"],
"note": "Requires a Z-Wave radio/controller connected to the HomeCore Green.",
},
],
},
{
"id": "philips_hue",
"name": "Philips Hue",
"category": "Lighting",
"description": "Hue lights, lamps, light strips, switches, remotes and motion sensors.",
"capabilities": ["Lighting", "Scenes", "Sensors"],
"variants": [
{
"id": "hue_bridge",
"name": "Hue Bridge",
"handlers": ["hue"],
"note": "Recommended. HomeCore discovers the Hue Bridge and guides you through pairing.",
},
{
"id": "hue_bluetooth",
"name": "Hue Bluetooth",
"handlers": ["hue_ble"],
"note": "For supported Hue Bluetooth lights when a compatible Bluetooth adapter is available.",
},
],
},
{
"id": "hikvision_axpro",
"name": "Hikvision AX PRO",
"category": "Alarm & security",
"description": "AX PRO / AX Hub alarm panel, zones and supported wireless detectors.",
"capabilities": ["Alarm", "Zones", "Sensors"],
"variants": [
{
"id": "hikvision_axpro",
"name": "Hikvision AX PRO",
"handlers": ["hikvision_axpro"],
"note": "Uses the HomeCore-compatible AX PRO local driver when installed.",
"unavailable": "The AX PRO driver is not installed on this Green yet.",
}
],
},
{
"id": "yale",
"name": "Yale",
"category": "Alarm & security",
"description": "Yale Smart Living/Sync alarms plus supported Yale Home devices.",
"capabilities": ["Alarm", "Locks", "Sensors", "Doorbell"],
"variants": [
{
"id": "yale_smart_living",
"name": "Yale Smart Living / Sync Alarm",
"handlers": ["yale_smart_alarm"],
"note": "For the established Yale Smart Living / Sync alarm platform.",
},
{
"id": "yale_home",
"name": "Yale Home",
"handlers": ["yale"],
"note": "For Yale Home locks, doorbells and supported Yale Home devices.",
},
{
"id": "yale_new_alarm",
"name": "New Yale Smart Alarm",
"handlers": ["yale_smart_alarm_ce"],
"note": "Uses the HomeCore-compatible driver for the newer Yale Home alarm range when installed.",
"unavailable": "The new Yale Smart Alarm driver is not installed on this Green yet.",
},
],
},
{
"id": "heatmiser",
"name": "Heatmiser",
"category": "Heating",
"description": "Heatmiser thermostats, heating zones and supported NeoHub systems.",
"capabilities": ["Heating", "Thermostats", "Zones"],
"variants": [
{
"id": "heatmiser_neo",
"name": "Heatmiser Neo / NeoHub",
"handlers": ["heatmiserneo"],
"note": "Uses the HomeCore-compatible NeoHub driver when installed.",
"unavailable": "The Heatmiser Neo driver is not installed on this Green yet.",
},
{
"id": "heatmiser_legacy",
"name": "Heatmiser legacy gateway",
"handlers": ["heatmiser"],
"note": "For legacy Heatmiser gateways when a setup handler is available.",
"unavailable": "This Heatmiser system requires the HomeCore Heatmiser driver.",
},
],
},
{
"id": "sky_glass",
"name": "Sky Glass",
"category": "Entertainment",
"description": "Sky Glass television and compatible Sky entertainment controls.",
"capabilities": ["TV", "Media"],
"variants": [
{
"id": "sky_glass",
"name": "Sky Glass",
"handlers": ["sky_glass", "sky_stream"],
"note": "HomeCore will use a compatible Sky Glass/Stream driver when one is installed.",
"unavailable": "A compatible Sky Glass driver is not installed on this Green yet.",
}
],
},
{
"id": "hive",
"name": "Hive",
"category": "Heating & smart devices",
"description": "Hive heating, hot water, radiator valves, plugs, lights and supported sensors.",
"capabilities": ["Heating", "Hot water", "Lighting", "Plugs", "Sensors"],
"variants": [
{
"id": "hive",
"name": "Hive",
"handlers": ["hive"],
"note": "Sign in with the Hive account owner. Hive requires two-factor authentication.",
}
],
},
]
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\n\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 device_class in (
"door", "window", "opening", "motion", "occupancy",
"tamper", "lock", "problem", "safety", "smoke", "gas",
):
return "security"
return "sensor"
if domain == "sensor":
if device_class == "temperature":
return "heating"
return "sensor"
if domain in ("number", "select"):
return "sensor"
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, include_unadded=False):
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)
# Event entities are only customer-facing when they represent
# doorbell/intercom activity. Binary sensors are retained because
# alarm ecosystems expose zones and detectors through them.
if category == "other" and domain == "event":
return None
override = (product.get("devices") or {}).get(entity_id, {})
# A device becomes part of HomeCore only when the customer explicitly
# adds it. Legacy configured entries from 1.0.1 count as added.
added = bool(override.get("added"))
if override and "added" not in override:
added = True
if not include_unadded and not added:
return None
if not include_unadded and 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")),
"added": added,
"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 all_homeassistant_states():
states = ha_request("GET", "/states") or []
if not isinstance(states, list):
raise RuntimeError("Home Assistant did not return an entity-state list")
return states
def get_devices(ctx=None):
product = load_product()
out = []
for raw in all_homeassistant_states():
item = normalize(raw, product, include_unadded=False)
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 get_discovered_devices(ctx=None):
product = load_product()
states = all_homeassistant_states()
out = []
for raw in states:
item = normalize(raw, product, include_unadded=True)
if not item:
continue
if ctx and not can_view(ctx, item["category"]):
continue
out.append(item)
out.sort(
key=lambda x: (
0 if x.get("added") else 1,
x["category"],
x["name"].lower(),
)
)
return states, 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/device-discovery")
def device_discovery(request: Request):
ctx = auth_context(request)
require_owner(ctx)
states, discovered = get_discovered_devices(ctx)
return {
"total_states": len(states),
"supported_count": len(discovered),
"added_count": sum(1 for x in discovered if x.get("added")),
"devices": discovered,
}
def flow_handlers():
handlers = ha_request(
"GET",
"/config/config_entries/flow_handlers",
timeout=30,
) or []
return set(str(x) for x in handlers)
def public_catalog():
handlers = flow_handlers()
result = []
for brand in HOMECORE_DEVICE_CATALOG:
item = {
key: value
for key, value in brand.items()
if key != "variants"
}
variants = []
for variant in brand.get("variants", []):
chosen = next(
(
handler
for handler in variant.get("handlers", [])
if handler in handlers
),
None,
)
variants.append(
{
"id": variant["id"],
"name": variant["name"],
"note": variant.get("note", ""),
"available": bool(chosen),
"handler": chosen,
"unavailable": variant.get(
"unavailable",
"This HomeCore device driver is not available on this Green.",
),
}
)
item["variants"] = variants
item["available"] = any(x["available"] for x in variants)
result.append(item)
return result
def find_catalog_variant(brand_id, variant_id):
for brand in HOMECORE_DEVICE_CATALOG:
if brand["id"] != brand_id:
continue
variants = brand.get("variants", [])
variant = next(
(x for x in variants if x["id"] == variant_id),
variants[0] if len(variants) == 1 else None,
)
if not variant:
raise HTTPException(400, "Choose a HomeCore device type")
return brand, variant
raise HTTPException(404, "HomeCore device brand not found")
@router.get("/api/v1/device-catalog")
def device_catalog(request: Request):
ctx = auth_context(request)
require_owner(ctx)
try:
catalogue = public_catalog()
except Exception as exc:
raise HTTPException(
502,
"HomeCore could not read the device-driver catalogue from the controller: "
+ str(exc),
)
return {
"catalog": catalogue,
"expandable": True,
}
@router.post("/api/v1/device-setup/start")
async def device_setup_start(request: Request):
ctx = auth_context(request)
require_owner(ctx)
body = await request.json()
brand_id = str(body.get("brand_id") or "")
variant_id = str(body.get("variant_id") or "")
brand, variant = find_catalog_variant(brand_id, variant_id)
handlers = flow_handlers()
handler = next(
(
name
for name in variant.get("handlers", [])
if name in handlers
),
None,
)
if not handler:
raise HTTPException(
409,
variant.get(
"unavailable",
"This HomeCore device driver is not available on this Green.",
),
)
try:
step = ha_request(
"POST",
"/config/config_entries/flow",
payload={
"handler": handler,
"show_advanced_options": False,
},
timeout=60,
)
except httpx.HTTPStatusError as exc:
detail = ""
try:
detail = exc.response.json().get("message", "")
except Exception:
detail = exc.response.text
raise HTTPException(
exc.response.status_code,
detail or f"Could not start {brand['name']} setup",
)
activity(
"device.setup.started",
{
"brand": brand["name"],
"variant": variant["name"],
"handler": handler,
},
)
return {
"brand": brand["name"],
"variant": variant["name"],
"handler": handler,
"step": step,
}
@router.get("/api/v1/device-setup/flow/{flow_id}")
def device_setup_get(flow_id: str, request: Request):
ctx = auth_context(request)
require_owner(ctx)
try:
return ha_request(
"GET",
f"/config/config_entries/flow/{flow_id}",
timeout=60,
)
except httpx.HTTPStatusError as exc:
raise HTTPException(
exc.response.status_code,
"HomeCore device setup session is no longer available",
)
@router.post("/api/v1/device-setup/flow/{flow_id}")
async def device_setup_continue(flow_id: str, request: Request):
ctx = auth_context(request)
require_owner(ctx)
body = await request.json()
try:
step = ha_request(
"POST",
f"/config/config_entries/flow/{flow_id}",
payload=body,
timeout=120,
)
except httpx.HTTPStatusError as exc:
detail = ""
try:
detail = exc.response.json().get("message", "")
except Exception:
detail = exc.response.text
raise HTTPException(
exc.response.status_code,
detail or "HomeCore could not continue device setup",
)
if isinstance(step, dict) and step.get("type") == "create_entry":
activity(
"device.setup.completed",
{
"handler": step.get("handler", ""),
"title": step.get("title", ""),
},
)
return step
@router.delete("/api/v1/device-setup/flow/{flow_id}")
def device_setup_cancel(flow_id: str, request: Request):
ctx = auth_context(request)
require_owner(ctx)
try:
ha_request(
"DELETE",
f"/config/config_entries/flow/{flow_id}",
timeout=30,
)
except Exception:
pass
return {"ok": True}
@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,
"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 "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 "added" in body:
value["added"] = bool(
body["added"]
)
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/status")
def cloud_status(request: Request):
auth_context(request)
api = options()["cloud_api"].rstrip("/")
try:
response = httpx.get(api + "/health", timeout=5)
return {
"connected": response.status_code < 500,
"status": "connected" if response.status_code < 500 else "offline",
}
except Exception:
return {"connected": False, "status": "offline"}
@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.