host/home-assistant-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v2.5.0 — rename_entity, toggle_automation, trigger_automation + NL маршрутизация

Browse Source
This commit is contained in:
host
2026-07-29 19:59:41 +03:00
parent 40ed449663
commit 0f7f060408
1 changed files with 201 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
home_assistant_tools.py
+200 -86
View File
@@ -2,7 +2,7 @@
title: Home Assistant Smart Control
description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка.
author: ChatGPT (refactored by MiniMax-M3)
version: 2.4.0
version: 2.5.0
license: MIT
"""
@@ -66,7 +66,6 @@ class Tools:
"устройство", "все", "весь", "вся",
}
# Домены и их алиасы (русский -> домен HA)
_DOMAIN_ALIASES = {
"binary_sensor": ["binary_sensor", "бинарн", "датчик", "датчики", "сенсор", "сенсоры"],
"sensor": ["sensor", "показател", "значение", "значения"],
@@ -261,6 +260,22 @@ class Tools:
return domain
return None
def _find_entity_by_name(self, name: str, domain: Optional[str] = None) -> Optional[str]:
"""Найти первый entity_id, у которого friendly_name или id совпадает с name."""
states = self._get_states(force=True)
name_lower = name.lower().strip()
for s in states:
eid = s["entity_id"]
if domain and not eid.startswith(domain + "."):
continue
fn = (self._attr(s, "friendly_name", "") or "").lower()
if fn == name_lower:
return eid
# Частичное совпадение
if name_lower in fn or name_lower in eid.lower():
return eid
return None
# ========================================================================
# Парсинг чисел
# ========================================================================
@@ -289,26 +304,21 @@ class Tools:
return " * " + fn + room + ": " + str(state) + str(unit)
# ========================================================================
# Мониторинг (ответы на вопросы о состоянии дома)
# Мониторинг
# ========================================================================
def _monitor_windows_doors(self, tokens: List[str]) -> List[str]:
"""Какие окна/двери открыты."""
cfg = self._MONITOR_DOMAINS["window_door"]
states = self._get_states()
items: List[Tuple[Dict[str, Any], bool, bool]] = []
for s in states:
eid = s["entity_id"]
is_relevant_domain = eid.startswith("binary_sensor.")
device_class = self._attr(s, "device_class", "")
is_door_window = device_class in ("door", "window", "garage_door", "lock")
text = self._entity_text(s)
has_keyword = any(k in text for k in cfg["keywords"])
if not ((is_relevant_domain and is_door_window) or has_keyword):
continue
cur = s.get("state", "")
is_open = cur in cfg["open_states"]
is_closed = cur in cfg["closed_states"]
@@ -330,7 +340,6 @@ class Tools:
lines.append("Закрыто: " + str(len(closed_items)) + " шт.")
if unknown:
lines.append("Неизвестно: " + str(len(unknown)) + " шт.")
if tokens:
filtered = []
for line in lines:
@@ -338,17 +347,13 @@ class Tools:
filtered.append(line)
if filtered:
return filtered
return lines
def _monitor_lights(self, tokens: List[str]) -> List[str]:
"""Какой свет горит."""
cfg = self._MONITOR_DOMAINS["light_status"]
entities = self._find_entities(tokens if tokens else cfg["keywords"], "light")
if not entities:
return ["свет не найден"]
on_items: List[str] = []
off_items: List[str] = []
for eid in entities:
@@ -357,7 +362,6 @@ class Tools:
on_items.append(eid)
else:
off_items.append(eid)
lines: List[str] = []
if on_items:
lines.append("Горит: " + str(len(on_items)) + " шт.")
@@ -371,55 +375,37 @@ class Tools:
return lines
def _monitor_climate(self, tokens: List[str]) -> List[str]:
"""Какая температура и состояние климата.
Берём ТОЛЬКО climate.* или sensor с device_class=temperature/humidity.
"""
states = self._get_states()
items: List[Dict[str, Any]] = []
for s in states:
eid = s["entity_id"]
dc = self._attr(s, "device_class", "")
# Только climate или sensor с релевантным device_class
if eid.startswith("climate."):
items.append(s)
elif eid.startswith("sensor.") and dc in (
"temperature", "humidity", "current_temperature",
):
items.append(s)
# Фильтр по комнате/токенам
if tokens:
items = [
s for s in items
if any(t in self._entity_text(s) for t in tokens)
]
if not items:
return ["климат-устройства не найдены"]
lines = ["**Климат:**"]
for s in items[:20]:
lines.append(self._format_entity(s))
return lines
def _monitor_battery(self, tokens: List[str]) -> List[str]:
"""Какие батарейки садятся.
Учитываем:
- sensor.* с device_class=battery и state как число
- binary_sensor.* с device_class=battery (on = садится)
"""
states = self._get_states()
low_battery: List[Tuple[Dict[str, Any], float]] = []
binary_low: List[Dict[str, Any]] = []
for s in states:
eid = s["entity_id"]
dc = self._attr(s, "device_class", "")
cur = s.get("state", "")
# Числовой сенсор батареи
if eid.startswith("sensor.") and dc == "battery":
if cur in ("unavailable", "unknown", "none", ""):
continue
@@ -430,18 +416,13 @@ class Tools:
if level < 30:
low_battery.append((s, level))
continue
# Бинарный сенсор "батарея разряжена"
if eid.startswith("binary_sensor.") and dc == "battery":
if cur == "on":
binary_low.append(s)
continue
if not low_battery and not binary_low:
return ["Все батарейки в норме (>=30%)"]
lines = ["**Низкий заряд:**"]
# Сортируем по возрастанию
low_battery.sort(key=lambda x: x[1])
for s, level in low_battery[:15]:
lines.append(self._format_entity(s))
@@ -450,10 +431,8 @@ class Tools:
return lines
def _monitor_motion(self, tokens: List[str]) -> List[str]:
"""Где есть движение прямо сейчас."""
states = self._get_states()
active: List[Dict[str, Any]] = []
for s in states:
eid = s["entity_id"]
if not eid.startswith("binary_sensor."):
@@ -463,27 +442,22 @@ class Tools:
continue
if s.get("state") == "on":
active.append(s)
if not active:
return ["Нигде нет движения (все датчики молчат)"]
if tokens:
active = [s for s in active if any(t in self._entity_text(s) for t in tokens)]
lines = ["**Движение обнаружено:**"]
for s in active[:15]:
lines.append(self._format_entity(s))
return lines
def _monitor_air_quality(self, tokens: List[str]) -> List[str]:
"""Качество воздуха: TVOC, CO2, PM2.5/10, влажность."""
states = self._get_states()
items: List[Dict[str, Any]] = []
target_dc = {
"volatile_organic_compounds", "carbon_dioxide",
"pm25", "pm10", "humidity", "aqi",
}
for s in states:
eid = s["entity_id"]
if not eid.startswith("sensor."):
@@ -492,32 +466,25 @@ class Tools:
name = (self._attr(s, "friendly_name", "") or "").lower()
if dc in target_dc or any(k in name for k in ["tvoc", "co2", "влажн", "воздух", "pm2", "pm10"]):
items.append(s)
if not items:
return ["датчики качества воздуха не найдены"]
if tokens:
items = [s for s in items if any(t in self._entity_text(s) for t in tokens)]
lines = ["**Качество воздуха:**"]
for s in items[:15]:
lines.append(self._format_entity(s))
return lines
def _monitor_media(self, tokens: List[str]) -> List[str]:
"""Состояние медиаплееров."""
cfg = self._MONITOR_DOMAINS["media_status"]
entities = self._find_entities(tokens if tokens else cfg["keywords"], "media_player")
if not entities:
return ["медиаплееры не найдены"]
playing: List[str] = []
idle: List[str] = []
for eid in entities:
state = self._get_state(eid)
(playing if state == "playing" else idle).append(eid)
lines = ["**Медиа:**"]
for eid in playing:
for s in self._get_states():
@@ -538,11 +505,9 @@ class Tools:
return lines
def _monitor_presence(self, tokens: List[str]) -> List[str]:
"""Кто дома."""
states = self._get_states()
home: List[str] = []
away: List[str] = []
for s in states:
eid = s["entity_id"]
if not (eid.startswith("person.") or eid.startswith("device_tracker.")):
@@ -553,7 +518,6 @@ class Tools:
home.append(name)
else:
away.append(name)
lines = ["**Люди дома:**"]
if home:
for n in home:
@@ -565,7 +529,6 @@ class Tools:
return lines
def _monitor_full_report(self) -> List[str]:
"""Общий отчёт по дому одной фразой."""
lines = ["**Отчёт по дому:**", ""]
lines.extend(self._monitor_presence([]))
lines.append("")
@@ -599,9 +562,7 @@ class Tools:
warnings: List[str] = []
try:
# ============================================================
# Маршрутизация 'покажи все / список устройств' -> list_devices
# ============================================================
if any(w in cmd for w in [
"покажи", "список устройств", "какие устройств",
"что есть", "инвентар", "все устройств",
@@ -609,15 +570,44 @@ class Tools:
target = self._resolve_domain(cmd)
return self.list_devices(domain=target or "", limit=80)
# ============================================================
# Общий отчёт
# ============================================================
if "отчет" in cmd or "отчёт" in cmd or "обзор" in cmd or "все вместе" in cmd:
return "\n".join(self._monitor_full_report())
# ============================================================
# РЕЖИМ МОНИТОРИНГА (вопросы о состоянии)
# ============================================================
# Переименование
if "переимен" in cmd:
# Ожидаемый формат: "переименуй <entity_id или имя> в <новое имя>"
# или: "переименуй <entity_id> как <новое имя>"
m = re.search(r"переимен\w*\s+(.+?)\s+(?:в|как|на)\s+(.+)$", cmd)
if m:
target = m.group(1).strip().strip('"').strip("'")
new_name = m.group(2).strip().strip('"').strip("'")
return self.rename_entity(
entity_id=target if "." in target else "",
name=new_name,
friendly_name=target if "." not in target else "",
)
return (
"Формат: `переименуй <entity_id или имя> в <новое имя>`\n"
"Пример: `переименуй automation.сертификат в Обновление SSL`"
)
# Включение/выключение автоматизации
if "автоматизац" in cmd and any(w in cmd for w in ("включ", "выключ", "активир", "деактивир")):
target = ""
for s in self._get_states(force=True):
eid = s["entity_id"]
if not eid.startswith("automation."):
continue
fn = (self._attr(s, "friendly_name", "") or "").lower()
if fn and fn in cmd:
target = eid
break
if target:
turn_off = any(w in cmd for w in ("выключ", "деактивир"))
return self.toggle_automation(entity_id=target, turn_off=turn_off)
warnings.append("автоматизация не найдена в команде")
is_question = (
cmd.startswith(("как", "сколько", "что", "кто", "где", "какая", "какой", "какое", "какие"))
or "?" in cmd
@@ -632,33 +622,26 @@ class Tools:
if any(k in cmd for k in ["окн", "двер", "форточ", "калитк", "ворота", "гараж"]):
actions.extend(self._monitor_windows_doors(tokens))
matched = True
elif any(k in cmd for k in ["свет", "лампочк", "люстр"]) and (
"горит" in cmd or "включ" in cmd or "лишн" in cmd
):
actions.extend(self._monitor_lights(tokens))
matched = True
elif any(k in cmd for k in ["воздух", "tvoc", "co2", "влажн", "качеств", "pm2", "pm10"]):
actions.extend(self._monitor_air_quality(tokens))
matched = True
elif any(k in cmd for k in ["движен", "двигает", "хот", "кто-то ход"]):
actions.extend(self._monitor_motion(tokens))
matched = True
elif any(k in cmd for k in ["температур", "климат", "тепл", "холод", "кондиционер", "батаре"]):
actions.extend(self._monitor_climate(tokens))
matched = True
elif "батарейк" in cmd or "заряд" in cmd:
actions.extend(self._monitor_battery(tokens))
matched = True
elif any(k in cmd for k in ["музык", "медиа", "плеер", "колонк", "телевизор"]):
actions.extend(self._monitor_media(tokens))
matched = True
elif "дома" in cmd or "присутств" in cmd:
actions.extend(self._monitor_presence(tokens))
matched = True
@@ -667,7 +650,6 @@ class Tools:
actions.append("**Общий статус дома:**")
actions.extend(self._monitor_lights([]))
actions.extend(self._monitor_climate([]))
else:
actions.extend(self._handle_control(cmd, tokens, warnings))
@@ -686,6 +668,8 @@ class Tools:
" - кто дома?\n"
" - батарейки садятся?\n"
" - покажи все binary_sensor\n"
" - переименуй automation.сертификат в Обновление SSL\n"
" - выключи автоматизацию Гирлянда крыльцо\n"
" - отчёт по дому"
)
if warnings:
@@ -708,7 +692,6 @@ class Tools:
def _handle_control(
self, cmd: str, tokens: List[str], warnings: List[str]
) -> List[str]:
"""Обработка команд действия (вкл/выкл/установить/открыть/...)."""
actions: List[str] = []
# Свет
@@ -759,7 +742,7 @@ class Tools:
label = "выключен" if turn_off else "включён"
actions.append("свет " + label + " (" + str(count) + " шт.)")
# Розетки / выключатели / реле
# Розетки / реле / выключатели
if any(k in cmd for k in ["розетк", "выключател", "устройств", "реле"]):
entities = self._find_entities(tokens, "switch")
if not entities:
@@ -832,21 +815,18 @@ class Tools:
count += 1
if count:
actions.append("пауза (" + str(count) + " шт.)")
elif "дальше" in cmd or "следующ" in cmd:
for e in entities:
self._call_service("media_player", "media_next_track", {"entity_id": e})
count += 1
if count:
actions.append("следующий трек")
elif "назад" in cmd or "предыдущ" in cmd:
for e in entities:
self._call_service("media_player", "media_previous_track", {"entity_id": e})
count += 1
if count:
actions.append("предыдущий трек")
elif "громче" in cmd or "тише" in cmd:
for e in entities:
self._call_service(
@@ -857,7 +837,6 @@ class Tools:
count += 1
if count:
actions.append("громкость (" + str(count) + " шт.)")
else:
for e in entities:
if self._get_state(e) != "playing":
@@ -866,7 +845,7 @@ class Tools:
if count:
actions.append("воспроизведение (" + str(count) + " шт.)")
# Шторы / жалюзи
# Шторы
if any(k in cmd for k in ["штор", "жалюз", "роллет", "гардин"]):
entities = self._find_entities(tokens, "cover")
if not entities:
@@ -931,7 +910,7 @@ class Tools:
self._call_service("vacuum", "start", {"entity_id": e})
actions.append("пылесос уборка")
# Кнопки (button.press)
# Кнопки
if "кнопк" in cmd and any(w in cmd for w in ("нажм", "кликн", "тапн", "триггерн", "активир")):
entities = self._find_entities(tokens, "button")
if entities:
@@ -1006,7 +985,6 @@ class Tools:
device_class = device_class.lower().strip()
only_with_state = only_with_state.lower().strip()
# Фильтрация
filtered: List[Dict[str, Any]] = []
for s in states:
eid = s["entity_id"]
@@ -1014,7 +992,6 @@ class Tools:
cur_state = (s.get("state", "") or "").lower()
dc = (attrs.get("device_class", "") or "").lower()
# Фильтр по домену
if domain:
if domain == "window_door":
if dc not in ("door", "window", "garage_door", "lock") and \
@@ -1025,15 +1002,10 @@ class Tools:
if not eid.startswith(domain + "."):
continue
# Фильтр по device_class
if device_class and dc != device_class:
continue
# Фильтр по состоянию
if only_with_state and cur_state != only_with_state:
continue
# Скрыть unavailable/unknown
if not show_unavailable and cur_state in ("unavailable", "unknown", "none"):
continue
@@ -1047,7 +1019,6 @@ class Tools:
+ ", state=" + (only_with_state or "любой") + ")."
)
# Группируем по домену
by_domain: Dict[str, List[Dict[str, Any]]] = {}
for s in filtered:
d = s["entity_id"].split(".", 1)[0]
@@ -1065,3 +1036,146 @@ class Tools:
lines.append("... показано " + str(limit) + " из " + str(len(filtered)) + ". Увеличьте `limit` для полного списка.")
return "\n".join(lines)
# ========================================================================
# 🏷 Переименование сущностей
# ========================================================================
def rename_entity(
self,
entity_id: str = "",
name: str = "",
friendly_name: str = "",
) -> str:
"""
Переименовать сущность в Home Assistant (меняет friendly_name).
Работает для automation, script, light, switch, scene и др.
Можно указать либо entity_id (точное имя), либо friendly_name (для поиска).
Args:
entity_id: точный id сущности (например "automation.сертификат").
Если пусто — ищем по friendly_name.
name: новое имя.
friendly_name: текущее friendly_name для поиска (если entity_id не указан).
Returns:
Подтверждение или сообщение об ошибке.
"""
if not self.valves.ha_token:
return "Не задан HA Token (Valves)."
if not name:
return "Укажите новое имя (параметр name)."
target = entity_id
if not target or "." not in target:
# Поиск по friendly_name
search_name = friendly_name or entity_id
target = self._find_entity_by_name(search_name)
if not target:
return "Не нашёл сущность с именем: " + search_name
if self.valves.dry_run:
return f"[DRY-RUN] Переименовал бы {target} -> {name}"
try:
self._request(
"POST",
"/api/services/homeassistant/update_entity",
json={"entity_id": target, "name": name},
)
self._get_states(force=True) # сброс кэша
return f"Переименовано: {target} -> '{name}'"
except Exception as e:
return f"Ошибка переименования {target}: {e}"
# ========================================================================
# ⚙️ Управление автоматизациями
# ========================================================================
def toggle_automation(
self,
entity_id: str = "",
friendly_name: str = "",
turn_off: bool = False,
) -> str:
"""
Включить или выключить автоматизацию.
Args:
entity_id: точный id (например "automation.сертификат").
Если пусто — ищем по friendly_name.
friendly_name: имя для поиска.
turn_off: True — выключить, False — включить.
Returns:
Подтверждение или сообщение об ошибке.
"""
if not self.valves.ha_token:
return "Не задан HA Token (Valves)."
target = entity_id
if not target or "." not in target:
search_name = friendly_name or entity_id
target = self._find_entity_by_name(search_name, domain="automation")
if not target:
return "Не нашёл автоматизацию: " + search_name
service = "turn_off" if turn_off else "turn_on"
action_label = "выключена" if turn_off else "включена"
if self.valves.dry_run:
return f"[DRY-RUN] Автоматизация {target} была бы {action_label}"
try:
self._request(
"POST",
"/api/services/automation/" + service,
json={"entity_id": target},
)
self._get_states(force=True)
return f"Автоматизация {target} {action_label}"
except Exception as e:
return f"Ошибка: {e}"
def trigger_automation(
self,
entity_id: str = "",
friendly_name: str = "",
skip_condition: bool = True,
) -> str:
"""
Запустить автоматизацию вручную (триггернуть).
Args:
entity_id: точный id (например "automation.сертификат").
Если пусто — ищем по friendly_name.
friendly_name: имя для поиска.
skip_condition: пропустить ли условия (по умолчанию True).
Returns:
Подтверждение или сообщение об ошибке.
"""
if not self.valves.ha_token:
return "Не задан HA Token (Valves)."
target = entity_id
if not target or "." not in target:
search_name = friendly_name or entity_id
target = self._find_entity_by_name(search_name, domain="automation")
if not target:
return "Не нашёл автоматизацию: " + search_name
if self.valves.dry_run:
return f"[DRY-RUN] Триггернул бы {target}"
try:
data: Dict[str, Any] = {"entity_id": target}
if skip_condition:
data["skip_condition"] = True
self._request(
"POST",
"/api/services/automation/trigger",
json=data,
)
return f"Триггер запущен: {target}"
except Exception as e:
return f"Ошибка: {e}"