Ross
662 lines
20 KiB
Python
662 lines
20 KiB
Python
import json
|
|
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
|
|
|
|
from anatomy.models import Modality
|
|
|
|
from generic.models import Examination, Condition, Sign, ExamBase, Plane, Contrast
|
|
|
|
# from generic.models import Examination, Site, Condition, Sign
|
|
|
|
from easy_thumbnails.files import get_thumbnailer
|
|
from easy_thumbnails.exceptions import InvalidImageFormatError
|
|
|
|
import pydicom
|
|
import datetime
|
|
|
|
|
|
|
|
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)])
|
|
|
|
|
|
class Long(models.Model):
|
|
# author = models.ForeignKey('auth.User', on_delete=models.CASCADE)
|
|
# image = models.ImageField()
|
|
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_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.description, ", ".join(examinations))
|
|
|
|
def GetExams(self):
|
|
e = self.exams.all().values_list("name", flat=True)
|
|
exams = ", ".join(e)
|
|
return exams
|
|
|
|
def GetImages(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 GetImageUrls(self):
|
|
return ",".join(
|
|
["https://penracourses.org.uk{}".format(i.url) for i in self.GetImages()]
|
|
)
|
|
|
|
def GetUnmarkedAnswerCount(self):
|
|
answers = self.cid_user_answers.all()
|
|
return len([ans for ans in answers if not ans.is_marked()])
|
|
|
|
def GetUnmarkedAnswers(self):
|
|
answers = self.cid_user_answers.all()
|
|
return [ans for ans in answers if not ans.is_marked()]
|
|
|
|
def get_json(self, question_id):
|
|
# self == q?
|
|
q = get_object_or_404(Long, pk=question_id)
|
|
|
|
#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 = datetime.datetime.now()
|
|
with open(path, "w+") as f:
|
|
|
|
# We manually create the json for long questions to reducem memroy usade
|
|
f.write("""{{
|
|
"title": "{}",
|
|
"type": "long",
|
|
"generated": "{}",
|
|
""".format(q.history, timestamp.isoformat())
|
|
)
|
|
|
|
|
|
f.write('"images" : [')
|
|
#images = []
|
|
image_titles = []
|
|
|
|
question_series = q.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...)
|
|
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))
|
|
#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.GetImages()
|
|
|
|
|
|
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']
|
|
|
|
|
|
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",
|
|
)
|
|
|
|
|
|
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.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 = [
|
|
"https://penracourses.org.uk{}".format(i.image.url)
|
|
for i in self.images.all()
|
|
]
|
|
|
|
return ",".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_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
|
|
|
|
# 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 Note(models.Model):
|
|
long = models.ForeignKey(
|
|
Long, related_name="long_notes", on_delete=models.CASCADE, null=True
|
|
)
|
|
# author = models.ForeignKey(User,
|
|
author = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL, related_name="long_notes", on_delete=models.CASCADE
|
|
)
|
|
note = models.TextField()
|
|
created_on = models.DateTimeField(auto_now_add=True)
|
|
|
|
def get_absolute_url(self):
|
|
self.pk = self.long_id
|
|
return reverse("longs:question_detail", kwargs={"pk": self.pk})
|
|
|
|
def __str__(self):
|
|
return "{} [{}] {}".format(self.long, self.author, self.created_on)
|
|
|
|
|
|
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")
|
|
|
|
|
|
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 secondse (75 minutes)",
|
|
default=4500,
|
|
)
|
|
|
|
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):
|
|
questions = self.exam_questions.all()
|
|
|
|
exam_questions = defaultdict(dict)
|
|
|
|
exam_order = []
|
|
|
|
n = 1
|
|
for q in questions:
|
|
exam_order.append(q.id)
|
|
|
|
exam_questions[q.id] = q.json_creation_time.isoformat()
|
|
|
|
continue
|
|
|
|
# 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_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,
|
|
"question_requests": exam_questions,
|
|
}
|
|
|
|
|
|
if self.time_limit:
|
|
exam_json["exam_time"] = self.time_limit
|
|
|
|
return exam_json
|
|
|
|
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
|
|
)
|
|
|
|
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_answer_score(self):
|
|
if self.score == "":
|
|
return ""
|
|
return float(self.score)
|