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 import string from collections import defaultdict from helpers.images import image_as_base64, pretty_print_dicom from generic.models import ( CidUser, CidUserGroup, ExamCollection, ExamUserStatus, Examination, ExamBase, Plane, Contrast, QuestionBase, QuestionNote, SeriesBase, SeriesImageBase, UserAnswerBase, UserUserGroup, Modality ) # 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 = models.ManyToManyField("LongSeries", through="SeriesDetail", related_name="long") # question_file = models.FileField(upload_to=question_file_directory_path, blank=True, null=True) def get_app_name(self): return "longs" def get_absolute_url(self): return reverse("longs:question_detail", kwargs={"pk": self.pk}) 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, str]: """ Long cases are receive a mark between 4 and 8 This has changed to a 3 if unanswered Therefore unmarked = 3 """ return (3, ("Not answered",) * 5) # def GetNonFeedbackQuestionImages(self): # return self.get_images() def test_image_validator(file): pass class LongSeriesImage(SeriesImageBase): image = models.FileField(upload_to=image_directory_path) series = models.ForeignKey( "LongSeries", related_name="images", on_delete=models.CASCADE, null=True ) class SeriesDetail(models.Model): sort_order = models.IntegerField(default=1000) long = models.ForeignKey(Long, on_delete=models.CASCADE) longseries = models.ForeignKey("LongSeries", on_delete=models.CASCADE) class Meta: ordering = ("sort_order",) @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}) def get_link(self): return format_html("{}", self.get_absolute_url(), self) 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 ExamQuestionDetail(models.Model): sort_order = models.IntegerField(default=1000) exam = models.ForeignKey("Exam", on_delete=models.CASCADE) question = models.ForeignKey(Long, on_delete=models.CASCADE) class Meta: ordering = ("sort_order",) @reversion.register class Exam(ExamBase): app_name = "longs" exam_questions = models.ManyToManyField(Long, through=ExamQuestionDetail, related_name="exams") 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", ) markers = models.ManyToManyField( settings.AUTH_USER_MODEL, blank=True, null=True, help_text="Authorised markers for the exam", related_name="long_exam_markers", ) double_mark = models.BooleanField( default=False, 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, related_query_name="longs_exams") cid_user_exam = GenericRelation("generic.CidUserExam", related_query_name="longs_exams") examcollection = models.ForeignKey(ExamCollection, blank=True, null=True, on_delete=models.SET_NULL, related_name="longs_exams") def get_app_name(self): return "longs" 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.get_questions() 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") #UNANSWERED = "3", _("Unanswered") 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", ) candidate_feedback = models.TextField( null=True, blank=True, help_text="Feedback 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", ) 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, )