Files
penracourses/longs/models.py
T
2022-05-20 20:26:02 +01:00

843 lines
26 KiB
Python

import json
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
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 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, Examination, Condition, Sign, 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
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=u"{0}longs/".format(settings.MEDIA_ROOT),
# Url for file
base_url=u"{0}longs/".format(settings.MEDIA_URL),
)
def image_directory_path(instance, filename):
return u"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)
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):
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_json(self, based=True, feedback=False):
# self == q?
#q = get_object_or_404(Long, pk=question_id)
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,
}
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
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"
)
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": True,
"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 CidUserAnswer(models.Model):
"""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)",
)
# 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
#)
created = models.DateTimeField(auto_now_add=True)
updated = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['cid']
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(CidUserAnswer, 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=CidUserAnswer.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(
CidUserAnswer, related_name="mark", on_delete=models.SET_NULL, null=True, blank=True
)