wallpanell/custom_components/wall_panel/helpers.py
2026-03-25 13:48:26 +03:00

284 lines
8.8 KiB
Python
Executable File

"""Helper functions for Wall Panel."""
from __future__ import annotations
import json
from copy import deepcopy
from typing import Any
from .const import (
CONF_CONFIG,
CONF_FRONTEND_URL_PATH,
CONF_PANEL_URL,
CONF_REQUIRE_ADMIN,
CONF_SIDEBAR_ICON,
CONF_SIDEBAR_TITLE,
CONF_SYNC_TOKEN,
DEFAULT_FRONTEND_URL_PATH,
DEFAULT_PANEL_URL,
DEFAULT_SIDEBAR_ICON,
DEFAULT_SIDEBAR_TITLE,
)
def default_config() -> dict[str, Any]:
return {
"app": {
"title": "Wall Panel",
"poll_interval_ms": 5000,
"main_room_name": "Главная",
"main_room_icon": "mdi:home",
"edit_mode": False,
"battery_history_hours": 4320,
},
"home_assistant": {
"base_url": "",
"token": "",
"verify_ssl": True,
"sync_url": "",
"sync_token": "",
"sync_timeout": 10,
"sync_verify_ssl": True,
"sync_cache_seconds": 30,
"weather_entity_id": "",
"auto_label": "auto",
"auto_entity_ids": [],
},
"camera": {
"rtsp_url": "",
"stream_url": "",
"stream_mode": "hls",
"poster_url": "",
"popup_timeout_minutes": 3,
"trigger_entities": [],
},
"rooms": [],
}
def normalize_config(value: Any) -> dict[str, Any]:
config = deepcopy(default_config())
if not isinstance(value, dict):
return config
_deep_merge(config, value)
if not isinstance(config.get("rooms"), list):
config["rooms"] = []
return config
def config_to_json(config: dict[str, Any]) -> str:
return json.dumps(config, ensure_ascii=False, indent=2, sort_keys=False)
def parse_config_json(raw: str) -> dict[str, Any]:
data = json.loads(raw)
if not isinstance(data, dict):
raise ValueError("Config JSON must be an object")
return normalize_config(data)
def current_entry_config(entry) -> dict[str, Any]:
return normalize_config(entry.options.get(CONF_CONFIG))
def current_entry_panel(entry) -> dict[str, Any]:
return {
CONF_PANEL_URL: str(entry.options.get(CONF_PANEL_URL, DEFAULT_PANEL_URL) or ""),
CONF_SYNC_TOKEN: str(entry.options.get(CONF_SYNC_TOKEN, "") or ""),
CONF_SIDEBAR_TITLE: str(entry.options.get(CONF_SIDEBAR_TITLE, DEFAULT_SIDEBAR_TITLE) or DEFAULT_SIDEBAR_TITLE),
CONF_SIDEBAR_ICON: str(entry.options.get(CONF_SIDEBAR_ICON, DEFAULT_SIDEBAR_ICON) or DEFAULT_SIDEBAR_ICON),
CONF_FRONTEND_URL_PATH: str(entry.options.get(CONF_FRONTEND_URL_PATH, DEFAULT_FRONTEND_URL_PATH) or DEFAULT_FRONTEND_URL_PATH),
CONF_REQUIRE_ADMIN: bool(entry.options.get(CONF_REQUIRE_ADMIN, False)),
}
def save_settings(config: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
app = config.setdefault("app", {})
if "edit_mode" in payload:
app["edit_mode"] = bool(payload["edit_mode"])
if isinstance(payload.get("title"), str) and payload["title"].strip():
app["title"] = payload["title"].strip()
return config
def update_entity_override(config: dict[str, Any], room_id: str, entity_id: str, patch: dict[str, Any]) -> dict[str, Any]:
room = _ensure_room(config, room_id)
overrides = room.setdefault("entity_overrides", {})
current = overrides.get(entity_id, {})
if not isinstance(current, dict):
current = {}
merged = deepcopy(current)
for key, value in patch.items():
if value is not None:
merged[key] = value
overrides[entity_id] = merged
return config
def update_room_override(config: dict[str, Any], room_id: str, patch: dict[str, Any]) -> dict[str, Any]:
room = _ensure_room(config, room_id)
for key, value in patch.items():
if value is not None:
room[key] = value
return config
def update_room_layout_item(config: dict[str, Any], room_id: str, layout_item_id: str, patch: dict[str, Any]) -> dict[str, Any]:
room = _ensure_room(config, room_id)
items = room.setdefault("layout_items", [])
if not isinstance(items, list):
items = []
room["layout_items"] = items
current = None
for item in items:
if isinstance(item, dict) and str(item.get("id", "")) == layout_item_id:
current = item
break
if current is None:
current = {
"id": layout_item_id,
"type": "ghost",
}
items.append(current)
for key, value in patch.items():
if value is not None:
current[key] = value
_sort_room_layout_items(room)
return config
def create_room_layout_item(config: dict[str, Any], room_id: str, layout_item_id: str, order: int | None = None) -> dict[str, Any]:
return update_room_layout_item(config, room_id, layout_item_id, {
"order": order,
"type": "ghost",
})
def delete_room_layout_item(config: dict[str, Any], room_id: str, layout_item_id: str) -> dict[str, Any]:
room = _ensure_room(config, room_id)
items = room.get("layout_items", [])
if not isinstance(items, list):
room["layout_items"] = []
return config
room["layout_items"] = [
item for item in items
if not isinstance(item, dict) or str(item.get("id", "")) != layout_item_id
]
return config
def reorder_room_grid(config: dict[str, Any], room_id: str, entries: list[Any]) -> dict[str, Any]:
room = _ensure_room(config, room_id)
normalized: list[dict[str, str]] = []
for entry in entries:
if not isinstance(entry, dict):
continue
kind = str(entry.get("kind", "")).strip()
item_id = str(entry.get("id", "")).strip()
if kind not in {"entity", "layout"} or not item_id:
continue
normalized.append({"kind": kind, "id": item_id})
entity_overrides = room.setdefault("entity_overrides", {})
if not isinstance(entity_overrides, dict):
entity_overrides = {}
room["entity_overrides"] = entity_overrides
layout_items = room.setdefault("layout_items", [])
if not isinstance(layout_items, list):
layout_items = []
room["layout_items"] = layout_items
layout_by_id = {
str(item.get("id", "")): item
for item in layout_items
if isinstance(item, dict) and str(item.get("id", "")).strip()
}
order = 10000
for entry in normalized:
if entry["kind"] == "entity":
current = entity_overrides.get(entry["id"], {})
if not isinstance(current, dict):
current = {}
merged = deepcopy(current)
merged["order"] = order
entity_overrides[entry["id"]] = merged
else:
current = layout_by_id.get(entry["id"], {
"id": entry["id"],
"type": "ghost",
})
merged = deepcopy(current)
merged["id"] = entry["id"]
merged["type"] = "ghost"
merged["order"] = order
layout_by_id[entry["id"]] = merged
order += 10
room["layout_items"] = sorted(
layout_by_id.values(),
key=lambda item: (int(item.get("order", 9999) or 9999), str(item.get("id", ""))),
)
return config
def build_patch_payload(payload: dict[str, Any], keys: list[str]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key in keys:
if key in payload:
result[key] = payload[key]
return result
def _deep_merge(target: dict[str, Any], source: dict[str, Any]) -> None:
for key, value in source.items():
if isinstance(value, dict) and isinstance(target.get(key), dict):
_deep_merge(target[key], value)
else:
target[key] = deepcopy(value)
def _ensure_room(config: dict[str, Any], room_id: str) -> dict[str, Any]:
rooms = config.setdefault("rooms", [])
if not isinstance(rooms, list):
rooms = []
config["rooms"] = rooms
for room in rooms:
if isinstance(room, dict) and str(room.get("id", "")) == room_id:
room.setdefault("visible", True)
room.setdefault("entity_ids", [])
room.setdefault("entity_overrides", {})
room.setdefault("layout_items", [])
return room
room = {
"id": room_id,
"visible": True,
"entity_ids": [],
"entity_overrides": {},
"layout_items": [],
}
rooms.append(room)
return room
def _sort_room_layout_items(room: dict[str, Any]) -> None:
items = room.get("layout_items", [])
if not isinstance(items, list):
room["layout_items"] = []
return
room["layout_items"] = sorted(
[item for item in items if isinstance(item, dict)],
key=lambda item: (int(item.get("order", 9999) or 9999), str(item.get("id", ""))),
)