WC2-431: Decipher how to aggregate ETL based on current tableau

plugins/wfp/admin.py

@@ -1,11 +1,5 @@
 from django.contrib import admin
-
-from .models import (
-    Beneficiary,
-    Journey,
-    Step,
-    Visit,
-)
+from .models import Beneficiary, Journey, Step, Visit, MonthlyStatistics
 
 
 @admin.register(Beneficiary)
@@ -33,10 +27,12 @@ class JourneyAdmin(admin.ModelAdmin):
         "instance_id",
         "beneficiary",
     )
+    raw_id_fields = ["beneficiary"]
     list_filter = (
         "admission_criteria",
         "admission_type",
         "nutrition_programme",
+        "beneficiary__gender",
         "programme_type",
         "start_date",
         "end_date",
@@ -56,3 +52,34 @@ class VisitAdmin(admin.ModelAdmin):
 class StepAdmin(admin.ModelAdmin):
     list_display = ("id", "assistance_type", "quantity_given", "visit")
     list_filter = ("assistance_type", "visit__journey__programme_type", "visit__journey__beneficiary__account")
+
+
+@admin.register(MonthlyStatistics)
+class MonthlyStatisticsAdmin(admin.ModelAdmin):
+    list_filter = (
+        "account",
+        "month",
+        "gender",
+        "admission_criteria",
+        "admission_type",
+        "nutrition_programme",
+        "programme_type",
+        "exit_type",
+    )
+    list_display = (
+        "id",
+        "org_unit",
+        "account",
+        "month",
+        "year",
+        "gender",
+        "admission_criteria",
+        "admission_type",
+        "nutrition_programme",
+        "programme_type",
+        "exit_type",
+        "number_visits",
+        "given_sachet_rusf",
+        "given_sachet_rutf",
+        "given_quantity_csb",
+    )

plugins/wfp/aggregate_journeys.py

@@ -0,0 +1,62 @@
+from itertools import groupby
+from operator import itemgetter
+
+
+class AggregatedJourney:
+    def group_visit_by_admission(self, current_visit, row, assistance):
+        visit_by_admission_type = groupby(list(current_visit), key=itemgetter("visit__journey__admission_type"))
+        for admission_type, visit_admission_type in visit_by_admission_type:
+            if admission_type is not None:
+                row["admission_type"] = admission_type
+
+            visit_by_admission_criteria = groupby(
+                list(visit_admission_type), key=itemgetter("visit__journey__admission_criteria")
+            )
+            for admission_criteria, visit_admission_criteria in visit_by_admission_criteria:
+                if admission_type is not None:
+                    row["admission_criteria"] = admission_criteria
+
+                all_visits = [visit for visit in visit_admission_criteria]
+                program_type = all_visits[0]["visit__journey__programme_type"]
+                row["programme_type"] = program_type
+                for visit in all_visits:
+                    if visit.get("assistance_type") == "rutf":
+                        assistance["rutf_quantity"] = assistance.get("rutf_quantity", 0) + visit.get(
+                            "quantity_given", 0
+                        )
+                    if visit.get("assistance_type") == "rusf":
+                        assistance["rusf_quantity"] = assistance.get("rusf_quantity", 0) + visit.get(
+                            "quantity_given", 0
+                        )
+                    if visit.get("assistance_type") in ["csb", "csb1", "csb2"]:
+                        assistance["csb_quantity"] = assistance.get("csb_quantity", 0) + visit.get("quantity_given", 0)
+                    if visit.get("visit__journey__exit_type") is not None:
+                        row["exit_type"] = visit.get("visit__journey__exit_type")
+                row["number_visits"] = len(all_visits)
+        return row
+
+    def group_by_period(self, visits_by_period, org_unit, all_journeys, assistance):
+        row = None
+        for period, visit in visits_by_period:
+            if period is not None:
+                month_year = period.split("/")
+                row = {"year": month_year[0], "month": month_year[1]}
+
+            visit_by_gender = groupby(list(visit), key=itemgetter("visit__journey__beneficiary__gender"))
+            for gender, current__visit_by_gender in visit_by_gender:
+                row["gender"] = gender
+
+                visits_by_nutrition_program = groupby(
+                    list(current__visit_by_gender), key=itemgetter("visit__journey__nutrition_programme")
+                )
+                for nutrition_program, current_visit in visits_by_nutrition_program:
+                    if nutrition_program is not None:
+                        row["nutrition_programme"] = nutrition_program
+                    row = self.group_visit_by_admission(current_visit, row, assistance)
+
+            row["org_unit"] = org_unit
+            row["given_sachet_rusf"] = assistance.get("rusf_quantity", 0)
+            row["given_sachet_rutf"] = assistance.get("rutf_quantity", 0)
+            row["given_quantity_csb"] = assistance.get("csb_quantity", 0)
+            all_journeys.append(row)
+        return all_journeys

plugins/wfp/common.py

@@ -2,6 +2,11 @@
 from iaso.models import *
 from datetime import datetime, timedelta, date
 from dateutil.relativedelta import *
+from itertools import groupby
+from operator import itemgetter
+from django.db.models import Value, CharField
+from django.db.models.functions import Extract, Concat
+from plugins.wfp.aggregate_journeys import AggregatedJourney
 
 
 class ETL:
@@ -10,6 +15,7 @@ def __init__(self, types=None):
 
     def delete_beneficiaries(self):
         beneficiary = Beneficiary.objects.all().delete()
+        MonthlyStatistics.objects.all().delete()
 
         print("EXISTING BENEFICIARY DELETED", beneficiary[1]["wfp.Beneficiary"])
         print("EXISTING STEPS DELETED", beneficiary[1]["wfp.Step"])
@@ -216,9 +222,7 @@ def admission_type_converter(self, admission_type):
             return "referred_from_otp_sam"
         elif admission_type == "referred_from_tsfp":
             return "referred_from_tsfp_mam"
-        elif admission_type == "referred_from_sc_itp":
-            return "referred_from_sc"
-        elif admission_type == "returned_from_sc":
+        elif admission_type in ["referred_from_sc_itp", "returned_from_sc"]:
             return "referred_from_sc"
         elif admission_type == "returnee":
             return "returned_referral"
@@ -637,3 +641,72 @@ def save_entity_journey(self, journey, beneficiary, record, entity_type):
         journey.save()
 
         return journey
+
+    def save_monthly_journey(self, monthly_journey, account):
+        monthly_Statistic = MonthlyStatistics()
+        orgUnit = OrgUnit.objects.get(id=monthly_journey.get("org_unit"))
+
+        monthly_Statistic.org_unit = orgUnit
+        monthly_Statistic.gender = monthly_journey.get("gender")
+        monthly_Statistic.month = monthly_journey.get("month")
+        monthly_Statistic.year = monthly_journey.get("year")
+        monthly_Statistic.number_visits = monthly_journey.get("number_visits")
+        monthly_Statistic.programme_type = monthly_journey.get("programme_type")
+        monthly_Statistic.nutrition_programme = monthly_journey.get("nutrition_programme")
+        monthly_Statistic.admission_type = monthly_journey.get("admission_type")
+        monthly_Statistic.admission_criteria = monthly_journey.get("admission_criteria")
+        monthly_Statistic.given_sachet_rusf = monthly_journey.get("given_sachet_rusf")
+        monthly_Statistic.given_sachet_rutf = monthly_journey.get("given_sachet_rutf")
+        monthly_Statistic.given_quantity_csb = monthly_journey.get("given_quantity_csb")
+        monthly_Statistic.exit_type = monthly_journey.get("exit_type")
+        monthly_Statistic.account = account
+
+        monthly_Statistic.save()
+
+    def journey_with_visit_and_steps_per_visit(self, account, programme):
+        aggregated_journeys = []
+        journeys = (
+            Step.objects.select_related("visit", "visit__journey", "visit__org_unit_id")
+            .filter(visit__journey__programme_type=programme, visit__journey__beneficiary__account=account)
+            .values(
+                "visit__journey__admission_type",
+                "assistance_type",
+                "instance_id",
+                "quantity_given",
+                "visit",
+                "visit__id",
+                "visit__date",
+                "visit__journey",
+                "visit__org_unit_id",
+                "visit__journey__admission_criteria",
+                "visit__journey__nutrition_programme",
+                "visit__journey__programme_type",
+                "visit__journey__end_date",
+                "visit__journey__exit_type",
+                "visit__journey__beneficiary__gender",
+                "visit__journey__beneficiary__account",
+                year=Extract("visit__date", "year"),
+                month=Extract("visit__date", "month"),
+                period=Concat(
+                    Extract("visit__date", "year"),
+                    Value("/"),
+                    Extract("visit__date", "month"),
+                    output_field=CharField(),
+                ),
+            )
+            .order_by("visit__id")
+        )
+        data_by_journey = groupby(list(journeys), key=itemgetter("visit__org_unit_id"))
+
+        for org_unit, journeys in data_by_journey:
+            visits_by_period = groupby(journeys, key=itemgetter("period"))
+            assistance = {"rutf_quantity": 0, "rusf_quantity": 0, "csb_quantity": 0}
+            aggregated_journeys = AggregatedJourney().group_by_period(
+                visits_by_period, org_unit, aggregated_journeys, assistance
+            )
+
+        for index, journey in enumerate(aggregated_journeys):
+            logger.info(
+                f"---------------------------------------- Journey N° {(index+1)} -----------------------------------"
+            )
+            self.save_monthly_journey(journey, account)

plugins/wfp/management/commands/nigeria/Under5.py

@@ -121,6 +121,9 @@ def run(self):
         existing_beneficiaries = ETL().existing_beneficiaries()
         instances = self.group_visit_by_entity(entities)
 
+        # Cleaning monthly statistics then update the table with fresh data
+        MonthlyStatistics.objects.all().filter(account=account, programme_type="U5").delete()
+
         for index, instance in enumerate(instances):
             logger.info(
                 f"---------------------------------------- Beneficiary N° {(index+1)} {instance['entity_id']}-----------------------------------"

plugins/wfp/management/commands/south_sudan/Pbwg.py

@@ -18,6 +18,9 @@ def run(self):
         existing_beneficiaries = ETL().existing_beneficiaries()
         instances = self.group_visit_by_entity(entities)
 
+        # Cleaning monthly statistics then update the table with fresh data
+        MonthlyStatistics.objects.all().filter(account=account, programme_type="PLW").delete()
+
         for index, instance in enumerate(instances):
             logger.info(
                 f"---------------------------------------- Beneficiary N° {(index+1)} {instance['entity_id']}-----------------------------------"

plugins/wfp/management/commands/south_sudan/Under5.py

@@ -145,6 +145,9 @@ def run(self):
         existing_beneficiaries = ETL().existing_beneficiaries()
         instances = self.group_visit_by_entity(entities)
 
+        # Cleaning monthly statistics then update the table with fresh data
+        MonthlyStatistics.objects.all().filter(account=account, programme_type="U5").delete()
+
         for index, instance in enumerate(instances):
             logger.info(
                 f"---------------------------------------- Beneficiary N° {(index+1)} {instance['entity_id']}-----------------------------------"