# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""Treatment and treatment-operation domain models.
A :class:`TreatmentOperation` is the *recipe*: what materials go in,
what comes out, which emitter and energy carrier are involved, and how
emissions should be allocated to the outputs (by weight or manually).
A :class:`Treatment` is a *specific application* of that recipe on a
booking date, with its own input/output lines that can deviate from
the operation defaults. Operations are certified before they can be
referenced; certifying enforces that emission intensity is set and
output allocations sum to 100 %.
"""
from dataclasses import dataclass
from decimal import Decimal
from typing import List
from django.core.validators import MinValueValidator
from django.db import models
from django.db.models import Prefetch, Sum
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.emiflow.models import (
EmissionIntensity,
LCCValue,
EmissionBookingRecord,
)
from ferrosoft.apps.emiflow.models.catalog import Emitter, EnergyCarrier
from ferrosoft.apps.ferrobase.models import (
UUIDModel,
MeasurementUnit,
Material,
zero,
UnitCategory,
)
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.models.fields import NormalDecimalField
from ferrosoft.apps.ferrobase.services.conversion import normalize_unit
from ferrosoft.apps.ferrobase.templatetags.ferrobase import simple_format
[docs]
class AllocationMethod(models.IntegerChoices):
"""How emissions are distributed across a treatment's output materials."""
WEIGHT = 1, _("By Weight")
MANUALLY = 2, _("Manually")
[docs]
def describe(self) -> str:
"""Return a human-readable explanation of this allocation mode."""
if self == AllocationMethod.WEIGHT:
return _(
"Allocation is automatically determined by the weight assigned to material outputs."
)
if self == AllocationMethod.MANUALLY:
return _(
"Allocation is manually determined. Changing weights does not automatically change allocation."
)
raise ValueError("Cannot describe %s" % self)
[docs]
class TreatmentStatus(models.IntegerChoices):
"""Lifecycle state of a :class:`Treatment` (posted treatments are immutable)."""
OPEN = 1, _("Open")
POSTED = 2, _("Posted")
[docs]
class TreatmentOperationStatus(models.IntegerChoices):
"""Lifecycle state of a :class:`TreatmentOperation` (certified ops can be referenced by treatments)."""
OPEN = 1, _("Open")
CERTIFIED = 2, _("Certified")
[docs]
@dataclass
class MaterialAllocation:
"""One row in an :class:`Allocation`: a material and its fraction (0..1)."""
material: Material
fraction: Decimal
[docs]
@dataclass
class Allocation:
"""Snapshot of how a treatment operation's emissions split across its outputs."""
method: AllocationMethod
total_allocation: Decimal
total_weight: DecimalWithUnit
materials: List[MaterialAllocation]
[docs]
class TreatmentOperationManager(models.Manager):
"""Manager that prefetches inputs, outputs, emitter, and energy carrier."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related("emitter", "energy_carrier")
.prefetch_related(
Prefetch("inputs", TreatmentOperationInputLine.objects.all()),
Prefetch("outputs", TreatmentOperationOutputLine.objects.all()),
)
)
[docs]
def upsert(self, records):
"""Bulk insert-or-update operations using ``reference`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"name",
"emitter",
"energy_carrier",
"allocation_method",
],
unique_fields=["reference"],
)
[docs]
class TreatmentOperation(UUIDModel):
"""A reusable treatment recipe (inputs, outputs, emitter, allocation).
Operations move through :class:`TreatmentOperationStatus` from
*Open* to *Certified*; certification is gated by
:meth:`certification_errors` (intensity present + outputs allocated
100 %). Only certified operations are intended to be referenced by
:class:`Treatment` rows.
"""
class Meta:
verbose_name = _("Treatment Operation")
verbose_name_plural = _("Treatment Operations")
indexes = [
models.Index(
name="treatmentop_name",
fields=["name"],
)
]
constraints = [
models.UniqueConstraint(
name="uniq_treatment_op",
fields=["reference"],
)
]
objects = TreatmentOperationManager()
reference = models.CharField(
verbose_name=_("Reference"),
)
name = models.CharField(
verbose_name=_("Name"),
)
emitter = models.ForeignKey(
Emitter,
verbose_name=_("Emitter"),
on_delete=models.RESTRICT,
)
energy_carrier = models.ForeignKey(
EnergyCarrier,
verbose_name=_("Energy Carrier"),
on_delete=models.RESTRICT,
help_text=_("Leave blank to copy from selected emitter"),
)
emission_intensity = models.ForeignKey(
EmissionIntensity,
null=True,
on_delete=models.SET_NULL,
related_name="+",
verbose_name=_("Emission Intensity"),
)
allocation_method = models.IntegerField(
choices=AllocationMethod.choices,
default=AllocationMethod.WEIGHT,
verbose_name=_("Allocation Method"),
)
status = models.IntegerField(
choices=TreatmentOperationStatus.choices,
default=TreatmentOperationStatus.OPEN,
verbose_name=_("Status"),
)
def __str__(self):
if self.name == "":
return self.reference
return "%s - %s" % (self.reference, self.name)
[docs]
def get_absolute_url(self):
return reverse("emiflow:treatmentoperation_update", args=(self.pk,))
[docs]
def get_output_allocation(self) -> Allocation:
"""Bundle the current method, totals, and per-material allocations."""
total_weight, total_allocation = self.outputs.sum_allocation()
return Allocation(
method=AllocationMethod(self.allocation_method),
total_allocation=total_allocation,
total_weight=total_weight,
materials=[
MaterialAllocation(line.material, line.allocated_fraction)
for line in self.outputs.all()
],
)
[docs]
def is_status(self, status: TreatmentOperationStatus) -> bool:
"""Return ``True`` when the operation is in the given status."""
return TreatmentOperationStatus(self.status) == status
[docs]
def certification_errors(self) -> List[str]:
"""Return user-facing reasons the operation cannot yet be certified.
Currently checks that an :class:`EmissionIntensity` is attached
and that output allocations sum to exactly 100 %. Returns an
empty list when the operation is ready to certify.
"""
errors = []
if self.emission_intensity is None:
errors.append(_("Emission intensity is missing"))
unused_, allocation = self.outputs.sum_allocation()
one = Decimal(1)
allocation_percentage = (allocation * Decimal(100)).quantize(one)
if allocation != one:
errors.append(
_("%(percentage)s%% of emissions are allocated, should be 100%%.")
% {"percentage": str(allocation_percentage)}
)
return errors
[docs]
def certify(self) -> List[str]:
"""Try to certify the operation; return the blocking errors if any.
On success the status is moved to ``CERTIFIED`` and the empty
list is returned. On failure the status is left unchanged and
the caller-facing error messages from :meth:`certification_errors`
are returned.
"""
errors = self.certification_errors()
if len(errors) == 0:
self.status = TreatmentOperationStatus.CERTIFIED
self.save(update_fields=["status"])
return errors
[docs]
def reopen(self):
"""Move the operation back from ``CERTIFIED`` to ``OPEN``."""
self.status = TreatmentOperationStatus.OPEN
self.save(update_fields=["status"])
[docs]
def trigger_allocation(self):
"""Recompute ``allocated_fraction`` on every output line from its weight.
No-op when the operation uses manual allocation. Called from the
post-save / post-delete signals on
:class:`TreatmentOperationOutputLine` so the allocation stays
consistent whenever an output is added, edited, or removed.
"""
if AllocationMethod(self.allocation_method) != AllocationMethod.WEIGHT:
return
# Refresh is needed so that all outputs are returned, even those
# just inserted through save() method.
self.refresh_from_db()
# noinspection PyUnresolvedReferences
weight_sum, _ = self.outputs.sum_allocation()
updates = []
for output in self.outputs.all():
output_weight = normalize_unit(UnitCategory.MASS, output.weight)
output.allocated_fraction = output_weight.value / weight_sum.value
updates.append(output)
self.outputs.bulk_update(updates, ["allocated_fraction"])
[docs]
class TreatmentManager(models.Manager):
"""Manager that prefetches operation, inputs, outputs, and LCC values."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related(
"operation",
"operation__energy_carrier",
"operation__emitter",
)
.prefetch_related(
Prefetch("inputs", TreatmentInputLine.objects.all()),
Prefetch("outputs", TreatmentOutputLine.objects.all()),
Prefetch("lcc_values", TreatmentLCCValue.objects.get_ordered()),
)
)
[docs]
def upsert(self, records):
"""Bulk insert-or-update treatments using ``reference`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"booking_date",
"operation",
],
unique_fields=["reference"],
)
[docs]
class Treatment(UUIDModel):
"""A specific run of a :class:`TreatmentOperation` on a booking date.
Each treatment owns its own input and output lines so the realised
quantities can deviate from the recipe; emissions are posted via
:meth:`post_emission`, after which the treatment moves to
``POSTED`` and becomes effectively read-only.
"""
class Meta:
verbose_name = _("Treatment")
verbose_name_plural = _("Treatments")
constraints = [
models.UniqueConstraint(
name="uniq_treatment",
fields=["reference"],
)
]
indexes = [
models.Index(fields=["booking_date"]),
]
objects = TreatmentManager()
reference = models.CharField(
verbose_name=_("Reference"),
)
booking_date = models.DateField(
verbose_name=_("Booking Date"),
)
operation = models.ForeignKey(
TreatmentOperation,
verbose_name=_("Operation"),
on_delete=models.RESTRICT,
)
status = models.IntegerField(
choices=TreatmentStatus.choices,
default=TreatmentStatus.OPEN,
verbose_name=_("Status"),
)
def __str__(self):
return "%s - %s" % (self.reference, self.booking_date)
[docs]
def get_absolute_url(self):
return reverse("emiflow:treatment_update", args=(self.pk,))
[docs]
def is_status(self, status: TreatmentStatus) -> bool:
"""Return ``True`` when the treatment is in the given status."""
return TreatmentStatus(self.status) == status
[docs]
def copy_lines_from_operation(self):
"""Seed input and output lines from the parent operation.
Idempotent and one-sided: each side (inputs, outputs) is only
copied when no rows exist yet, so re-running the call after the
user has edited one side will not overwrite their edits.
"""
operation = self.operation
if self.inputs.all().count() == 0:
TreatmentInputLine.objects.bulk_create(
[
TreatmentInputLine(
material_id=line.material_id,
treatment=self,
weight_value=line.weight_value,
weight_unit_id=line.weight_unit_id,
)
for line in operation.inputs.all()
]
)
if self.outputs.all().count() == 0:
TreatmentOutputLine.objects.bulk_create(
[
TreatmentOutputLine(
allocated_fraction=line.allocated_fraction,
material_id=line.material_id,
treatment=self,
weight_value=line.weight_value,
weight_unit_id=line.weight_unit_id,
)
for line in operation.outputs.all()
]
)
[docs]
def post_emission(self) -> list[EmissionBookingRecord]:
"""Run the booking service, mark the treatment posted, return the records.
No-op (returns an empty list) when the treatment is not in the
``OPEN`` status — posting is a one-way transition.
"""
from ferrosoft.apps.emiflow.services.treatment import TreatmentBookingService
if not self.is_status(TreatmentStatus.OPEN):
return []
records = TreatmentBookingService().book(self)
self.status = TreatmentStatus.POSTED
self.save(update_fields=["status"])
return records
[docs]
class TreatmentLineQuerySet(models.QuerySet):
"""QuerySet adding a unit-aware ``sum_weight`` aggregation."""
[docs]
def sum_weight(self) -> DecimalWithUnit:
"""Return the sum of all line weights, normalised to metric tons."""
weights = (
self.values("weight_unit__id")
.annotate(weight_total=Sum("weight_value"))
.order_by("weight_unit__id")
)
# Total weight in Metric Ton
weight_total = DecimalWithUnit.zero()
for weight in weights:
weight_total += normalize_unit(
UnitCategory.MASS,
DecimalWithUnit.create(
weight["weight_total"], weight["weight_unit__id"]
),
)
return weight_total
[docs]
class TreatmentLineManager(models.Manager):
"""Manager using :class:`TreatmentLineQuerySet` with material/unit prefetch."""
[docs]
def get_queryset(self):
return TreatmentLineQuerySet(self.model, self._db).select_related(
"material", "weight_unit"
)
[docs]
class TreatmentLine(UUIDModel):
"""Abstract base for a material + weight line on either a treatment or operation."""
objects = TreatmentLineManager()
material: Material = models.ForeignKey(
Material,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Material"),
)
weight_value: Decimal = NormalDecimalField(
verbose_name=_("Weight"),
validators=[
MinValueValidator(Decimal(1)),
],
)
weight_unit: MeasurementUnit = models.ForeignKey(
MeasurementUnit,
on_delete=models.RESTRICT,
related_name="+",
verbose_name=_("Weight Unit"),
)
def __str__(self) -> str:
return "%s %s" % (self.material, simple_format(self.weight))
@property
def weight(self) -> DecimalWithUnit:
"""Line weight paired with its unit."""
return DecimalWithUnit(self.weight_value, self.weight_unit)
[docs]
class TreatmentOutputLineManager(TreatmentLineManager):
"""Manager that upserts output lines keyed by ``(treatment, material)``."""
[docs]
def upsert(self, records):
"""Bulk insert-or-update output lines using ``(treatment, material)`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"weight_value",
"weight_unit",
"allocated_fraction",
],
unique_fields=[
"treatment",
"material",
],
)
[docs]
class TreatmentOutputLine(TreatmentLine):
"""Material returned by a treatment."""
class Meta:
verbose_name = _("Material Output Line")
verbose_name_plural = _("Material Output Lines")
constraints = [
models.UniqueConstraint(
name="uniq_treatment_output_line",
fields=["treatment", "material"],
)
]
objects = TreatmentOutputLineManager()
treatment: Treatment = models.ForeignKey(
Treatment,
on_delete=models.CASCADE,
related_name="outputs",
verbose_name=_("Treatment"),
)
allocated_fraction = models.DecimalField(
decimal_places=4,
default=0,
max_digits=5,
verbose_name=_("Emission Allocation"),
)
[docs]
def get_absolute_url(self):
return reverse("emiflow:treatmentoutputline_update", args=[self.pk])
[docs]
class TreatmentOperationOutputLineManager(TreatmentLineManager):
"""Manager with allocation aggregation and upsert helpers."""
[docs]
def sum_allocation(self) -> tuple[DecimalWithUnit, Decimal]:
"""Return ``(total_weight_in_t, total_allocated_fraction)`` for the queryset.
Weights are summed per unit at the database layer and then
converted to metric tons in Python so a mix of input units does
not produce a meaningless total.
"""
groups = (
self.values("weight_unit__id")
.annotate(
allocation=Sum("allocated_fraction"),
weight=Sum("weight_value"),
)
.order_by("weight_unit__id")
)
weight_total = DecimalWithUnit.zero()
allocation_total = zero
for group in groups:
allocation_total += group["allocation"]
weight_total += normalize_unit(
UnitCategory.MASS,
DecimalWithUnit.create(group["weight"], group["weight_unit__id"]),
)
return weight_total, allocation_total
[docs]
def upsert(self, records):
"""Bulk insert-or-update output lines using ``(operation, material)`` as the conflict target."""
return self.bulk_create(
records,
update_conflicts=True,
update_fields=[
"weight_value",
"weight_unit",
"allocated_fraction",
],
unique_fields=[
"operation",
"material",
],
)
[docs]
class TreatmentOperationOutputLine(TreatmentLine):
"""Material returned by a treatment operation."""
class Meta:
verbose_name = _("Material Output Line")
verbose_name_plural = _("Material Output Lines")
constraints = [
models.UniqueConstraint(
name="uniq_treatment_op_output_line",
fields=["operation", "material"],
)
]
objects = TreatmentOperationOutputLineManager()
operation: TreatmentOperation = models.ForeignKey(
TreatmentOperation,
on_delete=models.CASCADE,
related_name="outputs",
verbose_name=_("Treatment Operation"),
)
allocated_fraction = models.DecimalField(
decimal_places=4,
default=0,
max_digits=5,
verbose_name=_("Emission Allocation"),
)
[docs]
def get_absolute_url(self):
return reverse("emiflow:treatmentoperationoutputline_update", args=[self.pk])
[docs]
class TreatmentLCCValue(LCCValue):
"""Custom lifecycle category values for treatments."""
treatment = models.ForeignKey(
Treatment,
on_delete=models.CASCADE,
related_name="lcc_values",
verbose_name=_("Treatment"),
)
[docs]
def get_absolute_url(self):
return reverse("emiflow:treatmentlccvalue_update", args=[self.pk])