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
from django.utils.html import format_html
from helpers.images import combine_dicom_images_side_by_side
from django.core.files import File
import tempfile


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 __str__(self):
        return "{}: {}".format(
            self.id,
            self.history,
        )

    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 combine_images_side_by_side(self, image1_id, image2_id):
        """
        Combines two DICOM images side by side and saves the new image.
        Marks the original images as feedback images.
        """

        img1 = self.images.get(id=image1_id)
        img2 = self.images.get(id=image2_id)

        dicom_path1 = os.path.join(settings.MEDIA_ROOT, img1.image.name)
        dicom_path2 = os.path.join(settings.MEDIA_ROOT, img2.image.name)

        # Combine images (returns a FileDataset)
        combined_dataset = combine_dicom_images_side_by_side(dicom_path1, dicom_path2)

        # Save new image directly to storage
        combined_image_name = f"combined_{img1.filename}_{img2.filename}.dcm"
        combined_image_path = os.path.join("shorts/picture", combined_image_name)
        full_path = os.path.join(settings.MEDIA_ROOT, combined_image_path)
        combined_dataset.save_as(full_path)
        with open(full_path, "rb") as f:
            new_image = QuestionImage(
            question=self,
            image=File(f, name=combined_image_path),
            feedback_image=False,
            is_dicom=True,
            )
            new_image.save()

        # Mark originals as feedback images
        img1.feedback_image = True
        img1.save()
        img2.feedback_image = True
        img2.save()

        return new_image

    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()

    def get_exams(self):
        e = self.exams.all().values_list("name", flat=True)
        exams = ", ".join(e)
        return exams

    def get_unmarked_user_answer_string(self, exam_pk: int = None):
        unmarked_answers = self.get_unmarked_user_answers(exam_pk)

        if not unmarked_answers:
            return "No answers to mark"
        return format_html(
            "<span class='warn'>{} answer unmarked:</span> {}".format(
                len(unmarked_answers), ", ".join(unmarked_answers)
            )
        )

    def get_unmarked_user_answers(self, exam_pk: int | None = None, marker=None):
        if exam_pk is None:
            user_answer_queryset = self.cid_user_answers.all()
        else:
            user_answer_queryset = self.cid_user_answers.filter(exam__id=exam_pk)

        unmarked_answers = user_answer_queryset.filter(score__isnull=True)


        return unmarked_answers


    def get_unmarked_user_answer_count(self, exam_pk=None, marker=None):
        if exam_pk is None:
            return self.cid_user_answers.all().count()

        return len(self.get_unmarked_user_answers(exam_pk, marker=marker))

    def get_user_answers(self, exam_pk=None, include_normal=True):
        if exam_pk is None:
            queryset = self.cid_user_answers.all()
        else:
            queryset = self.cid_user_answers.filter(exam__id=exam_pk)

        if not include_normal:
            queryset = queryset.filter(normal=False)

        user_answers = set([i.answer for i in queryset])

        return user_answers

    def get_examination_string(self):
        """Returns a string of the examinations associated with the question."""
        examinations = self.examination.all().values_list("examination", flat=True)
        return ", ".join(examinations)
        

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",
    )

    double_mark = models.BooleanField(
        default=False, help_text="Defines if an exam is expected to be double marked"
    )

    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):
        questions = self.get_questions()

        exam_questions = defaultdict(dict)

        exam_order = []

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

            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(q.get_examination_string())

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

            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": "short/{}".format(self.id),
            "cached": False,
            "exam_type": "short",
            "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

        print(exam_json)
        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
    )

    # If score is null then the answer is unmarked
    score = models.IntegerField(
        default=0, blank=True, validators=[MinValueValidator(0), MaxValueValidator(5)], null=True
    )

    candidate_feedback = models.TextField(
        null=True, blank=True, help_text="Feedback for the candidate"
    )

    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

    def get_answer_string(self):
        return self.answer

    def is_marked(self):
        if self.score is None:
            return False
        else:
            return True


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}"

class AnswerMarks(models.Model):
    # Same as in UserAnswer except null is not allowed
    score = models.IntegerField(
        default=0, blank=True, validators=[MinValueValidator(0), MaxValueValidator(5)]
    )

    mark_reason = models.TextField(
        null=True,
        blank=True,
        help_text="Reason for the given mark - not visible to candidates",
    )

    marker = models.ForeignKey(
        settings.AUTH_USER_MODEL, related_name="shorts_answers", on_delete=models.CASCADE
    )

    user_answer = models.ForeignKey(
        UserAnswer,
        related_name="mark",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )