84 lines
2.3 KiB
Python
84 lines
2.3 KiB
Python
from contextlib import asynccontextmanager
|
|
import logging
|
|
|
|
from fastapi import FastAPI, Form, Request
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from starlette.responses import RedirectResponse
|
|
|
|
import jinja2
|
|
|
|
from pnpdevice import settings
|
|
from pnpdevice.relays import Manager
|
|
import pnpdevice.discovery
|
|
import pnpdevice.relays
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
relays.connect()
|
|
yield
|
|
# relays.shutdown()
|
|
|
|
app = FastAPI(debug=settings.DEBUG, lifespan=lifespan)
|
|
relays = Manager(settings.CONFIGFILE, settings.RELAYS_FAKE)
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def index(request: Request):
|
|
return templates.TemplateResponse("index.template.html", {
|
|
"request": request,
|
|
"relays": relays,
|
|
})
|
|
|
|
@app.get("/relay/create")
|
|
def relay_create_get(request: Request):
|
|
"Get the form to create a new relay."
|
|
relays.load_all_pins()
|
|
used_pin_names = {relay.pin.name for relay in relays}
|
|
pins = [p for p in relays.pins if p.name not in used_pin_names]
|
|
sorted_pins = sorted(pins, key=lambda p: p.name)
|
|
return templates.TemplateResponse("relay-create.template.html", {
|
|
"relays": relays,
|
|
"request": request,
|
|
"pins": sorted_pins,
|
|
})
|
|
|
|
@app.post("/relay/create")
|
|
def relay_create_post(
|
|
request: Request,
|
|
pin_id: str = Form(...),
|
|
name: str = Form(...),
|
|
):
|
|
"Create a new relay from form POST."
|
|
chip_number, chip_name, line_number = pin_id.split("-")
|
|
chip_number = int(chip_number)
|
|
line_number = int(line_number)
|
|
relays.relay_add(chip_number, chip_name, line_number, name)
|
|
return RedirectResponse(status_code=303, url="/")
|
|
|
|
@app.get("/relay/{pin_id}")
|
|
def relay_detail_get(request: Request, pin_id: str):
|
|
"Get details on a single relay."
|
|
relay = relays.get_relay_by_pin_id(pin_id)
|
|
return templates.TemplateResponse("relay-detail.template.html", {
|
|
"request": request,
|
|
"relay": relay
|
|
})
|
|
|
|
@app.post("/relay/{pin_id}/state")
|
|
def relay_state_post(
|
|
request: Request,
|
|
pin_id: str,
|
|
state: bool = Form(...),
|
|
):
|
|
"Change the state of a relay."
|
|
relay = relays.get_relay_by_pin_id(pin_id)
|
|
relays.relay_set(relay, state)
|
|
return RedirectResponse(status_code=303, url="/")
|
|
|
|
async def status(request: Request):
|
|
return html("Status")
|