penracourses/shorts/models.py

from collections import defaultdict
import os
from django.db import models
from django.urls import reverse
from django.conf import settings
import pydicom
from atlas.models import Finding, Structure, Condition
from rad.settings import REMOTE_URL
from django.utils.translation import gettext_lazy as _
import dicognito
import json

# Create your models here.
from generic.models import (
    CidUser,
    CidUserGroup,
    ExamBase,
    ExamCollection,
    ExamUserStatus,
    FindingBase,
    QuestionBase,
    UserAnswerBase,
    UserUserGroup,
    get_pretty_json,
)
from django.contrib.contenttypes.fields import GenericRelation
from django.core.validators import MaxValueValidator, MinValueValidator
from django.utils.html import mark_safe

from rapids.models import Abnormality, Examination, Region

from helpers.images import get_image_hash, image_as_base64


def image_directory_path(instance, filename):
    # return u"{0}".format(filename)
    return "shorts/picture/{0}".format(filename)


class Question(QuestionBase):
    """ """

    history = models.TextField(
        null=True,
        blank=True,
        help_text="Single line history for the question, e.g. Age 14, male. Referral from ED. History: Painful wrist after falling off skateboard.",
    )

    examination = models.ManyToManyField(
        Examination, help_text="Name of the (primary) examination"
    )

    marking_guidance = models.TextField(
        null=True,
        blank=True,
        help_text="Marking guidance for the question. This is not shown to the candidate but is used by the marker to help them mark the question.",
    )

    author = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        help_text="Author of question",
        related_name="shorts_authored_questions",
    )

    def get_app_name(self):
        return "shorts"

    def get_findings_url(self):
        return reverse("shorts:question_findings", kwargs={"question_id": self.pk})

    def get_absolute_url(self):
        return reverse("shorts:question_detail", kwargs={"pk": self.pk})

    def get_images(self, feedback=False):
        qs = self.images.all()

        if not feedback:
            images = [i.image for i in qs if not i.feedback_image]
        else:
            images = [i.image for i in qs]

        return images

    def get_image_urls(self):
        return ",".join([f"{REMOTE_URL}{i.url}" for i in self.get_images()])

    def get_image_url_array(self):
        return json.dumps([f"{REMOTE_URL}{i.url}" for i in self.get_images()])

    def anonymise_images(self):
        anonymizer = dicognito.anonymizer.Anonymizer()

        for image in self.images.all():
            file_path = os.path.join(settings.MEDIA_ROOT, image.image.name)

            try:
                with pydicom.dcmread(file_path) as dataset:
                    anonymizer.anonymize(dataset)
                    dataset.save_as(file_path)
            except pydicom.errors.InvalidDicomError:
                pass

    def check_user_can_edit(self, user):
        if user.is_superuser:
            return True

        if self.author.filter(id=user.id).exists():
            return True

        if user.groups.filter(name="shorts_checker").exists():
            return True

        return False

    def get_best_sample_answer(self):
        return self.sample_answers.filter(score__gt=0).order_by("-score").first()

        

class QuestionImage(models.Model):
    question = models.ForeignKey(
        Question, related_name="images", on_delete=models.CASCADE
    )
    image = models.FileField(upload_to=image_directory_path)

    # Image annotations are stored as findings in shorts cases
    #image_annotations = models.TextField(
    #    blank=True,
    #    null=True,
    #    help_text="Stores a JSON representation of annotations to be applied by cornerstonetools",
    #)

    feedback_image = models.BooleanField(default=False)

    filename = models.CharField(
        max_length=255,
        null=True,
        blank=True,
        help_text="Name of the original file when uploaded",
    )

    description = models.CharField(max_length=255, null=True, blank=True)

    image_md5_hash = models.CharField(max_length=32, null=True, blank=True)

    is_dicom = models.BooleanField(default=False)

    def image_tag(self):
        if self.image:
            return mark_safe(
                '<img src="{}" class="admin-rapid-image" /><span class="admin-rapid-image-info">Click and hold to zoom<span>'.format(
                    self.image.url
                )
            )
        else:
            return ""

    image_tag.short_description = "Image"

    def save(self, *args, **kwargs):
        """Override save method to add image hash"""
        if self.image:
            image_hash, is_dicom = get_image_hash(
                self.image, hash_type="md5", direct_pixel_data=False
            )
            self.is_dicom = is_dicom
            self.image_md5_hash = image_hash

        # Hack for tests
        if image_hash != "12345ABCD":
            super().save(*args, **kwargs)  # Call the "real" save() method.


class SampleAnswer(models.Model):
    """Model that defines sample answers for the question.

    These are used to aid marking and for feedback purposes

    """

    question = models.ForeignKey(
        Question, related_name="sample_answers", on_delete=models.CASCADE
    )
    answer = models.TextField()

    score = models.IntegerField(
        default=0, blank=True, validators=[MinValueValidator(0), MaxValueValidator(5)]
    )

    proposed = models.BooleanField(default=False)

    def __str__(self):
        return self.answer


class ExamQuestionDetail(models.Model):
    exam = models.ForeignKey("Exam", on_delete=models.CASCADE)
    question = models.ForeignKey(Question, on_delete=models.CASCADE)

    sort_order = models.IntegerField(default=1000)

    class Meta:
        ordering = ("sort_order",)


class Exam(ExamBase):
    app_name = "shorts"

    exam_questions = models.ManyToManyField(
        Question, through=ExamQuestionDetail, related_name="exams"
    )

    time_limit = models.IntegerField(
        help_text="Exam time limit (in seconds). Default is 2 hours",
        default=2 * 60 * 60,
    )

    author = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        help_text="Author of exam",
        related_name="shorts_exam_author",
    )

    markers = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        help_text="Authorised markers for the exam",
        related_name="shorts_exam_markers",
    )

    valid_cid_users = models.ManyToManyField(
        CidUser, blank=True, related_name="shorts_exams"
    )

    valid_user_users = models.ManyToManyField(
        settings.AUTH_USER_MODEL, blank=True, related_name="user_shorts_exams"
    )

    cid_user_groups = models.ManyToManyField(
        CidUserGroup,
        blank=True,
        help_text="These groups define which candidates are able to be added to the exams/collection.",
        related_name="shorts_cid_user_groups",
    )

    user_user_groups = models.ManyToManyField(
        UserUserGroup,
        blank=True,
        help_text="These groups define which candidates are able to be added to the exams/collection.",
        related_name="shorts_user_user_groups",
    )

    exam_user_status = GenericRelation(
        ExamUserStatus, related_query_name="shorts_exams"
    )
    cid_user_exam = GenericRelation(
        "generic.CidUserExam", related_query_name="shorts_exams"
    )

    examcollection = models.ForeignKey(
        ExamCollection,
        blank=True,
        null=True,
        on_delete=models.SET_NULL,
        related_name="shorts_exams",
    )

    def get_app_name(self):
        return "shorts"

    def get_exam_json(self, based=True):
        # TODO

        return {}
        questions = self.get_questions()

        exam_questions = defaultdict(dict)

        exam_order = []

        q: Rapid
        for q in questions:
            exam_order.append(q.id)

            # Loop through rapidimage associations
            images = []
            feedback_images = []
            image_titles = []
            for i in q.images.all():
                if i.feedback_image == True:
                    if based:
                        feedback_images.append(image_as_base64(i.image))
                    else:
                        feedback_images.append(
                            "{}/{}".format(settings.REMOTE_URL, i.image.url)
                        )
                else:
                    if based:
                        images.append(image_as_base64(i.image))
                    else:
                        images.append("{}/{}".format(settings.REMOTE_URL, i.image.url))

                if i.description:
                    image_titles.append(i.description)
                else:
                    image_titles.append("")

            exam_questions[q.id] = {
                "images": images,
                # "feedback_image": [],
                # "annotations": [str(q.image_annotations)],
                "type": "rapid",
            }

            if not self.exam_mode:
                exam_questions[q.id]["normal"] = q.normal
                exam_questions[q.id]["feedback_images"] = feedback_images
                exam_questions[q.id]["answers"] = list(
                    q.get_correct_unstripped_answers()
                )

            if self.include_history and q.history:
                exam_questions[q.id]["history"] = q.history

            if any(image_titles):
                exam_questions[q.id]["image_titles"] = image_titles

            # if feedback_images:
            #    exam_questions[q.id]["feedback_image"] = feedback_images

        exam_json = {
            "eid": "rapid/{}".format(self.id),
            "cached": False,
            "exam_type": "rapid",
            "exam_name": self.name,
            "exam_mode": self.exam_mode,
            "exam_order": exam_order,
            "questions": exam_questions,
        }

        if self.time_limit:
            exam_json["exam_time"] = self.time_limit

        return exam_json

    def get_absolute_url(self):
        return reverse("shorts:exam_overview", kwargs={"pk": self.pk})


class UserAnswer(UserAnswerBase):
    """User answers by candidate"""

    app_name = "shorts"

    question = models.ForeignKey(
        Question, related_name="cid_user_answers", on_delete=models.CASCADE
    )

    answer = models.TextField(blank=True)

    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        blank=True,
        null=True,
        related_name="user_shorts_user_answers",
    )

    # Each user answer is associated with a particular exam
    exam = models.ForeignKey(
        Exam, related_name="cid_user_answers", on_delete=models.CASCADE, null=True
    )

    score = models.IntegerField(
        default=0, blank=True, validators=[MinValueValidator(0), MaxValueValidator(5)]
    )

    class CallStateOptions(models.TextChoices):
        CORRECT = "C", _("Correct Call")
        OVERCALL = "O", _("Overcall")
        UNDERCALL = "U", _("Undercall")
        INCORRECTCALL = "W", _("Incorrect Call")
        PARTIALCALL = "P", _("Partial Call")

    callstate = models.CharField(
        max_length=1, choices=CallStateOptions.choices, blank=True, null=True
    )

    def __str__(self):
        name = self.get_candidate_name()

        try:
            exam = self.exam
        except (Exam.DoesNotExist, KeyError) as e:
            exam = "None"
        return "{}/{}/{}: {}".format(
            exam, name, self.question.get_primary_answer(), self.answer
        )

    def save(self, *args, **kwargs):
        self.clean()
        return super(UserAnswer, self).save(*args, **kwargs)

    def clean(self):
        if self.answer:
            self.answer = self.answer.strip()
        else:
            self.answer = ""

    # def get_compare_string(self):
    #     # strip here should be unneccasry (providing clean is now working)
    #     s = self.answer.lower().strip()
    #     s = s.translate(str.maketrans('', '', string.punctuation))
    #     return s

    def get_answer(self):
        return self.answer

    def get_answer_callstate(self):
        if not self.callstate:
            return ""
        return self.callstate.label

    def get_answer_score(self, cached=True):
        return self.score


class QuestionFinding(FindingBase):
    question = models.ForeignKey(
        Question, related_name="findings", on_delete=models.SET_NULL, null=True
    )
    findings = models.ManyToManyField(Finding, blank=True)
    structures = models.ManyToManyField(Structure, blank=True)
    conditions = models.ManyToManyField(Condition, blank=True)


    def __str__(self) -> str:
        findings = self.findings.all().values_list("name")
        return f"Findings: {self.question.id}/{findings}/{self.description}"