Source code for ferrosoft.apps.emiflow.services.pcf

#  Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""Product Carbon Footprint (PCF) calculation across all chains for a material.

The :class:`ProductCarbonFootprintService` aggregates transport
emissions for every :class:`TransportChain` that carries a given
:class:`Material`, normalises units across chains, and reports the
emission-per-tonne intensity in ``kgCO2e``.
"""
from dataclasses import dataclass
from decimal import Decimal
from typing import List

from django.utils.translation import gettext as _

from ferrosoft.apps.emiflow.models import TransportChain
from ferrosoft.apps.emiflow.services.emissions import TransportChainCalculator
from ferrosoft.apps.ferrobase.models import Material
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_connection
from ferrosoft.decimal import zero




@dataclass
class LifecycleResult:
    """One row of a PCF table: a labelled lifecycle contribution."""

    label: str
    value: DecimalWithUnit





class ProductCarbonFootprintService:
    """Calculate the Product Carbon Footprint of one material across all transports.

    Uses a raw SQL query to find the transport chains carrying the
    material, then runs the chain calculator against each one and
    divides the summed emissions by the summed weight to express the
    result as ``kgCO2e`` per weight unit.
    """

    _CHAIN_QUERY = """SELECT tc.id
    FROM emiflow_transportchain tc
    JOIN public.emiflow_freightitem ef on tc.id = ef.chain_id
    WHERE ef.material_id = %s
    GROUP BY tc.id;
    """

    def __init__(self):
        self.calculator = TransportChainCalculator()



    def calculate_for(self, material: Material) -> List[LifecycleResult]:
        """Compute the PCF for ``material`` and return one transport row.

        Returns an empty list when no chains carry the material. Raises
        :class:`RuntimeError` when the source chains mix weight or
        emission units, since combining mismatched units would silently
        falsify the result.
        """
        # Collect chain IDs
        chains = []
        with tenant_db_connection().cursor() as cursor:
            for result in cursor.execute(self._CHAIN_QUERY, [str(material.pk)]):
                chains.append(result[0])

        sum_emission = zero
        sum_quantity = zero
        weight_unit = None
        emission_unit = None
        for chain in TransportChain.objects.filter(pk__in=chains):
            result = self.calculator.calculate_for(chain)

            # Ensure uniform units.
            if weight_unit is None:
                weight_unit = result.weight.try_unit().symbol
            elif weight_unit != result.weight.try_unit().symbol:
                raise RuntimeError("Different weight units")

            if emission_unit is None:
                emission_unit = result.emission_total.try_unit().symbol
            elif emission_unit != result.emission_total.try_unit().symbol:
                raise RuntimeError("Different emission units")

            sum_emission += result.emission_total.value
            sum_quantity += result.weight.value

        if sum_quantity == zero:
            lc_transport = zero
        else:
            lc_transport = Decimal(sum_emission / sum_quantity)

        if weight_unit is None or emission_unit is None:
            return []

        # Convert to kgCO2e, as it is more convenient for the eye.
        if emission_unit == "gCO2e":
            emission_unit = "kgCO2e"
            lc_transport /= Decimal(1000)

        return [
            LifecycleResult(
                label=_("Transport"),
                value=DecimalWithUnit.create(
                    lc_transport.quantize(Decimal("1.00")),
                    "%s/%s" % (emission_unit, weight_unit),
                ),
            )
        ]