penracourses/anatomy/models.py

from __future__ import annotations
import json
from django.contrib.contenttypes.fields import GenericRelation
from django.db import models
from django.utils import timezone
from django.contrib.postgres.fields import ArrayField

from django.core.files.storage import FileSystemStorage
from django.conf import settings
from django.utils.html import format_html

from django.urls import reverse

from django.utils.translation import gettext_lazy as _

from sortedm2m.fields import SortedManyToManyField

import string

from generic.models import CidUser, CidUserGroup, ExamUserStatus, Examination, ExamBase, QuestionBase, QuestionNote, UserAnswerBase, UserUserGroup, Modality
from atlas.models import Structure

from collections import defaultdict
from helpers.images import image_as_base64

import reversion

image_storage = FileSystemStorage(
    # Physical file location ROOT
    location=u"{0}/".format(settings.MEDIA_ROOT),
    # Url for file
    base_url=u"{0}/".format(settings.MEDIA_URL),
)


def image_directory_path(instance, filename):
    # file will be uploaded to MEDIA_ROOT/anatomy/picture/<filename>
    return u"picture/anatomy/{0}".format(filename)

def get_answer_compare(s: str) -> str:
    s = s.strip().translate(str.maketrans("", "", string.punctuation)).lower()
    s = " ".join(s.split())
    return s

class BodyPart(models.Model):
    bodypart = models.CharField(max_length=200)

    def __str__(self):
        return self.bodypart


class Region(models.Model):
    region = models.CharField(max_length=200)

    def __str__(self):
        return self.region


## TODO: ??? Move to generic app
#class Modality(models.Model):
#    modality = models.CharField(max_length=200)
#
#    def __str__(self):
#        return self.modality


class QuestionType(models.Model):
    text = models.CharField(max_length=400)

    def __str__(self):
        return self.text


@reversion.register
class AnatomyQuestion(QuestionBase):
    question_type = models.ForeignKey(
        QuestionType, on_delete=models.SET_NULL, null=True, default=1
    )

    image = models.ImageField(
        upload_to=image_directory_path,
        help_text="The image to use for the question. Ideally use use unmarked images and annotate (arrow) them on the test system. If you wish to reuse an image that is already uploaded 'clone' the question that contains it.",
    )

    image_annotations = models.TextField(
        blank=True,
        help_text="Stores a JSON representation of annotations to be applied by cornerstonetools",
    )

    description = models.CharField(
        max_length=400,
        help_text="Short description of the image e.g. 'Sagittal CT Chest, Abdomen & Pelvis', will be displayed as the title.",
    )

    answer_help = models.TextField(default="", blank=True, null=True, help_text="Helpful information for marking")

    answer_suggest_incorrect = ArrayField(models.CharField(max_length=255, blank=True), default=list)

    examination = models.ForeignKey(
        Examination, on_delete=models.SET_NULL, null=True, blank=True
    )
    modality = models.ForeignKey(
        Modality,
        on_delete=models.SET_NULL,
        null=True,
        help_text="Modality of the image",
    )
    region = models.ForeignKey(
        Region,
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
        help_text="Region the image covers",
    )
    body_part = models.ForeignKey(
        BodyPart, on_delete=models.SET_NULL, null=True, blank=True
    )
    structure = models.ForeignKey(
        Structure, on_delete=models.SET_NULL, null=True, blank=True
    )

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

    class Meta:
        permissions = ()

    def __str__(self):
        # Get first answer
        return "{}/{}: {} [{}, {}]".format(
            self.pk,
            self.question_type,
            self.get_primary_answer(),
            self.modality,
            self.structure,
        )

    def get_link(self):
        return format_html("<a href='{}'>{}</a>", self.get_absolute_url(), self)

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

    def get_primary_answer(self):
        if hasattr(self, "prefetched_primary_answer") and self.prefetched_primary_answer:
            ans = self.prefetched_primary_answer[0]
        else:
            ans = self.answers.filter(
                proposed=False, status=Answer.MarkOptions.CORRECT
            ).first()
        # ans = self.answers.first()

        if ans is None:
            return "None yet..."
        else:
            return ans.answer

        if (
            self.answers.filter(
                proposed=False, status=Answer.MarkOptions.CORRECT
            ).count()
            > 0
        ):
            return (
                self.answers.filter(proposed=False, status=Answer.MarkOptions.CORRECT)
                .first()
                .answer
            )
        else:
            return "None yet..."

    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=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):
        """_summary_

        Args:
            exam_pk (_type_, optional): _description_. Defaults to None.

        Returns:
            set: set of unmarked user strings
        """
        if exam_pk is None:
            user_answers = set([i.answer_compare for i in self.cid_user_answers.all()])
        else:
            user_answers = set(
                [
                    i.answer_compare
                    for i in self.cid_user_answers.filter(exam__id=exam_pk)
                ]
            )

        user_answers.discard("")

        unmarked_answers = user_answers - self.get_marked_answers()

        return unmarked_answers

    def get_unmarked_user_answer_count(self, exam_pk=None):
        # if exam_pk is None:
        #    return self.cid_user_answers.filter(
        #        score=Answer.MarkOptions.UNMARKED
        #    ).count()
        # else:
        #    if hasattr(self, "prefetched_unmarked_cid_answers"):
        #        return len(self.prefetched_unmarked_cid_answers)
        #
        #    else:
        #        return self.cid_user_answers.filter(
        #            score=Answer.MarkOptions.UNMARKED, exam__id=exam_pk
        #        ).count()
        return len(self.get_unmarked_user_answers(exam_pk))

    def get_marked_answers(self):
        return set(
            [
                i.answer_compare
                for i in self.answers.filter(proposed=False)
                # for i in self.answers.all()
                if i.status != i.MarkOptions.UNMARKED
            ]
        )

    def get_annotations(self):
        return self.image_annotations

    def get_title(self):
        return "{}".format(self.description)

    def get_correct_unstripped_answers(self):
        return set(
            [
                str(i)
                for i in self.answers.filter(proposed=False)
                # for i in self.answers.all()
                if i.status == i.MarkOptions.CORRECT
            ]
        )

    def get_question_json(self, based: bool=True, answers: bool=True, feedback=False):
        """Returns a json representation of the question"""

        images = []
        annotations = []

        if based:
            images.append(image_as_base64(self.image))
        else:
            images.append("{}/{}".format(settings.REMOTE_URL, self.image.url))

        if self.image_annotations:
            annotations.append(str(self.image_annotations))

        json = {
            "images": images,
            # "feedback_image": [],
            "annotations": annotations,
            "type": "anatomy",
            "title": self.get_title(),
            "question": str(self.question_type),
        }

        if answers:
            json["answers"] = list(self.get_correct_unstripped_answers())

        #if feedback:
        #    json["feedback"] = self.feedback

        return json

    def get_image_url(self):
        return "https://www.penracourses.org.uk{}".format(self.image.url)

    def get_image_url_array(self):
        return json.dumps(["https://www.penracourses.org.uk{}".format(self.image.url)])

    def get_image_annotations(self):
        return json.dumps([self.image_annotations])


@reversion.register
class Answer(models.Model):
    question = models.ForeignKey(
        AnatomyQuestion, related_name="answers", on_delete=models.CASCADE
    )
    answer = models.TextField(max_length=500)
    answer_compare = models.TextField(max_length=500)

    class MarkOptions(models.TextChoices):
        UNMARKED = "", _("Unmarked")
        INCORRECT = "0", _("Incorrect")
        HALF_MARK = "1", _("Half mark")
        CORRECT = "2", _("Correct")

    status = models.CharField(
        max_length=1, choices=MarkOptions.choices, default=MarkOptions.UNMARKED
    )

    proposed = models.BooleanField(default=False)

    def __str__(self):
        return self.answer

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

    def clean(self):
        if self.answer:
            self.answer = self.answer.strip()
            self.answer_compare = get_answer_compare(self.answer)

    # def get_compare_string(self):
    #     s = self.answer.lower().strip()
    #     s = s.translate(str.maketrans('', '', string.punctuation))
    #     return s


# class HalfMarkAnswers(models.Model):
#    question = models.ForeignKey(AnatomyQuestion,
#                                 related_name="half_mark_answers",
#                                 on_delete=models.CASCADE)
#    answer = models.CharField(max_length=500)
#
#    def __str__(self):
#        return self.answer
#
#    def clean(self):
#        if self.answer:
#            self.answer.strip()
#
#
# class IncorrectAnswers(models.Model):
#    question = models.ForeignKey(AnatomyQuestion,
#                                 related_name="incorrect_answers",
#                                 on_delete=models.CASCADE)
#    answer = models.CharField(max_length=500)
#
#    def __str__(self):
#        return self.answer
#
#    def clean(self):
#        if self.answer:
#            self.answer.strip()


@reversion.register
class Exam(ExamBase):
    app_name = "anatomy"

    exam_questions = SortedManyToManyField(
        AnatomyQuestion, related_name="exams", blank="true"
    )
    time_limit = models.IntegerField(
        help_text="Exam time limit (in seconds)", default=5400
    )

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

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

    valid_user_users = models.ManyToManyField(
        settings.AUTH_USER_MODEL, blank=True, related_name="user_anatomy_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="anatomy_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="anatomy_user_user_groups"
    )

    exam_user_status = GenericRelation(ExamUserStatus)

    def get_exam_json(self, based=True):
        questions = self.exam_questions.all()

        exam_questions = defaultdict(dict)

        exam_order = []

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

            if based:
                image = image_as_base64(q.image)
            else:
                image = q.image.url

            annotations = []
            if str(q.image_annotations):
                annotations = [str(q.image_annotations)]

            exam_questions[q.id] = {
                "title": q.get_title(),
                "question": str(q.question_type),
                "images": [image],
                "annotations": annotations,
                "type": "anatomy",
            }

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

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


@reversion.register
class UserAnswer(UserAnswerBase):
    """User answers by candidate"""
    app_name = "anatomy"

    question = models.ForeignKey(
        AnatomyQuestion, related_name="cid_user_answers", on_delete=models.CASCADE
    )
    answer = models.TextField(max_length=500, blank=True)

    answer_compare = models.TextField(max_length=500, blank=True)

    user = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True, related_name="user_anatomy_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.CharField(
        max_length=1,
        choices=Answer.MarkOptions.choices,
        default=Answer.MarkOptions.UNMARKED,
        blank=True,
    )

    def __str__(self):
        try:
            exam = self.exam
        except (Exam.DoesNotExist, KeyError) as e:
            exam = "None"
        return "{}/{}: {}".format(
            exam, self.cid, self.answer
        )

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

    def clean(self):
        if self.answer:
            self.answer_compare = get_answer_compare(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_string(self):
        return self.answer

    def get_answer_score(self, cached=True):
        if cached and self.score:
            if self.score == Answer.MarkOptions.CORRECT:
                mark = 2
            elif self.score == Answer.MarkOptions.HALF_MARK:
                mark = 1
            elif self.score == Answer.MarkOptions.INCORRECT:
                mark = 0

            return mark

        q = self.question
        ans = self.answer_compare

        if ans == "":
            self.score == Answer.MarkOptions.INCORRECT
            self.save()
            return 0

        marked_ans = q.answers.filter(answer_compare__iexact=ans).first()

        mark = "unmarked"
        if marked_ans is not None:
            self.score = marked_ans.status
            self.save()

            if marked_ans.status == Answer.MarkOptions.CORRECT:
                mark = 2
            elif marked_ans.status == Answer.MarkOptions.HALF_MARK:
                mark = 1
            elif marked_ans.status == Answer.MarkOptions.INCORRECT:
                mark = 0

        return mark

        if (
            q.answers.filter(
                answer_compare__iexact=ans, status=Answer.MarkOptions.CORRECT
            ).first()
            is not None
        ):
            mark = 2
        elif (
            q.answers.filter(
                answer_compare__iexact=ans, status=Answer.MarkOptions.HALF_MARK
            ).first()
            is not None
        ):
            mark = 1
        elif q.answers.filter(
            answer_compare__iexact=ans, status=Answer.MarkOptions.INCORRECT
        ).first():
            mark = 0
        else:
            mark = "unmarked"
        return mark