Source code for analytics_data_api.v0.views.problems
"""
API methods for module level data.
"""
from collections import defaultdict
from itertools import groupby
from django.db import OperationalError
from rest_framework import generics
from analytics_data_api.utils import matching_tuple
from analytics_data_api.v0.models import (
GradeDistribution,
ProblemFirstLastResponseAnswerDistribution,
ProblemResponseAnswerDistribution,
SequentialOpenDistribution,
)
from analytics_data_api.v0.serializers import (
ConsolidatedAnswerDistributionSerializer,
ConsolidatedFirstLastAnswerDistributionSerializer,
GradeDistributionSerializer,
SequentialOpenDistributionSerializer,
)
from analytics_data_api.v0.views.utils import raise_404_if_none
[docs]class ProblemResponseAnswerDistributionView(generics.ListAPIView):
"""
Get the distribution of student answers to a specific problem.
**Example request**
GET /api/v0/problems/{problem_id}/answer_distribution
**Response Values**
Returns a collection for each unique answer given to specified
problem. Each collection contains:
* course_id: The ID of the course for which data is returned.
* module_id: The ID of the problem.
* part_id: The ID for the part of the problem. For multi-part
problems, a collection is returned for each part.
* correct: Whether the answer was correct (``true``) or not
(``false``).
* count: The number of times the answer in this collection was
given.
* value_id: The ID of the answer in this collection.
* answer_value: An answer for this problem.
* problem_display_name: The display name for the specified problem.
* question_text: The question for the specified problem.
* variant: For randomized problems, the random seed used. If problem
is not randomized, value is null.
* created: The date the count was computed.
"""
serializer_class = ConsolidatedAnswerDistributionSerializer
allow_empty = False
@classmethod
def consolidate_answers(cls, problem):
""" Attempt to consolidate erroneously randomized answers. """
answer_sets = defaultdict(list)
match_tuple_sets = defaultdict(set)
for answer in problem:
answer.consolidated_variant = False
answer_sets[answer.value_id].append(answer)
match_tuple_sets[answer.value_id].add(matching_tuple(answer))
# If a part has more than one unique tuple of matching fields, do not consolidate.
for _, match_tuple_set in match_tuple_sets.items():
if len(match_tuple_set) > 1:
return problem
consolidated_answers = []
for _, answers in answer_sets.items():
consolidated_answer = None
if len(answers) == 1:
consolidated_answers.append(answers[0])
continue
for answer in answers:
if consolidated_answer:
if isinstance(consolidated_answer, ProblemResponseAnswerDistribution):
consolidated_answer.count += answer.count
else:
consolidated_answer.first_response_count += answer.first_response_count
consolidated_answer.last_response_count += answer.last_response_count
else:
consolidated_answer = answer
consolidated_answer.variant = None
consolidated_answer.consolidated_variant = True
consolidated_answers.append(consolidated_answer)
return consolidated_answers
@raise_404_if_none
def get_queryset(self):
"""Select all the answer distribution response having to do with this usage of the problem."""
problem_id = self.kwargs.get('problem_id')
try:
queryset = list(ProblemResponseAnswerDistribution.objects.filter(module_id=problem_id).order_by('part_id'))
except OperationalError:
self.serializer_class = ConsolidatedFirstLastAnswerDistributionSerializer
queryset = list(ProblemFirstLastResponseAnswerDistribution.objects.filter(
module_id=problem_id).order_by('part_id'))
consolidated_rows = []
for _, part in groupby(queryset, lambda x: x.part_id):
consolidated_rows += self.consolidate_answers(list(part))
return consolidated_rows
[docs]class GradeDistributionView(generics.ListAPIView):
"""
Get the distribution of grades for a specific problem.
**Example request**
GET /api/v0/problems/{problem_id}/grade_distribution
**Response Values**
Returns a collection for each unique grade given to a specified
problem. Each collection contains:
* course_id: The ID of the course for which data is returned.
* module_id: The ID of the problem.
* grade: The grade being counted in this collection.
* count: The number of times the grade in this collection was
given.
* max_grade: The highest possible grade for this problem.
* created: The date the count was computed.
"""
serializer_class = GradeDistributionSerializer
allow_empty = False
@raise_404_if_none
def get_queryset(self):
"""Select all grade distributions for a particular module"""
problem_id = self.kwargs.get('problem_id')
return GradeDistribution.objects.filter(module_id=problem_id)
[docs]class SequentialOpenDistributionView(generics.ListAPIView):
"""
Get the number of views of a subsection, or sequential, in the course.
**Example request**
GET /api/v0/problems/{module_id}/sequential_open_distribution
**Response Values**
Returns a collection that contains the number of views of the specified
problem. The collection contains:
* course_id: The ID of the course for which data is returned.
* module_id: The ID of the subsection, or sequential.
* count: The number of times the subsection was viewed.
* created: The date the count computed.
"""
serializer_class = SequentialOpenDistributionSerializer
allow_empty = False
@raise_404_if_none
def get_queryset(self):
"""Select the view count for a specific module"""
module_id = self.kwargs.get('module_id')
return SequentialOpenDistribution.objects.filter(module_id=module_id)