models.py

            verbose_name=_("Page Title"), null=True, blank=True)

    bookmarked = models.BooleanField(default=False,
            help_text=_("A user-facing 'marking' feature to allow participants to "
                "easily return to pages that still need their attention."),
            verbose_name=_("Bookmarked"))

    class Meta:
        verbose_name = _("Flow page data")
        verbose_name_plural = _("Flow page data")

    def __str__(self) -> str:
        # flow page data
        return (_("Data for page '%(group_id)s/%(page_id)s' "
                "(page ordinal %(page_ordinal)s) in %(flow_session)s") % {
                    "group_id": self.group_id,
                    "page_id": self.page_id,
                    "page_ordinal": self.page_ordinal,
                    "flow_session": self.flow_session})

    # Django's templates are a little daft. No arithmetic--really?
    def previous_ordinal(self):
        return self.page_ordinal - 1

    def next_ordinal(self):
        return self.page_ordinal + 1

    def human_readable_ordinal(self):
        return self.page_ordinal + 1

# }}}


# {{{ flow page visit

class FlowPageVisit(models.Model):
    # This is redundant (because the FlowSession is available through
    # page_data), but it helps the admin site understand the link
    # and provide editing.
    flow_session = models.ForeignKey(FlowSession, db_index=True,
            verbose_name=_("Flow session"), on_delete=models.CASCADE)

    page_data = models.ForeignKey(FlowPageData, db_index=True,
            verbose_name=_("Page data"), on_delete=models.CASCADE)
    visit_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Visit time"))
    remote_address = models.GenericIPAddressField(null=True, blank=True,
            verbose_name=_("Remote address"))

    user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
            blank=True, related_name="visitor",
            verbose_name=_("User"), on_delete=models.SET_NULL)
    impersonated_by = models.ForeignKey(settings.AUTH_USER_MODEL,
            null=True, blank=True, related_name="impersonator",
            verbose_name=_("Impersonated by"), on_delete=models.SET_NULL)

    is_synthetic = models.BooleanField(default=False,
            help_text=_("Synthetic flow page visits are generated for "
            "unvisited pages once a flow is finished. This is needed "
            "since grade information is attached to a visit, and it "
            "needs a place to go."),
            verbose_name=_("Is synthetic"))

    answer = JSONField(null=True, blank=True,
            # Show correct characters in admin for non ascii languages.
            dump_kwargs={"ensure_ascii": False},
            # Translators: "Answer" is a Noun.
            verbose_name=_("Answer"))

    # is_submitted_answer may seem redundant with answers being
    # non-NULL, but it isn't. This supports saved (but as
    # yet ungraded) answers.

    # NULL means it's not an answer at all.
    #   (Should coincide with 'answer is None')
    # True means it's a final, submitted answer fit for grading.
    # False means it's just a saved answer.
    is_submitted_answer = models.BooleanField(
            # Translators: determine whether the answer is a final,
            # submitted answer fit for grading.
            verbose_name=_("Is submitted answer"),
            null=True)

    def __str__(self) -> str:
        result = (
                # Translators: flow page visit
                _("'%(group_id)s/%(page_id)s' in '%(session)s' "
                "on %(time)s")
                % {"group_id": self.page_data.group_id,
                    "page_id": self.page_data.page_id,
                    "session": self.flow_session,
                    "time": self.visit_time})

        if self.answer is not None:
            # Translators: flow page visit: if an answer is
            # provided by user then append the string.
            result += str(_(" (with answer)"))

        return result

    class Meta:
        verbose_name = _("Flow page visit")
        verbose_name_plural = _("Flow page visits")
        # These must be distinguishable, to figure out what came later.
        unique_together = (("page_data", "visit_time"),)

    def get_most_recent_grade(self) -> FlowPageVisitGrade | None:

        grades = self.grades.order_by("-grade_time")[:1]

        for grade in grades:
            return grade

        return None

    def get_most_recent_feedback(self):
        grade = self.get_most_recent_grade()

        if grade is None:
            return None
        else:
            return get_feedback_for_grade(grade)

    def is_impersonated(self):
        if self.impersonated_by:
            return True
        else:
            return False

# }}}


#  {{{ flow page visit grade

class FlowPageVisitGrade(models.Model):
    visit = models.ForeignKey(FlowPageVisit, related_name="grades",
            verbose_name=_("Visit"), on_delete=models.CASCADE)

    # NULL means "autograded"
    grader = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, blank=True,
            verbose_name=_("Grader"), on_delete=models.SET_NULL)
    grade_time = models.DateTimeField(db_index=True, default=now,
            verbose_name=_("Grade time"))

    graded_at_git_commit_sha = models.CharField(
            max_length=200, null=True, blank=True,
            verbose_name=_("Graded at git commit SHA"))

    grade_data = JSONField(null=True, blank=True,
            # Show correct characters in admin for non ascii languages.
            dump_kwargs={"ensure_ascii": False},
            verbose_name=_("Grade data"))

    # This data should be recomputable, but we'll cache it here,
    # because it might be very expensive (container-launch expensive
    # for code questions, for example) to recompute.

    max_points = models.FloatField(null=True, blank=True,
            # Translators: max point in grade
            help_text=_("Point value of this question when receiving "
            "full credit."),
            verbose_name=_("Max points"))
    correctness = models.FloatField(null=True, blank=True,
            # Translators: correctness in grade
            help_text=_("Real number between zero and one (inclusively) "
            "indicating the degree of correctness of the answer."),
            verbose_name=_("Correctness"))

    # This JSON object has fields corresponding to
    # :class:`course.page.AnswerFeedback`, except for
    # :attr:`course.page.AnswerFeedback.correctness`, which is stored
    # separately for efficiency.

    feedback = JSONField(null=True, blank=True,
            # Show correct characters in admin for non ascii languages.
            dump_kwargs={"ensure_ascii": False},
            # Translators: "Feedback" stands for the feedback of answers.
            verbose_name=_("Feedback"))

    def percentage(self) -> float | None:
        if self.correctness is not None:
            return 100*self.correctness
        else:
            return None

    def value(self):
        if self.correctness is not None and self.max_points is not None:
            return self.correctness * self.max_points
        else:
            return None

    class Meta:
        verbose_name = _("Flow page visit grade")
        verbose_name_plural = _("Flow page visit grades")
        # These must be distinguishable, to figure out what came later.
        unique_together = (("visit", "grade_time"),)

        ordering = ("visit", "grade_time")

    def __str__(self) -> str:
        # information on FlowPageVisitGrade class
        # Translators: return the information of the grade of a user
        # by percentage.
        return _("grade of %(visit)s: %(percentage)s") % {
                "visit": self.visit, "percentage": self.percentage()}

# }}}


# {{{ bulk feedback

class FlowPageBulkFeedback(models.Model):
    # We're only storing one of these per page, because
    # they're 'bulk' (i.e. big, like plots or program output)
    page_data = models.OneToOneField(FlowPageData,
            verbose_name=_("Page data"), on_delete=models.CASCADE)
    grade = models.ForeignKey(FlowPageVisitGrade,
            verbose_name=_("Grade"), on_delete=models.CASCADE)

    bulk_feedback = JSONField(null=True, blank=True,
            # Show correct characters in admin for non ascii languages.
            dump_kwargs={"ensure_ascii": False},
            verbose_name=_("Bulk feedback"))


BULK_FEEDBACK_FILENAME_KEY = "_rl_stor_fn"


def update_bulk_feedback(page_data: FlowPageData, grade: FlowPageVisitGrade,
        bulk_feedback_json: Any) -> None:

    import json
    import zlib
    compressed_bulk_json_str = zlib.compress(
            json.dumps(bulk_feedback_json).encode("utf-8"))

    from django.db import transaction
    with transaction.atomic():
        try:
            fp_bulk_feedback = FlowPageBulkFeedback.objects.get(page_data=page_data)

            if (isinstance(fp_bulk_feedback.bulk_feedback, dict)
                    and (BULK_FEEDBACK_FILENAME_KEY
                        in fp_bulk_feedback.bulk_feedback)):
                storage_fn_to_delete = fp_bulk_feedback.bulk_feedback[
                        BULK_FEEDBACK_FILENAME_KEY]

                def delete_bulk_fb_file():
                    print(f"DELETING {storage_fn_to_delete}!")
                    settings.RELATE_BULK_STORAGE.delete(storage_fn_to_delete)

                transaction.on_commit(delete_bulk_fb_file)

        except ObjectDoesNotExist:
            fp_bulk_feedback = FlowPageBulkFeedback(page_data=page_data)

        # Half the sector size on Linux
        if len(compressed_bulk_json_str) >= 256:
            username = "anon"
            flow_session = page_data.flow_session
            if flow_session.participation is not None:
                username = flow_session.participation.user.username

            fn_pattern = (
                    "bulk-feedback/"
                    f"{flow_session.course.identifier}/"
                    f"{flow_session.flow_id}/"
                    f"{page_data.page_id}/"
                    f"{username}"
                    f".json_zlib")

            from django.core.files.base import ContentFile
            saved_name = settings.RELATE_BULK_STORAGE.save(
                    fn_pattern,
                    ContentFile(compressed_bulk_json_str))

            bulk_feedback_json = {BULK_FEEDBACK_FILENAME_KEY: saved_name}

        fp_bulk_feedback.grade = grade
        fp_bulk_feedback.bulk_feedback = bulk_feedback_json
        fp_bulk_feedback.save()


def get_feedback_for_grade(
        grade: FlowPageVisitGrade | None) -> AnswerFeedback | None:

    if grade is None:
        return None

    try:
        bulk_feedback_json = FlowPageBulkFeedback.objects.get(
                page_data=grade.visit.page_data,
                grade=grade).bulk_feedback
    except ObjectDoesNotExist:
        bulk_feedback_json = None

    if (bulk_feedback_json is not None
            and isinstance(bulk_feedback_json, dict)
            and (BULK_FEEDBACK_FILENAME_KEY in bulk_feedback_json)):
        import json
        import zlib
        try:
            with settings.RELATE_BULK_STORAGE.open(
                    bulk_feedback_json[BULK_FEEDBACK_FILENAME_KEY]
                    ) as inf:
                bulk_feedback_json = json.loads(
                        zlib.decompress(inf.read()).decode("utf-8"))
        except FileNotFoundError:
            bulk_feedback_json = None

    from course.page.base import AnswerFeedback
    return AnswerFeedback.from_json(grade.feedback, bulk_feedback_json)

# }}}


# {{{ deprecated flow rule exception stuff

def validate_stipulations(stip):  # pragma: no cover (deprecated and not tested)
    if stip is None:
        return

    if not isinstance(stip, dict):
        raise ValidationError(_("stipulations must be a dictionary"))
    allowed_keys = {"credit_percent", "allowed_session_count"}
    if not set(stip.keys()) <= allowed_keys:
        raise ValidationError(
                string_concat(
                    _("unrecognized keys in stipulations"),
                    ": %s")
                % ", ".join(set(stip.keys()) - allowed_keys))

    if "credit_percent" in stip and not isinstance(
            stip["credit_percent"], int | float):
        raise ValidationError(_("credit_percent must be a float"))
    if ("allowed_session_count" in stip
            and (
                not isinstance(stip["allowed_session_count"], int)
                or stip["allowed_session_count"] < 0)):
        raise ValidationError(
                _("'allowed_session_count' must be a non-negative integer"))


class FlowAccessException(models.Model):  # pragma: no cover (deprecated and not tested)  # noqa
    # deprecated

    participation = models.ForeignKey(Participation, db_index=True,
            verbose_name=_("Participation"), on_delete=models.CASCADE)
    flow_id = models.CharField(max_length=200, blank=False, null=False,
            verbose_name=_("Flow ID"))
    expiration = models.DateTimeField(blank=True, null=True,
            verbose_name=_("Expiration"))

    stipulations = JSONField(blank=True, null=True,
            # Translators: help text for stipulations in FlowAccessException
            # (deprecated)
            help_text=_("A dictionary of the same things that can be added "
            "to a flow access rule, such as allowed_session_count or "
            "credit_percent. If not specified here, values will default "
            "to the stipulations in the course content."),
            validators=[validate_stipulations],
            dump_kwargs={"ensure_ascii": False},
            verbose_name=_("Stipulations"))

    creator = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
            verbose_name=_("Creator"), on_delete=models.SET_NULL)
    creation_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Creation time"))

    is_sticky = models.BooleanField(
            default=False,
            # Translators: deprecated
            help_text=_("Check if a flow started under this "
            "exception rule set should stay "
            "under this rule set until it is expired."),
            # Translators: deprecated
            verbose_name=_("Is sticky"))

    comment = models.TextField(blank=True, null=True,
            verbose_name=_("Comment"))

    def __str__(self) -> str:
        return (
                # Translators: flow access exception in admin (deprecated)
                _("Access exception for '%(user)s' to '%(flow_id)s' "
                "in '%(course)s'") %
                {
                    "user": self.participation.user,
                    "flow_id": self.flow_id,
                    "course": self.participation.course
                    })


class FlowAccessExceptionEntry(models.Model):  # pragma: no cover (deprecated and not tested)  # noqa
    # deprecated

    exception = models.ForeignKey(FlowAccessException,
            related_name="entries",
            verbose_name=_("Exception"), on_delete=models.CASCADE)
    permission = models.CharField(max_length=50,
            choices=FLOW_PERMISSION_CHOICES,
            verbose_name=_("Permission"))

    class Meta:
        # Translators: FlowAccessExceptionEntry (deprecated)
        verbose_name_plural = _("Flow access exception entries")

    def __str__(self) -> str:
        return self.permission

# }}}


# {{{ flow rule exception

class FlowRuleException(models.Model):
    flow_id = models.CharField(max_length=200, blank=False, null=False,
            verbose_name=_("Flow ID"))
    participation = models.ForeignKey(Participation, db_index=True,
            verbose_name=_("Participation"), on_delete=models.CASCADE)
    expiration = models.DateTimeField(blank=True, null=True,
            verbose_name=_("Expiration"))

    creator = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
            verbose_name=_("Creator"), on_delete=models.SET_NULL)
    creation_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Creation time"))

    comment = models.TextField(blank=True, null=True,
            verbose_name=_("Comment"))

    kind = models.CharField(max_length=50, blank=False, null=False,
            choices=FLOW_RULE_KIND_CHOICES,
            verbose_name=_("Kind"))
    rule = YAMLField(blank=False, null=False,
            verbose_name=_("Rule"))
    active = models.BooleanField(default=True,
            verbose_name=pgettext_lazy(
                "Is the flow rule exception activated?", "Active"))

    def __str__(self) -> str:
        return (
                # Translators: For FlowRuleException
                _("%(kind)s exception %(exception_id)s for '%(user)s' to "
                "'%(flow_id)s' in '%(course)s'")
                % {
                    "kind": self.kind,
                    "user": self.participation.user,
                    "flow_id": self.flow_id,
                    "course": self.participation.course,
                    "exception_id":
                        " id %d" % self.id if self.id is not None else ""})

    def clean(self) -> None:
        super().clean()

        if self.kind not in dict(FLOW_RULE_KIND_CHOICES).keys():
            raise ValidationError(
                # Translators: the rule refers to FlowRuleException rule
                string_concat(_("invalid exception rule kind"), ": ", self.kind))

        if (self.kind == flow_rule_kind.grading
                and self.expiration is not None):
            raise ValidationError(_("grading rules may not expire"))

        from course.content import (
            get_course_commit_sha,
            get_course_repo,
            get_flow_desc,
        )
        from course.validation import (
            ValidationContext,
            ValidationError as ContentValidationError,
            validate_session_access_rule,
            validate_session_grading_rule,
            validate_session_start_rule,
        )
        from relate.utils import dict_to_struct
        rule = dict_to_struct(self.rule)

        with get_course_repo(self.participation.course) as repo:
            commit_sha = get_course_commit_sha(
                    self.participation.course, self.participation)
            ctx = ValidationContext(
                    repo=repo,
                    commit_sha=commit_sha)

            flow_desc = get_flow_desc(repo,
                    self.participation.course,
                    self.flow_id, commit_sha)

        tags: list = []
        grade_identifier = None
        if hasattr(flow_desc, "rules"):
            tags = cast(list, getattr(flow_desc.rules, "tags", []))
            grade_identifier = flow_desc.rules.grade_identifier

        try:
            if self.kind == flow_rule_kind.start:
                validate_session_start_rule(ctx, str(self), rule, tags)
            elif self.kind == flow_rule_kind.access:
                validate_session_access_rule(ctx, str(self), rule, tags)
            elif self.kind == flow_rule_kind.grading:
                validate_session_grading_rule(
                        ctx, str(self), rule, tags,
                        grade_identifier)
            else:  # pragma: no cover. This won't happen
                raise ValueError("invalid exception rule kind")

        except ContentValidationError as e:
            # the rule refers to FlowRuleException rule
            raise ValidationError(
                string_concat(_("invalid existing_session_rules"), ": ", str(e)))

    class Meta:
        verbose_name = _("Flow rule exception")
        verbose_name_plural = _("Flow rule exceptions")

# }}}


# {{{ grading

class GradingOpportunity(models.Model):
    course = models.ForeignKey(Course,
            verbose_name=_("Course"), on_delete=models.CASCADE)

    identifier = models.CharField(max_length=200, blank=False, null=False,
            # Translators: format of identifier for GradingOpportunity
            help_text=_("A symbolic name for this grade. "
            "lower_case_with_underscores, no spaces."),
            verbose_name=_("Grading opportunity ID"),
            validators=[
                RegexValidator(
                    "^"+GRADING_OPP_ID_REGEX+"$",
                    message=_(
                        "Identifier may only contain letters, "
                        "numbers, and hyphens ('-').")),
                    ])
    name = models.CharField(max_length=200, blank=False, null=False,
            # Translators: name for GradingOpportunity
            help_text=_("A human-readable identifier for the grade."),
            verbose_name=_("Grading opportunity name"))
    flow_id = models.CharField(max_length=200, blank=True, null=True,
            help_text=_("Flow identifier that this grading opportunity "
            "is linked to, if any"),
            verbose_name=_("Flow ID"))

    aggregation_strategy = models.CharField(max_length=20,
            choices=GRADE_AGGREGATION_STRATEGY_CHOICES,
            # Translators: strategy on how the grading of multiple sessioins
            # are aggregated.
            verbose_name=_("Aggregation strategy"))

    due_time = models.DateTimeField(default=None, blank=True, null=True,
            verbose_name=_("Due time"))
    creation_time = models.DateTimeField(default=now,
            verbose_name=_("Creation time"))

    shown_in_grade_book = models.BooleanField(default=True,
            verbose_name=_("Shown in grade book"))
    shown_in_participant_grade_book = models.BooleanField(default=True,
            verbose_name=_("Shown in student grade book"))
    result_shown_in_participant_grade_book = models.BooleanField(default=True,
            verbose_name=_("Result shown in student grade book"))

    page_scores_in_participant_gradebook = models.BooleanField(default=False,
            verbose_name=_("Scores for individual pages are shown "
                "in the participants' grade book"))
    hide_superseded_grade_history_before = models.DateTimeField(
            verbose_name=_("Hide superseded grade history before"),
            blank=True, null=True,
            help_text=_(
                "Grade changes dated before this date that are "
                "superseded by later grade changes will not be shown to "
                "participants. "
                "This can help avoid discussions about pre-release grading "
                "adjustments. "
                "May be blank. In that case, the entire grade history is "
                "shown."))

    class Meta:
        verbose_name = _("Grading opportunity")
        verbose_name_plural = _("Grading opportunities")
        ordering = ("course", "due_time", "identifier")
        unique_together = (("course", "identifier"),)

    def __str__(self) -> str:
        return (
                # Translators: For GradingOpportunity
                _("%(opportunity_name)s (%(opportunity_id)s) in %(course)s")
                % {
                    "opportunity_name": self.name,
                    "opportunity_id": self.identifier,
                    "course": self.course})

    def get_aggregation_strategy_descr(self):
        return dict(GRADE_AGGREGATION_STRATEGY_CHOICES).get(
                self.aggregation_strategy)


class GradeChange(models.Model):
    """Per 'grading opportunity', each participant may accumulate multiple grades
    that are aggregated according to :attr:`GradingOpportunity.aggregation_strategy`.

    In addition, for each opportunity, grade changes are grouped by their 'attempt'
    identifier, where later grades with the same :attr:`attempt_id` supersede earlier
    ones.
    """
    opportunity = models.ForeignKey(GradingOpportunity,
            verbose_name=_("Grading opportunity"), on_delete=models.CASCADE)

    participation = models.ForeignKey(Participation,
            verbose_name=_("Participation"), on_delete=models.CASCADE)

    state = models.CharField(max_length=50,
            choices=GRADE_STATE_CHANGE_CHOICES,
            # Translators: something like 'status'.
            verbose_name=_("State"))

    attempt_id = models.CharField(max_length=50, null=True, blank=True,
            default="main",
            # Translators: help text of "attempt_id" in GradeChange class
            help_text=_("Grade changes are grouped by their 'attempt ID' "
            "where later grades with the same attempt ID supersede earlier "
            "ones."),
            verbose_name=_("Attempt ID"))

    points = models.DecimalField(max_digits=10, decimal_places=2,
            blank=True, null=True,
            verbose_name=_("Points"))
    max_points = models.DecimalField(max_digits=10, decimal_places=2,
            verbose_name=_("Max points"))

    comment = models.TextField(blank=True, null=True,
            verbose_name=_("Comment"))

    due_time = models.DateTimeField(default=None, blank=True, null=True,
            verbose_name=_("Due time"))

    creator = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
            verbose_name=_("Creator"), on_delete=models.SET_NULL)
    grade_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Grade time"))

    flow_session = models.ForeignKey(FlowSession, null=True, blank=True,
            related_name="grade_changes",
            verbose_name=_("Flow session"), on_delete=models.SET_NULL)

    class Meta:
        verbose_name = _("Grade change")
        verbose_name_plural = _("Grade changes")
        ordering = ("opportunity", "participation", "grade_time")

    def str(self):
        # Translators: information for GradeChange
        return _("%(participation)s %(state)s on %(opportunityname)s") % {
            "participation": self.participation,
            "state": self.state,
            "opportunityname": self.opportunity.name}

    def clean(self) -> None:
        super().clean()

        if self.opportunity.course != self.participation.course:
            raise ValidationError(_("Participation and opportunity must live "
                    "in the same course"))

    def percentage(self) -> Decimal | None:
        if (self.max_points is not None
                and self.points is not None
                and self.max_points != 0):
            return 100*self.points/self.max_points
        else:
            return None

    def get_state_desc(self):
        return dict(GRADE_STATE_CHANGE_CHOICES).get(
                self.state)

    # may be set by GradeStateMachine
    # FIXME: This is kind of a nasty thing to do
    is_superseded: bool

# }}}


# {{{ grade state machine

class GradeStateMachine:
    opportunity: GradingOpportunity | None
    state: str | None
    due_time: datetime.datetime | None
    last_graded_time: datetime.datetime | None
    last_report_time: datetime.datetime | None
    _last_grade_change_time: datetime.datetime | None

    valid_percentages: list[float | Decimal]
    attempt_id_to_gchange: dict[str, GradeChange]

    def __init__(self) -> None:
        self.opportunity = None

        self.state = None
        self._clear_grades()
        self.due_time = None
        self.last_graded_time = None
        self.last_report_time = None

        # applies to *all* grade changes
        self._last_grade_change_time = None

    def _clear_grades(self) -> None:

        self.state = None
        self.last_grade_time = None
        self.valid_percentages = []
        self.attempt_id_to_gchange: dict[str, GradeChange] = {}

    def _consume_grade_change(self,
            gchange: GradeChange, set_is_superseded: bool) -> None:

        if self.opportunity is None:
            opp = self.opportunity = gchange.opportunity
            assert opp is not None
            self.due_time = opp.due_time
        else:
            assert self.opportunity.pk == gchange.opportunity.pk

        assert self.opportunity is not None

        # check that times are increasing
        if self._last_grade_change_time is not None:
            assert gchange.grade_time >= self._last_grade_change_time
        self._last_grade_change_time = gchange.grade_time

        if gchange.state == grade_state_change_types.graded:
            if self.state == grade_state_change_types.unavailable:
                raise ValueError(
                        _("cannot accept grade once opportunity has been "
                            "marked 'unavailable'"))
            if self.state == grade_state_change_types.exempt:
                raise ValueError(
                        _("cannot accept grade once opportunity has been "
                        "marked 'exempt'"))

            # if self.due_time is not None and gchange.grade_time > self.due_time:
                # raise ValueError("cannot accept grade after due date")

            self.state = gchange.state
            if gchange.attempt_id is not None:
                if (set_is_superseded
                        and gchange.attempt_id in self.attempt_id_to_gchange):
                    self.attempt_id_to_gchange[gchange.attempt_id] \
                            .is_superseded = True
                self.attempt_id_to_gchange[gchange.attempt_id] = gchange
            else:
                self.valid_percentages.append(not_none(gchange.percentage()))

            self.last_graded_time = gchange.grade_time

        elif gchange.state == grade_state_change_types.unavailable:
            self._clear_grades()
            self.state = gchange.state

        elif gchange.state == grade_state_change_types.do_over:
            self._clear_grades()

        elif gchange.state == grade_state_change_types.exempt:
            self._clear_grades()
            self.state = gchange.state

        elif gchange.state == grade_state_change_types.report_sent:
            self.last_report_time = gchange.grade_time

        elif gchange.state == grade_state_change_types.extension:
            self.due_time = gchange.due_time

        elif gchange.state in [
                grade_state_change_types.grading_started,
                grade_state_change_types.retrieved,
                ]:
            pass
        else:
            raise RuntimeError(
                    _("invalid grade change state '%s'") % gchange.state)

    def consume(self, iterable: Iterable[GradeChange],
            set_is_superseded: bool = False) -> GradeStateMachine:

        for gchange in iterable:
            gchange.is_superseded = False
            self._consume_grade_change(gchange, set_is_superseded)

        valid_grade_changes = sorted(
                (gchange
                for gchange in self.attempt_id_to_gchange.values()
                if gchange.percentage() is not None),
                key=lambda gchange: gchange.grade_time)

        self.valid_percentages.extend(
                not_none(gchange.percentage())
                for gchange in valid_grade_changes)

        del self.attempt_id_to_gchange

        return self

    def percentage(self) -> float | Decimal | None:
        """
        :return: a percentage of achieved points, or *None*
        """
        if self.opportunity is None or not self.valid_percentages:
            return None

        strategy = self.opportunity.aggregation_strategy

        if strategy == grade_aggregation_strategy.max_grade:
            return max(self.valid_percentages)
        elif strategy == grade_aggregation_strategy.min_grade:
            return min(self.valid_percentages)
        elif strategy == grade_aggregation_strategy.avg_grade:
            return sum(self.valid_percentages)/len(self.valid_percentages)
        elif strategy == grade_aggregation_strategy.use_earliest:
            return self.valid_percentages[0]
        elif strategy == grade_aggregation_strategy.use_latest:
            return self.valid_percentages[-1]
        else:
            raise ValueError(
                    _("invalid grade aggregation strategy '%s'") % strategy)

    def stringify_state(self):
        if self.state is None:
            return "- ∅ -"
        elif self.state == grade_state_change_types.exempt:
            return _("(exempt)")
        elif self.state == grade_state_change_types.graded:
            if self.valid_percentages:
                result = f"{self.percentage():.1f}%"
                if len(self.valid_percentages) > 1:
                    result += " (/%d)" % len(self.valid_percentages)
                return result
            else:
                return "- ∅ -"
        else:
            return _("(other state)")

    def stringify_machine_readable_state(self):
        if self.state is None:
            return "NONE"
        elif self.state == grade_state_change_types.exempt:
            return "EXEMPT"
        elif self.state == grade_state_change_types.graded:
            if self.valid_percentages:
                return f"{self.percentage():.3f}"
            else:
                return "NONE"
        else:
            return "OTHER_STATE"

    def stringify_percentage(self):
        if self.state == grade_state_change_types.graded:
            if self.valid_percentages:
                return f"{self.percentage():.1f}"
            else:
                return ""
        else:
            return ""
# }}}


# {{{ flow <-> grading integration

def get_flow_grading_opportunity(
        course: Course, flow_id: str, flow_desc: FlowDesc,
        grade_identifier: str, grade_aggregation_strategy: str
        ) -> GradingOpportunity:

    default_name = (
            # Translators: display the name of a flow
            _("Flow: %(flow_desc_title)s")
            % {"flow_desc_title": flow_desc.title})

    gopp, created = GradingOpportunity.objects.get_or_create(
            course=course,
            identifier=grade_identifier,
            defaults={
                "name": default_name,
                "flow_id": flow_id,
                "aggregation_strategy": grade_aggregation_strategy,
                })

    # update gopp.name when flow_desc.title changed
    if not created:
        if gopp.name != default_name:
            gopp.name = default_name
            gopp.save()

    return gopp

# }}}


# {{{ XMPP log

class InstantMessage(models.Model):
    participation = models.ForeignKey(Participation,
            verbose_name=_("Participation"), on_delete=models.CASCADE)
    text = models.CharField(max_length=200,
            verbose_name=_("Text"))
    time = models.DateTimeField(default=now,
            verbose_name=_("Time"))

    class Meta:
        verbose_name = _("Instant message")
        verbose_name_plural = _("Instant messages")
        ordering = ("participation__course", "time")

    def __str__(self) -> str:
        return f"{self.participation}: {self.text}"

# }}}


# {{{ exams

class Exam(models.Model):
    course = models.ForeignKey(Course,
            verbose_name=_("Course"), on_delete=models.CASCADE)
    description = models.CharField(max_length=200,
            verbose_name=_("Description"))
    flow_id = models.CharField(max_length=200,
            verbose_name=_("Flow ID"))
    active = models.BooleanField(
            default=True,
            verbose_name=_("Active"),
            help_text=_(
                "Currently active, i.e. may be used to log in "
                "via an exam ticket"))
    listed = models.BooleanField(
            verbose_name=_("Listed"),
            default=True,
            help_text=_("Shown in the list of current exams"))

    no_exams_before = models.DateTimeField(
            verbose_name=_("No exams before"))
    no_exams_after = models.DateTimeField(
            null=True, blank=True,
            verbose_name=_("No exams after"))

    class Meta:
        verbose_name = _("Exam")
        verbose_name_plural = _("Exams")
        ordering = ("course", "no_exams_before",)

    def __str__(self) -> str:
        return _("Exam  %(description)s in %(course)s") % {
                "description": self.description,
                "course": self.course,
                }


class ExamTicket(models.Model):
    exam = models.ForeignKey(Exam,
            verbose_name=_("Exam"), on_delete=models.CASCADE)

    participation = models.ForeignKey(Participation, db_index=True,
            verbose_name=_("Participation"), on_delete=models.CASCADE)

    creator = models.ForeignKey(settings.AUTH_USER_MODEL, null=True,
            verbose_name=_("Creator"), on_delete=models.SET_NULL)
    creation_time = models.DateTimeField(default=now,
            verbose_name=_("Creation time"))
    usage_time = models.DateTimeField(
            verbose_name=_("Usage time"),
            help_text=_("Date and time of first usage of ticket"),
            null=True, blank=True)

    state = models.CharField(max_length=50,
            choices=EXAM_TICKET_STATE_CHOICES,
            verbose_name=_("Exam ticket state"))

    code = models.CharField(max_length=50)

    valid_start_time = models.DateTimeField(
            verbose_name=_("End valid period"),
            help_text=_("If not blank, date and time at which this exam ticket "
                "starts being valid/usable"),
            null=True, blank=True)
    valid_end_time = models.DateTimeField(
            verbose_name=_("End valid period"),
            help_text=_("If not blank, date and time at which this exam ticket "
                "stops being valid/usable"),
            null=True, blank=True)
    restrict_to_facility = models.CharField(max_length=200, blank=True, null=True,
            verbose_name=_("Restrict to facility"),
            help_text=_("If not blank, this exam ticket may only be used in the "
                "given facility"))

    require_login = models.BooleanField(