penracourses/longs/models.py

import json
from pathlib import Path
from rad.settings import REMOTE_URL
from django.db.models.fields.files import ImageField
from django.db.models.fields.related import ForeignKey
from django.db import models
from django.shortcuts import get_object_or_404
from django.utils import timezone


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 django.utils.html import mark_safe

from django.core.exceptions import ValidationError

from sortedm2m.fields import SortedManyToManyField

import string
from collections import defaultdict
from helpers.images import image_as_base64, pretty_print_dicom

from anatomy.models import Modality

from generic.models import (
    CidUser,
    CidUserGroup,
    ExamUserStatus,
    Examination,
    #Condition,
    #Sign,
    ExamBase,
    Plane,
    Contrast,
    QuestionNote,
    UserAnswerBase,
    UserUserGroup,
)

# from generic.models import Examination, Site, Condition, Sign

from easy_thumbnails.files import get_thumbnailer
from easy_thumbnails.exceptions import InvalidImageFormatError

import pydicom
import pydicom.errors
import datetime
from django.utils import timezone

import reversion

from django.contrib.contenttypes.fields import GenericRelation


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


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


def findMiddle(input_list):
    middle = float(len(input_list)) / 2
    if middle % 2 != 0:
        return input_list[int(middle - 0.5)]
    else:
        return input_list[int(middle)]
        return (input_list[int(middle)], input_list[int(middle - 1)])


@reversion.register
class Long(models.Model):
    description = models.TextField(
        blank=True,
        help_text="Description of the case, for admin organisation, will not be visible when taking",
    )

    history = models.TextField(null=True, blank=True)

    feedback = models.TextField(null=True, blank=True)

    # TODO: merge with atlas condition / signs / finding
    #condition = tagulous.models.TagField(
    #    to=Condition,
    #    blank=True,
    #    help_text='Associated condition. Will allow searching / filtering and tips / hints to be displayed. Conditions with spaces must be enclosed in quotes "..."',
    #)

    #sign = tagulous.models.TagField(
    #    to=Sign, blank=True, help_text="Radiological signs in the question"
    #)

    # Model answers
    model_observations = models.TextField(null=True, blank=True)
    model_interpretation = models.TextField(null=True, blank=True)
    model_principle_diagnosis = models.TextField(null=True, blank=True)
    model_differential_diagnosis = models.TextField(null=True, blank=True)
    model_management = models.TextField(null=True, blank=True)

    mark_scheme = models.TextField(null=True, blank=True)

    DEFAULT_SITE_ID = 1
    # site = models.ManyToManyField(
    #    Site,
    #    blank=True,
    #    default=DEFAULT_SITE_ID,
    #    help_text=
    #    "If we know the source of the image")
    verified = models.BooleanField(default=False)
    created_date = models.DateTimeField(default=timezone.now)
    published_date = models.DateTimeField(blank=True, null=True)
    author = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        help_text="Author of question",
        related_name="long_authored_questions",
    )

    scrapped = models.BooleanField(
        default=False, help_text="Question has been scrapped and will not be shown"
    )

    open_access = models.BooleanField(
        help_text="If a question should be freely available to browse", default=True
    )

    recreate_json = models.BooleanField(
        help_text="If the json cache needs updating", default=False
    )

    json_creation_time = models.DateTimeField(blank=True, default=None, null=True)

    question_json_id = models.IntegerField(
        default=1, help_text="Auto incrementing json creation number"
    )

    series = SortedManyToManyField("LongSeries", related_name="long")

    notes = GenericRelation(QuestionNote)

    # question_file = models.FileField(upload_to=question_file_directory_path, blank=True, null=True)

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

    def get_authors(self):
        """Returns a comma seperated text list of authors"""
        authors = ", ".join([i.username for i in self.author.all()])
        return authors

    def get_author_objects(self):
        """Returns a comma seperated text list of authors"""
        authors = [i for i in self.author.all()]
        return authors

    def __str__(self):
        return "{} / {}".format(self.pk, self.description)
        examinations = [series.get_examination() for series in self.series.all()]
        return "{}/{} : {}".format(self.pk, self.description, ", ".join(examinations))

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

    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 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=exam_pk, marker=marker))

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

        unmarked_answers = [ans for ans in answers if not ans.is_marked()]

        if marker is None:
            return unmarked_answers

        # If marker is specified we check for what they have marked
        marker_unmarked = []
        for answer in unmarked_answers:
            if answer.mark.filter(marker=marker).count() < 1:
                marker_unmarked.append(answer)

        return marker_unmarked

    def get_question_model_answers(self):
        """Returns a dict of the question model answers"""
        return {
            "observations": self.model_observations,
            "interpretation": self.model_interpretation,
            "principle diagnosis": self.model_principle_diagnosis,
            "differential_diagnosis": self.model_differential_diagnosis,
            "management": self.model_management,
        }


    def get_question_json(self, based: bool=True, answers: bool=True, feedback: bool=False):
        """Returns a json representation of the question
        
        if based = False return is a dict
        otherwise a url to the json file

        answers arg is only parsed if based = False
        """
        
        question_id = self.pk

        if not based:
            image_titles = []
            images = []
            question_series = self.series.all()
            # Loop through longimage associations
            for i, series in enumerate(question_series):
                # image_array = []
                # for i in series.images.all():
                #    image_array.append(image_as_base64(i.image))
                #    #image_array.append(i.image.url)

                # We are still limited by the size of a series (although this should not be too big or computers will crash...)
                images.append(
                    [
                        "{}/{}".format(settings.REMOTE_URL, i.image.url)
                        for i in series.images.all()
                    ]
                )
                # image_array = [i.image.url for i in series.images.all()]
                # images.append(url)
                image_titles.append(series.get_examination())

            question_json = {
                "title": self.history,
                "images": images,
                "image_titles": image_titles,
                # "feedback_image": [],
                # "annotations": [str(q.image_annotations)],
                "type": "long",
                "cached": False,
            }

            if answers:
                question_json["answers"] = self.get_question_model_answers()

            return question_json

        # exam_order.append(q.id)
        path = "{0}longs/questions/{1}.json".format(settings.MEDIA_ROOT, question_id)
        url = "{0}longs/questions/{1}.json".format(settings.MEDIA_URL, question_id)

        timestamp = timezone.now()

        self.question_json_id += 1

        # Make sure path exists (we should probably just do this once on startup)
        Path(path).parents[0].mkdir(parents=True, exist_ok=True)

        with open(path, "w+") as f:

            print("start writing file")
            # We manually create the json for long questions to reducem memroy usade
            f.write(
                """{{
            "title": "{}",
            "type": "long",
            "generated": "{}",
            "question_json_id": "{}",
            """.format(
                    self.history, timestamp.isoformat(), self.question_json_id
                )
            )

            f.write('"images" : [')
            # images = []
            image_titles = []

            question_series = self.series.all()
            # Loop through longimage associations
            for i, series in enumerate(question_series):
                print("gen series", i)
                # image_array = []
                # for i in series.images.all():
                #    image_array.append(image_as_base64(i.image))
                #    #image_array.append(i.image.url)

                # We are still limited by the size of a series (although this should not be too big or computers will crash...)
                # image_array = [image_as_base64(i.image) for i in series.images.all()]
                ##image_array = [i.image.url for i in series.images.all()]
                # f.write(json.dumps(image_array))
                f.write('["')

                first = True

                # Generate a single image at a time
                for img in series.images.all():
                    if not first:
                        f.write(', "')
                    f.write(image_as_base64(img.image))
                    # for debugging
                    # f.write(img.image.url)
                    f.write('"')
                    first = False
                f.write("]")

                # images.append(url)
                if i != len(question_series) - 1:
                    f.write(",")
                image_titles.append(series.get_examination())
            f.write("],")

            f.write('"image_titles" : {} }}'.format(json.dumps(image_titles)))

        self.json_creation_time = timestamp
        self.save()

        # exam_question = {
        #    "title": q.history,
        #    "images": images,
        #    "image_titles": image_titles,
        #    #"feedback_image": [],
        #    #"annotations": [str(q.image_annotations)],
        #    "type": "long",
        #    "cached": False,
        # }

        return url

    # def GetNonFeedbackQuestionImages(self):
    # return self.get_images()


def test_image_validator(file):
    pass


class LongSeriesImage(models.Model):
    image = models.FileField(upload_to=image_directory_path)
    position = models.IntegerField(default=0)
    upload_filename = models.CharField(max_length=255, blank=True)
    series = models.ForeignKey(
        "LongSeries", related_name="images", on_delete=models.SET_NULL, null=True
    )

    class Meta:
        ordering = ["position"]

    def get_dicom_info(self):
        try:
            info = pretty_print_dicom(pydicom.read_file(self.image))
        except pydicom.errors.InvalidDicomError:
            info = "File is not a dicom."
        return info


@reversion.register
class LongSeries(models.Model):
    modality = models.ForeignKey(
        Modality, related_name="series_modality", on_delete=models.SET_NULL, null=True
    )
    examination = models.ForeignKey(
        Examination,
        help_text="Name of the examination, this appears as the thumbnail title on the test system",
        related_name="series_examination",
        on_delete=models.SET_NULL,
        null=True,
    )
    plane = models.ForeignKey(
        Plane,
        help_text="Plane of the examination",
        related_name="series_plane",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    contrast = models.ForeignKey(
        Contrast,
        help_text="MRI / CT contrast",
        related_name="series_contrast",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    # long = models.ManyToManyField(
    #    "Long",
    #    help_text="The question(s) this series should be associated with",
    #    related_name="series",
    #    blank=True,
    # )
    description = models.TextField(
        blank=True,
        help_text="Description of stack, for admin organisation, will not be visible when taking",
    )

    author = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        related_name="long_series",
    )

    open_access = models.BooleanField(
        help_text="If a question should be freely available to browse", default=True
    )

    def __str__(self):
        if self.long:
            long_id = ", ".format([long.pk for long in self.long.all()])
            # long_id = self.long.pk
        else:
            long_id = "None"
        return "{}/{} : {} [{}]".format(
            self.pk, self.get_examination(), self.description, long_id
        )

    def get_author_objects(self):
        """Returns a comma seperated text list of authors"""
        if self.author:
            return self.author.all()
        else:
            return ["None"]
        if self.long:
            authors = [i for i in self.long.author.all()]
        else:
            authors = []

        return authors

    def get_author_display(self):
        return ", ".join([i.username for i in self.get_author_objects()])

    def get_examination(self):
        """Returns a comma seperated text list of regions"""
        return str(self.examination)

    def get_examination_full(self):
        examination = ""
        plane = ""
        contrast = ""

        if self.examination:
            examination = self.examination
        if self.plane:
            plane = " {}".format(self.plane)
        if self.contrast:
            contrast = " {}".format(self.contrast)
        return "{}{}{}".format(examination, plane, contrast)

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

    def get_image_urls(self, feedback=False):
        images = [f"{REMOTE_URL}{i.image.url}" for i in self.images.all()]

        return ",".join(images)

    def get_image_url_array(self, feedback=False):
        images = [f"{REMOTE_URL}{i.image.url}" for i in self.images.all()]

        return json.dumps(images)

    def get_thumbnail(self):
        images = self.images.all()

        if len(images) < 1:
            return "No images", 0

        img = findMiddle(images).image
        try:
            thumbnailer = get_thumbnailer(img)
            thumbnail = thumbnailer["exam-list"]
        except InvalidImageFormatError:
            return format_html('<span title="{}">Invalid image url<span>', img), len(
                images
            )
        return format_html('<img src="/media/{}" />', thumbnail), len(images)

    def get_block(self):
        examination = self.get_examination_full()
        thumb, image_number = self.get_thumbnail()
        return format_html(
            "<div>{}<br/>{}<br/>Images: {}</div>", examination, thumb, image_number
        )

    def order_by_upload_filename(self):
        images = self.images.all()

        filenames = []
        map = {}

        for i in images:
            filenames.append(i.upload_filename)
            map[i.upload_filename] = i

        filenames = sorted(filenames)

        n = 1
        for f in filenames:
            i = map[f]
            i.position = n
            i.save()
            n = n + 1

    def order_by_dicom(self, field="SliceLocation"):
        images = self.images.all()

        files = []

        for i in images:
            files.append((i, pydicom.dcmread(i.image.path)))

        # print("file count: {}".format(len(files)))

        # skip files with no SliceLocation (eg scout views)
        slices = []
        map = {}
        skipcount = 0
        for i, f in files:
            if hasattr(f, field):
                slices.append(f)
                # map[f.SliceLocation] = i
                map[f[field].value] = i
            else:
                skipcount = skipcount + 1

        print("skipped, no {}: {}".format(field, skipcount))

        # ensure they are in the correct order
        slices = sorted(slices, key=lambda s: s[field].value)

        # print(slices)
        n = 1
        for f in slices:
            i = map[f[field].value]
            i.position = n
            i.save()
            n = n + 1

    def get_total_image_size(self):
        images = self.images.all()

        size = 0

        for i in images:
            size += i.image.size

        return size


# class LongImage(models.Model):
#    long = models.ForeignKey(Long,
#                              related_name="images",
#                              on_delete=models.CASCADE)
#    image = models.ImageField(upload_to=image_directory_path, storage=image_storage)
#
#    feedback_image = models.BooleanField(default=False)
#
#    def image_tag(self):
#        if self.image:
#            return mark_safe(
#                '<img src="{}" class="admin-long-image" /><span class="admin-long-image-info">Click and hold to zoom<span>'
#                .format(self.image.url))
#        else:
#            return ""
#
#    image_tag.short_description = 'Image'


class LongCreationDefault(models.Model):
    # author = models.OneToOneField(User,
    author = models.OneToOneField(
        settings.AUTH_USER_MODEL, related_name="long_default", on_delete=models.CASCADE
    )

    # site = models.ManyToManyField(
    #    Site,
    #    blank=True,
    #    default=1,
    #    help_text="Default site to use when creating a new long")

    def get_absolute_url(self):
        return reverse("longs:long_create")


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

    exam_questions = SortedManyToManyField(Long, related_name="exams", blank="true")

    time_limit = models.IntegerField(
        help_text="Exam time limit (in seconds). Default is 4500 seconds (75 minutes)",
        default=4500,
    )

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

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

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

    valid_user_users = models.ManyToManyField(
        settings.AUTH_USER_MODEL, blank=True, related_name="user_longs_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="longs_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="longs_user_user_groups",
    )

    exam_user_status = GenericRelation(ExamUserStatus)

    def get_exam_question_json(self, question_id):
        q = get_object_or_404(Long, pk=question_id)

        # exam_order.append(q.id)

        # Loop through longimage associations
        images = []
        image_titles = []
        for series in q.series.all():
            # image_array = []
            # for i in series.images.all():
            #    image_array.append(image_as_base64(i.image))
            #    #image_array.append(i.image.url)
            image_array = [image_as_base64(i.image) for i in series.images.all()]
            images.append(image_array)
            image_titles.append(series.get_examination())

        exam_question = {
            "title": q.history,
            "images": images,
            "image_titles": image_titles,
            # "feedback_image": [],
            # "annotations": [str(q.image_annotations)],
            "type": "long",
            "images_json": True,
        }

        return exam_question

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

        exam_questions = defaultdict(dict)

        exam_order = []

        n = 1
        for q in questions:
            exam_order.append(q.id)
            if based:

                # If it is a new question we need to for a question json refresh
                if q.json_creation_time is None:
                    q.get_question_json()
                exam_questions[q.id] = q.question_json_id

            else:
                # TODO: combine with question_json
                images = []
                image_titles = []
                for series in q.series.all():
                    # image_array = []
                    # for i in series.images.all():
                    #    image_array.append(image_as_base64(i.image))
                    #    #image_array.append(i.image.url)
                    image_array = [
                        "{}/{}".format(settings.REMOTE_URL, i.image.url)
                        for i in series.images.all()
                    ]
                    images.append(image_array)
                    image_titles.append(series.get_examination())

                # exam_questions[q.id] = n

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

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

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

        if based:
            exam_json["question_requests"] = 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"""

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

    # Answers
    answer_observations = models.TextField(null=True, blank=True)
    answer_interpretation = models.TextField(null=True, blank=True)
    answer_principle_diagnosis = models.TextField(null=True, blank=True)
    answer_differential_diagnosis = models.TextField(null=True, blank=True)
    answer_management = models.TextField(null=True, blank=True)

    cid = models.BigIntegerField(
        blank=True,
        null=True,
        help_text="Candidate ID (limitied by BigIntegerField size)",
    )

    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.CASCADE,
        blank=True,
        null=True,
        related_name="user_longs_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
    )

    class ScoreOptions(models.TextChoices):
        UNMARKED = "", _("Unmarked")
        FOUR = 4, _("4")
        FOUR_HALF = 4.5, _("4.5")
        FIVE = 5, _("5")
        FIVE_HALF = 5.5, _("5.5")
        SIX = 6, _("6")
        SIX_HALF = 6.5, _("6.5")
        SEVEN = 7, _("7")
        SEVEN_HALF = 7.5, _("7.5")
        EIGHT = 8, _("8")

    score = models.CharField(
        max_length=3,
        choices=ScoreOptions.choices,
        default=ScoreOptions.UNMARKED,
        blank=True,
        help_text="The final score for the candidate",
    )

    # mark = models.ManyToManyField(
    #    "AnswerMarks", related_name="user_answer", blank=True
    # )
    # secondary_mark = models.ForeignKey(
    #    "AnswerMarks", related_name="user_answer", on_delete=models.SET_NULL, null=True, blank=True
    # )



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

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

    def clean(self):
        for ans in (
            self.answer_observations,
            self.answer_interpretation,
            self.answer_principle_diagnosis,
            self.answer_differential_diagnosis,
            self.answer_management,
        ):
            if ans:
                ans = ans.strip()

    def is_marked(self):
        if self.score == "":
            return False
        else:
            return True

    def get_mark_status(self):
        mark_objects = self.mark.all()
        count = mark_objects.count()
        if count == 0:
            return "Unmarked"

        if count == 1:
            return f"Marked by {mark_objects.first().marker} (awaiting second marker)"

        s = ", ".join(f"{str(i.marker)} ({i.score})" for i in mark_objects)
        return f"Marked by {s}"

    def get_markers(self):
        mark_objects = self.mark.all()
        if not mark_objects:
            return "None"
        return "/".join(f"{str(i.marker)}" for i in mark_objects)

    def get_mark_scores(self):
        mark_objects = self.mark.all()
        count = mark_objects.count()
        if count < 2:
            return "hidden"
        return "/".join(f"{str(i.score)}" for i in mark_objects)

    def get_answer_score(self):
        if self.score == "":
            return ""
        return float(self.score)

    def discrepant_answers(self):
        marks = set(self.mark.values_list("score", flat=True))
        if len(marks) > 1:
            return True
        return False

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



@reversion.register
class AnswerMarks(models.Model):
    score = models.CharField(max_length=3, choices=UserAnswer.ScoreOptions.choices)

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

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

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

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