Controllers.smart_plug

  1"""
  2Контроллер умной розетки Яндекс Умного дома.
  3
  4Позволяет получать состояние розетки (напряжение, ток, потребление)
  5и управлять ей (включить/выключить) через Yandex IoT API.
  6"""
  7
  8import time
  9import requests
 10
 11from langfuse.decorators import observe
 12from loguru import logger
 13
 14from Controllers.controller import Controller
 15from DataSources import YandexDataSource
 16
 17
 18class SmartPlugController(Controller, YandexDataSource):
 19    """Контроллер для управления умной розеткой Яндекс Умного дома.
 20
 21    Поддерживает получение состояния (напряжение, сила тока,
 22    потребляемая мощность) и включение/выключение розетки.
 23
 24    Attributes:
 25        device_id: Идентификатор устройства в Яндекс Умном доме.
 26        round_digits: Кортеж из трёх значений для округления (вольты, амперы, ватты).
 27        state_retry_count: Количество попыток получения состояния.
 28        state_retry_delay: Задержка между попытками в секундах.
 29        state: Последнее полученное состояние розетки.
 30    """
 31
 32    _default_round_digits = (0, 2, 0)
 33
 34    def __init__(
 35        self, device_id, api_key, round_digits=None, state_retry_count=3, state_retry_delay=2
 36    ):
 37        """Инициализирует контроллер розетки.
 38
 39        Args:
 40            device_id: Идентификатор устройства в Яндекс Умном доме.
 41            api_key: Токен авторизации Yandex IoT API.
 42            round_digits: Кортеж с количеством знаков после запятой.
 43            state_retry_count: Количество попыток получения состояния.
 44            state_retry_delay: Задержка между попытками в секундах.
 45        """
 46
 47        self.device_id = device_id
 48
 49        Controller.__init__(self)
 50        YandexDataSource.__init__(self, api_key)
 51
 52        self.round_digits = round_digits or self._default_round_digits
 53        self.state_retry_count = state_retry_count
 54        self.state_retry_delay = state_retry_delay
 55        self.state = self.get_state()
 56
 57    def __str__(self):
 58        """Возвращает строковое представление состояния розетки.
 59
 60        Returns:
 61            str: Строка вида "розетка: включена, напряжение: 220 вольт, ...".
 62        """
 63        # используем .get() и разные кавычки, чтобы избежать синтаксических ошибок
 64        power = self.state.get("power", False)
 65        vr, ar, cr = self.round_digits
 66        voltage = round(self.state.get("voltage", 0), vr)
 67        voltage = int(voltage) if vr == 0 else voltage
 68        amperage = round(self.state.get("amperage", 0), ar)
 69        amperage = int(amperage) if ar == 0 else amperage
 70        consumption = round(self.state.get("power_consumption", 0), cr)
 71        consumption = int(consumption) if cr == 0 else consumption
 72        return f"розетка: {'включена' if power else 'выключена'}, напряжение: {voltage} вольт, сила тока: {amperage} ампер, потребление: {consumption} ватт"
 73
 74    @observe(name="Функция для получения состояния розетки")
 75    def get_state(self):
 76        """Получает состояние розетки через Yandex IoT API.
 77
 78        Выполняет GET-запрос с повторными попытками. Разбирает
 79        capabilities и properties устройства: напряжение, ток,
 80        потребляемую мощность.
 81
 82        Returns:
 83            dict: Словарь с ключами device_id, name, online, power,
 84                  voltage, amperage, power_consumption. При ошибках
 85                  возвращает значения по умолчанию.
 86        """
 87
 88        for attempt in range(self.state_retry_count):
 89            try:
 90                response = requests.get(
 91                    self.url + f"/{self.device_id}", headers=self.yandex_headers, timeout=10
 92                )
 93                response.raise_for_status()
 94
 95                device = response.json()
 96
 97                state = {
 98                    "device_id": self.device_id,
 99                    "name": device.get("name", ""),
100                    "online": device.get("state", "unknown"),
101                }
102
103                # Обрабатываем capabilities и properties как в первой функции
104                for capability in device.get("capabilities", []):
105                    if capability.get("type") == "devices.capabilities.on_off":
106                        state["power"] = capability.get("state", {}).get("value", False)
107
108                for property in device.get("properties", []):
109                    prop_type = property.get("type")
110                    value = property.get("state", {}).get("value")
111
112                    if prop_type == "devices.properties.float":
113                        parameters = property.get("parameters", {})
114                        instance = parameters.get("instance", "")
115
116                        if "voltage" in instance:
117                            state["voltage"] = value
118                        elif "amperage" in instance:
119                            state["amperage"] = value
120                        elif "power" in instance:
121                            state["power_consumption"] = value
122
123                self.state = state
124                return state
125            except requests.exceptions.RequestException as e:
126                logger.error(f"Попытка {attempt + 1}/{self.state_retry_count}: {e}")
127                if attempt < self.state_retry_count - 1:
128                    time.sleep(self.state_retry_delay)
129
130        self.state = {
131            "device_id": self.device_id,
132            "name": "",
133            "online": "unknown",
134            "power": False,
135            "voltage": 0,
136            "amperage": 0,
137            "power_consumption": 0,
138        }
139        logger.error(
140            f"Не удалось получить состояние розетки после {self.state_retry_count} попыток"
141        )
142        return self.state
143
144    @observe(name="Управление розеткой (включение/выключение)")
145    def control_plug(self, turn_on):
146        """Включает или выключает розетку через Yandex IoT API.
147
148        Отправляет POST-запрос с действием on_off на устройство.
149
150        Args:
151            turn_on: True для включения, False для выключения.
152
153        Returns:
154            bool: True если команда выполнена успешно, False при ошибке.
155        """
156        try:
157            payload = {
158                "devices": [
159                    {
160                        "id": self.device_id,
161                        "actions": [
162                            {
163                                "type": "devices.capabilities.on_off",
164                                "state": {"instance": "on", "value": turn_on},
165                            }
166                        ],
167                    }
168                ]
169            }
170
171            response = requests.post(
172                self.url + "/actions", headers=self.yandex_headers, json=payload, timeout=10
173            )
174
175            response.raise_for_status()
176
177            action_text = "включена" if turn_on else "выключена"
178            logger.debug(f"Розетка успешно {action_text}")
179
180            return True
181
182        except requests.exceptions.RequestException as e:
183            logger.error(f"Ошибка при управлении розеткой: {e}")
184            if hasattr(e, "response") and e.response is not None:
185                logger.error(f"Статус код: {e.response.status_code}")
186                logger.error(f"Ответ: {e.response.text}")
187            return False
188        except Exception as e:
189            logger.error(f"Неожиданная ошибка при управлении розеткой: {e}")
190            return False