Controllers.climate

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