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.detail import DetailView 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 ( ExamAuthorForm, RapidForm, MarkRapidQuestionForm, ImageFormSet, RegionForm, AbnormalityForm, ExaminationForm, AnswerFormSet, AnswerUpdateFormSet, ExamForm, ) from .models import ( Rapid, Abnormality, RapidImage, Region, Examination, Exam, Answer, CidUserAnswer, ) from generic.views import AuthorRequiredMixin, ExamCloneMixin, ExamViews, GenericViewBase from reversion.views import RevisionMixin 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, mixins from rest_framework.pagination import PageNumberPagination from .serializers import ( ExamSerializer, QuestionAnswerSerializer, RapidLateralitySerializer, RapidSerializer, UserAnswerSerializer, ) from generic.mixins import SuperuserRequiredMixin logger = logging.getLogger(__name__) def normaliseScore(score): if score == 49: return 4.5 elif score in [50, 51]: return 5 elif score in [52, 53]: return 5.5 elif score in [54]: return 6 elif score in [55, 56]: return 6.5 elif score in [57, 58]: return 7 elif score in [59]: return 7.5 elif score in [60]: return 8 return 4 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 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 rapid.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.subspecialty.first().name.all()) return render(request, "rapids/question_detail.html", {"question": rapid}) 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) @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(RevisionMixin, 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( RevisionMixin, 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 #wanted_exams = form["exams"].data #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""" 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 rapid.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('' % (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( '' % (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( '' % (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( '' % (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 @login_required def mark_all(request, exam_pk, sk): return mark(request, exam_pk, sk, unmarked_exam_answers_only=False) @login_required def mark_review(request, exam_pk, sk): return mark(request, exam_pk, sk, unmarked_exam_answers_only=False, review=True) # @user_passes_test(user_is_admin, login_url="/accounts/login") @login_required def mark(request, exam_pk, sk, unmarked_exam_answers_only=True, review=False): exam = get_object_or_404(Exam, pk=exam_pk) if not exam.exam_mode: raise Http404("Packet not in exam mode") mark_url = "rapids:mark" if review: mark_url = "rapids:mark_review" elif not unmarked_exam_answers_only: mark_url = "rapids:mark_all" 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") 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(mark_url, exam_pk=exam_pk, sk=n + 1) cd = form.cleaned_data marked_answers = json.loads(cd.get("marked_answers")) # This should probably use json (or something else...) # ajax..... to_run = ( ("correct", Answer.MarkOptions.CORRECT), ("half-correct", Answer.MarkOptions.HALF_MARK), ("incorrect", Answer.MarkOptions.INCORRECT), ) for status_string, status in to_run: for ans in marked_answers[status_string]: 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 = status a.save() if "next" in request.POST: return redirect(mark_url, exam_pk=exam_pk, sk=n + 1) elif "previous" in request.POST: return redirect(mark_url, exam_pk=exam_pk, sk=n - 1) # Reset user answers (relies on the functional javascript) unmarked_user_answers = set() else: form = MarkRapidQuestionForm() # answers_dict = {} # for ans in question.answers.all(): # answers_dict[ans.answer_compare] = ans correct_answers = [] half_mark_answers = [] incorrect_answers = [] review_user_answers = [] unmarked_answers_bool = False unmarked_user_answers = [] # this could be improved if question.normal: incorrect_answers = question.get_user_answers(exam_pk=exam_pk) if "" in incorrect_answers: incorrect_answers.remove("") if "normal" in incorrect_answers: incorrect_answers.remove("normal") else: if review: unmarked_user_answers = question.get_user_answers( exam_pk=exam_pk, include_normal=False ) elif unmarked_exam_answers_only: unmarked_user_answers = question.get_unmarked_user_answers(exam_pk=exam_pk) else: unmarked_user_answers = question.get_unmarked_user_answers() if not unmarked_exam_answers_only: 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 ) if review: for ans in unmarked_user_answers: # duplicated from user answer model.... marked_ans = question.answers.filter(answer_compare__iexact=ans).first() mark = "unmarked" if marked_ans is not None: if marked_ans.status == Answer.MarkOptions.CORRECT: mark = "correct" elif marked_ans.status == Answer.MarkOptions.HALF_MARK: mark = "half-correct" elif marked_ans.status == Answer.MarkOptions.INCORRECT: mark = "incorrect" if mark == "unmarked": unmarked_answers_bool = True review_user_answers.append((ans, mark)) return render( request, "rapids/mark.html", { "exam": exam, "form": form, "review": review, "question": question, "question_number": n, "question_details": question_details, "unmarked_answers_bool": unmarked_answers_bool, "user_answers": unmarked_user_answers, "review_user_answers": review_user_answers, "correct_answers": correct_answers, "half_mark_answers": half_mark_answers, "incorrect_answers": incorrect_answers, "unmarked_exam_answers_only": unmarked_exam_answers_only, }, ) def exam_scores_cid_user(request, pk, cid, passcode): exam = get_object_or_404(Exam, pk=pk) if not exam.exam_mode: raise Http404("Packet not in exam mode") # TODO:Need some kind of test for cid questions = exam.exam_questions.all()#.prefetch_related("cid_user_answers", "answers") #cid_user_answers = list(CidUserAnswer.objects.filter(cid=cid, exam__id=pk).prefetch_related("question")) #cid_user_answers_q_map = {} #for ans in cid_user_answers: # cid_user_answers_q_map[ans.question] = ans answers_and_marks = [] answers_marks = [] answers = [] view_all_results = False if request.user.groups.filter(name="view_all_results").exists(): view_all_results = True for q in questions: # Get user answer user_answer = q.cid_user_answers.filter(cid=cid, exam__id=pk).first() #user_answer = cid_user_answers_q_map[q] 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.get_primary_answer() if not exam.publish_results and not view_all_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) normalised_score = normaliseScore(total_score) max_score = len(questions) * 2 return render( request, "rapids/exam_scores_user.html", { "exam": exam, "cid": cid, "passcode": passcode, "questions": questions, "answers": answers, "answers_marks": answers_marks, "total_score": total_score, "normalised_score": normalised_score, "max_score": max_score, "answers_and_marks": answers_and_marks, "view_all_results": view_all_results, }, ) class QuestionDelete(AuthorOrCheckerRequiredMixin, DeleteView): model = Rapid success_url = reverse_lazy("rapids:rapid_view") GenericExamViews = ExamViews( Exam, Rapid, Answer, CidUserAnswer, "rapids", "rapid" ) class ExamCreate(RevisionMixin, LoginRequiredMixin, CreateView): model = Exam form_class = ExamForm def form_valid(self, form): self.object = form.save(commit=False) self.object.save() form.instance.author.add(self.request.user.id) return super().form_valid(form) class ExamUpdate( RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView ): model = Exam form_class = ExamForm def form_valid(self, form): self.object = form.save(commit=False) self.object.save() form.instance.author.add(self.request.user.id) return super().form_valid(form) class ExamAuthorUpdate(RevisionMixin, LoginRequiredMixin, AuthorRequiredMixin, UpdateView): model = Exam form_class = ExamAuthorForm class ExamDelete(RevisionMixin, AuthorOrCheckerRequiredMixin, DeleteView): model = Exam template_name = "exam_confirm_delete.html" success_url = reverse_lazy("rapids:index") class UserAnswerView(LoginRequiredMixin, DetailView): model = CidUserAnswer class UserAnswerTableView(LoginRequiredMixin, SingleTableMixin, FilterView): model = CidUserAnswer table_class = RapidUserAnswerTable template_name = "rapids/user_answer_question_view.html" filterset_class = RapidUserAnswerFilter class UserAnswerDelete(SuperuserRequiredMixin, DeleteView): model = CidUserAnswer template_name = "user_answer_delete.html" success_url = reverse_lazy("rapids:user_answer_table_view") class LargeResultsSetPagination(PageNumberPagination): page_size = 1000 page_size_query_param = "page_size" max_page_size = 10000 class StandardResultsSetPagination(PageNumberPagination): page_size = 100 page_size_query_param = "page_size" max_page_size = 1000 class ExamViewSet(RevisionMixin, 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 UserAnswerViewSet(viewsets.ModelViewSet): queryset = CidUserAnswer.objects.all() # .order_by('name') serializer_class = UserAnswerSerializer permission_classes = [permissions.IsAuthenticated] class RapidViewSet( RevisionMixin, mixins.CreateModelMixin, mixins.RetrieveModelMixin, mixins.UpdateModelMixin, mixins.ListModelMixin, mixins.DestroyModelMixin, viewsets.GenericViewSet, ): queryset = Rapid.objects.all() # .order_by('name') serializer_class = RapidSerializer permission_classes = [permissions.IsAuthenticated] pagination_class = StandardResultsSetPagination class RapidLateralityViewSet( RevisionMixin, mixins.CreateModelMixin, mixins.RetrieveModelMixin, mixins.UpdateModelMixin, mixins.ListModelMixin, viewsets.GenericViewSet, ): queryset = Rapid.objects.all() # .order_by('name') serializer_class = RapidLateralitySerializer permission_classes = [permissions.IsAuthenticated] pagination_class = StandardResultsSetPagination # class AddSuggestedAnswer(LoginRequiredMixin, CreateView): # model = QuestionNote # form_class = QuestionNoteForm # # #fields = ("author", "note_type", "note") # # # def get_form_kwargs(self): # kwargs = super(AddQuestionNote, self).get_form_kwargs() # # update the kwargs for the form init method with yours # kwargs.update(self.kwargs) # self.kwargs contains all url conf params # return kwargs # # def get_initial(self): # pk = self.kwargs["pk"] # question_type = self.kwargs["question_type"] # if question_type == "rapid": # question = Rapid # content_type = ContentType.objects.get(model="rapid") # elif question_type == "anatomy": # question = AnatomyQuestion # content_type = ContentType.objects.get(model="anatomyquestion") # elif question_type == "long": # question = Long # content_type = ContentType.objects.get(model="long") # else: # raise PermissionError() # # get_object_or_404(question, pk=pk) # return { "content_type" : content_type, # "object_id" : pk } GenericViews = GenericViewBase("rapids", Rapid, CidUserAnswer, Exam) @user_is_author_or_rapid_checker def question_anonymise_dicom(request, pk): question = get_object_or_404(Rapid, pk=pk) question.anonymise_images() return redirect("rapids:question_detail", pk=pk) @user_is_author_or_rapid_checker def question_save_annotation(request, pk): if request.is_ajax() and request.method == "POST": question = get_object_or_404(Rapid, pk=pk) image_annotations = json.loads(request.POST.get("annotation")) question_images = question.images.filter(feedback_image=False) for n, image in enumerate(question_images): annotation = image_annotations[n] if annotation != "undefined": image.image_annotations = annotation else: image.image_annotations = "" image.save() data = {"status": "success"} return JsonResponse(data, status=200) else: data = {"status": "error"} return JsonResponse(data, status=400) class ExamClone(ExamCloneMixin, ExamCreate): """Clone exam view""" @login_required def get_question_by_hash(request): if request.is_ajax() and request.method == "POST": hash = request.POST.get("hash") try: question = RapidImage.objects.get(image_md5_hash=hash) data = { "status": "success", "id": question.pk, "url": question.rapid.get_absolute_url(), } return JsonResponse(data, status=200) except RapidImage.DoesNotExist: data = {"status": "success", "id": False} return JsonResponse(data, status=200) # TODO: better permissions @login_required def question_json_unbased(request, pk): """ No (file based) caching is enabled for unbased quesitons """ question = get_object_or_404(Rapid, pk=pk) question_json = question.get_question_json(based=False) return JsonResponse(question_json) @login_required def question_json(request, pk): question = get_object_or_404(Rapid, pk=pk) question_json = question.get_question_json(based=True) return JsonResponse(question_json) @login_required def question_viewer(request): # question = get_object_or_404(Rapid, pk=pk) ids = request.GET.get("ids").split(",") return render( request, "rapids/question_viewer.html", { "ids": ids, }, )