"""Transport-chain domain models.
A :class:`TransportChain` is one consignment from consignor to consignee
broken into ordered :class:`TransportChainElement` (TCE) hops. Each TCE
points at a reusable :class:`TransportOperation` (the activity
definition) and may carry geo-points for shortest-feasible or great-
circle distance calculations. The chain's freight is described by
:class:`FreightItem` rows (material + weight). Custom per-lifecycle
emissions are attached via :class:`TransportLCCValue`.
"""
from abc import ABC
from dataclasses import dataclass
from datetime import datetime, timedelta
from decimal import Decimal
from uuid import UUID
from django.core import validators
from django.db import models
from django.db.models import Prefetch, Sum
from django.db.models.aggregates import Max
from django.templatetags.l10n import localize
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.emiflow.models.booking import EmissionBookingRecord, LCCValue
from ferrosoft.apps.emiflow.models.operation import (
OperationEpitype,
TransportOperation,
)
from ferrosoft.apps.ferrobase.models import (
UUIDModel,
BusinessPartner,
MeasurementUnit,
Material,
AddressMixin,
CoordinatesMixin,
StoredFile,
IntegerChoicesMixin,
UnitCategory,
)
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.models.fields import NormalDecimalField
from ferrosoft.apps.ferrobase.services.conversion import normalize_unit
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_connection
from ferrosoft.decimal import zero
[docs]
class BookingError(RuntimeError):
"""BookingError is raised by TransportChain.book.
Its message is translated and should be shown to the user."""
pass
[docs]
class BookingStatus(models.IntegerChoices):
"""Lifecycle state of a :class:`TransportChain`.
Posted chains are immutable to keep the booking journal consistent.
"""
OPEN = 1, _("Open")
POSTED = 2, _("Posted")
[docs]
class DistanceType(IntegerChoicesMixin, models.IntegerChoices):
"""How the activity distance for a TCE is determined."""
ACTUAL_DISTANCE = 1, _("Actual Distance")
SHORTEST_FEASIBLE = 2, _("Shortest Feasible Distance (SFD)")
GREAT_CIRCLE = 3, _("Great Circle Distance (GCD)")
[docs]
class Incoterm(models.TextChoices):
"""Standard Incoterms® 2020 three-letter codes."""
EXW = "EXW", _("EXW - Ex Works")
FCA = "FCA", _("FCA - Free Carrier")
FAS = "FAS", _("FAS - Free Alongside Ship")
FOB = "FOB", _("FOB - Free On Board")
CFR = "CFR", _("CFR - Cost And FReight")
CIF = "CIF", _("CIF - Cost Insurance Freight")
DAP = "DAP", _("DAP - Delivered At Place")
DPU = "DPU", _("DPU - Delivered at Place Unloaded")
CPT = "CPT", _("CPT - Carriage Paid To")
CIP = "CIP", _("CIP - Carriage Insurance Paid")
DDP = "DDP", _("DDP - Delivered Duty Paid")
[docs]
class TransportChainType(models.IntegerChoices):
"""Direction of a transport chain relative to the booking organisation."""
INCOMING = 1, _("Incoming")
OUTGOING = 2, _("Outgoing")
DROP_SHIPMENT = 3, _("Drop Shipment")
[docs]
class TransportChainVisitor(ABC):
"""Visitor protocol traversed by :meth:`TransportChain.accept`.
Default implementations are no-ops; subclasses override only the
callbacks they care about. Used by booking and report-rendering code
to walk the chain without coupling either side to the iteration
order.
"""
[docs]
def before_chain(self, chain: "TransportChain"):
"""Called once before any freight items or elements are visited."""
pass
[docs]
def on_freight_item(self, item: "FreightItem"):
"""Called for each freight item, in material-code order."""
pass
[docs]
def on_element(self, element: "TransportChainElement"):
"""Called for each chain element, in ``position`` order."""
pass
[docs]
def after_chain(self, chain: "TransportChain"):
"""Called once after all freight items and elements have been visited."""
pass
[docs]
class TransportChainManager(models.Manager):
"""Manager that upserts transport chains keyed by ``reference``."""
[docs]
def upsert(self, records):
"""Bulk insert-or-update chains using ``reference`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"booking_date",
"status",
"transport_type",
"incoterms",
"simulation",
"consignor",
"consignee",
],
unique_fields=[
"reference",
],
)
[docs]
class TransportChainManagerPrefetched(TransportChainManager):
"""Variant of :class:`TransportChainManager` that prefetches elements and LCC values."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related("consignor", "consignee")
.prefetch_related(
Prefetch("elements", TransportChainElement.prefetched_objects.all()),
Prefetch("lcc_values", TransportLCCValue.objects.all()),
)
)
[docs]
class TransportChain(UUIDModel):
"""A consignment from consignor to consignee, modelled as ordered hops.
Each hop is a :class:`TransportChainElement` (TCE); the chain's cargo
is split across one or more :class:`FreightItem` rows by material.
Once :attr:`status` is :class:`POSTED <BookingStatus.POSTED>` the
chain is immutable — ``delete()`` raises and edit forms are
disabled. Generated booking documents are kept for
:attr:`EXPIRE_DOCUMENT_ON` and garbage-collected afterwards.
"""
class Meta:
verbose_name = _("Transport chain")
verbose_name_plural = _("Transport chains")
indexes = [
models.Index(fields=["booking_date"]),
models.Index(fields=["status"]),
]
constraints = [
models.UniqueConstraint(
name="uniq_transport_chain",
fields=["reference"],
)
]
objects = TransportChainManager()
prefetched_objects = TransportChainManagerPrefetched()
EXPIRE_DOCUMENT_ON = timedelta(days=7)
_copy_booking_record = """INSERT INTO emiflow_emissionbookingrecord
(id, lifecycle_category, lifecycle_category_name, entity_reference, entity_id,
creation_time, modification_time, booking_date, simulation,
emission_total_value, emission_operation_value,
emission_unit_id, emission_unit_symbol,
weight_value, weight_unit_id, weight_unit_symbol,
material_id, material_code, material_name,
incoming_partner_id, incoming_partner_name,
outgoing_partner_id, outgoing_partner_name)
SELECT gen_random_uuid(),
src.lifecycle_category, src.lifecycle_category_name,
%s, %s,
src.creation_time, src.modification_time, src.booking_date, src.simulation,
src.emission_total_value, src.emission_operation_value,
src.emission_unit_id, src.emission_unit_symbol,
src.weight_value, src.weight_unit_id, src.weight_unit_symbol,
src.material_id, src.material_code, src.material_name,
src.incoming_partner_id, src.incoming_partner_name,
src.outgoing_partner_id, src.outgoing_partner_name
FROM emiflow_emissionbookingrecord src
WHERE entity_id = %s
"""
reference = models.CharField(
verbose_name=_("Reference ID"),
)
booking_date = models.DateField(
verbose_name=_("Booking date"),
)
status = models.PositiveIntegerField(
verbose_name=_("Booking status"),
choices=BookingStatus.choices,
default=BookingStatus.OPEN,
)
transport_type = models.PositiveIntegerField(
verbose_name=_("Transport Type"),
choices=TransportChainType.choices,
)
incoterms = models.CharField(
verbose_name=_("Incoterms"),
choices=Incoterm.choices,
null=True,
blank=True,
)
simulation = models.BooleanField(
verbose_name=_("Simulation"),
default=False,
)
consignor = models.ForeignKey(
BusinessPartner,
verbose_name=_("Consignor"),
related_name="+",
on_delete=models.RESTRICT,
null=True,
blank=True,
)
consignee = models.ForeignKey(
BusinessPartner,
verbose_name=_("Consignee"),
related_name="+",
on_delete=models.RESTRICT,
null=True,
blank=True,
)
documents = models.ManyToManyField(
StoredFile,
verbose_name=_("Documents"),
related_name="transport_chains",
)
def __str__(self):
return "%s %s" % (self._meta.verbose_name, self.reference)
[docs]
@classmethod
def default_document_expiry(cls, creation_time: datetime) -> datetime:
"""Return the expiry timestamp for documents created at ``creation_time``."""
return creation_time + cls.EXPIRE_DOCUMENT_ON
[docs]
def accept(self, visitor: TransportChainVisitor):
"""Visit this chain and its related objects in canonical order.
The visitor receives callbacks in this sequence:
1. ``before_chain(self)``
2. ``on_freight_item(item)`` for each freight item.
3. ``on_element(element)`` for each transport chain element.
4. ``after_chain(self)``
"""
visitor.before_chain(self)
for item in self.freight_items.all():
visitor.on_freight_item(item)
for element in self.elements_ordered():
visitor.on_element(element)
visitor.after_chain(self)
[docs]
def copy(self, new_reference: str) -> "TransportChain":
"""Deep-copy this chain (elements, geo-points, freight) under a new reference.
Existing emission booking records for the original chain are
re-inserted attached to the new chain via a raw SQL ``INSERT
... SELECT`` to avoid recomputing them. The new chain inherits
the original's status, incoterms, simulation flag, and partners.
"""
from ferrosoft.apps.emiflow.services import builder
builder = builder.TransportChainBuilder.new()
(
builder.reference(new_reference)
.booking_date(self.booking_date)
.status(self.status)
.transport_type(self.transport_type)
.incoterms(self.incoterms)
.simulation(self.simulation)
.consignor(self.consignor)
.consignee(self.consignee)
)
for element in self.elements_ordered():
element_builder = (
builder.add_element()
.position(element.position)
.epitype(element.epitype)
.activity_distance_type(element.activity_distance_type)
.activity_distance(element.activity_distance)
.operation(element.operation)
)
point = getattr(self, "geo_destination", None)
if isinstance(point, TCEGeoPoint):
(
element_builder.add_geo_destination()
.address(point.address)
.coordinates(point.coordinates)
)
point = getattr(self, "geo_origin", None)
if isinstance(point, TCEGeoPoint):
(
element_builder.add_geo_origin()
.address(point.address)
.coordinates(point.coordinates)
)
for item in self.freight_items_ordered():
(builder.add_freight_item().weight(item.weight).material(item.material))
chain = builder.save()
with tenant_db_connection().cursor() as cursor:
cursor.execute(
self._copy_booking_record,
[new_reference, str(chain.pk), str(self.pk)],
)
return chain
[docs]
def delete(self, *args, **kwargs):
"""Refuse to delete unless the chain is still ``OPEN``.
Posted chains have associated booking records and must not be
removed; the call raises :class:`RuntimeError` to prevent the
delete from being silently lost.
"""
if not self.is_status(BookingStatus.OPEN):
raise RuntimeError("Operation not permitted")
super().delete(*args, **kwargs)
[docs]
def elements_ordered(self):
"""Return chain elements in ``position`` order with related operation data prefetched."""
return (
self.elements.order_by("position")
.select_related(
"operation",
"operation__emitter",
"operation__energy_carrier",
"activity_distance_unit",
)
.prefetch_related(
"geo_origin",
"geo_destination",
)
)
[docs]
def freight_items_ordered(self):
"""Return freight items sorted by material code with unit and material prefetched."""
return self.freight_items.order_by("material__code").select_related(
"weight_unit",
"material",
)
[docs]
def get_absolute_url(self):
return reverse("emiflow:transportchain_update", args=[self.pk])
[docs]
def is_status(self, status: BookingStatus) -> bool:
"""Return ``True`` when the chain is in the given booking status."""
return self.status == status.value
[docs]
def next_element_position(self) -> int:
"""Return the next free ``position`` value for a new chain element."""
result = (
self.elements.values("chain")
.annotate(next_position=Max("position") + 1)
.order_by("chain")
.first()
)
if result is None:
return 1
return result["next_position"]
[docs]
def post_emission(self) -> list[EmissionBookingRecord]:
"""Run the booking service and return the booking records it produced."""
from ferrosoft.apps.emiflow.services.emissions import TransportBookingService
return TransportBookingService().book(self)
[docs]
def sum_weight(self) -> DecimalWithUnit:
"""Return the chain's total freight weight, normalised to metric tons."""
return FreightItem.objects.sum_weight(self.pk)
[docs]
class TCEManager(models.Manager):
"""Manager that upserts TCEs keyed by ``(chain, position)``."""
[docs]
def upsert(self, records):
"""Bulk insert-or-update TCEs using ``(chain, position)`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"epitype",
"activity_distance_type",
"activity_distance_value",
"activity_distance_unit",
"distance_adjustment_factor",
"operation",
],
unique_fields=[
"chain",
"position",
],
)
[docs]
class TCEManagerPrefetched(TCEManager):
"""Variant of :class:`TCEManager` that joins the operation and distance unit."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related(
"operation",
"activity_distance_unit",
)
)
[docs]
class TransportChainElement(UUIDModel):
"""TransportChainElement describes a single transport or hub operation in a transport chain.
Some characteristics for emissions calculation are defined in the operation category.
Attributes:
activity_distance_type: Informs how activity distance was determined.
"""
class Meta:
verbose_name = _("Transport chain element")
constraints = [
models.UniqueConstraint(
name="uniq_tce",
fields=[
"chain",
"position",
],
)
]
objects = TCEManager()
prefetched_objects = TCEManagerPrefetched()
chain = models.ForeignKey(
TransportChain,
on_delete=models.CASCADE,
related_name="elements",
)
position = models.PositiveIntegerField(
verbose_name=_("Position"),
)
epitype = models.PositiveIntegerField(
verbose_name=_("Type of Operation"),
choices=OperationEpitype.choices,
)
activity_distance_type = models.PositiveIntegerField(
verbose_name=_("Distance Type"),
help_text=_(
"Based on the distance type, additional information is required. For actual distance, the distance must be entered. For shortest feasible and great circle distance, origin and destination must be entered."
),
choices=DistanceType.choices,
null=True,
)
activity_distance_value = NormalDecimalField(
verbose_name=_("Activity Distance Value"),
default=zero,
validators=[
validators.MinValueValidator(0),
],
)
activity_distance_unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
null=True,
)
distance_adjustment_factor = NormalDecimalField(
verbose_name=_("Distance adjustment factor"),
default=1,
)
operation = models.ForeignKey(
TransportOperation,
on_delete=models.RESTRICT,
related_name="transport_chain_elements",
verbose_name=_("Operation"),
null=True,
)
def __str__(self):
return "%s %s:%d" % (
self._meta.verbose_name,
self.chain.reference,
self.position,
)
@property
def activity_distance(self):
"""Activity distance normalised to the canonical distance unit (4 dp).
Returns a zero :class:`DecimalWithUnit` unchanged so callers can
treat unset and zero distances uniformly.
"""
distance = DecimalWithUnit(
self.activity_distance_value, self.activity_distance_unit
)
if distance == DecimalWithUnit.zero() or distance.unit is None:
return distance
return normalize_unit(UnitCategory.DISTANCE, distance).round(4)
@property
def activity_distance_parameters(self) -> "ActivityDistanceParameters":
"""Bundle the distance, DAF, and geo-points needed to compute the activity distance.
Raises:
RuntimeError: When called on a hub (non-transport) element —
hub TCEs use freight mass as activity, not distance.
"""
if OperationEpitype(self.epitype) != OperationEpitype.TRANSPORT:
raise RuntimeError(
"Activity distance parameters are only supported with TRANSPORT epitype"
)
return ActivityDistanceParameters(
DistanceType(self.activity_distance_type),
self.activity_distance,
self.distance_adjustment_factor,
getattr(self, "geo_origin", None),
getattr(self, "geo_destination", None),
)
@property
def show_distance_adjustment_factor(self):
"""Whether the form should expose the DAF input (only for actual distance)."""
if self.activity_distance_type is None:
return False
return DistanceType(self.activity_distance_type) == DistanceType.ACTUAL_DISTANCE
@property
def show_geo_points(self):
"""Whether the form should expose origin/destination geo-points.
Geo-points only matter for great-circle and shortest-feasible
distance computations; actual distance uses the entered number.
"""
dtype = DistanceType(self.activity_distance_type)
return (
dtype == DistanceType.GREAT_CIRCLE
or dtype == DistanceType.SHORTEST_FEASIBLE
)
[docs]
def get_absolute_url(self):
return reverse("emiflow:transportchainelement_update", args=[self.pk])
[docs]
class TCEGeoPoint(AddressMixin, CoordinatesMixin, UUIDModel):
"""Abstract base for the two TCE geo-point models (origin, destination)."""
[docs]
class TCEGeoPointManager(models.Manager):
"""Manager that upserts geo-points keyed by ``tce``."""
[docs]
def upsert(self, records):
"""Bulk insert-or-update geo-points using ``tce`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"salutation",
"first_name",
"last_name",
"company",
"additional_lines",
"street",
"house_no",
"city",
"postal_code",
"country",
"latitude",
"longitude",
],
unique_fields=["tce"],
)
[docs]
class TCEGeoOrigin(TCEGeoPoint, UUIDModel):
"""
Geographical origin of a transport chain element.
Note that TCE's don't necessarily have a GEO origin.
It depends on TCE's activity_distance_type.
"""
objects = TCEGeoPointManager()
tce = models.OneToOneField(
TransportChainElement,
on_delete=models.CASCADE,
related_name="geo_origin",
)
def __str__(self):
return "TCE origin at %s;%s" % (self.latitude, self.longitude)
[docs]
class TCEGeoDestination(TCEGeoPoint, UUIDModel):
"""
Geographical destination of a transport chain element.
Note that TCE's don't necessarily have a GEO destination.
It depends on TCE's activity_distance_type.
"""
objects = TCEGeoPointManager()
tce = models.OneToOneField(
TransportChainElement,
on_delete=models.CASCADE,
related_name="geo_destination",
)
def __str__(self):
return "TCE origin at %s;%s" % (self.latitude, self.longitude)
[docs]
@dataclass
class ActivityDistanceParameters:
"""
Includes activity distance along with some related parameters.
Note that "activity distance" only has meaning for TCE's of epitype
TRANSPORT. For those, activity distance is transport activity.
For HUB TCE's on the other hand, no activity distance exists. Freight item
mass is the transport activity in this case.
Attributes:
daf: Distance Adjustment Factor
"""
distance_type: DistanceType
distance: DecimalWithUnit
daf: Decimal
origin: TCEGeoOrigin | None
destination: TCEGeoDestination | None
def __str__(self):
return "%s %s (DAF %s) from %s to %s" % (
self.distance_type.label,
self.distance,
self.daf,
"-" if not self.origin else self.origin.address.format_lines(", "),
(
"-"
if not self.destination
else self.destination.address.format_lines(", ")
),
)
[docs]
class FreightItemManager(models.Manager):
"""Manager with upsert and weight-aggregation helpers."""
[docs]
def upsert(self, records):
"""Bulk insert-or-update freight items using ``(chain, material)`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"weight_value",
"weight_unit",
"material",
],
unique_fields=[
"chain",
"material",
],
)
[docs]
def sum_weight(self, chain_id: UUID):
"""Return the total freight weight for ``chain_id``, normalised to metric tons.
Weights are summed per unit at the database layer and then
unit-converted in Python so chains carrying a mix of units
(e.g. kg + t) produce a single, meaningful total.
"""
weights = (
self.filter(chain_id=chain_id)
.values("weight_unit__id")
.annotate(weight_sum=Sum("weight_value"))
.order_by("weight_unit__id")
)
# Mixed weights cause headaches, need to be normalized.
weight_sum = DecimalWithUnit.zero("t")
for weight in weights:
weight_sum += normalize_unit(
UnitCategory.MASS,
DecimalWithUnit.create(weight["weight_sum"], weight["weight_unit__id"]),
)
return weight_sum
[docs]
class FreightItem(UUIDModel):
"""Quantified material allocated to a transport chain."""
class Meta:
verbose_name = _("Freight item")
constraints = [
models.UniqueConstraint(
name="freight_item_uniq_material",
fields=["chain", "material"],
),
]
objects = FreightItemManager()
chain = models.ForeignKey(
TransportChain,
on_delete=models.CASCADE,
related_name="freight_items",
verbose_name=_("Transport Chain"),
)
weight_value = NormalDecimalField(
verbose_name=_("Weight"),
validators=[
validators.MinValueValidator(0),
],
)
weight_unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Weight Unit"),
)
material = models.ForeignKey(
Material,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Material"),
)
def __str__(self):
return "%s %s %s %s" % (
self._meta.verbose_name,
self.material,
localize(self.weight_value),
self.weight_unit.symbol,
)
@property
def weight(self):
"""Freight item's weight paired with its unit."""
return DecimalWithUnit(self.weight_value, self.weight_unit)
[docs]
def get_absolute_url(self):
return reverse("emiflow:freightitem_update", args=[self.pk])
[docs]
class TransportLCCValue(LCCValue):
"""Custom lifecycle category values for transports."""
chain = models.ForeignKey(
TransportChain,
on_delete=models.CASCADE,
related_name="lcc_values",
verbose_name=_("Transport"),
)
[docs]
def get_absolute_url(self):
return reverse("emiflow:transportlccvalue_update", args=[self.pk])