Spaces:
Running
Running
File size: 5,327 Bytes
04dc214 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | """
Device Monitoring Service
Handles device connection monitoring, status tracking, and health checks
"""
from typing import Dict, List, Optional
from datetime import datetime
from src.models.edge_device import EdgeDevice
class DeviceMonitoringService:
"""
Service class to handle device monitoring, connection status tracking,
and health checks for connected physical AI devices.
"""
def __init__(self):
# In-memory storage for device statuses (in production, use database)
self._device_statuses: Dict[str, Dict] = {}
self._connection_history: Dict[str, List[Dict]] = {}
async def register_device(self, device: EdgeDevice) -> bool:
"""
Register a new device in the monitoring system
"""
device_id = device.id
self._device_statuses[device_id] = {
'status': 'registered',
'last_seen': datetime.utcnow().isoformat(),
'connection_attempts': 0,
'last_connection_attempt': None,
'health_score': 100
}
if device_id not in self._connection_history:
self._connection_history[device_id] = []
return True
async def update_device_status(self, device_id: str, status: str) -> bool:
"""
Update the status of a connected device
"""
if device_id not in self._device_statuses:
return False
self._device_statuses[device_id]['status'] = status
self._device_statuses[device_id]['last_seen'] = datetime.utcnow().isoformat()
# Record connection event
self._record_connection_event(device_id, status)
return True
async def get_device_status(self, device_id: str) -> Optional[Dict]:
"""
Get the current status of a specific device
"""
if device_id not in self._device_statuses:
return None
return self._device_statuses[device_id]
async def get_all_device_statuses(self) -> Dict[str, Dict]:
"""
Get statuses for all monitored devices
"""
return self._device_statuses.copy()
async def heartbeat_received(self, device_id: str) -> bool:
"""
Process a heartbeat signal from a device to confirm it's alive
"""
if device_id not in self._device_statuses:
return False
self._device_statuses[device_id]['status'] = 'online'
self._device_statuses[device_id]['last_seen'] = datetime.utcnow().isoformat()
self._device_statuses[device_id]['connection_attempts'] += 1
return True
async def device_disconnected(self, device_id: str) -> bool:
"""
Mark a device as disconnected
"""
if device_id not in self._device_statuses:
return False
self._device_statuses[device_id]['status'] = 'offline'
self._record_connection_event(device_id, 'disconnected')
return True
async def is_device_online(self, device_id: str) -> bool:
"""
Check if a specific device is currently online
"""
if device_id not in self._device_statuses:
return False
return self._device_statuses[device_id]['status'] == 'online'
async def get_connection_history(self, device_id: str, limit: int = 10) -> List[Dict]:
"""
Get connection history for a specific device
"""
if device_id not in self._connection_history:
return []
# Return the most recent events
return self._connection_history[device_id][-limit:]
async def check_device_health(self, device_id: str) -> Dict:
"""
Perform a health check on a specific device
"""
if device_id not in self._device_statuses:
return {'status': 'unknown', 'error': 'Device not found'}
status_info = self._device_statuses[device_id]
# Calculate health based on last seen time and status
health_score = status_info['health_score']
if status_info['status'] == 'offline':
health_score = max(0, health_score - 20)
elif status_info['status'] == 'error':
health_score = max(0, health_score - 30)
return {
'deviceId': device_id,
'status': status_info['status'],
'lastSeen': status_info['last_seen'],
'healthScore': health_score,
'connected': status_info['status'] == 'online',
'connectionAttempts': status_info['connection_attempts']
}
def _record_connection_event(self, device_id: str, event_type: str):
"""
Internal method to record connection events
"""
if device_id not in self._connection_history:
self._connection_history[device_id] = []
event = {
'timestamp': datetime.utcnow().isoformat(),
'eventType': event_type,
'deviceId': device_id
}
self._connection_history[device_id].append(event)
# Keep only the last 100 events to prevent memory issues
if len(self._connection_history[device_id]) > 100:
self._connection_history[device_id] = self._connection_history[device_id][-100:]
# Global instance of the service
device_monitor_service = DeviceMonitoringService() |