import os 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, FileResponse, JSONResponse from sqlalchemy.orm import Session from typing import List, Dict from pydantic import BaseModel 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) # Read environment (development or production) ENVIRONMENT = os.getenv("ENVIRONMENT", "production") # Initialize FastAPI app VERSION = "0.3.2" app = FastAPI( title="Seismo Fleet Manager", description="Backend API for managing seismograph fleet status", version=VERSION ) # 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") # Add custom context processor to inject environment variable into all templates @app.middleware("http") async def add_environment_to_context(request: Request, call_next): """Middleware to add environment variable to request state""" request.state.environment = ENVIRONMENT response = await call_next(request) return response # Override TemplateResponse to include environment and version in context original_template_response = templates.TemplateResponse def custom_template_response(name, context=None, *args, **kwargs): if context is None: context = {} context["environment"] = ENVIRONMENT context["version"] = VERSION return original_template_response(name, context, *args, **kwargs) templates.TemplateResponse = custom_template_response # 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}) # ===== PWA ROUTES ===== @app.get("/sw.js") async def service_worker(): """Serve service worker with proper headers for PWA""" return FileResponse( "backend/static/sw.js", media_type="application/javascript", headers={ "Service-Worker-Allowed": "/", "Cache-Control": "no-cache" } ) @app.get("/offline-db.js") async def offline_db_script(): """Serve offline database script""" return FileResponse( "backend/static/offline-db.js", media_type="application/javascript", headers={"Cache-Control": "no-cache"} ) # Pydantic model for sync edits class EditItem(BaseModel): id: int unitId: str changes: Dict timestamp: int class SyncEditsRequest(BaseModel): edits: List[EditItem] @app.post("/api/sync-edits") async def sync_edits(request: SyncEditsRequest, db: Session = Depends(get_db)): """Process offline edit queue and sync to database""" from backend.models import RosterUnit results = [] synced_ids = [] for edit in request.edits: try: # Find the unit unit = db.query(RosterUnit).filter_by(id=edit.unitId).first() if not unit: results.append({ "id": edit.id, "status": "error", "reason": f"Unit {edit.unitId} not found" }) continue # Apply changes for key, value in edit.changes.items(): if hasattr(unit, key): # Handle boolean conversions if key in ['deployed', 'retired']: setattr(unit, key, value in ['true', True, 'True', '1', 1]) else: setattr(unit, key, value if value != '' else None) db.commit() results.append({ "id": edit.id, "status": "success" }) synced_ids.append(edit.id) except Exception as e: db.rollback() results.append({ "id": edit.id, "status": "error", "reason": str(e) }) synced_count = len(synced_ids) return JSONResponse({ "synced": synced_count, "total": len(request.edits), "synced_ids": synced_ids, "results": results }) @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.get("status", "Unknown"), "age": unit_data.get("age", "N/A"), "last_seen": unit_data.get("last", "Never"), "deployed": unit_data.get("deployed", False), "note": unit_data.get("note", ""), "device_type": unit_data.get("device_type", "seismograph"), "address": unit_data.get("address", ""), "coordinates": unit_data.get("coordinates", ""), "project_id": unit_data.get("project_id", ""), "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.get("status", "N/A"), "age": unit_data.get("age", "N/A"), "last_seen": unit_data.get("last", "Never"), "deployed": unit_data.get("deployed", False), "note": unit_data.get("note", ""), "device_type": unit_data.get("device_type", "seismograph"), "address": unit_data.get("address", ""), "coordinates": unit_data.get("coordinates", ""), "project_id": unit_data.get("project_id", ""), "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": f"Seismo Fleet Manager v{VERSION}", "status": "running", "version": VERSION } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8001)