# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""Emission calculation and booking for treatments.
:class:`TreatmentEmissionCalculator` multiplies the treatment's output
mass by the operation's emission intensity (for both *operation* and
*energy provision* categories) to derive the total emission.
:class:`TreatmentBookingService` then splits that total across output
lines using each line's ``allocated_fraction`` and creates one
:class:`EmissionBookingRecord` per output and per attached
:class:`TreatmentLCCValue`.
"""
from dataclasses import dataclass
from decimal import Decimal
from typing import Mapping
from django.utils.translation import gettext as _
from ferrosoft.apps.emiflow.models import (
Treatment,
EmissionCategory,
EmissionBookingRecord,
PredefinedLCC,
LifecycleCategory,
)
from ferrosoft.apps.ferrobase.models import UnitCategory
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.services.conversion import (
normalize_unit,
NormalizationError,
)
[docs]
@dataclass
class EmissionResult:
"""One category's calculated emission paired with the intensity that produced it."""
emission: DecimalWithUnit
intensity: DecimalWithUnit
[docs]
@dataclass
class TreatmentResult:
"""Per-treatment activity (output mass) and per-category emission results."""
activity: DecimalWithUnit
emissions: Mapping[EmissionCategory, EmissionResult]
[docs]
class TreatmentEmissionCalculator:
"""Compute operation and energy-provision emissions for a treatment."""
[docs]
def calculate_for(self, treatment: Treatment) -> TreatmentResult:
"""Calculate emissions for ``treatment``.
Raises:
NormalizationError: When the operation has no emission
intensity, or the intensity unit cannot be normalised
to ``kgCO2e/t``.
"""
op = treatment.operation
# Material mass in metric ton.
m = treatment.outputs.all().sum_weight()
# Emission intensities for both operation and energy provision.
intensity = op.emission_intensity
if intensity is None:
raise NormalizationError(_("Emission intensity is undefined"))
ei_op = self._normalize_intensity(intensity.operation_intensity)
ei_ep = self._normalize_intensity(intensity.energy_provision_intensity)
return TreatmentResult(
activity=m,
emissions={
EmissionCategory.OPERATION: EmissionResult(
emission=normalize_unit(
UnitCategory.EMISSION_VALUE, self._emission(m, ei_op)
),
intensity=ei_op,
),
EmissionCategory.ENERGY_PROVISION: EmissionResult(
emission=normalize_unit(
UnitCategory.EMISSION_VALUE, self._emission(m, ei_ep)
),
intensity=ei_ep,
),
},
)
@staticmethod
def _normalize_intensity(intensity: DecimalWithUnit) -> DecimalWithUnit:
"""Normalise intensity to ``kgCO2e/t``; raise for unsupported units."""
match intensity.try_unit().symbol:
case "kgCO2e/t":
return intensity
case "gCO2e/t":
return DecimalWithUnit.create(
intensity.value / Decimal(1000), "kgCO2e/t"
)
case _:
raise NormalizationError(_("Unsupported intensity unit"))
@staticmethod
def _emission(m: DecimalWithUnit, ei: DecimalWithUnit) -> DecimalWithUnit:
"""Multiply mass by intensity to produce an emission value.
Args:
m: Mass in metric tons.
ei: Emission intensity in ``kgCO2e/t``.
Returns:
DecimalWithUnit: Emission in ``kgCO2e``.
"""
return DecimalWithUnit.create(
m.value * ei.value,
ei.try_unit().fraction[0],
)
[docs]
class TreatmentBookingService:
"""Persist :class:`EmissionBookingRecord` rows for a treatment.
For every output line, the treatment's total emission is multiplied
by the line's ``allocated_fraction`` to produce one booking record;
each attached :class:`TreatmentLCCValue` is similarly split across
output lines.
"""
[docs]
def book(self, treatment: Treatment) -> list[EmissionBookingRecord]:
"""Calculate emissions for ``treatment`` and bulk-upsert booking records.
Returns the list of persisted records (one per output line plus
one per LCC value per output line).
"""
# For booking records, operation and total emissions are needed.
result = TreatmentEmissionCalculator().calculate_for(treatment)
em_operation = result.emissions[EmissionCategory.OPERATION].emission
em_energy_provision = result.emissions[
EmissionCategory.ENERGY_PROVISION
].emission
em_total = em_operation + em_energy_provision
# Load predefined LCC to write its possibly customized name into
# emission booking records.
treatment_category = LifecycleCategory.objects.get(
pk=PredefinedLCC.TREATMENT.value
)
# LCC values are all loaded at once into memory, because they are few
# and are looped over multiple times.
lcc_values = list(treatment.lcc_values.all())
# Records are accumulated to be saved in bulk.
records = []
# A booking record has to be created for each output line,
# and emissions partially assigned according to allocation.
for line in treatment.outputs.all():
material = line.material
weight = normalize_unit(UnitCategory.MASS, line.weight)
frac_operation = em_operation * line.allocated_fraction
frac_total = em_total * line.allocated_fraction
records.append(
EmissionBookingRecord(
booking_date=treatment.booking_date,
lifecycle_category=treatment_category.pk,
lifecycle_category_name=treatment_category.name,
entity_id=treatment.pk,
entity_reference=treatment.reference,
simulation=False,
emission_total_value=frac_total.value,
emission_operation_value=frac_operation.value,
emission_unit_id=frac_total.try_unit().pk,
emission_unit_symbol=frac_total.try_unit().symbol,
weight_value=weight.value,
weight_unit_id=weight.try_unit().pk,
weight_unit_symbol=weight.try_unit().symbol,
material_id=[material.pk],
material_code=[material.code],
material_name=[material.name],
)
)
# Also create booking records for each related LCC.
for lcc in lcc_values:
total_emission = (
normalize_unit(UnitCategory.EMISSION_VALUE, lcc.total_emission)
* line.allocated_fraction
)
operation_emission = (
normalize_unit(UnitCategory.EMISSION_VALUE, lcc.operation_emission)
* line.allocated_fraction
)
records.append(
EmissionBookingRecord(
booking_date=treatment.booking_date,
lifecycle_category=lcc.lifecycle_category.pk,
lifecycle_category_name=lcc.lifecycle_category.name,
entity_id=treatment.pk,
entity_reference=treatment.reference,
simulation=False,
emission_total_value=total_emission.value,
emission_operation_value=operation_emission.value,
emission_unit_id=total_emission.try_unit().pk,
emission_unit_symbol=total_emission.try_unit().symbol,
weight_value=weight.value,
weight_unit_id=weight.try_unit().pk,
weight_unit_symbol=weight.try_unit().symbol,
material_id=[material.pk],
material_code=[material.code],
material_name=[material.name],
)
)
return EmissionBookingRecord.objects.upsert(records)