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 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, based=True): # self == q? q = get_object_or_404(Long, pk=question_id) if not based: image_titles = [] images = [] 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...) 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": q.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 = 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('Invalid image url', img), len(images) return format_html('', thumbnail), len(images) def get_block(self): examination = self.get_examination_full() thumb, image_number = self.get_thumbnail() return format_html( "
{}
{}
Images: {}
", 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( # 'Click and hold to zoom' # .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, based=False): questions = self.exam_questions.all() exam_questions = defaultdict(dict) exam_order = [] n = 1 for q in questions: exam_order.append(q.id) if based: exam_questions[q.id] = q.json_creation_time.isoformat() 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, #"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 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)