""" 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.functions import NumGeometries from django.db.models import Count, Sum, Q from codelist.models import KonovaCode from codelist.settings import CODELIST_LAW_ID from compensation.models import Compensation, Payment, EcoAccountDeduction from intervention.models import Intervention from konova.models import Geometry class TimespanReport: office_id = -1 class InterventionReport: queryset = Intervention.objects.none() queryset_checked = Intervention.objects.none() queryset_recorded = Intervention.objects.none() # 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 def __init__(self, id: str): self.queryset = Intervention.objects.filter( responsible__conservation_office__id=id, deleted=None, ) self.queryset_checked = self.queryset.filter( checked__isnull=False ) self.queryset_recorded = self.queryset.filter( recorded__isnull=False ) self._create_report() 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_registration_office_unb = Compensation.objects.none() queryset_registration_office_unb_checked = Compensation.objects.none() queryset_registration_office_unb_recorded = Compensation.objects.none() 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() 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() 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): self.queryset = Compensation.objects.filter( intervention__responsible__conservation_office__id=id, deleted=None, ) self.queryset_checked = self.queryset.filter( intervention__checked__isnull=False ) self.queryset_recorded = self.queryset.filter( intervention__recorded__isnull=False ) self._create_report() 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( num=NumGeometries("geom") ).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_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_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, ) def __init__(self, office_id: str): self.office_id = office_id self.intervention_report = self.InterventionReport(self.office_id) self.compensation_report = self.CompensationReport(self.office_id)