#!/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.2 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 = { "включи", "выключи", "поставь", "установи", "сделай", "запусти", "на", "в", "во", "мне", "пожалуйста", "давай", "ну", "свет", "лампочк", "климат", "температуру", "шторы", "музыку", "режим", "сцену", "сценарий", "уют", "уютный", "теплый", "градус", "градуса", "градусов", "открой", "закрой", "опусти", "подними", "пауза", "играй", "останови", "включил", "выключил", "поставил", "комнате", "квартире", "доме", "гостиной", "спальне", "кухне", "детской", "ванной", "прихожей", "коридоре", "кабинете", "и", "а", "но", "или", "же", "бы", "ли", } _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": ["кто", "дома", "присутств"], }, } # ======================================================================== # 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 _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: 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: """Принимает команду на естественном языке и управляет домом / отвечает на вопросы.""" 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: 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