# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
from uuid import UUID
from rest_framework import serializers
from ferrosoft.apps.dataimport.convert import create_or_get_model
from ferrosoft.apps.emiflow.models import (
Treatment,
TreatmentOutputLine,
TreatmentInputLine,
TreatmentOperation,
AllocationMethod,
TreatmentOperationStatus,
EmissionIntensity,
IntensityCapableEntity,
TreatmentStatus,
TreatmentOperationInputLine,
TreatmentOperationOutputLine,
)
from ferrosoft.apps.emiflow.services.builder import BulkCreator, MaterialBuilder
from ferrosoft.apps.ferrobase.models import MeasurementUnit, Material
from ferrosoft.decimal import zero
[docs]
class TreatmentOperationSerializer(serializers.Serializer):
reference = serializers.CharField()
name = serializers.CharField()
emitter = serializers.UUIDField()
energyCarrier = serializers.UUIDField()
allocationMethod = serializers.ChoiceField(AllocationMethod.choices)
emissionIntensityOperation = serializers.DecimalField(
max_digits=12, decimal_places=4
)
emissionIntensityEnergyProvision = serializers.DecimalField(
max_digits=12, decimal_places=4
)
emissionIntensityUnitCode = serializers.CharField()
[docs]
class TreatmentOperationOutputLineSerializer(serializers.Serializer):
lineNo = serializers.IntegerField()
itemNo = serializers.CharField()
itemName = serializers.CharField()
weightValue = serializers.DecimalField(max_digits=12, decimal_places=4)
weightUnitCode = serializers.CharField()
emissionRatio = serializers.DecimalField(max_digits=5, decimal_places=4)
[docs]
class TreatmentSerializer(serializers.Serializer):
reference = serializers.CharField()
bookingDate = serializers.DateField()
operation = TreatmentOperationSerializer()
inputs = TreatmentOperationInputLineSerializer(many=True)
outputs = TreatmentOperationOutputLineSerializer(many=True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.bulk = BulkCreator(
create_order=[
TreatmentOperationInputLine,
TreatmentOperationOutputLine,
TreatmentInputLine,
TreatmentOutputLine,
]
)
[docs]
def create(self, validated_data: dict) -> Treatment:
operation = self.create_operation(validated_data["operation"])
def create_inner() -> Treatment:
return Treatment.objects.create(
reference=validated_data["reference"],
booking_date=validated_data["bookingDate"],
operation=operation,
status=TreatmentStatus.OPEN,
)
treatment = create_or_get_model(Treatment, create_inner)
for data in validated_data["inputs"]:
self.create_input_line(data, treatment, operation)
for data in validated_data["outputs"]:
self.create_output_line(data, treatment, operation)
self.bulk.upsert()
return treatment
[docs]
def create_operation(self, data: dict) -> TreatmentOperation:
def create() -> TreatmentOperation:
return TreatmentOperation.objects.create(
reference=data["reference"],
name=data["name"],
emitter_id=data["emitter"],
energy_carrier_id=data["energyCarrier"],
allocation_method=data["allocationMethod"],
status=TreatmentOperationStatus.CERTIFIED,
)
operation = create_or_get_model(TreatmentOperation, create)
def create_intensity() -> EmissionIntensity:
unit = self.get_unit(data["emissionIntensityUnitCode"])
return EmissionIntensity.objects.create(
owner_type=IntensityCapableEntity.TREATMENT_OPERATION,
owner_id=operation.pk,
enable_distance=False,
operation_value=data["emissionIntensityOperation"],
energy_provision_value=data["emissionIntensityEnergyProvision"],
unit=unit,
quantity_value=zero,
quantity_unit_id=UUID("9afba255-1aba-41ca-94af-3eafc6c53ff7"),
mass_value=zero,
mass_unit_id=UUID("10af5832-ef89-4442-b1c4-c123cf00b24b"),
distance_value=zero,
distance_unit_id=UUID("f30a3508-3814-4c6c-8810-2480174e0d33"),
)
# Intensity is passed inline in operation.
intensity = create_or_get_model(EmissionIntensity, create_intensity)
operation.emission_intensity = intensity
operation.save(update_fields=["emission_intensity"])
return operation
[docs]
def create_output_line(
self,
data: dict,
treatment: Treatment,
operation: TreatmentOperation,
):
material = create_or_get_model(Material, self.create_material(data))
unit = self.get_unit(data["weightUnitCode"])
self.bulk.add_model(
TreatmentOutputLine(
treatment=treatment,
material=material,
weight_value=data["weightValue"],
weight_unit=unit,
allocated_fraction=data["emissionRatio"],
)
)
self.bulk.add_model(
TreatmentOperationOutputLine(
operation=operation,
material=material,
weight_value=data["weightValue"],
weight_unit=unit,
allocated_fraction=data["emissionRatio"],
)
)
[docs]
@staticmethod
def create_material(data: dict):
def create():
return (
MaterialBuilder.new().code(data["itemNo"]).name(data["itemName"]).save()
)
return create
[docs]
def get_unit(self, symbol: str) -> MeasurementUnit:
unit = MeasurementUnit.objects.filter(
# Filter case-insensitive because BC CODE type converts strings
# to UPPERCASE.
symbol__iexact=symbol,
).first()
if unit is None:
raise RuntimeError(f"Unit {symbol} not found")
return unit