# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""Emission booking records and lifecycle categories.
This module defines the immutable journal of calculated emissions:
* :class:`EmissionBookingRecord` — one row per ``(lifecycle category,
emitting entity, material)`` triple. Fields are denormalised (material
arrays, partner names) so reports can query a single table without
joining live operational data.
* :class:`LifecycleCategory` — user-defined categories plus the four
predefined ones (treatment, incoming/outgoing transport, drop
shipment) enumerated in :class:`PredefinedLCC`.
* :class:`LCCValue` — abstract base for manually entered per-category
emission values that subclasses attach to transports or treatments.
The :class:`EmissionBookingRecordQuerySet` exposes the aggregations used
by the reporting pipeline (totals by entity, by material, by lifecycle
category) — all keyed off the denormalised columns so they remain valid
even after the originating operation has changed.
"""
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Iterable, TypedDict
from uuid import UUID
from django.contrib.postgres.aggregates import StringAgg
from django.contrib.postgres.fields import ArrayField
from django.core.serializers.json import DjangoJSONEncoder
from django.db import models
from django.db.models import QuerySet, Func, F, ExpressionWrapper, Case, When, Value
from django.db.models.aggregates import Sum
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.metrics import TonnageCounter
from ferrosoft.apps.emiflow.models.reporting import ProportionalEmission
from ferrosoft.apps.ferrobase.models import UUIDModel, MeasurementUnit
from ferrosoft.apps.ferrobase.models.decimal import (
DecimalWithUnit,
DecimalWithUnitField,
)
from ferrosoft.apps.ferrobase.models.fields import BigDecimalField
from ferrosoft.apps.ferrobase.tenant.context import get_current_tenant
from ferrosoft.decimal import zero
[docs]
class PredefinedLCC(models.IntegerChoices):
"""Predefined Lifecycle Categories.
These correspond to (not necessarily 1-1) to models found in Emiflow,
for which emissions can be calculated."""
TREATMENT = 1, _("Treatment")
INCOMING_TRANSPORT = 2, _("Incoming Transport")
OUTGOING_TRANSPORT = 3, _("Outgoing Transport")
DROP_SHIPMENT = 4, _("Drop Shipment")
[docs]
@classmethod
def value_from(cls, source):
"""Map an upstream enum value to its corresponding lifecycle category.
Currently supports :class:`TransportChainType`; raises
:class:`ValueError` for any other input type so unmapped values
fail loudly during booking.
"""
from ferrosoft.apps.emiflow.models.transport import TransportChainType
_TRANSPORT_CHAIN_TYPE_MAPPING = {
TransportChainType.INCOMING: PredefinedLCC.INCOMING_TRANSPORT,
TransportChainType.OUTGOING: PredefinedLCC.OUTGOING_TRANSPORT,
TransportChainType.DROP_SHIPMENT: PredefinedLCC.DROP_SHIPMENT,
}
if isinstance(source, TransportChainType):
value = _TRANSPORT_CHAIN_TYPE_MAPPING.get(source, None)
if value is None:
raise ValueError("Unsupported transport chain type")
return value
raise ValueError("Cannot create entity type from %s" % type(source))
EmissionTotal = TypedDict(
"EmissionTotal",
{
"lifecycle_category": int,
"lifecycle_category_names": str,
"entity_id": UUID,
"emission_total": DecimalWithUnit,
"emission_operation": DecimalWithUnit,
"weight": DecimalWithUnit,
"average_total": DecimalWithUnit,
"average_operation": DecimalWithUnit,
},
)
PCFTotal = TypedDict(
"PCFTotal",
{
"material": str,
"lifecycle_category_names": str,
"emission_total": DecimalWithUnit,
"emission_operation": DecimalWithUnit,
"weight": DecimalWithUnit,
"average_total": DecimalWithUnit,
"average_operation": DecimalWithUnit,
},
)
LifecycleResult = TypedDict(
"LifecycleResult",
{
"lifecycle_category": int,
"lifecycle_category_names": str,
"emission_total": DecimalWithUnit,
"emission_operation": DecimalWithUnit,
"weight": DecimalWithUnit,
"average_total": DecimalWithUnit,
"average_operation": DecimalWithUnit,
},
)
[docs]
@dataclass
class LifecycleSum:
"""Running totals over one or more :class:`LifecycleResult` rows."""
emission_total: DecimalWithUnit = field(default_factory=DecimalWithUnit.zero)
emission_operation: DecimalWithUnit = field(default_factory=DecimalWithUnit.zero)
weight: DecimalWithUnit = field(default_factory=DecimalWithUnit.zero)
average_total: DecimalWithUnit = field(default_factory=DecimalWithUnit.zero)
average_operation: DecimalWithUnit = field(default_factory=DecimalWithUnit.zero)
_PROPORTIONAL_METHOD_MAPPING = {
ProportionalEmission.INCOMING_TRANSPORT: PredefinedLCC.INCOMING_TRANSPORT,
ProportionalEmission.OUTGOING_TRANSPORT: PredefinedLCC.OUTGOING_TRANSPORT,
}
[docs]
class EmissionBookingRecordQuerySet(models.QuerySet):
"""QuerySet that knows how to roll booking records up for reports.
Each aggregation method returns rows with ``emission_total``,
``emission_operation``, ``weight``, and the two ``average_*`` columns
annotated as :class:`DecimalWithUnit` values, so reports can format
them without re-attaching units in Python.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.MAX_DIGITS = 12
self.DECIMAL_PLACES = 4
self.emission_unit = MeasurementUnit(symbol="gCO2e")
self.weight_unit = MeasurementUnit(symbol="t")
self.average_unit = MeasurementUnit(symbol="gCO2e/t")
[docs]
def emission_total(self) -> Iterable[EmissionTotal]:
"""Group rows by ``(lifecycle_category, entity_id)`` and aggregate emissions."""
return (
self.values("lifecycle_category", "entity_id")
._annotate_emission_aggregates()
.order_by("lifecycle_category", "entity_id")
)
[docs]
def pcf_total(self) -> Iterable[PCFTotal]:
"""Group rows by material and aggregate emissions.
Materials are stored as a PostgreSQL array per booking record;
``unnest`` is applied so a row carrying multiple materials
contributes to each one independently. Useful for product
carbon footprint reporting — hence the name.
"""
# noinspection PyUnresolvedReferences,PyProtectedMember
return (
self.values(material=Func(F("material_name"), function="unnest"))
._annotate_emission_aggregates()
.order_by("material")
)
[docs]
def filter_material_name(self, name: str):
"""Restrict the queryset to rows whose material array contains ``name``."""
return self.filter(material_name__contains=[name])
[docs]
def pcf_lifecycles(self, material_name: str) -> Iterable[LifecycleResult]:
"""Group rows for one material by lifecycle category and aggregate.
Used by the PCF report, which is product-specific and therefore
always scoped to a single material name.
"""
# noinspection PyUnresolvedReferences,PyProtectedMember
return (
self.filter_material_name(material_name)
.values("lifecycle_category")
._annotate_emission_aggregates()
.order_by("lifecycle_category")
)
def _annotate_emission_aggregates(self) -> QuerySet:
"""Annotate the queryset with summed emissions plus per-tonne averages."""
return self.annotate(
**self.get_emission_aggregate_columns(),
average_total=ExpressionWrapper(
Case(
When(weight=Value(zero), then=Value(zero)),
default=F("emission_total") / F("weight"),
),
output_field=DecimalWithUnitField(
max_digits=self.MAX_DIGITS,
decimal_places=self.DECIMAL_PLACES,
unit=self.average_unit,
),
),
average_operation=ExpressionWrapper(
Case(
When(weight=Value(zero), then=Value(zero)),
default=F("emission_operation") / F("weight"),
),
output_field=DecimalWithUnitField(
max_digits=self.MAX_DIGITS,
decimal_places=self.DECIMAL_PLACES,
unit=self.average_unit,
),
),
lifecycle_category_names=StringAgg(
"lifecycle_category_name", ", ", distinct=True
),
)
[docs]
def get_emission_aggregate_columns(self) -> dict:
"""Return ``Sum`` expressions for total, operation, and weight columns.
Exposed for callers that need to compose their own aggregations
on top of the shared unit-aware output fields.
"""
return {
"emission_total": Sum(
"emission_total_value",
output_field=DecimalWithUnitField(
max_digits=self.MAX_DIGITS,
decimal_places=self.DECIMAL_PLACES,
unit=self.emission_unit,
),
),
"emission_operation": Sum(
"emission_operation_value",
output_field=DecimalWithUnitField(
max_digits=self.MAX_DIGITS,
decimal_places=self.DECIMAL_PLACES,
unit=self.emission_unit,
),
),
"weight": Sum(
"weight_value",
output_field=DecimalWithUnitField(
max_digits=self.MAX_DIGITS,
decimal_places=self.DECIMAL_PLACES,
unit=self.weight_unit,
),
),
}
[docs]
class EmissionBookingRecordManager(models.Manager):
"""Manager exposing aggregation, lookup, and upsert helpers."""
[docs]
@staticmethod
def calculate_proportions(
method: ProportionalEmission, lifecycles: Iterable[LifecycleResult]
) -> Iterable[LifecycleResult]:
"""Scale lifecycle totals against the chosen authoritative leg's weight.
Used by PCF reports when the customer wants the incoming or
outgoing transport weight to drive the emissions allocation.
Lifecycle rows with weights larger than the authoritative leg
are scaled down so they share the same denominator; smaller
weights are left untouched. Returns the input unchanged when
``method`` is ``DISABLED`` or no authoritative row is present.
"""
# Method tells which lifecycle category includes the authoritative
# weight. If method doesn't map to a lifecycle category, then no
# proportional calculation needs to be done.
auth_lifecycle = _PROPORTIONAL_METHOD_MAPPING.get(method, None)
if auth_lifecycle is None:
return lifecycles
auth_weight: DecimalWithUnit | None = None
for lifecycle in lifecycles:
if lifecycle["lifecycle_category"] == auth_lifecycle.value:
auth_weight = lifecycle["weight"]
if auth_weight is None:
return lifecycles
for lifecycle in lifecycles:
if lifecycle["weight"] > auth_weight:
factor = auth_weight.value / lifecycle["weight"].value
lifecycle["weight"] = auth_weight
lifecycle["emission_total"] *= factor
lifecycle["emission_operation"] *= factor
return lifecycles
[docs]
def get_queryset(self) -> EmissionBookingRecordQuerySet:
return EmissionBookingRecordQuerySet(self.model, using=self._db)
[docs]
def find_first(self, lifecycle_category: PredefinedLCC, entity_id: UUID):
"""Return the first booking record for the given category and entity."""
return self.filter(
lifecycle_category=lifecycle_category,
entity_id=entity_id,
).first()
[docs]
def sum_emissions(
self, lifecycle_category: PredefinedLCC, entity_id: UUID
) -> LifecycleSum:
"""Aggregate the entity's emissions in one lifecycle category."""
return (
self.filter(
lifecycle_category=lifecycle_category,
entity_id=entity_id,
)
.emission_total()
.first()
)
[docs]
@staticmethod
def sum_lifecycles(results: Iterable[LifecycleResult]) -> LifecycleSum:
"""Sum a sequence of :class:`LifecycleResult` rows into a single :class:`LifecycleSum`.
Convenience over Python-side iteration for callers that already
have the per-lifecycle rows in memory (e.g. PCF totals).
"""
lsum = LifecycleSum()
for item in results:
lsum.emission_total += item["emission_total"]
lsum.emission_operation += item["emission_operation"]
lsum.weight += item["weight"]
lsum.average_total += item["average_total"]
lsum.average_operation += item["average_operation"]
return lsum
[docs]
def upsert(self, records) -> "list[EmissionBookingRecord]":
"""Bulk insert-or-update booking records and bump the tenant tonnage counter.
``(lifecycle_category, entity_id, material_id)`` is the conflict
target; on conflict every denormalised field is refreshed.
Simulation rows are excluded from the tonnage counter so dry
runs don't burn through a customer's subscription quota.
"""
# Increase counter for tonnage reporting.
weight_sum = sum([rec.weight_value for rec in records if not rec.simulation])
tenant = get_current_tenant()
TonnageCounter.labels(
tenant="None" if tenant is None else tenant.name,
organization="None" if tenant is None else tenant.organization.name,
).inc(float(weight_sum))
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"booking_date",
"lifecycle_category_name",
"entity_reference",
"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",
],
unique_fields=[
"lifecycle_category",
"entity_id",
"material_id",
],
)
[docs]
class EmissionBookingRecord(UUIDModel):
"""One booking entry: an emission figure tied to a lifecycle category and entity.
Records are wide-and-denormalised: material codes/names, partner
names, and unit symbols are copied into the row so report
aggregations stay valid even if the underlying operation is later
edited or deleted. ``extra`` holds optional contextual data
(transport/activity distance, intensity at booking time) used by
detail views.
"""
class Meta:
indexes = [
models.Index(fields=["booking_date"]),
models.Index(fields=["lifecycle_category"]),
models.Index(fields=["lifecycle_category_name"]),
models.Index(fields=["entity_id"]),
models.Index(fields=["material_id"]),
models.Index(fields=["incoming_partner_id"]),
models.Index(fields=["outgoing_partner_id"]),
]
constraints = [
models.UniqueConstraint(
name="uniq_ebr",
fields=[
"lifecycle_category",
"entity_id",
"material_id",
],
),
]
objects = EmissionBookingRecordManager()
creation_time = models.DateTimeField(
default=now_utc,
verbose_name=_("Creation Time"),
)
modification_time = models.DateTimeField(
verbose_name=_("Update Time"),
null=True,
blank=True,
)
booking_date = models.DateField(
verbose_name=_("Booking Date"),
)
lifecycle_category = models.PositiveIntegerField(
verbose_name=_("Lifecycle Category ID"),
)
lifecycle_category_name = models.CharField(
verbose_name=_("Lifecycle Category"),
)
entity_id = models.UUIDField(
verbose_name=_("Entity ID"),
)
entity_reference = models.CharField(
verbose_name=_("Entity Reference"),
)
simulation = models.BooleanField(
verbose_name=_("Simulation"),
)
emission_total_value = BigDecimalField(
verbose_name=_("Emission Total"),
)
emission_operation_value = BigDecimalField(
verbose_name=_("Emission Operation"),
)
emission_unit_id = models.UUIDField(
verbose_name=_("Emission Unit ID"),
)
emission_unit_symbol = models.CharField(
verbose_name=_("Emission Unit Symbol"),
)
weight_value = BigDecimalField(
verbose_name=_("Weight"),
)
weight_unit_id = models.UUIDField(
verbose_name=_("Weight Unit ID"),
)
weight_unit_symbol = models.CharField(
verbose_name=_("Weight Unit Symbol"),
)
material_id = ArrayField(
models.UUIDField(
verbose_name=_("Material ID"),
),
)
material_code = ArrayField(
models.CharField(
verbose_name=_("Material Code"),
),
)
material_name = ArrayField(
models.CharField(
verbose_name=_("Material Name"),
)
)
incoming_partner_id = models.UUIDField(
verbose_name=_("Incoming Partner ID"),
null=True,
blank=True,
)
incoming_partner_name = models.CharField(
verbose_name=_("Incoming Partner Name"),
null=True,
blank=True,
)
outgoing_partner_id = models.UUIDField(
verbose_name=_("Outgoing Partner ID"),
null=True,
blank=True,
)
outgoing_partner_name = models.CharField(
verbose_name=_("Outgoing Partner Name"),
null=True,
blank=True,
)
extra = models.JSONField(
default=dict,
encoder=DjangoJSONEncoder,
verbose_name=_("Extra Data"),
)
@property
def emission_total(self):
"""Total emission paired with the denormalised unit symbol."""
return DecimalWithUnit(
self.emission_total_value,
MeasurementUnit(symbol=self.emission_unit_symbol),
)
@property
def emission_operation(self):
"""Operation-only emission paired with the denormalised unit symbol."""
return DecimalWithUnit(
self.emission_operation_value,
MeasurementUnit(symbol=self.emission_unit_symbol),
)
@property
def weight(self):
"""Booking weight paired with the denormalised unit symbol."""
return DecimalWithUnit(
self.weight_value,
MeasurementUnit(symbol=self.weight_unit_symbol),
)
@property
def transport_distance(self) -> DecimalWithUnit:
"""Transport distance recorded alongside this booking (transport records only)."""
data = self.extra.get("transport_distance", {})
return self._du_from_dict(data)
@property
def activity_distance(self) -> DecimalWithUnit:
"""Activity distance recorded alongside this booking, when applicable."""
data = self.extra.get("activity_distance", {})
return self._du_from_dict(data)
@property
def emission_intensity_operation(self) -> DecimalWithUnit:
"""Operation-only emission intensity that produced this booking."""
data = self.extra.get("emission_intensity", {})
data = data.get("operation", {})
return self._du_from_dict(data)
@property
def emission_intensity_total(self) -> DecimalWithUnit:
"""Total emission intensity that produced this booking."""
data = self.extra.get("emission_intensity", {})
data = data.get("total", {})
return self._du_from_dict(data)
@staticmethod
def _du_from_dict(data: dict) -> DecimalWithUnit:
"""Build a :class:`DecimalWithUnit` from the ``{value, unit_symbol}`` JSON shape."""
value = data.get("value")
return DecimalWithUnit(
Decimal(value) if value is not None else zero,
MeasurementUnit(symbol=data.get("unit_symbol", "")),
)
[docs]
def save(self, *args, **kwargs):
"""Stamp ``modification_time`` and forward to the default save."""
self.modification_time = now_utc()
super().save(*args, **kwargs)
[docs]
class LifecycleCategoryManager(models.Manager):
"""Manager that upserts lifecycle categories by primary key."""
[docs]
def upsert(self, objects):
"""Bulk insert-or-update categories using ``id`` as the conflict target."""
return self.bulk_create(
objects,
update_conflicts=True,
update_fields=["name", "predefined"],
unique_fields=["id"],
)
[docs]
def upsert_predefined(self):
"""Ensure the four :class:`PredefinedLCC` categories exist with their canonical IDs."""
return self.upsert(
[
LifecycleCategory(pk=lcc.value, name=str(lcc.label), predefined=True)
for lcc in PredefinedLCC
],
)
[docs]
class LifecycleCategory(models.Model):
"""A named bucket that booking records are attributed to.
The primary key is intentionally a small integer (not a UUID) so the
four :class:`PredefinedLCC` enum values can be used as the canonical
PKs for the system-provided categories.
"""
# Not inheriting from UUIDModel to keep PK compatible with PredefinedLCC enum.
class Meta:
verbose_name = _("Lifecycle Category")
verbose_name_plural = _("Lifecycle Categories")
objects = LifecycleCategoryManager()
name = models.CharField(
verbose_name=_("Name"),
)
predefined = models.BooleanField(
default=False,
help_text=_("Predefined lifecycle categories cannot be deleted."),
verbose_name=_("Predefined"),
)
def __str__(self):
return self.name
[docs]
def get_absolute_url(self):
return reverse("emiflow:lifecyclecategory_update", args=[self.pk])
[docs]
class LCCValueManager(models.Manager):
"""Manager that eagerly joins the category and emission unit."""
[docs]
def get_queryset(self):
return (
super().get_queryset().select_related("lifecycle_category", "emission_unit")
)
[docs]
def get_ordered(self):
"""Return values sorted by lifecycle-category name."""
return self.order_by("lifecycle_category__name")
[docs]
class LCCValue(UUIDModel):
"""Abstract base for manually entered per-lifecycle emission contributions.
Concrete subclasses (:class:`TransportLCCValue`,
:class:`TreatmentLCCValue`) add the foreign key to the owning entity.
Only custom (non-predefined) categories are intended to be selected
on the form.
"""
objects = LCCValueManager()
lifecycle_category = models.ForeignKey(
LifecycleCategory,
help_text=_("Only custom lifecycles can be selected."),
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Lifecycle Category"),
)
operation_emission_value = BigDecimalField(
verbose_name=_("Operation Emission"),
)
total_emission_value = BigDecimalField(
verbose_name=_("Total Emission"),
)
emission_unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Emission Unit"),
)
def __str__(self):
return "%s %s" % (self._meta.verbose_name, self.lifecycle_category.name)
@property
def operation_emission(self):
"""Operation emission value paired with its unit."""
return DecimalWithUnit(self.operation_emission_value, self.emission_unit)
@property
def total_emission(self):
"""Total emission value paired with its unit."""
return DecimalWithUnit(self.total_emission_value, self.emission_unit)