from fastapi import FastAPI, Request, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from fastapi.responses import HTMLResponse from sqlalchemy.orm import Session from backend.database import engine, Base, get_db from backend.routers import roster, units, photos, roster_edit, dashboard, dashboard_tabs from backend.services.snapshot import emit_status_snapshot from backend.models import IgnoredUnit # 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) from backend.routers import settings app.include_router(settings.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("/settings", response_class=HTMLResponse) async def settings_page(request: Request): """Settings page for roster management""" return templates.TemplateResponse("settings.html", {"request": request}) @app.get("/partials/roster-deployed", response_class=HTMLResponse) async def roster_deployed_partial(request: Request): """Partial template for deployed units tab""" from datetime import datetime snapshot = emit_status_snapshot() units_list = [] for unit_id, unit_data in snapshot["active"].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/roster-benched", response_class=HTMLResponse) async def roster_benched_partial(request: Request): """Partial template for benched units tab""" from datetime import datetime snapshot = emit_status_snapshot() units_list = [] for unit_id, unit_data in snapshot["benched"].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 ID units_list.sort(key=lambda x: x["id"]) return templates.TemplateResponse("partials/roster_table.html", { "request": request, "units": units_list, "timestamp": datetime.now().strftime("%H:%M:%S") }) @app.get("/partials/roster-retired", response_class=HTMLResponse) async def roster_retired_partial(request: Request): """Partial template for retired units tab""" from datetime import datetime snapshot = emit_status_snapshot() units_list = [] for unit_id, unit_data in snapshot["retired"].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 ID units_list.sort(key=lambda x: x["id"]) return templates.TemplateResponse("partials/retired_table.html", { "request": request, "units": units_list, "timestamp": datetime.now().strftime("%H:%M:%S") }) @app.get("/partials/roster-ignored", response_class=HTMLResponse) async def roster_ignored_partial(request: Request, db: Session = Depends(get_db)): """Partial template for ignored units tab""" from datetime import datetime ignored = db.query(IgnoredUnit).all() ignored_list = [] for unit in ignored: ignored_list.append({ "id": unit.id, "reason": unit.reason or "", "ignored_at": unit.ignored_at.strftime("%Y-%m-%d %H:%M:%S") if unit.ignored_at else "Unknown" }) # Sort by ID ignored_list.sort(key=lambda x: x["id"]) return templates.TemplateResponse("partials/ignored_table.html", { "request": request, "ignored_units": ignored_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)