445 lines
10 KiB
Python
445 lines
10 KiB
Python
import logging
|
|
from contextlib import asynccontextmanager
|
|
from typing import Literal, Annotated
|
|
from fastapi import FastAPI, Request, HTTPException, Path
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from app.my_dataclasses import (
|
|
TION_MAC,
|
|
MIN_FAN_SPEED,
|
|
MAX_FAN_SPEED,
|
|
MIN_TARGET_TEMP,
|
|
MAX_TARGET_TEMP,
|
|
AIR_MODE_OUTSIDE,
|
|
AIR_MODE_RECIRCULATION,
|
|
SCHEDULE_FILE,
|
|
)
|
|
|
|
from app.tion import (
|
|
TionController,
|
|
TionService,
|
|
)
|
|
|
|
from schedule import (
|
|
ScheduleService,
|
|
ScheduledSettings,
|
|
load_schedule,
|
|
)
|
|
|
|
|
|
|
|
def configure_logging() -> None:
|
|
noisy_loggers = (
|
|
"asyncio",
|
|
"bleak",
|
|
"bleak.backends",
|
|
"bleak.backends.winrt",
|
|
"bleak.backends.winrt.client",
|
|
"tion_btle",
|
|
"tion_btle.tion",
|
|
"tion_btle.s4",
|
|
"tion_btle.light_family",
|
|
)
|
|
|
|
for logger_name in noisy_loggers:
|
|
logging.getLogger(logger_name).setLevel(logging.WARNING)
|
|
|
|
configure_logging()
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Tion
|
|
# ----------------------------------------------------------------------
|
|
|
|
controller = TionController(TION_MAC)
|
|
|
|
service = TionService(
|
|
controller,
|
|
poll_interval=5,
|
|
)
|
|
schedule_service: ScheduleService | None = None
|
|
schedule_load_error: str | None = None
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Application lifecycle
|
|
# ----------------------------------------------------------------------
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
global schedule_service
|
|
global schedule_load_error
|
|
|
|
await service.start()
|
|
|
|
try:
|
|
try:
|
|
config = load_schedule(SCHEDULE_FILE)
|
|
|
|
schedule_service = ScheduleService(
|
|
config,
|
|
service,
|
|
check_interval=5,
|
|
)
|
|
|
|
await schedule_service.start()
|
|
schedule_load_error = None
|
|
|
|
except Exception as exc:
|
|
schedule_service = None
|
|
schedule_load_error = f"{type(exc).__name__}: {exc}"
|
|
yield
|
|
|
|
finally:
|
|
if schedule_service is not None:
|
|
await schedule_service.stop()
|
|
|
|
await service.stop()
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# FastAPI
|
|
# ----------------------------------------------------------------------
|
|
|
|
app = FastAPI(
|
|
title="Tion Controller",
|
|
version="0.1.0",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(
|
|
request: Request,
|
|
exc: RequestValidationError,
|
|
):
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={
|
|
"message": "Invalid request",
|
|
"errors": exc.errors(),
|
|
},
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Helpers
|
|
# ----------------------------------------------------------------------
|
|
|
|
def get_status() -> dict:
|
|
"""
|
|
Получить текущее состояние сервиса.
|
|
|
|
Bluetooth-запрос здесь не выполняется.
|
|
"""
|
|
|
|
state = service.state
|
|
|
|
return {
|
|
"online": service.online,
|
|
"running": service.running,
|
|
|
|
"last_seen": (
|
|
service.last_seen.isoformat()
|
|
if service.last_seen is not None
|
|
else None
|
|
),
|
|
|
|
"last_error": service.last_error,
|
|
|
|
"tion": (
|
|
state.to_dict()
|
|
if state is not None
|
|
else None
|
|
),
|
|
|
|
"schedule": get_schedule_status(),
|
|
}
|
|
|
|
|
|
async def execute_command(operation, override: ScheduledSettings | None = None):
|
|
|
|
try:
|
|
# Если расписание реально работает,
|
|
# ручная команда становится
|
|
# temporary override.
|
|
if (
|
|
override is not None
|
|
and schedule_service is not None
|
|
and schedule_service.running
|
|
and schedule_service.config.enabled
|
|
):
|
|
await schedule_service.apply_override(override)
|
|
|
|
# Если расписание выключено или
|
|
# недоступно — обычное ручное управление.
|
|
else:
|
|
await service.execute(operation)
|
|
|
|
except Exception as exc:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail={
|
|
"message": "Tion unavailable",
|
|
"error": (
|
|
f"{type(exc).__name__}: {exc}"
|
|
),
|
|
},
|
|
) from exc
|
|
|
|
return get_status()
|
|
|
|
|
|
def occurrence_to_dict(occurrence) -> dict | None:
|
|
|
|
if occurrence is None:
|
|
return None
|
|
|
|
return {
|
|
"when": occurrence.when.isoformat(),
|
|
"weekday": occurrence.weekday,
|
|
"template": occurrence.template,
|
|
"action": occurrence.point.action.type.value,
|
|
"settings": (
|
|
occurrence
|
|
.point
|
|
.action
|
|
.settings
|
|
.to_dict()
|
|
),
|
|
}
|
|
|
|
|
|
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,
|
|
"override_active": False,
|
|
"override_until": None,
|
|
"override_settings": {},
|
|
"scheduled_settings": {},
|
|
"last_error": schedule_load_error,
|
|
}
|
|
|
|
resolution = schedule_service.resolve()
|
|
|
|
current = resolution.current
|
|
|
|
current_action = None
|
|
|
|
if current is not None:
|
|
current_action = (current.point.action.type.value)
|
|
|
|
return {
|
|
"available": True,
|
|
"enabled": resolution.enabled,
|
|
"running": schedule_service.running,
|
|
"current_action": current_action,
|
|
|
|
"current": occurrence_to_dict(resolution.current),
|
|
"next": occurrence_to_dict(resolution.next),
|
|
|
|
"auto_active": resolution.auto_active,
|
|
"scheduled_settings": (
|
|
resolution
|
|
.scheduled_settings
|
|
.to_dict()
|
|
),
|
|
|
|
"override_active":schedule_service.override_active,
|
|
|
|
"override_until": (
|
|
schedule_service
|
|
.override_until
|
|
.isoformat()
|
|
if schedule_service.override_until
|
|
else None
|
|
),
|
|
|
|
"override_settings": (
|
|
schedule_service
|
|
.override_settings
|
|
.to_dict()
|
|
),
|
|
|
|
"last_error": (
|
|
schedule_service.last_error
|
|
),
|
|
}
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Status
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.get("/api/status")
|
|
async def status():
|
|
return get_status()
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Power
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.post("/api/tion/power/{state}")
|
|
async def set_power(state: Literal["on", "off"]):
|
|
|
|
enabled = state == "on"
|
|
|
|
if enabled:
|
|
operation = lambda tion: tion.power_on()
|
|
else:
|
|
operation = lambda tion: tion.power_off()
|
|
|
|
return await execute_command(
|
|
operation,
|
|
ScheduledSettings(power=enabled),
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Fan speed
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.post("/api/tion/speed/{speed}")
|
|
async def set_speed(
|
|
speed: Annotated[
|
|
int,
|
|
Path(
|
|
ge=MIN_FAN_SPEED,
|
|
le=MAX_FAN_SPEED,
|
|
),
|
|
] ):
|
|
|
|
|
|
|
|
return await execute_command(
|
|
lambda tion: tion.set_speed(speed),
|
|
ScheduledSettings(speed=speed),
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Heater
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.post("/api/tion/heater/{state}")
|
|
async def set_heater(state: Literal["on", "off"]):
|
|
|
|
enabled = state == "on"
|
|
|
|
if enabled:
|
|
operation = lambda tion: tion.heater_on()
|
|
else:
|
|
operation = lambda tion: tion.heater_off()
|
|
|
|
|
|
return await execute_command(
|
|
operation,
|
|
ScheduledSettings(heater=enabled),
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Target temperature
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.post("/api/tion/temperature/{temperature}")
|
|
|
|
async def set_temperature(
|
|
temperature: Annotated[
|
|
int,
|
|
Path(
|
|
ge=MIN_TARGET_TEMP,
|
|
le=MAX_TARGET_TEMP,
|
|
),
|
|
],):
|
|
|
|
return await execute_command(
|
|
lambda tion:
|
|
tion.set_target_temperature(temperature),
|
|
ScheduledSettings(target_temp=temperature),
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Air mode
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.post("/api/tion/mode/{mode}")
|
|
async def set_mode(
|
|
mode: str,
|
|
):
|
|
if mode not in (
|
|
AIR_MODE_OUTSIDE,
|
|
AIR_MODE_RECIRCULATION,
|
|
):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail={
|
|
"message": "Invalid air mode",
|
|
"allowed": [
|
|
AIR_MODE_OUTSIDE,
|
|
AIR_MODE_RECIRCULATION,
|
|
],
|
|
},
|
|
)
|
|
|
|
return await execute_command(
|
|
lambda tion:
|
|
tion.set_air_mode(mode),
|
|
|
|
ScheduledSettings(
|
|
mode=mode,
|
|
),
|
|
)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Sound
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.post("/api/tion/sound/{value}")
|
|
async def set_sound(
|
|
value: Literal["on", "off"],
|
|
):
|
|
enabled = value == "on"
|
|
|
|
if enabled:
|
|
operation = lambda tion: tion.sound_on()
|
|
else:
|
|
operation = lambda tion: tion.sound_off()
|
|
|
|
return await execute_command(
|
|
operation,
|
|
ScheduledSettings(
|
|
sound=enabled,
|
|
),
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Light
|
|
# ----------------------------------------------------------------------
|
|
|
|
@app.post("/api/tion/light/{value}")
|
|
async def set_light(
|
|
value: Literal["on", "off"],
|
|
):
|
|
enabled = value == "on"
|
|
|
|
if enabled:
|
|
operation = lambda tion: tion.light_on()
|
|
else:
|
|
operation = lambda tion: tion.light_off()
|
|
|
|
return await execute_command(
|
|
operation,
|
|
ScheduledSettings(
|
|
light=enabled,
|
|
),
|
|
) |