Source code for ferrosoft.apps.emiflow.forms.transport
# Copyright (c) 2024 Ferrosoft GmbH. All rights reserved.
from django import forms
from django.forms.widgets import HiddenInput
from django.http import QueryDict
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.emiflow.forms.mixins import CopyEnergyCarrierFromEmitter
from ferrosoft.apps.emiflow.models import (
TransportChain,
FreightItem,
TransportChainElement,
DistanceType,
TCEGeoOrigin,
TCEGeoDestination,
IntensityCapableEntity,
OperationEpitype,
TCEGeoPoint,
)
from ferrosoft.apps.emiflow.models.operation import TransportOperation
from ferrosoft.apps.emiflow.services.visitor import GeocodingRoutingVisitor
from ferrosoft.apps.ferrobase.forms import UnitChoiceField
from ferrosoft.apps.ferrobase.models import (
IDSequence,
CommonLimit,
UnitCollection,
Country,
UnitCursor,
)
from ferrosoft.apps.ferrobase.models.registry import ModelRegistry
from ferrosoft.apps.ferrobase.services.idgenerator import IdGeneratorPostgres
from ferrosoft.apps.ferromaps.services.addresses import Address, CoordinatesNotFound
[docs]
class ReportGeneratorForm(forms.Form):
filter_form = forms.CharField(
widget=forms.HiddenInput(
attrs={"id": "reportDownloadFilters"},
)
)
[docs]
class TransportChainForm(forms.ModelForm):
[docs]
class Meta:
model = TransportChain
fields = [
"reference",
"booking_date",
"transport_type",
"incoterms",
"simulation",
"consignor",
"consignee",
]
widgets = {
"booking_date": forms.DateInput(
attrs={"type": "date"},
format="%Y-%m-%d",
),
}
[docs]
@classmethod
def with_defaults(cls):
tc_generator = IdGeneratorPostgres.create(IDSequence.TRANSPORT_CHAIN)
return cls(
initial={
"reference": tc_generator.next_id(),
"booking_date": now_utc(),
},
)
[docs]
class FreightItemForm(forms.ModelForm):
[docs]
class Meta:
model = FreightItem
fields = [
"chain",
"material",
"weight_value",
"weight_unit",
]
widgets = {"chain": forms.HiddenInput()}
weight_unit = UnitChoiceField(
units=UnitCollection.MASS,
initial=UnitCursor("t"),
)
[docs]
class TransportChainElementForm(forms.ModelForm):
DRYRUN_KEY = "dryrun"
DRYRUN_VALUE = "1"
[docs]
class Meta:
model = TransportChainElement
fields = [
"chain",
"epitype",
"operation",
"activity_distance_type",
"activity_distance_value",
"activity_distance_unit",
"distance_adjustment_factor",
]
widgets = {
"chain": HiddenInput,
"epitype": HiddenInput,
}
activity_distance_unit = UnitChoiceField(
label=_("Distance unit"),
units=UnitCollection.DISTANCE,
initial=UnitCursor("km"),
required=False,
)
# Following are geo fields, which are only applicable to distance types SFD/GCD, therefore not required by default.
origin_latitude = forms.DecimalField(
label=_("Latitude"),
required=False,
widget=forms.HiddenInput(),
)
origin_longitude = forms.DecimalField(
label=_("Longitude"),
required=False,
widget=forms.HiddenInput(),
)
origin_street = forms.CharField(
label=_("Street"),
max_length=int(CommonLimit.NORMAL_CHAR),
required=False,
)
origin_house_number = forms.CharField(
label=_("House Number"),
max_length=int(CommonLimit.TINY_CHAR),
required=False,
)
origin_city = forms.CharField(
label=_("City"),
max_length=int(CommonLimit.NORMAL_CHAR),
required=False,
)
origin_postal_code = forms.CharField(
label=_("Postal Code"),
max_length=int(CommonLimit.NORMAL_CHAR),
required=False,
)
origin_country = forms.ModelChoiceField(
label=_("Country"),
queryset=Country.get_ordered(),
required=False,
)
destination_latitude = forms.DecimalField(
label=_("Latitude"),
required=False,
widget=forms.HiddenInput(),
)
destination_longitude = forms.DecimalField(
label=_("Longitude"),
required=False,
widget=forms.HiddenInput(),
)
destination_street = forms.CharField(
label=_("Street"),
max_length=int(CommonLimit.NORMAL_CHAR),
required=False,
)
destination_house_number = forms.CharField(
label=_("House Number"),
max_length=int(CommonLimit.TINY_CHAR),
required=False,
)
destination_city = forms.CharField(
label=_("City"),
max_length=int(CommonLimit.NORMAL_CHAR),
required=False,
)
destination_postal_code = forms.CharField(
label=_("Postal Code"),
max_length=int(CommonLimit.NORMAL_CHAR),
required=False,
)
destination_country = forms.ModelChoiceField(
label=_("Country"),
queryset=Country.get_ordered(),
required=False,
)
def __init__(self, *args, **kwargs):
self.visitor = GeocodingRoutingVisitor.get_instance(
force_distance_calculation=True
)
# Some of this form's fields are dependent on related objects, which
# need to be loaded manually.
instance = kwargs.pop("instance", None)
data: QueryDict | None = kwargs.pop("data", None)
initial: dict = kwargs.pop("initial", None) or {}
if instance is not None:
# Data must be mutatable.
if data is not None:
data = data.copy()
origin = getattr(instance, "geo_origin", None)
if isinstance(origin, TCEGeoPoint):
self._set_geo_point_data(
"origin", origin, data if data is not None else initial
)
destination = getattr(instance, "geo_destination", None)
if isinstance(destination, TCEGeoPoint):
self._set_geo_point_data(
"destination", destination, data if data is not None else initial
)
super().__init__(*args, instance=instance, data=data, initial=initial, **kwargs)
epitype = OperationEpitype.try_parse(
self.data.get("epitype", self.initial.get("epitype", None)),
default=OperationEpitype.TRANSPORT,
)
dtype = DistanceType.try_parse(
self.data.get(
"activity_distance_type",
self.initial.get("activity_distance_type", None),
),
default=DistanceType.ACTUAL_DISTANCE,
)
self.show_geo_inputs = (
dtype == DistanceType.GREAT_CIRCLE
or dtype == DistanceType.SHORTEST_FEASIBLE
)
# Constrain operations based on epitype.
self.fields["operation"].queryset = TransportOperation.objects.filter(
activity_type=epitype
).order_by("reference")
# Hide some inputs unless needed as indicated by data.
self.fields["distance_adjustment_factor"].widget.input_type = "hidden"
self.fields["activity_distance_value"].widget.input_type = "hidden"
self.fields["activity_distance_unit"].widget = forms.HiddenInput()
if epitype == OperationEpitype.TRANSPORT:
self.enable_activity_distance = True
post_action = (
reverse("emiflow:transportchainelement_create")
if self.instance.position is None
else reverse(
"emiflow:transportchainelement_update", args=[self.instance.pk]
)
)
self.fields["activity_distance_type"].widget = forms.Select(
attrs={
# Re-render form when changing the distance type.
"hx-post": post_action,
"hx-target": "#elementFormContent",
},
# Include an empty option, so the user is required to choose a distance type.
choices=lambda: [("", "---------")] + list(DistanceType.choices),
)
# Show some input for ACTUAL_DISTANCE.
if dtype == DistanceType.ACTUAL_DISTANCE:
self.fields["distance_adjustment_factor"].widget.input_type = "number"
self.fields["activity_distance_value"].widget.input_type = "number"
self.fields["activity_distance_unit"].required = True
self.fields["activity_distance_unit"].widget = forms.Select(
choices=self.fields["activity_distance_unit"].choices
)
elif epitype == OperationEpitype.HUB:
self.enable_activity_distance = False
self.fields["activity_distance_type"].required = False
self.fields["activity_distance_value"].required = False
self.fields["distance_adjustment_factor"].required = False
pass
else:
raise ValueError("Unknown epitype")
def _set_geo_point_data(self, prefix: str, point: TCEGeoPoint, data: dict):
data[f"{prefix}_latitude"] = str(point.latitude)
data[f"{prefix}_longitude"] = str(point.longitude)
data[f"{prefix}_street"] = point.street
data[f"{prefix}_house_number"] = point.house_no
data[f"{prefix}_city"] = point.city
data[f"{prefix}_postal_code"] = point.postal_code
data[f"{prefix}_country"] = str(point.country_id)
[docs]
def clean(self):
cleaned_data = super().clean()
dtype = DistanceType.try_parse(cleaned_data.get("activity_distance_type", None))
location_based = (
dtype == DistanceType.SHORTEST_FEASIBLE
or dtype == DistanceType.GREAT_CIRCLE
)
origin_unset = (
cleaned_data.get("origin_longitude", None) is None
or cleaned_data.get("origin_latitude", None) is None
)
destination_unset = (
cleaned_data.get("destination_longitude", None) is None
or cleaned_data.get("destination_latitude", None) is None
)
if location_based and origin_unset:
# User may have entered address manually, therefore no coordinates are set.
origin_unset = not self._geocode_address(cleaned_data, "origin")
if location_based and destination_unset:
# User may have entered address manually, therefore no coordinates are set.
destination_unset = not self._geocode_address(cleaned_data, "destination")
if location_based and origin_unset and destination_unset:
raise forms.ValidationError(_("Please enter origin and destination"))
@staticmethod
def _geocode_address(cleaned_data: dict, prefix: str) -> bool:
fields = {
"street": cleaned_data.get(f"{prefix}_street"),
"house_no": cleaned_data.get(f"{prefix}_house_number"),
"city": cleaned_data.get(f"{prefix}_city"),
"postal_code": cleaned_data.get(f"{prefix}_postal_code"),
"country": cleaned_data.get(f"{prefix}_country"),
}
# Do not attempt geocoding with empty fields
fields_are_set = any(fields.values())
if not fields_are_set:
return False
address = Address(
street=fields["street"],
house_no=fields["house_no"],
city=fields["city"],
postal_code=fields["postal_code"],
country="" if fields["country"] is None else fields["country"].iso_code,
)
try:
location = address.geocode()
cleaned_data[f"{prefix}_latitude"] = location.coords.lat
cleaned_data[f"{prefix}_longitude"] = location.coords.lon
return True
except CoordinatesNotFound:
return False
[docs]
def save(self, commit=True):
instance: TransportChainElement = super().save(commit=False)
if instance.position is None:
instance.position = instance.chain.next_element_position()
# Ensure a unit is set even if user does not specify one (which is the case if the input was hidden).
if instance.activity_distance_unit_id is None:
instance.activity_distance_unit = ModelRegistry.Unit.get(UnitCursor("km"))
dat = self.cleaned_data
origin: TCEGeoOrigin | None = None
destination: TCEGeoDestination | None = None
dtype = DistanceType.try_parse(dat.get("activity_distance_type", None))
if (
dtype == DistanceType.SHORTEST_FEASIBLE
or dtype == DistanceType.GREAT_CIRCLE
):
origin = TCEGeoOrigin(
tce=instance,
longitude=dat["origin_longitude"],
latitude=dat["origin_latitude"],
street=dat["origin_street"],
house_no=dat["origin_house_number"],
city=dat["origin_city"],
postal_code=dat["origin_postal_code"],
country=dat["origin_country"],
)
destination = TCEGeoDestination(
tce=instance,
longitude=dat["destination_longitude"],
latitude=dat["destination_latitude"],
street=dat["destination_street"],
house_no=dat["destination_house_number"],
city=dat["destination_city"],
postal_code=dat["destination_postal_code"],
country=dat["destination_country"],
)
if commit:
instance.save()
if origin is not None and destination is not None:
# Not really bulk with only one object, but this in conjunction
# with update_conflicts=True causes a real INSERT ON CONFLICT DO
# UPDATE query, which is better than Django's update_or_create.
TCEGeoOrigin.objects.bulk_create(
[origin],
update_conflicts=True,
update_fields=[
"longitude",
"latitude",
"street",
"house_no",
"city",
"postal_code",
"country",
],
unique_fields=["tce"],
)
TCEGeoDestination.objects.bulk_create(
[destination],
update_conflicts=True,
update_fields=[
"longitude",
"latitude",
"street",
"house_no",
"city",
"postal_code",
"country",
],
unique_fields=["tce"],
)
# Geocode and route now to be able to present activity distance to the user.
self.visitor.on_element(instance)
return instance
[docs]
class TransportOperationForm(CopyEnergyCarrierFromEmitter, forms.ModelForm):
[docs]
class Meta:
model = TransportOperation
fields = [
"reference",
"name",
"activity_type",
"emitter",
"energy_carrier",
]
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Override energy_carrier required attribute, to be able
# to copy energy carrier from selected emitter. Require
# energy carrier selectively in clean method (see mixin).
self.fields["energy_carrier"].required = False
[docs]
def save(self, commit=True):
instance = super().save(commit=False)
if commit:
# Copy emission intensity if one exists in the emitter.
emitter_intensity = instance.emitter.emission_intensity
if emitter_intensity is not None:
instance.emission_intensity = emitter_intensity.copy_with_owner(
IntensityCapableEntity.TRANSPORT_OPERATION,
instance.pk,
)
instance.save()
return instance