# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
import logging
from decimal import Decimal
from typing import Callable, Type
from asgiref.sync import sync_to_async
from django.utils.translation import gettext as _
from ferrosoft.apps.dataimport.convert import create_or_get_model
from ferrosoft.apps.dataimport.models import LogHighlight
from ferrosoft.apps.dataimport.queue import ModelProcessorArgs, ProcessorArgs
from ferrosoft.apps.dataimport.websocket.updater import ImportJobUIUpdater
from ferrosoft.apps.emiflow.dataimport.base import (
ImportState,
process_transport_chain,
get_transport_chain_defaults,
)
from ferrosoft.apps.emiflow.models import (
MultiLineImportLine,
TransportChain,
BookingStatus,
FreightItem,
TransportChainElement,
TransportChainDefaults,
DistanceType,
MultiLineBookingRecord,
OperationEpitype,
TransportChainType,
Incoterm,
)
from ferrosoft.apps.emiflow.services.builder import (
TransportChainBuilder,
FreightItemBuilder,
TransportChainElementBuilder,
)
from ferrosoft.apps.emiflow.services.emissions import TransportBookingService
from ferrosoft.apps.emiflow.services.visitor import GeocodingRoutingError
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.services.taskqueue import (
background_task_tenant,
tenant_task,
wait_for_finish,
)
[docs]
def model_processor(pargs: ModelProcessorArgs):
process_transport_chain(pargs, MultiLineImportLine, _create_transport_chain)
def _create_transport_chain(
model: MultiLineImportLine, state: ImportState, pargs: ModelProcessorArgs
):
line: MultiLineImportLine = model
defaults = get_transport_chain_defaults(line, state, pargs)
if defaults is None:
return
chain = _create_with_reference(TransportChain, model.reference, _create_chain(line))
create_or_get_model(FreightItem, _create_freight_item(line, chain))
create_or_get_model(TransportChainElement, _create_tce(line, chain, defaults))
# Save chain to be booked later on.
MultiLineBookingRecord.add_transport(chain, pargs.job.pk)
def _create_with_reference[T](
model_type: Type[T], reference: str, creator: Callable[[str], T]
) -> T:
return create_or_get_model(model_type, lambda: creator(reference))
def _create_chain(model: MultiLineImportLine):
"""Create a TransportChain."""
def creator(reference) -> TransportChain:
return (
TransportChainBuilder.new()
.status(BookingStatus.OPEN)
.reference(reference)
.booking_date(model.booking_date)
.transport_type(TransportChainType(model.transport_type))
.incoterms(None if model.incoterms is None else Incoterm(model.incoterms))
.consignor(model.consignor)
.consignee(model.consignee)
.save()
)
return creator
def _create_freight_item(
line: MultiLineImportLine,
chain: TransportChain,
):
def creator() -> FreightItem:
return (
FreightItemBuilder.new()
.chain(chain)
.weight(line.weight)
.material(line.material)
.save()
)
return creator
def _create_tce(
line: MultiLineImportLine, chain: TransportChain, defaults: TransportChainDefaults
):
def creator() -> TransportChainElement:
builder = (
TransportChainElementBuilder.new()
.chain(chain)
.epitype(OperationEpitype.TRANSPORT)
.position(line.position)
.operation(defaults.transport_operation)
.activity_distance_type(DistanceType(defaults.distance_type))
.activity_distance(DecimalWithUnit.create(0, "km"))
.distance_adjustment_factor(Decimal("1.0"))
)
# Implicitly create GEO records.
builder.add_geo_origin().address(line.start_address())
builder.add_geo_destination().address(line.destination_address())
return builder.save()
return creator
[docs]
def book_transports(pargs: ProcessorArgs):
"""Book all transport chains recorded into MultiLineBookingRecord's."""
job = pargs.job
group_name = "transport_booking_%s" % str(job.pk)
group_count = 0
updater = ImportJobUIUpdater(job.pk)
try:
updater.send_log_entry(
job.pk,
_(
"Transports are being enqueued for booking. This might take a while for large datasets."
),
0,
LogHighlight.INFO,
)
for chain_id in MultiLineBookingRecord.get_transports(job.pk):
# Book transports in subtasks, to make use of concurrency.
background_task_tenant(
"ferrosoft.apps.emiflow.dataimport.multiline.book_single_transport",
chain_id,
job.pk,
result_group=group_name,
task_stream="emiflow_booking",
)
group_count += 1
# Wait until all transports are booked. Results don't matter.
updater.send_log_entry(
job.pk,
_("Enqueued %(count)d transports for booking.") % {"count": group_count},
0,
LogHighlight.INFO,
)
if group_count > 0:
wait_for_finish(result_group=group_name, count=group_count)
finally:
MultiLineBookingRecord.clear_transports(job.pk)
[docs]
@sync_to_async
@tenant_task()
def book_single_transport(chain_id, job_id):
"""Book a single transport chain.
This includes geocoding and routing of TransportChainElements."""
updater = ImportJobUIUpdater(job_id)
chain = TransportChain.objects.get(pk=chain_id)
try:
TransportBookingService().book(chain)
updater.send_log_entry(
job_id,
_("Booked transport %(reference)s") % {"reference": chain.reference},
0,
LogHighlight.SUCCESS,
)
except GeocodingRoutingError as e:
updater.send_log_entry(
job_id,
_("Transport %(reference)s has an address error: %(error)s")
% {
"reference": chain.reference,
"error": str(e),
},
0,
LogHighlight.DANGER,
)
except Exception:
logging.exception("Failed booking transport chain")
updater.send_log_entry(
job_id,
_("Failed booking transport %(reference)s")
% {"reference": chain.reference},
0,
LogHighlight.DANGER,
)