From 5cc125db3f85b231ae3fe97425ddaeff9995cbdf Mon Sep 17 00:00:00 2001 From: Fedorov Dmitriy Date: Sun, 13 Sep 2026 19:39:07 +0300 Subject: [PATCH] =?UTF-8?q?work:=20=D0=9F=D0=BE=D0=BB=D0=BD=D0=BE=D1=81?= =?UTF-8?q?=D1=82=D1=8C=D1=8E=20=D1=80=D0=B5=D0=B0=D0=BB=D0=B8=D0=B7=D0=BE?= =?UTF-8?q?=D0=B2=D0=B0=D0=BB=20=D0=BB=D0=BE=D0=B3=D0=B8=D0=BA=D1=83=20?= =?UTF-8?q?=D1=83=D0=BF=D1=80=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D1=8F=20?= =?UTF-8?q?=D0=B1=D1=80=D0=B8=D0=B7=D0=B5=D1=80=D0=BE=D0=BC.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/my_dataclasses.py | 26 ++++ app/tion/__init__.py | 7 ++ app/tion/controller.py | 219 ++++++++++++++++++++++++++++++++++ app/tion/models.py | 56 +++++++++ my_dataclasses.py | 8 -- requirements.txt | 1 + tests/test-tion.py | 2 +- tests/test_tion_controller.py | 63 ++++++++++ 8 files changed, 373 insertions(+), 9 deletions(-) create mode 100644 app/my_dataclasses.py create mode 100644 app/tion/__init__.py create mode 100644 app/tion/controller.py create mode 100644 app/tion/models.py delete mode 100644 my_dataclasses.py create mode 100644 requirements.txt create mode 100644 tests/test_tion_controller.py diff --git a/app/my_dataclasses.py b/app/my_dataclasses.py new file mode 100644 index 0000000..ef9f5a1 --- /dev/null +++ b/app/my_dataclasses.py @@ -0,0 +1,26 @@ +from dataclasses import asdict, dataclass +from typing import Any, Mapping + +# Константы и классы + +# Константы + +# MAC Bluetooth бризера +TION_MAC = "d1:74:9b:eb:ee:a6" + +MIN_FAN_SPEED = 1 +MAX_FAN_SPEED = 6 + +MIN_TARGET_TEMP = 5 +MAX_TARGET_TEMP = 30 + +AIR_MODE_OUTSIDE = "outside" +AIR_MODE_RECIRCULATION = "recirculation" + +SUPPORTED_AIR_MODES = { + AIR_MODE_OUTSIDE, + AIR_MODE_RECIRCULATION, +} + +#Классы + diff --git a/app/tion/__init__.py b/app/tion/__init__.py new file mode 100644 index 0000000..f2fb64f --- /dev/null +++ b/app/tion/__init__.py @@ -0,0 +1,7 @@ +from .controller import TionController +from .models import TionState + +__all__ = [ + "TionController", + "TionState", +] diff --git a/app/tion/controller.py b/app/tion/controller.py new file mode 100644 index 0000000..b273ee9 --- /dev/null +++ b/app/tion/controller.py @@ -0,0 +1,219 @@ +import asyncio +from typing import Any + +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(mac) + + # Не позволяем двум частям программы одновременно работать с 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.connection_status == "connected" + + @property + def state(self) -> TionState | None: + """ + Последнее известное состояние без обращения к Bluetooth. + """ + return self._state + + """ + Соединение + """ + async def connect(self) -> None: + """ + Открыть постоянное BLE-соединение. + """ + async with self._lock: + if self._started: + return + + await self._device.connect() + self._started = True + + async def disconnect(self) -> None: + """ + Закрыть BLE-соединение. + """ + async with self._lock: + if not self._started: + return + + try: + await self._device.disconnect() + finally: + self._started = False + + """ + Состояние + """ + + 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() \ No newline at end of file diff --git a/app/tion/models.py b/app/tion/models.py new file mode 100644 index 0000000..06c0677 --- /dev/null +++ b/app/tion/models.py @@ -0,0 +1,56 @@ +from dataclasses import asdict, dataclass +from typing import Any, Mapping + + +def _is_on(value: Any) -> bool: + return value == "on" + + +@dataclass(frozen=True, slots=True) +class TionState: + power: bool + heater: bool + heating: bool + sound: bool + mode: str + + out_temp: int + in_temp: int + target_temp: int + + fan_speed: int + filter_remain: float + + device_time: str + request_error_code: int + model: str + + light: bool | None = None + + @classmethod + def from_raw(cls, data: Mapping[str, Any]) -> "TionState": + light_raw = data.get("light") + + return cls( + power=_is_on(data["state"]), + heater=_is_on(data["heater"]), + heating=_is_on(data["heating"]), + sound=_is_on(data["sound"]), + mode=str(data["mode"]), + + out_temp=int(data["out_temp"]), + in_temp=int(data["in_temp"]), + target_temp=int(data["heater_temp"]), + + fan_speed=int(data["fan_speed"]), + filter_remain=float(data["filter_remain"]), + + device_time=str(data["time"]), + request_error_code=int(data["request_error_code"]), + model=str(data["model"]), + + light=None if light_raw is None else _is_on(light_raw), + ) + + def to_dict(self) -> dict: + return asdict(self) \ No newline at end of file diff --git a/my_dataclasses.py b/my_dataclasses.py deleted file mode 100644 index 2d68a11..0000000 --- a/my_dataclasses.py +++ /dev/null @@ -1,8 +0,0 @@ - - -# Константы и классы - -# Константы - -# MAC Bluetooth бризера -TION_MAC = "d1:74:9b:eb:ee:a6" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a83753b --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +tion-btle==3.3.6 \ No newline at end of file diff --git a/tests/test-tion.py b/tests/test-tion.py index fa38831..98d66de 100644 --- a/tests/test-tion.py +++ b/tests/test-tion.py @@ -5,7 +5,7 @@ import json import logging from tion_btle import TionS4 -from my_dataclasses import * +from app.my_dataclasses import * def print_state(state: dict): diff --git a/tests/test_tion_controller.py b/tests/test_tion_controller.py new file mode 100644 index 0000000..9cba0b3 --- /dev/null +++ b/tests/test_tion_controller.py @@ -0,0 +1,63 @@ +import asyncio +import json +import logging + +from app.tion import TionController +from app.my_dataclasses import * + + + + +def print_state(title: str, state) -> None: + print() + print("=" * 50) + print(title) + print("=" * 50) + + print(json.dumps( + state.to_dict(), + indent=2, + ensure_ascii=False + )) + + +async def main(): + # Убираем лишнее логирование + logging.getLogger("asyncio").setLevel(logging.WARNING) + logging.getLogger("bleak").setLevel(logging.WARNING) + logging.getLogger("tion_btle").setLevel(logging.WARNING) + + tion = TionController(TION_MAC) + + print("Подключение к Tion...") + + async with tion: + + print(f"Connected: {tion.connected}") + + # Читаем исходное состояние + state = await tion.get_state() + print_state("Исходное состояние", state) + + # Включаем + state = await tion.power_on() + print_state("После POWER ON", state) + + await asyncio.sleep(1) + + # Скорость 2 + state = await tion.set_speed(2) + print_state("После SPEED 2", state) + + await asyncio.sleep(1) + + # Скорость 3 + state = await tion.set_speed(3) + print_state("После SPEED 3", state) + + print() + print(f"Connected after exit: {tion.connected}") + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file