From 0f7f0604087b6f9fd88651ee23b263a8de7ddf6a Mon Sep 17 00:00:00 2001 From: host Date: Wed, 29 Jul 2026 19:59:41 +0300 Subject: [PATCH] =?UTF-8?q?v2.5.0=20=E2=80=94=20rename=5Fentity,=20toggle?= =?UTF-8?q?=5Fautomation,=20trigger=5Fautomation=20+=20NL=20=D0=BC=D0=B0?= =?UTF-8?q?=D1=80=D1=88=D1=80=D1=83=D1=82=D0=B8=D0=B7=D0=B0=D1=86=D0=B8?= =?UTF-8?q?=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- home_assistant_tools.py | 288 ++++++++++++++++++++++++++++------------ 1 file changed, 201 insertions(+), 87 deletions(-) diff --git a/home_assistant_tools.py b/home_assistant_tools.py index a2893d7..2fecf0a 100644 --- a/home_assistant_tools.py +++ b/home_assistant_tools.py @@ -2,7 +2,7 @@ title: Home Assistant Smart Control description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка. author: ChatGPT (refactored by MiniMax-M3) -version: 2.4.0 +version: 2.5.0 license: MIT """ @@ -66,7 +66,6 @@ class Tools: "устройство", "все", "весь", "вся", } - # Домены и их алиасы (русский -> домен HA) _DOMAIN_ALIASES = { "binary_sensor": ["binary_sensor", "бинарн", "датчик", "датчики", "сенсор", "сенсоры"], "sensor": ["sensor", "показател", "значение", "значения"], @@ -261,6 +260,22 @@ class Tools: 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 + # ======================================================================== # Парсинг чисел # ======================================================================== @@ -289,26 +304,21 @@ class Tools: 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"] @@ -330,7 +340,6 @@ class Tools: lines.append("Закрыто: " + str(len(closed_items)) + " шт.") if unknown: lines.append("Неизвестно: " + str(len(unknown)) + " шт.") - if tokens: filtered = [] for line in lines: @@ -338,17 +347,13 @@ class Tools: 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: @@ -357,7 +362,6 @@ class Tools: on_items.append(eid) else: off_items.append(eid) - lines: List[str] = [] if on_items: lines.append("Горит: " + str(len(on_items)) + " шт.") @@ -371,55 +375,37 @@ class Tools: return lines def _monitor_climate(self, tokens: List[str]) -> List[str]: - """Какая температура и состояние климата. - Берём ТОЛЬКО climate.* или sensor с device_class=temperature/humidity. - """ states = self._get_states() items: List[Dict[str, Any]] = [] - for s in states: eid = s["entity_id"] dc = self._attr(s, "device_class", "") - - # Только climate или sensor с релевантным 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]: - """Какие батарейки садятся. - Учитываем: - - sensor.* с device_class=battery и state как число - - binary_sensor.* с device_class=battery (on = садится) - """ 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 @@ -430,18 +416,13 @@ class Tools: 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)) @@ -450,10 +431,8 @@ class Tools: 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."): @@ -463,27 +442,22 @@ class Tools: 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]: - """Качество воздуха: TVOC, CO2, PM2.5/10, влажность.""" 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."): @@ -492,32 +466,25 @@ class Tools: 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(): @@ -538,11 +505,9 @@ class Tools: 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.")): @@ -553,7 +518,6 @@ class Tools: home.append(name) else: away.append(name) - lines = ["**Люди дома:**"] if home: for n in home: @@ -565,7 +529,6 @@ class Tools: return lines def _monitor_full_report(self) -> List[str]: - """Общий отчёт по дому одной фразой.""" lines = ["**Отчёт по дому:**", ""] lines.extend(self._monitor_presence([])) lines.append("") @@ -599,9 +562,7 @@ class Tools: warnings: List[str] = [] try: - # ============================================================ # Маршрутизация 'покажи все / список устройств' -> list_devices - # ============================================================ if any(w in cmd for w in [ "покажи", "список устройств", "какие устройств", "что есть", "инвентар", "все устройств", @@ -609,15 +570,44 @@ class Tools: 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 @@ -632,33 +622,26 @@ class Tools: 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 @@ -667,7 +650,6 @@ class Tools: actions.append("**Общий статус дома:**") actions.extend(self._monitor_lights([])) actions.extend(self._monitor_climate([])) - else: actions.extend(self._handle_control(cmd, tokens, warnings)) @@ -686,6 +668,8 @@ class Tools: " - кто дома?\n" " - батарейки садятся?\n" " - покажи все binary_sensor\n" + " - переименуй automation.сертификат в Обновление SSL\n" + " - выключи автоматизацию Гирлянда крыльцо\n" " - отчёт по дому" ) if warnings: @@ -708,7 +692,6 @@ class Tools: def _handle_control( self, cmd: str, tokens: List[str], warnings: List[str] ) -> List[str]: - """Обработка команд действия (вкл/выкл/установить/открыть/...).""" actions: List[str] = [] # Свет @@ -759,7 +742,7 @@ class Tools: 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: @@ -832,21 +815,18 @@ class Tools: 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( @@ -857,7 +837,6 @@ class Tools: count += 1 if count: actions.append("громкость (" + str(count) + " шт.)") - else: for e in entities: if self._get_state(e) != "playing": @@ -866,7 +845,7 @@ class Tools: if count: actions.append("воспроизведение (" + str(count) + " шт.)") - # Шторы / жалюзи + # Шторы if any(k in cmd for k in ["штор", "жалюз", "роллет", "гардин"]): entities = self._find_entities(tokens, "cover") if not entities: @@ -931,7 +910,7 @@ class Tools: self._call_service("vacuum", "start", {"entity_id": e}) actions.append("пылесос уборка") - # Кнопки (button.press) + # Кнопки if "кнопк" in cmd and any(w in cmd for w in ("нажм", "кликн", "тапн", "триггерн", "активир")): entities = self._find_entities(tokens, "button") if entities: @@ -1006,7 +985,6 @@ class Tools: 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"] @@ -1014,7 +992,6 @@ class Tools: 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 \ @@ -1025,15 +1002,10 @@ class Tools: if not eid.startswith(domain + "."): continue - # Фильтр по device_class if device_class and dc != device_class: continue - - # Фильтр по состоянию if only_with_state and cur_state != only_with_state: continue - - # Скрыть unavailable/unknown if not show_unavailable and cur_state in ("unavailable", "unknown", "none"): continue @@ -1047,7 +1019,6 @@ class Tools: + ", state=" + (only_with_state or "любой") + ")." ) - # Группируем по домену by_domain: Dict[str, List[Dict[str, Any]]] = {} for s in filtered: d = s["entity_id"].split(".", 1)[0] @@ -1064,4 +1035,147 @@ class Tools: lines.append("") lines.append("... показано " + str(limit) + " из " + str(len(filtered)) + ". Увеличьте `limit` для полного списка.") - return "\n".join(lines) \ No newline at end of file + 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}" \ No newline at end of file