144 lines
4.4 KiB
Python
144 lines
4.4 KiB
Python
from fastapi import FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from fastapi.responses import HTMLResponse
|
|
|
|
from backend.database import engine, Base
|
|
from backend.routers import roster, units, photos, roster_edit, dashboard, dashboard_tabs
|
|
from backend.services.snapshot import emit_status_snapshot
|
|
|
|
# Create database tables
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# Initialize FastAPI app
|
|
app = FastAPI(
|
|
title="Seismo Fleet Manager",
|
|
description="Backend API for managing seismograph fleet status",
|
|
version="0.1.1"
|
|
)
|
|
|
|
# Configure CORS
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Mount static files
|
|
app.mount("/static", StaticFiles(directory="backend/static"), name="static")
|
|
|
|
# Setup Jinja2 templates
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
# Include API routers
|
|
app.include_router(roster.router)
|
|
app.include_router(units.router)
|
|
app.include_router(photos.router)
|
|
app.include_router(roster_edit.router)
|
|
app.include_router(dashboard.router)
|
|
app.include_router(dashboard_tabs.router)
|
|
|
|
|
|
|
|
# Legacy routes from the original backend
|
|
from backend import routes as legacy_routes
|
|
app.include_router(legacy_routes.router)
|
|
|
|
|
|
# HTML page routes
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def dashboard(request: Request):
|
|
"""Dashboard home page"""
|
|
return templates.TemplateResponse("dashboard.html", {"request": request})
|
|
|
|
|
|
@app.get("/roster", response_class=HTMLResponse)
|
|
async def roster_page(request: Request):
|
|
"""Fleet roster page"""
|
|
return templates.TemplateResponse("roster.html", {"request": request})
|
|
|
|
|
|
@app.get("/unit/{unit_id}", response_class=HTMLResponse)
|
|
async def unit_detail_page(request: Request, unit_id: str):
|
|
"""Unit detail page"""
|
|
return templates.TemplateResponse("unit_detail.html", {
|
|
"request": request,
|
|
"unit_id": unit_id
|
|
})
|
|
|
|
|
|
@app.get("/partials/roster-table", response_class=HTMLResponse)
|
|
async def roster_table_partial(request: Request):
|
|
"""Partial template for roster table (HTMX)"""
|
|
from datetime import datetime
|
|
snapshot = emit_status_snapshot()
|
|
|
|
units_list = []
|
|
for unit_id, unit_data in snapshot["units"].items():
|
|
units_list.append({
|
|
"id": unit_id,
|
|
"status": unit_data["status"],
|
|
"age": unit_data["age"],
|
|
"last_seen": unit_data["last"],
|
|
"deployed": unit_data["deployed"],
|
|
"note": unit_data.get("note", ""),
|
|
"device_type": unit_data.get("device_type", "seismograph"),
|
|
"last_calibrated": unit_data.get("last_calibrated"),
|
|
"next_calibration_due": unit_data.get("next_calibration_due"),
|
|
"deployed_with_modem_id": unit_data.get("deployed_with_modem_id"),
|
|
"ip_address": unit_data.get("ip_address"),
|
|
"phone_number": unit_data.get("phone_number"),
|
|
"hardware_model": unit_data.get("hardware_model"),
|
|
})
|
|
|
|
# Sort by status priority (Missing > Pending > OK) then by ID
|
|
status_priority = {"Missing": 0, "Pending": 1, "OK": 2}
|
|
units_list.sort(key=lambda x: (status_priority.get(x["status"], 3), x["id"]))
|
|
|
|
return templates.TemplateResponse("partials/roster_table.html", {
|
|
"request": request,
|
|
"units": units_list,
|
|
"timestamp": datetime.now().strftime("%H:%M:%S")
|
|
})
|
|
|
|
|
|
@app.get("/partials/unknown-emitters", response_class=HTMLResponse)
|
|
async def unknown_emitters_partial(request: Request):
|
|
"""Partial template for unknown emitters (HTMX)"""
|
|
snapshot = emit_status_snapshot()
|
|
|
|
unknown_list = []
|
|
for unit_id, unit_data in snapshot.get("unknown", {}).items():
|
|
unknown_list.append({
|
|
"id": unit_id,
|
|
"status": unit_data["status"],
|
|
"age": unit_data["age"],
|
|
"fname": unit_data.get("fname", ""),
|
|
})
|
|
|
|
# Sort by ID
|
|
unknown_list.sort(key=lambda x: x["id"])
|
|
|
|
return templates.TemplateResponse("partials/unknown_emitters.html", {
|
|
"request": request,
|
|
"unknown_units": unknown_list
|
|
})
|
|
|
|
|
|
@app.get("/health")
|
|
def health_check():
|
|
"""Health check endpoint"""
|
|
return {
|
|
"message": "Seismo Fleet Manager v0.1.1",
|
|
"status": "running",
|
|
"version": "0.1.1"
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8001)
|