from typing import NamedTuple from django.db.models import Case, Count, IntegerField, Sum, Value, When from django.db.models.functions import Coalesce from vbv_lernwelt.core.admin import User from vbv_lernwelt.course.models import CourseCompletion, CourseCompletionStatus from vbv_lernwelt.learnpath.models import LearningUnit from vbv_lernwelt.self_evaluation_feedback.models import ( CourseCompletionFeedback, SelfEvaluationFeedback, ) class AssessmentCounts(NamedTuple): pass_count: int fail_count: int unknown_count: int @property def total_count(self): return self.pass_count + self.fail_count + self.unknown_count def get_self_evaluation_feedback_counts( feedback: SelfEvaluationFeedback, ): course_completion_feedback = CourseCompletionFeedback.objects.filter( feedback=feedback ).aggregate( pass_count=Coalesce( Sum( Case( When( feedback_assessment=CourseCompletionStatus.SUCCESS.value, then=Value(1), ), output_field=IntegerField(), ) ), Value(0), ), fail_count=Coalesce( Sum( Case( When( feedback_assessment=CourseCompletionStatus.FAIL.value, then=Value(1), ), output_field=IntegerField(), ) ), Value(0), ), unknown_count=Coalesce( Sum( Case( When( feedback_assessment=CourseCompletionStatus.UNKNOWN.value, then=Value(1), ), output_field=IntegerField(), ) ), Value(0), ), ) return AssessmentCounts( pass_count=course_completion_feedback.get("pass_count", 0), fail_count=course_completion_feedback.get("fail_count", 0), unknown_count=course_completion_feedback.get("unknown_count", 0), ) def get_self_assessment_counts( learning_unit: LearningUnit, user: User ) -> AssessmentCounts: performance_criteria = learning_unit.performancecriteria_set.all() completion_counts = CourseCompletion.objects.filter( page__in=performance_criteria, user=user ).aggregate( pass_count=Count( Case( When(completion_status=CourseCompletionStatus.SUCCESS.value, then=1), output_field=IntegerField(), ) ), fail_count=Count( Case( When(completion_status=CourseCompletionStatus.FAIL.value, then=1), output_field=IntegerField(), ) ), unknown_count=Count( Case( When(completion_status=CourseCompletionStatus.UNKNOWN.value, then=1), output_field=IntegerField(), ) ), ) pass_count = completion_counts.get("pass_count", 0) fail_count = completion_counts.get("fail_count", 0) unknown_count = completion_counts.get("unknown_count", 0) # not yet completed performance criteria are unknown if pass_count + fail_count + unknown_count < performance_criteria.count(): unknown_count += performance_criteria.count() - ( pass_count + fail_count + unknown_count ) return AssessmentCounts( pass_count=pass_count, fail_count=fail_count, unknown_count=unknown_count, ) def calculate_aggregate(counts: [AssessmentCounts]): return AssessmentCounts( pass_count=sum(x.pass_count for x in counts), fail_count=sum(x.fail_count for x in counts), unknown_count=sum(x.unknown_count for x in counts), )