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 ObjectDoesNotExist, PermissionDenied, ViewDoesNotExist from django.contrib.auth.mixins import LoginRequiredMixin from django.views.generic.detail import DetailView from generic.mixins import SuperuserRequiredMixin 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 ( LongForm, LongSeriesForm, LongSeriesImageFormSet, MarkLongQuestionForm, SeriesFormSet, NoteForm, ExaminationForm, ExamForm, ) from .models import ( Long, LongSeries, Note, Examination, Exam, CidUserAnswer, ) from .tables import LongTable, LongSeriesTable, UserAnswerTable from .filters import LongFilter, LongSeriesFilter, UserAnswerFilter from django_tables2 import SingleTableView, SingleTableMixin from django_filters.views import FilterView from .decorators import user_is_author_or_long_checker, user_is_author_or_long_checker_or_long_marker, user_is_author_or_long_series_checker_or_long_marker, user_is_long_checker, user_is_author_or_long_series_checker, user_is_long_marker, user_is_author_or_long_checker_or_long_marker from collections import defaultdict import json import statistics import plotly.express as px from django.core.cache import cache from django.conf import settings from helpers.images import image_as_base64 import logging from copy import deepcopy from django.forms.models import model_to_dict from longs.forms import LongCreationDefaultForm from longs.models import LongCreationDefault from generic.views import ExamViews from reversion.views import RevisionMixin import reversion from rest_framework import viewsets from .serializers import ExamSerializer from zipview.views import BaseZipView import os 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="long_checker").exists() or self.request.user.is_superuser: return obj if self.request.user not in obj.get_author_objects(): raise PermissionDenied() # or Http404 return obj # def index(request): # other_longs = Long.objects.exclude(author=request.user.pk) # user_longs = Long.objects.filter(author=request.user.pk) # return render( # request, # "longs/index.html", # {"other_longs": other_longs, "user_longs": user_longs}, # ) @login_required @user_is_author_or_long_checker_or_long_marker def question_detail(request, pk): long = get_object_or_404(Long, pk=pk) # if request.user not in long.author.all(): # raise PermissionDenied # logging.debug(long.long_notes.first()) # logging.debug(long.subspecialty.first().name.all()) return render(request, "longs/question_detail.html", {"question": long}) @login_required @user_is_author_or_long_series_checker_or_long_marker def long_series_detail(request, pk): series = get_object_or_404(LongSeries, pk=pk) # if request.user not in long.author.all(): # raise PermissionDenied # logging.debug(long.long_notes.first()) # logging.debug(long.subspecialty.first().name.all()) return render(request, "longs/long_series.html", {"series": series}) @login_required @user_is_author_or_long_checker def long_split(request, pk): long = get_object_or_404(Long, pk=pk) images = long.images.all() old_abnormality = long.abnormality.all() old_region = long.region.all() old_examination = long.examination.all() old_site = long.site.all() old_author = long.author.all() if not images: raise Http404 for n in range(len(images) - 1): long.pk = None long.save() long.abnormality.set(old_abnormality) long.region.set(old_region) long.examination.set(old_examination) long.site.set(old_site) long.author.set(old_author) images[n].long = long images[n].save() # images[-1].long # if request.user not in long.author.all(): # raise PermissionDenied # logging.debug(long.long_notes.first()) # logging.debug(long.subspecialty.first().name.all()) return render(request, "longs/question_detail.html", {"question": long}) @login_required @user_is_author_or_long_checker def author_detail(request, pk): # logging.debug(Author.objects.all()) # author = get_object_or_404(Author, pk=pk) author = User.objects.get(pk=pk) longs = Long.objects.filter(author=pk) return render( request, "longs/category_detail.html", {"category": author, "longs": longs} ) @user_is_author_or_long_checker def author_list(request): authors = User.objects.all() return render(request, "longs/author_list.html", {"authors": authors}) class LongCreationDefaultView(LoginRequiredMixin, UpdateView): model = LongCreationDefault form_class = LongCreationDefaultForm # 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 `Long` instance exists # before going any further. # """ # self.pk = get_object_or_404(Long, pk=kwargs['pk']) # return super().dispatch(request, *args, **kwargs) def get_object(self, queryset=None): obj, create = LongCreationDefault.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 `Long` instance exists before going any further. """ self.pk = get_object_or_404(Long, pk=kwargs["pk"]) return super().dispatch(request, *args, **kwargs) def form_valid(self, form): note = form.save(commit=False) note.long = 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 #@user_is_author_or_long_checker #def long_delete(request, pk): #q = get_object_or_404(Long, pk=pk) #q.delete() class LongDelete(RevisionMixin, AuthorOrCheckerRequiredMixin, DeleteView): model = Long success_url = reverse_lazy("longs:long_view") class LongSeriesDelete(RevisionMixin, AuthorOrCheckerRequiredMixin, DeleteView): model = LongSeries success_url = reverse_lazy("longs:long_series_view") @login_required @user_is_author_or_long_checker def long_clone(request, pk): new_item = get_object_or_404(Long, 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 = LongForm(request.POST or None, instance=new_item) series_formset = SeriesFormSet() if form.is_valid(): form.instance.author.add(request.user.id) # logger.debug(formset.is_valid()) if series_formset.is_valid(): response = super().form_valid(form) series_formset.instance = obj series_formset.save() return response else: return super().form_invalid(form) context = { "form": form, "series_formset": series_formset, # other context } return render(request, "longs/long_form.html", context) class LongSeriesCreate(RevisionMixin, LoginRequiredMixin, CreateView): model = LongSeries form_class = LongSeriesForm def get_initial(self): # print(self.request) if "pk" in self.kwargs: initial = super().get_initial() long = get_object_or_404(Long, pk=self.kwargs["pk"]) initial['long'] = long.id return initial def get_context_data(self, **kwargs): context = super(LongSeriesCreate, self).get_context_data(**kwargs) if self.request.POST: context["image_formset"] = LongSeriesImageFormSet( self.request.POST, self.request.FILES ) context["image_formset"].full_clean() else: context["image_formset"] = LongSeriesImageFormSet() 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) series_formset = context["image_formset"] if series_formset.is_valid(): response = super().form_valid(form) series_formset.instance = self.object series_formset.save() return response else: return super().form_invalid(form) class LongSeriesUpdate(RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView): model = LongSeries form_class = LongSeriesForm # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] def get_context_data(self, **kwargs): context = super(LongSeriesUpdate, self).get_context_data(**kwargs) if self.request.POST: context["image_formset"] = LongSeriesImageFormSet( self.request.POST, self.request.FILES, instance=self.object ) context["image_formset"].full_clean() else: context["image_formset"] = LongSeriesImageFormSet(instance=self.object) return context def form_valid(self, form): self.object = form.save(commit=False) self.object.save() # Do this in the form.save() #new_exams = list(form.cleaned_data['exams'].values_list("pk", flat=True)) #self.object.exams.clear() #self.object.exams.add(*new_exams) form.instance.author.add(self.request.user.id) context = self.get_context_data(form=form) image_formset = context["image_formset"] # logger.debug(formset.is_valid()) if image_formset.is_valid(): response = super().form_valid(form) image_formset.instance = self.object image_formset.save() return response else: return super().form_invalid(form) class LongCreateBase(RevisionMixin, LoginRequiredMixin, CreateView): model = Long form_class = LongForm def get_context_data(self, **kwargs): context = super(LongCreateBase, self).get_context_data(**kwargs) if self.request.POST: context["series_formset"] = SeriesFormSet( self.request.POST, self.request.FILES ) context["series_formset"].full_clean() else: context["series_formset"] = SeriesFormSet() return context def form_valid(self, form): self.object = form.save(commit=False) # Add exam objects self.object.save() # Do this in the form.save() #new_exams = list(form.cleaned_data['exams'].values_list("pk", flat=True)) #self.object.exams.clear() #self.object.exams.add(*new_exams) form.instance.author.add(self.request.user.id) context = self.get_context_data(form=form) series_formset = context["series_formset"] if series_formset.is_valid(): response = super().form_valid(form) series_formset.instance = self.object series_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("longs:long_clone", pk=self.object.pk) else: return super().form_invalid(form) # @login_required class LongCreate(LongCreateBase): def get_initial(self): # There has to be a better way... try: s = (i.pk for i in self.request.user.long_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 LongUpdate(RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView): model = Long form_class = LongForm # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] def get_context_data(self, **kwargs): context = super(LongUpdate, self).get_context_data(**kwargs) if self.request.POST: context["series_formset"] = SeriesFormSet( self.request.POST, self.request.FILES, instance=self.object ) context["series_formset"].full_clean() else: context["series_formset"] = SeriesFormSet(instance=self.object) return context def form_valid(self, form): self.object = form.save(commit=False) self.object.save() # Do this in the form.save() #new_exams = list(form.cleaned_data['exams'].values_list("pk", flat=True)) #self.object.exams.clear() #self.object.exams.add(*new_exams) form.instance.author.add(self.request.user.id) context = self.get_context_data(form=form) series_formset = context["series_formset"] # logger.debug(formset.is_valid()) if series_formset.is_valid(): response = super().form_valid(form) series_formset.instance = self.object series_formset.save() return response else: return super().form_invalid(form) class LongClone(LongCreateBase): """Clones a existing long""" # 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(Long, pk=self.kwargs["pk"]) initial_data = model_to_dict(old_object, exclude=["id"]) return initial_data @login_required @user_is_author_or_long_checker def long_scrap(request, pk): try: long = Long.objects.get(pk=pk) except Long.DoesNotExist: raise Http404("Long does not exist") long.scrapped = not long.scrapped long.save() return HttpResponseRedirect(reverse("longs: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, "longs/create_simple.html", {'form': form}) @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, "longs/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("/") class LongView(LoginRequiredMixin, SingleTableMixin, FilterView): model = Long table_class = LongTable template_name = "longs/view.html" filterset_class = LongFilter class LongSeriesView(LoginRequiredMixin, SingleTableMixin, FilterView): model = LongSeries table_class = LongSeriesTable template_name = "longs/view.html" filterset_class = LongSeriesFilter 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 False, 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 False, JsonResponse({"success": False, "error": "invalid cid"}) exam = get_object_or_404(Exam, pk=eid) #if not exam.active: # return False, 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"] # Long cases seperate sections by qidn qidn = answer["qidn"] # If the user answer does not exist if not exiting_answers: ans = CidUserAnswer(cid=answer["cid"]) ans.question_id = answer["qid"] ans.exam_id = eid # If the answer already exists or we have started populating it else: # Update an existing answer # should never be more than one (famous last words) ans = exiting_answers[0] if qidn == "1": ans.answer_observations = posted_answer elif qidn == "2": ans.answer_interpretation = posted_answer elif qidn == "3": ans.answer_principle_diagnosis = posted_answer elif qidn == "4": ans.answer_differential_diagnosis = posted_answer elif qidn == "5": ans.answer_management = posted_answer # Mark as unmarked ans.score = ans.ScoreOptions.UNMARKED ans.full_clean() ans.save() return True, None @login_required @user_is_long_marker @reversion.create_revision() def mark_answer(request, pk, sk, tk): 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") try: answer = question.cid_user_answers.get(cid=tk, exam__id=pk) except ObjectDoesNotExist: raise Http404("User answer does not exist") try: unmarked = question.get_unmarked_answers() next_unmarked_id = unmarked[0].cid if next_unmarked_id == tk: next_unmarked_id = unmarked[1].cid except IndexError: next_unmarked_id = False if request.method == "POST": form = MarkLongQuestionForm(request.POST) if form.is_valid(): # If skip button is pressed skip the question without marking # Skip is problematic if we skip to the next unmarked answer #if "skip" in request.POST: # return redirect("longs:mark_answer", pk=pk, sk=sk, tk=next_unmarked_id) # Extract score from form and save it to the object answer.score = form.cleaned_data["score"] answer.save() if "next" in request.POST: return redirect("longs:mark_answer", pk=pk, sk=sk, tk=next_unmarked_id) if "save" in request.POST: return redirect("longs:mark_answer", pk=pk, sk=sk, tk=tk) #elif "previous" in request.POST: # return redirect("longs:mark", pk=pk, sk=n - 1) else: form = MarkLongQuestionForm(initial={'score': answer.score}) return render( request, "longs/mark_answer.html", { "exam": exam, "form": form, "answer": answer, "question": question, "question_details": question_details, "next_unmarked_id": next_unmarked_id, "unmarked": unmarked, }, ) # @user_passes_test(user_is_admin, login_url="/accounts/login") @login_required @user_is_long_marker 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") user_answers = question.cid_user_answers.filter(exam__id=pk) unmarked_count = user_answers.filter(score=CidUserAnswer.ScoreOptions.UNMARKED).count() return render( request, "longs/mark.html", { "exam": exam, "question": question, "question_details": question_details, "user_answers": user_answers, "unmarked_count": unmarked_count, }, ) @login_required @user_is_long_marker 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, exam__id=pk) .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, exam__id=pk).first() if not s: # skip if no answer, (score 4) user_answers_marks[cid].append(4) #user_answers[cid].append("") by_question[q].append(("", 4)) continue else: ans = "" answer_score = s.get_answer_score() if answer_score == "": 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 != ""] ) 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, "longs/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 = [] answer_text = [] for q in questions: # Get user answer user_answer = q.cid_user_answers.filter(cid=cid, exam__id=pk).first() if not user_answer or user_answer is None: # skip if no answer #answers_marks.append("") #answers.append("") answer_score = 4 #ans = "Not answered" answer_text.append((("Not answered"),("Not answered"),("Not answered"),("Not answered"),("Not answered"))) else: answer_score = user_answer.get_answer_score() answer_text.append((user_answer.answer_observations, user_answer.answer_interpretation, user_answer.answer_principle_diagnosis, user_answer.answer_differential_diagnosis, user_answer.answer_management)) if not exam.publish_results: answer_score = 0 #answers.append(ans) answers_marks.append(answer_score) #answers_and_marks.append((ans, answer_score, correct_answer)) answered = [i for i in answers_marks if i != ""] if "" in answers_marks: total_score = sum(answered) unmarked_number = len(answers_marks) - len(answered) total_score = "{} ({} unmarked)".format(total_score, unmarked_number) else: total_score = sum(answered) max_score = len(questions) * 2 return render( request, "longs/exam_scores_user.html", { "exam": exam, "cid": cid, "questions": questions, #"answers": answers, "answers_marks": answers_marks, "total_score": total_score, "max_score": max_score, "answer_text": answer_text #"answers_and_marks": answers_and_marks, }, ) @login_required @user_is_author_or_long_series_checker def long_series_order_dicom(request, pk): series = get_object_or_404(LongSeries, pk=pk) try: series.order_by_dicom() except: return HttpResponse('