119 lines
2.5 KiB
Python
119 lines
2.5 KiB
Python
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()) |