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

feat: v.1.1.0 work: добавил режим ручного управления. Теперь можно отключать расписание!

Browse Source 
POST api/schedule/pause
POST api/schedule/resume
This commit is contained in:
Fedorov Dmitriy
2026-09-19 19:05:55 +03:00
parent b02d18d283
commit a5c9d07db5
7 changed files with 784 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api.py
+73 -18
View File
@@ -4,6 +4,8 @@ from typing import Literal, Annotated
from fastapi import FastAPI, Request, HTTPException, Path
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pathlib import Path as FilePath
from app.qingping.service import QingpingService
from app.my_dataclasses import (
TION_MAC,
@@ -39,7 +41,12 @@ from schedule import (
load_schedule,
)
from app.qingping.service import QingpingService
SCHEDULE_STATE_FILE = (
FilePath(SCHEDULE_FILE)
.with_name("schedule_state.json")
)
def configure_logging() -> None:
noisy_loggers = (
@@ -104,6 +111,7 @@ async def lifespan(app: FastAPI):
config,
service,
check_interval=5,
state_path=SCHEDULE_STATE_FILE,
)
await schedule_service.start()
@@ -277,21 +285,27 @@ def get_effective_temperature() -> int:
async def execute_command(operation, override: ScheduledSettings | None = None):
try:
# Если расписание реально работает,
# ручная команда становится
# temporary override.
# При активном расписании ручная команда
# становится temporary override.
#
# Но если расписание paused,
# мы находимся в полноценном ручном режиме,
# поэтому команда идёт напрямую в Tion.
if (
override is not None
and schedule_service is not None
and schedule_service.running
and schedule_service.config.enabled
override is not None
and schedule_service is not None
and schedule_service.running
and schedule_service.config.enabled
and not schedule_service.paused
):
await schedule_service.apply_override(override)
await schedule_service.apply_override(
override
)
# Если расписание выключено или
# недоступно — обычное ручное управление.
else:
await service.execute(operation)
await service.execute(
operation
)
except Exception as exc:
raise HTTPException(
@@ -331,18 +345,19 @@ def get_schedule_status() -> dict:
if schedule_service is None:
return {
"available": False,
"enabled": False,
"running": False,
"current_action": None,
"current": None,
"next": None,
"auto_active": False,
"auto_fallback_speed": None,
"auto_target_temp": None,
"available": False,
"current_action": None,
"current": None,
"enabled": False,
"next": None,
"override_active": False,
"override_until": None,
"override_settings": {},
"paused": False,
"running": False,
"scheduled_settings": {},
"last_error": schedule_load_error,
}
@@ -385,6 +400,8 @@ def get_schedule_status() -> dict:
"enabled": resolution.enabled,
"running": schedule_service.running,
"paused": schedule_service.paused,
"current_action": current_action,
"current_time": current_time,
@@ -721,4 +738,42 @@ async def clear_schedule_override():
"ok": True,
"schedule": get_schedule_status(),
"auto": get_auto_status(),
}
# ----------------------------------------------------------------------
# Переход в полностью manual
# ----------------------------------------------------------------------
@app.post("/api/schedule/pause")
async def pause_schedule():
if schedule_service is None:
raise HTTPException(
status_code=503,
detail="Schedule service unavailable",
)
await schedule_service.pause()
return {
"ok": True,
"schedule": get_schedule_status(),
}
# ----------------------------------------------------------------------
# Возвращаем AUTO
# ----------------------------------------------------------------------
@app.post("/api/schedule/resume")
async def resume_schedule():
if schedule_service is None:
raise HTTPException(
status_code=503,
detail="Schedule service unavailable",
)
await schedule_service.resume()
return {
"ok": True,
"schedule": get_schedule_status(),
}
app/auto/controller.py
+34
View File
@@ -88,6 +88,40 @@ class AutoController:
resolution = self._schedule.resolve(now)
# ----------------------------------------------
# Расписание поставлено на паузу.
#
# Это полноценный ручной режим:
# AUTO вообще не имеет права управлять Tion.
# ----------------------------------------------
if getattr(
self._schedule,
"paused",
False,
):
self._state = "suspended"
self._reason = "schedule_paused"
# После перехода в ручной режим пользователь
# может изменить эти параметры напрямую.
# Поэтому старые значения AUTO больше
# нельзя считать достоверными.
self._target_speed = 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
# ----------------------------------------------
# Активен ручной override.
#
config/schedule.yaml
+5 -1
View File
@@ -17,7 +17,6 @@ templates:
heater: on
target_temp: 21
- time: "09:30"
action:
type: auto
@@ -52,6 +51,11 @@ templates:
speed: 2
target_temp: 25
- time: "19:02"
action:
type: set
speed: 6
- time: "22:00"
action:
type: set
schedule/service.py
+109 -1
View File
@@ -1,5 +1,6 @@
import asyncio
import json
from pathlib import Path
from app.tion.service import TionService
from datetime import (
@@ -29,6 +30,7 @@ WEEKDAYS = (
)
class ScheduleService:
def __init__(
@@ -36,6 +38,7 @@ class ScheduleService:
config,
tion: TionService | None = None,
check_interval: float = 5.0,
state_path: str | Path | None = None,
):
self._config = config
self._tion = tion
@@ -59,6 +62,21 @@ class ScheduleService:
# Ошибка именно ScheduleService.
self._last_error: str | None = None
self._paused = False
self._state_path = (
Path(state_path)
if state_path is not None
else None
)
self._load_runtime_state()
@property
def paused(self) -> bool:
return self._paused
@property
def running(self) -> bool:
@@ -87,6 +105,62 @@ class ScheduleService:
def config(self) -> ScheduleConfig:
return self._config
def _load_runtime_state(self) -> None:
if self._state_path is None:
return
if not self._state_path.exists():
return
try:
data = json.loads(
self._state_path.read_text(
encoding="utf-8"
)
)
except (
OSError,
json.JSONDecodeError,
):
return
paused = data.get("paused")
if isinstance(paused, bool):
self._paused = paused
def _save_runtime_state(self) -> None:
if self._state_path is None:
return
data = {
"paused": self._paused,
}
self._state_path.parent.mkdir(
parents=True,
exist_ok=True,
)
temp_path = (
self._state_path.with_suffix(
self._state_path.suffix + ".tmp"
)
)
temp_path.write_text(
json.dumps(
data,
indent=2,
),
encoding="utf-8",
)
temp_path.replace(
self._state_path
)
async def apply_override(self, settings: ScheduledSettings) -> None:
@@ -165,6 +239,36 @@ class ScheduleService:
# Сразу пересчитываем и применяем текущее состояние расписания.
await self._process()
async def pause(self) -> None:
if self._paused:
return
# Сначала запрещаем расписанию вмешиваться.
self._paused = True
# Temporary override больше не имеет смысла:
# мы переходим в полноценный ручной режим.
self._clear_override()
self._last_applied_when = None
self._save_runtime_state()
async def resume(self) -> None:
if not self._paused:
return
self._paused = False
self._save_runtime_state()
# Сразу применяем состояние расписания,
# актуальное именно сейчас.
await self._process()
def resolve(
self,
now: datetime | None = None,
@@ -378,6 +482,10 @@ class ScheduleService:
async def _process(self) -> None:
#Если ручное управление то сразу выходим
if self._paused:
return
now = datetime.now()
resolution = self.resolve(now)
tests/test_auto_schedule_pause.py
+181
View File
@@ -0,0 +1,181 @@
import asyncio
from types import SimpleNamespace
from app.auto.controller import AutoController
class FakeScheduleService:
def __init__(self):
self.paused = True
self.override_active = False
def resolve(self, now):
return SimpleNamespace(
enabled=True,
auto_active=True,
auto_fallback_speed=1,
auto_target_temp=20,
)
class FakeQingpingService:
def __init__(self):
self.online = True
self.state = SimpleNamespace(
co2=1200,
temperature=19.0,
)
class FakeTionController:
def __init__(self):
self.calls = []
async def set_speed(self, speed):
self.calls.append(
("set_speed", speed)
)
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_target_temperature(
self,
temperature,
):
self.calls.append(
(
"set_target_temperature",
temperature,
)
)
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_paused_schedule_suspends_auto():
schedule = FakeScheduleService()
qingping = FakeQingpingService()
tion = FakeTionService()
controller = AutoController(
schedule_service=schedule,
qingping_service=qingping,
tion_service=tion,
)
# Имитируем состояние,
# которое AUTO помнил до pause.
controller._target_speed = 3
controller._auto_speed = 3
controller._target_heater = True
controller._auto_heater = True
controller._target_temperature = 30
controller._temperature = 19.0
controller._temperature_source = (
"qingping"
)
await controller._process()
status = controller.status()
# AUTO должен быть остановлен.
assert (
status["state"]
== "suspended"
)
assert (
status["reason"]
== "schedule_paused"
)
# Никаких команд Tion.
assert tion.controller.calls == []
# Старое владение AUTO забыто.
assert (
controller._target_speed
is None
)
assert (
controller._auto_speed
is None
)
assert (
controller._target_heater
is None
)
assert (
controller._auto_heater
is None
)
assert (
controller._target_temperature
is None
)
assert (
controller._temperature
is None
)
assert (
controller._temperature_source
is None
)
async def main():
await test_paused_schedule_suspends_auto()
print()
print("=" * 70)
print(
"ALL AUTO SCHEDULE PAUSE "
"TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
tests/test_schedule_pause.py
+203
View File
@@ -0,0 +1,203 @@
import asyncio
from datetime import datetime, timedelta
from pathlib import Path
from tempfile import TemporaryDirectory
from schedule import (
ScheduleService,
ScheduledSettings,
load_schedule,
)
SCHEDULE_YAML = """
version: 1
enabled: true
timezone: local
templates:
test:
- time: "00:00"
action:
type: set
speed: 2
days:
mon: test
tue: test
wed: test
thu: test
fri: test
sat: test
sun: test
"""
class FakeTionController:
def __init__(self):
self.calls = []
async def set_speed(
self,
speed: int,
):
self.calls.append(
(
"set_speed",
speed,
)
)
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
async def test_paused_schedule_does_not_control_tion():
temp_dir, config = (
load_test_schedule()
)
try:
tion = FakeTionService()
service = ScheduleService(
config,
tion=tion,
)
assert service.paused is False
await service.pause()
assert service.paused is True
# Даже если фоновый цикл вызывает _process(),
# никаких команд быть не должно.
await service._process()
assert tion.controller.calls == []
# Возвращаем расписание.
#
# resume() должен сразу применить
# текущее состояние расписания.
await service.resume()
assert service.paused is False
assert (
"set_speed",
2,
) in tion.controller.calls
finally:
temp_dir.cleanup()
async def test_pause_clears_override():
temp_dir, config = (
load_test_schedule()
)
try:
service = ScheduleService(
config
)
# Имитируем уже существующий
# temporary override.
service._override_settings = (
ScheduledSettings(
speed=5,
)
)
service._override_until = (
datetime.now()
+ timedelta(hours=1)
)
service._override_pending = True
assert (
service.override_settings.speed
== 5
)
await service.pause()
assert service.paused is True
assert (
service.override_settings.speed
is None
)
assert (
service.override_until
is None
)
assert (
service._override_pending
is False
)
finally:
temp_dir.cleanup()
async def main():
await test_paused_schedule_does_not_control_tion()
await test_pause_clears_override()
print()
print("=" * 70)
print(
"ALL SCHEDULE PAUSE TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())
tests/test_schedule_pause_persistence.py
+179
View File
@@ -0,0 +1,179 @@
import asyncio
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: "00:00"
action:
type: set
speed: 2
days:
mon: test
tue: test
wed: test
thu: test
fri: test
sat: test
sun: test
"""
def create_config(
directory: Path,
):
schedule_path = (
directory
/ "schedule.yaml"
)
schedule_path.write_text(
SCHEDULE_YAML,
encoding="utf-8",
)
return load_schedule(
schedule_path
)
async def test_pause_survives_restart():
with TemporaryDirectory() as temp_dir:
directory = Path(temp_dir)
config = create_config(
directory
)
state_path = (
directory
/ "schedule_state.json"
)
# --------------------------------------
# Первый экземпляр сервиса.
# --------------------------------------
service1 = ScheduleService(
config,
state_path=state_path,
)
assert service1.paused is False
await service1.pause()
assert service1.paused is True
assert state_path.exists()
# --------------------------------------
# Имитируем перезапуск приложения:
# создаём новый ScheduleService.
# --------------------------------------
service2 = ScheduleService(
config,
state_path=state_path,
)
assert service2.paused is True
async def test_resume_survives_restart():
with TemporaryDirectory() as temp_dir:
directory = Path(temp_dir)
config = create_config(
directory
)
state_path = (
directory
/ "schedule_state.json"
)
service1 = ScheduleService(
config,
state_path=state_path,
)
await service1.pause()
assert service1.paused is True
await service1.resume()
assert service1.paused is False
# Новый экземпляр после resume
# тоже должен быть активным.
service2 = ScheduleService(
config,
state_path=state_path,
)
assert service2.paused is False
async def test_missing_state_file_defaults_to_active():
with TemporaryDirectory() as temp_dir:
directory = Path(temp_dir)
config = create_config(
directory
)
state_path = (
directory
/ "missing_state.json"
)
assert not state_path.exists()
service = ScheduleService(
config,
state_path=state_path,
)
assert service.paused is False
async def main():
await test_pause_survives_restart()
await test_resume_survives_restart()
await test_missing_state_file_defaults_to_active()
print()
print("=" * 70)
print(
"ALL SCHEDULE PAUSE "
"PERSISTENCE TESTS PASSED"
)
print("=" * 70)
if __name__ == "__main__":
asyncio.run(main())