Controllers.smart_plug

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