This commit is contained in:
2026-04-11 03:54:15 +03:30
parent 883573004c
commit 36d6b05a7f
68 changed files with 3487 additions and 841 deletions
+102 -1
View File
@@ -1,7 +1,16 @@
from collections import Counter
from copy import deepcopy
from farm_hub.models import FarmHub
from notifications.models import FarmNotification
from .models import FarmAlert
from .mock_data import (
ANOMALY_DETECTION_CARD,
ARM_ALERTS_TRACKER,
FARM_ALERTS_TIMELINE,
RECOMMENDATIONS_LIST,
)
from .models import AnomalyDetection, FarmAlert, Recommendation
class AlertService:
@@ -47,3 +56,95 @@ class AlertService:
level=level_map.get(alert.color, "info"),
metadata={"alert_uuid": str(alert.uuid), "color": alert.color},
)
def get_alert_tracker_data(farm=None):
if farm is None:
return deepcopy(ARM_ALERTS_TRACKER)
alerts = list(FarmAlert.objects.filter(farm=farm, is_active=True)[:20])
if not alerts:
return deepcopy(ARM_ALERTS_TRACKER)
counts = Counter(alert.title for alert in alerts)
alert_stats = []
for title, count in counts.most_common(3):
sample = next((alert for alert in alerts if alert.title == title), None)
alert_stats.append(
{
"title": title,
"count": str(count),
"avatarColor": sample.color if sample else "info",
"avatarIcon": sample.avatar_icon or "tabler-bell",
}
)
return {
"totalAlerts": len(alerts),
"radialBarValue": min(len(alerts) * 10, 100),
"alertStats": alert_stats,
}
def get_alert_timeline_data(farm=None):
if farm is None:
return deepcopy(FARM_ALERTS_TIMELINE)
alerts = list(FarmAlert.objects.filter(farm=farm)[:10])
if not alerts:
return deepcopy(FARM_ALERTS_TIMELINE)
return {
"alerts": [
{
"title": alert.title,
"description": alert.description,
"time": alert.created_at.strftime("%Y-%m-%d %H:%M"),
"color": alert.color,
}
for alert in alerts
]
}
def get_anomaly_detection_data(farm=None):
if farm is None:
return deepcopy(ANOMALY_DETECTION_CARD)
anomalies = list(AnomalyDetection.objects.filter(farm=farm)[:10])
if not anomalies:
return deepcopy(ANOMALY_DETECTION_CARD)
return {
"anomalies": [
{
"sensor": anomaly.sensor,
"value": anomaly.value,
"expected": anomaly.expected,
"deviation": anomaly.deviation,
"severity": anomaly.severity,
}
for anomaly in anomalies
]
}
def get_recommendations_list_data(farm=None):
if farm is None:
return deepcopy(RECOMMENDATIONS_LIST)
recommendations = list(Recommendation.objects.filter(farm=farm)[:10])
if not recommendations:
return deepcopy(RECOMMENDATIONS_LIST)
return {
"recommendations": [
{
"title": recommendation.title,
"subtitle": recommendation.subtitle,
"avatarIcon": recommendation.avatar_icon or "tabler-bulb",
"avatarColor": recommendation.avatar_color or "info",
}
for recommendation in recommendations
]
}