""" Seismograph Dashboard API Router Provides endpoints for the seismograph-specific dashboard """ from datetime import date, datetime, timedelta from fastapi import APIRouter, Request, Depends, Query, Form, HTTPException from fastapi.responses import HTMLResponse from sqlalchemy.orm import Session from backend.database import get_db from backend.models import RosterUnit, UnitHistory, UserPreferences from backend.templates_config import templates router = APIRouter(prefix="/api/seismo-dashboard", tags=["seismo-dashboard"]) @router.get("/stats", response_class=HTMLResponse) async def get_seismo_stats(request: Request, db: Session = Depends(get_db)): """ Returns HTML partial with seismograph statistics summary """ # Get all seismograph units seismos = db.query(RosterUnit).filter_by( device_type="seismograph", retired=False ).all() total = len(seismos) deployed = sum(1 for s in seismos if s.deployed) benched = sum(1 for s in seismos if not s.deployed and not s.out_for_calibration) out_for_calibration = sum(1 for s in seismos if s.out_for_calibration) # Count modems assigned to deployed seismographs with_modem = sum(1 for s in seismos if s.deployed and s.deployed_with_modem_id) without_modem = deployed - with_modem return templates.TemplateResponse( "partials/seismo_stats.html", { "request": request, "total": total, "deployed": deployed, "benched": benched, "out_for_calibration": out_for_calibration, "with_modem": with_modem, "without_modem": without_modem } ) @router.get("/units", response_class=HTMLResponse) async def get_seismo_units( request: Request, db: Session = Depends(get_db), search: str = Query(None), sort: str = Query("id"), order: str = Query("asc"), status: str = Query(None), modem: str = Query(None) ): """ Returns HTML partial with filterable and sortable seismograph unit list """ query = db.query(RosterUnit).filter_by( device_type="seismograph", retired=False ) # Apply search filter if search: query = query.filter( (RosterUnit.id.ilike(f"%{search}%")) | (RosterUnit.note.ilike(f"%{search}%")) | (RosterUnit.address.ilike(f"%{search}%")) ) # Apply status filter if status == "deployed": query = query.filter(RosterUnit.deployed == True) elif status == "benched": query = query.filter(RosterUnit.deployed == False, RosterUnit.out_for_calibration == False) elif status == "out_for_calibration": query = query.filter(RosterUnit.out_for_calibration == True) # Apply modem filter if modem == "with": query = query.filter(RosterUnit.deployed_with_modem_id.isnot(None)) elif modem == "without": query = query.filter(RosterUnit.deployed_with_modem_id.is_(None)) # Apply sorting sort_column_map = { "id": RosterUnit.id, "status": RosterUnit.deployed, "modem": RosterUnit.deployed_with_modem_id, "location": RosterUnit.address, "last_calibrated": RosterUnit.last_calibrated, "notes": RosterUnit.note } sort_column = sort_column_map.get(sort, RosterUnit.id) if order == "desc": query = query.order_by(sort_column.desc()) else: query = query.order_by(sort_column.asc()) seismos = query.all() return templates.TemplateResponse( "partials/seismo_unit_list.html", { "request": request, "units": seismos, "search": search or "", "sort": sort, "order": order, "status": status or "", "modem": modem or "", "today": date.today() } ) def _get_calibration_interval(db: Session) -> int: prefs = db.query(UserPreferences).first() if prefs and prefs.calibration_interval_days: return prefs.calibration_interval_days return 365 def _row_context(request: Request, unit: RosterUnit) -> dict: return {"request": request, "unit": unit, "today": date.today()} @router.get("/unit/{unit_id}/view-row", response_class=HTMLResponse) async def get_seismo_view_row(unit_id: str, request: Request, db: Session = Depends(get_db)): unit = db.query(RosterUnit).filter(RosterUnit.id == unit_id).first() if not unit: raise HTTPException(status_code=404, detail="Unit not found") return templates.TemplateResponse("partials/seismo_row_view.html", _row_context(request, unit)) @router.get("/unit/{unit_id}/edit-row", response_class=HTMLResponse) async def get_seismo_edit_row(unit_id: str, request: Request, db: Session = Depends(get_db)): unit = db.query(RosterUnit).filter(RosterUnit.id == unit_id).first() if not unit: raise HTTPException(status_code=404, detail="Unit not found") return templates.TemplateResponse("partials/seismo_row_edit.html", _row_context(request, unit)) @router.post("/unit/{unit_id}/quick-update", response_class=HTMLResponse) async def quick_update_seismo_unit( unit_id: str, request: Request, db: Session = Depends(get_db), status: str = Form(...), last_calibrated: str = Form(""), note: str = Form(""), ): unit = db.query(RosterUnit).filter(RosterUnit.id == unit_id).first() if not unit: raise HTTPException(status_code=404, detail="Unit not found") # --- Status --- old_deployed = unit.deployed old_out_for_cal = unit.out_for_calibration if status == "deployed": unit.deployed = True unit.out_for_calibration = False elif status == "out_for_calibration": unit.deployed = False unit.out_for_calibration = True else: unit.deployed = False unit.out_for_calibration = False if unit.deployed != old_deployed or unit.out_for_calibration != old_out_for_cal: old_status = "deployed" if old_deployed else ("out_for_calibration" if old_out_for_cal else "benched") db.add(UnitHistory( unit_id=unit_id, change_type="deployed_change", field_name="status", old_value=old_status, new_value=status, source="manual", )) # --- Last calibrated --- old_cal = unit.last_calibrated if last_calibrated: try: new_cal = datetime.strptime(last_calibrated, "%Y-%m-%d").date() except ValueError: raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD") unit.last_calibrated = new_cal unit.next_calibration_due = new_cal + timedelta(days=_get_calibration_interval(db)) else: unit.last_calibrated = None unit.next_calibration_due = None if unit.last_calibrated != old_cal: db.add(UnitHistory( unit_id=unit_id, change_type="calibration_status_change", field_name="last_calibrated", old_value=old_cal.strftime("%Y-%m-%d") if old_cal else None, new_value=last_calibrated or None, source="manual", )) # --- Note --- old_note = unit.note unit.note = note or None if unit.note != old_note: db.add(UnitHistory( unit_id=unit_id, change_type="note_change", field_name="note", old_value=old_note, new_value=unit.note, source="manual", )) db.commit() db.refresh(unit) return templates.TemplateResponse("partials/seismo_row_view.html", _row_context(request, unit))