"""Staging models for bulk transport-chain imports.
Rows here represent one input line that the
:mod:`ferrosoft.apps.emiflow.dataimport` services then turn into real
:class:`TransportChain` records. :class:`SimpleImportLine` and
:class:`MultiLineImportLine` differ in whether one row maps to a whole
transport chain or to one element of a multi-hop chain;
:class:`MultiLineBookingRecord` tracks which chains a given import job
produced so it can post (or roll back) emissions atomically.
"""
from uuid import UUID
from django.db import models
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.dataimport.models import ImportJob
from ferrosoft.apps.emiflow.models.catalog import VehicleType
from ferrosoft.apps.emiflow.models.transport import (
TransportChain,
TransportChainType,
Incoterm,
)
from ferrosoft.apps.ferrobase.models import (
UUIDModel,
BusinessPartner,
Material,
MeasurementUnit,
)
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.models.fields import NormalDecimalField
from ferrosoft.apps.ferromaps.services.addresses import Address
[docs]
class ImportLine(UUIDModel):
"""Abstract base for one row staged for transport-chain creation.
Captures everything required to build a single :class:`FreightItem`
on a :class:`TransportChain`: routing (start/destination address),
cargo (material and weight), and chain metadata (incoterms, vehicle
type, partners).
"""
reference = models.CharField(
verbose_name=_("Reference"),
help_text=_(
"User supplied reference ID for the transport chain. Typically a number with a prefix."
),
)
booking_date = models.DateField(
verbose_name=_("Booking Date"),
null=True,
)
transport_type = models.PositiveIntegerField(
verbose_name=_("Transport Type"),
choices=TransportChainType.choices,
null=True,
)
incoterms = models.CharField(
verbose_name=_("Incoterms"),
choices=Incoterm.choices,
null=True,
blank=True,
)
vehicle_type = models.PositiveIntegerField(
verbose_name=_("Vehicle type"),
help_text=_(
"Vehicle type determines which attributes to use for emissions calculation. "
"These attributes have to be configured in the settings."
),
choices=VehicleType.choices,
null=True,
blank=True,
)
consignor = models.ForeignKey(
BusinessPartner,
on_delete=models.CASCADE,
related_name="+",
null=True,
blank=True,
verbose_name=_("Consignor"),
help_text=_(
"Consignor is for informational purposes only. Leave blank if the consignor is unknown."
),
)
consignee = models.ForeignKey(
BusinessPartner,
on_delete=models.CASCADE,
related_name="+",
null=True,
blank=True,
verbose_name=_("Consignee"),
help_text=_(
"Consignee is for informational purposes only. Leave blank if the consignee is unknown."
),
)
material = models.ForeignKey(
Material,
on_delete=models.CASCADE,
related_name="+",
)
weight_value = NormalDecimalField(
verbose_name=_("Weight Value"),
)
weight_unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Weight unit"),
)
start_street = models.CharField(
verbose_name=_("Start street"),
)
start_house_no = models.CharField(
verbose_name=_("Start house no."),
blank=True,
default="",
)
start_city = models.CharField(
verbose_name=_("Start city"),
)
start_postal_code = models.CharField(
verbose_name=_("Start postal code"),
)
start_country = models.ForeignKey(
"ferrobase.Country",
on_delete=models.CASCADE,
verbose_name=_("Start country"),
null=True,
related_name="+",
)
destination_street = models.CharField(
verbose_name=_("Destination street"),
)
destination_house_no = models.CharField(
verbose_name=_("Destination house no."),
blank=True,
default="",
)
destination_city = models.CharField(
verbose_name=_("Destination city"),
)
destination_postal_code = models.CharField(
verbose_name=_("Destination postal code"),
)
destination_country = models.ForeignKey(
"ferrobase.Country",
on_delete=models.CASCADE,
verbose_name=_("Destination country"),
null=True,
related_name="+",
)
@property
def weight(self) -> DecimalWithUnit:
"""Cargo weight paired with its unit."""
return DecimalWithUnit(self.weight_value, self.weight_unit)
[docs]
def start_address(self) -> Address:
"""Return the start location as an :class:`Address` value object."""
return Address(
first_name="",
last_name="",
salutation="",
company="",
additional_lines="",
street=self.start_street,
house_no=self.start_house_no,
city=self.start_city,
postal_code=self.start_postal_code,
country=self.start_country.iso_code,
)
[docs]
def destination_address(self) -> Address:
"""Return the destination location as an :class:`Address` value object."""
return Address(
first_name="",
last_name="",
salutation="",
company="",
additional_lines="",
street=self.destination_street,
house_no=self.destination_house_no,
city=self.destination_city,
postal_code=self.destination_postal_code,
country=self.destination_country.iso_code,
)
[docs]
class SimpleImportLine(ImportLine, UUIDModel):
"""One import row = one transport chain with a single route step and material movement.
Reference is unique per row, so re-importing the same reference
updates the staged line in place.
"""
class Meta:
constraints = [
models.UniqueConstraint(
fields=["reference"],
name="emiflow_uniq_sil",
),
]
def __str__(self):
return _("Import line for transport %(reference)s") % {
"reference": self.reference
}
[docs]
class MultiLineImportLine(ImportLine, UUIDModel):
"""One import row = one transport-chain element (TCE) for a multi-hop chain.
Lines with the same :attr:`reference` build up the chain in
:attr:`position` order. Multiple materials per position are
supported; the unique constraint includes ``material`` so the same
leg can carry several materials.
"""
class Meta:
constraints = [
models.UniqueConstraint(
fields=["reference", "position", "material"],
name="emiflow_uniq_mil",
),
]
position = models.PositiveIntegerField(
verbose_name=_("Position Number"),
help_text=_(
"Position number of transport chain elements (TCE). "
"TCE's define route sections of multi-hop transports. "
"If you don't have multi-hop transports, set this field to 1. "
"Don't set this field to a position number of materials, "
"which would result in duplicated TCE's and too much emission."
),
)
def __str__(self):
return _("Import line for TCE %(reference)s %(position)d") % {
"reference": self.reference,
"position": self.position,
}
[docs]
class MultiLineBookingRecord(UUIDModel):
"""Tracks which transport chains a multi-line import job has produced.
Used to post emissions for all chains an import created, and to clear
the bookkeeping rows once posting is complete.
"""
job = models.ForeignKey(
ImportJob,
on_delete=models.CASCADE,
related_name="+",
)
transport = models.ForeignKey(
TransportChain,
on_delete=models.CASCADE,
related_name="+",
)
[docs]
@classmethod
def add_transport(cls, transport: TransportChain, job_id: UUID):
"""Record that ``transport`` was produced by ``job_id``."""
return cls.objects.create(
job_id=job_id,
transport=transport,
)
[docs]
@classmethod
def get_transports(cls, job_id: UUID):
"""Return the IDs of all transport chains produced by ``job_id``."""
return cls.objects.filter(job__id=job_id).values_list(
"transport__id", flat=True
)
[docs]
@classmethod
def clear_transports(cls, job_id: UUID):
"""Drop the bookkeeping rows for ``job_id`` (usually after emissions are posted)."""
cls.objects.filter(job__id=job_id).delete()