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

v2.5.1 — фикс фильтра окон/дверей: строго по device_class (door/window/garage_door/lock)

Browse Source
This commit is contained in:
host
2026-07-29 20:06:17 +03:00
parent 410be3f3f0
commit 6c1c1a11ed
1 changed files with 41 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
home_assistant_tools.py
+41 -156
View File
@@ -2,7 +2,7 @@
title: Home Assistant Smart Control
description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка.
author: ChatGPT (refactored by MiniMax-M3)
version: 2.5.0
version: 2.5.1
license: MIT
"""
@@ -17,41 +17,20 @@ logger = logging.getLogger(__name__)
class Tools:
# ========================================================================
# Конфигурация (Valves - задаётся в UI Open WebUI)
# ========================================================================
class Valves(BaseModel):
ha_url: str = Field(
default="https://ha.nasweb.ru",
description="URL Home Assistant (например https://ha.nasweb.ru)",
)
ha_token: str = Field(
default="",
description="Long-Lived Access Token из профиля HA",
)
verify_ssl: bool = Field(
default=False,
description="Проверять SSL-сертификат HA (false для самоподписанных)",
)
ha_url: str = Field(default="https://ha.nasweb.ru", description="URL Home Assistant")
ha_token: str = Field(default="", description="Long-Lived Access Token из профиля HA")
verify_ssl: bool = Field(default=False, description="Проверять SSL-сертификат HA")
request_timeout: int = Field(default=10, description="Таймаут HTTP-запроса, сек")
max_retries: int = Field(default=2, description="Количество ретраев при сетевых сбоях")
cache_ttl: int = Field(default=5, description="TTL кэша состояний HA, сек")
confirm_global_actions: bool = Field(
default=True,
description="Спрашивать подтверждение для команд на ВСЕ сущности домена",
)
dry_run: bool = Field(
default=False,
description="Не выполнять команды, только показывать что БУДЕТ сделано",
)
confirm_global_actions: bool = Field(default=True, description="Спрашивать подтверждение для команд на ВСЕ сущности домена")
dry_run: bool = Field(default=False, description="Не выполнять команды, только показывать что БУДЕТ сделано")
def __init__(self):
self.valves = self.Valves()
self._states_cache: Optional[Tuple[float, List[Dict[str, Any]]]] = None
# ========================================================================
# Стоп-слова и ключевые слова
# ========================================================================
_STOP_WORDS = {
"включи", "выключи", "поставь", "установи", "сделай", "запусти",
"на", "в", "во", "мне", "пожалуйста", "давай", "ну",
@@ -99,9 +78,11 @@ class Tools:
_MONITOR_DOMAINS = {
"window_door": {
"domains": ["binary_sensor"],
"keywords": ["окн", "двер", "форточ", "калитк", "ворота", "гараж"],
"open_states": ("open", "on", "unlocked", "detected"),
"closed_states": ("closed", "off", "locked", "clear"),
# Допустимые device_class для окон/дверей/замков.
# Если в HA не задан device_class — сущность игнорируется.
"valid_device_classes": ("door", "window", "garage_door", "lock"),
},
"light_status": {
"domains": ["light"],
@@ -136,9 +117,6 @@ class Tools:
},
}
# ========================================================================
# HTTP / API
# ========================================================================
def _headers(self) -> Dict[str, str]:
return {
"Authorization": "Bearer " + self.valves.ha_token,
@@ -146,7 +124,6 @@ class Tools:
}
def _request(self, method: str, path: str, **kwargs) -> requests.Response:
"""HTTP-запрос с ретраями и логированием."""
url = self.valves.ha_url + path
last_exc: Optional[Exception] = None
kwargs.setdefault("verify", self.valves.verify_ssl)
@@ -171,7 +148,6 @@ class Tools:
raise RuntimeError("Home Assistant недоступен: " + str(last_exc))
def _get_states(self, force: bool = False) -> List[Dict[str, Any]]:
"""Получить состояния HA с TTL-кэшем."""
now = time.time()
if (
not force
@@ -186,7 +162,6 @@ class Tools:
return states
def _call_service(self, domain: str, service: str, data: Dict[str, Any]) -> bool:
"""Вызов сервиса HA. Возвращает True если выполнено, False если dry-run."""
logger.info(
"%s -> %s.%s %s",
"[DRY] " if self.valves.dry_run else "",
@@ -197,17 +172,12 @@ class Tools:
self._request("POST", "/api/services/" + domain + "/" + service, json=data)
return True
# ========================================================================
# Поиск сущностей
# ========================================================================
def _tokenize(self, text: str) -> List[str]:
"""Токенизация: фильтр стоп-слов, приведение к нижнему регистру."""
text = text.lower().strip().replace("ё", "е")
tokens = re.findall(r"[а-яa-z0-9]+", text)
return [t for t in tokens if t not in self._STOP_WORDS and len(t) >= 2]
def _entity_text(self, s: Dict[str, Any]) -> str:
"""Полный текст сущности: id + friendly_name + area."""
attrs = s.get("attributes", {}) or {}
return " ".join([
s.get("entity_id", ""),
@@ -222,10 +192,8 @@ class Tools:
domain: Optional[str] = None,
require_all: bool = False,
) -> List[str]:
"""Поиск сущностей по ключевым словам."""
states = self._get_states()
result: List[str] = []
for s in states:
eid = s["entity_id"]
if domain and not eid.startswith(domain + "."):
@@ -238,22 +206,18 @@ class Tools:
matches = sum(1 for kw in keywords if kw in haystack)
if (require_all and matches == len(keywords)) or (not require_all and matches > 0):
result.append(eid)
return result
def _get_state(self, entity_id: str) -> Optional[str]:
"""Получить текущее состояние конкретной сущности."""
for s in self._get_states():
if s["entity_id"] == entity_id:
return s.get("state")
return None
def _attr(self, s: Dict[str, Any], key: str, default=None) -> Any:
"""Безопасное извлечение атрибута."""
return (s.get("attributes", {}) or {}).get(key, default)
def _resolve_domain(self, cmd: str) -> Optional[str]:
"""Определить домен HA по русской команде."""
for domain, aliases in self._DOMAIN_ALIASES.items():
for a in aliases:
if a in cmd:
@@ -261,7 +225,6 @@ class Tools:
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:
@@ -271,16 +234,11 @@ class Tools:
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
# ========================================================================
# Парсинг чисел
# ========================================================================
def _extract_number(self, text: str, default: Optional[float] = None) -> Optional[float]:
"""Извлечь число из текста: арабские цифры или русские слова."""
m = re.search(r"-?\d+(?:[.,]\d+)?", text)
if m:
return float(m.group(0).replace(",", "."))
@@ -290,11 +248,7 @@ class Tools:
return float(self._RU_NUMS[word])
return default
# ========================================================================
# Форматирование ответа
# ========================================================================
def _format_entity(self, s: Dict[str, Any]) -> str:
"""Красиво отформатировать одну сущность для вывода."""
attrs = s.get("attributes", {}) or {}
fn = attrs.get("friendly_name", s["entity_id"])
state = s.get("state", "?")
@@ -307,17 +261,24 @@ class Tools:
# Мониторинг
# ========================================================================
def _monitor_windows_doors(self, tokens: List[str]) -> List[str]:
"""
Какие окна/двери открыты.
Строгий фильтр: только binary_sensor с device_class в
(door, window, garage_door, lock). Это исключает ложные
срабатывания на датчики движения, батареи штор, шторы Aqara и т.п.
"""
cfg = self._MONITOR_DOMAINS["window_door"]
valid_dc = cfg["valid_device_classes"]
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.")
if not eid.startswith("binary_sensor."):
continue
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):
if device_class not in valid_dc:
continue
cur = s.get("state", "")
is_open = cur in cfg["open_states"]
@@ -325,7 +286,7 @@ class Tools:
items.append((s, is_open, is_closed))
if not items:
return ["датчики окон/дверей не найдены"]
return ["датчики окон/дверей не найдены (проверьте device_class в HA)"]
open_items = [s for s, is_open, _ in items if is_open]
closed_items = [s for s, _, is_closed in items if is_closed]
@@ -340,6 +301,7 @@ class Tools:
lines.append("Закрыто: " + str(len(closed_items)) + " шт.")
if unknown:
lines.append("Неизвестно: " + str(len(unknown)) + " шт.")
if tokens:
filtered = []
for line in lines:
@@ -347,6 +309,7 @@ class Tools:
filtered.append(line)
if filtered:
return filtered
return lines
def _monitor_lights(self, tokens: List[str]) -> List[str]:
@@ -549,7 +512,6 @@ class Tools:
# Главная точка входа
# ========================================================================
def smart(self, command: str) -> str:
"""Принимает команду на естественном языке и управляет домом / отвечает на вопросы."""
if not self.valves.ha_token:
return (
"Не задан HA Token. Укажите его в настройках инструмента (Valves) "
@@ -562,7 +524,6 @@ class Tools:
warnings: List[str] = []
try:
# Маршрутизация 'покажи все / список устройств' -> list_devices
if any(w in cmd for w in [
"покажи", "список устройств", "какие устройств",
"что есть", "инвентар", "все устройств",
@@ -570,14 +531,10 @@ 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("'")
@@ -592,7 +549,6 @@ class Tools:
"Пример: `переименуй automation.сертификат в Обновление SSL`"
)
# Включение/выключение автоматизации
if "автоматизац" in cmd and any(w in cmd for w in ("включ", "выключ", "активир", "деактивир")):
target = ""
for s in self._get_states(force=True):
@@ -694,7 +650,6 @@ class Tools:
) -> List[str]:
actions: List[str] = []
# Свет
if "свет" in cmd or "лампочк" in cmd:
entities = self._find_entities(tokens, "light")
if not entities:
@@ -742,7 +697,6 @@ 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:
@@ -766,7 +720,6 @@ class Tools:
label = "выключены" if turn_off else "включены"
actions.append("розетки " + label + " (" + str(count) + " шт.)")
# Климат
if any(k in cmd for k in ["температур", "градус", "тепл", "холод", "климат"]):
temp = self._extract_number(cmd, default=22)
entities = self._find_entities(tokens, "climate")
@@ -801,7 +754,6 @@ class Tools:
)
actions.append("режим охлаждения")
# Медиа
if any(k in cmd for k in ["музык", "медиа", "плеер", "колонк"]):
entities = self._find_entities(tokens, "media_player")
if not entities:
@@ -845,7 +797,6 @@ 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:
@@ -863,7 +814,6 @@ class Tools:
label = "закрыты" if is_close else "открыты"
actions.append("шторы " + label + " (" + str(count) + " шт.)")
# Замки
if any(k in cmd for k in ["замок", "заблокир", "закрыт"]):
entities = self._find_entities(tokens, "lock")
if entities:
@@ -878,7 +828,6 @@ class Tools:
"замки " + ("закрыты" if is_lock else "открыты") + " (" + str(len(entities)) + " шт.)"
)
# Вентиляторы
if "вентилятор" in cmd or "кулер" in cmd:
entities = self._find_entities(tokens, "fan")
if entities:
@@ -893,7 +842,6 @@ class Tools:
"вентиляторы " + ("выключены" if is_off else "включены") + " (" + str(len(entities)) + " шт.)"
)
# Пылесосы
if "пылесос" in cmd or "робот" in cmd:
entities = self._find_entities(tokens, "vacuum")
if entities:
@@ -910,7 +858,6 @@ class Tools:
self._call_service("vacuum", "start", {"entity_id": e})
actions.append("пылесос уборка")
# Кнопки
if "кнопк" in cmd and any(w in cmd for w in ("нажм", "кликн", "тапн", "триггерн", "активир")):
entities = self._find_entities(tokens, "button")
if entities:
@@ -922,7 +869,6 @@ class Tools:
else:
warnings.append("кнопки не найдены")
# Сцены / скрипты
if any(k in cmd for k in ["сцен", "режим", "запуст"]):
all_scripts = self._find_entities([], "script")
matched: List[str] = []
@@ -942,7 +888,7 @@ class Tools:
return actions
# ========================================================================
# Диагностика / справочник устройств (отдельный инструмент)
# Диагностика / справочник устройств
# ========================================================================
def list_devices(
self,
@@ -952,30 +898,8 @@ class Tools:
show_unavailable: bool = True,
limit: int = 50,
) -> str:
"""
Получить список устройств из Home Assistant с фильтрами.
Используется для инвентаризации: какие устройства есть, в каком состоянии,
какие из них offline. Полезно для отладки и проверки интеграций.
Args:
domain: фильтр по домену (light, switch, binary_sensor, sensor, climate, ...).
Спецзначение "window_door" — окна и двери.
Пусто — все домены.
device_class: фильтр по device_class атрибуту
(door, window, motion, battery, temperature, ...).
Пусто — все классы.
only_with_state: оставить только с указанным состоянием (on/off/open/closed/...).
Пусто — любые.
show_unavailable: показывать ли устройства со state="unavailable"/"unknown".
limit: максимум устройств в ответе.
Returns:
Список устройств в формате markdown, сгруппированный по доменам.
"""
if not self.valves.ha_token:
return "Не задан HA Token (Valves)."
try:
states = self._get_states(force=True)
except Exception as e:
@@ -994,9 +918,10 @@ class Tools:
if domain:
if domain == "window_door":
if dc not in ("door", "window", "garage_door", "lock") and \
not any(k in (eid + " " + (attrs.get("friendly_name", "") or "").lower())
for k in ["окн", "двер", "форточ", "балкон", "калитк", "ворота", "гараж"]):
# Строгий фильтр: только binary_sensor с правильным device_class
if not eid.startswith("binary_sensor."):
continue
if dc not in ("door", "window", "garage_door", "lock"):
continue
else:
if not eid.startswith(domain + "."):
@@ -1038,7 +963,7 @@ class Tools:
return "\n".join(lines)
# ========================================================================
# 🏷 Переименование сущностей
# Переименование
# ========================================================================
def rename_entity(
self,
@@ -1046,21 +971,6 @@ class Tools:
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:
@@ -1068,14 +978,13 @@ class Tools:
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}"
return "[DRY-RUN] Переименовал бы " + target + " -> " + name
try:
self._request(
@@ -1083,13 +992,13 @@ class Tools:
"/api/services/homeassistant/update_entity",
json={"entity_id": target, "name": name},
)
self._get_states(force=True) # сброс кэша
return f"Переименовано: {target} -> '{name}'"
self._get_states(force=True)
return "Переименовано: " + target + " -> '" + name + "'"
except Exception as e:
return f"Ошибка переименования {target}: {e}"
return "Ошибка переименования " + target + ": " + str(e)
# ========================================================================
# ⚙️ Управление автоматизациями
# Управление автоматизациями
# ========================================================================
def toggle_automation(
self,
@@ -1097,18 +1006,6 @@ class Tools:
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)."
@@ -1123,7 +1020,7 @@ class Tools:
action_label = "выключена" if turn_off else "включена"
if self.valves.dry_run:
return f"[DRY-RUN] Автоматизация {target} была бы {action_label}"
return "[DRY-RUN] Автоматизация " + target + " была бы " + action_label
try:
self._request(
@@ -1132,9 +1029,9 @@ class Tools:
json={"entity_id": target},
)
self._get_states(force=True)
return f"Автоматизация {target} {action_label}"
return "Автоматизация " + target + " " + action_label
except Exception as e:
return f"Ошибка: {e}"
return "Ошибка: " + str(e)
def trigger_automation(
self,
@@ -1142,18 +1039,6 @@ class Tools:
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)."
@@ -1165,7 +1050,7 @@ class Tools:
return "Не нашёл автоматизацию: " + search_name
if self.valves.dry_run:
return f"[DRY-RUN] Триггернул бы {target}"
return "[DRY-RUN] Триггернул бы " + target
try:
data: Dict[str, Any] = {"entity_id": target}
@@ -1176,6 +1061,6 @@ class Tools:
"/api/services/automation/trigger",
json=data,
)
return f"Триггер запущен: {target}"
return "Триггер запущен: " + target
except Exception as e:
return f"Ошибка: {e}"
return "Ошибка: " + str(e)