# Copyright (c) 2024 Ferrosoft GmbH. All rights reserved.
"""Document generators that turn booking records into renderable reports.
Each :class:`DocumentGenerator` subclass binds a :class:`Report` (and
its filtered queryset of booking records) to a list of
:class:`DocumentOutput` objects. The base ``generate()`` renders each
output's template into HTML, and ``prepare_report_models()`` persists
the rendered HTML as short-lived :class:`StoredFile`-backed reports
that can later be turned into PDF by the upstream reporting service.
:class:`DocumentGeneratorRegistry` maps :class:`ReportType` values to
the matching generator class.
"""
from abc import ABC, abstractmethod
from datetime import datetime
from attrs import define
from django.core.files.base import ContentFile
from django.db.models import (
ExpressionWrapper,
Case,
When,
F,
DecimalField,
Value,
QuerySet,
)
from django.db.models.aggregates import Count, Sum
from django.db.models.fields.json import KT
from django.db.models.functions import Cast
from django.template import loader, Template
from django.utils.translation import gettext as _
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.emiflow.models import (
Report,
EmissionBookingRecord,
ProportionalEmission,
PredefinedLCC,
ReportTemplate,
ReportType,
)
from ferrosoft.apps.emiflow.tables import (
EmissionBookingRecordTable,
WarehouseOverviewTable,
)
from ferrosoft.apps.ferrobase.models import (
LocalUser,
MeasurementUnit,
)
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnitField
from ferrosoft.apps.reporting.models import Report as UpstreamReport
from ferrosoft.apps.reporting.services import Params, prepare_report
from ferrosoft.decimal import zero
from ferrosoft.report import generate_report_pdf
[docs]
def generate_report_document(report: UpstreamReport) -> bytes:
"""Read the report's stored HTML and convert it to PDF bytes."""
with report.document.content.open("rb") as f:
return generate_report_pdf(f.read())
[docs]
@define
class DocumentOutput:
"""Document outputs are produced by report generators."""
template_name: str
"""Template from which to generate the document"""
title: str = _("Report")
"""Document title"""
content: str = ""
"""HTML content (set by the generator)"""
[docs]
def get_context_data(self, **kwargs) -> dict:
"""Get context data in addition to report generator context data."""
return {}
[docs]
@define
class DocumentGenerator(ABC):
"""Generate HTML document for Emiflow reports."""
report: Report
"""The report for which the content is to be generated."""
booking_records: QuerySet[EmissionBookingRecord] = []
"""Reports center around booking records.
They can be used in get_context_data to make queries."""
template: ReportTemplate | None = None
"""Content customization parameters"""
[docs]
@abstractmethod
def get_document_outputs(self) -> list[DocumentOutput]:
"""Return which outputs to create.
A generator can return one or more outputs. Each non-empty output
is saved into a StoredFile associated with a Report model. The
user can select which output to view or to export to PDF.
See also generate method on how the outputs are used."""
return []
[docs]
def get_context_data(self) -> dict:
"""Get context for report template rendering.
This provides the base context for all document outputs.
DocumentOutput itself also provides a get_contex_data method,
which adds to the base context."""
return {}
[docs]
def generate(self, template: ReportTemplate | None = None) -> list[DocumentOutput]:
"""Generate document outputs.
Outputs are retrieved via get_document_outputs, which must be implemented
in child classes. For each output, the template defined in the output
is rendered and the resulting HTML is stored in output content field.
Args:
template: Content customization parameters overriding default parameters.
Returns: List of outputs to be saved as StoredFile associated with a Report model.
"""
outputs = self.get_document_outputs()
context = self.get_context_data()
for output in outputs:
output.content = format_html(
output.template_name,
context=context | output.get_context_data(generator=self),
title=output.title,
params=template or self.template,
)
return outputs
[docs]
def prepare_report_models(
self,
outputs: list[DocumentOutput],
owner: LocalUser,
) -> list[UpstreamReport]:
"""Create Report models (of the reporting app) with the content of given outputs."""
reports = []
for output in outputs:
report = prepare_report(
generate_report_document,
params=Params(
owner=owner,
file_name=output.title + ".html",
# HTML content is short-lived, as it is used for display
# in the user session only.
expiry_days=1,
media_type="text/html",
),
)
# Save HTML content into Report's file, so the content doesn't
# have to be passed through the task queue.
report.document.content.save("content.html", ContentFile(output.content))
reports.append(report)
return reports
[docs]
@define
class PCFDocumentOutput(DocumentOutput):
"""Output for one PCF report — one document per material."""
material_name: str = ""
[docs]
def get_context_data(self, **kwargs):
"""Aggregate booking records for this material and build the table context.
Partner columns are hidden from the rendered table because PCF
reports are typically forwarded to third parties.
"""
assert self.material_name != ""
generator: DocumentGenerator = kwargs.get("generator", None)
lifecycles = EmissionBookingRecord.objects.calculate_proportions(
ProportionalEmission(generator.report.proportional_emissions),
generator.booking_records.pcf_lifecycles(self.material_name),
)
records_table = EmissionBookingRecordTable(
generator.booking_records.filter_material_name(self.material_name),
orderable=False,
)
# Remove partner columns for privacy, as the report may be passed on.
records_table.columns.hide("incoming_partner_name")
records_table.columns.hide("outgoing_partner_name")
return {
"lifecycles": lifecycles,
"material_name": self.material_name,
"records_table": records_table,
"total": EmissionBookingRecord.objects.sum_lifecycles(lifecycles),
}
[docs]
@define
class PCFDocumentGenerator(DocumentGenerator):
"""Generator that emits one PCF document per distinct material."""
[docs]
def get_document_outputs(self) -> list[DocumentOutput]:
"""Return one :class:`PCFDocumentOutput` per material in the booking records."""
# One document per material, as PCF is inherently product based.
return [
PCFDocumentOutput(
material_name=total["material"],
template_name="emiflow/report/pcf_report.html",
title=_("Product Carbon Footprint (PCF) for %(material)s")
% {
"material": total["material"],
},
)
for total in self.booking_records.pcf_total()
]
[docs]
@define
class TransportEmissionsDocumentGenerator(DocumentGenerator):
"""Generator for ISO 14083 transport-emissions reports (summary + listing)."""
[docs]
def get_document_outputs(self) -> list[DocumentOutput]:
"""Return the two standard outputs: summary and full listing."""
return [
DocumentOutput(
template_name="emiflow/report/transport_emissions/summary.html",
title=_("Transport Emissions Summary"),
),
DocumentOutput(
template_name="emiflow/report/transport_emissions/full.html",
title=_("Transport Emissions Listing"),
),
]
[docs]
def get_context_data(self) -> dict:
"""Aggregate transport bookings (distance, intensity, totals) into the template context.
Restricts the records to the three transport-related lifecycle
categories (incoming, outgoing, drop shipment) and computes
distance- and intensity-based aggregates from the denormalised
``extra`` JSON fields. Unit symbols are hard-coded because
booking records always store transport distances and weights
in those units.
"""
# For transport emissions, only these lifecycle categories are of interest.
booking_records = self.booking_records.filter(
lifecycle_category__in=[
PredefinedLCC.INCOMING_TRANSPORT,
PredefinedLCC.OUTGOING_TRANSPORT,
PredefinedLCC.DROP_SHIPMENT,
]
)
# The following summary query uses hard coded units, which is fine, since
# booking records always use these units.
summary = booking_records.aggregate(
count=Count("*"),
transport_distance=Sum(
Cast(
KT("extra__transport_distance__value"),
output_field=DecimalField(max_digits=12, decimal_places=4),
),
output_field=DecimalWithUnitField(
max_digits=12,
decimal_places=4,
unit=MeasurementUnit(symbol="km"),
),
),
activity_distance=Sum(
Cast(
KT("extra__activity_distance__value"),
output_field=DecimalField(max_digits=12, decimal_places=4),
),
output_field=DecimalWithUnitField(
max_digits=12,
decimal_places=4,
unit=MeasurementUnit(symbol="tkm"),
),
),
intensity_total=ExpressionWrapper(
Case(
When(activity_distance=Value(zero), then=Value(zero)),
default=Sum("emission_total_value") / F("activity_distance"),
),
output_field=DecimalWithUnitField(
max_digits=12,
decimal_places=4,
unit=MeasurementUnit(symbol="gCO2e/tkm"),
),
),
intensity_operation=ExpressionWrapper(
Case(
When(activity_distance=Value(zero), then=Value(zero)),
default=Sum("emission_operation_value") / F("activity_distance"),
),
output_field=DecimalWithUnitField(
max_digits=12,
decimal_places=4,
unit=MeasurementUnit(symbol="gCO2e/tkm"),
),
),
**booking_records.get_emission_aggregate_columns(),
)
return {"summary": summary, "records": booking_records}
[docs]
@define
class WarehouseOverviewGenerator(DocumentGenerator):
"""Generator for the warehouse overview report (one table grouped by material)."""
[docs]
def get_document_outputs(self) -> list[DocumentOutput]:
"""Return a single output containing the warehouse overview table."""
return [
DocumentOutput(
template_name="emiflow/report/warehouse_overview.html",
title=_("Warehouse Overview"),
),
]
[docs]
def get_context_data(self) -> dict:
"""Provide the warehouse overview table, sourced from the PCF aggregate query."""
return {"table": WarehouseOverviewTable(self.booking_records.pcf_total())}
[docs]
@define
class DocumentGeneratorRegistry:
"""Maps :class:`ReportType` to the matching :class:`DocumentGenerator` class."""
generator_classes = {
ReportType.PCF: PCFDocumentGenerator,
ReportType.TRANSPORT: TransportEmissionsDocumentGenerator,
ReportType.WAREHOUSE: WarehouseOverviewGenerator,
}
[docs]
@classmethod
def get_generator(cls, report_type: ReportType, **kwargs) -> DocumentGenerator:
"""Instantiate the generator registered for ``report_type``.
``kwargs`` are forwarded to the generator class constructor
(typically ``report``, ``booking_records``, ``template``).
"""
generator_class = cls.generator_classes.get(report_type, None)
assert generator_class is not None
return generator_class(**kwargs)