""" title: Home Assistant Smart Control description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка. author: ChatGPT (refactored by MiniMax-M3) version: 2.5.3 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: class Valves(BaseModel): ha_url: str = Field(default="https://ha.nasweb.ru", description="URL Home Assistant") ha_token: str = Field(default="", description="Long-Lived Access Token из профиля HA") verify_ssl: bool = Field(default=False, description="Проверять SSL-сертификат 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() self._states_cache: Optional[Tuple[float, List[Dict[str, Any]]]] = None _STOP_WORDS = { "включи", "выключи", "поставь", "установи", "сделай", "запусти", "на", "в", "во", "мне", "пожалуйста", "давай", "ну", "свет", "лампочк", "климат", "температуру", "шторы", "музыку", "режим", "сцену", "сценарий", "уют", "уютный", "теплый", "градус", "градуса", "градусов", "открой", "закрой", "опусти", "подними", "пауза", "играй", "останови", "включил", "выключил", "поставил", "комнате", "квартире", "доме", "гостиной", "спальне", "кухне", "детской", "ванной", "прихожей", "коридоре", "кабинете", "и", "а", "но", "или", "же", "бы", "ли", "покажи", "список", "какие", "есть", "устройств", "устройства", "устройство", "все", "весь", "вся", } # Алиасы людей — препятствуют "глобальной" команде света _PEOPLE_NAMES = {"дима", "диан", "елен", "анатоли", "милен", "admin"} # Sub-entities switch, которые НЕ являются реальными розетками/выключателями. # Обычно создаются Zigbee2MQTT или прошивкой устройств. _SWITCH_SUB_ENTITIES = ( "power_outage_memory", "interlock", "do_not_disturb", "child_lock", "permit_join", "curtain_motor_reverse", "reverse_direction", "alarm", "indicator_light", "indicator", "fan_horizontal_swing", "fan_alarm", "physical_control_locked", "alarm_trigger", "led_disabled", "linkquality", ) _DOMAIN_ALIASES = { "binary_sensor": ["binary_sensor", "бинарн", "датчик", "датчики", "сенсор", "сенсоры"], "sensor": ["sensor", "показател", "значение", "значения"], "light": ["light", "свет", "лампочк", "лампа", "люстр", "свитч", "спот", "led"], "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", "скрипт"], "scene": ["scene", "сцен"], "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"], "open_states": ("open", "on", "unlocked", "detected"), "closed_states": ("closed", "off", "locked", "clear"), "valid_device_classes": ("door", "window", "garage_door", "lock"), }, "light_status": { "domains": ["light"], "keywords": ["свет", "лампочк", "люстр"], }, "climate_status": { "domains": ["climate", "sensor"], "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"], }, "switches": { "domains": ["switch"], "keywords": ["розетк", "выключател", "реле"], }, "system": { "domains": ["binary_sensor"], "keywords": ["бэкап", "backup", "stale", "updates", "updates_package", "node", "status", "connection", "gateway", "bridge"], }, } 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: 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]]: 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: 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: 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 _is_sub_entity(self, eid: str) -> bool: """Проверяет, является ли switch/light sub-entity (служебной).""" eid_lower = eid.lower() return any(sub in eid_lower for sub in self._SWITCH_SUB_ENTITIES) def _find_entities( self, keywords: List[str], domain: Optional[str] = None, require_all: bool = False, skip_sub_entities: 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 if skip_sub_entities and self._is_sub_entity(eid): 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]: for domain in ["scene", "script", "automation", "binary_sensor"]: for a in self._DOMAIN_ALIASES.get(domain, []): if a in cmd: return domain for domain, aliases in self._DOMAIN_ALIASES.items(): if domain in ("scene", "script", "automation", "binary_sensor"): continue 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]: 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 "" if state and unit and unit not in str(state): state_str = str(state) + " " + str(unit) else: state_str = str(state) return " * " + fn + room + ": " + state_str # ======================================================================== # Мониторинг # ======================================================================== def _monitor_windows_doors(self, tokens: List[str]) -> List[str]: cfg = self._MONITOR_DOMAINS["window_door"] valid_dc = cfg["valid_device_classes"] states = self._get_states() items: List[Tuple[Dict[str, Any], bool, bool]] = [] for s in states: eid = s["entity_id"] if not eid.startswith("binary_sensor."): continue device_class = self._attr(s, "device_class", "") if device_class not in valid_dc: 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 ["датчики окон/дверей не найдены (проверьте device_class в HA)"] 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("**Закрыто:**") for s in closed_items: lines.append(self._format_entity(s)) if unknown: lines.append("**Неизвестно** (state не в open/closed):") for s in unknown: lines.append(self._format_entity(s)) 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] = [] unavailable_items: List[str] = [] for eid in entities: state = self._get_state(eid) if state == "on": on_items.append(eid) elif state == "unavailable": unavailable_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)) + " шт.") if unavailable_items: lines.append("**Недоступно:** " + str(len(unavailable_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_switches(self, tokens: List[str]) -> List[str]: """Какие розетки/реле включены (без sub-entities).""" cfg = self._MONITOR_DOMAINS["switches"] entities = self._find_entities( tokens if tokens else cfg["keywords"], "switch", skip_sub_entities=True, ) if not entities: return ["розетки не найдены"] on_items: List[str] = [] off_items: List[str] = [] unavailable_items: List[str] = [] for eid in entities: state = self._get_state(eid) if state == "on": on_items.append(eid) elif state == "unavailable": unavailable_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)) + " шт.") if unavailable_items: lines.append("**Недоступно:** " + str(len(unavailable_items)) + " шт.") return lines def _monitor_scenes(self) -> List[str]: """Список сцен с последним временем активации.""" states = self._get_states() scenes = [s for s in states if s["entity_id"].startswith("scene.")] if not scenes: return ["Сцены не найдены"] lines = ["**Сцены:**"] for s in scenes: eid = s["entity_id"] fn = self._attr(s, "friendly_name", eid) state = s.get("state", "") attrs = s.get("attributes", {}) or {} # Для сцен state содержит timestamp последней активации или "unknown" last_activated = attrs.get("last_activated", "") description = "" if state == "unknown": description = " (ни разу не активировалась)" elif last_activated: description = " — последняя: " + str(last_activated)[:19] lines.append(" * " + fn + description) return lines def _monitor_system(self) -> List[str]: states = self._get_states() items: List[str] = [] warnings: List[str] = [] keys_node = ("node_", "node ") keys_update = ("updates", "package") keys_backup = ("backup", "stale") keys_conn = ("connection", "gateway", "bridge") for s in states: eid = s["entity_id"] if not eid.startswith("binary_sensor."): continue name = (self._attr(s, "friendly_name", "") or "").lower() cur = s.get("state", "") text = (eid + " " + name).lower() if any(k in text for k in keys_backup): if cur == "on": warnings.append("Бэкапы устарели: " + self._attr(s, "friendly_name", eid)) continue if any(k in text for k in keys_update): if cur == "on": warnings.append("Доступны обновления: " + self._attr(s, "friendly_name", eid)) continue if any(k in text for k in keys_node): if cur == "off": warnings.append("Нода offline: " + self._attr(s, "friendly_name", eid)) else: items.append(self._format_entity(s)) continue if any(k in text for k in keys_conn): if cur == "off": warnings.append("Соединение потеряно: " + self._attr(s, "friendly_name", eid)) else: items.append(self._format_entity(s)) continue lines = ["**Системный статус:**"] if items: lines.append("OK:") for s in items: lines.append(s) if warnings: lines.append("") lines.append("⚠️ **Требует внимания:**") for w in warnings: lines.append(" - " + w) if not items and not warnings: 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_switches([])) 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([])) lines.append("") lines.extend(self._monitor_scenes()) lines.append("") lines.extend(self._monitor_system()) 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: 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 ["розетк", "выключател", "реле"]): actions.extend(self._monitor_switches(tokens)) matched = True elif any(k in cmd for k in ["свет", "лампочк", "люстр", "спот", "led"]) 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 "батарейк" in cmd or "аккумулятор" 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_climate(tokens)) matched = True elif any(k in cmd for k in ["музык", "медиа", "плеер", "колонк", "телевизор"]): actions.extend(self._monitor_media(tokens)) matched = True elif "сцен" in cmd and ("какие" in cmd or "список" in cmd or "покажи" in cmd): actions.extend(self._monitor_scenes()) matched = True elif "дома" in cmd or "присутств" in cmd or "кто" in cmd: actions.extend(self._monitor_presence(tokens)) matched = True elif any(k in cmd for k in ["систем", "нод", "бэкап", "backup", "обновлен", "мост"]): actions.extend(self._monitor_system()) 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" " - что в розетках включено?\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 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 [ "спальн", "гостин", "кухн", "детск", "ванн", "прихож", "коридор", "кабинет", "зал", "веранд", ] + list(self._PEOPLE_NAMES) ) 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", skip_sub_entities=True, ) 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 ["сцен"]) and "запуст" in cmd: all_scenes = self._find_entities([], "scene") matched: List[str] = [] for s in all_scenes: 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("scene", "turn_on", {"entity_id": s}) actions.append("SCENE " + s) else: warnings.append("подходящая сцена не найдена среди " + str(len(all_scenes))) if "скрипт" in cmd or "запуст" in cmd: all_scripts = self._find_entities([], "script") matched = [] 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, skip_sub_entities: bool = False, ) -> str: 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 not eid.startswith("binary_sensor."): continue if dc not in ("door", "window", "garage_door", "lock"): continue else: if not eid.startswith(domain + "."): continue if skip_sub_entities and self._is_sub_entity(eid): 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: 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: 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 "[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 "Переименовано: " + target + " -> '" + name + "'" except Exception as e: return "Ошибка переименования " + target + ": " + str(e) # ======================================================================== # Управление автоматизациями # ======================================================================== def toggle_automation( self, entity_id: str = "", friendly_name: str = "", turn_off: bool = False, ) -> str: 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 "[DRY-RUN] Автоматизация " + target + " была бы " + action_label try: self._request( "POST", "/api/services/automation/" + service, json={"entity_id": target}, ) self._get_states(force=True) return "Автоматизация " + target + " " + action_label except Exception as e: return "Ошибка: " + str(e) def trigger_automation( self, entity_id: str = "", friendly_name: str = "", skip_condition: bool = True, ) -> str: 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 "[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 "Триггер запущен: " + target except Exception as e: return "Ошибка: " + str(e)