229 lines
7.5 KiB
Python
229 lines
7.5 KiB
Python
"""
|
|
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))
|