# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
from decimal import Decimal
from typing import Any
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http.request import HttpRequest
from django.shortcuts import redirect
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from django.views.generic import DetailView
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.emiflow.filter import TreatmentFilter, TreatmentOperationFilter
from ferrosoft.apps.emiflow.forms.treatment import (
TreatmentForm,
TreatmentInputLineForm,
TreatmentOperationForm,
TreatmentOperationInputLineForm,
TreatmentOperationOutputLineForm,
TreatmentOutputLineForm,
)
from ferrosoft.apps.emiflow.models import (
EmitterEpitype,
Treatment,
TreatmentOperation,
EmissionBookingRecord,
PredefinedLCC,
IntensityCapableEntity,
TreatmentInputLine,
TreatmentOperationInputLine,
TreatmentOperationOutputLine,
TreatmentOutputLine,
Emitter,
)
from ferrosoft.apps.emiflow.models.treatment import (
AllocationMethod,
TreatmentOperationStatus,
TreatmentStatus,
)
from ferrosoft.apps.emiflow.tables import (
AllocationEditorTable,
TreatmentInputLineTable,
TreatmentOperationTable,
TreatmentOutputLineTable,
TreatmentTable,
TreatmentOperationOutputLineTable,
TreatmentOperationInputLineTable,
TreatmentLCCValueTable,
)
from ferrosoft.apps.emiflow.views import TreatmentLCCValueCreate
from ferrosoft.apps.emiflow.views.intensity import intensity_create_url
from ferrosoft.apps.ferrobase.auth.roles import Permissions, Verb
from ferrosoft.apps.ferrobase.forms.copy import copy_form_factory, copy_fields
from ferrosoft.apps.ferrobase.models import IDSequence
from ferrosoft.apps.ferrobase.services.idgenerator import (
IdGeneratorPostgres,
IdGenerator,
)
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_transaction
from ferrosoft.apps.ferrobase.views.generic import (
CreateView,
DeleteView,
PlatformActivityMixin,
TableView,
UpdateView,
model_crud_routes,
CopyView,
)
from ferrosoft.strparse import parse_decimal_noexcept, parse_int_noexcept
[docs]
class AllocationEditView(LoginRequiredMixin, PlatformActivityMixin, DetailView):
activity_title = _("Emission Allocation")
help_topic = "treatment_operation_allocation"
model = TreatmentOperation
template_name = "emiflow/treatmentoperation/allocation.html"
[docs]
def get_context_data(self, **kwargs) -> dict[str, Any]:
return super().get_context_data(**kwargs) | {
"allocation": self.object.get_output_allocation(),
"allocation_methods": AllocationMethod,
"allocation_table": AllocationEditorTable(self.object.outputs.all()),
"proportion_visible": AllocationMethod(self.object.allocation_method)
== AllocationMethod.MANUALLY,
"proportion_visible_id": AllocationMethod.MANUALLY.value,
}
[docs]
def post(self, request: HttpRequest, *args, **kwargs):
# POST data is not accessed through Django forms API,
# because it provides no good way to access multiple values.
object: TreatmentOperation = self.get_object()
object.allocation_method = AllocationMethod(
parse_int_noexcept(request.POST.get("method")) or AllocationMethod.WEIGHT
)
object.save(update_fields=["allocation_method"])
# Accumulates output lines for which allocated_fraction is to be updated in DB.
updated_lines = []
if object.allocation_method == AllocationMethod.MANUALLY:
for line in object.outputs.all():
fraction = parse_decimal_noexcept(
request.POST.get(f"fraction[{str(line.material_id)}]")
)
if fraction is not None:
line.allocated_fraction = (fraction / Decimal(100)).quantize(
Decimal("1.0000")
)
updated_lines.append(line)
if len(updated_lines) > 0:
TreatmentOperationOutputLine.objects.bulk_update(
updated_lines, ["allocated_fraction"]
)
return redirect("emiflow:treatmentoperation_update", pk=object.pk)
[docs]
class TreatmentCreate(CreateView):
form_class = TreatmentForm
help_topic = "new_treatment"
model = Treatment
[docs]
class TreatmentRead(TableView):
filterset_class = TreatmentFilter
help_topic = "treatments"
model = Treatment
ordering = "-reference"
table_class = TreatmentTable
template_name = "emiflow/treatment/index.html"
[docs]
class TreatmentUpdate(UpdateView):
form_class = TreatmentForm
help_topic = "new_treatment"
model = Treatment
template_name = "emiflow/treatment/update.html"
[docs]
def get_context_data(self, **kwargs):
is_posted = self.object.is_status(TreatmentStatus.POSTED)
return super().get_context_data(**kwargs) | {
"disable_submit": is_posted,
"enable_posting": not is_posted,
"emission": EmissionBookingRecord.objects.sum_emissions(
PredefinedLCC.TREATMENT,
self.object.pk,
),
"input_line_table": TreatmentInputLineTable(
self.object.inputs.all(),
disable_update=is_posted,
),
"output_line_table": TreatmentOutputLineTable(
self.object.outputs.all(),
disable_update=is_posted,
),
"lcc_values_table": TreatmentLCCValueTable(
self.object.lcc_values.all(),
create_url=reverse(
"emiflow:treatmentlccvalue_create",
query={
TreatmentLCCValueCreate.parent_query: str(self.object.pk),
},
),
disable_update=is_posted,
),
}
[docs]
def post(self, request, *args, **kwargs):
# Beside normal Django form submission, there is also this
# special form instructing to post the treatment.
if request.POST.get("post_treatment", None) == "1":
self.object = self.get_object()
with tenant_db_transaction():
self.object.post_emission()
return redirect(self.object.get_absolute_url())
return super().post(request, *args, **kwargs)
[docs]
class TreatmentDelete(DeleteView):
model = Treatment
treatment_urls = model_crud_routes(
Treatment,
create_view=TreatmentCreate,
read_view=TreatmentRead,
update_view=TreatmentUpdate,
delete_view=TreatmentDelete,
)
[docs]
class TreatmentOperationCopy(CopyView):
[docs]
@staticmethod
def prepare_copy(source: TreatmentOperation, target: TreatmentOperation, **kwargs):
def copy_emission_intensity():
# Copy emission intensity as-is and let user adjust it in update view.
source_intensity = getattr(source, "emission_intensity", None)
if source_intensity is not None:
target_intensity = source_intensity.copy_with_owner(
IntensityCapableEntity.TREATMENT_OPERATION, target.pk
)
target.emission_intensity = target_intensity
target.save(update_fields=["emission_intensity"])
# Likewise for material inputs and outputs.
copies = []
for source_line in source.inputs.all():
target_line = TreatmentOperationInputLine()
copy_fields(
source_line,
target_line,
fields=[
"material",
"weight_value",
"weight_unit",
],
operation=target,
)
copies.append(target_line)
TreatmentOperationInputLine.objects.bulk_create(copies)
copies = []
for source_line in source.outputs.all():
target_line = TreatmentOperationOutputLine()
copy_fields(
source_line,
target_line,
fields=[
"material",
"weight_value",
"weight_unit",
"allocated_fraction",
],
operation=target,
)
copies.append(target_line)
TreatmentOperationOutputLine.objects.bulk_create(copies)
# Require user to certify the copied operation. This is fine UX-wise,
# since copies likely need further modification via update view.
target.status = TreatmentOperationStatus.OPEN
target.reference = IdGenerator.get_instance(
IDSequence.TREATMENT_OPERATION
).next_id()
return [copy_emission_intensity]
form_class = copy_form_factory(
TreatmentOperation,
fields=[
"name",
"emitter",
"energy_carrier",
],
fields_to_copy=["allocation_method"],
preparator=prepare_copy,
)
model = TreatmentOperation
[docs]
class TreatmentOperationCreate(CreateView):
form_class = TreatmentOperationForm
form_template = "emiflow/treatmentoperation/form.html"
help_topic = "new_treatment_operation"
model = TreatmentOperation
[docs]
class TreatmentOperationRead(TableView):
filterset_class = TreatmentOperationFilter
help_topic = "treatment_operations"
model = TreatmentOperation
ordering = "-reference"
table_class = TreatmentOperationTable
[docs]
class TreatmentOperationUpdate(UpdateView):
add_copy_action = True
form_class = TreatmentOperationForm
form_template = "emiflow/treatmentoperation/form.html"
help_topic = "new_treatment_operation"
model = TreatmentOperation
template_name = "emiflow/treatmentoperation/update.html"
[docs]
def get_context_data(self, **kwargs):
cert_errors = self.object.certification_errors()
is_certified = self.object.is_status(TreatmentOperationStatus.CERTIFIED)
return super().get_context_data(**kwargs) | {
"allocation_method_description": AllocationMethod(
self.object.allocation_method
).describe(),
"certification_errors": cert_errors,
"disable_submit": is_certified,
"disable_update": is_certified,
"enable_certification": self.object.is_status(TreatmentOperationStatus.OPEN)
and len(cert_errors) == 0,
"is_certified": is_certified,
"input_line_table": TreatmentOperationInputLineTable(
self.object.inputs.all(),
create_url=reverse(
"emiflow:treatmentoperation_inputline_create",
kwargs={"operation": self.object.pk},
),
disable_update=is_certified,
),
"intensity_create_url": intensity_create_url(
IntensityCapableEntity.TREATMENT_OPERATION,
self.object.pk,
self.object.energy_carrier_id,
EmitterEpitype(self.object.emitter.epitype),
),
"output_line_table": TreatmentOperationOutputLineTable(
self.object.outputs.all(),
create_url=reverse(
"emiflow:treatmentoperation_outputline_create",
kwargs={"operation": self.object.pk},
),
disable_update=is_certified,
),
}
[docs]
def post(self, request, *args, **kwargs):
# Beside normal Django form submission, there are also these
# special forms instructing to certify or reopen operation.
if request.POST.get("certify", None) == "1":
self.object = self.get_object()
self.object.certify()
return redirect(self.object.get_absolute_url())
if request.POST.get("reopen", None) == "1":
self.object = self.get_object()
self.object.reopen()
return redirect(self.object.get_absolute_url())
return super().post(request, *args, **kwargs)
[docs]
class TreatmentOperationDelete(DeleteView):
help_topic = "treatment_operations"
model = TreatmentOperation
treatment_operation_urls = model_crud_routes(
TreatmentOperation,
copy_view=TreatmentOperationCopy,
create_view=TreatmentOperationCreate,
read_view=TreatmentOperationRead,
update_view=TreatmentOperationUpdate,
delete_view=TreatmentOperationDelete,
)
treatment_input_line_urls = model_crud_routes(
TreatmentInputLine,
# Views other than UPDATE are not of interest,
# as lines are listed inline in Treatment view,
# created by copying from TreamentOperation,
# and deletion is undesired.
update_view=TreatmentInputLineUpdate,
)
[docs]
class TreatmentOutputLineUpdate(UpdateView):
delete_view = None
form_class = TreatmentOutputLineForm
help_topic = "treatment_output_material"
model = TreatmentOutputLine
related_links_default = False
[docs]
def cancel_view(self):
return self.object.treatment.get_absolute_url()
[docs]
def get_success_url(self) -> str:
return self.object.treatment.get_absolute_url()
treatment_output_line_urls = model_crud_routes(
TreatmentOutputLine,
# Views other than UPDATE are not of interest,
# as lines are listed inline in Treatment view,
# created by copying from TreamentOperation,
# and deletion is undesired.
update_view=TreatmentOutputLineUpdate,
)
treatment_operation_input_line_urls = model_crud_routes(
TreatmentOperationInputLine,
create_view=TreatmentOperationInputLineCreate,
update_view=TreatmentOperationInputLineUpdate,
delete_view=TreatmentOperationInputLineDelete,
)
[docs]
class TreatmentOperationOutputLineCreate(CreateView):
form_class = TreatmentOperationOutputLineForm
help_topic = "treatment_operation_output_material"
model = TreatmentOperationOutputLine
[docs]
def cancel_view(self):
return reverse(
"emiflow:treatmentoperation_update",
kwargs={"pk": self.request.resolver_match.kwargs["operation"]},
)
[docs]
def get_success_url(self):
return self.object.operation.get_absolute_url()
[docs]
class TreatmentOperationOutputLineUpdate(UpdateView):
form_class = TreatmentOperationOutputLineForm
help_topic = "treatment_operation_output_material"
model = TreatmentOperationOutputLine
related_links_default = False
[docs]
def cancel_view(self):
return self.object.operation.get_absolute_url()
[docs]
def get_success_url(self):
return self.object.operation.get_absolute_url()
[docs]
class TreatmentOperationOutputLineDelete(DeleteView):
help_topic = "treatment_operation_output_material"
model = TreatmentOperationOutputLine
[docs]
def cancel_view(self):
return self.object.operation.get_absolute_url()
[docs]
def get_success_url(self):
return self.object.operation.get_absolute_url()
treatment_operation_output_line_urls = model_crud_routes(
TreatmentOperationOutputLine,
create_view=TreatmentOperationOutputLineCreate,
update_view=TreatmentOperationOutputLineUpdate,
delete_view=TreatmentOperationOutputLineDelete,
)