# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
import logging
import re
from typing import List
from uuid import UUID
from django.utils.translation import gettext as _
from rest_framework import serializers
from ferrosoft.apps.emiflow.models import (
TCEGeoOrigin,
TransportChainElement,
FreightItem,
TransportChain,
TCEGeoDestination,
Settings,
DistanceType,
OperationEpitype,
EmissionBookingRecord,
)
from ferrosoft.apps.emiflow.services.vehicle import detect_vehicle_type
from ferrosoft.apps.ferrobase.models import (
Material,
MeasurementUnit,
Country,
UnitCursor,
)
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.models.registry import ModelRegistry
from ferrosoft.apps.ferrobase.services.conversion import convert_unit
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_transaction
_UUID_PATTERN = re.compile(
r"^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$",
re.IGNORECASE,
)
def _is_uuid(value):
return re.search(_UUID_PATTERN, value) is not None
def _convert_measurement_unit(symbol: str) -> str:
ids = (
MeasurementUnit.objects
.filter(symbol=symbol)
.values_list("id", flat=True)
) # fmt: skip
if len(ids) == 0:
raise serializers.ValidationError(
{
"weight_unit": _("Unknown measurement unit: %s") % symbol,
}
)
return str(ids[0])
def _convert_country(needle: str | None) -> str:
if needle is None:
raise serializers.ValidationError(
{
"country": _("Country is required"),
}
)
if _is_uuid(needle):
return needle
ids = (
Country.objects
.filter(iso_code=needle.upper())
.values_list("id", flat=True)
) # fmt: skip
if len(ids) == 0:
raise serializers.ValidationError(
{
"country": _("Unknown country: %s") % needle,
}
)
return str(ids[0])
def _upsert_model(model_class, create_fields: dict):
update_fields = create_fields.copy()
pk = update_fields.pop("id", None)
if pk is None:
raise serializers.ValidationError(_("ID is required"))
model, *rest = model_class.objects.update_or_create(
pk=pk, defaults=update_fields, create_defaults=create_fields
)
return model
def _delete_orphan_models(
model_class, chain: TransportChain, instance_fields: list[dict]
):
model_ids = [field["id"] for field in instance_fields]
amount, _ = (
model_class.objects.filter(chain=chain).exclude(pk__in=model_ids).delete()
)
if amount > 0:
logging.info("deleted %d orphan %s", amount, model_class.__name__)
[docs]
class TCEGeoPointSerializer(serializers.ModelSerializer):
id = serializers.UUIDField(required=True)
"""Writable ID field for client generated IDs."""
[docs]
def to_internal_value(self, data):
return super().to_internal_value(
data | {"country": _convert_country(data.get("country"))}
)
[docs]
class ListTransportChainElementSerializer(serializers.ListSerializer):
[docs]
@staticmethod
def create_custom(
validated_data: dict | List[dict], chain_id: UUID, freight_items: List[dict]
):
if isinstance(validated_data, dict):
return TransportChainElementSerializer.create_custom(
validated_data, chain_id, freight_items
)
elif isinstance(validated_data, list):
return [
TransportChainElementSerializer.create_custom(
data, chain_id, freight_items
)
for data in validated_data
]
else:
raise ValueError("unexpected type %s" % type(validated_data))
[docs]
class TransportChainElementSerializer(serializers.ModelSerializer):
id = serializers.UUIDField(required=True)
"""Writable ID field for client generated IDs."""
geo_origin = TCEGeoPointSerializer(required=False)
geo_destination = TCEGeoPointSerializer(required=False)
# If the following fields are not supplied, TransportChainSerializer sets them.
# They are not set in this class, because it requires a field of transport chain.
operation = serializers.UUIDField(required=False)
"""If this field is not supplied, it is copied from TransportChainDefaults."""
activity_distance_type = serializers.ChoiceField(
choices=DistanceType.choices,
required=False,
)
"""If this field is not supplied, it is copied from TransportChainDefaults."""
activity_distance_value = serializers.DecimalField(
required=False,
max_digits=12,
decimal_places=4,
)
"""If not set, defaults to 0."""
activity_distance_unit = serializers.UUIDField(required=False)
"""If not set, defaults to KILOMETER."""
[docs]
def to_internal_value(self, data):
ad_unit = data.get("activity_distance_unit", None)
if isinstance(ad_unit, str) and not _is_uuid(ad_unit):
data["activity_distance_unit"] = _convert_measurement_unit(ad_unit)
return super().to_internal_value(data)
[docs]
@staticmethod
def create_custom(validated_data: dict, chain_id: UUID, freight_items: List[dict]):
geo_origin = validated_data.pop("geo_origin", None)
geo_destination = validated_data.pop("geo_destination", None)
# Set operation/category from default settings.
if (
validated_data.get("operation") is None
or validated_data.get("activity_distance_type") is None
or validated_data.get("activity_distance_value") is None
or validated_data.get("activity_distance_unit") is None
):
validated_data = _set_tce_fields_from_defaults(
validated_data, freight_items
)
element = _upsert_model(
TransportChainElement,
validated_data
| {
"chain_id": chain_id,
"epitype": OperationEpitype.TRANSPORT,
},
)
if isinstance(geo_origin, dict):
_upsert_model(TCEGeoOrigin, geo_origin | {"tce": element})
if isinstance(geo_destination, dict):
_upsert_model(TCEGeoDestination, geo_destination | {"tce": element})
return element
[docs]
class EmissionBookingRecordSerializer(serializers.ModelSerializer):
[docs]
class ListFreightItemSerializer(serializers.ListSerializer):
[docs]
@staticmethod
def create_custom(validated_data: dict | List[dict], chain_id: UUID):
if isinstance(validated_data, dict):
return FreightItemSerializer.create_custom(validated_data, chain_id)
elif isinstance(validated_data, list):
return [
FreightItemSerializer.create_custom(data, chain_id)
for data in validated_data
]
else:
raise ValueError("unexpected type %s" % type(validated_data))
[docs]
class FreightItemSerializer(serializers.ModelSerializer):
id = serializers.UUIDField(required=True)
"""Writable ID field for client generated IDs."""
_material_fields = [
"reference",
"name",
]
[docs]
def validate_material(self, value):
if isinstance(value, dict):
errors = []
for field in self._material_fields:
if not value.get(field):
errors.append(_("Material field '%s' is required.") % field)
if errors:
raise serializers.ValidationError(errors)
return value
[docs]
def to_internal_value(self, data):
# Accept material fields passed inline, to make it easy for clients to post freight items.
# They don't have to determine if material already exists.
material = data.get("material", None)
if isinstance(material, dict):
data["material"] = self._convert_inline_material(material)
# Convert weight_unit symbol to MeasurementUnit UUID, to make it easier for the client.
# Although, measurement unit is not created implicitly.
weight_unit = data.get("weight_unit", None)
if isinstance(weight_unit, str) and not _is_uuid(weight_unit):
data["weight_unit"] = _convert_measurement_unit(weight_unit)
return super().to_internal_value(data)
@staticmethod
def _convert_inline_material(material: dict) -> str:
material_ids = (
Material
.objects
.filter(code=material["reference"])
.values_list("id", flat=True)
) # fmt: skip
if len(material_ids) > 0:
return str(material_ids[0])
material = Material.objects.create(
code=material["reference"],
name=material["name"],
)
return str(material.pk)
[docs]
@staticmethod
def create_custom(validated_data: dict, chain_id: UUID):
# These fields were used in previous models, remove them for compatibility.
validated_data.pop("shipment", None)
validated_data.pop("consignment", None)
return _upsert_model(FreightItem, validated_data | {"chain_id": chain_id})
[docs]
class TCEPartialSerializer(serializers.Serializer):
chain_id = serializers.UUIDField()
element = TransportChainElementSerializer()
freight_item = FreightItemSerializer()
[docs]
def create(self, validated_data):
dat = validated_data
element = self.fields["element"].create_custom(
dat["element"], dat["chain_id"], [dat["freight_item"]]
)
freight_item = self.fields["freight_item"].create_custom(
dat["freight_item"], dat["chain_id"]
)
return {
"chain_id": dat["chain_id"],
"element": element,
"freight_item": freight_item,
}
[docs]
class TransportChainSerializer(serializers.ModelSerializer):
id = serializers.UUIDField(required=True)
"""Writable ID field for client generated IDs."""
elements = TransportChainElementSerializer(many=True)
freight_items = FreightItemSerializer(many=True)
[docs]
def create(self, validated_data):
"""Create or update transport chain along with related objects.
Objects are updated if they already exist with the given IDs."""
with tenant_db_transaction():
# As pop removes key as well so We use pop() instead of get() because these nested fields are
# explicitly handled below and should not be passed to the TransportChain constructor.
elements = validated_data.pop("elements", [])
freight_items = validated_data.pop("freight_items", [])
chain = _upsert_model(TransportChain, validated_data)
# Handle TransportChainElement + nested geo points
_delete_orphan_models(TransportChainElement, chain, elements)
for element_fields in elements:
self.fields["elements"].create_custom(
element_fields, chain.pk, freight_items
)
_delete_orphan_models(FreightItem, chain, freight_items)
for freight_fields in freight_items:
self.fields["freight_items"].create_custom(freight_fields, chain.pk)
return chain
def _set_tce_fields_from_defaults(fields: dict, freight_items: List[dict]) -> dict:
# Sum weight for vehicle type detection
weight = DecimalWithUnit.zero(UnitCursor("t"))
for item in freight_items:
weight += convert_unit(
DecimalWithUnit.create(
item.get("weight_value", 0),
item.get("weight_unit"),
),
UnitCursor("t"),
)
vehicle_type = detect_vehicle_type(weight)
defaults = Settings.get_instance().get_transport_chain_defaults(vehicle_type)
if not defaults.are_defaults_set:
raise RuntimeError(
_("Transport defaults are not defined for vehicle type %(name)s")
% {"name": vehicle_type.label}
)
return fields | {
"operation": defaults.transport_operation,
"activity_distance_type": defaults.distance_type,
"activity_distance_value": fields.get("activity_distance_type", "0"),
"activity_distance_unit": fields.get("activity_distance_unit")
or ModelRegistry.Unit.get(UnitCursor("km")),
}