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-14 21:24:22 +03:00
parent 4c665d9108
commit dec14b1d12
5 changed files with 357 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/schedule.yaml
+28 -18
View File
@@ -36,23 +36,47 @@ templates:
type: set
speed: 3
- time: "17:30"
- time: "19:00"
action:
type: set
power: on
speed: 6
heater: on
target_temp: 20
- time: "20:30"
action:
type: auto
- time: "20:31"
action:
type: set
speed: 3
- time: "23:00"
action:
type: set
speed: 1
heater: off
- time: "23:58"
action:
type: set
speed: 3
weekend:
- time: "00:00"
action:
type: set
power: off
speed: 6
- time: "00:02"
action:
type: set
speed: 1
- time: "09:00"
action:
@@ -67,22 +91,6 @@ templates:
action:
type: auto
- time: "22:54"
action:
type: set
power: on
speed: 2
- time: "22:55"
action:
type: set
speed: 4
- time: "22:56"
action:
type: set
speed: 6
- time: "23:30"
action:
type: set
@@ -90,6 +98,8 @@ templates:
heater: off
days:
mon: workday
schedule/models.py
+1 -4
View File
@@ -19,10 +19,7 @@ class ScheduledSettings:
sound: bool | None = None
light: bool | None = None
def merged(
self,
newer: "ScheduledSettings",
) -> "ScheduledSettings":
def merged(self, newer: "ScheduledSettings") -> "ScheduledSettings":
"""
Наложить более новые настройки поверх старых.
schedule/service.py
+202 -38
View File
@@ -42,12 +42,123 @@ class ScheduleService:
self._running = False
self._task = None
self._last_applied_when = None
# Последняя успешно применённая точка расписания.
self._last_applied_when: datetime | None = None
# Temporary override.
self._override_settings = ScheduledSettings()
self._override_until: datetime | None = None
# True означает:
# override записан, но применить его к Tion
# пока не удалось.
self._override_pending = False
# Ошибка именно ScheduleService.
self._last_error: str | None = None
@property
def running(self) -> bool:
return self._running
@property
def last_error(self) -> str | None:
return self._last_error
@property
def override_settings(self) -> ScheduledSettings:
return self._override_settings
@property
def override_until(self) -> datetime | None:
return self._override_until
@property
def override_active(self) -> bool:
if self._override_until is None:
return False
return datetime.now() < self._override_until
@property
def config(self) -> ScheduleConfig:
return self._config
async def apply_override(self, settings: ScheduledSettings) -> None:
if settings.empty:
raise ValueError(
"Override settings cannot be empty"
)
if self._tion is None:
raise RuntimeError(
"TionService is required"
)
now = datetime.now()
resolution = self.resolve(now)
if not resolution.enabled:
raise RuntimeError(
"Schedule is disabled"
)
if resolution.current is None:
raise RuntimeError(
"Current schedule point not found"
)
if resolution.next is None:
raise RuntimeError(
"Next schedule point not found"
)
# Если старый override уже закончился,
# перед созданием нового очищаем его.
if ( self._override_until is not None
and now >= self._override_until
):
self._clear_override()
# Добавляем новые ручные параметры
# к уже существующему override.
self._override_settings = ( self._override_settings.merged(settings) )
# Override действует строго до
# следующей точки расписания.
self._override_until = resolution.next.when
# Считаем его неприменённым,
# пока команда реально не прошла.
self._override_pending = True
try:
await self._apply_settings(settings)
except Exception as exc:
self._last_error = (
f"{type(exc).__name__}: {exc}"
)
# Override остаётся сохранённым.
# Фоновый цикл попробует ещё раз.
raise
else:
self._override_pending = False
self._last_error = None
def _clear_override(self) -> None:
self._override_settings = ScheduledSettings()
self._override_until = None
self._override_pending = False
def resolve(
self,
now: datetime | None = None,
@@ -216,21 +327,24 @@ class ScheduleService:
return
if self._tion is None:
raise RuntimeError(
"TionService is required to start schedule"
)
raise RuntimeError("TionService is required to start schedule")
self._running = True
# При запуске сразу приводим Tion
# к текущему состоянию расписания.
await self._process(initial=True)
try:
await self._process()
except Exception as exc:
self._last_error = f"{type(exc).__name__}: {exc}"
self._task = asyncio.create_task(self._loop())
self._task = asyncio.create_task(
self._loop()
)
async def stop(self) -> None:
if not self._running:
return
self._running = False
if self._task is not None:
@@ -243,21 +357,57 @@ class ScheduleService:
self._task = None
async def _loop(self) -> None:
while self._running:
await asyncio.sleep(
self._check_interval
)
await self._process()
try:
await self._process()
async def _process(
self,
*,
initial: bool = False,
) -> None:
except Exception as exc:
self._last_error = f"{type(exc).__name__}: {exc}"
resolution = self.resolve()
# 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()
resolution = self.resolve(now)
if not resolution.enabled:
return
@@ -267,39 +417,53 @@ class ScheduleService:
if current is None:
return
# Эта точка уже применена.
# --------------------------------------------------
# 1. Проверяем окончание temporary override
# --------------------------------------------------
if (
not initial
and self._last_applied_when == current.when
self._override_until is not None
and now >= self._override_until
):
self._clear_override()
# --------------------------------------------------
# 2. Пока override действует,
# расписание Tion не трогает
# --------------------------------------------------
if self.override_active:
# Если override ранее не удалось применить,
# пробуем снова.
if self._override_pending:
await self._apply_settings(self._override_settings)
self._override_pending = False
self._last_error = None
return
# AUTO пока только распознаём.
# Сам AutoController подключим позже.
if (
current.point.action.type
== ScheduleActionType.AUTO
):
self._last_applied_when = current.when
# --------------------------------------------------
# 3. Если текущая точка уже успешно применена,
# ничего не делаем
# --------------------------------------------------
if self._last_applied_when == current.when:
return
if initial:
# После запуска восстанавливаем всё
# накопленное состояние расписания.
settings = resolution.scheduled_settings
else:
# При обычном переходе применяем
# только параметры новой точки.
settings = current.point.action.settings
# --------------------------------------------------
# 4. Приводим Tion к ПОЛНОМУ актуальному
# состоянию расписания
# --------------------------------------------------
await self._apply_settings(settings)
await self._apply_settings(resolution.scheduled_settings)
# Ставим только ПОСЛЕ успешного применения.
self._last_applied_when = current.when
self._last_error = None
async def _apply_settings(
self,
settings: ScheduledSettings,
) -> None:
async def _apply_settings(self, settings: ScheduledSettings) -> None:
if self._tion is None:
return
tests/test_schedule_live.py
+7 -1
View File
@@ -1,6 +1,7 @@
import asyncio
import logging
from pathlib import Path
import datetime
from app.my_dataclasses import TION_MAC
from app.tion import TionController, TionService
@@ -52,11 +53,16 @@ async def main():
if tion.state:
print(
f"{datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")} "
f"power={tion.state.power} "
f"speed={tion.state.fan_speed} "
f"heater={tion.state.heater}"
f"heater={tion.state.heater} "
f"target_temp={tion.state.target_temp} "
f"last_error={tion.last_error} "
)
finally:
await schedule.stop()
await tion.stop()
tests/test_schedule_override.py
+119
View File
@@ -0,0 +1,119 @@
import asyncio
import logging
from datetime import datetime
from pathlib import Path
from app.my_dataclasses import TION_MAC
from app.tion import TionController, TionService
from schedule import (
ScheduleService,
ScheduledSettings,
load_schedule,
)
logging.disable(logging.CRITICAL)
PROJECT_ROOT = Path(__file__).resolve().parents[1]
SCHEDULE_FILE = (
PROJECT_ROOT
/ "config"
/ "schedule.yaml"
)
async def main():
config = load_schedule(SCHEDULE_FILE)
controller = TionController(TION_MAC)
tion = TionService(
controller,
poll_interval=5,
)
schedule = ScheduleService(
config,
tion,
check_interval=5,
)
await tion.start()
await schedule.start()
try:
print()
print("Schedule started")
resolution = schedule.resolve()
print(
f"Current: "
f"{resolution.current.when if resolution.current else None}"
)
print(
f"Next: "
f"{resolution.next.when if resolution.next else None}"
)
# print()
# print("Applying temporary override:")
# print("speed=6")
# print()
#
# await schedule.apply_override(
# ScheduledSettings(
# speed=6,
# )
# )
while True:
await asyncio.sleep(5)
if ( datetime.now() > datetime(2026, 9, 18, 20, 30, 30)
and datetime.now() < datetime(2026, 9, 18, 20, 30, 36) ):
print("Applying temporary override:")
print("heater=True")
print("target_temp=25")
await schedule.apply_override(
ScheduledSettings(
heater=True,
target_temp=25,
)
)
now = datetime.now().strftime(
"%y-%m-%d %H:%M:%S"
)
state = tion.state
if state is None:
print(
f"{now} state=None"
)
continue
print(
f"{now} "
f"power={state.power} "
f"speed={state.fan_speed} "
f"heater={state.heater} "
f"target_temp={state.target_temp} "
f"override={schedule.override_active} "
f"until={schedule.override_until}"
)
finally:
await schedule.stop()
await tion.stop()
if __name__ == "__main__":
asyncio.run(main())