Controllers.climate

  1"""
  2Контроллер датчика температуры и влажности.
  3
  4Взаимодействует с API Яндекс Умного дома для получения показаний
  5температуры и влажности с климатического датчика.
  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 ClimateController(Controller, YandexDataSource):
 19    """Контроллер для датчика температуры и влажности Яндекс Умного дома.
 20
 21    Получает состояние датчика (температуру, влажность, онлайн-статус)
 22    через Yandex IoT API с повторными попытками при ошибках.
 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: Строка вида "Название: 25°C, влажность 40%".
 62        """
 63        name = self.state.get("name", "Датчик")
 64        temperature = round(self.state.get("temperature", 0), self.round_digits[0])
 65        temperature = int(temperature) if self.round_digits[0] == 0 else temperature
 66        humidity = round(self.state.get("humidity", 0), self.round_digits[0])
 67        humidity = int(humidity) if self.round_digits[0] == 0 else humidity
 68        return f"{name}: {temperature}°C, влажность {humidity}%"
 69
 70    @observe(name="Функция для получения состояния датчика температуры и влажности")
 71    def get_state(self):
 72        """Получает состояние датчика температуры и влажности через Yandex IoT API.
 73
 74        Выполняет GET-запрос к API Яндекс Умного дома с повторными
 75        попытками. Разбирает capabilities и properties устройства.
 76
 77        Returns:
 78            dict: Словарь с ключами device_id, name, online, power,
 79                  temperature, humidity. При ошибках возвращает значения по умолчанию.
 80        """
 81
 82        for attempt in range(self.state_retry_count):
 83            try:
 84                response = requests.get(
 85                    self.url + f"/{self.device_id}", headers=self.yandex_headers, timeout=10
 86                )
 87                response.raise_for_status()
 88
 89                device = response.json()
 90
 91                state = {
 92                    "device_id": self.device_id,
 93                    "name": device.get("name", ""),
 94                    "online": device.get("state", "unknown"),
 95                }
 96
 97                # Обрабатываем capabilities и properties как в первой функции
 98                for capability in device.get("capabilities", []):
 99                    if capability.get("type") == "devices.capabilities.on_off":
100                        state["power"] = capability.get("state", {}).get("value", False)
101
102                for property in device.get("properties", []):
103                    prop_type = property.get("type")
104                    value = property.get("state", {}).get("value")
105
106                    if prop_type == "devices.properties.float":
107                        parameters = property.get("parameters", {})
108                        instance = parameters.get("instance", "")
109
110                        if "temperature" in instance:
111                            state["temperature"] = value
112                        elif "humidity" in instance:
113                            state["humidity"] = value
114
115                self.state = state
116                return state
117            except requests.exceptions.RequestException as e:
118                logger.error(f"Попытка {attempt + 1}/{self.state_retry_count}: {e}")
119                if attempt < self.state_retry_count - 1:
120                    time.sleep(self.state_retry_delay)
121
122        self.state = {
123            "device_id": self.device_id,
124            "name": "",
125            "online": "unknown",
126            "power": False,
127            "temperature": 0,
128            "humidity": 0,
129        }
130        logger.error(
131            f"Не удалось получить состояние датчика после {self.state_retry_count} попыток"
132        )
133        return self.state