""" title: Home Assistant Smart Control description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка. author: ChatGPT (refactored by MiniMax-M3) version: 2.5.0 license: MIT """ import re import time import logging import requests from typing import Dict, Any, List, Optional, Tuple from pydantic import BaseModel, Field 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 для самоподписанных)", ) 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() self._states_cache: Optional[Tuple[float, List[Dict[str, Any]]]] = None # ======================================================================== # Стоп-слова и ключевые слова # ======================================================================== _STOP_WORDS = { "включи", "выключи", "поставь", "установи", "сделай", "запусти", "на", "в", "во", "мне", "пожалуйста", "давай", "ну", "свет", "лампочк", "климат", "температуру", "шторы", "музыку", "режим", "сцену", "сценарий", "уют", "уютный", "теплый", "градус", "градуса", "градусов", "открой", "закрой", "опусти", "подними", "пауза", "играй", "останови", "включил", "выключил", "поставил", "комнате", "квартире", "доме", "гостиной", "спальне", "кухне", "детской", "ванной", "прихожей", "коридоре", "кабинете", "и", "а", "но", "или", "же", "бы", "ли", "покажи", "список", "какие", "есть", "устройств", "устройства", "устройство", "все", "весь", "вся", } _DOMAIN_ALIASES = { "binary_sensor": ["binary_sensor", "бинарн", "датчик", "датчики", "сенсор", "сенсоры"], "sensor": ["sensor", "показател", "значение", "значения"], "light": ["light", "свет", "лампочк", "лампа", "люстр", "свитч"], "switch": ["switch", "розетк", "выключател", "реле"], "climate": ["climate", "климат", "кондиционер", "термостат"], "cover": ["cover", "штор", "жалюз", "роллет", "гардин"], "media_player": ["media_player", "медиа", "плеер", "колонк", "телевизор"], "lock": ["lock", "замок"], "fan": ["fan", "вентилятор", "кулер"], "vacuum": ["vacuum", "пылесос", "робот"], "person": ["person", "люди", "пользовател"], "device_tracker": ["device_tracker", "трекер", "геолокац"], "button": ["button", "кнопк"], "automation": ["automation", "автоматизац"], "script": ["script", "скрипт", "сцен", "сценарий"], "input_boolean": ["input_boolean", "переключател", "тумблер"], } _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"), }, "light_status": { "domains": ["light"], "keywords": ["свет", "лампочк", "люстр"], }, "climate_status": { "domains": ["climate"], "keywords": ["температур", "климат", "тепл", "холод", "кондиционер", "батаре"], }, "battery": { "domains": ["sensor", "binary_sensor"], "keywords": ["батарейк", "аккумулятор", "заряд"], }, "media_status": { "domains": ["media_player"], "keywords": ["музык", "медиа", "плеер", "колонк", "телевизор", "тв"], }, "presence": { "domains": ["person", "device_tracker"], "keywords": ["кто", "дома", "присутств"], }, "motion": { "domains": ["binary_sensor"], "keywords": ["движен", "двигать", "присутств"], "device_classes": ["motion", "occupancy"], }, "air_quality": { "domains": ["sensor"], "keywords": ["воздух", "tvoc", "co2", "pm2", "pm10", "влажн", "увлажн"], "device_classes": ["volatile_organic_compounds", "carbon_dioxide", "pm25", "pm10", "humidity", "aqi"], }, } # ======================================================================== # 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 kwargs.setdefault("verify", self.valves.verify_ssl) 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 _resolve_domain(self, cmd: str) -> Optional[str]: """Определить домен HA по русской команде.""" for domain, aliases in self._DOMAIN_ALIASES.items(): for a in aliases: if a in cmd: return domain return None def _find_entity_by_name(self, name: str, domain: Optional[str] = None) -> Optional[str]: """Найти первый entity_id, у которого friendly_name или id совпадает с name.""" states = self._get_states(force=True) name_lower = name.lower().strip() for s in states: eid = s["entity_id"] if domain and not eid.startswith(domain + "."): continue fn = (self._attr(s, "friendly_name", "") or "").lower() if fn == name_lower: return eid # Частичное совпадение if name_lower in fn or name_lower in eid.lower(): return eid return None # ======================================================================== # Парсинг чисел # ======================================================================== 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]: states = self._get_states() items: List[Dict[str, Any]] = [] for s in states: eid = s["entity_id"] dc = self._attr(s, "device_class", "") if eid.startswith("climate."): items.append(s) elif eid.startswith("sensor.") and dc in ( "temperature", "humidity", "current_temperature", ): items.append(s) if tokens: items = [ s for s in items if any(t in self._entity_text(s) for t in tokens) ] if not items: return ["климат-устройства не найдены"] lines = ["**Климат:**"] for s in items[:20]: lines.append(self._format_entity(s)) return lines def _monitor_battery(self, tokens: List[str]) -> List[str]: states = self._get_states() low_battery: List[Tuple[Dict[str, Any], float]] = [] binary_low: List[Dict[str, Any]] = [] for s in states: eid = s["entity_id"] dc = self._attr(s, "device_class", "") cur = s.get("state", "") if eid.startswith("sensor.") and dc == "battery": if cur in ("unavailable", "unknown", "none", ""): continue try: level = float(cur) except (ValueError, TypeError): continue if level < 30: low_battery.append((s, level)) continue if eid.startswith("binary_sensor.") and dc == "battery": if cur == "on": binary_low.append(s) continue if not low_battery and not binary_low: return ["Все батарейки в норме (>=30%)"] lines = ["**Низкий заряд:**"] low_battery.sort(key=lambda x: x[1]) for s, level in low_battery[:15]: lines.append(self._format_entity(s)) for s in binary_low[:10]: lines.append(" ! " + self._attr(s, "friendly_name", s["entity_id"]) + ": разряжена") return lines def _monitor_motion(self, tokens: List[str]) -> List[str]: states = self._get_states() active: List[Dict[str, Any]] = [] for s in states: eid = s["entity_id"] if not eid.startswith("binary_sensor."): continue dc = self._attr(s, "device_class", "") if dc not in ("motion", "occupancy"): continue if s.get("state") == "on": active.append(s) if not active: return ["Нигде нет движения (все датчики молчат)"] if tokens: active = [s for s in active if any(t in self._entity_text(s) for t in tokens)] lines = ["**Движение обнаружено:**"] for s in active[:15]: lines.append(self._format_entity(s)) return lines def _monitor_air_quality(self, tokens: List[str]) -> List[str]: states = self._get_states() items: List[Dict[str, Any]] = [] target_dc = { "volatile_organic_compounds", "carbon_dioxide", "pm25", "pm10", "humidity", "aqi", } for s in states: eid = s["entity_id"] if not eid.startswith("sensor."): continue dc = self._attr(s, "device_class", "") name = (self._attr(s, "friendly_name", "") or "").lower() if dc in target_dc or any(k in name for k in ["tvoc", "co2", "влажн", "воздух", "pm2", "pm10"]): items.append(s) if not items: return ["датчики качества воздуха не найдены"] if tokens: items = [s for s in items if any(t in self._entity_text(s) for t in tokens)] lines = ["**Качество воздуха:**"] for s in items[:15]: lines.append(self._format_entity(s)) return lines def _monitor_media(self, tokens: List[str]) -> List[str]: cfg = self._MONITOR_DOMAINS["media_status"] entities = self._find_entities(tokens if tokens else cfg["keywords"], "media_player") if not entities: return ["медиаплееры не найдены"] playing: List[str] = [] idle: List[str] = [] for eid in entities: state = self._get_state(eid) (playing if state == "playing" else idle).append(eid) lines = ["**Медиа:**"] for eid in playing: for s in self._get_states(): 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 _monitor_full_report(self) -> List[str]: lines = ["**Отчёт по дому:**", ""] lines.extend(self._monitor_presence([])) lines.append("") lines.extend(self._monitor_windows_doors([])) lines.append("") lines.extend(self._monitor_lights([])) lines.append("") lines.extend(self._monitor_climate([])) lines.append("") lines.extend(self._monitor_air_quality([])) lines.append("") lines.extend(self._monitor_motion([])) lines.append("") lines.extend(self._monitor_battery([])) return lines # ======================================================================== # Главная точка входа # ======================================================================== def smart(self, command: str) -> str: """Принимает команду на естественном языке и управляет домом / отвечает на вопросы.""" 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: # Маршрутизация 'покажи все / список устройств' -> list_devices if any(w in cmd for w in [ "покажи", "список устройств", "какие устройств", "что есть", "инвентар", "все устройств", ]): 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: # Ожидаемый формат: "переименуй в <новое имя>" # или: "переименуй как <новое имя>" m = re.search(r"переимен\w*\s+(.+?)\s+(?:в|как|на)\s+(.+)$", cmd) if m: target = m.group(1).strip().strip('"').strip("'") new_name = m.group(2).strip().strip('"').strip("'") return self.rename_entity( entity_id=target if "." in target else "", name=new_name, friendly_name=target if "." not in target else "", ) return ( "Формат: `переименуй в <новое имя>`\n" "Пример: `переименуй automation.сертификат в Обновление SSL`" ) # Включение/выключение автоматизации if "автоматизац" in cmd and any(w in cmd for w in ("включ", "выключ", "активир", "деактивир")): target = "" for s in self._get_states(force=True): eid = s["entity_id"] if not eid.startswith("automation."): continue fn = (self._attr(s, "friendly_name", "") or "").lower() if fn and fn in cmd: target = eid break if target: turn_off = any(w in cmd for w in ("выключ", "деактивир")) return self.toggle_automation(entity_id=target, turn_off=turn_off) warnings.append("автоматизация не найдена в команде") is_question = ( cmd.startswith(("как", "сколько", "что", "кто", "где", "какая", "какой", "какое", "какие")) or "?" in cmd 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 ["воздух", "tvoc", "co2", "влажн", "качеств", "pm2", "pm10"]): actions.extend(self._monitor_air_quality(tokens)) matched = True elif any(k in cmd for k in ["движен", "двигает", "хот", "кто-то ход"]): actions.extend(self._monitor_motion(tokens)) matched = True elif any(k in cmd for k in ["температур", "климат", "тепл", "холод", "кондиционер", "батаре"]): actions.extend(self._monitor_climate(tokens)) matched = True elif "батарейк" in cmd or "заряд" in cmd: actions.extend(self._monitor_battery(tokens)) matched = True elif any(k in cmd for k in ["музык", "медиа", "плеер", "колонк", "телевизор"]): actions.extend(self._monitor_media(tokens)) matched = True elif "дома" in cmd or "присутств" in cmd: actions.extend(self._monitor_presence(tokens)) matched = True 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" " - какая температура в спальне?\n" " - как воздух в квартире?\n" " - где есть движение?\n" " - кто дома?\n" " - батарейки садятся?\n" " - покажи все binary_sensor\n" " - переименуй automation.сертификат в Обновление SSL\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: 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 "кнопк" in cmd and any(w in cmd for w in ("нажм", "кликн", "тапн", "триггерн", "активир")): entities = self._find_entities(tokens, "button") if entities: count = 0 for e in entities: self._call_service("button", "press", {"entity_id": e}) count += 1 actions.append("кнопки нажаты (" + str(count) + " шт.)") else: warnings.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 # ======================================================================== # Диагностика / справочник устройств (отдельный инструмент) # ======================================================================== def list_devices( self, domain: str = "", device_class: str = "", only_with_state: str = "", 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: return "Не удалось получить состояния: " + str(e) domain = domain.lower().strip() device_class = device_class.lower().strip() only_with_state = only_with_state.lower().strip() filtered: List[Dict[str, Any]] = [] for s in states: eid = s["entity_id"] attrs = s.get("attributes", {}) or {} cur_state = (s.get("state", "") or "").lower() dc = (attrs.get("device_class", "") or "").lower() if domain: if domain == "window_door": if dc not in ("door", "window", "garage_door", "lock") and \ not any(k in (eid + " " + (attrs.get("friendly_name", "") or "").lower()) for k in ["окн", "двер", "форточ", "балкон", "калитк", "ворота", "гараж"]): continue else: if not eid.startswith(domain + "."): continue if device_class and dc != device_class: continue if only_with_state and cur_state != only_with_state: continue if not show_unavailable and cur_state in ("unavailable", "unknown", "none"): continue filtered.append(s) if not filtered: return ( "Устройств не найдено по фильтру " "(domain=" + (domain or "любой") + ", class=" + (device_class or "любой") + ", state=" + (only_with_state or "любой") + ")." ) by_domain: Dict[str, List[Dict[str, Any]]] = {} for s in filtered: d = s["entity_id"].split(".", 1)[0] by_domain.setdefault(d, []).append(s) lines = ["**Найдено " + str(len(filtered)) + " устройств** (показано до " + str(limit) + "):"] for d in sorted(by_domain.keys()): lines.append("") lines.append("### " + d + " (" + str(len(by_domain[d])) + " шт.)") for s in by_domain[d][:limit]: lines.append(self._format_entity(s)) if len(filtered) > limit: lines.append("") lines.append("... показано " + str(limit) + " из " + str(len(filtered)) + ". Увеличьте `limit` для полного списка.") return "\n".join(lines) # ======================================================================== # 🏷 Переименование сущностей # ======================================================================== def rename_entity( self, entity_id: str = "", name: str = "", friendly_name: str = "", ) -> str: """ Переименовать сущность в Home Assistant (меняет friendly_name). Работает для automation, script, light, switch, scene и др. Можно указать либо entity_id (точное имя), либо friendly_name (для поиска). Args: entity_id: точный id сущности (например "automation.сертификат"). Если пусто — ищем по friendly_name. name: новое имя. friendly_name: текущее friendly_name для поиска (если entity_id не указан). Returns: Подтверждение или сообщение об ошибке. """ if not self.valves.ha_token: return "Не задан HA Token (Valves)." if not name: return "Укажите новое имя (параметр name)." target = entity_id if not target or "." not in target: # Поиск по friendly_name search_name = friendly_name or entity_id target = self._find_entity_by_name(search_name) if not target: return "Не нашёл сущность с именем: " + search_name if self.valves.dry_run: return f"[DRY-RUN] Переименовал бы {target} -> {name}" try: self._request( "POST", "/api/services/homeassistant/update_entity", json={"entity_id": target, "name": name}, ) self._get_states(force=True) # сброс кэша return f"Переименовано: {target} -> '{name}'" except Exception as e: return f"Ошибка переименования {target}: {e}" # ======================================================================== # ⚙️ Управление автоматизациями # ======================================================================== def toggle_automation( self, entity_id: str = "", friendly_name: str = "", turn_off: bool = False, ) -> str: """ Включить или выключить автоматизацию. Args: entity_id: точный id (например "automation.сертификат"). Если пусто — ищем по friendly_name. friendly_name: имя для поиска. turn_off: True — выключить, False — включить. Returns: Подтверждение или сообщение об ошибке. """ if not self.valves.ha_token: return "Не задан HA Token (Valves)." target = entity_id if not target or "." not in target: search_name = friendly_name or entity_id target = self._find_entity_by_name(search_name, domain="automation") if not target: return "Не нашёл автоматизацию: " + search_name service = "turn_off" if turn_off else "turn_on" action_label = "выключена" if turn_off else "включена" if self.valves.dry_run: return f"[DRY-RUN] Автоматизация {target} была бы {action_label}" try: self._request( "POST", "/api/services/automation/" + service, json={"entity_id": target}, ) self._get_states(force=True) return f"Автоматизация {target} {action_label}" except Exception as e: return f"Ошибка: {e}" def trigger_automation( self, entity_id: str = "", friendly_name: str = "", skip_condition: bool = True, ) -> str: """ Запустить автоматизацию вручную (триггернуть). Args: entity_id: точный id (например "automation.сертификат"). Если пусто — ищем по friendly_name. friendly_name: имя для поиска. skip_condition: пропустить ли условия (по умолчанию True). Returns: Подтверждение или сообщение об ошибке. """ if not self.valves.ha_token: return "Не задан HA Token (Valves)." target = entity_id if not target or "." not in target: search_name = friendly_name or entity_id target = self._find_entity_by_name(search_name, domain="automation") if not target: return "Не нашёл автоматизацию: " + search_name if self.valves.dry_run: return f"[DRY-RUN] Триггернул бы {target}" try: data: Dict[str, Any] = {"entity_id": target} if skip_condition: data["skip_condition"] = True self._request( "POST", "/api/services/automation/trigger", json=data, ) return f"Триггер запущен: {target}" except Exception as e: return f"Ошибка: {e}"