826 lines
36 KiB
Python
826 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
title: Home Assistant Smart Control
|
||
description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка.
|
||
author: ChatGPT (refactored by MiniMax-M3)
|
||
version: 2.2.0
|
||
license: MIT
|
||
"""
|
||
|
||
import re
|
||
import time
|
||
import logging
|
||
import requests
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
from pydantic import BaseModel, Field
|
||
|
||
# Логгер — выводит в консоль Open WebUI
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Tools:
|
||
# ========================================================================
|
||
# Конфигурация (Valves - задаётся в UI Open WebUI)
|
||
# ========================================================================
|
||
class Valves(BaseModel):
|
||
"""Конфигурация инструмента."""
|
||
|
||
ha_url: str = Field(
|
||
default="http://192.168.31.62:8123",
|
||
description="URL Home Assistant (например http://192.168.31.62:8123)",
|
||
)
|
||
ha_token: str = Field(
|
||
default="",
|
||
description="Long-Lived Access Token из профиля HA",
|
||
)
|
||
request_timeout: int = Field(default=10, description="Таймаут HTTP-запроса, сек")
|
||
max_retries: int = Field(default=2, description="Количество ретраев при сетевых сбоях")
|
||
cache_ttl: int = Field(default=5, description="TTL кэша состояний HA, сек")
|
||
confirm_global_actions: bool = Field(
|
||
default=True,
|
||
description="Спрашивать подтверждение для команд, действующих на ВСЕ сущности домена",
|
||
)
|
||
dry_run: bool = Field(
|
||
default=False,
|
||
description="Не выполнять команды, только показывать что БУДЕТ сделано",
|
||
)
|
||
|
||
def __init__(self):
|
||
self.valves = self.Valves()
|
||
# Кэш состояний с TTL
|
||
self._states_cache: Optional[Tuple[float, List[Dict[str, Any]]]] = None
|
||
|
||
# ========================================================================
|
||
# Стоп-слова и ключевые слова
|
||
# ========================================================================
|
||
_STOP_WORDS = {
|
||
"включи", "выключи", "поставь", "установи", "сделай", "запусти",
|
||
"на", "в", "во", "мне", "пожалуйста", "давай", "ну",
|
||
"свет", "лампочк", "климат", "температуру", "шторы", "музыку",
|
||
"режим", "сцену", "сценарий", "уют", "уютный", "тплый", "теплый",
|
||
"градус", "градуса", "градусов", "открой", "закрой", "опусти", "подними",
|
||
"пауза", "играй", "останови", "включил", "выключил", "поставил",
|
||
"комнате", "квартире", "доме", "гостиной", "спальне", "кухне",
|
||
"детской", "ванной", "прихожей", "коридоре", "кабинете",
|
||
"и", "а", "но", "или", "же", "бы", "ли",
|
||
}
|
||
|
||
# Русские числительные для парсинга
|
||
_RU_NUMS = {
|
||
"ноль": 0, "один": 1, "два": 2, "три": 3, "четыре": 4, "пять": 5,
|
||
"шесть": 6, "семь": 7, "восемь": 8, "девять": 9, "десять": 10,
|
||
"одиннадцать": 11, "двенадцать": 12, "тринадцать": 13, "четырнадцать": 14,
|
||
"пятнадцать": 15, "шестнадцать": 16, "семнадцать": 17, "восемнадцать": 18,
|
||
"девятнадцать": 19, "двадцать": 20, "двадцать один": 21, "двадцать два": 22,
|
||
"двадцать три": 23, "двадцать четыре": 24, "двадцать пять": 25,
|
||
"двадцать шесть": 26, "двадцать семь": 27, "двадцать восемь": 28,
|
||
"двадцать девять": 29, "тридцать": 30,
|
||
}
|
||
|
||
# Категории устройств для мониторинга
|
||
_MONITOR_DOMAINS = {
|
||
"window_door": {
|
||
"domains": ["binary_sensor"],
|
||
"keywords": ["окн", "двер", "форточ", "калитк", "ворота", "гараж"],
|
||
"open_states": ("open", "on", "unlocked", "detected"),
|
||
"closed_states": ("closed", "off", "locked", "clear"),
|
||
"title": "Окна и двери",
|
||
},
|
||
"light_status": {
|
||
"domains": ["light"],
|
||
"keywords": ["свет", "лампочк", "люстр"],
|
||
"active_states": ("on",),
|
||
"title": "Свет",
|
||
},
|
||
"climate_status": {
|
||
"domains": ["climate"],
|
||
"keywords": ["температур", "климат", "тепл", "холод", "кондиционер", "батаре"],
|
||
"title": "Климат",
|
||
},
|
||
"battery": {
|
||
"domains": ["sensor", "binary_sensor"],
|
||
"keywords": ["батарейк", "аккумулятор", "заряд"],
|
||
"title": "Батарейки",
|
||
},
|
||
"media_status": {
|
||
"domains": ["media_player"],
|
||
"keywords": ["музык", "медиа", "плеер", "колонк", "телевизор", "тв"],
|
||
"title": "Медиа",
|
||
},
|
||
"presence": {
|
||
"domains": ["person", "device_tracker"],
|
||
"keywords": ["кто", "дома", "присутств"],
|
||
"title": "Люди",
|
||
},
|
||
}
|
||
|
||
# ========================================================================
|
||
# HTTP / API
|
||
# ========================================================================
|
||
def _headers(self) -> Dict[str, str]:
|
||
return {
|
||
"Authorization": "Bearer " + self.valves.ha_token,
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
def _request(self, method: str, path: str, **kwargs) -> requests.Response:
|
||
"""HTTP-запрос с ретраями и логированием."""
|
||
url = self.valves.ha_url + path
|
||
last_exc: Optional[Exception] = None
|
||
for attempt in range(1, self.valves.max_retries + 2):
|
||
try:
|
||
resp = requests.request(
|
||
method, url,
|
||
headers=self._headers(),
|
||
timeout=self.valves.request_timeout,
|
||
**kwargs,
|
||
)
|
||
resp.raise_for_status()
|
||
return resp
|
||
except requests.RequestException as e:
|
||
last_exc = e
|
||
logger.warning(
|
||
"HA %s %s failed (attempt %d): %s",
|
||
method, path, attempt, e,
|
||
)
|
||
if attempt <= self.valves.max_retries:
|
||
time.sleep(0.5 * attempt)
|
||
raise RuntimeError("Home Assistant недоступен: " + str(last_exc))
|
||
|
||
def _get_states(self, force: bool = False) -> List[Dict[str, Any]]:
|
||
"""Получить состояния HA с TTL-кэшем."""
|
||
now = time.time()
|
||
if (
|
||
not force
|
||
and self._states_cache is not None
|
||
and now - self._states_cache[0] < self.valves.cache_ttl
|
||
):
|
||
return self._states_cache[1]
|
||
|
||
resp = self._request("GET", "/api/states")
|
||
states = resp.json()
|
||
self._states_cache = (now, states)
|
||
return states
|
||
|
||
def _call_service(self, domain: str, service: str, data: Dict[str, Any]) -> bool:
|
||
"""Вызов сервиса HA. Возвращает True если выполнено, False если dry-run."""
|
||
logger.info(
|
||
"%s -> %s.%s %s",
|
||
"[DRY] " if self.valves.dry_run else "",
|
||
domain, service, data,
|
||
)
|
||
if self.valves.dry_run:
|
||
return False
|
||
self._request("POST", "/api/services/" + domain + "/" + service, json=data)
|
||
return True
|
||
|
||
# ========================================================================
|
||
# Поиск сущностей
|
||
# ========================================================================
|
||
def _tokenize(self, text: str) -> List[str]:
|
||
"""Токенизация: фильтр стоп-слов, приведение к нижнему регистру."""
|
||
text = text.lower().strip().replace("ё", "е")
|
||
tokens = re.findall(r"[а-яa-z0-9]+", text)
|
||
return [t for t in tokens if t not in self._STOP_WORDS and len(t) >= 2]
|
||
|
||
def _entity_text(self, s: Dict[str, Any]) -> str:
|
||
"""Полный текст сущности: id + friendly_name + area."""
|
||
attrs = s.get("attributes", {}) or {}
|
||
return " ".join([
|
||
s.get("entity_id", ""),
|
||
attrs.get("friendly_name", "") or "",
|
||
attrs.get("area", "") or "",
|
||
attrs.get("area_id", "") or "",
|
||
]).lower()
|
||
|
||
def _find_entities(
|
||
self,
|
||
keywords: List[str],
|
||
domain: Optional[str] = None,
|
||
require_all: bool = False,
|
||
) -> List[str]:
|
||
"""Поиск сущностей по ключевым словам."""
|
||
states = self._get_states()
|
||
result: List[str] = []
|
||
|
||
for s in states:
|
||
eid = s["entity_id"]
|
||
if domain and not eid.startswith(domain + "."):
|
||
continue
|
||
haystack = self._entity_text(s)
|
||
if not keywords:
|
||
if domain:
|
||
result.append(eid)
|
||
continue
|
||
matches = sum(1 for kw in keywords if kw in haystack)
|
||
if (require_all and matches == len(keywords)) or (not require_all and matches > 0):
|
||
result.append(eid)
|
||
|
||
return result
|
||
|
||
def _get_state(self, entity_id: str) -> Optional[str]:
|
||
"""Получить текущее состояние конкретной сущности."""
|
||
for s in self._get_states():
|
||
if s["entity_id"] == entity_id:
|
||
return s.get("state")
|
||
return None
|
||
|
||
def _attr(self, s: Dict[str, Any], key: str, default=None) -> Any:
|
||
"""Безопасное извлечение атрибута."""
|
||
return (s.get("attributes", {}) or {}).get(key, default)
|
||
|
||
# ========================================================================
|
||
# Парсинг чисел
|
||
# ========================================================================
|
||
def _extract_number(self, text: str, default: Optional[float] = None) -> Optional[float]:
|
||
"""Извлечь число из текста: арабские цифры или русские слова."""
|
||
# Арабские цифры
|
||
m = re.search(r"-?\d+(?:[.,]\d+)?", text)
|
||
if m:
|
||
return float(m.group(0).replace(",", "."))
|
||
# Русские слова
|
||
text_norm = text.lower().replace("ё", "е")
|
||
# Сначала длинные сочетания
|
||
for word in sorted(self._RU_NUMS.keys(), key=len, reverse=True):
|
||
if word in text_norm:
|
||
return float(self._RU_NUMS[word])
|
||
return default
|
||
|
||
# ========================================================================
|
||
# Форматирование ответа
|
||
# ========================================================================
|
||
def _format_entity(self, s: Dict[str, Any]) -> str:
|
||
"""Красиво отформатировать одну сущность для вывода."""
|
||
attrs = s.get("attributes", {}) or {}
|
||
fn = attrs.get("friendly_name", s["entity_id"])
|
||
state = s.get("state", "?")
|
||
unit = attrs.get("unit_of_measurement", "")
|
||
area = attrs.get("area", "") or attrs.get("area_id", "")
|
||
room = " (" + area + ")" if area else ""
|
||
return " * " + fn + room + ": " + str(state) + str(unit)
|
||
|
||
# ========================================================================
|
||
# Мониторинг (ответы на вопросы о состоянии дома)
|
||
# ========================================================================
|
||
def _monitor_windows_doors(self, tokens: List[str]) -> List[str]:
|
||
"""Какие окна/двери открыты."""
|
||
cfg = self._MONITOR_DOMAINS["window_door"]
|
||
states = self._get_states()
|
||
|
||
items: List[Tuple[Dict[str, Any], bool, bool]] = []
|
||
for s in states:
|
||
eid = s["entity_id"]
|
||
is_relevant_domain = eid.startswith("binary_sensor.")
|
||
device_class = self._attr(s, "device_class", "")
|
||
is_door_window = device_class in ("door", "window", "garage_door", "lock")
|
||
|
||
text = self._entity_text(s)
|
||
has_keyword = any(k in text for k in cfg["keywords"])
|
||
|
||
if not ((is_relevant_domain and is_door_window) or has_keyword):
|
||
continue
|
||
|
||
cur = s.get("state", "")
|
||
is_open = cur in cfg["open_states"]
|
||
is_closed = cur in cfg["closed_states"]
|
||
items.append((s, is_open, is_closed))
|
||
|
||
if not items:
|
||
return ["датчики окон/дверей не найдены"]
|
||
|
||
open_items = [s for s, is_open, _ in items if is_open]
|
||
closed_items = [s for s, _, is_closed in items if is_closed]
|
||
unknown = [s for s, is_open, is_closed in items if not is_open and not is_closed]
|
||
|
||
lines: List[str] = []
|
||
if open_items:
|
||
lines.append("**Открыто:**")
|
||
for s in open_items:
|
||
lines.append(self._format_entity(s))
|
||
if closed_items:
|
||
lines.append("Закрыто: " + str(len(closed_items)) + " шт.")
|
||
if unknown:
|
||
lines.append("Неизвестно: " + str(len(unknown)) + " шт.")
|
||
|
||
# Фильтр по комнате если указан
|
||
if tokens:
|
||
filtered = []
|
||
for line in lines:
|
||
if any(t in line.lower() for t in tokens):
|
||
filtered.append(line)
|
||
if filtered:
|
||
return filtered
|
||
|
||
return lines
|
||
|
||
def _monitor_lights(self, tokens: List[str]) -> List[str]:
|
||
"""Какой свет горит."""
|
||
cfg = self._MONITOR_DOMAINS["light_status"]
|
||
entities = self._find_entities(tokens if tokens else cfg["keywords"], "light")
|
||
|
||
if not entities:
|
||
return ["свет не найден"]
|
||
|
||
on_items: List[str] = []
|
||
off_items: List[str] = []
|
||
for eid in entities:
|
||
state = self._get_state(eid)
|
||
if state == "on":
|
||
on_items.append(eid)
|
||
else:
|
||
off_items.append(eid)
|
||
|
||
lines: List[str] = []
|
||
if on_items:
|
||
lines.append("Горит: " + str(len(on_items)) + " шт.")
|
||
for eid in on_items[:15]:
|
||
for s in self._get_states():
|
||
if s["entity_id"] == eid:
|
||
lines.append(self._format_entity(s))
|
||
break
|
||
if off_items:
|
||
lines.append("Выключено: " + str(len(off_items)) + " шт.")
|
||
return lines
|
||
|
||
def _monitor_climate(self, tokens: List[str]) -> List[str]:
|
||
"""Какая температура и состояние климата."""
|
||
cfg = self._MONITOR_DOMAINS["climate_status"]
|
||
entities = self._find_entities(tokens if tokens else cfg["keywords"], "climate")
|
||
|
||
if not entities:
|
||
# Если ничего не нашли по climate, ищем sensor.temperature
|
||
entities = self._find_entities(tokens if tokens else ["температур"], "sensor")
|
||
|
||
if not entities:
|
||
return ["климат-устройства не найдены"]
|
||
|
||
lines = ["**Климат:**"]
|
||
for eid in entities[:20]:
|
||
for s in self._get_states():
|
||
if s["entity_id"] == eid:
|
||
lines.append(self._format_entity(s))
|
||
break
|
||
return lines
|
||
|
||
def _monitor_battery(self, tokens: List[str]) -> List[str]:
|
||
"""Какие батарейки садятся."""
|
||
states = self._get_states()
|
||
low_battery: List[Tuple[Dict[str, Any], float]] = []
|
||
|
||
for s in states:
|
||
eid = s["entity_id"]
|
||
device_class = self._attr(s, "device_class", "")
|
||
unit = self._attr(s, "unit_of_measurement", "")
|
||
|
||
is_battery = (
|
||
device_class == "battery"
|
||
or "battery" in eid
|
||
or unit == "%"
|
||
) and (
|
||
"battery" in eid
|
||
or device_class == "battery"
|
||
or "батарейк" in self._entity_text(s)
|
||
)
|
||
|
||
if not is_battery:
|
||
continue
|
||
|
||
try:
|
||
level = float(s.get("state", ""))
|
||
except (ValueError, TypeError):
|
||
continue
|
||
|
||
if level < 30:
|
||
low_battery.append((s, level))
|
||
|
||
if not low_battery:
|
||
return ["Все батарейки в норме (>=30%)"]
|
||
|
||
low_battery.sort(key=lambda x: x[1])
|
||
lines = ["**Низкий заряд:**"]
|
||
for s, level in low_battery[:15]:
|
||
lines.append(self._format_entity(s))
|
||
return lines
|
||
|
||
def _monitor_media(self, tokens: List[str]) -> List[str]:
|
||
"""Состояние медиаплееров."""
|
||
cfg = self._MONITOR_DOMAINS["media_status"]
|
||
entities = self._find_entities(tokens if tokens else cfg["keywords"], "media_player")
|
||
|
||
if not entities:
|
||
return ["медиаплееры не найдены"]
|
||
|
||
playing: List[str] = []
|
||
idle: List[str] = []
|
||
for eid in entities:
|
||
state = self._get_state(eid)
|
||
(playing if state == "playing" else idle).append(eid)
|
||
|
||
lines = ["**Медиа:**"]
|
||
for eid in playing:
|
||
for s in self._get_states():
|
||
if s["entity_id"] == eid:
|
||
fn = self._attr(s, "friendly_name", eid)
|
||
media_title = self._attr(s, "media_title", "")
|
||
artist = self._attr(s, "media_artist", "")
|
||
extra = ""
|
||
if media_title:
|
||
extra = " - " + artist + " - " + media_title
|
||
lines.append(" > " + fn + extra)
|
||
break
|
||
for eid in idle:
|
||
for s in self._get_states():
|
||
if s["entity_id"] == eid:
|
||
lines.append(" || " + self._attr(s, "friendly_name", eid))
|
||
break
|
||
return lines
|
||
|
||
def _monitor_presence(self, tokens: List[str]) -> List[str]:
|
||
"""Кто дома."""
|
||
states = self._get_states()
|
||
home: List[str] = []
|
||
away: List[str] = []
|
||
|
||
for s in states:
|
||
eid = s["entity_id"]
|
||
if not (eid.startswith("person.") or eid.startswith("device_tracker.")):
|
||
continue
|
||
name = self._attr(s, "friendly_name", eid)
|
||
state = s.get("state", "")
|
||
if state == "home":
|
||
home.append(name)
|
||
else:
|
||
away.append(name)
|
||
|
||
lines = ["**Люди дома:**"]
|
||
if home:
|
||
for n in home:
|
||
lines.append(" HOME " + n)
|
||
if away:
|
||
lines.append(" AWAY Не дома: " + ", ".join(away))
|
||
if not home and not away:
|
||
lines.append(" (информация отсутствует)")
|
||
return lines
|
||
|
||
# ========================================================================
|
||
# Главная точка входа
|
||
# ========================================================================
|
||
def smart(self, command: str) -> str:
|
||
"""
|
||
Принимает команду на естественном языке и управляет домом / отвечает на вопросы.
|
||
|
||
Управление:
|
||
- включи свет в спальне
|
||
- выключи весь свет
|
||
- поставь 23 градуса в детской
|
||
- открой шторы в гостиной
|
||
- запусти сцену вечерний отдых
|
||
- пауза музыки
|
||
|
||
Мониторинг:
|
||
- какие окна открыты?
|
||
- что горит лишнего?
|
||
- какая температура в спальне?
|
||
- батарейки садятся?
|
||
- кто дома?
|
||
"""
|
||
if not self.valves.ha_token:
|
||
return (
|
||
"Не задан HA Token. Укажите его в настройках инструмента (Valves) "
|
||
"в интерфейсе Open WebUI."
|
||
)
|
||
|
||
cmd = command.lower().strip()
|
||
tokens = self._tokenize(cmd)
|
||
actions: List[str] = []
|
||
warnings: List[str] = []
|
||
|
||
try:
|
||
# ============================================================
|
||
# Режим мониторинга (вопросы о состоянии)
|
||
# ============================================================
|
||
is_question = (
|
||
cmd.startswith(("как", "сколько", "что", "кто", "где", "какая", "какой", "какое", "какие"))
|
||
or "?" in cmd
|
||
or "сейчас" in cmd
|
||
or "состоян" in cmd
|
||
or "статус" in cmd
|
||
)
|
||
|
||
if is_question:
|
||
matched = False
|
||
|
||
if any(k in cmd for k in ["окн", "двер", "форточ", "калитк", "ворота", "гараж"]):
|
||
actions.extend(self._monitor_windows_doors(tokens))
|
||
matched = True
|
||
|
||
elif any(k in cmd for k in ["свет", "лампочк", "люстр"]) and (
|
||
"горит" in cmd or "включ" in cmd or "лишн" in cmd
|
||
):
|
||
actions.extend(self._monitor_lights(tokens))
|
||
matched = True
|
||
|
||
elif any(k in cmd for k in ["температур", "климат", "тепл", "холод", "кондиционер", "батаре"]):
|
||
actions.extend(self._monitor_climate(tokens))
|
||
matched = True
|
||
|
||
elif "батарейк" in cmd or "заряд" in cmd:
|
||
actions.extend(self._monitor_battery(tokens))
|
||
matched = True
|
||
|
||
elif any(k in cmd for k in ["музык", "медиа", "плеер", "колонк", "телевизор"]):
|
||
actions.extend(self._monitor_media(tokens))
|
||
matched = True
|
||
|
||
elif "дома" in cmd or "присутств" in cmd:
|
||
actions.extend(self._monitor_presence(tokens))
|
||
matched = True
|
||
|
||
if not matched:
|
||
# Общий статус: топ устройств по доменам
|
||
actions.append("**Общий статус дома:**")
|
||
actions.extend(self._monitor_lights([]))
|
||
actions.extend(self._monitor_climate([]))
|
||
|
||
# ============================================================
|
||
# Режим управления (команды действия)
|
||
# ============================================================
|
||
else:
|
||
actions.extend(self._handle_control(cmd, tokens, warnings))
|
||
|
||
# ============================================================
|
||
# Итог
|
||
# ============================================================
|
||
if not actions:
|
||
hint = (
|
||
"Примеры команд:\n"
|
||
" - включи свет в спальне\n"
|
||
" - выключи весь свет\n"
|
||
" - поставь 23 градуса в детской\n"
|
||
" - открой шторы в гостиной\n"
|
||
" - какие окна открыты?\n"
|
||
" - какая температура в спальне?\n"
|
||
" - кто дома?"
|
||
)
|
||
if warnings:
|
||
return "Не удалось выполнить. " + ". ".join(warnings) + "\n\n" + hint
|
||
return "Не понял команду.\n\n" + hint
|
||
|
||
prefix = "[DRY-RUN] " if self.valves.dry_run else ""
|
||
result = prefix + "\n".join(actions)
|
||
if warnings:
|
||
result += "\n\nWARN: " + "; ".join(warnings)
|
||
return result
|
||
|
||
except Exception as e:
|
||
logger.exception("smart() failed")
|
||
return "Ошибка: " + str(e)
|
||
|
||
# ========================================================================
|
||
# Обработка команд управления
|
||
# ========================================================================
|
||
def _handle_control(
|
||
self, cmd: str, tokens: List[str], warnings: List[str]
|
||
) -> List[str]:
|
||
"""Обработка команд действия (вкл/выкл/установить/открыть/...)."""
|
||
actions: List[str] = []
|
||
|
||
# Свет
|
||
if "свет" in cmd or "лампочк" in cmd:
|
||
entities = self._find_entities(tokens, "light")
|
||
if not entities:
|
||
warnings.append("светильники не найдены")
|
||
else:
|
||
turn_off = any(w in cmd for w in ("выключ", "отключ", "погас"))
|
||
brightness = self._extract_number(cmd)
|
||
is_global = not any(
|
||
t in cmd for t in [
|
||
"спальн", "гостин", "кухн", "детск", "ванн",
|
||
"прихож", "коридор", "кабинет", "зал",
|
||
]
|
||
) and len(entities) > 1
|
||
|
||
if is_global and self.valves.confirm_global_actions and not self.valves.dry_run:
|
||
return [
|
||
"Команда подействует на ВСЕ светильники ("
|
||
+ str(len(entities))
|
||
+ " шт.). Укажите комнату (напр. 'в спальне') или подтвердите."
|
||
]
|
||
|
||
count = 0
|
||
for e in entities:
|
||
cur = self._get_state(e)
|
||
if turn_off:
|
||
if cur == "off":
|
||
continue
|
||
self._call_service("light", "turn_off", {"entity_id": e})
|
||
else:
|
||
data: Dict[str, Any] = {"entity_id": e}
|
||
if brightness is not None:
|
||
if brightness <= 100:
|
||
data["brightness"] = int(brightness * 2.55)
|
||
else:
|
||
data["brightness"] = int(brightness)
|
||
if "уют" in cmd or "тплый" in cmd or "тепл" in cmd:
|
||
data["color_temp"] = 320
|
||
elif "холод" in cmd or "бел" in cmd:
|
||
data["color_temp"] = 250
|
||
self._call_service("light", "turn_on", data)
|
||
count += 1
|
||
|
||
if count:
|
||
label = "выключен" if turn_off else "включён"
|
||
actions.append("свет " + label + " (" + str(count) + " шт.)")
|
||
|
||
# Розетки / выключатели
|
||
if any(k in cmd for k in ["розетк", "выключател", "устройств"]):
|
||
entities = self._find_entities(tokens, "switch")
|
||
if not entities:
|
||
warnings.append("розетки не найдены")
|
||
else:
|
||
turn_off = any(w in cmd for w in ("выключ", "отключ"))
|
||
count = 0
|
||
for e in entities:
|
||
cur = self._get_state(e)
|
||
if turn_off and cur == "off":
|
||
continue
|
||
if not turn_off and cur == "on":
|
||
continue
|
||
self._call_service(
|
||
"switch",
|
||
"turn_off" if turn_off else "turn_on",
|
||
{"entity_id": e},
|
||
)
|
||
count += 1
|
||
if count:
|
||
label = "выключены" if turn_off else "включены"
|
||
actions.append("розетки " + label + " (" + str(count) + " шт.)")
|
||
|
||
# Климат
|
||
if any(k in cmd for k in ["температур", "градус", "тепл", "холод", "климат"]):
|
||
temp = self._extract_number(cmd, default=22)
|
||
entities = self._find_entities(tokens, "climate")
|
||
if not entities:
|
||
warnings.append("климат-устройства не найдены")
|
||
else:
|
||
count = 0
|
||
for e in entities:
|
||
self._call_service(
|
||
"climate",
|
||
"set_temperature",
|
||
{"entity_id": e, "temperature": temp},
|
||
)
|
||
count += 1
|
||
if count:
|
||
actions.append("установлено " + str(temp) + "C (" + str(count) + " шт.)")
|
||
|
||
# Режим работы (нагрев/охлаждение/авто)
|
||
if "нагрев" in cmd:
|
||
entities = self._find_entities(tokens, "climate")
|
||
for e in entities:
|
||
self._call_service(
|
||
"climate", "set_hvac_mode",
|
||
{"entity_id": e, "hvac_mode": "heat"},
|
||
)
|
||
actions.append("режим нагрева")
|
||
elif "охлажд" in cmd or "холод" in cmd:
|
||
entities = self._find_entities(tokens, "climate")
|
||
for e in entities:
|
||
self._call_service(
|
||
"climate", "set_hvac_mode",
|
||
{"entity_id": e, "hvac_mode": "cool"},
|
||
)
|
||
actions.append("режим охлаждения")
|
||
|
||
# Медиа
|
||
if any(k in cmd for k in ["музык", "медиа", "плеер", "колонк"]):
|
||
entities = self._find_entities(tokens, "media_player")
|
||
if not entities:
|
||
warnings.append("медиаплееры не найдены")
|
||
else:
|
||
count = 0
|
||
if any(w in cmd for w in ("пауза", "стоп", "останов")):
|
||
for e in entities:
|
||
if self._get_state(e) in ("playing", "paused"):
|
||
self._call_service("media_player", "media_pause", {"entity_id": e})
|
||
count += 1
|
||
if count:
|
||
actions.append("пауза (" + str(count) + " шт.)")
|
||
|
||
elif "дальше" in cmd or "следующ" in cmd:
|
||
for e in entities:
|
||
self._call_service("media_player", "media_next_track", {"entity_id": e})
|
||
count += 1
|
||
if count:
|
||
actions.append("следующий трек")
|
||
|
||
elif "назад" in cmd or "предыдущ" in cmd:
|
||
for e in entities:
|
||
self._call_service("media_player", "media_previous_track", {"entity_id": e})
|
||
count += 1
|
||
if count:
|
||
actions.append("предыдущий трек")
|
||
|
||
elif "громче" in cmd or "тише" in cmd:
|
||
for e in entities:
|
||
self._call_service(
|
||
"media_player",
|
||
"volume_up" if "громче" in cmd else "volume_down",
|
||
{"entity_id": e},
|
||
)
|
||
count += 1
|
||
if count:
|
||
actions.append("громкость (" + str(count) + " шт.)")
|
||
|
||
else:
|
||
for e in entities:
|
||
if self._get_state(e) != "playing":
|
||
self._call_service("media_player", "media_play", {"entity_id": e})
|
||
count += 1
|
||
if count:
|
||
actions.append("воспроизведение (" + str(count) + " шт.)")
|
||
|
||
# Шторы / жалюзи
|
||
if any(k in cmd for k in ["штор", "жалюз", "роллет", "гардин"]):
|
||
entities = self._find_entities(tokens, "cover")
|
||
if not entities:
|
||
warnings.append("шторы не найдены")
|
||
else:
|
||
is_close = any(w in cmd for w in ("закр", "опуст", "спуст"))
|
||
count = 0
|
||
for e in entities:
|
||
if is_close:
|
||
self._call_service("cover", "close_cover", {"entity_id": e})
|
||
else:
|
||
self._call_service("cover", "open_cover", {"entity_id": e})
|
||
count += 1
|
||
if count:
|
||
label = "закрыты" if is_close else "открыты"
|
||
actions.append("шторы " + label + " (" + str(count) + " шт.)")
|
||
|
||
# Замки
|
||
if any(k in cmd for k in ["замок", "заблокир", "закрыт"]):
|
||
entities = self._find_entities(tokens, "lock")
|
||
if entities:
|
||
is_lock = "закры" in cmd or "заблок" in cmd
|
||
for e in entities:
|
||
self._call_service(
|
||
"lock",
|
||
"lock" if is_lock else "unlock",
|
||
{"entity_id": e},
|
||
)
|
||
actions.append(
|
||
"замки " + ("закрыты" if is_lock else "открыты") + " (" + str(len(entities)) + " шт.)"
|
||
)
|
||
|
||
# Вентиляторы
|
||
if "вентилятор" in cmd or "кулер" in cmd:
|
||
entities = self._find_entities(tokens, "fan")
|
||
if entities:
|
||
is_off = "выключ" in cmd
|
||
for e in entities:
|
||
self._call_service(
|
||
"fan",
|
||
"turn_off" if is_off else "turn_on",
|
||
{"entity_id": e},
|
||
)
|
||
actions.append(
|
||
"вентиляторы " + ("выключены" if is_off else "включены") + " (" + str(len(entities)) + " шт.)"
|
||
)
|
||
|
||
# Пылесосы
|
||
if "пылесос" in cmd or "робот" in cmd:
|
||
entities = self._find_entities(tokens, "vacuum")
|
||
if entities:
|
||
if "пауза" in cmd:
|
||
for e in entities:
|
||
self._call_service("vacuum", "pause", {"entity_id": e})
|
||
actions.append("пылесос пауза")
|
||
elif "стоп" in cmd or "верн" in cmd:
|
||
for e in entities:
|
||
self._call_service("vacuum", "stop", {"entity_id": e})
|
||
actions.append("пылесос стоп")
|
||
elif "уборк" in cmd or "запуст" in cmd:
|
||
for e in entities:
|
||
self._call_service("vacuum", "start", {"entity_id": e})
|
||
actions.append("пылесос уборка")
|
||
|
||
# Сцены / скрипты
|
||
if any(k in cmd for k in ["сцен", "режим", "запуст"]):
|
||
all_scripts = self._find_entities([], "script")
|
||
matched: List[str] = []
|
||
for s in all_scripts:
|
||
sl = s.lower()
|
||
for t in tokens:
|
||
if len(t) >= 3 and t in sl:
|
||
matched.append(s)
|
||
break
|
||
if matched:
|
||
for s in matched:
|
||
self._call_service("script", "turn_on", {"entity_id": s})
|
||
actions.append("SCRIPT " + s)
|
||
else:
|
||
warnings.append("подходящая сцена не найдена среди " + str(len(all_scripts)))
|
||
|
||
return actions |