from typing import Any, Dict from datetime import datetime, timezone, timedelta from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from sqlalchemy import func from app.db.session import get_db from app.models.device import Device from app.models.telemetry import Telemetry from app.models.log import Log, LogLevel router = APIRouter() @router.get("/") def get_system_status( db: Session = Depends(get_db), ) -> Any: """ Get overall system status. """ # Count total devices total_devices = db.query(func.count(Device.id)).scalar() # Count active devices (seen in the last 24 hours) active_devices = db.query(func.count(Device.id)).filter( Device.last_seen >= datetime.now(timezone.utc) - timedelta(hours=24) ).scalar() # Count total telemetry entries total_telemetry = db.query(func.count(Telemetry.id)).scalar() # Count logs by level log_counts = db.query( Log.level, func.count(Log.id) ).group_by(Log.level).all() log_stats = {level.value: 0 for level in LogLevel} for level, count in log_counts: log_stats[level.value] = count # Get latest telemetry timestamp latest_telemetry = db.query(func.max(Telemetry.timestamp)).scalar() return { "devices": { "total": total_devices, "active": active_devices, }, "telemetry": { "total": total_telemetry, "latest": latest_telemetry, }, "logs": log_stats, "system_time": datetime.now(timezone.utc), } @router.get("/devices") def get_devices_status( db: Session = Depends(get_db), ) -> Any: """ Get status summary for all devices. """ # Get all devices with their latest status devices = db.query(Device).all() result = [] for device in devices: # Determine status based on last_seen status = "unknown" if device.last_seen: # Handle both timezone-aware and timezone-naive datetimes now = datetime.now(timezone.utc) last_seen = device.last_seen # If last_seen is timezone-naive, assume it's in UTC if last_seen.tzinfo is None: last_seen = last_seen.replace(tzinfo=timezone.utc) hours_since_last_seen = (now - last_seen).total_seconds() / 3600 if hours_since_last_seen < 1: status = "online" elif hours_since_last_seen < 24: status = "idle" else: status = "offline" # Get battery status battery_status = "unknown" if device.battery_level is not None: if device.battery_level < 20: battery_status = "critical" elif device.battery_level < 50: battery_status = "low" else: battery_status = "good" # Get signal status signal_status = "unknown" if device.signal_strength is not None: if device.signal_strength < -100: signal_status = "poor" elif device.signal_strength < -80: signal_status = "fair" else: signal_status = "good" # Get latest telemetry data for sensor readings latest_telemetry = db.query(Telemetry).filter( Telemetry.device_id == device.id ).order_by(Telemetry.timestamp.desc()).first() # Extract sensor data from latest telemetry sensors = {} if latest_telemetry: sensors = { "temperature": latest_telemetry.temperature, "humidity": latest_telemetry.humidity, "solar_voltage": latest_telemetry.solar_voltage, "battery_voltage": latest_telemetry.battery_voltage, } else: sensors = { "temperature": None, "humidity": None, "solar_voltage": None, "battery_voltage": None, } result.append({ "id": device.id, "name": device.name, "status": status, "battery": { "level": device.battery_level, "voltage": sensors["battery_voltage"], "status": battery_status, }, "signal": { "strength": device.signal_strength, "status": signal_status, }, "sensors": { "dht11": { "temperature": sensors["temperature"], "humidity": sensors["humidity"], }, "solar": { "voltage": sensors["solar_voltage"], }, "battery": { "voltage": sensors["battery_voltage"], } }, "location": { "latitude": device.latitude, "longitude": device.longitude, "altitude": device.altitude, }, "last_seen": device.last_seen, }) return result