ClimatController/schedule/service.py

import asyncio

from app.tion.service import TionService

from datetime import (
    date,
    datetime,
    timedelta,
)

from .models import (
    ScheduleActionType,
    ScheduleConfig,
    ScheduleOccurrence,
    ScheduleResolution,
    ScheduledSettings,
)


WEEKDAYS = (
    "mon",
    "tue",
    "wed",
    "thu",
    "fri",
    "sat",
    "sun",
)


class ScheduleService:

    def __init__(
            self,
            config,
            tion: TionService | None = None,
            check_interval: float = 5.0,
    ):
        self._config = config
        self._tion = tion
        self._check_interval = check_interval

        self._running = False
        self._task = None
        self._last_applied_when = None

    @property
    def config(self) -> ScheduleConfig:
        return self._config

    def resolve(
        self,
        now: datetime | None = None,
    ) -> ScheduleResolution:
        """
        Определить:
        - текущую точку;
        - следующую точку;
        - накопленные SET-настройки;
        - активен ли сейчас AUTO.
        """

        if now is None:
            now = datetime.now()

        if not self._config.enabled:
            return ScheduleResolution(
                enabled=False,
                current=None,
                next=None,
                scheduled_settings=ScheduledSettings(),
                auto_active=False,
            )

        # Недели назад достаточно,
        # поскольку расписание повторяется каждые 7 дней.
        start_date = (
            now.date()
            - timedelta(days=7)
        )

        end_date = (
            now.date()
            + timedelta(days=7)
        )

        occurrences = self._build_occurrences(
            start_date,
            end_date,
        )

        current = None
        next_point = None

        for occurrence in occurrences:

            if occurrence.when <= now:
                current = occurrence
                continue

            next_point = occurrence
            break

        if current is None:
            raise RuntimeError(
                "Could not determine current "
                "schedule point"
            )

        scheduled_settings = (
            self._calculate_settings(
                occurrences,
                current,
            )
        )

        auto_active = (
            current.point.action.type
            == ScheduleActionType.AUTO
        )

        return ScheduleResolution(
            enabled=True,
            current=current,
            next=next_point,
            scheduled_settings=scheduled_settings,
            auto_active=auto_active,
        )

    def _build_occurrences(
        self,
        start_date: date,
        end_date: date,
    ) -> list[ScheduleOccurrence]:

        result: list[ScheduleOccurrence] = []

        current_date = start_date

        while current_date <= end_date:

            weekday = WEEKDAYS[
                current_date.weekday()
            ]

            template_name = (
                self._config.days[weekday]
            )

            template = (
                self._config.templates[
                    template_name
                ]
            )

            for point in template:

                when = datetime.combine(
                    current_date,
                    point.at,
                )

                result.append(
                    ScheduleOccurrence(
                        when=when,
                        weekday=weekday,
                        template=template_name,
                        point=point,
                    )
                )

            current_date += timedelta(days=1)

        result.sort(
            key=lambda occurrence:
            occurrence.when
        )

        return result

    def _calculate_settings(
        self,
        occurrences: list[
            ScheduleOccurrence
        ],
        current: ScheduleOccurrence,
    ) -> ScheduledSettings:
        """
        Восстановить последние SET-значения
        каждого параметра расписания.
        """

        settings = ScheduledSettings()

        for occurrence in occurrences:

            if occurrence.when > current.when:
                break

            action = occurrence.point.action

            if (
                action.type
                != ScheduleActionType.SET
            ):
                continue

            settings = settings.merged(
                action.settings
            )

        return settings

    async def start(self) -> None:
        if self._running:
            return

        if self._tion is None:
            raise RuntimeError(
                "TionService is required to start schedule"
            )

        self._running = True

        # При запуске сразу приводим Tion
        # к текущему состоянию расписания.
        await self._process(initial=True)

        self._task = asyncio.create_task(
            self._loop()
        )

    async def stop(self) -> None:
        self._running = False

        if self._task is not None:
            self._task.cancel()

            try:
                await self._task
            except asyncio.CancelledError:
                pass

            self._task = None

    async def _loop(self) -> None:
        while self._running:
            await asyncio.sleep(
                self._check_interval
            )

            await self._process()

    async def _process(
            self,
            *,
            initial: bool = False,
    ) -> None:

        resolution = self.resolve()

        if not resolution.enabled:
            return

        current = resolution.current

        if current is None:
            return

        # Эта точка уже применена.
        if (
                not initial
                and self._last_applied_when == current.when
        ):
            return

        # AUTO пока только распознаём.
        # Сам AutoController подключим позже.
        if (
                current.point.action.type
                == ScheduleActionType.AUTO
        ):
            self._last_applied_when = current.when
            return

        if initial:
            # После запуска восстанавливаем всё
            # накопленное состояние расписания.
            settings = resolution.scheduled_settings
        else:
            # При обычном переходе применяем
            # только параметры новой точки.
            settings = current.point.action.settings

        await self._apply_settings(settings)

        self._last_applied_when = current.when

    async def _apply_settings(
            self,
            settings: ScheduledSettings,
    ) -> None:

        if self._tion is None:
            return

        # Если нужно включить Tion —
        # сначала включаем.
        if settings.power is True:
            await self._tion.execute(
                lambda tion: tion.power_on()
            )

        if settings.mode is not None:
            await self._tion.execute(
                lambda tion:
                tion.set_air_mode(settings.mode)
            )

        if settings.target_temp is not None:
            await self._tion.execute(
                lambda tion:
                tion.set_target_temperature(
                    settings.target_temp
                )
            )

        if settings.heater is not None:
            if settings.heater:
                await self._tion.execute(
                    lambda tion: tion.heater_on()
                )
            else:
                await self._tion.execute(
                    lambda tion: tion.heater_off()
                )

        if settings.speed is not None:
            await self._tion.execute(
                lambda tion:
                tion.set_speed(settings.speed)
            )

        if settings.sound is not None:
            if settings.sound:
                await self._tion.execute(
                    lambda tion: tion.sound_on()
                )
            else:
                await self._tion.execute(
                    lambda tion: tion.sound_off()
                )

        if settings.light is not None:
            if settings.light:
                await self._tion.execute(
                    lambda tion: tion.light_on()
                )
            else:
                await self._tion.execute(
                    lambda tion: tion.light_off()
                )

        # Если нужно выключить —
        # выключаем последним.
        if settings.power is False:
            await self._tion.execute(
                lambda tion: tion.power_off()
            )