From 969796ccc43af01e2e9c43eadc3ea4f242e2a1d0 Mon Sep 17 00:00:00 2001 From: host Date: Wed, 29 Jul 2026 19:01:31 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9E=D0=B1=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=BE=20=D0=B4=D0=BB=D1=8F=20HTTPS:=20=D0=B4=D0=B5=D1=84?= =?UTF-8?q?=D0=BE=D0=BB=D1=82=D0=BD=D1=8B=D0=B9=20URL=20https://ha.nasweb.?= =?UTF-8?q?ru,=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=20verify?= =?UTF-8?q?=5Fssl?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- home_assistant_tools.py | 825 +--------------------------------------- 1 file changed, 6 insertions(+), 819 deletions(-) diff --git a/home_assistant_tools.py b/home_assistant_tools.py index 13bbe90..628394d 100644 --- a/home_assistant_tools.py +++ b/home_assistant_tools.py @@ -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 \ No newline at end of file + request_timeout: int = Field(default=10, description="Таймаут HTTP-запроса, сек") \ No newline at end of file