ClimatController/app/tion/controller.py

import asyncio
from typing import Any
from bleak import BleakScanner
from tion_btle import TionS4

from .models import TionState

from app.my_dataclasses import *


class TionController:
    """
    Высокоуровневый контроллер Tion 4S.

    Один экземпляр TionController должен владеть одним BLE-соединением
    с бризером на протяжении всей работы приложения.
    """

    def __init__(self, mac: str):
        self._mac = mac
        self._device:  TionS4 | None = None

        # Не позволяем двум частям программы одновременно работать с BLE.
        self._lock = asyncio.Lock()

        # Показывает, что мы удерживаем внешнее соединение tion-btle.
        self._started = False

        # Последнее успешно прочитанное состояние.
        self._state: TionState | None = None

    @property
    def mac(self) -> str:
        return self._mac

    @property
    def connected(self) -> bool:
        return (
                self._device is not None
                and self._device.connection_status == "connected"
        )

    async def _reset_device(self) -> None:
        """
        Полностью уничтожить текущий BLE transport.

        Важно:
        tion-btle не вызывает BleakClient.disconnect(),
        если WinRT уже считает устройство disconnected.

        Поэтому при полном reset принудительно закрываем
        BleakClient напрямую.
        """

        old_device = self._device

        # Controller сразу больше не считает старый объект рабочим.
        self._device = None
        self._started = False

        if old_device is None:
            return

        try:
            # Нам нужен именно настоящий BleakClient.disconnect().
            #
            # Не old_device.disconnect(), потому что tion-btle
            # может пропустить физический cleanup при
            # connection_status == "disc".
            await old_device._btle.disconnect()

        except Exception:
            # Старый transport всё равно больше использоваться
            # не будет.
            pass

    """
    Соединение
    """

    async def connect(self) -> None:
        async with self._lock:
            if self.connected:
                self._started = True
                return

            # Полностью закрываем всё, что осталось
            # от предыдущего соединения.
            await self._reset_device()

            # Получаем свежий BLEDevice.
            ble_device = await BleakScanner.find_device_by_address(
                self._mac,
                timeout=5.0,
            )

            if ble_device is None:
                raise ConnectionError(
                    f"Tion {self._mac} not found"
                )

            # Новый TionS4 = новый BleakClient.
            device = TionS4(ble_device)

            try:
                await device.connect()

            except Exception:
                # ВАЖНО:
                # освобождаем даже частично созданную
                # WinRT/GATT-сессию.
                try:
                    await device._btle.disconnect()
                except Exception:
                    pass

                raise

            self._device = device
            self._started = True

    async def disconnect(self) -> None:
        """
        Закрыть BLE-соединение.
        """
        async with self._lock:
            await self._reset_device()


    """
    Состояние
    """

    async def get_state(self) -> TionState:
        """
        Получить реальное текущее состояние бризера.
        """
        self._ensure_started()

        async with self._lock:
            raw_state = await self._device.get()

            state = TionState.from_raw(raw_state)
            self._state = state

            return state

    """
    Питание
    """

    async def power_on(self) -> TionState:
        return await self._set({
            "state": "on"
        })

    async def power_off(self) -> TionState:
        return await self._set({
            "state": "off"
        })

    """
    Скорость вентилятора 
    """

    async def set_speed(self, speed: int) -> TionState:
        if not MIN_FAN_SPEED <= speed <= MAX_FAN_SPEED:
            raise ValueError("Tion fan speed must be between 1 and 6")

        return await self._set({
            "fan_speed": speed
        })

    """
    Целевая температура
    """

    async def set_target_temperature(self, temperature: int) -> TionState:
        if not MIN_TARGET_TEMP <= temperature <= MAX_TARGET_TEMP:
            raise ValueError(
                f"Tion target temperature must be between "
                f"{MIN_TARGET_TEMP} and {MAX_TARGET_TEMP} °C"
            )

        return await self._set({
            "heater_temp": temperature
        })

    """
    Нагрев
    """

    async def heater_on(self) -> TionState:
        return await self._set({
            "heater": "on"
        })

    async def heater_off(self) -> TionState:
        return await self._set({
            "heater": "off"
        })

    """
    Забор воздуха
    """

    async def set_air_mode(self, mode: str) -> TionState:
        if mode not in SUPPORTED_AIR_MODES:
            raise ValueError(
                f"Unsupported Tion air mode: {mode}. "
                f"Available modes: {sorted(SUPPORTED_AIR_MODES)}"
            )

        return await self._set({
            "mode": mode
        })

    """
    Звук
    """
    async def sound_on(self) -> TionState:
        return await self._set({
            "sound": "on"
        })

    async def sound_off(self) -> TionState:
        return await self._set({
            "sound": "off"
        })

    """
    Световая индикация
    """

    async def light_on(self) -> TionState:
        return await self._set({
            "light": "on"
        })

    async def light_off(self) -> TionState:
        return await self._set({
            "light": "off"
        })

    async def _set(self, settings: dict[str, Any]) -> TionState:
        """
        Отправить настройки в Tion и затем прочитать
        фактическое состояние устройства.
        """

        self._ensure_started()

        async with self._lock:
            await self._device.set(settings)

            raw_state = await self._device.get()

            state = TionState.from_raw(raw_state)
            self._state = state

            return state

    def _ensure_started(self) -> None:
        if not self._started:
            raise RuntimeError(
                "TionController is not connected. "
                "Call await controller.connect() first."
            )

    async def __aenter__(self) -> "TionController":
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_value, traceback) -> None:
        await self.disconnect()