518 lines
21 KiB
Python
518 lines
21 KiB
Python
from rest_framework import serializers
|
|
from django.contrib.auth.password_validation import validate_password
|
|
from .models import ComPortReading, User, Report, ScaleProtocol
|
|
from vehicles.models import Vehicle, VehicleExtra, ExtraCombination
|
|
from nomenclatures.models import (
|
|
Nomenclature, NomenclatureField, NomenclatureEntry, SelectorOption,
|
|
)
|
|
from nomenclatures.visibility import build_reveal_map, is_visible
|
|
|
|
|
|
class UserSerializer(serializers.ModelSerializer):
|
|
password = serializers.CharField(write_only=True, required=False)
|
|
|
|
class Meta:
|
|
model = User
|
|
fields = ['id', 'username', 'email', 'first_name', 'last_name',
|
|
'role', 'is_admin', 'is_active', 'date_joined', 'password',
|
|
'can_measure', 'can_manually_measure', 'can_manage_entities', 'can_edit_documents']
|
|
read_only_fields = ['id', 'date_joined']
|
|
extra_kwargs = {
|
|
'password': {'write_only': True}
|
|
}
|
|
|
|
def create(self, validated_data):
|
|
password = validated_data.pop('password', None)
|
|
user = User(**validated_data)
|
|
if password:
|
|
user.set_password(password)
|
|
user.save()
|
|
return user
|
|
|
|
def update(self, instance, validated_data):
|
|
password = validated_data.pop('password', None)
|
|
for attr, value in validated_data.items():
|
|
setattr(instance, attr, value)
|
|
if password:
|
|
instance.set_password(password)
|
|
instance.save()
|
|
return instance
|
|
|
|
|
|
class UserDetailSerializer(serializers.ModelSerializer):
|
|
"""Serializer for current user details (excludes password)"""
|
|
class Meta:
|
|
model = User
|
|
fields = ['id', 'username', 'email', 'first_name', 'last_name',
|
|
'role', 'is_admin', 'is_active', 'date_joined',
|
|
'can_measure', 'can_manually_measure', 'can_manage_entities', 'can_edit_documents']
|
|
read_only_fields = ['id', 'date_joined']
|
|
|
|
|
|
class ChangePasswordSerializer(serializers.Serializer):
|
|
"""Serializer for password change endpoint"""
|
|
old_password = serializers.CharField(required=True, write_only=True)
|
|
new_password = serializers.CharField(required=True, write_only=True)
|
|
|
|
def validate_old_password(self, value):
|
|
user = self.context['request'].user
|
|
if not user.check_password(value):
|
|
raise serializers.ValidationError("Old password is incorrect")
|
|
return value
|
|
|
|
def validate_new_password(self, value):
|
|
# Use Django's password validators
|
|
validate_password(value, self.context['request'].user)
|
|
return value
|
|
|
|
def save(self, **kwargs):
|
|
user = self.context['request'].user
|
|
user.set_password(self.validated_data['new_password'])
|
|
user.save()
|
|
return user
|
|
|
|
|
|
class ComPortReadingSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = ComPortReading
|
|
fields = ['id', 'port', 'data', 'timestamp', 'source_ip']
|
|
read_only_fields = ['id', 'timestamp']
|
|
|
|
|
|
def validate_vehicle_extra_data(value, lookup_only=False):
|
|
"""Validate a {nomenclature_code: value} dict against vehicle nomenclatures.
|
|
|
|
lookup_only=True rejects any key whose nomenclature is not kind='lookup'
|
|
or kind='selector' (the projection stored as a reusable combination).
|
|
"""
|
|
if not value:
|
|
return value
|
|
|
|
nomenclatures = {
|
|
n.code: n for n in
|
|
Nomenclature.objects.filter(applies_to='vehicle').prefetch_related(
|
|
'fields', 'entries', 'selector_options__reveals'
|
|
)
|
|
}
|
|
reveal_map = build_reveal_map()
|
|
|
|
errors = {}
|
|
for key, val in value.items():
|
|
if key not in nomenclatures:
|
|
errors[key] = f"Unknown nomenclature code '{key}'."
|
|
continue
|
|
|
|
nom = nomenclatures[key]
|
|
|
|
# A value for a nomenclature the current selector state hides is a
|
|
# contradiction: this is what actually enforces the rule, not the UI.
|
|
if val is not None and not is_visible(key, value, reveal_map):
|
|
errors[key] = (
|
|
f"'{key}' is not applicable for the current selector value."
|
|
)
|
|
continue
|
|
|
|
if nom.kind == Nomenclature.SELECTOR:
|
|
allowed = {o.value for o in nom.selector_options.all()}
|
|
if val is not None and val not in allowed:
|
|
errors[key] = f"Must be one of {sorted(allowed)}."
|
|
|
|
elif nom.kind == Nomenclature.LOOKUP:
|
|
if not isinstance(val, int):
|
|
errors[key] = "Must be a nomenclature entry ID (integer)."
|
|
elif not nom.entries.filter(id=val, is_active=True).exists():
|
|
errors[key] = f"Entry ID {val} not found in '{nom.code}'."
|
|
|
|
elif nom.kind == Nomenclature.FIELD:
|
|
if lookup_only:
|
|
errors[key] = "Only lookup and selector nomenclatures are allowed here."
|
|
continue
|
|
field_def = nom.fields.first()
|
|
if field_def:
|
|
if field_def.field_type == NomenclatureField.TEXT and not isinstance(val, str):
|
|
errors[key] = "Must be a string."
|
|
elif field_def.field_type == NomenclatureField.NUMBER and not isinstance(val, (int, float)):
|
|
errors[key] = "Must be a number."
|
|
elif field_def.field_type == NomenclatureField.BOOL and not isinstance(val, bool):
|
|
errors[key] = "Must be a boolean."
|
|
elif field_def.field_type == NomenclatureField.CHOICE and val not in (field_def.choices or []):
|
|
errors[key] = f"Must be one of {field_def.choices}."
|
|
|
|
if errors:
|
|
raise serializers.ValidationError(errors)
|
|
|
|
return value
|
|
|
|
|
|
class VehicleExtraSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = VehicleExtra
|
|
fields = ['id', 'data']
|
|
read_only_fields = ['id']
|
|
|
|
def validate_data(self, value):
|
|
return validate_vehicle_extra_data(value)
|
|
|
|
|
|
class VehicleSerializer(serializers.ModelSerializer):
|
|
extra = VehicleExtraSerializer(required=False, allow_null=True)
|
|
tare_user_name = serializers.CharField(source='tare_user.username', read_only=True, default=None)
|
|
gross_user_name = serializers.CharField(source='gross_user.username', read_only=True, default=None)
|
|
net_user_name = serializers.CharField(source='net_user.username', read_only=True, default=None)
|
|
void_user_name = serializers.CharField(source='void_user.username', read_only=True, default=None)
|
|
|
|
class Meta:
|
|
model = Vehicle
|
|
fields = ['id', 'vehicle_number', 'trailer1_number', 'trailer2_number', 'driver_pid', 'driver_name',
|
|
'tare', 'tare_date', 'tare_user', 'tare_user_name',
|
|
'tare_manual', 'tare_manual_user', 'tare_manual_date', 'tare_manual_note',
|
|
'gross', 'gross_date', 'gross_user', 'gross_user_name',
|
|
'gross_manual', 'gross_manual_user', 'gross_manual_date', 'gross_manual_note',
|
|
'net', 'net_date', 'net_user', 'net_user_name',
|
|
'doc_number', 'extra',
|
|
'is_void', 'void_date', 'void_user', 'void_user_name', 'void_reason']
|
|
read_only_fields = ['id', 'tare_date', 'tare_user', 'tare_user_name',
|
|
'tare_manual', 'tare_manual_user', 'tare_manual_date', 'tare_manual_note',
|
|
'gross_date', 'gross_user', 'gross_user_name',
|
|
'gross_manual', 'gross_manual_user', 'gross_manual_date', 'gross_manual_note',
|
|
'net', 'net_date', 'net_user', 'net_user_name',
|
|
'doc_number',
|
|
'is_void', 'void_date', 'void_user', 'void_user_name', 'void_reason']
|
|
|
|
def create(self, validated_data):
|
|
extra_data = validated_data.pop('extra', None)
|
|
vehicle = Vehicle.objects.create(**validated_data)
|
|
|
|
if extra_data:
|
|
VehicleExtra.objects.create(vehicle=vehicle, **extra_data)
|
|
|
|
return vehicle
|
|
|
|
def update(self, instance, validated_data):
|
|
extra_data = validated_data.pop('extra', None)
|
|
|
|
for attr, value in validated_data.items():
|
|
setattr(instance, attr, value)
|
|
instance.save()
|
|
|
|
# Handle VehicleExtra update/creation
|
|
if extra_data is not None:
|
|
if hasattr(instance, 'extra'):
|
|
for attr, value in extra_data.items():
|
|
setattr(instance.extra, attr, value)
|
|
instance.extra.save()
|
|
else:
|
|
VehicleExtra.objects.create(vehicle=instance, **extra_data)
|
|
|
|
return instance
|
|
|
|
def validate_vehicle_number(self, value):
|
|
if not value or not value.strip():
|
|
raise serializers.ValidationError("Vehicle number cannot be empty")
|
|
return value.strip()
|
|
|
|
def validate(self, attrs):
|
|
if self.instance is None and not attrs.get('driver_name'):
|
|
raise serializers.ValidationError({'driver_name': 'This field is required.'})
|
|
return attrs
|
|
|
|
|
|
class ExtraCombinationSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = ExtraCombination
|
|
fields = ['id', 'data', 'use_count', 'last_used']
|
|
read_only_fields = ['id', 'use_count', 'last_used']
|
|
|
|
def validate_data(self, value):
|
|
if not value:
|
|
raise serializers.ValidationError("Combination cannot be empty.")
|
|
return validate_vehicle_extra_data(value, lookup_only=True)
|
|
|
|
|
|
class ScaleProtocolSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = ScaleProtocol
|
|
fields = ['id', 'name', 'kind', 'poll_command', 'poll_interval',
|
|
'config', 'is_builtin', 'is_active']
|
|
read_only_fields = ['id', 'is_builtin']
|
|
|
|
def validate_poll_interval(self, value):
|
|
if value <= 0:
|
|
raise serializers.ValidationError("Poll interval must be greater than zero.")
|
|
return value
|
|
|
|
def validate(self, attrs):
|
|
kind = attrs.get('kind', getattr(self.instance, 'kind', None))
|
|
config = attrs.get('config', getattr(self.instance, 'config', None) or {})
|
|
|
|
if kind == ScaleProtocol.FIXED_COLUMN:
|
|
# A misconfigured offset silently clips a digit off every weight, so
|
|
# reject anything unusable rather than storing it.
|
|
errors = {}
|
|
for key in ('weight_start', 'weight_length', 'stable_pos'):
|
|
value = config.get(key)
|
|
if not isinstance(value, int) or isinstance(value, bool) or value < 0:
|
|
errors[key] = "Must be a non-negative integer."
|
|
if not isinstance(config.get('weight_length'), bool) and isinstance(config.get('weight_length'), int) and config['weight_length'] < 1:
|
|
errors['weight_length'] = "Must be at least 1."
|
|
stable_char = config.get('stable_char', ' ')
|
|
if not isinstance(stable_char, str) or len(stable_char) != 1:
|
|
errors['stable_char'] = "Must be exactly one character."
|
|
if errors:
|
|
raise serializers.ValidationError({'config': errors})
|
|
|
|
return attrs
|
|
|
|
|
|
class NomenclatureFieldSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = NomenclatureField
|
|
fields = ['id', 'key', 'label', 'field_type', 'required', 'choices', 'sort_order']
|
|
read_only_fields = ['id']
|
|
|
|
def validate_key(self, value):
|
|
if not value or not value.strip():
|
|
raise serializers.ValidationError("Field key cannot be empty")
|
|
return value.strip()
|
|
|
|
def validate_field_type(self, value):
|
|
valid_types = [choice[0] for choice in NomenclatureField.FIELD_TYPES]
|
|
if value not in valid_types:
|
|
raise serializers.ValidationError(
|
|
f"Invalid field type. Must be one of: {', '.join(valid_types)}"
|
|
)
|
|
return value
|
|
|
|
def validate(self, data):
|
|
if data.get('field_type') == NomenclatureField.CHOICE:
|
|
choices = data.get('choices')
|
|
if not choices or not isinstance(choices, list) or len(choices) == 0:
|
|
raise serializers.ValidationError({
|
|
'choices': 'Choices must be a non-empty list for choice fields.'
|
|
})
|
|
elif data.get('choices'):
|
|
raise serializers.ValidationError({
|
|
'choices': 'Only choice-type fields can have choices.'
|
|
})
|
|
return data
|
|
|
|
|
|
class NomenclatureEntrySerializer(serializers.ModelSerializer):
|
|
display_value = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = NomenclatureEntry
|
|
fields = ['id', 'data', 'is_active', 'display_value']
|
|
read_only_fields = ['id']
|
|
|
|
def get_display_value(self, obj):
|
|
display_key = obj.nomenclature.display_field or "name"
|
|
return obj.data.get(display_key, f"Entry #{obj.pk}")
|
|
|
|
def validate_data(self, value):
|
|
nomenclature = self.context.get('nomenclature')
|
|
if not nomenclature:
|
|
return value
|
|
|
|
field_defs = {f.key: f for f in nomenclature.fields.all()}
|
|
|
|
for key in value.keys():
|
|
if key not in field_defs:
|
|
raise serializers.ValidationError(
|
|
f"Unknown field '{key}'. Valid fields: {list(field_defs.keys())}"
|
|
)
|
|
|
|
for key, field_def in field_defs.items():
|
|
if field_def.required and key not in value:
|
|
raise serializers.ValidationError(
|
|
f"Required field '{key}' is missing."
|
|
)
|
|
|
|
for key, val in value.items():
|
|
if key not in field_defs:
|
|
continue
|
|
field_def = field_defs[key]
|
|
|
|
if field_def.field_type == NomenclatureField.TEXT and not isinstance(val, str):
|
|
raise serializers.ValidationError(f"Field '{key}' must be a string.")
|
|
elif field_def.field_type == NomenclatureField.NUMBER and not isinstance(val, (int, float)):
|
|
raise serializers.ValidationError(f"Field '{key}' must be a number.")
|
|
elif field_def.field_type == NomenclatureField.BOOL and not isinstance(val, bool):
|
|
raise serializers.ValidationError(f"Field '{key}' must be a boolean.")
|
|
elif field_def.field_type == NomenclatureField.CHOICE and val not in (field_def.choices or []):
|
|
raise serializers.ValidationError(f"Field '{key}' must be one of {field_def.choices}.")
|
|
|
|
return value
|
|
|
|
|
|
class SelectorOptionSerializer(serializers.ModelSerializer):
|
|
"""One selector choice; `reveals` is a list of nomenclature codes."""
|
|
|
|
reveals = serializers.SlugRelatedField(
|
|
many=True,
|
|
slug_field='code',
|
|
queryset=Nomenclature.objects.all(),
|
|
required=False,
|
|
)
|
|
|
|
class Meta:
|
|
model = SelectorOption
|
|
fields = ['id', 'value', 'label', 'reveals', 'sort_order']
|
|
read_only_fields = ['id']
|
|
|
|
def validate_value(self, value):
|
|
if not value or not value.strip():
|
|
raise serializers.ValidationError("Option value cannot be empty")
|
|
return value.strip()
|
|
|
|
|
|
class NomenclatureSerializer(serializers.ModelSerializer):
|
|
fields = NomenclatureFieldSerializer(many=True, required=False)
|
|
selector_options = SelectorOptionSerializer(many=True, required=False)
|
|
entry_count = serializers.IntegerField(source='entries.count', read_only=True)
|
|
|
|
class Meta:
|
|
model = Nomenclature
|
|
fields = ['id', 'code', 'name', 'applies_to', 'kind', 'display_field',
|
|
'sort_order', 'fields', 'selector_options', 'entry_count']
|
|
read_only_fields = ['id']
|
|
|
|
def create(self, validated_data):
|
|
fields_data = validated_data.pop('fields', [])
|
|
options_data = validated_data.pop('selector_options', [])
|
|
nomenclature = Nomenclature.objects.create(**validated_data)
|
|
|
|
for field_data in fields_data:
|
|
NomenclatureField.objects.create(nomenclature=nomenclature, **field_data)
|
|
|
|
self._write_options(nomenclature, options_data)
|
|
|
|
return nomenclature
|
|
|
|
def update(self, instance, validated_data):
|
|
fields_data = validated_data.pop('fields', None)
|
|
options_data = validated_data.pop('selector_options', None)
|
|
|
|
for attr, value in validated_data.items():
|
|
setattr(instance, attr, value)
|
|
instance.save()
|
|
|
|
if fields_data is not None:
|
|
instance.fields.all().delete()
|
|
for field_data in fields_data:
|
|
NomenclatureField.objects.create(nomenclature=instance, **field_data)
|
|
|
|
if options_data is not None:
|
|
instance.selector_options.all().delete()
|
|
self._write_options(instance, options_data)
|
|
|
|
return instance
|
|
|
|
@staticmethod
|
|
def _write_options(nomenclature, options_data):
|
|
"""Recreate the selector's options; `reveals` is m2m so it is set after save."""
|
|
for option_data in options_data:
|
|
reveals = option_data.pop('reveals', [])
|
|
option = SelectorOption.objects.create(selector=nomenclature, **option_data)
|
|
option.reveals.set(reveals)
|
|
|
|
def validate_code(self, value):
|
|
if not value or not value.strip():
|
|
raise serializers.ValidationError("Code cannot be empty")
|
|
return value.strip()
|
|
|
|
def validate_applies_to(self, value):
|
|
valid_choices = ['vehicle', 'container']
|
|
if value not in valid_choices:
|
|
raise serializers.ValidationError(
|
|
f"Invalid applies_to value. Must be one of: {', '.join(valid_choices)}"
|
|
)
|
|
return value
|
|
|
|
def validate(self, data):
|
|
kind = data.get('kind', getattr(self.instance, 'kind', Nomenclature.LOOKUP))
|
|
fields_data = data.get('fields', [])
|
|
options_data = data.get('selector_options')
|
|
|
|
if kind == Nomenclature.SELECTOR:
|
|
self._validate_selector_options(options_data or [])
|
|
elif options_data:
|
|
raise serializers.ValidationError({
|
|
'selector_options': "Only a 'selector' nomenclature can have options."
|
|
})
|
|
|
|
if kind == Nomenclature.FIELD and fields_data and len(fields_data) != 1:
|
|
raise serializers.ValidationError(
|
|
"A 'field' nomenclature must have exactly one field definition."
|
|
)
|
|
|
|
if kind == Nomenclature.LOOKUP:
|
|
display_field = data.get('display_field', getattr(self.instance, 'display_field', 'name'))
|
|
field_keys = [f['key'] for f in fields_data] if fields_data else []
|
|
if field_keys and display_field not in field_keys:
|
|
raise serializers.ValidationError(
|
|
f"display_field '{display_field}' must match one of the defined field keys."
|
|
)
|
|
|
|
return data
|
|
|
|
def _validate_selector_options(self, options_data):
|
|
"""No nesting, and no nomenclature revealed by two different selectors."""
|
|
errors = []
|
|
seen_values = set()
|
|
|
|
for option in options_data:
|
|
value = option.get('value')
|
|
if value in seen_values:
|
|
errors.append(f"Duplicate option value '{value}'.")
|
|
seen_values.add(value)
|
|
|
|
for revealed in option.get('reveals', []):
|
|
if revealed.kind == Nomenclature.SELECTOR:
|
|
errors.append(
|
|
f"'{revealed.code}' is a selector; selectors cannot reveal "
|
|
f"other selectors."
|
|
)
|
|
continue
|
|
|
|
other = (
|
|
SelectorOption.objects
|
|
.filter(reveals=revealed)
|
|
.exclude(selector=self.instance)
|
|
.select_related('selector')
|
|
.first()
|
|
)
|
|
if other:
|
|
errors.append(
|
|
f"'{revealed.code}' is already revealed by selector "
|
|
f"'{other.selector.code}'."
|
|
)
|
|
|
|
if errors:
|
|
raise serializers.ValidationError({'selector_options': errors})
|
|
|
|
|
|
class ReportSerializer(serializers.ModelSerializer):
|
|
created_by_name = serializers.CharField(source='created_by.username', read_only=True, default=None)
|
|
|
|
class Meta:
|
|
model = Report
|
|
fields = ['id', 'name', 'kind', 'page_width', 'page_height', 'api_endpoint',
|
|
'elements', 'page_setup', 'created_by', 'created_by_name', 'created_at', 'updated_at']
|
|
read_only_fields = ['id', 'created_by', 'created_by_name', 'created_at', 'updated_at']
|
|
|
|
def validate_name(self, value):
|
|
if not value or not value.strip():
|
|
raise serializers.ValidationError("Report name cannot be empty")
|
|
return value.strip()
|
|
|
|
def validate_elements(self, value):
|
|
if not isinstance(value, list):
|
|
raise serializers.ValidationError("Elements must be a list")
|
|
return value
|
|
|
|
def validate_page_setup(self, value):
|
|
if not isinstance(value, dict):
|
|
raise serializers.ValidationError("page_setup must be an object")
|
|
return value
|