Files
penracourses/longs/models.py
T
Ross 796ec1a072 .
2023-06-12 10:59:36 +01:00

766 lines
23 KiB
Python

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,
ExamBase,
Plane,
Contrast,
QuestionBase,
QuestionNote,
SeriesBase,
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)
@reversion.register
class Long(QuestionBase):
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)
# 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)
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"
)
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")
# 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 get_unanswered_mark_and_text(self) -> tuple[int, tuple[str, str, str, str, str]]:
"""
Long cases are receive a mark between 4 and 8
Therefore unmarked = 4
"""
return (4, ("Not answered",) * 5)
# 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(SeriesBase):
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,
)
author = models.ManyToManyField(
settings.AUTH_USER_MODEL,
blank=True,
related_name="long_series",
)
def __str__(self):
return "{}/{} : {}".format(self.pk, self.get_examination(), self.description)
def get_absolute_url(self):
return reverse("longs:series_detail", kwargs={"pk": self.pk})
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"""
app_name = "longs"
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) -> float | str:
"""Returns the answers score.
If the answer is unmarked the string "unmarked" is returned"""
if self.score == "":
return "unmarked"
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_answer(
self,
) -> tuple[
models.TextField,
models.TextField,
models.TextField,
models.TextField,
models.TextField,
]:
"""Returns a tuple containing the users answers (model fields)
"""
return (
self.answer_observations,
self.answer_interpretation,
self.answer_principle_diagnosis,
self.answer_differential_diagnosis,
self.answer_management,
)
def get_answer_string(self) -> str:
return f"""
Observations:
{self.answer_observations},
Interpretation:
{self.answer_interpretation},
Principle Diagnosis:
{self.answer_principle_diagnosis},
Differential:
{self.answer_differential_diagnosis},
Management:
{self.answer_management},
"""
@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,
)