88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
from rest_framework import serializers
|
|
from drf_spectacular.utils import extend_schema_field
|
|
from drf_spectacular.types import OpenApiTypes
|
|
from typing import Dict, Any
|
|
from sitemanagement.models import Sensor, Alert
|
|
|
|
class NotificationSerializer(serializers.ModelSerializer):
|
|
type = serializers.SerializerMethodField()
|
|
priority = serializers.SerializerMethodField()
|
|
timestamp = serializers.DateTimeField(source='created_at')
|
|
acknowledged = serializers.BooleanField(source='resolved', default=False)
|
|
|
|
class Meta:
|
|
model = Alert
|
|
fields = ('id', 'type', 'timestamp', 'acknowledged', 'priority')
|
|
|
|
def get_type(self, obj):
|
|
return 'critical' if obj.severity == 'critical' else 'warning'
|
|
|
|
def get_priority(self, obj):
|
|
return 'high' if obj.severity == 'critical' else 'medium'
|
|
|
|
class DetectorSerializer(serializers.ModelSerializer):
|
|
detector_id = serializers.SerializerMethodField()
|
|
type = serializers.SerializerMethodField()
|
|
name = serializers.SerializerMethodField()
|
|
object = serializers.SerializerMethodField()
|
|
status = serializers.SerializerMethodField()
|
|
zone = serializers.SerializerMethodField()
|
|
floor = serializers.SerializerMethodField()
|
|
notifications = NotificationSerializer(source='alerts', many=True)
|
|
|
|
class Meta:
|
|
model = Sensor
|
|
fields = ('detector_id', 'type', 'serial_number', 'name', 'object', 'status', 'zone', 'floor', 'notifications')
|
|
|
|
def get_detector_id(self, obj):
|
|
return obj.name or f"{obj.sensor_type.code}-{obj.id}"
|
|
|
|
def get_type(self, obj):
|
|
# маппинг типов сенсоров на нужные значения
|
|
sensor_type_mapping = {
|
|
'GA': 'fire_detector',
|
|
'PE': 'pressure_sensor',
|
|
'GLE': 'gas_detector'
|
|
}
|
|
return sensor_type_mapping.get(obj.sensor_type.code, 'unknown')
|
|
|
|
def get_name(self, obj):
|
|
return f"{obj.sensor_type.name} {obj.serial_number}"
|
|
|
|
def get_object(self, obj):
|
|
# получаем первую зону датчика и её объект
|
|
zone = obj.zones.first()
|
|
if zone:
|
|
return zone.object.title
|
|
return None
|
|
|
|
def get_status(self, obj):
|
|
# проверяем наличие активных алертов
|
|
latest_alert = obj.alerts.filter(resolved=False).first()
|
|
if latest_alert:
|
|
return latest_alert.severity # вернет 'warning' или 'critical'
|
|
return 'normal'
|
|
|
|
def get_zone(self, obj):
|
|
zone = obj.zones.first()
|
|
if zone:
|
|
return zone.name
|
|
return None
|
|
|
|
def get_floor(self, obj):
|
|
# получаем этаж из зоны
|
|
zone = obj.zones.first()
|
|
if zone:
|
|
return zone.floor
|
|
return None # если датчик не привязали к зоне
|
|
|
|
class DetectorsResponseSerializer(serializers.Serializer):
|
|
detectors = serializers.SerializerMethodField()
|
|
|
|
@extend_schema_field(OpenApiTypes.OBJECT)
|
|
def get_detectors(self, sensors) -> Dict[str, Any]:
|
|
detector_serializer = DetectorSerializer(sensors, many=True)
|
|
return {
|
|
sensor['detector_id']: sensor
|
|
for sensor in detector_serializer.data
|
|
} |