Skip to content
models.py 73.3 KiB
Newer Older
            verbose_name=_("Group ID"))
ifaint's avatar
ifaint committed
    page_id = models.CharField(max_length=200,
            verbose_name=_("Page ID"))
ifaint's avatar
ifaint committed
    data = JSONField(null=True, blank=True,
            # Show correct characters in admin for non ascii languages.
            dump_kwargs={"ensure_ascii": False},
            verbose_name=_("Data"))
    title = models.CharField(max_length=1000,
            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:
ifaint's avatar
ifaint committed
        verbose_name = _("Flow page data")
        verbose_name_plural = _("Flow page data")

    def __unicode__(self):
        # 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})
    __str__ = __unicode__
    # Django's templates are a little daft. No arithmetic--really?
        return self.page_ordinal - 1
        return self.page_ordinal + 1
    def human_readable_ordinal(self):
        return self.page_ordinal + 1

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.
ifaint's avatar
ifaint committed
    flow_session = models.ForeignKey(FlowSession, db_index=True,
            verbose_name=_("Flow session"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    page_data = models.ForeignKey(FlowPageData, db_index=True,
            verbose_name=_("Page data"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    visit_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Visit time"))
ifaint's avatar
ifaint committed
    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)
ifaint's avatar
ifaint committed
    is_synthetic = models.BooleanField(default=False,
ifaint's avatar
ifaint committed
            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"))
ifaint's avatar
ifaint committed
    answer = JSONField(null=True, blank=True,
            # Show correct characters in admin for non ascii languages.
            dump_kwargs={"ensure_ascii": False},
ifaint's avatar
ifaint committed
            # 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.
ifaint's avatar
ifaint committed
    is_submitted_answer = models.NullBooleanField(
            # Translators: determine whether the answer is a final,
            # submitted answer fit for grading.
            verbose_name=_("Is submitted answer"))
    def __unicode__(self):
Dong Zhuang's avatar
Dong Zhuang committed
                # 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:
Andreas Klöckner's avatar
Andreas Klöckner committed
            # Translators: flow page visit: if an answer is
Dong Zhuang's avatar
Dong Zhuang committed
            # provided by user then append the string.
            result += str(_(" (with answer)"))
    __str__ = __unicode__
    class Meta:
ifaint's avatar
ifaint committed
        verbose_name = _("Flow page visit")
ifaint's avatar
ifaint committed
        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):
        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):
ifaint's avatar
ifaint committed
    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)
ifaint's avatar
ifaint committed
    grade_time = models.DateTimeField(db_index=True, default=now,
            verbose_name=_("Grade time"))
    graded_at_git_commit_sha = models.CharField(
ifaint's avatar
ifaint committed
            max_length=200, null=True, blank=True,
            verbose_name=_("Graded at git commit SHA"))
ifaint's avatar
ifaint committed
    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,
ifaint's avatar
ifaint committed
            help_text=_("Point value of this question when receiving "
ifaint's avatar
ifaint committed
            "full credit."),
            verbose_name=_("Max points"))
    correctness = models.FloatField(null=True, blank=True,
ifaint's avatar
ifaint committed
            help_text=_("Real number between zero and one (inclusively) "
ifaint's avatar
ifaint committed
            "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.

ifaint's avatar
ifaint committed
    feedback = JSONField(null=True, blank=True,
            # Show correct characters in admin for non ascii languages.
            dump_kwargs={"ensure_ascii": False},
ifaint's avatar
ifaint committed
            # Translators: "Feedback" stands for the feedback of answers.
            verbose_name=_("Feedback"))
    def percentage(self):
        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

ifaint's avatar
ifaint committed
        verbose_name = _("Flow page visit grade")
ifaint's avatar
ifaint committed
        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 __unicode__(self):
        # 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()}
    __str__ = __unicode__

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)
ifaint's avatar
ifaint committed
    page_data = models.OneToOneField(FlowPageData,
            verbose_name=_("Page data"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    grade = models.ForeignKey(FlowPageVisitGrade,
            verbose_name=_("Grade"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    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"))
def update_bulk_feedback(page_data, grade, bulk_feedback_json):
    # type: (FlowPageData, FlowPageVisitGrade, Any) -> None
    FlowPageBulkFeedback.objects.update_or_create(
            page_data=page_data,
            defaults=dict(
                grade=grade,
                bulk_feedback=bulk_feedback_json))


def get_feedback_for_grade(grade):
Dong Zhuang's avatar
Dong Zhuang committed
    # type: (Optional[FlowPageVisitGrade]) -> Optional[AnswerFeedback]

    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

    from course.page.base import AnswerFeedback  # noqa: F811
Dong Zhuang's avatar
Dong Zhuang committed
    return AnswerFeedback.from_json(grade.feedback, bulk_feedback_json)
Dong Zhuang's avatar
Dong Zhuang committed
# {{{ deprecated exception stuff
Dong Zhuang's avatar
Dong Zhuang committed
def validate_stipulations(stip):  # pragma: no cover (deprecated and not tested)
    if not isinstance(stip, dict):
        raise ValidationError(_("stipulations must be a dictionary"))
    allowed_keys = set(["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"))
Dong Zhuang's avatar
Dong Zhuang committed
class FlowAccessException(models.Model):  # pragma: no cover (deprecated and not tested)  # noqa
ifaint's avatar
ifaint committed
    participation = models.ForeignKey(Participation, db_index=True,
            verbose_name=_("Participation"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    flow_id = models.CharField(max_length=200, blank=False, null=False,
            verbose_name=_("Flow ID"))
ifaint's avatar
ifaint committed
    expiration = models.DateTimeField(blank=True, null=True,
            verbose_name=_("Expiration"))
    stipulations = JSONField(blank=True, null=True,
            # Translators: help text for stipulations in FlowAccessException
            # (deprecated)
ifaint's avatar
ifaint committed
            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 "
ifaint's avatar
ifaint committed
            "to the stipulations in the course content."),
ifaint's avatar
ifaint committed
            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)
ifaint's avatar
ifaint committed
    creation_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Creation time"))
    is_sticky = models.BooleanField(
            default=False,
ifaint's avatar
ifaint committed
            # Translators: deprecated
ifaint's avatar
ifaint committed
            help_text=_("Check if a flow started under this "
            "exception rule set should stay "
ifaint's avatar
ifaint committed
            "under this rule set until it is expired."),
            # Translators: deprecated
            verbose_name=_("Is sticky"))
ifaint's avatar
ifaint committed
    comment = models.TextField(blank=True, null=True,
            verbose_name=_("Comment"))
    def __unicode__(self):
Dong Zhuang's avatar
Dong Zhuang committed
                # 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
                    })
    __str__ = __unicode__
Dong Zhuang's avatar
Dong Zhuang committed
class FlowAccessExceptionEntry(models.Model):  # pragma: no cover (deprecated and not tested)  # noqa
    exception = models.ForeignKey(FlowAccessException,
ifaint's avatar
ifaint committed
            related_name="entries",
            verbose_name=_("Exception"), on_delete=models.CASCADE)
    permission = models.CharField(max_length=50,
ifaint's avatar
ifaint committed
            choices=FLOW_PERMISSION_CHOICES,
            verbose_name=_("Permission"))
        # Translators: FlowAccessExceptionEntry (deprecated)
ifaint's avatar
ifaint committed
        verbose_name_plural = _("Flow access exception entries")
    def __unicode__(self):
        return self.permission

    __str__ = __unicode__
# }}}


class FlowRuleException(models.Model):
ifaint's avatar
ifaint committed
    flow_id = models.CharField(max_length=200, blank=False, null=False,
            verbose_name=_("Flow ID"))
ifaint's avatar
ifaint committed
    participation = models.ForeignKey(Participation, db_index=True,
            verbose_name=_("Participation"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    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)
ifaint's avatar
ifaint committed
    creation_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Creation time"))
ifaint's avatar
ifaint committed
    comment = models.TextField(blank=True, null=True,
            verbose_name=_("Comment"))

    kind = models.CharField(max_length=50, blank=False, null=False,
ifaint's avatar
ifaint committed
            choices=FLOW_RULE_KIND_CHOICES,
            verbose_name=_("Kind"))
ifaint's avatar
ifaint committed
    rule = YAMLField(blank=False, null=False,
            verbose_name=_("Rule"))
ifaint's avatar
ifaint committed
    active = models.BooleanField(default=True,
            verbose_name=pgettext_lazy(
                "Is the flow rule exception activated?", "Active"))
                # Translators: For FlowRuleException
Dong Zhuang's avatar
Dong Zhuang committed
                _("%(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,
Dong Zhuang's avatar
Dong Zhuang committed
                    "course": self.participation.course,
                    "exception_id":
                        " id %d" % self.id if self.id is not None else ""})
    __str__ = __unicode__
        # type: () -> None
        super(FlowRuleException, self).clean()

Dong Zhuang's avatar
Dong Zhuang committed
        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):
ifaint's avatar
ifaint committed
            raise ValidationError(_("grading rules may not expire"))

        from course.validation import (
                ValidationError as ContentValidationError,
                validate_session_access_rule,
                validate_session_grading_rule,
                ValidationContext)
        from course.content import (get_course_repo,
                get_course_commit_sha,
                get_flow_desc)

Andreas Klöckner's avatar
Andreas Klöckner committed
        from relate.utils import dict_to_struct
        rule = dict_to_struct(self.rule)

Dong Zhuang's avatar
Dong Zhuang committed
        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 = []  # type: List
        grade_identifier = None
        if hasattr(flow_desc, "rules"):
            tags = cast(list, getattr(flow_desc.rules, "tags", []))
            grade_identifier = flow_desc.rules.grade_identifier
            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)
Dong Zhuang's avatar
Dong Zhuang committed
            else:  # pragma: no cover. This won't happen
                raise ValueError("invalid exception rule kind")

        except ContentValidationError as e:
            # the rule refers to FlowRuleException rule
Dong Zhuang's avatar
Dong Zhuang committed
            raise ValidationError(
                string_concat(_("invalid existing_session_rules"), ": ", str(e)))
ifaint's avatar
ifaint committed
        verbose_name = _("Flow rule exception")
ifaint's avatar
ifaint committed
        verbose_name_plural = _("Flow rule exceptions")
# }}}


# {{{ grading

class GradingOpportunity(models.Model):
ifaint's avatar
ifaint committed
    course = models.ForeignKey(Course,
            verbose_name=_("Course"), on_delete=models.CASCADE)

    identifier = models.CharField(max_length=200, blank=False, null=False,
ifaint's avatar
ifaint committed
            # Translators: format of identifier for GradingOpportunity
ifaint's avatar
ifaint committed
            help_text=_("A symbolic name for this grade. "
ifaint's avatar
ifaint committed
            "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 hypens ('-').")),
                    ])
    name = models.CharField(max_length=200, blank=False, null=False,
ifaint's avatar
ifaint committed
            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,
ifaint's avatar
ifaint committed
            help_text=_("Flow identifier that this grading opportunity "
ifaint's avatar
ifaint committed
            "is linked to, if any"),
            verbose_name=_("Flow ID"))
    aggregation_strategy = models.CharField(max_length=20,
ifaint's avatar
ifaint committed
            choices=GRADE_AGGREGATION_STRATEGY_CHOICES,
            # Translators: strategy on how the grading of mutiple sessioins
            verbose_name=_("Aggregation strategy"))
ifaint's avatar
ifaint committed
    due_time = models.DateTimeField(default=None, blank=True, null=True,
            verbose_name=_("Due time"))
ifaint's avatar
ifaint committed
    creation_time = models.DateTimeField(default=now,
            verbose_name=_("Creation time"))
ifaint's avatar
ifaint committed
    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."))
ifaint's avatar
ifaint committed
        verbose_name = _("Grading opportunity")
        verbose_name_plural = _("Grading opportunities")
        ordering = ("course", "due_time", "identifier")
        unique_together = (("course", "identifier"),)
Dong Zhuang's avatar
Dong Zhuang committed
                # Translators: For GradingOpportunity
                _("%(opportunity_name)s (%(opportunity_id)s) in %(course)s")
                % {
                    "opportunity_name": self.name,
                    "opportunity_id": self.identifier,
                    "course": self.course})
    __str__ = __unicode__
    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.
    """
ifaint's avatar
ifaint committed
    opportunity = models.ForeignKey(GradingOpportunity,
            verbose_name=_("Grading opportunity"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    participation = models.ForeignKey(Participation,
            verbose_name=_("Participation"), on_delete=models.CASCADE)

    state = models.CharField(max_length=50,
ifaint's avatar
ifaint committed
            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",
ifaint's avatar
ifaint committed
            # Translators: help text of "attempt_id" in GradeChange class
ifaint's avatar
ifaint committed
            help_text=_("Grade changes are grouped by their 'attempt ID' "
            "where later grades with the same attempt ID supersede earlier "
ifaint's avatar
ifaint committed
            "ones."),
            verbose_name=_("Attempt ID"))
    points = models.DecimalField(max_digits=10, decimal_places=2,
ifaint's avatar
ifaint committed
            blank=True, null=True,
            verbose_name=_("Points"))
ifaint's avatar
ifaint committed
    max_points = models.DecimalField(max_digits=10, decimal_places=2,
            verbose_name=_("Max points"))
ifaint's avatar
ifaint committed
    comment = models.TextField(blank=True, null=True,
            verbose_name=_("Comment"))
ifaint's avatar
ifaint committed
    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)
ifaint's avatar
ifaint committed
    grade_time = models.DateTimeField(default=now, db_index=True,
            verbose_name=_("Grade time"))
    flow_session = models.ForeignKey(FlowSession, null=True, blank=True,
ifaint's avatar
ifaint committed
            related_name="grade_changes",
            verbose_name=_("Flow session"), on_delete=models.SET_NULL)
ifaint's avatar
ifaint committed
        verbose_name = _("Grade change")
        verbose_name_plural = _("Grade changes")
        ordering = ("opportunity", "participation", "grade_time")

    def __unicode__(self):
        # Translators: information for GradeChange
        return _("%(participation)s %(state)s on %(opportunityname)s") % {
            "participation": self.participation,
            "state": self.state,
            "opportunityname": self.opportunity.name}
    __str__ = __unicode__
        if self.opportunity.course != self.participation.course:
ifaint's avatar
ifaint committed
            raise ValidationError(_("Participation and opportunity must live "
                    "in the same course"))
    def percentage(self):
        # type: () -> Optional[float]

        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)

# }}}


# {{{ grade state machine

class GradeStateMachine(object):
    def __init__(self):
        # type: () -> 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):
        self.state = None
        self.last_grade_time = None
        self.valid_percentages = []  # type: List[GradeChange]
        self.attempt_id_to_gchange = {}  # type: Dict[Text, GradeChange]
    def _consume_grade_change(self, gchange, set_is_superseded):
        # type: (GradeChange, 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:
Dong Zhuang's avatar
Dong Zhuang committed
            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 "
ifaint's avatar
ifaint committed
                        "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:
Andreas Klöckner's avatar
Andreas Klöckner committed
                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(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, set_is_superseded=False):
        # type: (Iterable[GradeChange], bool) -> 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(
                cast(GradeChange, gchange.percentage())
        del self.attempt_id_to_gchange
        # type: () -> Optional[float]

        """
        :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 u"- ∅ -"
        elif self.state == grade_state_change_types.exempt:
Dong Zhuang's avatar
Dong Zhuang committed
            return _("(exempt)")
        elif self.state == grade_state_change_types.graded:
            if self.valid_percentages:
                result = "%.1f%%" % self.percentage()
                if len(self.valid_percentages) > 1:
                    result += " (/%d)" % len(self.valid_percentages)
                return result
            else:
                return u"- ∅ -"
Dong Zhuang's avatar
Dong Zhuang committed
            return _("(other state)")
    def stringify_machine_readable_state(self):
        if self.state is None:
            return u"NONE"
        elif self.state == grade_state_change_types.exempt:
            return "EXEMPT"
        elif self.state == grade_state_change_types.graded:
            if self.valid_percentages:
                return "%.3f" % self.percentage()
            else:
                return u"NONE"
        else:
            return u"OTHER_STATE"

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


# {{{ flow <-> grading integration

Andreas Klöckner's avatar
Andreas Klöckner committed
def get_flow_grading_opportunity(course, flow_id, flow_desc,
        grade_identifier, grade_aggregation_strategy):
    # type: (Course, Text, FlowDesc, Text, Text) -> 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,
Andreas Klöckner's avatar
Andreas Klöckner committed
            identifier=grade_identifier,
                flow_id=flow_id,
Andreas Klöckner's avatar
Andreas Klöckner committed
                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()


# {{{ XMPP log

class InstantMessage(models.Model):
ifaint's avatar
ifaint committed
    participation = models.ForeignKey(Participation,
            verbose_name=_("Participation"), on_delete=models.CASCADE)
ifaint's avatar
ifaint committed
    text = models.CharField(max_length=200,
            verbose_name=_("Text"))
ifaint's avatar
ifaint committed
    time = models.DateTimeField(default=now,
            verbose_name=_("Time"))
ifaint's avatar
ifaint committed
        verbose_name = _("Instant message")
        verbose_name_plural = _("Instant messages")
        ordering = ("participation__course", "time")

    def __unicode__(self):
        return "%s: %s" % (self.participation, self.text)

    __str__ = __unicode__

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 __unicode__(self):
        return _("Exam  %(description)s in %(course)s") % {
                "description": self.description,
                "course": self.course,
    __str__ = __unicode__

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(
            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, db_index=True, unique=True)

    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"))

    class Meta:
        verbose_name = _("Exam ticket")
        verbose_name_plural = _("Exam tickets")
        ordering = ("exam__course", "-creation_time")
        permissions = (
                ("can_issue_exam_tickets", _("Can issue exam tickets to student")),
                )
    def __unicode__(self):
        return _("Exam  ticket for %(participation)s in %(exam)s") % {
                "participation": self.participation,
                "exam": self.exam,
    __str__ = __unicode__
    def clean(self):
        super(ExamTicket, self).clean()

        try:
            if self.exam.course != self.participation.course:
                raise ValidationError(_("Participation and exam must live "
                        "in the same course"))
        except ObjectDoesNotExist:
            pass

# vim: foldmethod=marker