Richie
5c230a267c
build_systems / build-brain (pull_request) Successful in 46s
build_systems / build-bob (pull_request) Successful in 46s
build_systems / build-leviathan (pull_request) Successful in 54s
build_systems / build-rhapsody-in-green (pull_request) Successful in 58s
build_systems / build-jeeves (pull_request) Successful in 2m33s
treefmt / nix fmt (push) Successful in 4s
build_systems / build-brain (push) Successful in 30s
build_systems / build-bob (push) Successful in 9s
pytest / pytest (push) Successful in 27s
build_systems / build-leviathan (push) Successful in 40s
build_systems / build-rhapsody-in-green (push) Successful in 44s
build_systems / build-jeeves (push) Successful in 2m25s
treefmt / nix fmt (pull_request) Successful in 4s
pytest / pytest (pull_request) Successful in 25s
90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
"""FastAPI heater control service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
from typing import TYPE_CHECKING, Annotated
|
|
|
|
import typer
|
|
import uvicorn
|
|
from fastapi import FastAPI, HTTPException
|
|
|
|
from python.common import configure_logger
|
|
from python.heater.controller import HeaterController
|
|
from python.heater.models import ActionResult, DeviceConfig, HeaterStatus
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def create_app(config: DeviceConfig) -> FastAPI:
|
|
"""Create FastAPI application."""
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|
app.state.controller = HeaterController(config)
|
|
yield
|
|
|
|
app = FastAPI(
|
|
title="Heater Control API",
|
|
description="Fast local control for Tuya heater",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
@app.get("/status")
|
|
def get_status() -> HeaterStatus:
|
|
return app.state.controller.status()
|
|
|
|
@app.post("/on")
|
|
def heater_on() -> ActionResult:
|
|
result = app.state.controller.turn_on()
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
return result
|
|
|
|
@app.post("/off")
|
|
def heater_off() -> ActionResult:
|
|
result = app.state.controller.turn_off()
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
return result
|
|
|
|
@app.post("/toggle")
|
|
def heater_toggle() -> ActionResult:
|
|
result = app.state.controller.toggle()
|
|
if not result.success:
|
|
raise HTTPException(status_code=500, detail=result.error)
|
|
return result
|
|
|
|
return app
|
|
|
|
|
|
def serve(
|
|
host: Annotated[str, typer.Option("--host", "-h", help="Host to bind to")],
|
|
port: Annotated[int, typer.Option("--port", "-p", help="Port to bind to")] = 8124,
|
|
log_level: Annotated[str, typer.Option("--log-level", "-l", help="Log level")] = "INFO",
|
|
device_id: Annotated[str | None, typer.Option("--device-id", envvar="TUYA_DEVICE_ID")] = None,
|
|
device_ip: Annotated[str | None, typer.Option("--device-ip", envvar="TUYA_DEVICE_IP")] = None,
|
|
local_key: Annotated[str | None, typer.Option("--local-key", envvar="TUYA_LOCAL_KEY")] = None,
|
|
) -> None:
|
|
"""Start the heater control API server."""
|
|
configure_logger(log_level)
|
|
|
|
logger.info("Starting heater control API server")
|
|
|
|
if not device_id or not device_ip or not local_key:
|
|
error = "Must provide device ID, IP, and local key"
|
|
raise typer.Exit(error)
|
|
|
|
config = DeviceConfig(device_id=device_id, ip=device_ip, local_key=local_key)
|
|
|
|
app = create_app(config)
|
|
uvicorn.run(app, host=host, port=port)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
typer.run(serve)
|