# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""Emission intensity values shared across transport ops, treatment ops, and emitters.
An :class:`EmissionIntensity` is the calculated *gCO2e per tkm* (or
*gCO2e per t* for treatments) figure with all of the input variables it
was computed from. The ``owner_type`` + ``owner_id`` pair forms a weak
polymorphic reference so the same intensity row can belong to any of
the three intensity-capable model classes; cleanup is handled by the
post-delete signals in :mod:`ferrosoft.apps.emiflow.models.signals`.
"""
from decimal import Decimal
from uuid import UUID
from django.db import models
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from ferrosoft.decimal import zero
from ferrosoft.apps.ferrobase.models import UUIDModel, MeasurementUnit, ReadOnlyMixin
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.models.fields import NormalDecimalField
[docs]
class IntensityCapableEntity(models.IntegerChoices):
"""Enumerate model types for which emission intensity can be defined.
See owner_type and owner_id in EmissionIntensity."""
TRANSPORT_OPERATION = 1, _("Transport Operation")
TREATMENT_OPERATION = 2, _("Treatment Operation")
EMITTER = 3, _("Emitter")
[docs]
def is_distance_based(self) -> bool:
"""Return ``True`` when this owner type factors distance into the intensity.
Treatment operations don't move freight, so their intensities are
not divided by a distance term.
"""
return self.value != self.TREATMENT_OPERATION.value
[docs]
class EmissionIntensityManager(models.Manager):
"""Manager that prefetches units and the two emission factor relations."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related(
"unit",
"factor_operation_value",
"factor_energy_provision_value",
"quantity_unit",
"mass_unit",
"distance_unit",
)
)
[docs]
def delete_of_owner(self, owner_type: IntensityCapableEntity, owner_id: UUID):
"""Delete the intensity row(s) owned by ``(owner_type, owner_id)``.
Used by post-delete signal handlers to keep this weakly-referenced
table in sync with its owning entities.
"""
self.filter(owner_type=owner_type, owner_id=owner_id).delete()
[docs]
class EmissionIntensity(ReadOnlyMixin, UUIDModel):
"""Calculated emission intensity together with its input variables.
The row records both *output* intensities (operation and energy
provision values plus their shared unit) and *input* variables
(factors, quantity of fuel, mass of materials, optional distance) so
the calculation is fully reproducible. The
:attr:`~ReadOnlyMixin.read_only` flag is ``True`` for intensities
derived from imported catalog data (which must be copied via
:meth:`copy_with_owner` before modification) and ``False`` for
user-computed intensities, which remain editable in place.
"""
class Meta:
verbose_name = _("Emission Intensity")
verbose_name_plural = _("Emission Intensities")
constraints = [
models.UniqueConstraint(
name="uniq_emission_intensity",
fields=["owner_type", "owner_id"],
),
]
objects = EmissionIntensityManager()
# Reference to owning entity. This is a weak ref, because emission intensity
# can be defined for many entities. Because of this weak reference,
# post_delete signal handlers are defined for all owner model classes, to
# ensure EmissionIntensity gets removed along with owning entity.
owner_type = models.IntegerField(
verbose_name=_("Owner Type"),
choices=IntensityCapableEntity.choices,
)
owner_id = models.UUIDField(
verbose_name=_("Owner ID"),
)
enable_distance = models.BooleanField(
default=True,
verbose_name=_("Enable Distance"),
)
# Output variables
operation_value = NormalDecimalField(
verbose_name=_("Operation Emission Intensity"),
)
energy_provision_value = NormalDecimalField(
verbose_name=_("Energy Provision Emission Intensity"),
)
unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Emission Intensity Unit"),
null=True,
)
# Input variables
factor_operation_value = models.ForeignKey(
"emiflow.EmissionFactor",
on_delete=models.RESTRICT,
related_name="+",
null=True,
verbose_name=_("Operation Emission Factor"),
)
factor_energy_provision_value = models.ForeignKey(
"emiflow.EmissionFactor",
on_delete=models.RESTRICT,
related_name="+",
null=True,
verbose_name=_("Energy Provision Emission Factor"),
)
quantity_value = NormalDecimalField(
verbose_name=_("Quantity"),
help_text=_("Quantity of fuel or electricity"),
)
quantity_unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Quantity Unit"),
)
mass_value = NormalDecimalField(
verbose_name=_("Mass"),
help_text=_("Mass of all materials treated"),
)
mass_unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Mass Unit"),
)
# Distance may not be relevant for some emitters, in which case
# the distance can be set to 0 km, and it won't factor into the intensity.
distance_value = NormalDecimalField(
verbose_name=_("Distance"),
help_text=_(
"Transport or hub activity distance. Set to 0 if distance is not of interest."
),
)
distance_unit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Distance Unit"),
)
def __str__(self):
return "Operation: %s; Energy Provision: %s" % (
self.operation_intensity,
self.energy_provision_intensity,
)
@property
def operation_intensity(self) -> DecimalWithUnit:
"""Operation emission intensity as a :class:`DecimalWithUnit`."""
return DecimalWithUnit(self.operation_value, self.unit)
@property
def energy_provision_intensity(self) -> DecimalWithUnit:
"""Energy-provision emission intensity as a :class:`DecimalWithUnit`."""
return DecimalWithUnit(self.energy_provision_value, self.unit)
@property
def quantity(self) -> DecimalWithUnit:
"""Fuel or electricity quantity used in the calculation."""
return DecimalWithUnit(self.quantity_value, self.quantity_unit)
@property
def mass(self) -> DecimalWithUnit:
"""Total mass of treated materials used in the calculation."""
return DecimalWithUnit(self.mass_value, self.mass_unit)
@property
def distance(self) -> DecimalWithUnit:
"""Activity distance used in the calculation (zero when not distance-based)."""
return DecimalWithUnit(self.distance_value, self.distance_unit)
@property
def owner(self):
"""Resolve and return the owning entity via the weak ``owner_type``/``owner_id`` pair.
Returns:
The owning :class:`TransportOperation`, :class:`TreatmentOperation`,
or :class:`Emitter` instance.
Raises:
ValueError: If ``owner_type`` is not a recognised
:class:`IntensityCapableEntity` value.
"""
# Import these here to avoid cyclic dependencies.
from ferrosoft.apps.emiflow.models.operation import TransportOperation
from ferrosoft.apps.emiflow.models.treatment import TreatmentOperation
from ferrosoft.apps.emiflow.models.catalog import Emitter
match IntensityCapableEntity(self.owner_type):
case IntensityCapableEntity.TRANSPORT_OPERATION:
return TransportOperation.objects.get(pk=self.owner_id)
case IntensityCapableEntity.TREATMENT_OPERATION:
return TreatmentOperation.objects.get(pk=self.owner_id)
case IntensityCapableEntity.EMITTER:
return Emitter.objects.get(pk=self.owner_id)
case _:
raise ValueError("Unsupported owner type")
@property
def grade(self) -> int:
"""Grade total intensity on a 0..3 scale (higher is better).
Thresholds are centred around the GLEC public-extract average of
117 gCO2e/tkm: 0–80 → 3, 81–140 → 2, 141–200 → 1, above 200 → 0.
Returns:
int: Grade number from 0 (worst) to 3 (best).
"""
int_total = self.operation_intensity + self.energy_provision_intensity
# These arbitrary numbers center around the average of 117 gCO2e/tkm in
# the GLEC public extract.
if int_total.value <= Decimal(80):
return 3
elif int_total.value <= Decimal(140):
return 2
elif int_total.value <= Decimal(200):
return 1
else:
return 0
[docs]
def get_absolute_url(self):
return reverse("emiflow:emissionintensity_update", args=[self.pk])
[docs]
def update_foreign_key(self):
"""Point the owner's ``emission_intensity`` FK at this row.
The relation is duplicated (weak ``owner_id`` here plus a real FK
on the owner) so list views can ``select_related`` it; this method
keeps the FK side in sync after a new intensity is inserted.
Raises:
ValueError: If the resolved owner has no ``emission_intensity``
attribute.
"""
owner = self.owner
if hasattr(owner, "emission_intensity"):
owner.emission_intensity = self
owner.save(update_fields=["emission_intensity"])
else:
raise ValueError("Unsupported owner type")
[docs]
def copy_with_owner(
self, owner_type: IntensityCapableEntity, owner_id: UUID
) -> "EmissionIntensity":
"""Create a copy of this intensity attributed to a new owner.
Distance is zeroed out for owners that are not distance-based
(see :meth:`IntensityCapableEntity.is_distance_based`).
Args:
owner_type: The new owner's entity type.
owner_id: Primary key of the new owner.
Returns:
EmissionIntensity: The newly persisted copy.
"""
enable_distance = owner_type.is_distance_based()
return EmissionIntensity.objects.create(
owner_type=owner_type,
owner_id=owner_id,
operation_value=self.operation_value,
energy_provision_value=self.energy_provision_value,
enable_distance=enable_distance,
unit=self.unit,
factor_operation_value=self.factor_operation_value,
factor_energy_provision_value=self.factor_energy_provision_value,
quantity_value=self.quantity_value,
quantity_unit=self.quantity_unit,
mass_value=self.mass_value,
mass_unit=self.mass_unit,
distance_value=self.distance_value if enable_distance else zero,
distance_unit=self.distance_unit,
)