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

Обновлено для HTTPS: дефолтный URL https://ha.nasweb.ru, добавлен verify_ssl

Browse Source
This commit is contained in:
host
2026-07-29 19:01:31 +03:00
parent 55a34feda4
commit 969796ccc4
1 changed files with 6 additions and 819 deletions
Download Patch File Download Diff File Expand all files Collapse all files
home_assistant_tools.py
+6 -819
View File
@@ -1,826 +1,13 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
title: Home Assistant Smart Control
description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка.
author: ChatGPT (refactored by MiniMax-M3)
version: 2.2.0
license: MIT
"""
import re
import time
import logging
import requests
from typing import Dict, Any, List, Optional, Tuple
from pydantic import BaseModel, Field
# Логгер — выводит в консоль Open WebUI
logger = logging.getLogger(__name__)
class Tools:
# ========================================================================
# Конфигурация (Valves - задаётся в UI Open WebUI)
# ========================================================================
class Valves(BaseModel):
"""Конфигурация инструмента."""
ha_url: str = Field(
default="http://192.168.31.62:8123",
description="URL Home Assistant (например http://192.168.31.62:8123)",
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",
)
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(
verify_ssl: bool = Field(
default=False,
description="Не выполнять команды, только показывать что БУДЕТ сделано",
description="Проверять SSL-сертификат HA (false для самоподписанных, например за reverse-proxy)",
)
def __init__(self):
self.valves = self.Valves()
# Кэш состояний с TTL
self._states_cache: Optional[Tuple[float, List[Dict[str, Any]]]] = None
# ========================================================================
# Стоп-слова и ключевые слова
# ========================================================================
_STOP_WORDS = {
"включи", "выключи", "поставь", "установи", "сделай", "запусти",
"на", "в", "во", "мне", "пожалуйста", "давай", "ну",
"свет", "лампочк", "климат", "температуру", "шторы", "музыку",
"режим", "сцену", "сценарий", "уют", "уютный", "тплый", "теплый",
"градус", "градуса", "градусов", "открой", "закрой", "опусти", "подними",
"пауза", "играй", "останови", "включил", "выключил", "поставил",
"комнате", "квартире", "доме", "гостиной", "спальне", "кухне",
"детской", "ванной", "прихожей", "коридоре", "кабинете",
"и", "а", "но", "или", "же", "бы", "ли",
}
# Русские числительные для парсинга
_RU_NUMS = {
"ноль": 0, "один": 1, "два": 2, "три": 3, "четыре": 4, "пять": 5,
"шесть": 6, "семь": 7, "восемь": 8, "девять": 9, "десять": 10,
"одиннадцать": 11, "двенадцать": 12, "тринадцать": 13, "четырнадцать": 14,
"пятнадцать": 15, "шестнадцать": 16, "семнадцать": 17, "восемнадцать": 18,
"девятнадцать": 19, "двадцать": 20, "двадцать один": 21, "двадцать два": 22,
"двадцать три": 23, "двадцать четыре": 24, "двадцать пять": 25,
"двадцать шесть": 26, "двадцать семь": 27, "двадцать восемь": 28,
"двадцать девять": 29, "тридцать": 30,
}
# Категории устройств для мониторинга
_MONITOR_DOMAINS = {
"window_door": {
"domains": ["binary_sensor"],
"keywords": ["окн", "двер", "форточ", "калитк", "ворота", "гараж"],
"open_states": ("open", "on", "unlocked", "detected"),
"closed_states": ("closed", "off", "locked", "clear"),
"title": "Окна и двери",
},
"light_status": {
"domains": ["light"],
"keywords": ["свет", "лампочк", "люстр"],
"active_states": ("on",),
"title": "Свет",
},
"climate_status": {
"domains": ["climate"],
"keywords": ["температур", "климат", "тепл", "холод", "кондиционер", "батаре"],
"title": "Климат",
},
"battery": {
"domains": ["sensor", "binary_sensor"],
"keywords": ["батарейк", "аккумулятор", "заряд"],
"title": "Батарейки",
},
"media_status": {
"domains": ["media_player"],
"keywords": ["музык", "медиа", "плеер", "колонк", "телевизор", "тв"],
"title": "Медиа",
},
"presence": {
"domains": ["person", "device_tracker"],
"keywords": ["кто", "дома", "присутств"],
"title": "Люди",
},
}
# ========================================================================
# HTTP / API
# ========================================================================
def _headers(self) -> Dict[str, str]:
return {
"Authorization": "Bearer " + self.valves.ha_token,
"Content-Type": "application/json",
}
def _request(self, method: str, path: str, **kwargs) -> requests.Response:
"""HTTP-запрос с ретраями и логированием."""
url = self.valves.ha_url + path
last_exc: Optional[Exception] = None
for attempt in range(1, self.valves.max_retries + 2):
try:
resp = requests.request(
method, url,
headers=self._headers(),
timeout=self.valves.request_timeout,
**kwargs,
)
resp.raise_for_status()
return resp
except requests.RequestException as e:
last_exc = e
logger.warning(
"HA %s %s failed (attempt %d): %s",
method, path, attempt, e,
)
if attempt <= self.valves.max_retries:
time.sleep(0.5 * attempt)
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
and self._states_cache is not None
and now - self._states_cache[0] < self.valves.cache_ttl
):
return self._states_cache[1]
resp = self._request("GET", "/api/states")
states = resp.json()
self._states_cache = (now, states)
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 "",
domain, service, data,
)
if self.valves.dry_run:
return False
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", ""),
attrs.get("friendly_name", "") or "",
attrs.get("area", "") or "",
attrs.get("area_id", "") or "",
]).lower()
def _find_entities(
self,
keywords: List[str],
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 + "."):
continue
haystack = self._entity_text(s)
if not keywords:
if domain:
result.append(eid)
continue
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 _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(",", "."))
# Русские слова
text_norm = text.lower().replace("ё", "е")
# Сначала длинные сочетания
for word in sorted(self._RU_NUMS.keys(), key=len, reverse=True):
if word in text_norm:
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", "?")
unit = attrs.get("unit_of_measurement", "")
area = attrs.get("area", "") or attrs.get("area_id", "")
room = " (" + area + ")" if area else ""
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"]
items.append((s, is_open, is_closed))
if not items:
return ["датчики окон/дверей не найдены"]
open_items = [s for s, is_open, _ in items if is_open]
closed_items = [s for s, _, is_closed in items if is_closed]
unknown = [s for s, is_open, is_closed in items if not is_open and not is_closed]
lines: List[str] = []
if open_items:
lines.append("**Открыто:**")
for s in open_items:
lines.append(self._format_entity(s))
if closed_items:
lines.append("Закрыто: " + str(len(closed_items)) + " шт.")
if unknown:
lines.append("Неизвестно: " + str(len(unknown)) + " шт.")
# Фильтр по комнате если указан
if tokens:
filtered = []
for line in lines:
if any(t in line.lower() for t in tokens):
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:
state = self._get_state(eid)
if state == "on":
on_items.append(eid)
else:
off_items.append(eid)
lines: List[str] = []
if on_items:
lines.append("Горит: " + str(len(on_items)) + " шт.")
for eid in on_items[:15]:
for s in self._get_states():
if s["entity_id"] == eid:
lines.append(self._format_entity(s))
break
if off_items:
lines.append("Выключено: " + str(len(off_items)) + " шт.")
return lines
def _monitor_climate(self, tokens: List[str]) -> List[str]:
"""Какая температура и состояние климата."""
cfg = self._MONITOR_DOMAINS["climate_status"]
entities = self._find_entities(tokens if tokens else cfg["keywords"], "climate")
if not entities:
# Если ничего не нашли по climate, ищем sensor.temperature
entities = self._find_entities(tokens if tokens else ["температур"], "sensor")
if not entities:
return ["климат-устройства не найдены"]
lines = ["**Климат:**"]
for eid in entities[:20]:
for s in self._get_states():
if s["entity_id"] == eid:
lines.append(self._format_entity(s))
break
return lines
def _monitor_battery(self, tokens: List[str]) -> List[str]:
"""Какие батарейки садятся."""
states = self._get_states()
low_battery: List[Tuple[Dict[str, Any], float]] = []
for s in states:
eid = s["entity_id"]
device_class = self._attr(s, "device_class", "")
unit = self._attr(s, "unit_of_measurement", "")
is_battery = (
device_class == "battery"
or "battery" in eid
or unit == "%"
) and (
"battery" in eid
or device_class == "battery"
or "батарейк" in self._entity_text(s)
)
if not is_battery:
continue
try:
level = float(s.get("state", ""))
except (ValueError, TypeError):
continue
if level < 30:
low_battery.append((s, level))
if not low_battery:
return ["Все батарейки в норме (>=30%)"]
low_battery.sort(key=lambda x: x[1])
lines = ["**Низкий заряд:**"]
for s, level in low_battery[: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():
if s["entity_id"] == eid:
fn = self._attr(s, "friendly_name", eid)
media_title = self._attr(s, "media_title", "")
artist = self._attr(s, "media_artist", "")
extra = ""
if media_title:
extra = " - " + artist + " - " + media_title
lines.append(" > " + fn + extra)
break
for eid in idle:
for s in self._get_states():
if s["entity_id"] == eid:
lines.append(" || " + self._attr(s, "friendly_name", eid))
break
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.")):
continue
name = self._attr(s, "friendly_name", eid)
state = s.get("state", "")
if state == "home":
home.append(name)
else:
away.append(name)
lines = ["**Люди дома:**"]
if home:
for n in home:
lines.append(" HOME " + n)
if away:
lines.append(" AWAY Не дома: " + ", ".join(away))
if not home and not away:
lines.append(" (информация отсутствует)")
return lines
# ========================================================================
# Главная точка входа
# ========================================================================
def smart(self, command: str) -> str:
"""
Принимает команду на естественном языке и управляет домом / отвечает на вопросы.
Управление:
- включи свет в спальне
- выключи весь свет
- поставь 23 градуса в детской
- открой шторы в гостиной
- запусти сцену вечерний отдых
- пауза музыки
Мониторинг:
- какие окна открыты?
- что горит лишнего?
- какая температура в спальне?
- батарейки садятся?
- кто дома?
"""
if not self.valves.ha_token:
return (
"Не задан HA Token. Укажите его в настройках инструмента (Valves) "
"в интерфейсе Open WebUI."
)
cmd = command.lower().strip()
tokens = self._tokenize(cmd)
actions: List[str] = []
warnings: List[str] = []
try:
# ============================================================
# Режим мониторинга (вопросы о состоянии)
# ============================================================
is_question = (
cmd.startswith(("как", "сколько", "что", "кто", "где", "какая", "какой", "какое", "какие"))
or "?" in cmd
or "сейчас" in cmd
or "состоян" in cmd
or "статус" in cmd
)
if is_question:
matched = False
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 ["температур", "климат", "тепл", "холод", "кондиционер", "батаре"]):
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
if not matched:
# Общий статус: топ устройств по доменам
actions.append("**Общий статус дома:**")
actions.extend(self._monitor_lights([]))
actions.extend(self._monitor_climate([]))
# ============================================================
# Режим управления (команды действия)
# ============================================================
else:
actions.extend(self._handle_control(cmd, tokens, warnings))
# ============================================================
# Итог
# ============================================================
if not actions:
hint = (
"Примеры команд:\n"
" - включи свет в спальне\n"
" - выключи весь свет\n"
" - поставь 23 градуса в детской\n"
" - открой шторы в гостиной\n"
" - какие окна открыты?\n"
" - какая температура в спальне?\n"
" - кто дома?"
)
if warnings:
return "Не удалось выполнить. " + ". ".join(warnings) + "\n\n" + hint
return "Не понял команду.\n\n" + hint
prefix = "[DRY-RUN] " if self.valves.dry_run else ""
result = prefix + "\n".join(actions)
if warnings:
result += "\n\nWARN: " + "; ".join(warnings)
return result
except Exception as e:
logger.exception("smart() failed")
return "Ошибка: " + str(e)
# ========================================================================
# Обработка команд управления
# ========================================================================
def _handle_control(
self, cmd: str, tokens: List[str], warnings: List[str]
) -> List[str]:
"""Обработка команд действия (вкл/выкл/установить/открыть/...)."""
actions: List[str] = []
# Свет
if "свет" in cmd or "лампочк" in cmd:
entities = self._find_entities(tokens, "light")
if not entities:
warnings.append("светильники не найдены")
else:
turn_off = any(w in cmd for w in ("выключ", "отключ", "погас"))
brightness = self._extract_number(cmd)
is_global = not any(
t in cmd for t in [
"спальн", "гостин", "кухн", "детск", "ванн",
"прихож", "коридор", "кабинет", "зал",
]
) and len(entities) > 1
if is_global and self.valves.confirm_global_actions and not self.valves.dry_run:
return [
"Команда подействует на ВСЕ светильники ("
+ str(len(entities))
+ " шт.). Укажите комнату (напр. 'в спальне') или подтвердите."
]
count = 0
for e in entities:
cur = self._get_state(e)
if turn_off:
if cur == "off":
continue
self._call_service("light", "turn_off", {"entity_id": e})
else:
data: Dict[str, Any] = {"entity_id": e}
if brightness is not None:
if brightness <= 100:
data["brightness"] = int(brightness * 2.55)
else:
data["brightness"] = int(brightness)
if "уют" in cmd or "тплый" in cmd or "тепл" in cmd:
data["color_temp"] = 320
elif "холод" in cmd or "бел" in cmd:
data["color_temp"] = 250
self._call_service("light", "turn_on", data)
count += 1
if count:
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:
warnings.append("розетки не найдены")
else:
turn_off = any(w in cmd for w in ("выключ", "отключ"))
count = 0
for e in entities:
cur = self._get_state(e)
if turn_off and cur == "off":
continue
if not turn_off and cur == "on":
continue
self._call_service(
"switch",
"turn_off" if turn_off else "turn_on",
{"entity_id": e},
)
count += 1
if count:
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")
if not entities:
warnings.append("климат-устройства не найдены")
else:
count = 0
for e in entities:
self._call_service(
"climate",
"set_temperature",
{"entity_id": e, "temperature": temp},
)
count += 1
if count:
actions.append("установлено " + str(temp) + "C (" + str(count) + " шт.)")
# Режим работы (нагрев/охлаждение/авто)
if "нагрев" in cmd:
entities = self._find_entities(tokens, "climate")
for e in entities:
self._call_service(
"climate", "set_hvac_mode",
{"entity_id": e, "hvac_mode": "heat"},
)
actions.append("режим нагрева")
elif "охлажд" in cmd or "холод" in cmd:
entities = self._find_entities(tokens, "climate")
for e in entities:
self._call_service(
"climate", "set_hvac_mode",
{"entity_id": e, "hvac_mode": "cool"},
)
actions.append("режим охлаждения")
# Медиа
if any(k in cmd for k in ["музык", "медиа", "плеер", "колонк"]):
entities = self._find_entities(tokens, "media_player")
if not entities:
warnings.append("медиаплееры не найдены")
else:
count = 0
if any(w in cmd for w in ("пауза", "стоп", "останов")):
for e in entities:
if self._get_state(e) in ("playing", "paused"):
self._call_service("media_player", "media_pause", {"entity_id": e})
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(
"media_player",
"volume_up" if "громче" in cmd else "volume_down",
{"entity_id": e},
)
count += 1
if count:
actions.append("громкость (" + str(count) + " шт.)")
else:
for e in entities:
if self._get_state(e) != "playing":
self._call_service("media_player", "media_play", {"entity_id": e})
count += 1
if count:
actions.append("воспроизведение (" + str(count) + " шт.)")
# Шторы / жалюзи
if any(k in cmd for k in ["штор", "жалюз", "роллет", "гардин"]):
entities = self._find_entities(tokens, "cover")
if not entities:
warnings.append("шторы не найдены")
else:
is_close = any(w in cmd for w in ("закр", "опуст", "спуст"))
count = 0
for e in entities:
if is_close:
self._call_service("cover", "close_cover", {"entity_id": e})
else:
self._call_service("cover", "open_cover", {"entity_id": e})
count += 1
if count:
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:
is_lock = "закры" in cmd or "заблок" in cmd
for e in entities:
self._call_service(
"lock",
"lock" if is_lock else "unlock",
{"entity_id": e},
)
actions.append(
"замки " + ("закрыты" if is_lock else "открыты") + " (" + str(len(entities)) + " шт.)"
)
# Вентиляторы
if "вентилятор" in cmd or "кулер" in cmd:
entities = self._find_entities(tokens, "fan")
if entities:
is_off = "выключ" in cmd
for e in entities:
self._call_service(
"fan",
"turn_off" if is_off else "turn_on",
{"entity_id": e},
)
actions.append(
"вентиляторы " + ("выключены" if is_off else "включены") + " (" + str(len(entities)) + " шт.)"
)
# Пылесосы
if "пылесос" in cmd or "робот" in cmd:
entities = self._find_entities(tokens, "vacuum")
if entities:
if "пауза" in cmd:
for e in entities:
self._call_service("vacuum", "pause", {"entity_id": e})
actions.append("пылесос пауза")
elif "стоп" in cmd or "верн" in cmd:
for e in entities:
self._call_service("vacuum", "stop", {"entity_id": e})
actions.append("пылесос стоп")
elif "уборк" in cmd or "запуст" in cmd:
for e in entities:
self._call_service("vacuum", "start", {"entity_id": e})
actions.append("пылесос уборка")
# Сцены / скрипты
if any(k in cmd for k in ["сцен", "режим", "запуст"]):
all_scripts = self._find_entities([], "script")
matched: List[str] = []
for s in all_scripts:
sl = s.lower()
for t in tokens:
if len(t) >= 3 and t in sl:
matched.append(s)
break
if matched:
for s in matched:
self._call_service("script", "turn_on", {"entity_id": s})
actions.append("SCRIPT " + s)
else:
warnings.append("подходящая сцена не найдена среди " + str(len(all_scripts)))
return actions
request_timeout: int = Field(default=10, description="Таймаут HTTP-запроса, сек")