170 lines
6.9 KiB
Python
170 lines
6.9 KiB
Python
"""Home Assistant как python_tool с перечислимыми действиями.
|
||
|
||
Токен HA живёт в gateway и в контекст модели не попадает. Действия:
|
||
найти сущности, посмотреть состояние, вызвать сервис, перечислить сервисы
|
||
домена. Ничего про конфигурацию, пользователей и интеграции - только
|
||
REST `/api/states` и `/api/services`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from dataclasses import dataclass
|
||
from typing import Any, Literal
|
||
|
||
import aiohttp
|
||
|
||
__all__ = ["Action", "HomeAssistant"]
|
||
|
||
Action = Literal["search", "state", "call", "services"]
|
||
|
||
SEARCH_LIMIT = 60
|
||
ATTR_LIMIT = 25
|
||
# Не влезает в ответ и не нужно модели: гигантские списки и бинарники.
|
||
NOISY_ATTRS = frozenset({"entity_picture", "icon", "supported_color_modes", "options"})
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class HomeAssistant:
|
||
url: str
|
||
token: str
|
||
timeout: float = 20.0
|
||
max_chars: int = 8_000
|
||
|
||
async def ha(
|
||
self,
|
||
action: Action,
|
||
*,
|
||
query: str | None = None,
|
||
entity: str | None = None,
|
||
domain: str | None = None,
|
||
service: str | None = None,
|
||
data: dict[str, Any] | None = None,
|
||
) -> str:
|
||
"""Home Assistant - умный дом: свет, кнопки ИК-пульта, сцены, скрипты, датчики.
|
||
|
||
action:
|
||
- search: сущности по подстроке в id или имени (query; пусто - все),
|
||
сгруппированы по домену. Начни отсюда: точные entity_id нужны везде.
|
||
- state: состояние и атрибуты одной сущности (entity).
|
||
- call: вызвать сервис (domain, service, entity опционально, data -
|
||
доп. поля: brightness, color_temp, …). Примеры: light/turn_on,
|
||
light/turn_off, script/turn_on, scene/turn_on, button/press.
|
||
- services: какие сервисы есть у домена (domain) и их поля.
|
||
|
||
Делай то, о чём попросили: «выключи свет» - это одна сущность или
|
||
группа, не весь дом.
|
||
"""
|
||
match action:
|
||
case "search":
|
||
return await self._search(query or "")
|
||
case "state":
|
||
return await self._state(_need(entity, "entity"))
|
||
case "call":
|
||
return await self._call(
|
||
_need(domain, "domain"), _need(service, "service"), entity, data
|
||
)
|
||
case "services":
|
||
return await self._services(_need(domain, "domain"))
|
||
|
||
async def _search(self, query: str) -> str:
|
||
states = await self._get("/api/states")
|
||
q = query.lower()
|
||
hits = [
|
||
s
|
||
for s in states
|
||
if q in s["entity_id"].lower()
|
||
or q in str(s.get("attributes", {}).get("friendly_name", "")).lower()
|
||
]
|
||
if not hits:
|
||
return "ничего не нашлось"
|
||
by_domain: dict[str, list[dict[str, Any]]] = {}
|
||
for s in sorted(hits, key=lambda s: s["entity_id"]):
|
||
by_domain.setdefault(s["entity_id"].split(".", 1)[0], []).append(s)
|
||
lines = []
|
||
shown = 0
|
||
for dom, items in by_domain.items():
|
||
lines.append(f"{dom}:")
|
||
for s in items:
|
||
if shown >= SEARCH_LIMIT:
|
||
lines.append(f" … ещё {len(hits) - shown}, уточни query")
|
||
return "\n".join(lines)
|
||
name = s.get("attributes", {}).get("friendly_name") or ""
|
||
lines.append(f" {s['entity_id']:40} {s['state']:12} {name}")
|
||
shown += 1
|
||
return "\n".join(lines)
|
||
|
||
async def _state(self, entity: str) -> str:
|
||
s = await self._get(f"/api/states/{entity}")
|
||
attrs = {
|
||
k: v for k, v in (s.get("attributes") or {}).items() if k not in NOISY_ATTRS
|
||
}
|
||
lines = [f"{s['entity_id']}: {s['state']} (с {s.get('last_changed', '?')})"]
|
||
for k, v in list(attrs.items())[:ATTR_LIMIT]:
|
||
lines.append(f" {k}: {json.dumps(v, ensure_ascii=False)[:200]}")
|
||
return self._clip("\n".join(lines))
|
||
|
||
async def _call(
|
||
self, domain: str, service: str, entity: str | None, data: dict[str, Any] | None
|
||
) -> str:
|
||
body: dict[str, Any] = dict(data or {})
|
||
if entity:
|
||
body["entity_id"] = entity
|
||
changed = await self._post(f"/api/services/{domain}/{service}", body)
|
||
if not changed:
|
||
return f"{domain}.{service}: выполнено, состояния не менялись"
|
||
lines = [f"{domain}.{service}: выполнено, изменилось:"]
|
||
lines.extend(f" {s['entity_id']}: {s['state']}" for s in changed[:20])
|
||
return "\n".join(lines)
|
||
|
||
async def _services(self, domain: str) -> str:
|
||
for block in await self._get("/api/services"):
|
||
if block.get("domain") == domain:
|
||
lines = [f"{domain}:"]
|
||
for name, spec in sorted((block.get("services") or {}).items()):
|
||
fields = ", ".join((spec.get("fields") or {}).keys())
|
||
lines.append(
|
||
f" {name}: {spec.get('description') or ''}".rstrip()
|
||
+ (f" [{fields}]" if fields else "")
|
||
)
|
||
return self._clip("\n".join(lines))
|
||
return f"домена {domain} нет"
|
||
|
||
async def _get(self, path: str) -> Any:
|
||
return await self._request("GET", path, None)
|
||
|
||
async def _post(self, path: str, body: dict[str, Any]) -> Any:
|
||
return await self._request("POST", path, body)
|
||
|
||
async def _request(
|
||
self, method: str, path: str, body: dict[str, Any] | None
|
||
) -> Any:
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
async with (
|
||
aiohttp.ClientSession() as http,
|
||
http.request(
|
||
method,
|
||
f"{self.url.rstrip('/')}{path}",
|
||
json=body,
|
||
headers=headers,
|
||
timeout=aiohttp.ClientTimeout(total=self.timeout),
|
||
) as response,
|
||
):
|
||
text = await response.text()
|
||
if response.status >= 400:
|
||
msg = f"Home Assistant {method} {path}: {response.status} {text[:300]}"
|
||
raise RuntimeError(msg)
|
||
return json.loads(text) if text else None
|
||
|
||
def _clip(self, text: str) -> str:
|
||
if len(text) <= self.max_chars:
|
||
return text
|
||
return text[: self.max_chars] + "\n…[обрезано]"
|
||
|
||
|
||
def _need(value: str | None, name: str) -> str:
|
||
if not value:
|
||
msg = f"для этого действия нужен параметр {name}"
|
||
raise ValueError(msg)
|
||
return value
|