import json import os import pathlib from typing import Tuple from django.http import Http404, HttpRequest from generic.mixins import AuthorMixin 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 pydicom import pydicom.multival from pydicom.errors import InvalidDicomError 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 django.contrib.contenttypes.models import ContentType from sortedm2m.fields import SortedManyToManyField import string from collections import defaultdict from helpers.images import ( get_image_dicom_hash, get_image_hash, image_as_base64, pretty_print_dicom, ) from generic.models import ( CidUser, CidUserExam, CidUserGroup, ExamOrCollectionGenericBase, Examination, # Condition, ExamBase, Plane, Contrast, QuestionNote, SeriesBase, SeriesImageBase, Modality, UserUserGroup, ) # from generic.models import Examination, Site, Condition, Sign import pydicom.errors import datetime from django.utils import timezone import reversion from django.contrib.contenttypes.fields import GenericRelation from django.core.validators import MaxValueValidator, MinValueValidator image_storage = FileSystemStorage( # Physical file location ROOT location="{0}atlas/".format(settings.MEDIA_ROOT), # Url for file base_url="{0}atlas/".format(settings.MEDIA_URL), ) def image_directory_path(instance, filename): return "atlas/picture/{0}".format(filename) def uncategorised_dicom_directory_path(instance, filename): return "atlas/dicom/{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 SynMixin(object): # class Meta: # abstract = True def __str__(self) -> str: if self.primary: return self.name else: return f"{self.name} [syn]" def get_old_str(self) -> str: if self.primary: if self.synonym.count() == 0: return self.name else: synonyms = ",".join([i.name for i in self.synonym.all()]) return f"{self.name} ({synonyms})" else: return f"{self.name} [syn]" def get_synonym(self): if self.primary: return "[Primary]" else: s = self.synonym.filter(primary=True).values_list("name", flat=True) return ", ".join(s) def get_synonym_link(self): if self.primary: return "[Primary]" else: syns = self.synonym.filter(primary=True) return ", ".join( [f"{s.name}" for s in syns] ) def get_link(self): return format_html("{}", self.get_absolute_url(), self.name) class Finding(SynMixin, models.Model): name = models.CharField(max_length=255, unique=True) synonym = models.ManyToManyField("self", blank=True) primary = models.BooleanField(default="True") def get_absolute_url(self): return reverse("atlas:finding_detail", kwargs={"pk": self.pk}) class Condition(SynMixin, models.Model): name = models.CharField(max_length=255, unique=True) synonym = models.ManyToManyField( "self", blank=True, help_text="Use if a direct synonym for the condition exists, e.g. 'Wegener granulomatosis' and 'Granulomatosis with Polyangitis'.", ) parent = models.ManyToManyField( "self", blank=True, related_name="child", symmetrical=False, through="ConditionRelationship", help_text="Use if the condition could be considered a subset of another, e.g. 'Alzheimer disease' and 'Dementia'.", ) primary = models.BooleanField( default="True", help_text="Sets if this should be the canonical item, all other synonyms will then redirect to this by default.", ) subspecialty = models.ManyToManyField( "subspecialty", blank=True, help_text="Sets the subspecialty(/ies) that this condition falls under.", ) rcr_curriculum_map = models.ManyToManyField( "self", blank=True, limit_choices_to={"rcr_curriculum": True}, help_text="Specifies the related RCR curriculum condition (if it exists).", ) rcr_curriculum = models.BooleanField( default=False, help_text="The condition is from the (non exhaustive) RCR curriculum", ) def get_absolute_url(self): return reverse("atlas:condition_detail", kwargs={"pk": self.pk}) def get_children(self): return self.child.all() def get_descendents(self): """TODO""" def get_parents(self): return self.parent.all() class ConditionRelationship(models.Model): child = models.ForeignKey(Condition, on_delete=models.CASCADE) parent = models.ForeignKey(Condition, on_delete=models.CASCADE, related_name="+") relationship_type = models.CharField(max_length=255, blank=True, null=True) def __str__(self): return f"{self.parent} -> {self.child}" class Presentation(models.Model): name = models.CharField(max_length=255) subspecialty = models.ManyToManyField("subspecialty", blank=True) def __str__(self) -> str: if self.subspecialty: return ( f"{self.name} ({', '.join([str(i) for i in self.subspecialty.all()])})" ) return self.name def get_absolute_url(self): return reverse("atlas:presentation_detail", kwargs={"pk": self.pk}) def get_link(self): return format_html("{}", self.get_absolute_url(), self.name) class Subspecialty(models.Model): name = models.CharField(max_length=255) def __str__(self) -> str: return self.name def get_absolute_url(self): return reverse("atlas:subspecialty_detail", kwargs={"pk": self.pk}) def get_link(self): return format_html("{}", self.get_absolute_url(), self.name) class PathologicalProcess(models.Model): name = models.CharField(max_length=255) def __str__(self) -> str: return self.name def get_absolute_url(self): return reverse("atlas:pathological_process_detail", kwargs={"pk": self.pk}) def get_link(self): return format_html("{}", self.get_absolute_url(), self.name) class Differential(models.Model): condition = models.ForeignKey(Condition, on_delete=models.CASCADE) case = models.ForeignKey( "Case", on_delete=models.CASCADE, related_name="differentialcase" ) text = models.TextField(null=True, blank=True) class Structure(SynMixin, models.Model): name = models.CharField(max_length=255, unique=True) synonym = models.ManyToManyField("self", blank=True) primary = models.BooleanField(default="True") def get_absolute_url(self): return reverse("atlas:structure_detail", kwargs={"pk": self.pk}) @reversion.register class Case(models.Model, AuthorMixin): # class SubspecialtyChoices(models.TextChoices): # BREAST = "BR", _("Breast") # CARDIAC = "CA", _("Cardiac") # GASTRO = "GI", _("Gastrointestinal and hepatobiliary") # HEADNECK = "HN", _("Head and Neck") # MSK = "MS", _("Musculoskeletal") # NEURO = "NE", _("Neuroradiology") # OBSGYN = "OG", _("Obstectric and Gynaecological") # PAED = "PA", _("Paediatric") # URO = "UR", _("Uroradiology") # VASC = "VA", _("Vascular") # HAEMONC = "HA", _("Haemotology and Oncology") class CertaintyChoices(models.IntegerChoices): NONE = 0 POSSIBLE = 1 LIKELY = 2 ALMOST_CERTAIN = 3 CERTAIN = 4 title = models.CharField(max_length=255, help_text="Title of the case", default="") description = models.TextField( blank=True, help_text="Description of the case", ) history = models.TextField( null=True, blank=True, help_text="A (brief) summary of the relevant history" ) discussion = models.TextField(null=True, blank=True) report = models.TextField( null=True, blank=True, help_text="A model (sample) report for the case." ) # findings = models.TextField(null=True, blank=True) # subspecialty = models.CharField(max_length=2, choices=SubspecialtyChoices.choices) subspecialty = models.ManyToManyField(Subspecialty, blank=True) condition = models.ManyToManyField(Condition, blank=True) presentation = models.ManyToManyField(Presentation, blank=True) pathological_process = models.ManyToManyField(PathologicalProcess, blank=True) differential = models.ManyToManyField( Condition, through=Differential, related_name="casedifferential" ) diagnostic_certainty = models.IntegerField( choices=CertaintyChoices.choices, default=CertaintyChoices.NONE ) 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 the case", related_name="atlas_authored_cases", ) editor = models.ManyToManyField( settings.AUTH_USER_MODEL, blank=True, related_name="atlas_edited_cases", ) archive = models.BooleanField( default=False, help_text="Case has been archived, this will remain on the system but will not be shown", ) open_access = models.BooleanField( help_text="If a case should be freely available to browse", default=True ) series = SortedManyToManyField("Series", related_name="case") notes = GenericRelation(QuestionNote) previous_case = models.OneToOneField( "Case", on_delete=models.SET_NULL, null=True, related_name="next_case", blank=True, ) # question_file = models.FileField(upload_to=question_file_directory_path, blank=True, null=True) def get_absolute_url(self): return reverse("atlas:case_detail", kwargs={"pk": self.pk}) def get_link(self): # return f"{self.pk}: {self.title}" return format_html("{}", self.get_absolute_url(), str(self)) def __str__(self): return f"{self.pk}: {self.title}" def get_series_blocks(self): html = "" for s in self.series.all(): html += s.get_block() return html def get_long_str(self): examinations = [series.get_examination() for series in self.series.all()] return f"{self.pk}:{self.title} {'/'.join([str(c) for c in self.condition.all()])} [{', '.join(examinations)}]" def get_case_dicom_json(self, case_title_as_patient_name = True): series_json = [] for series in self.series.all(): series_json.append(series.get_series_dicom_json()) patient_name = "" if case_title_as_patient_name: patient_name = self.title return { "studies": [ { "StudyInstanceUID": "1.3.6.1.4.1.14519.5.2.1.6279.6001.298806137288633453246975630178", "StudyDate": "20000101", "StudyTime": "", "PatientName": patient_name, "PatientID": "LIDC-IDRI-0001", "AccessionNumber": "", "PatientAge": "", "PatientSex": "", "series": series_json, } ] } pass def extract_image_dicom_json_from_ds(ds, url, image_index): to_keep = [ "Columns", "Rows", "InstanceNumber", "SOPClassUID", "PhotometricInterpretation", "BitsAllocated", "BitsStored", "PixelRepresentation", "SamplesPerPixel", "PixelSpacing", "HighBit", "ImageOrientationPatient", "ImagePositionPatient", "FrameOfReferenceUID", "ImageType", "Modality", "SOPInstanceUID", "SeriesInstanceUID", "StudyInstanceUID", "WindowCenter", "WindowWidth", "SeriesDate", ] d = {} for key in to_keep: if key in ds: val = ds[key].value if type(val) == pydicom.multival.MultiValue: d[key] = list(val) else: d[key] = val # Is it worth trying on fake dicom tags?..... if d == {}: d["SOPInstanceUID"] = f"1.2.840.1111.{image_index}" d["SeriesInstanceUID"] = f"1.2.840.1112.1" return {"metadata": d, "url": f"dicomweb:{url}"} class SeriesImage(SeriesImageBase): image = models.FileField(upload_to=image_directory_path, null=True) series = models.ForeignKey( "Series", related_name="images", on_delete=models.CASCADE, null=True ) replaced = models.ForeignKey( "self", on_delete=models.SET_NULL, null=True, blank=True, help_text="Reference to the object that has replaced this one.", ) def get_dicom_data(self): try: with pydicom.dcmread(self.image) as d: return d except InvalidDicomError: return {} # def get_image_dicom_json(self, image_index): # try: # with pydicom.dcmread(self.image) as ds: # return extract_image_dicom_json_from_ds(ds, url=f"{REMOTE_URL}{self.image.url}") # except FileNotFoundError: # return [] # except InvalidDicomError: # return {"url": f"dicomweb:{REMOTE_URL}{self.image.url}", "metadata": { # "SOPInstanceUID" : f"1.2.840.1111.{image_index}" # }} class SeriesFinding(models.Model): series = models.ForeignKey( "Series", related_name="findings", on_delete=models.SET_NULL, null=True ) description = models.TextField( null=True, blank=True, help_text="Findings on the series / stack" ) findings = models.ManyToManyField(Finding, blank=True) structures = models.ManyToManyField(Structure, blank=True) annotation_json = models.TextField(null=True, blank=True) viewport_json = models.TextField(null=True, blank=True) def __str__(self) -> str: findings = self.findings.all().values_list("name") return f"{self.series.id}/{findings}/{self.description}" def annotation_as_string(self) -> str: return json.dumps(self.annotation_json) @reversion.register class Series(SeriesBase): modality = models.ForeignKey( Modality, related_name="atlas_series_modality", on_delete=models.SET_NULL, null=True, blank=True, ) examination = models.ForeignKey( Examination, help_text="Name of the examination", related_name="atlas_series_examination", on_delete=models.SET_NULL, null=True, blank=True, ) plane = models.ForeignKey( Plane, help_text="Plane of the examination", related_name="atlas_series_plane", on_delete=models.SET_NULL, null=True, blank=True, ) contrast = models.ForeignKey( Contrast, help_text="MRI / CT contrast", related_name="atlas_series_contrast", on_delete=models.SET_NULL, null=True, blank=True, ) author = models.ManyToManyField( settings.AUTH_USER_MODEL, blank=True, related_name="series", ) series_instance_uid = models.CharField(max_length=255, blank=True, null=True) # findings = models.TextField(null=True, blank=True, help_text="Findings on the series / stack") def get_full_str(self): if self.case: case_id = ", ".join([str(case) for case in self.case.all()]) # case_id = self.case.pk else: case_id = "None" return "{}/{} : {} [{}]".format( self.pk, self.get_examination_full(), self.description, case_id ) def get_absolute_url(self): return reverse("atlas:series_detail", kwargs={"pk": self.pk}) def get_link(self): return format_html( "{}", self.get_absolute_url(), self.get_full_str() ) def get_series_dicom_json(self): instances = [] series_json = {} to_keep = ["SeriesInstanceUID", "SeriesNumber", "Modality", "SliceThickness"] # TODO: clean up (this is rather convoluted....) image: SeriesImage for series_n, image in enumerate(self.images.filter(removed=False)): ds = image.get_dicom_data() if series_n == 0: for tag in to_keep: if tag in ds: val = ds[tag].value if type(val) == pydicom.multival.MultiValue: series_json[tag] = list(val) else: series_json[tag] = val instances.append( extract_image_dicom_json_from_ds( ds, url=f"{REMOTE_URL}{image.image.url}", image_index=series_n ) ) # else: # instances.append(image.get_image_dicom_json(image_index)) description = f"{self.examination} ({self.plane})" if self.contrast and self.contrast is not None: description = f"{description} / {self.contrast}" series_json["SeriesDescription"] = description series_json["instances"] = instances return series_json def get_ohif_dicom_json(self, case_title_as_patient_name = True): series_json = [] series_json.append(self.get_series_dicom_json()) patient_name = "" if case_title_as_patient_name: patient_name = self.title return { "studies": [ { "StudyInstanceUID": "1.3.6.1.4.1.14519.5.2.1.6279.6001.298806137288633453246975630178", "StudyDate": "20000101", "StudyTime": "", "PatientName": patient_name, "PatientID": "LIDC-IDRI-0001", "AccessionNumber": "", "PatientAge": "", "PatientSex": "", "series": series_json, } ] } class CaseCollection(ExamOrCollectionGenericBase): app_name = "atlas" cases = models.ManyToManyField(Case, through="CaseDetail") show_title_pre = models.BooleanField( default=False, help_text="Show the title of the cases (pre exam)" ) show_history_pre = models.BooleanField( default=False, help_text="Show the history of the cases (pre exam)" ) show_description_pre = models.BooleanField( default=False, help_text="Show the description of the cases (pre exam)" ) show_discussion_pre = models.BooleanField( default=False, help_text="Show the case discussion (pre exam)" ) show_report_pre = models.BooleanField( default=False, help_text="Show the case report (pre exam)" ) show_title_post = models.BooleanField( default=False, help_text="Show the title of the cases (post exam)" ) show_history_post = models.BooleanField( default=False, help_text="Show the history of the cases (post exam)" ) show_description_post = models.BooleanField( default=False, help_text="Show the description of the cases (post exam)" ) show_discussion_post = models.BooleanField( default=False, help_text="Show the case discussion (post exam)" ) show_report_post = models.BooleanField( default=False, help_text="Show the case report (post exam)" ) author = models.ManyToManyField( settings.AUTH_USER_MODEL, blank=True, help_text="Author of the collection", related_name="casecollection_authored_cases", ) valid_cid_users = models.ManyToManyField( CidUser, blank=True, related_name="casecollection_exams" ) valid_user_users = models.ManyToManyField( settings.AUTH_USER_MODEL, blank=True, related_name="user_casecollection_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="casecollection_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="casecollection_user_user_groups", ) exam_mode = models.BooleanField(default=False) # This should override the publish setting self_review = models.BooleanField( default=False, help_text="If true allows users self complete and review cases in a self directed way.", ) class COLLECTION_TYPE_CHOICES(models.TextChoices): REVIEW = ( "REV", _("Review"), ) REPORT = ( "REP", _("Report"), ) VIVA = ( "VIV", _("Viva"), ) collection_type = models.CharField( max_length=3, choices=COLLECTION_TYPE_CHOICES.choices, default=COLLECTION_TYPE_CHOICES.REPORT, ) class VIEWER_MODE_CHOICES(models.TextChoices): BUILT_IN = ( "BUILT_IN", _("Built in viewer"), ) OHIF = ( "OHIF", _("OHIF Viewer"), ) BUILT_IN_WITH_OHIF = ( "BUILT_IN_WITH_OHIF", _("Built in viewer with OHIF Viewer option"), ) viewer_mode = models.CharField( max_length=100, choices=VIEWER_MODE_CHOICES.choices, default=VIEWER_MODE_CHOICES.BUILT_IN, ) def get_absolute_url(self): return reverse("atlas:collection_detail", kwargs={"pk": self.pk}) def get_take_url(self): return reverse("atlas:collection_take_start", kwargs={"pk": self.pk}) def __str__(self) -> str: return self.name def show_built_in_viewer(self) -> bool: return self.viewer_mode in ( self.VIEWER_MODE_CHOICES.BUILT_IN, self.VIEWER_MODE_CHOICES.BUILT_IN_WITH_OHIF, ) def show_ohif_viewer_link(self) -> bool: return self.viewer_mode in ( self.VIEWER_MODE_CHOICES.OHIF, self.VIEWER_MODE_CHOICES.BUILT_IN_WITH_OHIF, ) def review_only(self) -> bool: """Returns True if a users cannot submit responses Returns: bool: _description_ """ if self.collection_type == self.COLLECTION_TYPE_CHOICES.REVIEW: return True return False def in_review_mode(self) -> bool: return self.review_only() or self.publish_results # def can_answer(self) -> bool: # """Returns true if the user is able to answer questions in a collection # (independent of current collection state (e.g. published / not published))""" # if self.collection_type == self.COLLECTION_TYPE_CHOICES.REPORT: # return True # return False def get_case_by_index( self, case_index: int, case_count=False ) -> Case | Tuple[Case, int]: # Seems to be a bug when case_number is 0 or 1 the same (first) object gets returned # forces a list fixes (but is likely inefficient) cases = list( self.cases.all().order_by("casedetail__sort_order").prefetch_related() ) try: case = cases[case_index] except IndexError: # Catch an invalid case_number s = f"Invalid case number: {case_index}" raise Http404(s) if case_count: return case, len(cases) else: return case def get_case_take_url(self, case): cases = list( self.cases.all().order_by("casedetail__sort_order").prefetch_related() ) return reverse( "atlas:collection_case_view_take_user", kwargs={"pk": self.pk, "case_number": cases.index(case)}, ) class CaseDetail(models.Model): case = models.ForeignKey(Case, on_delete=models.CASCADE) collection = models.ForeignKey(CaseCollection, on_delete=models.CASCADE) sort_order = models.IntegerField(default=1000) class Meta: ordering = ("sort_order",) class BaseReportAnswer(models.Model): question = models.ForeignKey(CaseDetail, on_delete=models.CASCADE) answer = models.TextField(blank=True) feedback = models.TextField(blank=True) created = models.DateTimeField(auto_now_add=True) updated = models.DateTimeField(auto_now=True) score = models.IntegerField( blank=True, null=True, validators=[MaxValueValidator(10), MinValueValidator(0)] ) def save(self, *args, **kwargs): self.clean() return super(BaseReportAnswer, self).save(*args, **kwargs) def clean(self): if self.answer: self.answer = self.answer.strip() def get_answer_score(self): return self.score def set_cid_or_user(self, cid=None, user=None): if cid is not None: self.cid = cid elif user is not None: self.user = user else: raise ValueError("No cid or user specified") class Meta: abstract = True class CidReportAnswer(BaseReportAnswer): cid = models.BigIntegerField( blank=False, help_text="Candidate ID", ) class UserReportAnswer(BaseReportAnswer): user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE) class SelfReview(models.Model): """Holds data regarding a users self review / reflection""" user_exam = models.ForeignKey(CidUserExam, on_delete=models.CASCADE) case = models.ForeignKey(Case, on_delete=models.CASCADE) comments = models.TextField(null=True, blank=True) findings = models.IntegerField( null=True, blank=True, validators=[MaxValueValidator(5), MinValueValidator(1)] ) interpretation = models.IntegerField( null=True, blank=True, validators=[MaxValueValidator(5), MinValueValidator(1)] ) review_date = models.DateTimeField(auto_now_add=True) review_update_date = models.DateTimeField(auto_now=True) def get_absolute_url(self): return self.user_exam.exam.get_case_take_url(self.case) def get_display_block(self): html = """
""" edit_url = reverse( "atlas:self_review_update", kwargs={ "user_exam_id": self.user_exam.pk, "case_id": self.case.pk, "pk": self.pk, }, ) # edit_url = "TEST" return format_html( html, comments=self.comments, findings=self.findings, interpretation=self.interpretation, date=self.review_date, id=self.pk, edit_url=edit_url, ) class UncategorisedDicom(models.Model): # We use image to maintain consitency across apps image = models.FileField(upload_to=uncategorised_dicom_directory_path) user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, help_text="The user that uploaded the file", ) series = models.ForeignKey( "Series", related_name="imported_dicoms", on_delete=models.SET_NULL, null=True, blank=True, ) series_instance_uid = models.CharField(max_length=255, blank=True, null=True) image_blake3_hash = models.CharField(max_length=64, null=True, blank=True) created_date = models.DateTimeField(default=timezone.now) basic_dicom_tags = models.JSONField(null=True, blank=True) def check_for_duplicates(self, image_hash): duplicate = None if obj := UncategorisedDicom.objects.filter( image_blake3_hash=image_hash ).first(): # duplicate = obj return obj print(image_hash) if obj := SeriesImage.objects.filter(image_blake3_hash=image_hash).first(): duplicate = obj return duplicate def save(self, *args, **kwargs): """Override save method to add image hash""" if self.image: ds = self.get_dicom_info() self.series_instance_uid = ds["SeriesInstanceUID"].value # if self.check_for_duplicates(): # return DuplicateDicom image_blake3_hash = get_image_dicom_hash( None, dataset=ds, hash_type="blake3", direct_pixel_data=True ) self.image_blake3_hash = image_blake3_hash duplicate = self.check_for_duplicates(image_blake3_hash) self.basic_dicom_tags = self.get_basic_dicom_tags(dataset=ds) if duplicate is not None: if duplicate != self: raise DuplicateDicom # Hack for tests if image_blake3_hash != "1234": super().save(*args, **kwargs) # Call the "real" save() method. def get_dicom_info(self): try: ds = pydicom.read_file(self.image) return ds except pydicom.errors.InvalidDicomError: return {"SeriesInstanceUID": "1234"} def get_basic_dicom_tags(self, dataset=None): try: if dataset is None: dataset = pydicom.read_file(self.image) to_include = ( "StudyDescription", "Modality", "ProtocolName", "SeriesDescription", "SeriesInstanceUID", "BodyPartExamined", "SliceThickness", ) tags = {} for tag in to_include: if tag in dataset: tags[tag] = dataset[tag].value else: tags[tag] = "" return tags except pydicom.errors.InvalidDicomError: return None class DuplicateDicom(Exception): pass