""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: michel.peltriaux@sgdnord.rlp.de Created on: 18.10.21 """ from django.contrib.gis.db.models import MultiPolygonField from django.contrib.gis.db.models.functions import NumGeometries from django.db.models import Count, Sum, Q from django.db.models.functions import Cast from analysis.settings import LKOMPVZVO_PUBLISH_DATE from codelist.models import KonovaCode from codelist.settings import CODELIST_LAW_ID from compensation.models import Compensation, Payment, EcoAccountDeduction, EcoAccount from intervention.models import Intervention from konova.models import Geometry from konova.sub_settings.django_settings import BASE_DIR, DEFAULT_DATE_FORMAT class TimespanReport: """ Holds multiple report elements for a timespan report """ office_id = -1 date_from = -1 date_to = -1 # Excel map is used to map a cell value ("A1") to an attribute excel_map = {} excel_template_path = f"{BASE_DIR}/analysis/utils/excel/excel_report.xlsx" class InterventionReport: queryset = Intervention.objects.none() queryset_checked = Intervention.objects.none() queryset_recorded = Intervention.objects.none() queryset_count = -1 queryset_checked_count = -1 queryset_recorded_count = -1 # Law related law_sum = -1 law_sum_checked = -1 law_sum_recorded = -1 evaluated_laws = None # Compensations related compensation_sum = -1 compensation_sum_checked = -1 compensation_sum_recorded = -1 payment_sum = -1 payment_sum_checked = -1 payment_sum_recorded = -1 deduction_sum = -1 deduction_sum_checked = -1 deduction_sum_recorded = -1 excel_map = {} def __init__(self, id: str, date_from: str, date_to: str): self.queryset = Intervention.objects.filter( responsible__conservation_office__id=id, legal__registration_date__gt=LKOMPVZVO_PUBLISH_DATE, deleted=None, created__timestamp__gte=date_from, created__timestamp__lte=date_to, ) self.queryset_checked = self.queryset.filter( checked__isnull=False ) self.queryset_recorded = self.queryset.filter( recorded__isnull=False ) self.queryset_count = self.queryset.count() self.queryset_checked_count = self.queryset_checked.count() self.queryset_recorded_count = self.queryset_recorded.count() self._create_report() self._define_excel_map() def _define_excel_map(self): """ Define the excel map, which holds values for each placeholder used in the template Returns: """ self.excel_map = { "i_checked": self.queryset_checked_count, "i_recorded": self.queryset_recorded_count, "i_total": self.queryset_count, "i_compensations_checked": self.compensation_sum_checked, "i_compensations_recorded": self.compensation_sum_recorded, "i_compensations_total": self.compensation_sum, "i_payments_recorded": self.payment_sum_recorded, "i_payments_checked": self.payment_sum_checked, "i_payments_total": self.payment_sum, "i_deductions_recorded": self.deduction_sum_recorded, "i_deductions_checked": self.deduction_sum_checked, "i_deductions_total": self.deduction_sum, "i_laws_iter": { "iterable": self.evaluated_laws, "attrs": [ "short_name", "num_checked", "num_recorded", "num", ] }, "i_laws_checked": self.law_sum_checked, "i_laws_recorded": self.law_sum_recorded, "i_laws_total": self.law_sum, } def _create_report(self): """ Creates all report information Returns: """ self._evaluate_laws() self._evaluate_compensations() def _evaluate_laws(self): """ Analyzes the intervention-law distribution Returns: """ # Count interventions based on law # Fetch all KonovaCodes for laws, sorted alphabetically laws = KonovaCode.objects.filter( is_archived=False, is_leaf=True, code_lists__in=[CODELIST_LAW_ID], ).order_by( "long_name" ) # Fetch all law ids which are used by any .legal object of an intervention object intervention_laws_total = self.queryset.values_list("legal__laws__id") intervention_laws_checked = self.queryset.filter(checked__isnull=False).values_list("legal__laws__id") intervention_laws_recorded = self.queryset.filter(recorded__isnull=False).values_list( "legal__laws__id") # Count how often which law id appears in the above list, return only the long_name of the law and the resulting # count (here 'num'). This is for keeping the db fetch as small as possible # Compute the sum for total, checked and recorded self.evaluated_laws = laws.annotate( num=Count("id", filter=Q(id__in=intervention_laws_total)), num_checked=Count("id", filter=Q(id__in=intervention_laws_checked)), num_recorded=Count("id", filter=Q(id__in=intervention_laws_recorded)), ).values_list("short_name", "long_name", "num_checked", "num_recorded", "num", named=True) self.law_sum = self.evaluated_laws.aggregate(sum_num=Sum("num"))["sum_num"] self.law_sum_checked = self.evaluated_laws.aggregate(sum_num_checked=Sum("num_checked"))["sum_num_checked"] self.law_sum_recorded = self.evaluated_laws.aggregate(sum_num_recorded=Sum("num_recorded"))["sum_num_recorded"] def _evaluate_compensations(self): """ Analyzes the types of compensation distribution Returns: """ # Count all compensations comps = Compensation.objects.filter( intervention__in=self.queryset ) self.compensation_sum = comps.count() self.compensation_sum_checked = comps.filter(intervention__checked__isnull=False).count() self.compensation_sum_recorded = comps.filter(intervention__recorded__isnull=False).count() # Count all payments payments = Payment.objects.filter( intervention__in=self.queryset ) self.payment_sum = payments.count() self.payment_sum_checked = payments.filter(intervention__checked__isnull=False).count() self.payment_sum_recorded = payments.filter(intervention__recorded__isnull=False).count() # Count all deductions deductions = EcoAccountDeduction.objects.filter( intervention__in=self.queryset ) self.deduction_sum = deductions.count() self.deduction_sum_checked = deductions.filter(intervention__checked__isnull=False).count() self.deduction_sum_recorded = deductions.filter(intervention__recorded__isnull=False).count() class CompensationReport: queryset = Compensation.objects.none() queryset_checked = Compensation.objects.none() queryset_recorded = Compensation.objects.none() queryset_count = -1 queryset_checked_count = -1 queryset_recorded_count = -1 queryset_registration_office_unb = Compensation.objects.none() queryset_registration_office_unb_checked = Compensation.objects.none() queryset_registration_office_unb_recorded = Compensation.objects.none() queryset_registration_office_unb_count = -1 queryset_registration_office_unb_checked_count = -1 queryset_registration_office_unb_recorded_count = -1 num_single_surfaces_total_unb = -1 queryset_registration_office_tbp = Compensation.objects.none() queryset_registration_office_tbp_checked = Compensation.objects.none() queryset_registration_office_tbp_recorded = Compensation.objects.none() queryset_registration_office_tbp_count = -1 queryset_registration_office_tbp_checked_count = -1 queryset_registration_office_tbp_recorded_count = -1 num_single_surfaces_total_tbp = -1 queryset_registration_office_other = Compensation.objects.none() queryset_registration_office_other_checked = Compensation.objects.none() queryset_registration_office_other_recorded = Compensation.objects.none() queryset_registration_office_other_count = -1 queryset_registration_office_other_checked_count = -1 queryset_registration_office_other_recorded_count = -1 num_single_surfaces_total_other = -1 num_single_surfaces_total = -1 num_single_surfaces_recorded = -1 # Code list id for 'Träger der Bauleitplanung' parent id_tbp = 1943695 # Code list id for 'untere Naturschutzbehörde' id_unb = 1943087 # Code list id for 'obere Naturschutzbehörde' id_onb = 1943084 def __init__(self, id: str, date_from: str, date_to: str): self.queryset = Compensation.objects.filter( intervention__responsible__conservation_office__id=id, intervention__legal__registration_date__gt=LKOMPVZVO_PUBLISH_DATE, deleted=None, intervention__created__timestamp__gte=date_from, intervention__created__timestamp__lte=date_to, ) self.queryset_checked = self.queryset.filter( intervention__checked__isnull=False ) self.queryset_recorded = self.queryset.filter( intervention__recorded__isnull=False ) self.queryset_count = self.queryset.count() self.queryset_checked_count = self.queryset_checked.count() self.queryset_recorded_count = self.queryset_recorded.count() self._create_report() self._define_excel_map() def _define_excel_map(self): """ Define the excel map, which holds values for each placeholder used in the template Returns: """ self.excel_map = { "c_unb_checked": self.queryset_registration_office_unb_checked_count, "c_unb_recorded": self.queryset_registration_office_unb_recorded_count, "c_unb": self.queryset_registration_office_unb_count, "c_surfaces_unb": self.num_single_surfaces_total_unb, "c_tbp_checked": self.queryset_registration_office_tbp_checked_count, "c_tbp_recorded": self.queryset_registration_office_tbp_recorded_count, "c_tbp": self.queryset_registration_office_tbp_count, "c_surfaces_tbp": self.num_single_surfaces_total_tbp, "c_other_checked": self.queryset_registration_office_other_checked_count, "c_other_recorded": self.queryset_registration_office_other_recorded_count, "c_other": self.queryset_registration_office_other_count, "c_surfaces_other": self.num_single_surfaces_total_other, "c_checked": self.queryset_checked_count, "c_recorded": self.queryset_recorded_count, "c_total": self.queryset_count, "c_surfaces": self.num_single_surfaces_total, } def _create_report(self): """ Creates all report information Returns: """ self._evaluate_compensation_responsibility() self._evaluate_surfaces() def _evaluate_surfaces(self): """ Evaluates the surfaces of compensation Multipolygon fields Returns: """ # Evaluate all surfaces ids = self.queryset.values_list("geometry_id") self.num_single_surfaces_total = self._count_geometry_surfaces(ids) # Evaluate surfaces where the conservation office is the registration office as well ids = self.queryset_registration_office_unb.values_list("geometry_id") self.num_single_surfaces_total_unb = self._count_geometry_surfaces(ids) # Evaluates surfaces where the registration office is a Träger Bauleitplanung ids = self.queryset_registration_office_tbp.values_list("geometry_id") self.num_single_surfaces_total_tbp = self._count_geometry_surfaces(ids) # Evaluates surfaces where any other registration office is responsible ids = self.queryset_registration_office_other.values_list("geometry_id") self.num_single_surfaces_total_other = self._count_geometry_surfaces(ids) def _count_geometry_surfaces(self, ids: list): """ Wraps counting of geometry surfaces from a given list of ids Args: ids (list): List of geometry ids Returns: """ # Now select all geometries matching the ids # Then perform a ST_NumGeometries variant over all geometries # Then sum up all of the calculated surface numbers return Geometry.objects.filter( id__in=ids ).annotate( geom_cast=Cast("geom", MultiPolygonField()) ).annotate( num=NumGeometries("geom_cast") ).aggregate( num_geoms=Sum("num") )["num_geoms"] or 0 def _evaluate_compensation_responsibility(self): """ Evaluates compensations based on different responsibility areas unb -> Untere Naturschutzbehörde Holds entries where conservation_office and registration_office basically are the same tbp -> Träger Bauleitplanung Holds entries where registration_office is a Träger der Bauleitplanung other -> Other registration offices Holds all other entries Returns: """ self.queryset_registration_office_unb = self.queryset.filter( intervention__responsible__registration_office__parent__id=self.id_unb ) self.queryset_registration_office_unb_recorded = self.queryset_registration_office_unb.filter( intervention__recorded__isnull=False, ) self.queryset_registration_office_unb_checked = self.queryset_registration_office_unb.filter( intervention__checked__isnull=False, ) self.queryset_registration_office_unb_count = self.queryset_registration_office_unb.count() self.queryset_registration_office_unb_checked_count = self.queryset_registration_office_unb_checked.count() self.queryset_registration_office_unb_recorded_count = self.queryset_registration_office_unb_recorded.count() self.queryset_registration_office_tbp = self.queryset.filter( intervention__responsible__registration_office__parent__id=self.id_tbp ) self.queryset_registration_office_tbp_recorded = self.queryset_registration_office_tbp.filter( intervention__recorded__isnull=False, ) self.queryset_registration_office_tbp_checked = self.queryset_registration_office_tbp.filter( intervention__checked__isnull=False, ) self.queryset_registration_office_tbp_count = self.queryset_registration_office_tbp.count() self.queryset_registration_office_tbp_checked_count = self.queryset_registration_office_tbp_checked.count() self.queryset_registration_office_tbp_recorded_count = self.queryset_registration_office_tbp_recorded.count() self.queryset_registration_office_other = self.queryset.exclude( Q(id__in=self.queryset_registration_office_tbp) | Q(id__in=self.queryset_registration_office_unb) ) self.queryset_registration_office_other_recorded = self.queryset_registration_office_other.filter( intervention__recorded__isnull=False, ) self.queryset_registration_office_other_checked = self.queryset_registration_office_other.filter( intervention__checked__isnull=False, ) self.queryset_registration_office_other_count = self.queryset_registration_office_other.count() self.queryset_registration_office_other_checked_count = self.queryset_registration_office_other_checked.count() self.queryset_registration_office_other_recorded_count = self.queryset_registration_office_other_recorded.count() class EcoAccountReport: queryset = EcoAccount.objects.none() queryset_recorded = EcoAccount.objects.none() queryset_count = -1 queryset_recorded_count = -1 queryset_deductions = EcoAccountDeduction.objects.none() queryset_deductions_recorded = EcoAccountDeduction.objects.none() queryset_has_deductions = EcoAccountDeduction.objects.none() queryset_deductions_count = -1 queryset_deductions_recorded_count = -1 queryset_has_deductions_count = -1 # Total size of deductions deductions_sq_m = -1 recorded_deductions_sq_m = -1 def __init__(self, id: str, date_from: str, date_to: str): # First fetch all eco account for this office self.queryset = EcoAccount.objects.filter( responsible__conservation_office__id=id, deleted=None, created__timestamp__gte=date_from, created__timestamp__lte=date_to, ) self.queryset_recorded = self.queryset.filter( recorded__isnull=False ) # Fetch all related deductions self.queryset_deductions = EcoAccountDeduction.objects.filter( account__id__in=self.queryset.values_list("id") ) # Fetch deductions for interventions which are already recorded self.queryset_deductions_recorded = self.queryset_deductions.filter( intervention__recorded__isnull=False ) self.queryset_count = self.queryset.count() self.queryset_recorded_count = self.queryset_recorded.count() self.queryset_deductions_count = self.queryset_deductions.count() self.queryset_deductions_recorded_count = self.queryset_deductions_recorded.count() self.queryset_has_deductions_count = self.queryset_has_deductions.count() self._create_report() self._define_excel_map() def _define_excel_map(self): """ Define the excel map, which holds values for each placeholder used in the template Returns: """ self.excel_map = { "acc_total": self.queryset_count, "acc_recorded": self.queryset_recorded_count, "acc_deduc_recorded": self.queryset_deductions_recorded_count, "acc_deduc_surface_recorded": self.recorded_deductions_sq_m, "acc_deduc_total": self.queryset_deductions_count, "acc_deduc_surface_total": self.deductions_sq_m, } def _create_report(self): """ Creates all report information Returns: """ self._evaluate_deductions() def _evaluate_deductions(self): self.deductions_sq_m = self.queryset_deductions.aggregate( sum=Sum("surface") )["sum"] or 0 self.recorded_deductions_sq_m = self.queryset_deductions_recorded.aggregate( sum=Sum("surface") )["sum"] or 0 class OldDataReport: """ Evaluates 'old data' (registered (zugelassen) before 16.06.2018) """ queryset_intervention = Intervention.objects.none() queryset_intervention_recorded = Intervention.objects.none() queryset_intervention_count = -1 queryset_intervention_recorded_count = -1 queryset_comps = Compensation.objects.none() queryset_comps_recorded = Compensation.objects.none() queryset_comps_count = -1 queryset_comps_recorded_count = -1 queryset_acc = EcoAccount.objects.none() queryset_acc_recorded = EcoAccount.objects.none() queryset_acc_count = -1 queryset_acc_recorded_count = -1 def __init__(self, id: str, date_from: str, date_to: str): self.queryset_intervention = Intervention.objects.filter( legal__registration_date__lte=LKOMPVZVO_PUBLISH_DATE, responsible__conservation_office__id=id, deleted=None, created__timestamp__gte=date_from, created__timestamp__lte=date_to, ) self.queryset_intervention_recorded = self.queryset_intervention.filter( recorded__isnull=False ) self.queryset_intervention_count = self.queryset_intervention.count() self.queryset_intervention_recorded_count = self.queryset_intervention_recorded.count() self.queryset_comps = Compensation.objects.filter( intervention__in=self.queryset_intervention ) self.queryset_comps_recorded = Compensation.objects.filter( intervention__in=self.queryset_intervention_recorded, ) self.queryset_comps_count = self.queryset_comps.count() self.queryset_comps_recorded_count = self.queryset_comps_recorded.count() self.queryset_acc = EcoAccount.objects.filter( #legal__registration_date__lte=LKOMPVZVO_PUBLISH_DATE, responsible__conservation_office__id=id, deleted=None, created__timestamp__gte=date_from, created__timestamp__lte=date_to, ) self.queryset_acc_recorded = self.queryset_acc.filter( recorded__isnull=False, ) self.queryset_acc_count = self.queryset_acc.count() self.queryset_acc_recorded_count = self.queryset_acc_recorded.count() self._define_excel_map() def _define_excel_map(self): """ Define the excel map, which holds values for each placeholder used in the template Returns: """ self.excel_map = { "old_i_recorded": self.queryset_intervention_recorded_count, "old_i_total": self.queryset_intervention_count, "old_c_recorded": self.queryset_comps_recorded_count, "old_c_total": self.queryset_comps_count, "old_ea_recorded": self.queryset_acc_recorded_count, "old_ea_total": self.queryset_acc_count, } def __init__(self, office_id: str, date_from: str, date_to: str): self.office_id = office_id self.date_from = date_from self.date_to = date_to self.intervention_report = self.InterventionReport(self.office_id, date_from, date_to) self.compensation_report = self.CompensationReport(self.office_id, date_from, date_to) self.eco_account_report = self.EcoAccountReport(self.office_id, date_from, date_to) self.old_data_report = self.OldDataReport(self.office_id, date_from, date_to) # Build excel map self.excel_map = { "date_from": date_from.strftime(DEFAULT_DATE_FORMAT), "date_to": date_to.strftime(DEFAULT_DATE_FORMAT), } self.excel_map.update(self.intervention_report.excel_map) self.excel_map.update(self.compensation_report.excel_map) self.excel_map.update(self.eco_account_report.excel_map) self.excel_map.update(self.old_data_report.excel_map)