115 lines
3.1 KiB
Python
115 lines
3.1 KiB
Python
from contextlib import asynccontextmanager
|
|
from nt import device_encoding
|
|
from typing import Annotated
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from sqlmodel import SQLModel, Session, col, create_engine, select
|
|
from netbrite import NetBrite, NetbriteConnectionException, Zone
|
|
from db import (
|
|
APIAddNetBriteDevice,
|
|
APIGetNetBriteDevice,
|
|
APIGetZone,
|
|
APIMessages,
|
|
BaseNetBriteDevice,
|
|
BaseZone,
|
|
)
|
|
|
|
sqlite_file_name = "devices.db"
|
|
sqlite_url = f"sqlite:///{sqlite_file_name}"
|
|
|
|
connect_args = {"check_same_thread": False}
|
|
engine = create_engine(sqlite_url, connect_args=connect_args)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
print("creating tables")
|
|
create_db_and_tables()
|
|
yield
|
|
|
|
|
|
def create_db_and_tables():
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
|
|
def get_session():
|
|
with Session(engine) as session:
|
|
yield session
|
|
|
|
|
|
SessionDep = Annotated[Session, Depends(get_session)]
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
devices: list[NetBrite] = []
|
|
|
|
|
|
@app.post("/api/device")
|
|
def create_device(device: APIAddNetBriteDevice, session: SessionDep):
|
|
statement = select(BaseNetBriteDevice).where(
|
|
col(BaseNetBriteDevice.address) == device.address
|
|
)
|
|
result = session.exec(statement)
|
|
if result.first() != None:
|
|
raise HTTPException(400, "Device is already added")
|
|
|
|
try:
|
|
netbrite_device = NetBrite(device.address, device.port)
|
|
devices.append(netbrite_device)
|
|
except NetbriteConnectionException as exc:
|
|
raise HTTPException(400, "Failed to connect to device")
|
|
|
|
dbdevice = BaseNetBriteDevice(address=device.address, port=device.port)
|
|
session.add(dbdevice)
|
|
session.commit()
|
|
session.refresh(dbdevice)
|
|
|
|
return device
|
|
|
|
|
|
@app.get("/api/devices")
|
|
def get_devices(session: SessionDep):
|
|
db_devices: dict[int, BaseNetBriteDevice] = {}
|
|
|
|
statement = select(BaseZone, BaseNetBriteDevice).join(BaseNetBriteDevice)
|
|
results = session.exec(statement)
|
|
|
|
for zone, device in results:
|
|
if device.id == None:
|
|
continue
|
|
if not device.id in db_devices:
|
|
connected = devices.inde
|
|
get_device = APIGetNetBriteDevice(device., zones={})
|
|
db_devices[device.id] = device
|
|
|
|
db_devices[device.id].zones_list
|
|
|
|
|
|
for device in devices:
|
|
|
|
zones: dict[str, APIGetZone] = {}
|
|
for index, zone in device.zones_list.items():
|
|
x, y, xend, yend = zone.rect
|
|
width = xend - x
|
|
height = yend - y
|
|
|
|
zones[index] = APIGetZone(
|
|
x=x,
|
|
y=y,
|
|
width=width,
|
|
height=height,
|
|
scroll_speed=zone.scroll_speed,
|
|
pause_duration=zone.pause_duration,
|
|
volume=zone.volume,
|
|
default_font=zone.default_font,
|
|
default_color=zone.default_color,
|
|
default_message=zone.initial_text,
|
|
)
|
|
|
|
return devices
|
|
|
|
|
|
@app.get("/api/devices/{device_index}")
|
|
def get_device(device_index: int):
|
|
if device_index > len(devices) - 1:
|
|
raise HTTPException(400, "Device doesn't exist")
|
|
|
|
return devices[device_index]
|