161 lines
6.7 KiB
Python
Executable File
161 lines
6.7 KiB
Python
Executable File
"""HTTP views for Wall Panel."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from typing import Any
|
|
|
|
from aiohttp import web
|
|
from homeassistant.components.http import HomeAssistantView
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
from .const import CONF_CONFIG, CONF_SYNC_TOKEN, DOMAIN
|
|
from .helpers import (
|
|
build_patch_payload,
|
|
config_to_json,
|
|
create_room_layout_item,
|
|
current_entry_config,
|
|
delete_room_layout_item,
|
|
normalize_config,
|
|
reorder_room_grid,
|
|
save_settings,
|
|
update_entity_override,
|
|
update_room_layout_item,
|
|
update_room_override,
|
|
)
|
|
|
|
|
|
def _entry_from_hass(hass: HomeAssistant, entry_id: str):
|
|
return hass.data.get(DOMAIN, {}).get(entry_id, {}).get("entry")
|
|
|
|
|
|
def _request_token(request: web.Request) -> str:
|
|
header = request.headers.get("X-Wall-Panel-Token", "").strip()
|
|
if header:
|
|
return header
|
|
|
|
auth = request.headers.get("Authorization", "").strip()
|
|
if auth.lower().startswith("bearer "):
|
|
return auth[7:].strip()
|
|
|
|
return request.query.get("token", "").strip()
|
|
|
|
|
|
def _authorized(entry, request: web.Request) -> bool:
|
|
expected = str(entry.options.get(CONF_SYNC_TOKEN, "") or "").strip()
|
|
if not expected:
|
|
return False
|
|
|
|
return secrets.compare_digest(_request_token(request), expected)
|
|
|
|
|
|
def _response(data: Any, status: int = 200) -> web.Response:
|
|
if isinstance(data, str):
|
|
return web.Response(text=data, status=status, content_type="text/plain; charset=utf-8")
|
|
|
|
return web.json_response(data, status=status)
|
|
|
|
|
|
def _save_entry_config(hass: HomeAssistant, entry, config: dict[str, Any]) -> None:
|
|
options = dict(entry.options)
|
|
options[CONF_CONFIG] = config
|
|
hass.config_entries.async_update_entry(entry, options=options)
|
|
hass.data.setdefault(DOMAIN, {}).setdefault(entry.entry_id, {})["config"] = config
|
|
|
|
|
|
class WallPanelConfigView(HomeAssistantView):
|
|
"""Serve and update the canonical Wall Panel config."""
|
|
|
|
url = "/api/wall_panel/config/{entry_id}"
|
|
name = "api:wall_panel:config"
|
|
requires_auth = False
|
|
|
|
async def async_get(self, request: web.Request, entry_id: str) -> web.Response:
|
|
hass = request.app["hass"]
|
|
entry = _entry_from_hass(hass, entry_id)
|
|
if entry is None:
|
|
return _response({"ok": False, "error": "Unknown entry"}, 404)
|
|
if not _authorized(entry, request):
|
|
return _response({"ok": False, "error": "Unauthorized"}, 401)
|
|
|
|
config = current_entry_config(entry)
|
|
return _response(config)
|
|
|
|
async def async_post(self, request: web.Request, entry_id: str) -> web.Response:
|
|
hass = request.app["hass"]
|
|
entry = _entry_from_hass(hass, entry_id)
|
|
if entry is None:
|
|
return _response({"ok": False, "error": "Unknown entry"}, 404)
|
|
if not _authorized(entry, request):
|
|
return _response({"ok": False, "error": "Unauthorized"}, 401)
|
|
|
|
payload = await request.json()
|
|
if not isinstance(payload, dict):
|
|
return _response({"ok": False, "error": "Invalid payload"}, 400)
|
|
|
|
config = current_entry_config(entry)
|
|
action = str(payload.get("action", "") or "").strip().lower()
|
|
action_payload = payload.get("payload")
|
|
if not isinstance(action_payload, dict):
|
|
action_payload = payload
|
|
|
|
if action == "save-settings":
|
|
config = save_settings(config, action_payload)
|
|
elif action == "save-entity-override":
|
|
room_id = str(action_payload.get("room_id", "") or "").strip()
|
|
entity_id = str(action_payload.get("entity_id", "") or "").strip()
|
|
if not room_id or not entity_id:
|
|
return _response({"ok": False, "error": "room_id and entity_id are required"}, 400)
|
|
patch = build_patch_payload(action_payload, ["visible", "order", "card_type", "title", "icon"])
|
|
config = update_entity_override(config, room_id, entity_id, patch)
|
|
elif action == "save-space-override":
|
|
room_id = str(action_payload.get("room_id", "") or "").strip()
|
|
if not room_id:
|
|
return _response({"ok": False, "error": "room_id is required"}, 400)
|
|
patch = build_patch_payload(action_payload, ["visible", "order", "name", "icon", "temperature_sensor_entity_id"])
|
|
config = update_room_override(config, room_id, patch)
|
|
elif action == "create-room-layout-item":
|
|
room_id = str(action_payload.get("room_id", "") or "").strip()
|
|
if not room_id:
|
|
return _response({"ok": False, "error": "room_id is required"}, 400)
|
|
layout_item_id = str(action_payload.get("layout_item_id", "") or "").strip()
|
|
if not layout_item_id:
|
|
layout_item_id = f"slot_{secrets.token_hex(12)}"
|
|
order = action_payload.get("order")
|
|
config = create_room_layout_item(config, room_id, layout_item_id, int(order) if order is not None else None)
|
|
_save_entry_config(hass, entry, config)
|
|
return _response({
|
|
"ok": True,
|
|
"layout_item_id": layout_item_id,
|
|
"config": config,
|
|
})
|
|
elif action == "save-room-layout-item":
|
|
room_id = str(action_payload.get("room_id", "") or "").strip()
|
|
layout_item_id = str(action_payload.get("layout_item_id", "") or "").strip()
|
|
if not room_id or not layout_item_id:
|
|
return _response({"ok": False, "error": "room_id and layout_item_id are required"}, 400)
|
|
patch = build_patch_payload(action_payload, ["order"])
|
|
config = update_room_layout_item(config, room_id, layout_item_id, patch)
|
|
elif action == "delete-room-layout-item":
|
|
room_id = str(action_payload.get("room_id", "") or "").strip()
|
|
layout_item_id = str(action_payload.get("layout_item_id", "") or "").strip()
|
|
if not room_id or not layout_item_id:
|
|
return _response({"ok": False, "error": "room_id and layout_item_id are required"}, 400)
|
|
config = delete_room_layout_item(config, room_id, layout_item_id)
|
|
elif action == "reorder-room-grid":
|
|
room_id = str(action_payload.get("room_id", "") or "").strip()
|
|
entries = action_payload.get("entries", [])
|
|
if not room_id or not isinstance(entries, list):
|
|
return _response({"ok": False, "error": "room_id and entries are required"}, 400)
|
|
config = reorder_room_grid(config, room_id, entries)
|
|
elif isinstance(payload.get("config"), dict):
|
|
config = normalize_config(payload["config"])
|
|
else:
|
|
return _response({"ok": False, "error": "Unknown action"}, 404)
|
|
|
|
_save_entry_config(hass, entry, config)
|
|
return _response({
|
|
"ok": True,
|
|
"config": config,
|
|
})
|