Assistants.dog_heater

  1#!/usr/bin/env python3
  2"""
  3Ассистент контроля обогрева будки для собаки.
  4
  5Включает обогрев будки (через розетку) при понижении температуры
  6внутри будки ниже заданного порога.
  7"""
  8
  9import time
 10
 11from e_lib.assistants import Assistant
 12from langfuse import observe
 13from loguru import logger
 14
 15from Controllers import SmartPlugController
 16from Controllers.climate import ClimateController
 17from Messengers.bot import Bot
 18
 19
 20class DogHeaterAssistant(Assistant):
 21    """
 22    Контролирует обогрев будки для собаки.
 23
 24    Получает температуру из датчика климата в будке и
 25    включает/выключает обогреватель через розетку.
 26
 27    Attributes:
 28        smart_plug: Контроллер розетки для обогревателя.
 29        climate: Контроллер датчика температуры в будке.
 30        temp_threshold: Порог температуры для включения обогрева.
 31        bot: Бот для отправки уведомлений.
 32    """
 33
 34    def __init__(
 35        self,
 36        name: str,
 37        smart_plug: SmartPlugController,
 38        climate: ClimateController,
 39        temp_threshold: float,
 40        bot: Bot | None = None,
 41    ):
 42        super().__init__(name)
 43
 44        self.smart_plug = smart_plug
 45        self.climate = climate
 46        self.temp_threshold = temp_threshold
 47        self.bot = bot
 48
 49    @observe(name=f"Получение состояния {__qualname__}")
 50    def get_state(self) -> dict | None:
 51        """
 52        Получает состояние розетки обогревателя с повторными попытками.
 53        Проверяет консистентность состояния: включена — есть ток,
 54        выключена — тока нет.
 55
 56        Returns:
 57            dict: Состояние розетки или None при ошибке.
 58        """
 59        retry_count = self.smart_plug.state_retry_count
 60        retry_delay = self.smart_plug.state_retry_delay
 61
 62        state: dict | None = None
 63        for attempt in range(retry_count):
 64            state = self.smart_plug.get_state()
 65            if (
 66                state is not None
 67                and state.get("power")
 68                and state.get("amperage") is not None
 69                and state["amperage"] > 0
 70            ):
 71                return state
 72            if (
 73                state is not None
 74                and not state.get("power")
 75                and state.get("amperage") is not None
 76                and state.get("amperage") == 0
 77            ):
 78                return state
 79            logger.debug(
 80                f"Попытка {attempt + 1} не удалась, повтор через {retry_delay} секунд..."
 81            )
 82            time.sleep(retry_delay)
 83        if state is not None and state.get("power"):
 84            logger.warning(
 85                "Розетка включена, но потребление тока отсутствует или недоступно после нескольких попыток."
 86            )
 87        if state is not None and not state.get("power"):
 88            logger.warning(
 89                "Розетка выключена, но потребление тока присутствует или недоступно после нескольких попыток."
 90            )
 91
 92        return state
 93
 94    @observe(name=f"Цикл контроля для {__qualname__}")
 95    def run(self) -> None:
 96        """
 97        Выполняет один цикл контроля обогрева будки.
 98        Получает температуру из датчика в будке, сравнивает с порогом
 99        и при необходимости включает или выключает обогреватель.
100        """
101        logger.debug(f'Запуск цикла контроля температуры для "{self.name}"...')
102
103        state = self.climate.get_state()
104        if not state or state.get("temperature") is None:
105            logger.error(
106                f'Не удалось получить температуру, пропускаем цикл "{self.name}"'
107            )
108            return
109        temperature = state["temperature"]
110
111        should_turn_on = temperature <= self.temp_threshold
112
113        plug_state = self.smart_plug.get_state()
114        if not plug_state:
115            logger.error(
116                f'Не удалось получить состояние розетки, пропускаем цикл "{self.name}"'
117            )
118            return
119        if plug_state.get("power") != should_turn_on:
120            success = self.smart_plug.control_plug(should_turn_on)
121            if not success:
122                logger.error(
123                    f'Цикл "{self.name}" завершен с ошибкой управления розеткой'
124                )
125                return
126            # обновляем состояние после попытки
127            plug_state = self.get_state()
128            logger.info(
129                f"Температура {temperature}°C (порог: {self.temp_threshold}°C). "
130                + str(self.smart_plug)
131            )
132            if self.bot:
133                self.bot.send_message(
134                    text=f"Температура {temperature}°C (порог: {self.temp_threshold}°C). "
135                    + str(self.smart_plug)
136                )
137        logger.debug(
138            f'Цикл "{self.name}" завершен: температура {temperature}°C (порог: {self.temp_threshold}°C). '
139            + str(self.smart_plug)
140        )