From ae514137d7fe37ba79faf588ae3dd76e27589c06 Mon Sep 17 00:00:00 2001 From: host Date: Wed, 29 Jul 2026 18:51:01 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=20=D0=BE=D1=81=D0=BD=D0=BE=D0=B2=D0=BD=D0=BE=D0=B9=20?= =?UTF-8?q?=D0=B8=D0=BD=D1=81=D1=82=D1=80=D1=83=D0=BC=D0=B5=D0=BD=D1=82=20?= =?UTF-8?q?home=5Fassistant=5Ftools.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- home_assistant_tools.py | 826 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 826 insertions(+) create mode 100644 home_assistant_tools.py diff --git a/home_assistant_tools.py b/home_assistant_tools.py new file mode 100644 index 0000000..13bbe90 --- /dev/null +++ b/home_assistant_tools.py @@ -0,0 +1,826 @@ +#!/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_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( + default=False, + description="Не выполнять команды, только показывать что БУДЕТ сделано", + ) + + 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 \ No newline at end of file