Files
penracourses/rapids/views.py
T
Ross 4ea6d59552 .
2021-04-28 17:10:00 +01:00

971 lines
29 KiB
Python
Executable File

import json
from django.shortcuts import render, get_object_or_404, redirect
from django import forms
# from django.contrib.auth.models import User
from django.contrib.auth.decorators import login_required, user_passes_test
from django.contrib.auth.models import User
from django.core.exceptions import PermissionDenied
from django.contrib.auth.mixins import LoginRequiredMixin
from django.views.generic.edit import CreateView, UpdateView, DeleteView
from django.views.generic import ListView
from django.views.decorators.csrf import csrf_exempt
from django.urls import reverse_lazy, reverse
from django.http import Http404, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from .forms import (
RapidForm,
MarkRapidQuestionForm,
ImageFormSet,
NoteForm,
RegionForm,
AbnormalityForm,
ExaminationForm,
AnswerFormSet,
AnswerUpdateFormSet,
ExamForm
)
from .models import (
Rapid,
Note,
Abnormality,
Region,
Examination,
Exam,
Answer,
CidUserAnswer,
)
from generic.views import ExamViews
from .tables import RapidTable, RapidUserAnswerTable
from .filters import RapidFilter, RapidUserAnswerFilter
from django_tables2 import SingleTableView, SingleTableMixin
from django_filters.views import FilterView
from .decorators import user_is_author_or_rapid_checker
from collections import defaultdict
import json
import statistics
import plotly.express as px
from django.core.cache import cache
from helpers.images import image_as_base64
import logging
from copy import deepcopy
from django.forms.models import model_to_dict
from rapids.forms import RapidCreationDefaultForm
from rapids.models import RapidCreationDefault
from rest_framework import viewsets, permissions
from .serializers import ExamSerializer, QuestionAnswerSerializer, RapidSerializer
from generic.mixins import SuperuserRequiredMixin
logger = logging.getLogger(__name__)
class AuthorOrCheckerRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
if self.request.user.groups.filter(name="rapid_checker").exists():
return obj
if self.request.user not in obj.author.all():
raise PermissionDenied() # or Http404
return obj
@login_required
def question_detail(request, pk):
rapid = get_object_or_404(Rapid, pk=pk)
if not rapid.open_access:
if not request.user.groups.filter(name="rapid_checker").exists() and request.user not in rapid.author.all():
raise PermissionDenied()
# if request.user not in rapid.author.all():
# raise PermissionDenied
# logging.debug(rapid.rapid_notes.first())
# logging.debug(rapid.subspecialty.first().name.all())
return render(request, "rapids/question_detail.html", {"question": rapid})
@login_required
def rapid_split(request, pk):
rapid = get_object_or_404(Rapid, pk=pk)
if not request.user.groups.filter(name="rapid_checker").exists() and request.user not in question.author.all():
raise PermissionDenied()
images = rapid.images.all()
old_abnormality = rapid.abnormality.all()
old_region = rapid.region.all()
old_examination = rapid.examination.all()
old_site = rapid.site.all()
old_author = rapid.author.all()
if not images:
raise Http404
for n in range(len(images) - 1):
rapid.pk = None
rapid.save()
rapid.abnormality.set(old_abnormality)
rapid.region.set(old_region)
rapid.examination.set(old_examination)
rapid.site.set(old_site)
rapid.author.set(old_author)
images[n].rapid = rapid
images[n].save()
# images[-1].rapid
# if request.user not in rapid.author.all():
# raise PermissionDenied
# logging.debug(rapid.rapid_notes.first())
# logging.debug(rapid.subspecialty.first().name.all())
return render(request, "rapids/question_detail.html", {"question": rapid})
@login_required
def author_detail(request, pk):
# logging.debug(Author.objects.all())
# author = get_object_or_404(Author, pk=pk)
author = User.objects.get(pk=pk)
rapids = Rapid.objects.filter(author=pk)
return render(
request, "rapids/category_detail.html", {"category": author, "rapids": rapids}
)
def author_list(request):
authors = User.objects.all()
return render(request, "rapids/author_list.html", {"authors": authors})
class RapidCreationDefaultView(LoginRequiredMixin, UpdateView):
model = RapidCreationDefault
form_class = RapidCreationDefaultForm
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
# def dispatch(self, request, *args, **kwargs):
# """
# Overridden so we can make sure the `Rapid` instance exists
# before going any further.
# """
# self.pk = get_object_or_404(Rapid, pk=kwargs['pk'])
# return super().dispatch(request, *args, **kwargs)
def get_object(self, queryset=None):
obj, create = RapidCreationDefault.objects.get_or_create(
author=self.request.user
)
return obj
def form_valid(self, form):
model = form.save(commit=False)
model.author = self.request.user
model.save()
response = super().form_valid(form)
return response
# form.instance.author.add(self.request.user.id)
class AddNote(LoginRequiredMixin, CreateView):
model = Note
form_class = NoteForm
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
def dispatch(self, request, *args, **kwargs):
"""
Overridden so we can make sure the `Rapid` instance exists
before going any further.
"""
self.pk = get_object_or_404(Rapid, pk=kwargs["pk"])
return super().dispatch(request, *args, **kwargs)
def form_valid(self, form):
note = form.save(commit=False)
note.rapid = self.pk
note.author = self.request.user
note.save()
response = super().form_valid(form)
return response
# form.instance.author.add(self.request.user.id)
@login_required
def rapid_clone(request, pk):
new_item = get_object_or_404(Rapid, pk=pk)
new_item.pk = None # autogen a new pk (item_id)
# new_item.name = "Copy of " + new_item.name #need to change uniques
form = RapidForm(request.POST or None, instance=new_item)
image_formset = ImageFormSet()
answer_formset = AnswerFormSet()
if form.is_valid():
form.instance.author.add(request.user.id)
# logger.debug(formset.is_valid())
if image_formset.is_valid() and answer_formset.is_valid():
response = super().form_valid(form)
image_formset.instance = obj
image_formset.save()
answer_formset.instance = obj
answer_formset.save()
return response
else:
return super().form_invalid(form)
context = {
"form": form,
"image_formset": image_formset,
"answer_formset": answer_formset
# other context
}
return render(request, "rapids/rapid_form.html", context)
class RapidCreateBase(LoginRequiredMixin, CreateView):
model = Rapid
form_class = RapidForm
# Sending user object to the form, to verify which fields to display/remove (depending on group)
def get_form_kwargs(self):
kwargs = super(RapidCreateBase, self).get_form_kwargs()
kwargs.update({'user': self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(RapidCreateBase, self).get_context_data(**kwargs)
if self.request.POST:
context["image_formset"] = ImageFormSet(
self.request.POST, self.request.FILES
)
context["image_formset"].full_clean()
context["answer_formset"] = AnswerFormSet(self.request.POST)
context["answer_formset"].full_clean()
else:
context["image_formset"] = ImageFormSet()
context["answer_formset"] = AnswerFormSet()
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
image_formset = context["image_formset"]
answer_formset = context["answer_formset"]
if image_formset.is_valid() and answer_formset.is_valid():
response = super().form_valid(form)
image_formset.instance = self.object
image_formset.save()
answer_formset.instance = self.object
answer_formset.save()
# If the normal submit button is pressed we save as normal
if "submit" in self.request.POST:
return response
# else we redirect to the clone url
else:
return redirect("rapids:rapid_clone", pk=self.object.pk)
else:
return super().form_invalid(form)
# @login_required
class RapidCreate(RapidCreateBase):
#initial = {"laterality": Rapid.NONE}
def get_initial(self):
if "pk" in self.kwargs:
initial = super().get_initial()
exam = get_object_or_404(Exam, pk=self.kwargs["pk"])
initial['exams'] = [exam.id]
return initial
## There has to be a better way...
#try:
# s = (i.pk for i in self.request.user.rapid_default.site.all())
# self.initial.update({"site": s})
#except AttributeError:
# pass
#return self.initial
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
# self.object = form.save(commit=False)
# self.object.save()
# form.instance.author.add(self.request.user.id)
# return super().form_valid(form)
class RapidUpdate(LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView):
model = Rapid
form_class = RapidForm
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
def get_form_kwargs(self):
kwargs = super(RapidUpdate, self).get_form_kwargs()
kwargs.update({'user': self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(RapidUpdate, self).get_context_data(**kwargs)
if self.request.POST:
context["image_formset"] = ImageFormSet(
self.request.POST, self.request.FILES, instance=self.object
)
context["image_formset"].full_clean()
context["answer_formset"] = AnswerFormSet(
self.request.POST, instance=self.object
)
context["answer_formset"].full_clean()
else:
context["image_formset"] = ImageFormSet(instance=self.object)
context["answer_formset"] = AnswerFormSet(instance=self.object)
return context
def form_valid(self, form):
# save exam orders (there must be a better way to do this)
exam_orders = {}
for exam in self.object.exams.all():
exam_orders[exam] = list(exam.exam_questions.all())
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
image_formset = context["image_formset"]
answer_formset = context["answer_formset"]
# logger.debug(formset.is_valid())
if image_formset.is_valid() and answer_formset.is_valid():
response = super().form_valid(form)
image_formset.instance = self.object
image_formset.save()
answer_formset.instance = self.object
answer_formset.save()
# restore exam orders
for exam in self.object.exams.all():
if exam in exam_orders and self.object in exam_orders[exam]:
print(exam_orders[exam])
exam.exam_questions.set(exam_orders[exam])
exam.save()
return response
else:
return super().form_invalid(form)
class RapidClone(RapidCreateBase):
"""Clones a existing rapid"""
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
def get_initial(self):
# print(self.request)
old_object = get_object_or_404(Rapid, pk=self.kwargs["pk"])
initial_data = model_to_dict(old_object, exclude=["id"])
return initial_data
@login_required
@user_is_author_or_rapid_checker
def rapid_scrap(request, pk):
try:
rapid = Rapid.objects.get(pk=pk)
except Rapid.DoesNotExist:
raise Http404("Rapid does not exist")
if not request.user.groups.filter(name="rapid_checker").exists() and request.user not in question.author.all():
raise PermissionDenied()
rapid.scrapped = not rapid.scrapped
rapid.save()
return HttpResponseRedirect(reverse("rapids:question_detail", args=(pk,)))
# @login_required
# def edit_abnormality_popup(request):
# instance = get_object_or_404(Abnormality, pk = pk)
# form = AbnormalityForm(request.POST or None)
# if form.is_valid():
# instance = form.save()
# return HttpResponse('<script>opener.closePopup(window, "%s", "%s", "#id_abnormality");</script>' % (instance.pk, instance))
# return render(request, "rapids/create_simple.html", {'form': form})
@login_required
def create_abnormality(request):
form = AbnormalityForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_abnormality");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Abnormality"}
)
@csrf_exempt
def get_abnormality_id(request):
if request.is_ajax():
abnormality_name = request.GET["abnormality_name"]
abnormality_id = Abnormality.objects.get(name=abnormality_name).id
data = {
"abnormality_id": abnormality_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
@login_required
def create_examination(request):
form = ExaminationForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_examination");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Examination"}
)
@csrf_exempt
def get_examination_id(request):
if request.is_ajax():
examination_name = request.GET["examination_name"]
examination_id = Examination.objects.get(name=examination_name).id
data = {
"examination_id": examination_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
@login_required
def create_region(request):
form = RegionForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_region");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Region"}
)
@csrf_exempt
def get_region_id(request):
if request.is_ajax():
region_name = request.GET["region_name"]
region_id = Region.objects.get(name=region_name).id
data = {
"region_id": region_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
class RapidView(LoginRequiredMixin, SingleTableMixin, FilterView):
model = Rapid
table_class = RapidTable
template_name = "rapids/view.html"
filterset_class = RapidFilter
def loadJsonAnswer(answer):
eid = int(answer["eid"].split("/")[1])
# As access is not restricted make sure the data appears valid
if (not isinstance(answer["cid"], int)) or (
not isinstance(eid, int) or (not isinstance(answer["ans"], str))
):
return JsonResponse(
{"success": False, "error": "cid or eid or answers not defined"}
)
# The model should catch invalid data but this should be less intensive
max_int = 999999999999999999999
if answer["cid"] > max_int or eid > max_int:
return JsonResponse({"success": False, "error": "invalid cid"})
exam = get_object_or_404(Exam, pk=eid)
if not exam.active:
return JsonResponse(
{"success": False, "error": "No active exam: {}".format(eid)}
)
exiting_answers = CidUserAnswer.objects.filter(
question__id=answer["qid"], exam__id=eid, cid=answer["cid"]
)
posted_answer = answer["ans"]
# qidn 1 does not hold answer data (but should always arrive prior to 2)
if answer["qidn"] == "1":
posted_answer = ""
# Normal answers are just posted with the answer "Normal"
normal = False
if posted_answer == "Normal":
normal = True
if not exiting_answers:
ans = CidUserAnswer(answer=posted_answer, normal=normal, cid=answer["cid"])
ans.question_id = answer["qid"]
ans.exam_id = eid
ans.full_clean()
ans.save()
else:
# Update an existing answer
# should never be more than one (famous last words)
ans = exiting_answers[0]
if answer["qidn"] == "1":
ans.normal = normal
elif answer["qidn"] == "2":
ans.answer = posted_answer
ans.full_clean()
ans.save()
return True
# @user_passes_test(user_is_admin, login_url="/accounts/login")
@login_required
def mark(request, pk, sk):
exam = get_object_or_404(Exam, pk=pk)
questions = exam.exam_questions.all()
n = sk
question_details = {
"total": len(questions),
"current": n + 1,
}
try:
question = questions[sk]
except IndexError:
raise Http404("Exam question does not exist")
answers_dict = {}
for ans in question.answers.all():
answers_dict[ans.answer_compare] = ans
marked_answers_set = set(answers_dict.keys())
# correct_answers = [i.answer.lower() for i in question.answers.all()]
# half_correct_answers = [
# i.answer.lower() for i in question.half_mark_answers.all()
# ]
# incorrect_answers = [
# i.answer.lower() for i in question.incorrect_answers.all()
# ]
unmarked_user_answers = question.GetUnmarkedAnswers()
if request.method == "POST":
form = MarkRapidQuestionForm(request.POST)
# *******************************
# TODO: convert to JSON request
# *******************************
if form.is_valid():
# If skip button is pressed skip the question without marking
if "skip" in request.POST:
return redirect("rapids:mark", pk=pk, sk=n + 1)
cd = form.cleaned_data
# correct = cd.get("correct")
# half_correct = cd.get("half_correct")
# incorrect = cd.get("incorrect")
marked_answers = json.loads(cd.get("marked_answers"))
## This is mildly dangerous as we delete all answers
# question.answers.all().delete()
# question.half_mark_answers.all().delete()
# question.incorrect_answers.all().delete()
# This should probably use json (or something else...)
# ajax.....
for ans in marked_answers["correct"]:
ans = ans.strip()
if ans == "":
continue
a = Answer.objects.filter(
answer__iexact=ans, question_id=question.pk
).first()
if a is None:
a = Answer()
a.question_id = question.pk
a.answer = ans
a.status = Answer.MarkOptions.CORRECT
a.save()
# for ans in half_correct.split("--//--"):
for ans in marked_answers["half-correct"]:
ans = ans.strip()
if ans == "":
continue
a = Answer.objects.filter(
answer__iexact=ans, question_id=question.pk
).first()
if a is None:
a = Answer()
a.question_id = question.pk
a.answer = ans
a.status = Answer.MarkOptions.HALF_MARK
a.save()
for ans in marked_answers["incorrect"]:
ans = ans.strip()
if ans == "":
continue
a = Answer.objects.filter(
answer__iexact=ans, question_id=question.pk
).first()
if a is None:
a = Answer()
a.question_id = question.pk
a.answer = ans
a.status = Answer.MarkOptions.INCORRECT
a.save()
# answer = form.save(commit=False)
# answer.user = request.user
# answer.question = question
# answer.published_date = timezone.now()
# answer.save()
if "next" in request.POST:
return redirect("rapids:mark", pk=pk, sk=n + 1)
elif "previous" in request.POST:
return redirect("rapids:mark", pk=pk, sk=n - 1)
# Reset user answers (relies on the functional javascript)
unmarked_user_answers = set()
else:
form = MarkRapidQuestionForm()
correct_answers = question.answers.filter(status=Answer.MarkOptions.CORRECT)
half_mark_answers = question.answers.filter(status=Answer.MarkOptions.HALF_MARK)
incorrect_answers = question.answers.filter(status=Answer.MarkOptions.INCORRECT)
return render(
request,
"rapids/mark.html",
{
"exam": exam,
"form": form,
"question": question,
"question_details": question_details,
"user_answers": unmarked_user_answers,
"correct_answers": correct_answers,
"half_mark_answers": half_mark_answers,
"incorrect_answers": incorrect_answers,
},
)
@login_required
def exam_scores_cid(request, pk):
exam = get_object_or_404(Exam, pk=pk)
questions = exam.exam_questions.all()
cids = (
CidUserAnswer.objects.filter(question__in=questions)
.order_by("cid")
.values_list("cid", flat=True)
.distinct()
)
user_answers_and_marks = defaultdict(list)
user_answers_marks = defaultdict(list)
user_answers = defaultdict(list)
user_names = {}
by_question = defaultdict(list)
unmarked = set()
# Loop through all candidates
for cid in cids:
# Convoluted (probably...)
user_names[cid] = cid
for q in questions:
# Get user answer
s = q.cid_user_answers.filter(cid=cid).first()
if not s:
# skip if no answer
user_answers_marks[cid].append(0)
user_answers[cid].append("")
by_question[q].append(("", 0))
continue
elif s.normal:
ans = "Normal"
else:
ans = s.answer
answer_score = s.get_answer_score()
if answer_score == "unmarked":
index = exam.get_question_index(q)
unmarked.add(index)
user_answers[cid].append(ans)
user_answers_marks[cid].append(answer_score)
user_answers_and_marks[cid].append((ans, answer_score))
by_question[q].append((ans, answer_score))
user_scores = {}
for user in user_answers_marks:
user_scores[user] = sum(
[i for i in user_answers_marks[user] if i != "unmarked"]
)
user_scores_list = list(user_scores.values())
if len(user_scores_list) < 1:
mean = 0
median = 0
mode = 0
fig_html = ""
else:
mean = statistics.mean(user_scores_list)
median = statistics.median(user_scores_list)
try:
mode = statistics.mode(user_scores_list)
except statistics.StatisticsError:
mode = "No unique mode"
df = user_scores_list
fig = px.histogram(
df,
x=0,
title="{}: distribution of scores".format(exam),
labels={"0": "Score"},
height=400,
width=600,
)
fig_html = fig.to_html()
max_score = len(questions) * 2
return render(
request,
"rapids/exam_scores.html",
{
"cids": cids,
"exam": exam,
"unmarked": unmarked,
"questions": questions,
"by_question": by_question,
"user_answers": dict(user_answers),
"user_answers_marks": dict(user_answers_marks),
"user_scores": user_scores,
"user_scores_list": user_scores_list,
"user_names": user_names,
"user_answers_and_marks": user_answers_and_marks,
"max_score": max_score,
"mean": mean,
"median": median,
"mode": mode,
"plot": fig_html,
},
)
def exam_scores_cid_user(request, pk, sk):
exam = get_object_or_404(Exam, pk=pk)
# TODO:Need some kind of test for cid
cid = sk
questions = exam.exam_questions.all()
answers_and_marks = []
answers_marks = []
answers = []
for q in questions:
# Get user answer
user_answer = q.cid_user_answers.filter(cid=cid).first()
if not user_answer or user_answer is None:
# skip if no answer
answers_marks.append(0)
#answers.append("")
answer_score = 0
ans = "Not answered"
else:
if user_answer.normal:
ans = "Normal"
else:
ans = user_answer.answer
answer_score = user_answer.get_answer_score()
correct_answer = q.GetPrimaryAnswer()
if not exam.publish_results:
correct_answer = "*****"
answer_score = 0
answers.append(ans)
answers_marks.append(answer_score)
answers_and_marks.append((ans, answer_score, correct_answer))
if "unmarked" in answers_marks:
answered = [i for i in answers_marks if type(i) == int]
total_score = sum(answered)
unmarked_number = len(answers_marks) - len(answered)
total_score = "{} ({} unmarked)".format(total_score, unmarked_number)
else:
total_score = sum(answers_marks)
max_score = len(questions) * 2
return render(
request,
"rapids/exam_scores_user.html",
{
"exam": exam,
"cid": cid,
"questions": questions,
"answers": answers,
"answers_marks": answers_marks,
"total_score": total_score,
"max_score": max_score,
"answers_and_marks": answers_and_marks,
},
)
class QuestionDelete(AuthorOrCheckerRequiredMixin, DeleteView):
model = Rapid
success_url = reverse_lazy("rapids:rapid_view")
RapidExamViews = ExamViews(Exam, Rapid, "rapids", "rapid", loadJsonAnswer)
class ExamCreate(LoginRequiredMixin, CreateView):
model = Exam
form_class = ExamForm
class ExamUpdate(LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView):
model = Exam
form_class = ExamForm
class RapidUserAnswerView(LoginRequiredMixin, SingleTableMixin, FilterView):
model = CidUserAnswer
table_class = RapidUserAnswerTable
template_name = "rapids/rapid_question_view.html"
filterset_class = RapidUserAnswerFilter
class UserAnswerDelete(SuperuserRequiredMixin, DeleteView):
model = CidUserAnswer
success_url = reverse_lazy("rapids:rapid_user_answer_view")
class ExamViewSet(viewsets.ModelViewSet):
#queryset = Exam.objects.all().order_by('name')
serializer_class = ExamSerializer
def get_queryset(self):
"""
This view should return a list exams available to a user.
"""
user = self.request.user
if user.groups.filter(name="rapid_checker").exists():
return Exam.objects.all()
return Exam.objects.filter(author__id=user.id)
class QuestionAnswerViewSet(viewsets.ModelViewSet):
queryset = Answer.objects.all()#.order_by('name')
serializer_class = QuestionAnswerSerializer
permission_classes = [permissions.IsAuthenticated]
class RapidViewSet(viewsets.ModelViewSet):
queryset = Rapid.objects.all()#.order_by('name')
serializer_class = RapidSerializer
permission_classes = [permissions.IsAuthenticated]