home-automation-backend/src/main.py

from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI

from src.components.poo_recorder import PooRecorder
from src.config import Config
from src.util.homeassistant import HomeAssistant
from src.util.location_recorder import LocationRecorder
from src.util.mqtt import MQTT
from src.util.notion import NotionAsync
from src.util.ticktick import TickTick

Config.init()

location_recorder_db = str(Path(__file__).resolve().parent / ".." / "location_recorder.db")

ticktick = TickTick()
notion = NotionAsync(token=Config.get_env(key="NOTION_TOKEN"))
mqtt = MQTT()
location_recorder = LocationRecorder(db_path=location_recorder_db)
homeassistant = HomeAssistant(ticktick=ticktick, location_recorder=location_recorder)
poo_recorder = PooRecorder(mqtt=mqtt, notion=notion, homeassistant=homeassistant)


@asynccontextmanager
async def _lifespan(_app: FastAPI):  # noqa: ANN202
    await mqtt.start()
    await location_recorder.create_db_engine()
    yield
    await mqtt.stop()


app = FastAPI(lifespan=_lifespan)


@app.get("/homeassistant/status")
async def get_status() -> dict:
    return {"Status": "Ok"}


@app.post("/homeassistant/publish")
async def homeassistant_publish(payload: HomeAssistant.Message) -> dict:
    return await homeassistant.process_message(message=payload)


# Poo recorder
@app.post("/poo/record")
async def record(record_detail: PooRecorder.RecordField) -> PooRecorder.RecordField:
    await poo_recorder.record(record_detail)
    return record_detail


# ticktick
@app.get("/ticktick/auth/code")
async def ticktick_auth(code: str, state: str) -> dict:
    if await ticktick.retrieve_access_token(code, state):
        return {"State": "Token Retrieved"}
    return {"State": "Token Retrieval Failed"}