pmc
7d05339e05
All checks were successful
Build and Deploy LTY / build-and-deploy (push) Successful in 54m40s
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
118 lines
4.9 KiB
Python
118 lines
4.9 KiB
Python
import time
|
||
from rest_framework import serializers
|
||
from django.core.cache import cache
|
||
from django.utils import timezone
|
||
from .models import DeviceType, DeviceBatch, Device, UserDevice
|
||
|
||
|
||
class DeviceTypeSerializer(serializers.ModelSerializer):
|
||
class Meta:
|
||
model = DeviceType
|
||
fields = '__all__'
|
||
|
||
|
||
class DeviceBatchSerializer(serializers.ModelSerializer):
|
||
device_type_name = serializers.ReadOnlyField(source='device_type.name')
|
||
|
||
class Meta:
|
||
model = DeviceBatch
|
||
fields = '__all__'
|
||
|
||
|
||
class DeviceSerializer(serializers.ModelSerializer):
|
||
device_type_name = serializers.ReadOnlyField(source='device_type.name')
|
||
batch_number = serializers.ReadOnlyField(source='batch.batch_number')
|
||
|
||
class Meta:
|
||
model = Device
|
||
fields = '__all__'
|
||
read_only_fields = ('device_code',)
|
||
|
||
|
||
class DeviceCreateSerializer(serializers.ModelSerializer):
|
||
firmware_version = serializers.CharField(max_length=50, required=False)
|
||
wifi_name = serializers.CharField(max_length=100, required=False)
|
||
wifi_password = serializers.CharField(max_length=100, required=False)
|
||
brightness = serializers.IntegerField(min_value=0, max_value=100, required=False)
|
||
|
||
class Meta:
|
||
model = Device
|
||
fields = [
|
||
'device_type', 'batch', 'serial_number', 'mac_address',
|
||
'firmware_version', 'wifi_name', 'wifi_password', 'brightness'
|
||
]
|
||
|
||
|
||
class DeviceBatchCreateSerializer(serializers.Serializer):
|
||
device_type = serializers.PrimaryKeyRelatedField(queryset=DeviceType.objects.all())
|
||
batch = serializers.PrimaryKeyRelatedField(queryset=DeviceBatch.objects.all())
|
||
start_serial = serializers.CharField(max_length=50)
|
||
count = serializers.IntegerField(min_value=1, max_value=1000)
|
||
mac_prefix = serializers.CharField(max_length=9, required=False) # 可选，MAC地址前缀
|
||
|
||
def validate(self, data):
|
||
# 检查批次数量是否足够
|
||
batch = data['batch']
|
||
if batch.device_type.id != data['device_type'].id:
|
||
raise serializers.ValidationError("所选批次不属于所选设备类型")
|
||
|
||
# 检查设备数量是否足够
|
||
existing_devices = Device.objects.filter(batch=batch).count()
|
||
if existing_devices + data['count'] > batch.quantity:
|
||
raise serializers.ValidationError(f"批次总量为{batch.quantity}，已创建{existing_devices}台，无法再创建{data['count']}台")
|
||
|
||
return data
|
||
|
||
|
||
class UserDeviceSerializer(serializers.ModelSerializer):
|
||
device_code = serializers.ReadOnlyField(source='device.device_code')
|
||
mac_address = serializers.ReadOnlyField(source='device.mac_address')
|
||
device_type = serializers.ReadOnlyField(source='device.device_type.name')
|
||
device_status = serializers.ReadOnlyField(source='device.status')
|
||
is_online = serializers.SerializerMethodField()
|
||
battery_level = serializers.ReadOnlyField(source='device.battery_level')
|
||
firmware_version = serializers.ReadOnlyField(source='device.firmware_version')
|
||
wifi_name = serializers.ReadOnlyField(source='device.wifi_name')
|
||
wifi_password = serializers.ReadOnlyField(source='device.wifi_password')
|
||
brightness = serializers.ReadOnlyField(source='device.brightness')
|
||
|
||
def get_is_online(self, obj):
|
||
"""根据Redis中的last_seen判断设备是否在线（3分钟内有上报）"""
|
||
last_seen = cache.get(f"device:last_seen:{obj.device.mac_address}")
|
||
if last_seen is None:
|
||
return False
|
||
return (time.time() - float(last_seen)) <= 180
|
||
|
||
class Meta:
|
||
model = UserDevice
|
||
fields = '__all__'
|
||
read_only_fields = ('user', 'bound_at')
|
||
|
||
|
||
class DeviceRegisterSerializer(serializers.Serializer):
|
||
mac_address = serializers.CharField(max_length=17)
|
||
device_type_code = serializers.CharField(max_length=10, required=False, default='T01')
|
||
firmware_version = serializers.CharField(max_length=50, required=False, default='')
|
||
|
||
def validate_mac_address(self, value):
|
||
if Device.objects.filter(mac_address=value).exists():
|
||
raise serializers.ValidationError("设备已注册")
|
||
return value
|
||
|
||
def validate_device_type_code(self, value):
|
||
if not DeviceType.objects.filter(code=value).exists():
|
||
raise serializers.ValidationError(f"设备类型 {value} 不存在")
|
||
return value
|
||
|
||
|
||
class DeviceBindSerializer(serializers.Serializer):
|
||
mac_address = serializers.CharField(max_length=17)
|
||
nickname = serializers.CharField(max_length=100, required=False)
|
||
is_primary = serializers.BooleanField(default=False, required=False)
|
||
|
||
def validate_mac_address(self, value):
|
||
try:
|
||
Device.objects.get(mac_address=value)
|
||
except Device.DoesNotExist:
|
||
raise serializers.ValidationError("设备不存在")
|
||
return value |