penracourses/atlas/models.py

import json
import os
import pathlib
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
import tagulous
import tagulous.models


import pydicom
import dicognito.anonymizer


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 ugettext_lazy as _
from django.utils.html import mark_safe

from django.core.exceptions import ValidationError
from django.contrib.contenttypes.models import ContentType

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,
    CidUserExam,
    Examination,
    # Condition,
    ExamBase,
    Plane,
    Contrast,
    QuestionNote,
)

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

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

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

import reversion

from django.contrib.contenttypes.fields import GenericRelation

from django.core.validators import MaxValueValidator, MinValueValidator


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


def image_directory_path(instance, filename):
    return "atlas/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)])


class SynMixin(object):
    # class Meta:
    #    abstract = True

    def __str__(self) -> str:
        if self.primary:
            return self.name
        else:
            return f"{self.name} [syn]"

    def get_old_str(self) -> str:
        if self.primary:
            if self.synonym.count() == 0:
                return self.name
            else:
                synonyms = ",".join([i.name for i in self.synonym.all()])
                return f"{self.name} ({synonyms})"
        else:
            return f"{self.name} [syn]"

    def get_synonym(self):
        if self.primary:
            return "[Primary]"
        else:
            s = self.synonym.filter(primary=True).values_list("name", flat=True)
            return ", ".join(s)

    def get_synonym_link(self):
        if self.primary:
            return "[Primary]"
        else:
            syns = self.synonym.filter(primary=True)
            return ", ".join(
                [f"<a href='{s.get_absolute_url()}'>{s.name}</a>" for s in syns]
            )

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


class Finding(SynMixin, models.Model):
    name = models.CharField(max_length=255, unique=True)
    synonym = models.ManyToManyField("self", blank=True)

    primary = models.BooleanField(default="True")

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


class Condition(SynMixin, models.Model):
    name = models.CharField(max_length=255, unique=True)

    synonym = models.ManyToManyField("self", blank=True)
    parent = models.ManyToManyField("self", blank=True)

    primary = models.BooleanField(default="True")

    subspecialty = models.ManyToManyField("subspecialty", blank=True)

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


class Presentation(models.Model):
    name = models.CharField(max_length=255)

    subspecialty = models.ManyToManyField("subspecialty", blank=True)

    def __str__(self) -> str:
        if self.subspecialty:
            return (
                f"{self.name} ({', '.join([str(i) for i in self.subspecialty.all()])})"
            )
        return self.name

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

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


class Subspecialty(models.Model):
    name = models.CharField(max_length=255)

    def __str__(self) -> str:
        return self.name

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

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


class PathologicalProcess(models.Model):
    name = models.CharField(max_length=255)

    def __str__(self) -> str:
        return self.name

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

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


class Differential(models.Model):
    condition = models.ForeignKey(Condition, on_delete=models.CASCADE)
    case = models.ForeignKey(
        "Case", on_delete=models.CASCADE, related_name="differentialcase"
    )
    text = models.TextField(null=True, blank=True)


class Structure(SynMixin, models.Model):
    name = models.CharField(max_length=255, unique=True)

    synonym = models.ManyToManyField("self", blank=True)

    primary = models.BooleanField(default="True")

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


@reversion.register
class Case(models.Model):
    # class SubspecialtyChoices(models.TextChoices):
    #    BREAST = "BR", _("Breast")
    #    CARDIAC = "CA", _("Cardiac")
    #    GASTRO = "GI", _("Gastrointestinal and hepatobiliary")
    #    HEADNECK = "HN", _("Head and Neck")
    #    MSK = "MS", _("Musculoskeletal")
    #    NEURO = "NE", _("Neuroradiology")
    #    OBSGYN = "OG", _("Obstectric and Gynaecological")
    #    PAED = "PA", _("Paediatric")
    #    URO = "UR", _("Uroradiology")
    #    VASC = "VA", _("Vascular")
    #    HAEMONC = "HA", _("Haemotology and Oncology")

    title = models.CharField(max_length=255, help_text="Title of the case", default="")
    # author = models.ForeignKey('auth.User', on_delete=models.CASCADE)
    # image = models.ImageField()
    description = models.TextField(
        blank=True,
        help_text="Description of the case",
    )

    history = models.TextField(
        null=True, blank=True, help_text="A (brief) summary of the relevant history"
    )
    discussion = models.TextField(null=True, blank=True)
    report = models.TextField(
        null=True, blank=True, help_text="A model (sample) report for the case."
    )

    # findings = models.TextField(null=True, blank=True)

    # subspecialty = models.CharField(max_length=2, choices=SubspecialtyChoices.choices)
    subspecialty = models.ManyToManyField(Subspecialty, blank=True)

    condition = models.ManyToManyField(Condition, blank=True)

    presentation = models.ManyToManyField(Presentation, blank=True)

    pathological_process = models.ManyToManyField(PathologicalProcess, blank=True)

    differential = models.ManyToManyField(
        Condition, through=Differential, related_name="casedifferential"
    )

    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 the case",
        related_name="atlas_authored_cases",
    )

    editor = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        related_name="atlas_edited_cases",
    )

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

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

    series = SortedManyToManyField("Series", related_name="case")

    notes = GenericRelation(QuestionNote)

    previous_case = models.OneToOneField(
        "Case",
        on_delete=models.SET_NULL,
        null=True,
        related_name="next_case",
        blank=True,
    )

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

    def get_absolute_url(self):
        return reverse("atlas:case_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 get_link(self):
        return format_html("<a href='{}'>{}</a>", self.get_absolute_url(), str(self))

    def __str__(self):
        return f"{self.pk}:{self.title}"

    def get_series_blocks(self):
        html = ""
        for s in self.series.all():
            html += s.get_block()

        return html

    def get_long_str(self):
        examinations = [series.get_examination() for series in self.series.all()]
        return f"{self.pk}:{self.title} {'/'.join([str(c) for c in self.condition.all()])} [{', '.join(examinations)}]"


class SeriesImage(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(
        "Series", 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


class SeriesFinding(models.Model):

    series = models.ForeignKey(
        "Series", related_name="findings", on_delete=models.SET_NULL, null=True
    )
    description = models.TextField(
        null=True, blank=True, help_text="Findings on the series / stack"
    )
    findings = models.ManyToManyField(Finding, blank=True)
    structures = models.ManyToManyField(Structure, blank=True)
    annotation_json = models.TextField(null=True, blank=True)
    viewport_json = models.TextField(null=True, blank=True)

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

    def annotation_as_string(self) -> str:
        return json.dumps(self.annotation_json)


@reversion.register
class Series(models.Model):
    modality = models.ForeignKey(
        Modality,
        related_name="atlas_series_modality",
        on_delete=models.SET_NULL,
        null=True,
    )
    examination = models.ForeignKey(
        Examination,
        help_text="Name of the examination",
        related_name="atlas_series_examination",
        on_delete=models.SET_NULL,
        null=True,
    )
    plane = models.ForeignKey(
        Plane,
        help_text="Plane of the examination",
        related_name="atlas_series_plane",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    contrast = models.ForeignKey(
        Contrast,
        help_text="MRI / CT contrast",
        related_name="atlas_series_contrast",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    description = models.TextField(
        blank=True,
        help_text="Description of stack",
    )

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

    # findings = models.TextField(null=True, blank=True, help_text="Findings on the series / stack")

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

    def __str__(self):
        return f"{self.pk}:{self.description}"

    def get_full_str(self):
        if self.case:
            case_id = ", ".format([case.pk for case in self.case.all()])
            # case_id = self.case.pk
        else:
            case_id = "None"
        return "{}/{} : {} [{}]".format(
            self.pk, self.get_examination_full(), self.description, case_id
        )

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

    #    def get_string(self):
    #        examination = self.get_examination_full()
    #        return

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

    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("atlas:series_detail", kwargs={"pk": self.pk})

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

        return ",".join(images)

    def get_image_url_array_not_json(self):
        return self.get_image_url_array(json_output=False)

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

        if json_output:
            return json.dumps(images)
        else:
            # This is a mess...
            return format_html('", "'.join(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_thumbnail_link(self):
        return format_html(
            "<a href='{}'>{}<a/>", self.get_absolute_url(), self.get_thumbnail()[0]
        )

    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

    def anonymise_images(self):
        # NOTE: this will not maintain the correct hashed filename
        # but that doesn't matter as we will never get the same anonymisation
        # even with the same dicom...
        anonymizer = dicognito.anonymizer.Anonymizer()

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

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


class CaseCollection(models.Model):
    name = models.CharField(max_length=255, unique=True)

    cases = models.ManyToManyField(Case, through="CaseDetail")

    publish_results = models.BooleanField(
        help_text="If a collection should published", default=False
    )

    active = models.BooleanField(
        help_text="If a collection should be available", default=True
    )

    show_title_pre = models.BooleanField(
        default=False, help_text="Show the title of the cases (pre exam)"
    )
    show_history_pre = models.BooleanField(
        default=False, help_text="Show the history of the cases (pre exam)"
    )
    show_description_pre = models.BooleanField(
        default=False, help_text="Show the description of the cases (pre exam)"
    )
    show_discussion_pre = models.BooleanField(
        default=False, help_text="Show the case discussion (pre exam)"
    )
    show_report_pre = models.BooleanField(
        default=False, help_text="Show the case report (pre exam)"
    )

    show_title_post = models.BooleanField(
        default=False, help_text="Show the title of the cases (post exam)"
    )
    show_history_post = models.BooleanField(
        default=False, help_text="Show the history of the cases (post exam)"
    )
    show_description_post = models.BooleanField(
        default=False, help_text="Show the description of the cases (post exam)"
    )
    show_discussion_post = models.BooleanField(
        default=False, help_text="Show the case discussion (post exam)"
    )
    show_report_post = models.BooleanField(
        default=False, help_text="Show the case report (post exam)"
    )

    author = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        help_text="Author of the collection",
        # related_name="atlas_authored_cases",
    )

    cid_users = GenericRelation("generic.CidUserExam")

    valid_users = models.ManyToManyField(
        CidUser, blank=True, related_name="casecollection_exams"
    )

    archive = models.BooleanField(default=False)
    exam_mode = models.BooleanField(default=False)

    class COLLECTION_TYPE_CHOICES(models.TextChoices):
        REVIEW = (
            "REV",
            _("Review"),
        )
        REPORT = (
            "REP",
            _("Report"),
        )

    collection_type = models.CharField(
        max_length=3,
        choices=COLLECTION_TYPE_CHOICES.choices,
        default=COLLECTION_TYPE_CHOICES.REPORT,
    )

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

    def get_take_url(self):
        return reverse("atlas:collection_take", kwargs={"pk": self.pk})

    def __str__(self) -> str:
        return self.name

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

    def check_cid_user(self, cid, passcode, request=None):
        if request is not None and request.user.is_superuser:
            return True

        if self.valid_users.exists():

            user = self.valid_users.filter(cid=cid).first()

            if not user or user.passcode != passcode:
                return False

        return True

    def get_or_create_cid_user_exam(self, cid_user, start_time=None):
        content_type = ContentType.objects.get_for_model(self)
        c = CidUserExam.objects.filter(
            content_type=content_type, object_id=self.pk, cid_user=cid_user
        ).first()
        if c:
            return c

        if start_time is None:
            start_time = timezone.now()
        new = CidUserExam(
            content_type=content_type,
            object_id=self.pk,
            cid_user=cid_user,
            start_time=start_time,
        )
        new.save()
        return new

    def get_cid_user_exams(self, cid_user=None):
        content_type = ContentType.objects.get_for_model(self)
        if cid_user is None:
            return CidUserExam.objects.filter(
                content_type=content_type,
                object_id=self.pk,
            )
        else:
            return CidUserExam.objects.filter(
                content_type=content_type, object_id=self.pk, cid_user=cid_user
            )


class CaseDetail(models.Model):
    case = models.ForeignKey(Case, on_delete=models.CASCADE)
    collection = models.ForeignKey(CaseCollection, on_delete=models.CASCADE)

    sort_order = models.IntegerField()

    class Meta:
        ordering = ("sort_order",)


class CidReportAnswer(models.Model):
    question = models.ForeignKey(CaseDetail, on_delete=models.CASCADE)

    answer = models.TextField(blank=True)

    cid = models.BigIntegerField(
        blank=False,
        help_text="Candidate ID",
    )

    feedback = models.TextField(blank=True)

    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(auto_now=True)

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

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

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

    def get_answer_score(self):
        return self.score