430 lines
14 KiB
Python
430 lines
14 KiB
Python
"""
|
|
Modem Dashboard Router
|
|
|
|
Provides API endpoints for the Field Modems management page.
|
|
"""
|
|
|
|
from fastapi import APIRouter, Request, Depends, Query
|
|
from fastapi.responses import HTMLResponse
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime
|
|
import subprocess
|
|
import time
|
|
import logging
|
|
|
|
from backend.database import get_db
|
|
from backend.models import RosterUnit
|
|
from backend.templates_config import templates
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/modem-dashboard", tags=["modem-dashboard"])
|
|
|
|
|
|
@router.get("/stats", response_class=HTMLResponse)
|
|
async def get_modem_stats(request: Request, db: Session = Depends(get_db)):
|
|
"""
|
|
Get summary statistics for modem dashboard.
|
|
Returns HTML partial with stat cards.
|
|
"""
|
|
# Query all modems
|
|
all_modems = db.query(RosterUnit).filter_by(device_type="modem").all()
|
|
|
|
# Get IDs of modems that have devices paired to them
|
|
paired_modem_ids = set()
|
|
devices_with_modems = db.query(RosterUnit).filter(
|
|
RosterUnit.deployed_with_modem_id.isnot(None),
|
|
RosterUnit.retired == False
|
|
).all()
|
|
for device in devices_with_modems:
|
|
if device.deployed_with_modem_id:
|
|
paired_modem_ids.add(device.deployed_with_modem_id)
|
|
|
|
# Count categories
|
|
total_count = len(all_modems)
|
|
retired_count = sum(1 for m in all_modems if m.retired)
|
|
|
|
# In use = deployed AND paired with a device
|
|
in_use_count = sum(1 for m in all_modems
|
|
if m.deployed and not m.retired and m.id in paired_modem_ids)
|
|
|
|
# Spare = deployed but NOT paired (available for assignment)
|
|
spare_count = sum(1 for m in all_modems
|
|
if m.deployed and not m.retired and m.id not in paired_modem_ids)
|
|
|
|
# Benched = not deployed and not retired
|
|
benched_count = sum(1 for m in all_modems if not m.deployed and not m.retired)
|
|
|
|
return templates.TemplateResponse("partials/modem_stats.html", {
|
|
"request": request,
|
|
"total_count": total_count,
|
|
"in_use_count": in_use_count,
|
|
"spare_count": spare_count,
|
|
"benched_count": benched_count,
|
|
"retired_count": retired_count
|
|
})
|
|
|
|
|
|
@router.get("/units", response_class=HTMLResponse)
|
|
async def get_modem_units(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
search: str = Query(None),
|
|
filter_status: str = Query(None), # "in_use", "spare", "benched", "retired"
|
|
):
|
|
"""
|
|
Get list of modem units for the dashboard.
|
|
Returns HTML partial with modem cards.
|
|
"""
|
|
query = db.query(RosterUnit).filter_by(device_type="modem")
|
|
|
|
# Filter by search term if provided
|
|
if search:
|
|
search_term = f"%{search}%"
|
|
query = query.filter(
|
|
(RosterUnit.id.ilike(search_term)) |
|
|
(RosterUnit.ip_address.ilike(search_term)) |
|
|
(RosterUnit.hardware_model.ilike(search_term)) |
|
|
(RosterUnit.phone_number.ilike(search_term)) |
|
|
(RosterUnit.location.ilike(search_term))
|
|
)
|
|
|
|
modems = query.order_by(
|
|
RosterUnit.retired.asc(),
|
|
RosterUnit.deployed.desc(),
|
|
RosterUnit.id.asc()
|
|
).all()
|
|
|
|
# Get paired device info for each modem
|
|
paired_devices = {}
|
|
devices_with_modems = db.query(RosterUnit).filter(
|
|
RosterUnit.deployed_with_modem_id.isnot(None),
|
|
RosterUnit.retired == False
|
|
).all()
|
|
for device in devices_with_modems:
|
|
if device.deployed_with_modem_id:
|
|
paired_devices[device.deployed_with_modem_id] = {
|
|
"id": device.id,
|
|
"device_type": device.device_type,
|
|
"deployed": device.deployed
|
|
}
|
|
|
|
# Annotate modems with paired device info
|
|
modem_list = []
|
|
for modem in modems:
|
|
paired = paired_devices.get(modem.id)
|
|
|
|
# Determine status category
|
|
if modem.retired:
|
|
status = "retired"
|
|
elif not modem.deployed:
|
|
status = "benched"
|
|
elif paired:
|
|
status = "in_use"
|
|
else:
|
|
status = "spare"
|
|
|
|
# Apply filter if specified
|
|
if filter_status and status != filter_status:
|
|
continue
|
|
|
|
modem_list.append({
|
|
"id": modem.id,
|
|
"ip_address": modem.ip_address,
|
|
"phone_number": modem.phone_number,
|
|
"hardware_model": modem.hardware_model,
|
|
"deployed": modem.deployed,
|
|
"retired": modem.retired,
|
|
"location": modem.location,
|
|
"project_id": modem.project_id,
|
|
"paired_device": paired,
|
|
"status": status
|
|
})
|
|
|
|
return templates.TemplateResponse("partials/modem_list.html", {
|
|
"request": request,
|
|
"modems": modem_list
|
|
})
|
|
|
|
|
|
@router.get("/{modem_id}/paired-device")
|
|
async def get_paired_device(modem_id: str, db: Session = Depends(get_db)):
|
|
"""
|
|
Get the device (SLM/seismograph) that is paired with this modem.
|
|
Returns JSON with device info or null if not paired.
|
|
"""
|
|
# Check modem exists
|
|
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
|
|
if not modem:
|
|
return {"status": "error", "detail": f"Modem {modem_id} not found"}
|
|
|
|
# Find device paired with this modem
|
|
device = db.query(RosterUnit).filter(
|
|
RosterUnit.deployed_with_modem_id == modem_id,
|
|
RosterUnit.retired == False
|
|
).first()
|
|
|
|
if device:
|
|
return {
|
|
"paired": True,
|
|
"device": {
|
|
"id": device.id,
|
|
"device_type": device.device_type,
|
|
"deployed": device.deployed,
|
|
"project_id": device.project_id,
|
|
"location": device.location or device.address
|
|
}
|
|
}
|
|
|
|
return {"paired": False, "device": None}
|
|
|
|
|
|
@router.get("/{modem_id}/paired-device-html", response_class=HTMLResponse)
|
|
async def get_paired_device_html(modem_id: str, request: Request, db: Session = Depends(get_db)):
|
|
"""
|
|
Get HTML partial showing the device paired with this modem.
|
|
Used by unit_detail.html for modems.
|
|
"""
|
|
# Check modem exists
|
|
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
|
|
if not modem:
|
|
return HTMLResponse('<p class="text-red-500">Modem not found</p>')
|
|
|
|
# Find device paired with this modem
|
|
device = db.query(RosterUnit).filter(
|
|
RosterUnit.deployed_with_modem_id == modem_id,
|
|
RosterUnit.retired == False
|
|
).first()
|
|
|
|
return templates.TemplateResponse("partials/modem_paired_device.html", {
|
|
"request": request,
|
|
"modem_id": modem_id,
|
|
"device": device
|
|
})
|
|
|
|
|
|
@router.get("/{modem_id}/ping")
|
|
async def ping_modem(modem_id: str, db: Session = Depends(get_db)):
|
|
"""
|
|
Test modem connectivity with a simple ping.
|
|
Returns response time and connection status.
|
|
"""
|
|
# Get modem from database
|
|
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
|
|
|
|
if not modem:
|
|
return {"status": "error", "detail": f"Modem {modem_id} not found"}
|
|
|
|
if not modem.ip_address:
|
|
return {"status": "error", "detail": f"Modem {modem_id} has no IP address configured"}
|
|
|
|
try:
|
|
# Ping the modem (1 packet, 2 second timeout)
|
|
start_time = time.time()
|
|
result = subprocess.run(
|
|
["ping", "-c", "1", "-W", "2", modem.ip_address],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=3
|
|
)
|
|
response_time = int((time.time() - start_time) * 1000) # Convert to milliseconds
|
|
|
|
if result.returncode == 0:
|
|
return {
|
|
"status": "success",
|
|
"modem_id": modem_id,
|
|
"ip_address": modem.ip_address,
|
|
"response_time_ms": response_time,
|
|
"message": "Modem is responding"
|
|
}
|
|
else:
|
|
return {
|
|
"status": "error",
|
|
"modem_id": modem_id,
|
|
"ip_address": modem.ip_address,
|
|
"detail": "Modem not responding to ping"
|
|
}
|
|
|
|
except subprocess.TimeoutExpired:
|
|
return {
|
|
"status": "error",
|
|
"modem_id": modem_id,
|
|
"ip_address": modem.ip_address,
|
|
"detail": "Ping timeout"
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Failed to ping modem {modem_id}: {e}")
|
|
return {
|
|
"status": "error",
|
|
"modem_id": modem_id,
|
|
"detail": str(e)
|
|
}
|
|
|
|
|
|
@router.get("/{modem_id}/diagnostics")
|
|
async def get_modem_diagnostics(modem_id: str, db: Session = Depends(get_db)):
|
|
"""
|
|
Get modem diagnostics (signal strength, data usage, uptime).
|
|
|
|
Currently returns placeholders. When ModemManager is available,
|
|
this endpoint will query it for real diagnostics.
|
|
"""
|
|
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
|
|
if not modem:
|
|
return {"status": "error", "detail": f"Modem {modem_id} not found"}
|
|
|
|
# TODO: Query ModemManager backend when available
|
|
return {
|
|
"status": "unavailable",
|
|
"message": "ModemManager integration not yet available",
|
|
"modem_id": modem_id,
|
|
"signal_strength_dbm": None,
|
|
"data_usage_mb": None,
|
|
"uptime_seconds": None,
|
|
"carrier": None,
|
|
"connection_type": None # LTE, 5G, etc.
|
|
}
|
|
|
|
|
|
@router.get("/{modem_id}/pairable-devices")
|
|
async def get_pairable_devices(
|
|
modem_id: str,
|
|
db: Session = Depends(get_db),
|
|
search: str = Query(None),
|
|
hide_paired: bool = Query(True)
|
|
):
|
|
"""
|
|
Get list of devices (seismographs and SLMs) that can be paired with this modem.
|
|
Used by the device picker modal in unit_detail.html.
|
|
"""
|
|
# Check modem exists
|
|
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
|
|
if not modem:
|
|
return {"status": "error", "detail": f"Modem {modem_id} not found"}
|
|
|
|
# Query seismographs and SLMs
|
|
query = db.query(RosterUnit).filter(
|
|
RosterUnit.device_type.in_(["seismograph", "sound_level_meter"]),
|
|
RosterUnit.retired == False
|
|
)
|
|
|
|
# Filter by search term if provided
|
|
if search:
|
|
search_term = f"%{search}%"
|
|
query = query.filter(
|
|
(RosterUnit.id.ilike(search_term)) |
|
|
(RosterUnit.project_id.ilike(search_term)) |
|
|
(RosterUnit.location.ilike(search_term)) |
|
|
(RosterUnit.address.ilike(search_term)) |
|
|
(RosterUnit.note.ilike(search_term))
|
|
)
|
|
|
|
devices = query.order_by(
|
|
RosterUnit.deployed.desc(),
|
|
RosterUnit.device_type.asc(),
|
|
RosterUnit.id.asc()
|
|
).all()
|
|
|
|
# Build device list
|
|
device_list = []
|
|
for device in devices:
|
|
# Skip already paired devices if hide_paired is True
|
|
is_paired_to_other = (
|
|
device.deployed_with_modem_id is not None and
|
|
device.deployed_with_modem_id != modem_id
|
|
)
|
|
is_paired_to_this = device.deployed_with_modem_id == modem_id
|
|
|
|
if hide_paired and is_paired_to_other:
|
|
continue
|
|
|
|
device_list.append({
|
|
"id": device.id,
|
|
"device_type": device.device_type,
|
|
"deployed": device.deployed,
|
|
"project_id": device.project_id,
|
|
"location": device.location or device.address,
|
|
"note": device.note,
|
|
"paired_modem_id": device.deployed_with_modem_id,
|
|
"is_paired_to_this": is_paired_to_this,
|
|
"is_paired_to_other": is_paired_to_other
|
|
})
|
|
|
|
return {"devices": device_list, "modem_id": modem_id}
|
|
|
|
|
|
@router.post("/{modem_id}/pair")
|
|
async def pair_device_to_modem(
|
|
modem_id: str,
|
|
db: Session = Depends(get_db),
|
|
device_id: str = Query(..., description="ID of the device to pair")
|
|
):
|
|
"""
|
|
Pair a device (seismograph or SLM) to this modem.
|
|
Updates the device's deployed_with_modem_id field.
|
|
"""
|
|
# Check modem exists
|
|
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
|
|
if not modem:
|
|
return {"status": "error", "detail": f"Modem {modem_id} not found"}
|
|
|
|
# Find the device
|
|
device = db.query(RosterUnit).filter(
|
|
RosterUnit.id == device_id,
|
|
RosterUnit.device_type.in_(["seismograph", "sound_level_meter"]),
|
|
RosterUnit.retired == False
|
|
).first()
|
|
if not device:
|
|
return {"status": "error", "detail": f"Device {device_id} not found"}
|
|
|
|
# Unpair any device currently paired to this modem
|
|
currently_paired = db.query(RosterUnit).filter(
|
|
RosterUnit.deployed_with_modem_id == modem_id
|
|
).all()
|
|
for paired_device in currently_paired:
|
|
paired_device.deployed_with_modem_id = None
|
|
|
|
# Pair the new device
|
|
device.deployed_with_modem_id = modem_id
|
|
db.commit()
|
|
|
|
return {
|
|
"status": "success",
|
|
"modem_id": modem_id,
|
|
"device_id": device_id,
|
|
"message": f"Device {device_id} paired to modem {modem_id}"
|
|
}
|
|
|
|
|
|
@router.post("/{modem_id}/unpair")
|
|
async def unpair_device_from_modem(modem_id: str, db: Session = Depends(get_db)):
|
|
"""
|
|
Unpair any device currently paired to this modem.
|
|
"""
|
|
# Check modem exists
|
|
modem = db.query(RosterUnit).filter_by(id=modem_id, device_type="modem").first()
|
|
if not modem:
|
|
return {"status": "error", "detail": f"Modem {modem_id} not found"}
|
|
|
|
# Find and unpair device
|
|
device = db.query(RosterUnit).filter(
|
|
RosterUnit.deployed_with_modem_id == modem_id
|
|
).first()
|
|
|
|
if device:
|
|
old_device_id = device.id
|
|
device.deployed_with_modem_id = None
|
|
db.commit()
|
|
return {
|
|
"status": "success",
|
|
"modem_id": modem_id,
|
|
"unpaired_device_id": old_device_id,
|
|
"message": f"Device {old_device_id} unpaired from modem {modem_id}"
|
|
}
|
|
|
|
return {
|
|
"status": "success",
|
|
"modem_id": modem_id,
|
|
"message": "No device was paired to this modem"
|
|
}
|