# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
import logging
from decimal import Decimal
from rest_framework import serializers
from ferrosoft.apps.dataimport.convert import create_or_get_model
from ferrosoft.apps.emiflow.models import (
TransportChain,
BookingStatus,
OperationEpitype,
TransportChainDefaults,
DistanceType,
TransportChainType,
Settings,
FreightItem,
TransportChainElement,
TCEGeoDestination,
TCEGeoOrigin,
)
from ferrosoft.apps.emiflow.models.tegos import (
OrderLineGroup,
PostingType,
FreightItemGroup,
)
from ferrosoft.apps.emiflow.services.builder import (
TransportChainBuilder,
TransportChainElementBuilder,
FreightItemBuilder,
MaterialBuilder,
BulkCreator,
)
from ferrosoft.apps.emiflow.services.vehicle import detect_vehicle_type
from ferrosoft.apps.ferrobase.models import Material, UnitCategory
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.services.conversion import normalize_unit
from ferrosoft.apps.ferromaps.services.addresses import Address
from ferrosoft.decimal import zero
logger = logging.getLogger(__name__)
[docs]
class AddressSerializer(serializers.Serializer):
id = serializers.UUIDField()
name = serializers.CharField()
address = serializers.CharField()
address2 = serializers.CharField(allow_blank=True)
postCode = serializers.CharField()
city = serializers.CharField()
countryCode = serializers.CharField()
[docs]
class OrderLineSerializer(serializers.Serializer):
id = serializers.UUIDField()
reference = serializers.CharField()
orderNo = serializers.CharField()
lineNo = serializers.CharField()
postingType = serializers.CharField()
materialReference = serializers.CharField(allow_blank=True)
materialName = serializers.CharField(allow_blank=True)
quantityToDispose = serializers.DecimalField(max_digits=12, decimal_places=4)
unitOfMeasureCode = serializers.CharField()
taskSite = AddressSerializer(allow_null=True)
[docs]
class OrderSerializer(serializers.Serializer):
id = serializers.UUIDField()
reference = serializers.CharField()
orderDate = serializers.DateField()
company = AddressSerializer()
orderLines = OrderLineSerializer(many=True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.bulk = BulkCreator(
create_order=[
Material,
TransportChain,
TransportChainElement,
FreightItem,
TCEGeoDestination,
TCEGeoOrigin,
]
)
[docs]
def create(self, validated_data: dict) -> list[TransportChain]:
settings = Settings.get_instance()
groups = self.group_lines(validated_data)
if len(groups) == 0:
# Silently discard payloads where filtering leaves no usable lines.
# Well, almost silently.
logger.warning(
f"No usable lines found in order {validated_data['reference']}"
)
return []
# Ensure stable order of groups (makes tests easier).
group_keys = list(groups.keys())
group_keys.sort()
chains = []
for key in group_keys:
group = groups[key]
chain = self.create_chain(validated_data, key, group)
chains.append(chain)
freight_items: list[FreightItem] = self.create_freight_items(group, chain)
freight_weight = sum(
map(lambda item: item.weight, freight_items), DecimalWithUnit.zero()
)
vehicle_type = detect_vehicle_type(freight_weight)
defaults = settings.get_transport_chain_defaults(vehicle_type)
if not defaults.are_defaults_set:
raise RuntimeError(
"Transport defaults for vehicle type %(label)s are not set"
% vehicle_type.label
)
for line in group.lines:
self.create_chain_elements(
line, validated_data["company"], chain, defaults
)
self.bulk.upsert()
return chains
[docs]
@staticmethod
def group_lines(validated_data: dict):
# One Tegos order corresponds to one or more transport chains.
# Grouping key is posting type and material reference.
groups: dict[str, OrderLineGroup] = {}
for line in validated_data["orderLines"]:
# If quantity field is zero, this is not an operational order line,
# so not of interest.
if line["quantityToDispose"] == zero:
continue
try:
posting_type = PostingType(line["postingType"])
except ValueError:
# Lines with invalid posting type are likely garbage.
continue
key = reference_key(str(posting_type), line["materialReference"])
if key not in groups:
groups[key] = OrderLineGroup(
posting_type=posting_type,
material_reference=line["materialReference"],
lines=[],
)
groups[key].lines.append(line)
return groups
[docs]
@staticmethod
def create_chain(
order: dict, group_key: str, group: OrderLineGroup
) -> TransportChain:
def create():
reference = reference_key(order["reference"], group_key)
return (
TransportChainBuilder.new()
.status(BookingStatus.OPEN)
.reference(reference)
.booking_date(order["orderDate"])
.transport_type(group.posting_type.to_transport_type())
.simulation(False)
.save()
)
return create_or_get_model(TransportChain, create)
[docs]
def create_freight_items(
self, group: OrderLineGroup, chain: TransportChain
) -> list[FreightItem]:
# Sum weight by material reference. There should only be one material
# per group, however, don't rely on it.
freight_groups: dict[str, FreightItemGroup] = {}
for line in group.lines:
reference = line["materialReference"]
if reference not in freight_groups:
freight_groups[reference] = FreightItemGroup(
weight=DecimalWithUnit.zero(),
material_name=line["materialName"],
)
freight_groups[reference].weight += normalize_unit(
UnitCategory.MASS,
DecimalWithUnit.create(
line["quantityToDispose"],
line["unitOfMeasureCode"].lower(),
),
)
freight_items = []
for material_ref, data in freight_groups.items():
material = create_or_get_model(
Material, self.create_material(material_ref, data)
)
freight_items.append(
(
FreightItemBuilder.new()
.chain(chain)
.weight(data.weight)
.material(material)
.add_bulk(self.bulk)
)
)
return freight_items
[docs]
@staticmethod
def create_material(material_ref, data):
def create():
return (
MaterialBuilder.new().code(material_ref).name(data.material_name).save()
)
return create
[docs]
def create_chain_elements(
self,
line: dict,
company: dict,
chain: TransportChain,
defaults: TransportChainDefaults,
):
builder = (
TransportChainElementBuilder.new()
.chain(chain)
.epitype(OperationEpitype.TRANSPORT)
.position(line["lineNo"])
.operation(defaults.transport_operation)
.activity_distance_type(DistanceType(defaults.distance_type))
.activity_distance(DecimalWithUnit.create(0, "km"))
.distance_adjustment_factor(Decimal(1))
)
builder.add_geo_origin().address(
self.get_origin_address(
TransportChainType(chain.transport_type), line, company
)
)
builder.add_geo_destination().address(
self.get_destination_address(
TransportChainType(chain.transport_type), line, company
)
)
return builder.add_bulk(self.bulk)
[docs]
def get_origin_address(
self, transport_type: TransportChainType, line: dict, company: dict
) -> Address:
match transport_type:
case TransportChainType.INCOMING:
return self.convert_to_address(line["taskSite"])
case TransportChainType.OUTGOING:
return self.convert_to_address(company)
case _:
raise ValueError("Invalid posting type")
[docs]
def get_destination_address(
self, transport_type: TransportChainType, line: dict, company: dict
):
match transport_type:
case TransportChainType.INCOMING:
return self.convert_to_address(company)
case TransportChainType.OUTGOING:
return self.convert_to_address(line["taskSite"])
case _:
raise ValueError("Invalid posting type")
[docs]
@staticmethod
def convert_to_address(data: dict) -> Address:
return Address(
additional_lines=data["address2"],
city=data["city"],
company=data["name"],
country=data["countryCode"],
postal_code=data["postCode"],
street=data["address"],
)
[docs]
def reference_key(*parts: str) -> str:
return ":".join(parts)