|
2 | 2 | Python APIs exposed for the progress tracking functionality of the course home API. |
3 | 3 | """ |
4 | 4 |
|
| 5 | +from __future__ import annotations |
| 6 | + |
5 | 7 | from django.contrib.auth import get_user_model |
6 | 8 | from opaque_keys.edx.keys import CourseKey |
| 9 | +from openedx.core.lib.grade_utils import round_away_from_zero |
| 10 | +from xmodule.graders import ShowCorrectness |
| 11 | +from datetime import datetime, timezone |
7 | 12 |
|
8 | 13 | from lms.djangoapps.courseware.courses import get_course_blocks_completion_summary |
| 14 | +from dataclasses import dataclass, field |
9 | 15 |
|
10 | 16 | User = get_user_model() |
11 | 17 |
|
12 | 18 |
|
| 19 | +@dataclass |
| 20 | +class _AssignmentBucket: |
| 21 | + """Holds scores and visibility info for one assignment type. |
| 22 | +
|
| 23 | + Attributes: |
| 24 | + assignment_type: Full assignment type name from the grading policy (for example, "Homework"). |
| 25 | + num_total: The total number of assignments expected to contribute to the grade before any |
| 26 | + drop-lowest rules are applied. |
| 27 | + last_grade_publish_date: The most recent date when grades for all assignments of assignment_type |
| 28 | + are released and included in the final grade. |
| 29 | + scores: Per-subsection fractional scores (each value is ``earned / possible`` and falls in |
| 30 | + the range 0–1). While awaiting published content we pad the list with zero placeholders |
| 31 | + so that its length always matches ``num_total`` until real scores replace them. |
| 32 | + visibilities: Mirrors ``scores`` index-for-index and records whether each subsection's |
| 33 | + correctness feedback is visible to the learner (``True``), hidden (``False``), or not |
| 34 | + yet populated (``None`` when the entry is a placeholder). |
| 35 | + included: Tracks whether each subsection currently counts toward the learner's grade as |
| 36 | + determined by ``SubsectionGrade.show_grades``. Values follow the same convention as |
| 37 | + ``visibilities`` (``True`` / ``False`` / ``None`` placeholders). |
| 38 | + assignments_created: Count of real subsections inserted into the bucket so far. Once this |
| 39 | + reaches ``num_total``, all placeholder entries have been replaced with actual data. |
| 40 | + """ |
| 41 | + assignment_type: str |
| 42 | + num_total: int |
| 43 | + last_grade_publish_date: datetime |
| 44 | + scores: list[float] = field(default_factory=list) |
| 45 | + visibilities: list[bool | None] = field(default_factory=list) |
| 46 | + included: list[bool | None] = field(default_factory=list) |
| 47 | + assignments_created: int = 0 |
| 48 | + |
| 49 | + @classmethod |
| 50 | + def with_placeholders(cls, assignment_type: str, num_total: int, now: datetime): |
| 51 | + """Create a bucket prefilled with placeholder (empty) entries.""" |
| 52 | + return cls( |
| 53 | + assignment_type=assignment_type, |
| 54 | + num_total=num_total, |
| 55 | + last_grade_publish_date=now, |
| 56 | + scores=[0] * num_total, |
| 57 | + visibilities=[None] * num_total, |
| 58 | + included=[None] * num_total, |
| 59 | + ) |
| 60 | + |
| 61 | + def add_subsection(self, score: float, is_visible: bool, is_included: bool): |
| 62 | + """Add a subsection’s score and visibility, replacing a placeholder if space remains.""" |
| 63 | + if self.assignments_created < self.num_total: |
| 64 | + if self.scores: |
| 65 | + self.scores.pop(0) |
| 66 | + if self.visibilities: |
| 67 | + self.visibilities.pop(0) |
| 68 | + if self.included: |
| 69 | + self.included.pop(0) |
| 70 | + self.scores.append(score) |
| 71 | + self.visibilities.append(is_visible) |
| 72 | + self.included.append(is_included) |
| 73 | + self.assignments_created += 1 |
| 74 | + |
| 75 | + def drop_lowest(self, num_droppable: int): |
| 76 | + """Remove the lowest scoring subsections, up to the provided num_droppable.""" |
| 77 | + while num_droppable > 0 and self.scores: |
| 78 | + idx = self.scores.index(min(self.scores)) |
| 79 | + self.scores.pop(idx) |
| 80 | + self.visibilities.pop(idx) |
| 81 | + self.included.pop(idx) |
| 82 | + num_droppable -= 1 |
| 83 | + |
| 84 | + def hidden_state(self) -> str: |
| 85 | + """Return whether kept scores are all, some, or none hidden.""" |
| 86 | + if not self.visibilities: |
| 87 | + return 'none' |
| 88 | + all_hidden = all(v is False for v in self.visibilities) |
| 89 | + some_hidden = any(v is False for v in self.visibilities) |
| 90 | + if all_hidden: |
| 91 | + return 'all' |
| 92 | + if some_hidden: |
| 93 | + return 'some' |
| 94 | + return 'none' |
| 95 | + |
| 96 | + def averages(self) -> tuple[float, float]: |
| 97 | + """Compute visible and included averages over kept scores. |
| 98 | +
|
| 99 | + Visible average uses only grades with visibility flag True in numerator; denominator is total |
| 100 | + number of kept scores (mirrors legacy behavior). Included average uses only scores that are |
| 101 | + marked included (show_grades True) in numerator with same denominator. |
| 102 | +
|
| 103 | + Returns: |
| 104 | + (earned_visible, earned_all) tuple of floats (0-1 each). |
| 105 | + """ |
| 106 | + if not self.scores: |
| 107 | + return 0.0, 0.0 |
| 108 | + visible_scores = [s for i, s in enumerate(self.scores) if self.visibilities[i]] |
| 109 | + included_scores = [s for i, s in enumerate(self.scores) if self.included[i]] |
| 110 | + earned_visible = (sum(visible_scores) / len(self.scores)) if self.scores else 0.0 |
| 111 | + earned_all = (sum(included_scores) / len(self.scores)) if self.scores else 0.0 |
| 112 | + return earned_visible, earned_all |
| 113 | + |
| 114 | + |
| 115 | +class _AssignmentTypeGradeAggregator: |
| 116 | + """Collects and aggregates subsection grades by assignment type.""" |
| 117 | + |
| 118 | + def __init__(self, course_grade, grading_policy: dict, has_staff_access: bool): |
| 119 | + """Initialize with course grades, grading policy, and staff access flag.""" |
| 120 | + self.course_grade = course_grade |
| 121 | + self.grading_policy = grading_policy |
| 122 | + self.has_staff_access = has_staff_access |
| 123 | + self.now = datetime.now(timezone.utc) |
| 124 | + self.policy_map = self._build_policy_map() |
| 125 | + self.buckets: dict[str, _AssignmentBucket] = {} |
| 126 | + |
| 127 | + def _build_policy_map(self) -> dict: |
| 128 | + """Convert grading policy into a lookup of assignment type → policy info.""" |
| 129 | + policy_map = {} |
| 130 | + for policy in self.grading_policy.get('GRADER', []): |
| 131 | + policy_map[policy.get('type')] = { |
| 132 | + 'weight': policy.get('weight', 0.0), |
| 133 | + 'short_label': policy.get('short_label', ''), |
| 134 | + 'num_droppable': policy.get('drop_count', 0), |
| 135 | + 'num_total': policy.get('min_count', 0), |
| 136 | + } |
| 137 | + return policy_map |
| 138 | + |
| 139 | + def _bucket_for(self, assignment_type: str) -> _AssignmentBucket: |
| 140 | + """Get or create a score bucket for the given assignment type.""" |
| 141 | + bucket = self.buckets.get(assignment_type) |
| 142 | + if bucket is None: |
| 143 | + num_total = self.policy_map.get(assignment_type, {}).get('num_total', 0) or 0 |
| 144 | + bucket = _AssignmentBucket.with_placeholders(assignment_type, num_total, self.now) |
| 145 | + self.buckets[assignment_type] = bucket |
| 146 | + return bucket |
| 147 | + |
| 148 | + def collect(self): |
| 149 | + """Gather subsection grades into their respective assignment buckets.""" |
| 150 | + for chapter in self.course_grade.chapter_grades.values(): |
| 151 | + for subsection_grade in chapter.get('sections', []): |
| 152 | + if not getattr(subsection_grade, 'graded', False): |
| 153 | + continue |
| 154 | + assignment_type = getattr(subsection_grade, 'format', '') or '' |
| 155 | + if not assignment_type: |
| 156 | + continue |
| 157 | + graded_total = getattr(subsection_grade, 'graded_total', None) |
| 158 | + earned = getattr(graded_total, 'earned', 0.0) if graded_total else 0.0 |
| 159 | + possible = getattr(graded_total, 'possible', 0.0) if graded_total else 0.0 |
| 160 | + earned = 0.0 if earned is None else earned |
| 161 | + possible = 0.0 if possible is None else possible |
| 162 | + score = (earned / possible) if possible else 0.0 |
| 163 | + is_visible = ShowCorrectness.correctness_available( |
| 164 | + subsection_grade.show_correctness, subsection_grade.due, self.has_staff_access |
| 165 | + ) |
| 166 | + is_included = subsection_grade.show_grades(self.has_staff_access) |
| 167 | + bucket = self._bucket_for(assignment_type) |
| 168 | + bucket.add_subsection(score, is_visible, is_included) |
| 169 | + visibilities_with_due_dates = [ShowCorrectness.PAST_DUE, ShowCorrectness.NEVER_BUT_INCLUDE_GRADE] |
| 170 | + if subsection_grade.show_correctness in visibilities_with_due_dates: |
| 171 | + if subsection_grade.due and subsection_grade.due > bucket.last_grade_publish_date: |
| 172 | + bucket.last_grade_publish_date = subsection_grade.due |
| 173 | + |
| 174 | + def build_results(self) -> dict: |
| 175 | + """Apply drops, compute averages, and return aggregated results and total grade.""" |
| 176 | + final_grades = 0.0 |
| 177 | + rows = [] |
| 178 | + for assignment_type, bucket in self.buckets.items(): |
| 179 | + policy = self.policy_map.get(assignment_type, {}) |
| 180 | + bucket.drop_lowest(policy.get('num_droppable', 0)) |
| 181 | + earned_visible, earned_all = bucket.averages() |
| 182 | + weight = policy.get('weight', 0.0) |
| 183 | + short_label = policy.get('short_label', '') |
| 184 | + row = { |
| 185 | + 'type': assignment_type, |
| 186 | + 'weight': weight, |
| 187 | + 'average_grade': round_away_from_zero(earned_visible, 4), |
| 188 | + 'weighted_grade': round_away_from_zero(earned_visible * weight, 4), |
| 189 | + 'short_label': short_label, |
| 190 | + 'num_droppable': policy.get('num_droppable', 0), |
| 191 | + 'last_grade_publish_date': bucket.last_grade_publish_date, |
| 192 | + 'has_hidden_contribution': bucket.hidden_state(), |
| 193 | + } |
| 194 | + final_grades += earned_all * weight |
| 195 | + rows.append(row) |
| 196 | + rows.sort(key=lambda r: r['weight']) |
| 197 | + return {'results': rows, 'final_grades': round_away_from_zero(final_grades, 4)} |
| 198 | + |
| 199 | + def run(self) -> dict: |
| 200 | + """Execute full pipeline (collect + aggregate) returning final payload.""" |
| 201 | + self.collect() |
| 202 | + return self.build_results() |
| 203 | + |
| 204 | + |
| 205 | +def aggregate_assignment_type_grade_summary( |
| 206 | + course_grade, |
| 207 | + grading_policy: dict, |
| 208 | + has_staff_access: bool = False, |
| 209 | +) -> dict: |
| 210 | + """ |
| 211 | + Aggregate subsection grades by assignment type and return summary data. |
| 212 | + Args: |
| 213 | + course_grade: CourseGrade object containing chapter and subsection grades. |
| 214 | + grading_policy: Dictionary representing the course's grading policy. |
| 215 | + has_staff_access: Boolean indicating if the user has staff access to view all grades. |
| 216 | + Returns: |
| 217 | + Dictionary with keys: |
| 218 | + results: list of per-assignment-type summary dicts |
| 219 | + final_grades: overall weighted contribution (float, 4 decimal rounding) |
| 220 | + """ |
| 221 | + aggregator = _AssignmentTypeGradeAggregator(course_grade, grading_policy, has_staff_access) |
| 222 | + return aggregator.run() |
| 223 | + |
| 224 | + |
13 | 225 | def calculate_progress_for_learner_in_course(course_key: CourseKey, user: User) -> dict: |
14 | 226 | """ |
15 | 227 | Calculate a given learner's progress in the specified course run. |
|
0 commit comments