grades.py

from __future__ import annotations


__copyright__ = "Copyright (C) 2014 Andreas Kloeckner"

__license__ = """
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""

import re
from decimal import Decimal
from typing import (
    TYPE_CHECKING,
    Any,
    cast,
)

from crispy_forms.layout import Submit
from django import forms, http
from django.contrib import messages
from django.core.exceptions import (
    ObjectDoesNotExist,
    PermissionDenied,
    SuspiciousOperation,
)
from django.db import connection, transaction
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from django.utils.timezone import now
from django.utils.translation import gettext, gettext_lazy as _, pgettext_lazy

from course.constants import participation_permission as pperm
from course.flow import adjust_flow_session_page_data
from course.models import (
    FlowPageVisit,
    FlowSession,
    GradeChange,
    GradeStateMachine,
    GradingOpportunity,
    Participation,
    grade_state_change_types,
    participation_status,
)
from course.utils import course_view, render_course_page
from course.views import get_now_or_fake_time
from relate.utils import (
    HTML5DateTimeInput,
    StyledForm,
    StyledModelForm,
    not_none,
    string_concat,
)


# {{{ for mypy

if TYPE_CHECKING:
    from course.content import FlowDesc
    from course.models import Course, FlowPageVisitGrade
    from course.utils import CoursePageContext

# }}}


# {{{ student grade book

@course_view
def view_participant_grades(pctx, participation_id=None):
    if pctx.participation is None:
        raise PermissionDenied(_("must be enrolled to view grades"))

    if participation_id is not None:
        grade_participation = Participation.objects.get(id=int(participation_id))
    else:
        grade_participation = pctx.participation

    is_privileged_view = pctx.has_permission(pperm.view_gradebook)

    if grade_participation != pctx.participation:
        if not is_privileged_view:
            raise PermissionDenied(_("may not view other people's grades"))

    # NOTE: It's important that these two queries are sorted consistently,
    # also consistently with the code below.

    gopp_extra_filter_kwargs = {}
    if not is_privileged_view:
        gopp_extra_filter_kwargs = {"shown_in_participant_grade_book": True}

    grading_opps = list(GradingOpportunity.objects
            .filter(
                course=pctx.course,
                shown_in_grade_book=True,
                **gopp_extra_filter_kwargs
                )
            .order_by("identifier"))

    grade_changes = list(GradeChange.objects
            .filter(
                participation=grade_participation,
                opportunity__pk__in=[gopp.pk for gopp in grading_opps],
                opportunity__shown_in_grade_book=True)
            .order_by(
                "participation__id",
                "opportunity__identifier",
                "grade_time")
            .select_related("participation")
            .select_related("participation__user")
            .select_related("opportunity"))

    idx = 0

    grade_table = []
    for opp in grading_opps:
        if not is_privileged_view:
            if not (opp.shown_in_grade_book
                    and opp.shown_in_participant_grade_book):
                continue
        else:
            if not opp.shown_in_grade_book:
                continue

        while (
                idx < len(grade_changes)
                and grade_changes[idx].opportunity.identifier < opp.identifier
                ):
            idx += 1

        my_grade_changes = []
        while (
                idx < len(grade_changes)
                and grade_changes[idx].opportunity.pk == opp.pk):
            my_grade_changes.append(grade_changes[idx])
            idx += 1

        state_machine = GradeStateMachine()
        state_machine.consume(my_grade_changes)

        grade_table.append(
                GradeInfo(
                    opportunity=opp,
                    grade_state_machine=state_machine))

    return render_course_page(pctx, "course/gradebook-participant.html", {
        "grade_table": grade_table,
        "grade_participation": grade_participation,
        "grading_opportunities": grading_opps,
        "grade_state_change_types": grade_state_change_types,
        "is_privileged_view": is_privileged_view,
        })

# }}}


# {{{ participant list

@course_view
def view_participant_list(pctx):
    if not pctx.has_permission(pperm.view_gradebook):
        raise PermissionDenied(_("may not view grade book"))

    participations = list(Participation.objects
            .filter(
                course=pctx.course)
            .order_by("id")
            .select_related("user"))

    return render_course_page(pctx, "course/gradebook-participant-list.html", {
        "participations": participations,
        })

# }}}


# {{{ grading opportunity list

@course_view
def view_grading_opportunity_list(pctx):
    if not pctx.has_permission(pperm.view_gradebook):
        raise PermissionDenied(_("may not view grade book"))

    grading_opps = list(GradingOpportunity.objects
            .filter(
                course=pctx.course,
                )
            .order_by("identifier"))

    return render_course_page(pctx, "course/gradebook-opp-list.html", {
        "grading_opps": grading_opps,
        })

# }}}


# {{{ teacher grade book

class GradeInfo:
    def __init__(self, opportunity: GradingOpportunity,
            grade_state_machine: GradeStateMachine) -> None:
        self.opportunity = opportunity
        self.grade_state_machine = grade_state_machine


def get_grade_table(course: Course) -> tuple[
        list[Participation], list[GradingOpportunity], list[list[GradeInfo]]]:

    # NOTE: It's important that these queries are sorted consistently,
    # also consistently with the code below.
    grading_opps = list(GradingOpportunity.objects
            .filter(
                course=course,
                shown_in_grade_book=True,
                )
            .order_by("identifier"))

    participations = list(Participation.objects
            .filter(
                course=course,
                status=participation_status.active)
            .order_by("id")
            .select_related("user"))

    grade_changes = list(GradeChange.objects
            .filter(
                opportunity__course=course,
                opportunity__shown_in_grade_book=True)
            .order_by(
                "participation__id",
                "opportunity__identifier",
                "grade_time")
            .select_related("participation")
            .select_related("participation__user")
            .select_related("opportunity"))

    idx = 0

    grade_table = []
    for participation in participations:
        while (
                idx < len(grade_changes)
                and grade_changes[idx].participation.id < participation.id):
            idx += 1

        grade_row = []
        for opp in grading_opps:
            while (
                    idx < len(grade_changes)
                    and grade_changes[idx].participation.pk == participation.pk
                    and grade_changes[idx].opportunity.identifier < opp.identifier
                    ):
                idx += 1

            my_grade_changes = []
            while (
                    idx < len(grade_changes)
                    and grade_changes[idx].opportunity.pk == opp.pk
                    and grade_changes[idx].participation.pk == participation.pk):
                my_grade_changes.append(grade_changes[idx])
                idx += 1

            state_machine = GradeStateMachine()
            state_machine.consume(my_grade_changes)

            grade_row.append(
                    GradeInfo(
                        opportunity=opp,
                        grade_state_machine=state_machine))

        grade_table.append(grade_row)

    return participations, grading_opps, grade_table


@course_view
def view_gradebook(pctx):
    if not pctx.has_permission(pperm.view_gradebook):
        raise PermissionDenied(_("may not view grade book"))

    participations, grading_opps, grade_table = get_grade_table(pctx.course)

    def grade_key(entry):
        (participation, _grades) = entry
        return (participation.user.last_name.lower(),
                    participation.user.first_name.lower())

    grade_table = sorted(zip(participations, grade_table, strict=True), key=grade_key)

    return render_course_page(pctx, "course/gradebook.html", {
        "grade_table": grade_table,
        "grading_opportunities": grading_opps,
        "participations": participations,
        "grade_state_change_types": grade_state_change_types,
        })


@course_view
def export_gradebook_csv(pctx):
    if not pctx.has_permission(pperm.batch_export_grade):
        raise PermissionDenied(_("may not batch-export grades"))

    participations, grading_opps, grade_table = get_grade_table(pctx.course)

    from io import StringIO
    csvfile = StringIO()

    import csv

    fieldnames = ["user_name", "last_name", "first_name"] + [
            gopp.identifier for gopp in grading_opps]

    writer = csv.writer(csvfile)

    writer.writerow(fieldnames)

    for participation, grades in zip(participations, grade_table, strict=True):
        writer.writerow([
            participation.user.username,
            participation.user.last_name,
            participation.user.first_name,
            ] + [grade_info.grade_state_machine.stringify_machine_readable_state()
                for grade_info in grades])

    response = http.HttpResponse(
            csvfile.getvalue().encode("utf-8"),
            content_type="text/plain; charset=utf-8")
    response["Content-Disposition"] = (
            f'attachment; filename="grades-{pctx.course.identifier}.csv"')
    return response

# }}}


# {{{ grades by grading opportunity

class OpportunitySessionGradeInfo:
    def __init__(self,
                 grade_state_machine,  # Optional[GradeStateMachine]
                 flow_session,  # Optional[FlowSession]
                 flow_id=None,  # Optional[Text]
                 grades=None,  # Optional[Any]
                 has_finished_session=False,  # bool
                 ) -> None:
        """
        :param grade_state_machine: a :class:`GradeStateMachine:` or None.
        :param flow_session: a :class:`FlowSession:` or None.
        :param flow_id: optional, a :class:`str:` or None. This is used to determine
        whether the instance is created by a flow-session-related opportunity.
        :param grades: optional, a :class:`list:` of float or None, representing the
        percentage grades of each page in the flow session.
        :param has_finished_session:  a :class:`bool:`, represent whether
        the related participation has finished a flow-session, if the opportunity
        is a flow-session related one. This is used to correctly order flow state,
        if the participation has at least one finished flow session, the in-progress
        flow session won't be ordered to top of the session state column.
        """
        self.grade_state_machine = grade_state_machine
        self.flow_session = flow_session
        self.flow_id = flow_id
        self.grades = grades
        self.has_finished_session = has_finished_session


class ModifySessionsForm(StyledForm):
    def __init__(self, session_rule_tags: list[str],
            *args: Any, **kwargs: Any) -> None:

        super().__init__(*args, **kwargs)

        self.fields["rule_tag"] = forms.ChoiceField(
                choices=tuple(
                    (rule_tag, str(rule_tag))
                    for rule_tag in session_rule_tags),
                label=_("Session tag"))
        self.fields["past_due_only"] = forms.BooleanField(
                required=False,
                initial=True,
                help_text=_("Only act on in-progress sessions that are past "
                "their access rule's due date (applies to 'expire')"),
                # Translators: see help text above.
                label=_("Past due only"))

        self.helper.add_input(
                Submit("expire", _("Impose deadline (Expire sessions)")))
        # self.helper.add_input(
        # Submit("end", _("End and grade all sessions")))
        self.helper.add_input(
                Submit("regrade", _("Regrade ended sessions")))
        self.helper.add_input(
                Submit("recalculate", _("Recalculate grades of ended sessions")))


RULE_TAG_NONE_STRING = "<<<NONE>>>"


def mangle_session_access_rule_tag(rule_tag: str | None) -> str:
    if rule_tag is None:
        return RULE_TAG_NONE_STRING
    else:
        return rule_tag


@course_view
def view_grades_by_opportunity(
        pctx: CoursePageContext, opp_id: str) -> http.HttpResponse:

    from course.views import get_now_or_fake_time
    now_datetime = get_now_or_fake_time(pctx.request)

    if not pctx.has_permission(pperm.view_gradebook):
        raise PermissionDenied(_("may not view grade book"))

    opportunity = get_object_or_404(GradingOpportunity, id=int(opp_id))

    if pctx.course != opportunity.course:
        raise SuspiciousOperation(_("opportunity from wrong course"))

    # {{{ batch sessions form

    batch_ops_allowed = (
            pctx.has_permission(pperm.batch_impose_flow_session_deadline)
            or pctx.has_permission(pperm.batch_end_flow_session)
            or pctx.has_permission(pperm.batch_regrade_flow_session)
            or pctx.has_permission(pperm.batch_recalculate_flow_session_grade)
            )

    batch_session_ops_form: ModifySessionsForm | None = None
    if batch_ops_allowed and opportunity.flow_id:
        cursor = connection.cursor()
        cursor.execute("select distinct access_rules_tag from course_flowsession "
                "where course_id = %s and flow_id = %s "
                "order by access_rules_tag", (pctx.course.id, opportunity.flow_id))
        session_rule_tags = [
                mangle_session_access_rule_tag(row[0]) for row in cursor.fetchall()]

        request = pctx.request
        if request.method == "POST":
            bsf = batch_session_ops_form = ModifySessionsForm(
                    session_rule_tags, request.POST, request.FILES)

            assert batch_session_ops_form is not None

            if "expire" in request.POST:
                op = "expire"
                if not pctx.has_permission(pperm.batch_impose_flow_session_deadline):
                    raise PermissionDenied(_("may not impose deadline"))

            elif "end" in request.POST:
                op = "end"
                if not pctx.has_permission(pperm.batch_end_flow_session):
                    raise PermissionDenied(_("may not batch-end flows"))

            elif "regrade" in request.POST:
                op = "regrade"
                if not pctx.has_permission(pperm.batch_regrade_flow_session):
                    raise PermissionDenied(_("may not batch-regrade flows"))

            elif "recalculate" in request.POST:
                op = "recalculate"
                if not pctx.has_permission(
                        pperm.batch_recalculate_flow_session_grade):
                    raise PermissionDenied(_("may not batch-recalculate grades"))

            else:
                raise SuspiciousOperation(_("invalid operation"))

            if bsf.is_valid():
                rule_tag = bsf.cleaned_data["rule_tag"]
                past_due_only = bsf.cleaned_data["past_due_only"]

                if rule_tag == RULE_TAG_NONE_STRING:
                    rule_tag = None

                from course.tasks import (
                    expire_in_progress_sessions,
                    finish_in_progress_sessions,
                    recalculate_ended_sessions,
                    regrade_flow_sessions,
                )

                if op == "expire":
                    async_res = expire_in_progress_sessions.delay(
                            pctx.course.id, opportunity.flow_id,
                            rule_tag, now_datetime,
                            past_due_only=past_due_only)

                    return redirect("relate-monitor_task", async_res.id)

                elif op == "end":
                    async_res = finish_in_progress_sessions.delay(
                            pctx.course.id, opportunity.flow_id,
                            rule_tag, now_datetime,
                            past_due_only=past_due_only)

                    return redirect("relate-monitor_task", async_res.id)

                elif op == "regrade":
                    async_res = regrade_flow_sessions.delay(
                            pctx.course.id, opportunity.flow_id,
                            rule_tag, inprog_value=False)

                    return redirect("relate-monitor_task", async_res.id)

                else:
                    assert op == "recalculate"
                    async_res = recalculate_ended_sessions.delay(
                            pctx.course.id, opportunity.flow_id,
                            rule_tag)

                    return redirect("relate-monitor_task", async_res.id)

        else:
            batch_session_ops_form = ModifySessionsForm(session_rule_tags)

    # }}}

    # NOTE: It's important that these queries are sorted consistently,
    # also consistently with the code below.

    participations = list(Participation.objects
            .filter(
                course=pctx.course,
                status=participation_status.active)
            .order_by("id")
            .select_related("user"))

    grade_changes = list(GradeChange.objects
            .filter(opportunity=opportunity)
            .order_by(
                "participation__id",
                "grade_time")
            .select_related("participation")
            .select_related("participation__user")
            .select_related("opportunity"))

    if opportunity.flow_id:
        flow_sessions: list[FlowSession] | None = list(FlowSession.objects
                .filter(
                    flow_id=opportunity.flow_id,
                    )
                .order_by(
                    "participation__id",
                    "start_time"
                    ))
    else:
        flow_sessions = None

    view_page_grades = pctx.request.GET.get("view_page_grades") == "1"

    gchng_idx = 0
    fsess_idx = 0

    finished_sessions = 0
    total_sessions = 0

    grade_table: list[tuple[Participation, OpportunitySessionGradeInfo]] = []
    for participation in participations:
        # Advance in grade change list
        while (
                gchng_idx < len(grade_changes)
                and grade_changes[gchng_idx].participation.pk < participation.pk):
            gchng_idx += 1

        my_grade_changes = []
        while (
                gchng_idx < len(grade_changes)
                and grade_changes[gchng_idx].participation.pk == participation.pk):
            my_grade_changes.append(grade_changes[gchng_idx])
            gchng_idx += 1

        state_machine = GradeStateMachine()
        state_machine.consume(my_grade_changes)

        # Advance in flow session list
        if flow_sessions is None:
            grade_table.append(
                    (participation, OpportunitySessionGradeInfo(
                        grade_state_machine=state_machine,
                        flow_session=None)))
        else:
            while (
                    fsess_idx < len(flow_sessions)
                    and (
                        flow_sessions[fsess_idx].participation is None
                        or (
                            not_none(flow_sessions[fsess_idx].participation).pk
                            < participation.pk))):
                fsess_idx += 1

            my_flow_sessions = []

            while (
                    fsess_idx < len(flow_sessions)
                    and flow_sessions[fsess_idx].participation is not None
                    and (
                        not_none(flow_sessions[fsess_idx].participation).pk
                        == participation.pk)):
                my_flow_sessions.append(flow_sessions[fsess_idx])
                fsess_idx += 1

            # When view_page_grades, participations with no started flow sessions
            # should not be included
            if not my_flow_sessions and not view_page_grades:
                grade_table.append(
                        (participation, OpportunitySessionGradeInfo(
                            grade_state_machine=None,
                            flow_id=opportunity.flow_id,
                            flow_session=None)))

            for fsession in my_flow_sessions:
                total_sessions += 1

                assert fsession is not None

                if not fsession.in_progress:
                    finished_sessions += 1

                grade_table.append(
                        (participation, OpportunitySessionGradeInfo(
                            grade_state_machine=state_machine,
                            flow_id=opportunity.flow_id,
                            flow_session=fsession,
                            has_finished_session=bool(finished_sessions))))

    if view_page_grades and len(grade_table) > 0 and opportunity.flow_id is not None:
        # Query grades for flow pages
        all_flow_sessions = [
                cast(FlowSession, info.flow_session)
                for _dummy1, info in grade_table]

        assert all(all_flow_sessions)
        max_page_count = max(not_none(fsess.page_count) for fsess in all_flow_sessions)
        page_numbers = list(range(1, 1 + max_page_count))

        from course.flow import assemble_page_grades
        page_grades: list[list[FlowPageVisitGrade | None]] \
            = assemble_page_grades(all_flow_sessions)

        for (_dummy2, grade_info), grade_list in \
                zip(grade_table, page_grades, strict=True):  # type: ignore
            # Not all pages exist in all sessions
            grades: list[tuple[int | None, FlowPageVisitGrade | None]] \
                = list(enumerate(grade_list))
            if len(grades) < max_page_count:
                grades.extend([(None, None)] * (max_page_count - len(grades)))
            grade_info.grades = grades
    else:
        page_numbers = []

    # No need to sort here, datatables resorts anyhow.

    return render_course_page(pctx, "course/gradebook-by-opp.html", {
        "opportunity": opportunity,
        "participations": participations,
        "grade_state_change_types": grade_state_change_types,
        "grade_table": grade_table,
        "batch_session_ops_form": batch_session_ops_form,
        "page_numbers": page_numbers,
        "view_page_grades": view_page_grades,

        "total_sessions": total_sessions,
        "finished_sessions": finished_sessions,
        })

# }}}


# {{{ reopen session UI

NONE_SESSION_TAG = "<<<NONE>>>"


class ReopenSessionForm(StyledForm):
    def __init__(self, flow_desc: FlowDesc, current_tag: str | None,
            *args: Any, **kwargs: Any) -> None:

        super().__init__(*args, **kwargs)

        rules = getattr(flow_desc, "rules", object())
        tags = getattr(rules, "tags", [])

        tags = [NONE_SESSION_TAG, *tags]
        self.fields["set_access_rules_tag"] = forms.ChoiceField(
                choices=[(tag, tag) for tag in tags],
                initial=(current_tag
                    if current_tag is not None
                    else NONE_SESSION_TAG),
                label=_("Set access rules tag"))

        self.fields["unsubmit_pages"] = forms.BooleanField(
                initial=True,
                required=False,
                label=_("Re-allow changes to already-submitted questions"))

        self.fields["comment"] = forms.CharField(
                widget=forms.Textarea, required=True,
                label=_("Comment"))

        self.helper.add_input(
                Submit(
                    "reopen", _("Reopen")))


@course_view
@transaction.atomic
def view_reopen_session(pctx: CoursePageContext, flow_session_id: str,
        opportunity_id: str) -> http.HttpResponse:

    if not pctx.has_permission(pperm.reopen_flow_session):
        raise PermissionDenied(_("may not reopen session"))

    request = pctx.request

    session = get_object_or_404(FlowSession, id=int(flow_session_id))

    from course.content import get_flow_desc
    try:
        flow_desc = get_flow_desc(pctx.repo, pctx.course, session.flow_id,
                pctx.course_commit_sha)
    except ObjectDoesNotExist:
        raise http.Http404()

    if request.method == "POST":
        form = ReopenSessionForm(flow_desc, session.access_rules_tag,
            request.POST, request.FILES)

        may_reopen = True
        if session.in_progress:
            messages.add_message(pctx.request, messages.ERROR,
                    _("Cannot reopen a session that's already in progress."))
            may_reopen = False

        if form.is_valid() and may_reopen:
            new_access_rules_tag = form.cleaned_data["set_access_rules_tag"]
            if new_access_rules_tag == NONE_SESSION_TAG:
                new_access_rules_tag = None

            session.access_rules_tag = new_access_rules_tag

            from relate.utils import as_local_time, format_datetime_local, local_now
            now_datetime = local_now()

            session.append_comment(
                    gettext("Session reopened at %(now)s by %(user)s, "
                        "previous completion time was '%(completion_time)s': "
                        "%(comment)s.") % {
                            "now": format_datetime_local(now_datetime),
                            "user": pctx.request.user,
                            "completion_time": format_datetime_local(
                                as_local_time(not_none(session.completion_time))),
                            "comment": form.cleaned_data["comment"]
                            })
            session.save()

            from course.flow import reopen_session
            reopen_session(now_datetime, session, suppress_log=True,
                    unsubmit_pages=form.cleaned_data["unsubmit_pages"])

            # anonymous sessions do not appear in the grading interface
            assert session.participation is not None

            return redirect("relate-view_single_grade",
                    pctx.course.identifier,
                    session.participation.id,
                    opportunity_id)

    else:
        form = ReopenSessionForm(flow_desc, session.access_rules_tag)

    return render(request, "generic-form.html", {
        "form": form,
        "form_description": _("Reopen session")
        })

# }}}


# {{{ view single grade

def average_grade(
            opportunity: GradingOpportunity
        ) -> tuple[float | Decimal | None, int]:

    grade_changes = (GradeChange.objects
            .filter(
                opportunity=opportunity,
                participation__roles__permissions__permission=(
                    pperm.included_in_grade_statistics))
            .order_by(
                "participation__id",
                "grade_time")
            .select_related("participation")
            .select_related("opportunity"))

    grades = []
    my_grade_changes: list[GradeChange] = []

    def finalize() -> None:

        if not my_grade_changes:
            return

        state_machine = GradeStateMachine()
        state_machine.consume(my_grade_changes)

        percentage = state_machine.percentage()
        if percentage is not None:
            grades.append(percentage)

        del my_grade_changes[:]

    last_participation = None
    for gchange in grade_changes:
        if last_participation != gchange.participation:
            finalize()
            last_participation = gchange.participation

        my_grade_changes.append(gchange)

    finalize()

    if grades:
        return sum(grades)/len(grades), len(grades)
    else:
        return None, 0


def get_single_grade_changes_and_state_machine(opportunity: GradingOpportunity,
        participation: Participation) -> tuple[list[GradeChange], GradeStateMachine]:

    grade_changes = list(
        GradeChange.objects.filter(
            opportunity=opportunity,
            participation=participation)
        .order_by("grade_time")
        .select_related("participation")
        .select_related("participation__user")
        .select_related("creator")
        .select_related("flow_session")
        .select_related("opportunity"))

    state_machine = GradeStateMachine()
    state_machine.consume(grade_changes, set_is_superseded=True)

    return grade_changes, state_machine


@course_view
def view_single_grade(pctx: CoursePageContext, participation_id: str,
        opportunity_id: str) -> http.HttpResponse:

    now_datetime = get_now_or_fake_time(pctx.request)

    participation = get_object_or_404(Participation,
            id=int(participation_id))

    if participation.course != pctx.course:
        raise SuspiciousOperation(_("participation does not match course"))

    opportunity = get_object_or_404(GradingOpportunity, id=int(opportunity_id))

    if pctx.course != opportunity.course:
        raise SuspiciousOperation(_("opportunity from wrong course"))

    my_grade = participation == pctx.participation
    is_privileged_view = pctx.has_permission(pperm.view_gradebook)

    if is_privileged_view:
        if not opportunity.shown_in_grade_book:
            messages.add_message(pctx.request, messages.INFO,
                    _("This grade is not shown in the grade book."))
        if not opportunity.shown_in_participant_grade_book:
            messages.add_message(pctx.request, messages.INFO,
                    _("This grade is not shown in the student grade book."))

    else:
        # unprivileged

        if not my_grade:
            raise PermissionDenied(_("may not view other people's grades"))

        if not (opportunity.shown_in_grade_book
                and opportunity.shown_in_participant_grade_book
                and opportunity.result_shown_in_participant_grade_book
                ):
            raise PermissionDenied(_("grade has not been released"))

    # {{{ modify sessions buttons

    request = pctx.request
    if pctx.request.method == "POST":
        action_re = re.compile(r"^([a-z]+)_([0-9]+)$")
        action_match = None
        for key in request.POST.keys():
            action_match = action_re.match(key)
            if action_match:
                break

        if not action_match:
            raise SuspiciousOperation(_("unknown action"))

        session = FlowSession.objects.get(id=int(action_match.group(2)))
        op = action_match.group(1)

        adjust_flow_session_page_data(
                pctx.repo, session, pctx.course.identifier,
                respect_preview=False)

        from course.flow import (
            expire_flow_session_standalone,
            finish_flow_session_standalone,
            recalculate_session_grade,
            regrade_session,
        )

        try:
            if op == "imposedl":
                if not pctx.has_permission(pperm.impose_flow_session_deadline):
                    raise PermissionDenied()

                expire_flow_session_standalone(
                        pctx.repo, pctx.course, session, now_datetime)
                messages.add_message(pctx.request, messages.SUCCESS,
                        _("Session deadline imposed."))

            elif op == "end":
                if not pctx.has_permission(pperm.end_flow_session):
                    raise PermissionDenied()

                finish_flow_session_standalone(
                        pctx.repo, pctx.course, session,
                        now_datetime=now_datetime)
                messages.add_message(pctx.request, messages.SUCCESS,
                        _("Session ended."))

            elif op == "regrade":
                if not pctx.has_permission(pperm.regrade_flow_session):
                    raise PermissionDenied()

                regrade_session(
                        pctx.repo, pctx.course, session)
                messages.add_message(pctx.request, messages.SUCCESS,
                        _("Session regraded."))

            elif op == "recalculate":
                if not pctx.has_permission(pperm.recalculate_flow_session_grade):
                    raise PermissionDenied()

                recalculate_session_grade(
                        pctx.repo, pctx.course, session)
                messages.add_message(pctx.request, messages.SUCCESS,
                        _("Session grade recalculated."))

            else:
                raise SuspiciousOperation(_("invalid session operation"))

        except KeyboardInterrupt as e:
            messages.add_message(pctx.request, messages.ERROR,
                    string_concat(
                        pgettext_lazy("Starting of Error message",
                            "Error"),
                        ": %(err_type)s %(err_str)s")
                    % {
                        "err_type": type(e).__name__,
                        "err_str": str(e)})

    # }}}

    grade_changes, state_machine = (
        get_single_grade_changes_and_state_machine(opportunity, participation))

    if opportunity.flow_id:
        flow_sessions = list(FlowSession.objects
                .filter(
                    participation=participation,
                    flow_id=opportunity.flow_id,
                    )
                .order_by("start_time"))

        from collections import namedtuple
        SessionProperties = namedtuple(
                "SessionProperties",
                ["due", "grade_description"])

        from course.content import get_flow_desc