38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from app.models.mongodb.device_log import DeviceLog
|
|
|
|
|
|
async def analyze_device_status(
|
|
device_id: str, start: datetime, end: datetime
|
|
) -> dict:
|
|
"""Analyze device status changes over a period."""
|
|
logs = await (
|
|
DeviceLog.find(
|
|
DeviceLog.device_id == device_id,
|
|
DeviceLog.event_type == "status_change",
|
|
DeviceLog.timestamp >= start,
|
|
DeviceLog.timestamp <= end,
|
|
)
|
|
.sort("+timestamp")
|
|
.to_list()
|
|
)
|
|
|
|
status_counts: dict[str, int] = {}
|
|
for log in logs:
|
|
status = log.payload.get("status", "unknown")
|
|
status_counts[status] = status_counts.get(status, 0) + 1
|
|
|
|
total_events = len(logs)
|
|
uptime_events = status_counts.get("online", 0)
|
|
uptime_ratio = uptime_events / total_events if total_events > 0 else 0.0
|
|
|
|
return {
|
|
"total_events": total_events,
|
|
"status_counts": status_counts,
|
|
"uptime_ratio": round(uptime_ratio, 4),
|
|
"period": {"start": start.isoformat(), "end": end.isoformat()},
|
|
}
|