Home/ClimatController
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

work: Добваил нагрев в автоматизацию

Browse Source
This commit is contained in:
Fedorov Dmitriy
2026-09-19 18:09:03 +03:00
parent bff8efb85b
commit c6d92c10ee
19 changed files with 2661 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api.py
+16 -11
View File
@@ -26,6 +26,8 @@ from app.auto import (
load_auto_config,
)
from app.auto.temperature_policy import TemperaturePolicy
from app.tion import (
TionController,
TionService,
@@ -108,9 +110,7 @@ async def lifespan(app: FastAPI):
schedule_load_error = None
try:
auto_config = load_auto_config(
AUTO_CONFIG_FILE
)
auto_config = load_auto_config(AUTO_CONFIG_FILE)
auto_policy = Co2SpeedPolicy(
base_speed=(
@@ -124,14 +124,15 @@ async def lifespan(app: FastAPI):
),
)
auto_interval = (
auto_config.check_interval
)
temperature_policy = TemperaturePolicy(auto_config.temperature)
auto_interval = (auto_config.check_interval)
auto_load_error = None
except Exception as exc:
auto_policy = None
temperature_policy = None
# Безопасный встроенный интервал нужен,
# потому что auto.yaml сейчас недоступен.
@@ -146,6 +147,7 @@ async def lifespan(app: FastAPI):
qingping_service=qingping_service,
tion_service=service,
policy=auto_policy,
temperature_policy=temperature_policy,
interval=auto_interval,
)
@@ -336,6 +338,8 @@ def get_schedule_status() -> dict:
"current": None,
"next": None,
"auto_active": False,
"auto_fallback_speed": None,
"auto_target_temp": None,
"override_active": False,
"override_until": None,
"override_settings": {},
@@ -393,11 +397,12 @@ def get_schedule_status() -> dict:
"next": occurrence_to_dict(resolution.next),
"auto_active": resolution.auto_active,
"scheduled_settings": (
resolution
.scheduled_settings
.to_dict()
),
"auto_fallback_speed": resolution.auto_fallback_speed,
"auto_target_temp": resolution.auto_target_temp,
"scheduled_settings": resolution.scheduled_settings.to_dict(),
"override_active":schedule_service.override_active,
app/auto/config.py
+81 -1
View File
@@ -32,6 +32,14 @@ class AutoConfig:
version: int
check_interval: float
co2: Co2Config
temperature: TemperatureConfig
@dataclass(
frozen=True,
slots=True,
)
class TemperatureConfig:
hysteresis: float
def _require_dict(name: str, value: Any) -> dict:
@@ -259,6 +267,7 @@ def load_auto_config(path: str | Path) -> AutoConfig:
"version",
"check_interval",
"co2",
"temperature",
}
if unknown:
@@ -294,6 +303,10 @@ def load_auto_config(path: str | Path) -> AutoConfig:
"check_interval must be > 0"
)
# --------------------------------------------------
# CO2
# --------------------------------------------------
if "co2" not in data:
raise ValueError(
"co2 config is required"
@@ -303,8 +316,75 @@ def load_auto_config(path: str | Path) -> AutoConfig:
data["co2"]
)
# --------------------------------------------------
# Temperature
# --------------------------------------------------
if "temperature" not in data:
raise ValueError(
"temperature config is required"
)
temperature_data = _require_dict(
"temperature",
data["temperature"],
)
unknown_temperature = (
set(temperature_data)
- {
"hysteresis",
}
)
if unknown_temperature:
raise ValueError(
f"Unknown temperature config fields: "
f"{sorted(unknown_temperature)}"
)
temperature_hysteresis = (
temperature_data.get(
"hysteresis"
)
)
if (
type(temperature_hysteresis)
not in {
int,
float,
}
):
raise ValueError(
"temperature.hysteresis "
"must be a number"
)
temperature_hysteresis = float(
temperature_hysteresis
)
if temperature_hysteresis < 0:
raise ValueError(
"temperature.hysteresis "
"must be >= 0"
)
temperature = TemperatureConfig(
hysteresis=temperature_hysteresis,
)
# --------------------------------------------------
# Result
# --------------------------------------------------
return AutoConfig(
version=version,
check_interval=float(check_interval),
check_interval=float(
check_interval
),
co2=co2,
temperature=temperature,
)
app/auto/controller.py
+255 -7
View File
@@ -1,6 +1,8 @@
import asyncio
from datetime import datetime
from app.auto.co2_policy import Co2SpeedPolicy
from app.auto.temperature_policy import TemperaturePolicy
from app.my_dataclasses import MAX_TARGET_TEMP
class AutoController:
@@ -10,12 +12,14 @@ class AutoController:
qingping_service,
tion_service,
policy: Co2SpeedPolicy | None = None,
temperature_policy: TemperaturePolicy | None = None,
interval: float = 5.0,
):
self._schedule = schedule_service
self._qingping = qingping_service
self._tion = tion_service
self._policy = policy
self._temperature_policy = temperature_policy
self._interval = interval
@@ -25,10 +29,17 @@ class AutoController:
self._reason: str | None = None
self._target_speed: int | None = None
self._last_error: str | None = None
self._auto_speed: int | None = None
self._target_heater: bool | None = None
self._auto_heater: bool | None = None
self._target_temperature: int | None = None
self._temperature: float | int | None = None
self._temperature_source: str | None = None
self._last_error: str | None = None
async def start(self) -> None:
@@ -86,10 +97,21 @@ class AutoController:
if self._schedule.override_active:
self._state = "suspended"
self._reason = "manual_override"
# AUTO больше не может считать,
# что знает реальное состояние speed/heater:
# пользователь мог изменить их вручную.
self._target_speed = None
self._last_error = None
self._auto_speed = None
self._target_heater = None
self._auto_heater = None
self._target_temperature = None
self._temperature = None
self._temperature_source = None
self._last_error = None
return
# ----------------------------------------------
@@ -102,10 +124,21 @@ class AutoController:
):
self._state = "inactive"
self._reason = None
# Управление возвращается ScheduleService.
# После этого AUTO уже не знает фактическое
# состояние speed/heater, поэтому забываем его.
self._target_speed = None
self._last_error = None
self._auto_speed = None
self._target_heater = None
self._auto_heater = None
self._target_temperature = None
self._temperature = None
self._temperature_source = None
self._last_error = None
return
fallback_speed = resolution.auto_fallback_speed
@@ -126,9 +159,10 @@ class AutoController:
self._reason = "auto_policy_unavailable"
self._auto_speed = None
await self._set_speed(
fallback_speed
)
await self._set_speed(fallback_speed)
if self._temperature_policy is not None:
await self._process_heater(resolution)
self._last_error = None
@@ -159,6 +193,10 @@ class AutoController:
)
self._auto_speed = target_speed
if self._temperature_policy is not None:
await self._process_heater(resolution)
self._last_error = None
return
@@ -181,9 +219,171 @@ class AutoController:
await self._set_speed(fallback_speed)
if self._temperature_policy is not None:
await self._process_heater(
resolution
)
self._last_error = None
async def _process_heater(self, resolution) -> None:
# --------------------------------------------------
# Берём целевую температуру из накопленных
# настроек расписания.
#
# Например:
#
# SET target_temp=20
# AUTO
#
# Во время AUTO target_temp остаётся 20.
# --------------------------------------------------
target_temp = resolution.auto_target_temp
# --------------------------------------------------
# Получаем фактическую температуру.
#
# _get_temperature() сам выбирает:
#
# 1. Qingping
# 2. Tion как резерв
# 3. None, если оба источника недоступны
# --------------------------------------------------
temperature, source = self._get_temperature()
self._temperature = temperature
self._temperature_source = source
# --------------------------------------------------
# Нет температурной policy.
#
# AUTO не умеет безопасно принять решение
# о нагреве → выключаем heater.
# --------------------------------------------------
if self._temperature_policy is None:
await self._set_heater(False)
self._auto_heater = False
return
# --------------------------------------------------
# Нет целевой температуры.
#
# Непонятно, до какой температуры греть.
# Поэтому heater OFF.
# --------------------------------------------------
if target_temp is None:
await self._set_heater(False)
self._auto_heater = False
return
# --------------------------------------------------
# Нет достоверной фактической температуры.
#
# Ни Qingping, ни резервный датчик Tion
# использовать нельзя.
#
# Heater OFF.
# --------------------------------------------------
if temperature is None:
await self._set_heater(False)
self._auto_heater = False
return
# --------------------------------------------------
# Есть всё необходимое:
#
# - фактическая температура;
# - target_temp;
# - TemperaturePolicy.
#
# Policy решает, нужно ли сейчас греть.
# --------------------------------------------------
heater_required = (
self._temperature_policy
.heater_required(
temperature=temperature,
target_temp=target_temp,
heater_on=(
self._auto_heater
is True
),
)
)
if heater_required:
# Наш внешний регулятор решил, что нужно греть.
#
# Сначала поднимаем внутреннюю уставку Tion
# до максимума, чтобы встроенный термостат
# не заблокировал нагрев по своему in_temp.
await self._set_target_temperature(MAX_TARGET_TEMP)
# После этого разрешаем нагреватель.
await self._set_heater(True)
else:
# Целевая температура по внешнему датчику
# достигнута — нагрев запрещаем.
await self._set_heater(False)
self._auto_heater = heater_required
def _get_temperature(self) -> tuple[float | int | None, str | None]:
# --------------------------------------------------
# Основной источник — Qingping.
#
# Используем температуру Qingping только если
# сам поток данных датчика сейчас считается online.
# --------------------------------------------------
qingping_state = self._qingping.state
if (
self._qingping.online
and qingping_state.temperature is not None
):
return (
qingping_state.temperature,
"qingping",
)
# --------------------------------------------------
# Qingping недоступен или температура отсутствует.
#
# Пробуем резервный датчик температуры Tion.
#
# Важно проверять tion.online:
# TionService может хранить последнее состояние,
# даже если связь с устройством уже потеряна.
# --------------------------------------------------
tion_state = self._tion.state
if (
self._tion.online
and tion_state is not None
and tion_state.in_temp is not None
):
return (
tion_state.in_temp,
"tion",
)
# --------------------------------------------------
# Достоверной температуры нет вообще.
# --------------------------------------------------
return None, None
async def _set_speed(self, speed: int) -> None:
if self._target_speed == speed:
@@ -196,13 +396,61 @@ class AutoController:
self._target_speed = speed
async def _set_heater(self,enabled: bool) -> None:
# Если AutoController уже установил именно такое
# состояние нагревателя, повторную команду не шлём.
if self._target_heater == enabled:
return
if enabled:
await self._tion.execute(
lambda controller:
controller.heater_on()
)
else:
await self._tion.execute(
lambda controller:
controller.heater_off()
)
# Запоминаем состояние только после того,
# как команда успешно выполнилась.
self._target_heater = enabled
async def _set_target_temperature(
self,
temperature: int,
) -> None:
if self._target_temperature == temperature:
return
await self._tion.execute(
lambda controller:
controller.set_target_temperature(
temperature
)
)
self._target_temperature = temperature
def status(self) -> dict:
return {
"state": self._state,
"reason": self._reason,
"target_speed": self._target_speed,
"auto_speed": self._auto_speed,
"target_heater": self._target_heater,
"auto_heater": self._auto_heater,
"temperature": self._temperature,
"temperature_source": self._temperature_source,
"last_error": self._last_error,
}
app/auto/temperature_policy.py
+40
View File
@@ -0,0 +1,40 @@
from .config import TemperatureConfig
class TemperaturePolicy:
def __init__(
self,
config: TemperatureConfig,
):
self._config = config
def heater_required(
self,
temperature: float,
target_temp: float,
heater_on: bool,
) -> bool:
# Нагреватель уже включён.
#
# Продолжаем греть, пока температура
# не достигла целевой.
if heater_on:
return (
temperature
< target_temp
)
# Нагреватель выключен.
#
# Повторно включаем его только после
# падения температуры ниже нижней
# границы гистерезиса.
return (
temperature
< (
target_temp
- self._config.hysteresis
)
)
config/auto.yaml
+3
View File
@@ -21,3 +21,6 @@ co2:
- ppm: 2000
speed: 6
temperature:
hysteresis: 0.5
config/schedule.yaml
+7 -7
View File
@@ -17,25 +17,25 @@ templates:
target_temp: 20
mode: outside
- time: "16:17"
- time: "18:01"
action:
type: set
speed: 1
- time: "16:18"
- time: "18:07"
action:
type: auto
speed: 1
target_temp: 30
- time: "16:20"
- time: "18:08"
action:
type: set
target_temp: 30
speed: 6
- time: "17:00"
action:
type: set
speed: 3
- time: "19:00"
action:
schedule/loader.py
+17 -13
View File
@@ -196,9 +196,7 @@ def _parse_action(
action_type_raw = data.get("type")
try:
action_type = ScheduleActionType(
action_type_raw
)
action_type = ScheduleActionType(action_type_raw)
except ValueError as exc:
raise ValueError(
@@ -208,32 +206,38 @@ def _parse_action(
if action_type == ScheduleActionType.AUTO:
extra = set(data) - {
allowed_fields = {
"type",
"speed",
"target_temp",
}
extra = set(data) - allowed_fields
if extra:
raise ValueError(
"AUTO action supports only "
f"'speed': {sorted(extra)}"
"'speed' and 'target_temp': "
f"{sorted(extra)}"
)
if "speed" not in data:
raise ValueError(
"AUTO action must contain "
"fallback speed"
"AUTO action requires 'speed'"
)
settings = _parse_settings(
{
"speed": data["speed"],
}
)
settings_data = {
"speed": data["speed"],
}
if "target_temp" in data:
settings_data["target_temp"] = (
data["target_temp"]
)
return ScheduleAction(
type=ScheduleActionType.AUTO,
settings=settings,
settings=_parse_settings(settings_data),
)
settings_data = {
schedule/models.py
+1
View File
@@ -118,3 +118,4 @@ class ScheduleResolution:
auto_active: bool
auto_fallback_speed: int | None
auto_target_temp: int | None = None
schedule/service.py
+13 -49
View File
@@ -191,15 +191,9 @@ class ScheduleService:
# Недели назад достаточно,
# поскольку расписание повторяется каждые 7 дней.
start_date = (
now.date()
- timedelta(days=7)
)
start_date = now.date() - timedelta(days=7)
end_date = (
now.date()
+ timedelta(days=7)
)
end_date = now.date() + timedelta(days=7)
occurrences = self._build_occurrences(
start_date,
@@ -231,17 +225,14 @@ class ScheduleService:
)
)
auto_active = (
current.point.action.type
== ScheduleActionType.AUTO
)
auto_active = current.point.action.type == ScheduleActionType.AUTO
auto_fallback_speed = None
auto_target_temp = None
if auto_active:
auto_fallback_speed = (
current.point.action.settings.speed
)
auto_fallback_speed = current.point.action.settings.speed
auto_target_temp = current.point.action.settings.target_temp
return ScheduleResolution(
enabled=True,
@@ -250,6 +241,7 @@ class ScheduleService:
scheduled_settings=scheduled_settings,
auto_active=auto_active,
auto_fallback_speed=auto_fallback_speed,
auto_target_temp=auto_target_temp,
)
def _build_occurrences(
@@ -384,39 +376,6 @@ class ScheduleService:
except Exception as exc:
self._last_error = f"{type(exc).__name__}: {exc}"
# 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
#
# settings = resolution.scheduled_settings
#
# await self._apply_settings(settings)
#
# self._last_applied_when = current.when
async def _process(self) -> None:
now = datetime.now()
@@ -477,7 +436,12 @@ class ScheduleService:
# Остальные параметры расписания
# (power, heater, temperature, mode...)
# должны продолжать работать.
settings = replace(settings, speed=None)
settings = replace(
settings,
speed=None,
heater=None,
target_temp=None,
)
await self._apply_settings(settings)
tests/test_auto_heater_command.py
+154
View File
@@ -0,0 +1,154 @@
import asyncio
from app.auto.controller import AutoController
class FakeTionController:
def __init__(self):
self.calls = []
async def heater_on(self):
self.calls.append(
("heater_on", None)
)
async def heater_off(self):
self.calls.append(
("heater_off", None)
)
class FakeTionService:
def __init__(self):
self.controller = (
FakeTionController()
)
async def execute(
self,
operation,
):
return await operation(
self.controller
)
def make_controller():
tion = FakeTionService()
controller = AutoController(
schedule_service=None,
qingping_service=None,
tion_service=tion,
)
return controller, tion
async def test_heater_on():
controller, tion = (
make_controller()
)
await controller._set_heater(
True
)
assert tion.controller.calls == [
("heater_on", None)
]
assert (
controller._target_heater
is True
)
async def test_duplicate_heater_on():
controller, tion = (
make_controller()
)
await controller._set_heater(
True
)
await controller._set_heater(
True
)
assert tion.controller.calls == [
("heater_on", None)
]
async def test_heater_off_after_on():
controller, tion = (
make_controller()
)
await controller._set_heater(
True
)
await controller._set_heater(
False
)
assert tion.controller.calls == [
("heater_on", None),
("heater_off", None),
]
assert (
controller._target_heater
is False
)
async def test_duplicate_heater_off():
controller, tion = (
make_controller()
)
await controller._set_heater(
False
)
await controller._set_heater(
False
)
assert tion.controller.calls == [
("heater_off", None)
]
async def main():
await test_heater_on()
await test_duplicate_heater_on()
await test_heater_off_after_on()
await test_duplicate_heater_off()
print()
print("=" * 70)
print(
"ALL AUTO HEATER COMMAND "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
tests/test_auto_heater_process.py
+304
View File
@@ -0,0 +1,304 @@
import asyncio
from types import SimpleNamespace
from app.auto.config import (
TemperatureConfig,
)
from app.auto.controller import (
AutoController,
)
from app.auto.temperature_policy import (
TemperaturePolicy,
)
class FakeQingpingService:
def __init__(self):
self.online = True
self.state = SimpleNamespace(
temperature=19.4,
)
class FakeTionController:
def __init__(self):
self.calls = []
async def heater_on(self):
self.calls.append(
("heater_on", None)
)
async def heater_off(self):
self.calls.append(
("heater_off", None)
)
class FakeTionService:
def __init__(self):
self.online = True
self.state = SimpleNamespace(
in_temp=18,
)
self.controller = (
FakeTionController()
)
async def execute(
self,
operation,
):
return await operation(
self.controller
)
def make_resolution(
target_temp=20.0,
):
return SimpleNamespace(
scheduled_settings=(
SimpleNamespace(
target_temp=target_temp,
)
)
)
def make_controller():
qingping = (
FakeQingpingService()
)
tion = FakeTionService()
policy = TemperaturePolicy(
TemperatureConfig(
hysteresis=0.5,
)
)
controller = AutoController(
schedule_service=None,
qingping_service=qingping,
tion_service=tion,
temperature_policy=policy,
)
return (
controller,
qingping,
tion,
)
async def test_qingping_heating():
controller, qingping, tion = (
make_controller()
)
resolution = make_resolution()
# 19.4 < 19.5
# Heater должен включиться.
await controller._process_heater(
resolution
)
assert tion.controller.calls == [
("heater_on", None)
]
assert (
controller._auto_heater
is True
)
assert (
controller._temperature
== 19.4
)
assert (
controller._temperature_source
== "qingping"
)
tion.controller.calls.clear()
# ----------------------------------------------
# 19.8
#
# Heater уже ON.
# До 20 градусов ещё не дошли.
# Продолжаем греть.
#
# Новую команду отправлять не нужно.
# ----------------------------------------------
qingping.state.temperature = 19.8
await controller._process_heater(
resolution
)
assert tion.controller.calls == []
assert (
controller._auto_heater
is True
)
# ----------------------------------------------
# 20.0
#
# Цель достигнута.
# ----------------------------------------------
qingping.state.temperature = 20.0
await controller._process_heater(
resolution
)
assert tion.controller.calls == [
("heater_off", None)
]
assert (
controller._auto_heater
is False
)
async def test_tion_temperature_fallback():
controller, qingping, tion = (
make_controller()
)
resolution = make_resolution()
# Qingping отключился.
qingping.online = False
# Используем Tion.
tion.state.in_temp = 19
await controller._process_heater(
resolution
)
assert tion.controller.calls == [
("heater_on", None)
]
assert (
controller._temperature
== 19
)
assert (
controller._temperature_source
== "tion"
)
assert (
controller._auto_heater
is True
)
async def test_no_temperature():
controller, qingping, tion = (
make_controller()
)
resolution = make_resolution()
qingping.online = False
tion.online = False
await controller._process_heater(
resolution
)
assert tion.controller.calls == [
("heater_off", None)
]
assert (
controller._temperature
is None
)
assert (
controller._temperature_source
is None
)
assert (
controller._auto_heater
is False
)
async def test_target_temperature_missing():
controller, qingping, tion = (
make_controller()
)
resolution = make_resolution(
target_temp=None
)
await controller._process_heater(
resolution
)
assert tion.controller.calls == [
("heater_off", None)
]
assert (
controller._auto_heater
is False
)
async def main():
await test_qingping_heating()
await test_tion_temperature_fallback()
await test_no_temperature()
await test_target_temperature_missing()
print()
print("=" * 70)
print(
"ALL AUTO HEATER PROCESS "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
tests/test_auto_heater_target_temperature.py
+138
View File
@@ -0,0 +1,138 @@
import asyncio
from types import SimpleNamespace
from app.auto.config import (
TemperatureConfig,
)
from app.auto.controller import (
AutoController,
)
from app.auto.temperature_policy import (
TemperaturePolicy,
)
class FakeQingpingService:
def __init__(self):
self.online = True
self.state = SimpleNamespace(
temperature=19.5,
)
class FakeTionController:
def __init__(self):
self.calls = []
async def heater_on(self):
self.calls.append(
("heater_on", None)
)
async def heater_off(self):
self.calls.append(
("heater_off", None)
)
class FakeTionService:
def __init__(self):
self.online = True
self.state = SimpleNamespace(
in_temp=19,
)
self.controller = (
FakeTionController()
)
async def execute(
self,
operation,
):
return await operation(
self.controller
)
async def test_auto_uses_auto_target_temperature():
qingping = FakeQingpingService()
tion = FakeTionService()
policy = TemperaturePolicy(
TemperatureConfig(
hysteresis=0.5,
)
)
controller = AutoController(
schedule_service=None,
qingping_service=qingping,
tion_service=tion,
temperature_policy=policy,
)
resolution = SimpleNamespace(
# Последний SET хотел 23°C.
scheduled_settings=SimpleNamespace(
target_temp=23,
),
# Но текущий AUTO явно хочет 20°C.
auto_target_temp=20,
)
await controller._process_heater(
resolution
)
# При 19.5°C и target=20°C:
#
# heater был OFF,
# нижняя граница = 19.5°C.
#
# Поэтому heater должен остаться OFF.
#
# Если бы контроллер ошибочно использовал
# SET target_temp=23, он бы включил heater.
assert tion.controller.calls == [
("heater_off", None)
]
assert (
controller._auto_heater
is False
)
assert (
controller._temperature
== 19.5
)
assert (
controller._temperature_source
== "qingping"
)
async def main():
await test_auto_uses_auto_target_temperature()
print()
print("=" * 70)
print(
"ALL AUTO HEATER TARGET "
"TEMPERATURE TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
tests/test_auto_heater_tion_target.py
+245
View File
@@ -0,0 +1,245 @@
import asyncio
from types import SimpleNamespace
from app.auto.config import TemperatureConfig
from app.auto.controller import AutoController
from app.auto.temperature_policy import TemperaturePolicy
from app.my_dataclasses import MAX_TARGET_TEMP
class FakeQingpingService:
def __init__(self):
self.online = True
self.state = SimpleNamespace(
temperature=19.0,
)
class FakeTionController:
def __init__(self):
self.calls = []
async def set_target_temperature(
self,
temperature: int,
):
self.calls.append(
(
"set_target_temperature",
temperature,
)
)
async def heater_on(self):
self.calls.append(
(
"heater_on",
None,
)
)
async def heater_off(self):
self.calls.append(
(
"heater_off",
None,
)
)
class FakeTionService:
def __init__(self):
self.online = True
self.state = SimpleNamespace(
in_temp=22,
)
self.controller = FakeTionController()
async def execute(
self,
operation,
):
return await operation(
self.controller
)
def make_controller():
qingping = FakeQingpingService()
tion = FakeTionService()
temperature_policy = TemperaturePolicy(
TemperatureConfig(
hysteresis=0.5,
)
)
controller = AutoController(
schedule_service=None,
qingping_service=qingping,
tion_service=tion,
temperature_policy=temperature_policy,
)
return (
controller,
qingping,
tion,
)
def make_resolution(
target_temp: int,
):
return SimpleNamespace(
auto_target_temp=target_temp,
)
async def test_heating_sets_tion_target_first():
controller, qingping, tion = (
make_controller()
)
resolution = make_resolution(
target_temp=20
)
# Qingping = 19.0
# AUTO target = 20
#
# Нужно греть.
#
# Сначала Tion.target_temp поднимается
# до технического максимума,
# затем включается heater.
await controller._process_heater(
resolution
)
assert tion.controller.calls == [
(
"set_target_temperature",
MAX_TARGET_TEMP,
),
(
"heater_on",
None,
),
]
assert (
controller._target_temperature
== MAX_TARGET_TEMP
)
assert (
controller._target_heater
is True
)
assert (
controller._auto_heater
is True
)
async def test_repeated_heating_does_not_repeat_commands():
controller, qingping, tion = (
make_controller()
)
resolution = make_resolution(
target_temp=20
)
await controller._process_heater(
resolution
)
tion.controller.calls.clear()
# Температура всё ещё ниже цели.
# AUTO продолжает хотеть нагрев,
# но повторно слать команды Tion не нужно.
qingping.state.temperature = 19.2
await controller._process_heater(
resolution
)
assert tion.controller.calls == []
async def test_target_reached_turns_heater_off():
controller, qingping, tion = (
make_controller()
)
resolution = make_resolution(
target_temp=20
)
# Сначала включаем нагрев.
await controller._process_heater(
resolution
)
tion.controller.calls.clear()
# Цель достигнута.
qingping.state.temperature = 20.0
await controller._process_heater(
resolution
)
# MAX_TARGET_TEMP обратно сейчас не меняем.
# Достаточно запретить нагрев.
assert tion.controller.calls == [
(
"heater_off",
None,
)
]
assert (
controller._auto_heater
is False
)
assert (
controller._target_heater
is False
)
async def main():
await test_heating_sets_tion_target_first()
await test_repeated_heating_does_not_repeat_commands()
await test_target_reached_turns_heater_off()
print()
print("=" * 70)
print(
"ALL AUTO HEATER TION TARGET "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
tests/test_auto_temperature_config.py
+359
View File
@@ -0,0 +1,359 @@
from pathlib import Path
from tempfile import TemporaryDirectory
from app.auto.config import (
load_auto_config,
)
BASE_CONFIG = """
version: 1
check_interval: 5.0
co2:
base_speed: 1
hysteresis: 100
thresholds:
- ppm: 800
speed: 2
temperature:
hysteresis: 0.5
"""
def load_config(text: str):
temp_dir = TemporaryDirectory()
path = (
Path(temp_dir.name)
/ "auto.yaml"
)
path.write_text(
text,
encoding="utf-8",
)
config = load_auto_config(path)
return temp_dir, config
def test_valid_temperature_config():
temp_dir, config = load_config(
BASE_CONFIG
)
try:
print()
print("=" * 70)
print("VALID TEMPERATURE CONFIG")
print("=" * 70)
print(
"hysteresis:",
config.temperature.hysteresis,
)
assert (
config.temperature.hysteresis
== 0.5
)
finally:
temp_dir.cleanup()
def test_temperature_config_required():
config_text = """
version: 1
check_interval: 5.0
co2:
base_speed: 1
hysteresis: 100
thresholds:
- ppm: 800
speed: 2
"""
try:
temp_dir, _ = load_config(
config_text
)
except ValueError as exc:
print()
print("=" * 70)
print("MISSING TEMPERATURE CONFIG")
print("=" * 70)
print(exc)
assert (
str(exc)
== "temperature config is required"
)
return
else:
temp_dir.cleanup()
raise AssertionError(
"Missing temperature config "
"must raise ValueError"
)
def test_temperature_must_be_mapping():
config_text = """
version: 1
check_interval: 5.0
co2:
base_speed: 1
hysteresis: 100
thresholds:
- ppm: 800
speed: 2
temperature: 0.5
"""
try:
temp_dir, _ = load_config(
config_text
)
except ValueError as exc:
print()
print("=" * 70)
print("TEMPERATURE MUST BE MAPPING")
print("=" * 70)
print(exc)
assert (
str(exc)
== "temperature must be an object"
)
return
else:
temp_dir.cleanup()
raise AssertionError(
"Invalid temperature mapping "
"must raise ValueError"
)
def test_temperature_hysteresis_must_be_number():
config_text = """
version: 1
check_interval: 5.0
co2:
base_speed: 1
hysteresis: 100
thresholds:
- ppm: 800
speed: 2
temperature:
hysteresis: abc
"""
try:
temp_dir, _ = load_config(
config_text
)
except ValueError as exc:
print()
print("=" * 70)
print(
"HYSTERESIS MUST BE NUMBER"
)
print("=" * 70)
print(exc)
assert (
str(exc)
== (
"temperature.hysteresis "
"must be a number"
)
)
return
else:
temp_dir.cleanup()
raise AssertionError(
"Non-numeric hysteresis "
"must raise ValueError"
)
def test_temperature_hysteresis_must_not_be_negative():
config_text = """
version: 1
check_interval: 5.0
co2:
base_speed: 1
hysteresis: 100
thresholds:
- ppm: 800
speed: 2
temperature:
hysteresis: -0.1
"""
try:
temp_dir, _ = load_config(
config_text
)
except ValueError as exc:
print()
print("=" * 70)
print(
"HYSTERESIS MUST NOT BE NEGATIVE"
)
print("=" * 70)
print(exc)
assert (
str(exc)
== (
"temperature.hysteresis "
"must be >= 0"
)
)
return
else:
temp_dir.cleanup()
raise AssertionError(
"Negative hysteresis "
"must raise ValueError"
)
def test_unknown_temperature_field():
config_text = """
version: 1
check_interval: 5.0
co2:
base_speed: 1
hysteresis: 100
thresholds:
- ppm: 800
speed: 2
temperature:
hysteresis: 0.5
something_else: 123
"""
try:
temp_dir, _ = load_config(
config_text
)
except ValueError as exc:
print()
print("=" * 70)
print(
"UNKNOWN TEMPERATURE FIELD"
)
print("=" * 70)
print(exc)
assert (
str(exc)
== (
"Unknown temperature config fields: "
"['something_else']"
)
)
return
else:
temp_dir.cleanup()
raise AssertionError(
"Unknown temperature field "
"must raise ValueError"
)
def test_temperature_is_allowed_top_level_field():
temp_dir, config = load_config(
BASE_CONFIG
)
try:
assert (
config.temperature.hysteresis
== 0.5
)
finally:
temp_dir.cleanup()
def main():
test_valid_temperature_config()
test_temperature_config_required()
test_temperature_must_be_mapping()
test_temperature_hysteresis_must_be_number()
test_temperature_hysteresis_must_not_be_negative()
test_unknown_temperature_field()
test_temperature_is_allowed_top_level_field()
print()
print("=" * 70)
print(
"ALL AUTO TEMPERATURE CONFIG "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
main()
tests/test_auto_temperature_policy.py
+116
View File
@@ -0,0 +1,116 @@
from app.auto.config import (
TemperatureConfig,
)
from app.auto.temperature_policy import (
TemperaturePolicy,
)
def make_policy() -> TemperaturePolicy:
return TemperaturePolicy(
TemperatureConfig(
hysteresis=0.5,
)
)
def test_heater_off_below_lower_limit():
policy = make_policy()
result = policy.heater_required(
temperature=19.4,
target_temp=20.0,
heater_on=False,
)
assert result is True
def test_heater_off_at_lower_limit():
policy = make_policy()
result = policy.heater_required(
temperature=19.5,
target_temp=20.0,
heater_on=False,
)
assert result is False
def test_heater_off_inside_hysteresis():
policy = make_policy()
result = policy.heater_required(
temperature=19.8,
target_temp=20.0,
heater_on=False,
)
assert result is False
def test_heater_on_below_target():
policy = make_policy()
result = policy.heater_required(
temperature=19.8,
target_temp=20.0,
heater_on=True,
)
assert result is True
def test_heater_on_at_target():
policy = make_policy()
result = policy.heater_required(
temperature=20.0,
target_temp=20.0,
heater_on=True,
)
assert result is False
def test_heater_on_above_target():
policy = make_policy()
result = policy.heater_required(
temperature=20.2,
target_temp=20.0,
heater_on=True,
)
assert result is False
def main():
test_heater_off_below_lower_limit()
test_heater_off_at_lower_limit()
test_heater_off_inside_hysteresis()
test_heater_on_below_target()
test_heater_on_at_target()
test_heater_on_above_target()
print()
print("=" * 70)
print(
"ALL AUTO TEMPERATURE POLICY "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
main()
tests/test_auto_temperature_source.py
+169
View File
@@ -0,0 +1,169 @@
from types import SimpleNamespace
from app.auto.controller import AutoController
class FakeQingpingService:
def __init__(
self,
online: bool,
temperature,
):
self.online = online
self.state = SimpleNamespace(
temperature=temperature,
)
class FakeTionService:
def __init__(
self,
online: bool,
in_temp,
):
self.online = online
self.state = SimpleNamespace(
in_temp=in_temp,
)
def make_controller(
*,
qingping_online: bool,
qingping_temperature,
tion_online: bool,
tion_temperature,
):
qingping = FakeQingpingService(
online=qingping_online,
temperature=qingping_temperature,
)
tion = FakeTionService(
online=tion_online,
in_temp=tion_temperature,
)
return AutoController(
schedule_service=None,
qingping_service=qingping,
tion_service=tion,
)
def test_qingping_priority():
controller = make_controller(
qingping_online=True,
qingping_temperature=19.7,
tion_online=True,
tion_temperature=18,
)
temperature, source = (
controller._get_temperature()
)
assert temperature == 19.7
assert source == "qingping"
def test_tion_fallback():
controller = make_controller(
qingping_online=False,
qingping_temperature=19.7,
tion_online=True,
tion_temperature=18,
)
temperature, source = (
controller._get_temperature()
)
assert temperature == 18
assert source == "tion"
def test_tion_fallback_when_qingping_temperature_missing():
controller = make_controller(
qingping_online=True,
qingping_temperature=None,
tion_online=True,
tion_temperature=18,
)
temperature, source = (
controller._get_temperature()
)
assert temperature == 18
assert source == "tion"
def test_offline_tion_is_not_used():
controller = make_controller(
qingping_online=False,
qingping_temperature=None,
tion_online=False,
# Значение специально оставляем.
# Оно имитирует старый state Tion.
tion_temperature=18,
)
temperature, source = (
controller._get_temperature()
)
assert temperature is None
assert source is None
def test_missing_tion_temperature():
controller = make_controller(
qingping_online=False,
qingping_temperature=None,
tion_online=True,
tion_temperature=None,
)
temperature, source = (
controller._get_temperature()
)
assert temperature is None
assert source is None
def main():
test_qingping_priority()
test_tion_fallback()
test_tion_fallback_when_qingping_temperature_missing()
test_offline_tion_is_not_used()
test_missing_tion_temperature()
print()
print("=" * 70)
print(
"ALL AUTO TEMPERATURE SOURCE "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
main()
tests/test_schedule_auto_heater.py
+327
View File
@@ -0,0 +1,327 @@
import asyncio
from datetime import datetime
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch
from schedule import (
ScheduleService,
load_schedule,
)
SCHEDULE_YAML = """
version: 1
enabled: true
timezone: local
templates:
test:
- time: "00:00"
action:
type: set
power: off
- time: "09:00"
action:
type: set
power: on
speed: 1
heater: on
target_temp: 20
- time: "10:00"
action:
type: auto
speed: 3
- time: "12:00"
action:
type: set
speed: 2
heater: off
days:
mon: test
tue: test
wed: test
thu: test
fri: test
sat: test
sun: test
"""
class FakeTionController:
def __init__(self):
self.calls = []
async def power_on(self):
self.calls.append(
("power_on", None)
)
async def power_off(self):
self.calls.append(
("power_off", None)
)
async def set_speed(self, speed: int):
self.calls.append(
("set_speed", speed)
)
# Оставляем оба варианта управления heater,
# чтобы fake соответствовал фактическому
# интерфейсу TionController.
async def heater_on(self):
self.calls.append(
("heater_on", None)
)
async def heater_off(self):
self.calls.append(
("heater_off", None)
)
async def set_heater(self, enabled: bool):
self.calls.append(
("set_heater", enabled)
)
async def set_target_temperature(
self,
temperature: int,
):
self.calls.append(
(
"set_target_temperature",
temperature,
)
)
class FakeTionService:
def __init__(self):
self.controller = FakeTionController()
async def execute(self, operation):
return await operation(
self.controller
)
def load_test_schedule():
temp_dir = TemporaryDirectory()
path = (
Path(temp_dir.name)
/ "schedule.yaml"
)
path.write_text(
SCHEDULE_YAML,
encoding="utf-8",
)
config = load_schedule(path)
return temp_dir, config
def heater_calls(calls):
return [
call
for call in calls
if call[0] in {
"heater_on",
"heater_off",
"set_heater",
}
]
async def process_at(
service: ScheduleService,
when: datetime,
):
with patch(
"schedule.service.datetime",
wraps=datetime,
) as mocked_datetime:
mocked_datetime.now.return_value = when
await service._process()
async def test_auto_does_not_control_heater():
temp_dir, config = (
load_test_schedule()
)
try:
tion = FakeTionService()
service = ScheduleService(
config,
tion=tion,
)
# --------------------------------------------------
# 09:30
#
# Обычный SET.
# heater=on должен принадлежать ScheduleService.
# --------------------------------------------------
await process_at(
service,
datetime(
2026,
9,
14,
9,
30,
),
)
print()
print(
"09:30 SET calls:",
tion.controller.calls,
)
calls = heater_calls(
tion.controller.calls
)
assert calls != [], (
"SET must control heater"
)
tion.controller.calls.clear()
# --------------------------------------------------
# 10:30
#
# AUTO.
#
# В накопленных scheduled_settings heater всё ещё
# должен быть True, потому что последняя SET-точка
# включила heater.
#
# Но ScheduleService НЕ должен отправлять
# никаких heater-команд.
# --------------------------------------------------
resolution = service.resolve(
datetime(
2026,
9,
14,
10,
30,
)
)
print()
print(
"10:30 scheduled heater:",
resolution.scheduled_settings.heater,
)
assert (
resolution.auto_active
is True
)
assert (
resolution.scheduled_settings.heater
is True
)
await process_at(
service,
datetime(
2026,
9,
14,
10,
30,
),
)
print(
"10:30 AUTO calls:",
tion.controller.calls,
)
calls = heater_calls(
tion.controller.calls
)
assert calls == [], (
"ScheduleService must not control "
"heater during AUTO"
)
tion.controller.calls.clear()
# --------------------------------------------------
# 12:30
#
# AUTO закончился.
# heater=off снова принадлежит ScheduleService.
# --------------------------------------------------
await process_at(
service,
datetime(
2026,
9,
14,
12,
30,
),
)
print()
print(
"12:30 SET calls:",
tion.controller.calls,
)
calls = heater_calls(
tion.controller.calls
)
assert calls != [], (
"ScheduleService must regain heater "
"control after AUTO"
)
finally:
temp_dir.cleanup()
async def main():
await test_auto_does_not_control_heater()
print()
print("=" * 70)
print(
"ALL AUTO HEATER OWNERSHIP TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
tests/test_schedule_auto_temperature_parser.py
+217
View File
@@ -0,0 +1,217 @@
from pathlib import Path
from tempfile import TemporaryDirectory
from schedule import (
ScheduleActionType,
load_schedule,
)
VALID_YAML = """
version: 1
enabled: true
timezone: local
templates:
test:
- time: "00:00"
action:
type: auto
speed: 3
target_temp: 20
days:
mon: test
tue: test
wed: test
thu: test
fri: test
sat: test
sun: test
"""
def load_config(text: str):
temp_dir = TemporaryDirectory()
path = (
Path(temp_dir.name)
/ "schedule.yaml"
)
path.write_text(
text,
encoding="utf-8",
)
config = load_schedule(path)
return temp_dir, config
def test_auto_target_temperature():
temp_dir, config = load_config(
VALID_YAML
)
try:
point = (
config.templates["test"][0]
)
action = point.action
assert (
action.type
== ScheduleActionType.AUTO
)
assert (
action.settings.speed
== 3
)
assert (
action.settings.target_temp
== 20
)
finally:
temp_dir.cleanup()
def test_auto_rejects_heater():
config_text = """
version: 1
enabled: true
timezone: local
templates:
test:
- time: "00:00"
action:
type: auto
speed: 3
target_temp: 20
heater: on
days:
mon: test
tue: test
wed: test
thu: test
fri: test
sat: test
sun: test
"""
try:
temp_dir, _ = load_config(
config_text
)
except ValueError as exc:
print()
print(
"Expected error:",
exc,
)
assert "heater" in str(exc)
return
else:
temp_dir.cleanup()
raise AssertionError(
"AUTO heater field must "
"raise ValueError"
)
def test_auto_speed_is_required():
config_text = """
version: 1
enabled: true
timezone: local
templates:
test:
- time: "00:00"
action:
type: auto
target_temp: 20
days:
mon: test
tue: test
wed: test
thu: test
fri: test
sat: test
sun: test
"""
try:
temp_dir, _ = load_config(
config_text
)
except ValueError as exc:
print()
print(
"Expected error:",
exc,
)
assert (
str(exc)
== "AUTO action requires 'speed'"
)
return
else:
temp_dir.cleanup()
raise AssertionError(
"AUTO without speed must "
"raise ValueError"
)
def main():
test_auto_target_temperature()
test_auto_rejects_heater()
test_auto_speed_is_required()
print()
print("=" * 70)
print(
"ALL AUTO TEMPERATURE PARSER "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
main()
tests/test_schedule_auto_temperature_resolution.py
+196
View File
@@ -0,0 +1,196 @@
from datetime import datetime
from pathlib import Path
from tempfile import TemporaryDirectory
from schedule import (
ScheduleService,
load_schedule,
)
SCHEDULE_YAML = """
version: 1
enabled: true
timezone: local
templates:
test:
- time: "09:00"
action:
type: set
power: on
speed: 1
target_temp: 23
- time: "10:00"
action:
type: auto
speed: 3
target_temp: 20
- time: "12:00"
action:
type: set
speed: 2
target_temp: 22
days:
mon: test
tue: test
wed: test
thu: test
fri: test
sat: test
sun: test
"""
def load_test_schedule():
temp_dir = TemporaryDirectory()
path = (
Path(temp_dir.name)
/ "schedule.yaml"
)
path.write_text(
SCHEDULE_YAML,
encoding="utf-8",
)
config = load_schedule(path)
return temp_dir, config
def test_auto_temperature_resolution():
temp_dir, config = (
load_test_schedule()
)
try:
service = ScheduleService(
config
)
# ----------------------------------------------
# 09:30 — обычный SET
# ----------------------------------------------
resolution = service.resolve(
datetime(
2026,
9,
14,
9,
30,
)
)
assert (
resolution.auto_active
is False
)
assert (
resolution.scheduled_settings.target_temp
== 23
)
assert (
resolution.auto_target_temp
is None
)
# ----------------------------------------------
# 10:30 — AUTO
#
# Последний SET по-прежнему хранит 23,
# но AUTO явно требует 20.
# ----------------------------------------------
resolution = service.resolve(
datetime(
2026,
9,
14,
10,
30,
)
)
assert (
resolution.auto_active
is True
)
assert (
resolution.auto_fallback_speed
== 3
)
assert (
resolution.scheduled_settings.target_temp
== 23
)
assert (
resolution.auto_target_temp
== 20
)
# ----------------------------------------------
# 12:30 — снова SET
#
# AUTO закончился.
# ----------------------------------------------
resolution = service.resolve(
datetime(
2026,
9,
14,
12,
30,
)
)
assert (
resolution.auto_active
is False
)
assert (
resolution.scheduled_settings.target_temp
== 22
)
assert (
resolution.auto_target_temp
is None
)
finally:
temp_dir.cleanup()
def main():
test_auto_temperature_resolution()
print()
print("=" * 70)
print(
"ALL AUTO TEMPERATURE RESOLUTION "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
main()