Assistants.climate

  1#!/usr/bin/env python3
  2"""
  3Ассистент контроля температуры и влажности.
  4
  5Периодически опрашивает датчик температуры/влажности и отправляет
  6уведомления через бота при выходе показателей за установленные пороги.
  7"""
  8
  9from e_lib.assistants import ThresholdMonitorAssistant
 10from langfuse import observe
 11from loguru import logger
 12
 13from Controllers.climate import ClimateController
 14
 15
 16class ClimateAssistant(ThresholdMonitorAssistant):
 17    """Контролирует температуру и влажность в помещении.
 18
 19    При выходе температуры или влажности за заданные пороги
 20    отправляет предупреждение через Telegram-бота.
 21
 22    Attributes:
 23        temp_threshold_low: Нижний порог температуры.
 24        temp_threshold_high: Верхний порог температуры.
 25        humidity_threshold_low: Нижний порог влажности.
 26        humidity_threshold_high: Верхний порог влажности.
 27        rounding: Количество знаков после запятой для округления.
 28        temperature_controller: Контроллер датчика температуры/влажности.
 29        bot: Бот для отправки уведомлений.
 30    """
 31
 32    def __init__(
 33        self,
 34        name,
 35        temperature_controller: ClimateController,
 36        temperature_threshold_low=23,
 37        temperature_threshold_high=25,
 38        humidity_threshold_low=25,
 39        humidity_threshold_high=50,
 40        rounding=0,
 41        bot=None,
 42    ):
 43        """Инициализирует ассистента контроля климата.
 44
 45        Args:
 46            name: Имя ассистента.
 47            temperature_controller: Контроллер датчика температуры/влажности.
 48            temperature_threshold_low: Нижний порог температуры.
 49            temperature_threshold_high: Верхний порог температуры.
 50            humidity_threshold_low: Нижний порог влажности.
 51            humidity_threshold_high: Верхний порог влажности.
 52            rounding: Количество знаков после запятой для округления.
 53            bot: Бот для отправки уведомлений.
 54        """
 55        super().__init__(
 56            name,
 57            threshold_low=temperature_threshold_low,
 58            threshold_high=temperature_threshold_high,
 59            rounding=rounding,
 60            notifier=bot,
 61            logger=logger,
 62        )
 63        self.temp_threshold_low = temperature_threshold_low
 64        self.temp_threshold_high = temperature_threshold_high
 65        self.humidity_threshold_low = humidity_threshold_low
 66        self.humidity_threshold_high = humidity_threshold_high
 67        self.rounding = rounding
 68        self.temperature_controller = temperature_controller
 69        self.bot = bot
 70
 71    def read_metric(self):
 72        """Получает текущую температуру как основную метрику.
 73
 74        Возвращается температура из датчика; метод нужен для соответствия
 75        контракту ThresholdMonitorAssistant, полный цикл контроля обеих
 76        метрик выполняет переопределённый run().
 77
 78        Returns:
 79            Температура в градусах Цельсия или None, если состояние
 80            получить не удалось.
 81        """
 82        state = self.temperature_controller.get_state()
 83        if state is None:
 84            return None
 85        return state["temperature"]
 86
 87    def _check_temp(self, temp):
 88        """Сравнивает температуру с порогами и уведомляет при выходе.
 89
 90        Args:
 91            temp: Текущее значение температуры.
 92        """
 93        rounded_temp = self._round(temp)
 94        if rounded_temp > self.temp_threshold_high:
 95            self.handle_high_temp(rounded_temp)
 96        elif rounded_temp < self.temp_threshold_low:
 97            self.handle_low_temp(rounded_temp)
 98
 99    def _check_humidity(self, humidity):
100        """Сравнивает влажность с порогами и уведомляет при выходе.
101
102        Args:
103            humidity: Текущее значение влажности.
104        """
105        rounded_humidity = self._round(humidity)
106        if rounded_humidity > self.humidity_threshold_high:
107            self.handle_high_humidity(rounded_humidity)
108        elif rounded_humidity < self.humidity_threshold_low:
109            self.handle_low_humidity(rounded_humidity)
110
111    def handle_high_temp(self, temp):
112        """Обрабатывает превышение верхнего порога температуры.
113
114        Args:
115            temp: Текущее значение температуры.
116        """
117        self._logger.warning(
118            f"Температура превысила порог: {temp}°C > {self.temp_threshold_high}°C"
119        )
120        if self.bot:
121            self.bot.send_message(
122                f"Температура превысила порог: {temp}°C больше {self.temp_threshold_high}°C"
123            )
124
125    def handle_low_temp(self, temp):
126        """Обрабатывает понижение температуры ниже порога.
127
128        Args:
129            temp: Текущее значение температуры.
130        """
131        self._logger.warning(
132            f"Температура ниже, чем порог: {temp}°C < {self.temp_threshold_low}°C"
133        )
134        if self.bot:
135            self.bot.send_message(
136                f"Температура ниже, чем порог: {temp}°C меньше {self.temp_threshold_low}°C"
137            )
138
139    def handle_high_humidity(self, humidity):
140        """Обрабатывает превышение верхнего порога влажности.
141
142        Args:
143            humidity: Текущее значение влажности.
144        """
145        self._logger.warning(
146            f"Влажность превысила порог: {humidity}% > {self.humidity_threshold_high}%"
147        )
148        if self.bot:
149            self.bot.send_message(
150                f"Влажность превысила порог: {humidity}% больше {self.humidity_threshold_high}%"
151            )
152
153    def handle_low_humidity(self, humidity):
154        """Обрабатывает понижение влажности ниже порога.
155
156        Args:
157            humidity: Текущее значение влажности.
158        """
159        self._logger.warning(
160            f"Влажность ниже, чем порог: {humidity}% < {self.humidity_threshold_low}%"
161        )
162        if self.bot:
163            self.bot.send_message(
164                f"Влажность ниже, чем порог: {humidity}% меньше {self.humidity_threshold_low}%"
165            )
166
167    @observe(name=f"Цикл контроля температуры и влажности {__qualname__}")
168    def run(self):
169        """Выполняет один цикл контроля температуры и влажности.
170
171        Получает показания с датчика, округляет их и сравнивает
172        с заданными порогами. При выходе за пределы вызывает
173        соответствующие обработчики.
174        """
175        self._logger.debug(
176            f'Запуск цикла контроля температуры и влажности для "{self.name}"...'
177        )
178        state = self.temperature_controller.get_state()
179        if state is None:
180            self._logger.error(
181                f'Не удалось получить состояние датчика, пропускаем цикл "{self.name}"'
182            )
183            return
184        self._check_temp(state["temperature"])
185        self._check_humidity(state["humidity"])