Assistants.voltage

  1"""
  2Ассистент контроля напряжения в электросети.
  3
  4Периодически опрашивает розетку с измерением напряжения и
  5отправляет уведомления при выходе за установленные пороги.
  6"""
  7
  8from langfuse.decorators import observe
  9from loguru import logger
 10
 11from Assistants.assistant import Assistant
 12from Controllers.smart_plug import SmartPlugController
 13from Messengers.bot import Bot
 14
 15
 16class VoltageAssistant(Assistant):
 17    """Контролирует напряжение в электросети.
 18
 19    Получает показания напряжения из розетки и сравнивает с
 20    верхним и нижним порогами. При выходе за пределы отправляет
 21    предупреждение через бота.
 22
 23    Attributes:
 24        voltage_threshold_low: Нижний порог напряжения.
 25        voltage_threshold_high: Верхний порог напряжения.
 26        rounding: Количество знаков после запятой для округления.
 27        smart_plug: Контроллер розетки.
 28        bot: Бот для отправки уведомлений.
 29    """
 30
 31    def __init__(
 32        self,
 33        name,
 34        smart_plug_controller: SmartPlugController,
 35        voltage_threshold_low=200,
 36        voltage_threshold_high=250,
 37        rounding=0,
 38        bot: Bot = None,
 39    ):
 40        """Инициализирует ассистента контроля напряжения.
 41
 42        Args:
 43            name: Имя ассистента.
 44            smart_plug_controller: Контроллер розетки.
 45            voltage_threshold_low: Нижний порог напряжения.
 46            voltage_threshold_high: Верхний порог напряжения.
 47            rounding: Количество знаков после запятой для округления.
 48            bot: Бот для отправки уведомлений.
 49        """
 50        super().__init__(name)
 51        self.voltage_threshold_low = voltage_threshold_low
 52        self.voltage_threshold_high = voltage_threshold_high
 53        self.rounding = rounding
 54        self.smart_plug = smart_plug_controller
 55        self.bot = bot
 56
 57    def handle_high_voltage(self, voltage):
 58        """Обрабатывает превышение верхнего порога напряжения.
 59
 60        Args:
 61            voltage: Текущее значение напряжения.
 62        """
 63        logger.warning(f"Напряжение достигло порога: {voltage}V >= {self.voltage_threshold_high}V")
 64        if self.bot:
 65            self.bot.send_message(
 66                f"Напряжение достигло порога: {voltage}V >= {self.voltage_threshold_high}V"
 67            )
 68        # Дополнительные действия, например, отправка уведомления
 69
 70    def handle_low_voltage(self, voltage):
 71        """Обрабатывает понижение напряжения ниже порога.
 72
 73        Args:
 74            voltage: Текущее значение напряжения.
 75        """
 76        logger.warning(f"Напряжение ниже, чем порог: {voltage}V < {self.voltage_threshold_low}V")
 77        if self.bot:
 78            self.bot.send_message(
 79                f"Напряжение достигла порога: {voltage}V <= {self.voltage_threshold_low}V"
 80            )
 81        # Дополнительные действия, например, отправка уведомления
 82
 83    @observe(name=f"Цикл контроля напряжения {__qualname__}")
 84    def run(self):
 85        """Выполняет один цикл контроля напряжения.
 86
 87        Получает состояние розетки, округляет напряжение и сравнивает
 88        с порогами. При выходе за пределы вызывает соответствующие
 89        обработчики.
 90        """
 91        logger.debug(f'Запуск цикла контроля напряжения для "{self.name}"...')
 92        state = self.smart_plug.get_state()
 93        if state is None:
 94            logger.error(f'Не удалось получить состояние розетки, пропускаем цикл "{self.name}"')
 95            return
 96        voltage = state["voltage"]
 97        rounded_voltage = round(voltage, self.rounding)
 98        rounded_voltage = int(voltage) if self.rounding == 0 else rounded_voltage
 99        if rounded_voltage >= self.voltage_threshold_high:
100            self.handle_high_voltage(rounded_voltage)
101        elif rounded_voltage < self.voltage_threshold_low:
102            self.handle_low_voltage(rounded_voltage)