work: Полностью реализовал логику управления бризером.

This commit is contained in:
Fedorov Dmitriy
2026-09-13 19:39:07 +03:00
parent 9241da63c3
commit 5cc125db3f
8 changed files with 373 additions and 9 deletions
+26
View File
@@ -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,
}
#Классы
+7
View File
@@ -0,0 +1,7 @@
from .controller import TionController
from .models import TionState
__all__ = [
"TionController",
"TionState",
]
+219
View File
@@ -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()
+56
View File
@@ -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)
-8
View File
@@ -1,8 +0,0 @@
# Константы и классы
# Константы
# MAC Bluetooth бризера
TION_MAC = "d1:74:9b:eb:ee:a6"
+1
View File
@@ -0,0 +1 @@
tion-btle==3.3.6
+1 -1
View File
@@ -5,7 +5,7 @@ import json
import logging import logging
from tion_btle import TionS4 from tion_btle import TionS4
from my_dataclasses import * from app.my_dataclasses import *
def print_state(state: dict): def print_state(state: dict):
+63
View File
@@ -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())