90 lines
3.5 KiB
Python
90 lines
3.5 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()
|
|
detector_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', 'detector_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': 'Инклинометр',
|
|
'PE': 'Тензометр',
|
|
'GLE': 'Гидроуровень',
|
|
}
|
|
code = (getattr(obj.sensor_type, 'code', '') or '').upper()
|
|
return sensor_type_mapping.get(code, (getattr(obj.sensor_type, 'name', '') or ''))
|
|
|
|
def get_detector_type(self, obj):
|
|
return (getattr(obj.sensor_type, 'code', '') or '').upper()
|
|
|
|
def get_name(self, obj):
|
|
sensor_type = getattr(obj, 'sensor_type', None) or getattr(obj, 'sensor_type', None)
|
|
serial = getattr(obj, 'serial_number', '') or ''
|
|
base_name = getattr(obj, 'name', '') or ''
|
|
return base_name or f"{getattr(obj.sensor_type, 'code', '')}-{serial}".strip('-')
|
|
|
|
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):
|
|
first_zone = obj.zones.first()
|
|
return first_zone.name if first_zone else None
|
|
|
|
def get_floor(self, obj):
|
|
first_zone = obj.zones.first()
|
|
return getattr(first_zone, 'floor', 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
|
|
} |