# Copyright (c) 2024 Ferrosoft GmbH. All rights reserved.
"""``django-filter`` ``FilterSet`` definitions for emiflow list views.
Each :class:`~django_filters.FilterSet` here backs a corresponding list
view: it parses the request query string, validates it against the model
fields declared in ``Meta``, and produces a filtered queryset plus a bound
form that templates render as the filter sidebar. Custom ``method`` filters
support filtering by related-object primary key, by membership in a
PostgreSQL array column, and by ``IN`` clauses for multi-select widgets.
"""
from asgiref.sync import sync_to_async
from django import forms
from django.db.models import Q
from django.http import QueryDict, HttpRequest
from django.utils.translation import gettext_lazy as _
from django_filters import (
FilterSet,
CharFilter,
DateFilter,
ChoiceFilter,
BooleanFilter,
ModelChoiceFilter,
DateTimeFilter,
MultipleChoiceFilter,
ModelMultipleChoiceFilter,
)
from ferrosoft.apps.emiflow.models import (
TransportChain,
EmissionFactor,
Incoterm,
TransportChainType,
Treatment,
TreatmentOperation,
Emitter,
EnergyCarrier,
Report,
EmissionBookingRecord,
PredefinedLCC,
LifecycleCategory,
)
from ferrosoft.apps.emiflow.models.operation import TransportOperation
from ferrosoft.apps.ferrobase.models import BusinessPartner, Material
def _filter_by_pk(qs, name, value):
"""Filter ``qs`` where field ``name`` equals ``value.pk``."""
return qs.filter(**{name: value.pk})
def _filter_by_pk_array(qs, name, value):
"""Filter ``qs`` where the PostgreSQL array field ``name`` contains ``value.pk``."""
lookup = "__".join([name, "contains"])
return qs.filter(**{lookup: [value.pk]})
def _filter_by_pk_multiple(qs, name, value):
"""Filter ``qs`` where field ``name`` is in the primary keys of ``value``."""
lookup = "__".join([name, "in"])
return qs.filter(**{lookup: [it.pk for it in value]})
[docs]
class BusinessPartnerFilter(FilterSet):
"""Filter business partners by name, street, city, or postal code prefix."""
[docs]
class EmissionBookingRecordFilter(FilterSet):
"""Filter emission booking records by lifecycle category, booking-date range, material, and partners.
Booking records carry materials in a PostgreSQL array column, so
``material_id`` uses :func:`_filter_by_pk_array` rather than a normal
foreign-key lookup. Consignor and consignee filters exclude partners
with an empty name.
"""
lifecycle_category = ModelMultipleChoiceFilter(
label=_("Lifecycle Category"),
queryset=LifecycleCategory.objects.order_by("name"),
widget=forms.CheckboxSelectMultiple,
method=_filter_by_pk_multiple,
)
booking_date_from = DateFilter(
label=_("Booking date from"),
widget=forms.DateInput(
attrs={"type": "date"},
format="%Y-%m-%d",
),
field_name="booking_date",
lookup_expr="gte",
)
booking_date_until = DateFilter(
label=_("Booking date until"),
widget=forms.DateInput(
attrs={"type": "date"},
format="%Y-%m-%d",
),
field_name="booking_date",
lookup_expr="lte",
)
material_id = ModelChoiceFilter(
label=_("Material"),
queryset=Material.objects.order_by("code"),
method=_filter_by_pk_array,
)
incoming_partner_id = ModelChoiceFilter(
label=_("Consignor"),
queryset=BusinessPartner.get_ordered().filter(~Q(name="")),
method=_filter_by_pk,
)
outgoing_partner_id = ModelChoiceFilter(
label=_("Consignee"),
queryset=BusinessPartner.get_ordered().filter(~Q(name="")),
method=_filter_by_pk,
)
[docs]
@classmethod
def get_user_filtered(cls, user_filter: str, request: HttpRequest | None = None):
"""Apply a persisted report filter to ``EmissionBookingRecord``.
Reports persist the filter query string the user originally
configured. This method re-applies that query string to produce the
same record set later. An empty ``user_filter`` returns no records,
forcing the user to select filters before a report is generated.
Args:
user_filter: The persisted query string (``request.GET.urlencode()``)
captured when the report was created.
request: Current request; when it carries a ``sort`` GET param,
that ordering is applied. Defaults to descending booking date.
Returns:
tuple[QuerySet, EmissionBookingRecordFilter]: The filtered,
ordered queryset and the bound filter instance.
"""
# Report includes filter query string with which emission booking records are filtered.
# Require user to select some filters.
booking_records = EmissionBookingRecord.objects.none()
if user_filter == "":
filter_set = EmissionBookingRecordFilter()
else:
filter_set = EmissionBookingRecordFilter(data=QueryDict(user_filter))
if filter_set.is_valid():
booking_records = filter_set.qs
# Sort parameter applies to booking records.
if request is not None and "sort" in request.GET:
booking_records = booking_records.order_by(request.GET["sort"])
else:
booking_records = booking_records.order_by("-booking_date")
return booking_records, filter_set
[docs]
@classmethod
async def aget_user_filtered(
cls, user_filter: str, request: HttpRequest | None = None
):
"""Async wrapper around :meth:`get_user_filtered` for ASGI views."""
return await sync_to_async(cls.get_user_filtered)(user_filter, request)
[docs]
class EmissionFactorFilter(FilterSet):
"""Filter emission factors by energy carrier name substring."""
[docs]
class EmitterFilter(FilterSet):
"""Filter emitters by name substring, catalog, type, vehicle type, or energy carrier."""
[docs]
class EnergyCarrierFilter(FilterSet):
"""Filter energy carriers by name substring or catalog."""
[docs]
class ReportFilter(FilterSet):
"""Filter reports by reference prefix, name substring, type, or creation-time range."""
creation_time_from = DateTimeFilter(
label=_("Creation time from"),
widget=forms.DateTimeInput(
attrs={"type": "datetime-local"},
format="%Y-%m-%dT%H:%M",
),
field_name="creation_time",
lookup_expr="gte",
)
creation_time_until = DateTimeFilter(
label=_("Creation time until"),
widget=forms.DateTimeInput(
attrs={"type": "datetime-local"},
format="%Y-%m-%dT%H:%M",
),
field_name="creation_time",
lookup_expr="lte",
)
[docs]
class TransportChainFilter(FilterSet):
"""Filter transport chains by type, reference, booking-date range, incoterms, and simulation flag."""
transport_type = ChoiceFilter(
label=_("Transport Type"),
field_name="transport_type",
choices=TransportChainType.choices,
)
reference_like = CharFilter(
label=_("Reference"),
field_name="reference",
lookup_expr="icontains",
)
booking_date_from = DateFilter(
label=_("Booking date from"),
widget=forms.DateInput(
attrs={"type": "date"},
format="%Y-%m-%d",
),
field_name="booking_date",
lookup_expr="gte",
)
booking_date_until = DateFilter(
label=_("Booking date until"),
widget=forms.DateInput(
attrs={"type": "date"},
format="%Y-%m-%d",
),
field_name="booking_date",
lookup_expr="lte",
)
incoterms = ChoiceFilter(
label=_("Incoterms"),
field_name="incoterms",
choices=Incoterm.choices,
)
simulation = BooleanFilter(
label=_("Simulation"),
field_name="simulation",
)
[docs]
class TransportOperationFilter(FilterSet):
"""Filter transport operations by reference, name, activity type, emitter, or energy carrier."""
[docs]
class TreatmentFilter(FilterSet):
"""Filter treatments by reference, related operation, related emitter/energy carrier, or booking-date range."""
operation = ModelChoiceFilter(
queryset=TreatmentOperation.objects.all(),
field_name="operation",
)
booking_date_from = DateFilter(
label=_("Booking date from"),
widget=forms.DateInput(
attrs={"type": "date"},
format="%Y-%m-%d",
),
field_name="booking_date",
lookup_expr="gte",
)
booking_date_until = DateFilter(
label=_("Booking date until"),
widget=forms.DateInput(
attrs={"type": "date"},
format="%Y-%m-%d",
),
field_name="booking_date",
lookup_expr="lte",
)
[docs]
class TreatmentOperationFilter(FilterSet):
"""Filter treatment operations by reference, name, status, emitter, or energy carrier."""
emitter = ModelChoiceFilter(
queryset=Emitter.objects.all(),
field_name="emitter",
)
energy_carrier = ModelChoiceFilter(
queryset=EnergyCarrier.objects.all(),
field_name="energy_carrier",
)