Ross
2108 lines
71 KiB
Python
2108 lines
71 KiB
Python
from functools import lru_cache
|
|
import json
|
|
import os
|
|
import pathlib
|
|
from typing import Tuple
|
|
|
|
from django.http import Http404, HttpRequest
|
|
from generic.mixins import AuthorMixin, QuestionMixin
|
|
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, mark_safe
|
|
|
|
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
|
|
|
|
import string
|
|
from collections import defaultdict
|
|
from helpers.images import (
|
|
get_image_dicom_hash,
|
|
get_image_hash,
|
|
image_as_base64,
|
|
pretty_print_dicom,
|
|
)
|
|
|
|
import mimetypes
|
|
|
|
|
|
from generic.models import (
|
|
CidUser,
|
|
CidUserExam,
|
|
CidUserGroup,
|
|
CimarCase,
|
|
ExamOrCollectionGenericBase,
|
|
ExamUserStatus,
|
|
Examination,
|
|
# Condition,
|
|
ExamBase,
|
|
FindingBase,
|
|
Plane,
|
|
Contrast,
|
|
QuestionNote,
|
|
SeriesBase,
|
|
SeriesImageBase,
|
|
Modality,
|
|
Site,
|
|
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.contrib.contenttypes.models import ContentType
|
|
|
|
from django.core.validators import MaxValueValidator, MinValueValidator
|
|
|
|
from loguru import logger
|
|
|
|
|
|
|
|
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:
|
|
# Use getattr to avoid AttributeError if the model no longer
|
|
# defines `primary` (we are migrating some models away from the
|
|
# old primary/synonym fields).
|
|
if getattr(self, "primary", False):
|
|
return self.name
|
|
else:
|
|
return f"{self.name} [syn]"
|
|
|
|
def get_old_str(self) -> str:
|
|
# Prefer using a model-specific get_synonyms (e.g. Condition.get_synonyms)
|
|
try:
|
|
syns = self.get_synonyms()
|
|
except Exception:
|
|
syns = None
|
|
|
|
# Use getattr for safety when models no longer have `primary`.
|
|
if getattr(self, "primary", False):
|
|
if syns is None:
|
|
synonyms_field = getattr(self, "synonym", None)
|
|
if synonyms_field is None or (hasattr(synonyms_field, "count") and synonyms_field.count() == 0):
|
|
return self.name
|
|
synonyms = ",".join([i.name for i in synonyms_field.all()])
|
|
return f"{self.name} ({synonyms})"
|
|
else:
|
|
if not syns:
|
|
return self.name
|
|
synonyms = ",".join([i.name for i in syns])
|
|
return f"{self.name} ({synonyms})"
|
|
else:
|
|
return f"{self.name} [syn]"
|
|
|
|
def get_synonym(self):
|
|
# Return a readable synonym string. For models that implement
|
|
# get_synonyms (Condition with canonical/aliases), use that.
|
|
try:
|
|
syns = self.get_synonyms()
|
|
except Exception:
|
|
syns = None
|
|
if getattr(self, "primary", False):
|
|
return "[Primary]"
|
|
|
|
if syns is None:
|
|
synonyms_field = getattr(self, "synonym", None)
|
|
if synonyms_field is None:
|
|
return ""
|
|
# If the old M2M exists, prefer those marked primary if such a
|
|
# attribute exists on the related objects.
|
|
try:
|
|
s = synonyms_field.filter(primary=True).values_list("name", flat=True)
|
|
except Exception:
|
|
s = synonyms_field.values_list("name", flat=True)
|
|
return ", ".join(s)
|
|
|
|
# prefer primary names among synonyms if present
|
|
primary_names = [i.name for i in syns if getattr(i, "primary", False)]
|
|
if primary_names:
|
|
return ", ".join(primary_names)
|
|
return ", ".join([i.name for i in syns])
|
|
|
|
def get_synonym_link(self):
|
|
try:
|
|
syns = self.get_synonyms()
|
|
except Exception:
|
|
syns = None
|
|
if getattr(self, "primary", False):
|
|
return "[Primary]"
|
|
|
|
if syns is None:
|
|
# fall back to all M2M synonyms (not just primary) for models
|
|
# that still use the old synonym field (e.g. Finding, Structure)
|
|
synonyms_field = getattr(self, "synonym", None)
|
|
if synonyms_field is None:
|
|
return ""
|
|
syns_qs = synonyms_field.all()
|
|
items = [f"<a href='{s.get_absolute_url()}'>{s.name}</a>" for s in syns_qs]
|
|
return ", ".join(items)
|
|
|
|
# render links for synonyms/canonical group
|
|
items = [f"<a href='{s.get_absolute_url()}'>{s.name}</a>" for s in syns]
|
|
return ", ".join(items)
|
|
|
|
def get_link(self):
|
|
return format_html("<a href='{}'>{}</a>", self.get_absolute_url(), self.name)
|
|
|
|
|
|
class Finding(SynMixin, models.Model):
|
|
name = models.CharField(max_length=255, unique=True)
|
|
# New canonical/alias field: if set, this Finding is an alias and points
|
|
# to the canonical/master Finding. We remove the old M2M `synonym` and
|
|
# `primary` boolean in favour of this single canonical FK.
|
|
canonical = models.ForeignKey(
|
|
"self",
|
|
null=True,
|
|
blank=True,
|
|
on_delete=models.SET_NULL,
|
|
related_name="aliases",
|
|
help_text="If set, this Finding is an alias and points to the canonical Finding.",
|
|
)
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("atlas:finding_detail", kwargs={"pk": self.pk})
|
|
|
|
@property
|
|
def canonical_finding(self):
|
|
return self.canonical if self.canonical else self
|
|
|
|
def get_synonyms(self):
|
|
"""Return other Findings that are aliases/synonyms for this concept.
|
|
|
|
Behaviour mirrors Condition.get_synonyms: if this Finding is an alias
|
|
(has canonical set) return the canonical and other aliases (excluding
|
|
self). If this Finding is canonical, return all aliases (excluding
|
|
self).
|
|
"""
|
|
master = self.canonical_finding
|
|
qs = Finding.objects.filter(models.Q(canonical=master) | models.Q(pk=master.pk)).exclude(pk=self.pk)
|
|
return qs
|
|
|
|
|
|
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'.",
|
|
)
|
|
# New canonical/alias field (Option B): if set, this Condition is an alias
|
|
# and points to the canonical/master Condition.
|
|
canonical = models.ForeignKey(
|
|
"self",
|
|
null=True,
|
|
blank=True,
|
|
on_delete=models.SET_NULL,
|
|
related_name="aliases",
|
|
help_text="If set, this Condition is an alias and points to the canonical Condition.",
|
|
)
|
|
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})
|
|
|
|
@property
|
|
def canonical_condition(self):
|
|
"""Return the canonical/master condition for this Condition (self if none)."""
|
|
return self.canonical if self.canonical else self
|
|
|
|
def get_synonyms(self):
|
|
"""Return other Conditions that are synonyms/aliases for this concept.
|
|
|
|
Behaviour:
|
|
- If this Condition is an alias (canonical set), return the canonical and
|
|
any other aliases (excluding self).
|
|
- If this Condition is canonical, return all aliases (and exclude self).
|
|
"""
|
|
master = self.canonical_condition
|
|
qs = Condition.objects.filter(models.Q(canonical=master) | models.Q(pk=master.pk)).exclude(pk=self.pk)
|
|
return qs
|
|
|
|
def get_children(self):
|
|
return self.child.all()
|
|
|
|
def get_descendents(self):
|
|
"""TODO"""
|
|
|
|
def get_parents(self):
|
|
return self.parent.all()
|
|
|
|
def get_with_subspecialty(self):
|
|
return f"{', '.join([str(i) for i in self.subspecialty.all()])} / {self.name} "
|
|
|
|
|
|
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("<a href='{}'>{}</a>", 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("<a href='{}'>{}</a>", 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("<a href='{}'>{}</a>", 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)
|
|
# Migrate to canonical/aliases model like Condition and Finding
|
|
canonical = models.ForeignKey(
|
|
"self",
|
|
null=True,
|
|
blank=True,
|
|
on_delete=models.SET_NULL,
|
|
related_name="aliases",
|
|
help_text="If set, this Structure is an alias and points to the canonical Structure.",
|
|
)
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("atlas:structure_detail", kwargs={"pk": self.pk})
|
|
|
|
@property
|
|
def canonical_structure(self):
|
|
return self.canonical if self.canonical else self
|
|
|
|
def get_synonyms(self):
|
|
master = self.canonical_structure
|
|
qs = Structure.objects.filter(models.Q(canonical=master) | models.Q(pk=master.pk)).exclude(pk=self.pk)
|
|
return qs
|
|
|
|
|
|
@reversion.register
|
|
class Case(models.Model, AuthorMixin, QuestionMixin):
|
|
# 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,
|
|
help_text="The subspecialties the case is associated with. Multiple subspecialties can be selected.",
|
|
)
|
|
|
|
condition = models.ManyToManyField(
|
|
Condition, blank=True, help_text="The condition(s) the case demonstrates."
|
|
)
|
|
|
|
presentation = models.ManyToManyField(
|
|
Presentation,
|
|
blank=True,
|
|
help_text="The presentation(s) the case is associated with.",
|
|
)
|
|
|
|
pathological_process = models.ManyToManyField(PathologicalProcess, blank=True)
|
|
|
|
differential = models.ManyToManyField(
|
|
Condition, through=Differential, related_name="casedifferential",
|
|
help_text="The differential diagnosis for the case.",
|
|
)
|
|
|
|
diagnostic_certainty = models.IntegerField(
|
|
choices=CertaintyChoices.choices, default=CertaintyChoices.NONE,
|
|
help_text="The diagnostic certainty of the case.",
|
|
)
|
|
|
|
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 = models.ManyToManyField(
|
|
"Series", through="SeriesDetail", related_name="case"
|
|
)
|
|
|
|
notes = GenericRelation(QuestionNote)
|
|
|
|
previous_case = models.OneToOneField(
|
|
"Case",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
related_name="next_case",
|
|
blank=True,
|
|
help_text="If this case is related to another case on the system (e.g. follow up), link them here.",
|
|
)
|
|
|
|
resource = models.ManyToManyField("Resource", through="CaseResource")
|
|
|
|
total_images_series_size = models.BigIntegerField(null=True, blank=True)
|
|
|
|
cimar_uuid = models.CharField(max_length=255, null=True, blank=True)
|
|
|
|
def get_ordered_series_details(self):
|
|
# If the series are already prefetched, use them directly
|
|
if hasattr(self, '_prefetched_objects_cache') and 'series' in self._prefetched_objects_cache:
|
|
# Get the through model and sort by sort_order
|
|
through_model = self.series.through
|
|
# Build a mapping from series pk to through instance
|
|
through_objs = list(through_model.objects.filter(case=self).select_related('series'))
|
|
through_objs.sort(key=lambda x: x.sort_order)
|
|
return through_objs
|
|
# Otherwise, query as before
|
|
return self.series.through.objects.filter(case=self).select_related('series').order_by('sort_order')
|
|
|
|
def get_ordered_series(self):
|
|
"""Returns the series in the case in order of the SeriesDetail sort_order"""
|
|
return [sd.series for sd in self.get_ordered_series_details()]
|
|
|
|
def get_app_name(self):
|
|
return "atlas"
|
|
|
|
# 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("<a href='{}'>{}</a>", self.get_absolute_url(), str(self))
|
|
|
|
def get_cimar_case(self, raise_404_if_not_found=False):
|
|
if raise_404_if_not_found:
|
|
return get_object_or_404(CimarCase, uuid=self.cimar_uuid)
|
|
|
|
try:
|
|
return CimarCase.objects.get(uuid=self.cimar_uuid)
|
|
except CimarCase.DoesNotExist:
|
|
return None
|
|
|
|
def get_cimar_case_details(self):
|
|
cimar_case = self.get_cimar_case(raise_404_if_not_found=True)
|
|
return cimar_case.study_details
|
|
|
|
|
|
# def get_base_template(self):
|
|
|
|
# def get_edit_template_links(self):
|
|
|
|
def __str__(self):
|
|
if self.open_access:
|
|
return f"{self.pk}: {self.title} [OA]"
|
|
else:
|
|
return f"{self.pk}: {self.title}"
|
|
|
|
def get_series_blocks(self):
|
|
html = ""
|
|
for s in self.series.all():
|
|
html += s.get_block()
|
|
|
|
return mark_safe(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, priors=None):
|
|
series_json = []
|
|
series: Series
|
|
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
|
|
|
|
studies_json = {
|
|
"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,
|
|
}
|
|
]
|
|
|
|
}
|
|
|
|
if priors is not None:
|
|
for prior in priors:
|
|
series_json = []
|
|
for series in prior.prior_case.series.all():
|
|
series_json.append(series.get_series_dicom_json())
|
|
studies_json["studies"].append(
|
|
{
|
|
"StudyInstanceUID": "1.3.6.1.4.1.14519.5.2.1.6279.6001.298806137288633453246975630177",
|
|
#"StudyDate": "20000101",
|
|
"StudyTime": "",
|
|
"PatientName": patient_name,
|
|
"PatientID": "LIDC-IDRI-0001",
|
|
"AccessionNumber": "",
|
|
"PatientAge": "",
|
|
"PatientSex": "",
|
|
"series": series_json,
|
|
"StudyDescription": f"Prior: {prior.relation_text}"
|
|
}
|
|
)
|
|
|
|
return studies_json
|
|
|
|
def get_viva_details(self):
|
|
return {
|
|
"title": self.title,
|
|
"description": self.description,
|
|
"history": self.history,
|
|
"discussion": self.discussion,
|
|
"report": self.report,
|
|
}
|
|
|
|
def get_viva_details_json(self):
|
|
return json.dumps(self.get_viva_details())
|
|
|
|
def get_all_prior_cases(self):
|
|
prior_cases = []
|
|
|
|
while self.previous_case is not None:
|
|
prior_cases.append(self.previous_case)
|
|
self = self.previous_case
|
|
|
|
return prior_cases
|
|
|
|
def get_series(self):
|
|
return self.series.all().prefetch_related("images", "examination", "plane")
|
|
|
|
def get_series_modalities(self) -> set:
|
|
"""Returns a list of the modalities of the series in the case"""
|
|
series = self.series.all()
|
|
modalities = [s.modality.short_code for s in series if s.modality is not None]
|
|
|
|
return set(modalities)
|
|
|
|
|
|
def get_total_series_images_size(self, refresh=False):
|
|
if self.total_images_series_size is not None and not refresh:
|
|
return self.total_images_series_size
|
|
|
|
series = self.series.all()
|
|
size = sum(s.get_total_image_size(refresh=refresh) for s in series)
|
|
|
|
return size
|
|
|
|
def check_user_can_edit(self, user) -> bool:
|
|
"""Check if the user can edit the case.
|
|
|
|
Args:
|
|
user (User): The user to check.
|
|
|
|
Returns:
|
|
bool: True if the user can edit the case, False otherwise.
|
|
"""
|
|
if user.is_superuser:
|
|
return True
|
|
|
|
if self.author.filter(id=user.id).exists():
|
|
return True
|
|
|
|
return False
|
|
|
|
def get_series_images_nested(self, as_json: bool =True, exclude_series_ids: None | list[int] = None):
|
|
"""
|
|
Returns a list of lists, where each inner list contains the image URLs
|
|
for a single series in this case, in the order of the series.
|
|
"""
|
|
|
|
series_qs = self.series.all()
|
|
if exclude_series_ids is not None:
|
|
series_qs = self.series.exclude(id__in=exclude_series_ids)
|
|
images = [
|
|
[image.image.url for image in series.images.all()]
|
|
for series in series_qs.order_by('seriesdetail__sort_order').prefetch_related('images')
|
|
]
|
|
|
|
if as_json:
|
|
return json.dumps(images)
|
|
else:
|
|
return images
|
|
|
|
def get_case_named_stacks(self):
|
|
def build_stacks_for(case_obj, prefix=None):
|
|
logger.debug(f"Building stacks for case {case_obj.pk} with prefix '{prefix}'")
|
|
stacks = []
|
|
for series in case_obj.get_ordered_series():
|
|
images = [f"{REMOTE_URL}{img.image.url}" for img in series.get_images()]
|
|
name = f"{prefix}: {series}" if prefix else str(series)
|
|
stacks.append({"name": name, "imageIds": images})
|
|
return stacks
|
|
|
|
results = []
|
|
|
|
logger.debug(f"Building stacks for case {self.pk}")
|
|
|
|
# main case entry
|
|
results.append(
|
|
{
|
|
"caseId": f"",
|
|
"studyId": f"Current Case",
|
|
"stacks": build_stacks_for(self, prefix=None),
|
|
}
|
|
)
|
|
|
|
return json.dumps(results)
|
|
|
|
|
|
class NormalCase(models.Model):
|
|
"""Marks a Case as a 'normal' example used in the Atlas normals section.
|
|
|
|
This version stores a canonical `age_days` integer (patient age in days).
|
|
We keep the examination/modality metadata and authored fields. The
|
|
age fields previously stored (`age_years`, `age_value`, `age_unit`) have
|
|
been removed in favour of the single canonical `age_days` value.
|
|
"""
|
|
case = models.OneToOneField(
|
|
Case,
|
|
on_delete=models.CASCADE,
|
|
related_name="normal_case",
|
|
help_text="The Case that is marked as a normal example.",
|
|
)
|
|
|
|
# Canonical internal representation: patient age in days
|
|
age_days = models.PositiveIntegerField(
|
|
null=True,
|
|
blank=True,
|
|
help_text="Patient age at scan in days (canonical).",
|
|
)
|
|
|
|
# Helpful constants for converting/displaying ages. These are not
|
|
# stored directly on the model but are used by forms/views/templates.
|
|
AGE_UNIT_YEARS = "years"
|
|
AGE_UNIT_MONTHS = "months"
|
|
AGE_UNIT_WEEKS = "weeks"
|
|
AGE_UNIT_DAYS = "days"
|
|
AGE_UNIT_CHOICES = [
|
|
(AGE_UNIT_YEARS, "years"),
|
|
(AGE_UNIT_MONTHS, "months"),
|
|
(AGE_UNIT_WEEKS, "weeks"),
|
|
(AGE_UNIT_DAYS, "days"),
|
|
]
|
|
|
|
examination = models.ForeignKey(
|
|
Examination,
|
|
null=True,
|
|
blank=True,
|
|
on_delete=models.SET_NULL,
|
|
related_name="normal_cases",
|
|
help_text="Primary examination for this normal case (optional).",
|
|
)
|
|
|
|
modality = models.ForeignKey(
|
|
Modality,
|
|
null=True,
|
|
blank=True,
|
|
on_delete=models.SET_NULL,
|
|
related_name="normal_cases",
|
|
help_text="Primary modality for this normal case (optional).",
|
|
)
|
|
|
|
added_by = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
null=True,
|
|
blank=True,
|
|
on_delete=models.SET_NULL,
|
|
related_name="added_normals",
|
|
)
|
|
|
|
added_date = models.DateTimeField(default=timezone.now)
|
|
|
|
notes = models.TextField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-added_date"]
|
|
|
|
def __str__(self):
|
|
return f"Normal: {self.case} ({self.display_age()})"
|
|
|
|
def get_absolute_url(self):
|
|
return self.case.get_absolute_url()
|
|
|
|
@staticmethod
|
|
def to_days(value: int | float, unit: str) -> int | None:
|
|
"""Convert a numeric value + unit into integer days (rounded).
|
|
|
|
Examples:
|
|
to_days(2, 'years') -> ~730
|
|
to_days(6, 'months') -> ~183
|
|
"""
|
|
try:
|
|
if value is None or unit is None:
|
|
return None
|
|
v = float(value)
|
|
if unit == NormalCase.AGE_UNIT_YEARS:
|
|
return int(round(v * 365.25))
|
|
if unit == NormalCase.AGE_UNIT_MONTHS:
|
|
return int(round(v * 30.4375))
|
|
if unit == NormalCase.AGE_UNIT_WEEKS:
|
|
return int(round(v * 7))
|
|
if unit == NormalCase.AGE_UNIT_DAYS:
|
|
return int(round(v))
|
|
except Exception:
|
|
return None
|
|
|
|
@staticmethod
|
|
def days_to_value_unit(days: int | None) -> tuple[int | None, str | None]:
|
|
"""Convert canonical days into a (value, unit) tuple for display.
|
|
|
|
Prefer years for ages >= 365 days, months for ages >= 30 days,
|
|
weeks for >=7 days, else days.
|
|
"""
|
|
if days is None:
|
|
return (None, None)
|
|
try:
|
|
d = int(days)
|
|
if d >= 365:
|
|
yrs = int(round(d / 365.25))
|
|
return (yrs, NormalCase.AGE_UNIT_YEARS)
|
|
if d >= 30:
|
|
months = int(round(d / 30.4375))
|
|
return (months, NormalCase.AGE_UNIT_MONTHS)
|
|
if d >= 7:
|
|
weeks = int(round(d / 7))
|
|
return (weeks, NormalCase.AGE_UNIT_WEEKS)
|
|
return (d, NormalCase.AGE_UNIT_DAYS)
|
|
except Exception:
|
|
return (None, None)
|
|
|
|
def display_age(self) -> str:
|
|
"""Return a human friendly age string derived from age_days."""
|
|
if self.age_days is None:
|
|
return "age unknown"
|
|
val, unit = NormalCase.days_to_value_unit(self.age_days)
|
|
if val is None or unit is None:
|
|
return "age unknown"
|
|
# abbreviate years to 'y' for backwards compatibility where helpful
|
|
if unit == NormalCase.AGE_UNIT_YEARS:
|
|
return f"{val}y"
|
|
return f"{val} {unit}"
|
|
|
|
def save(self, *args, **kwargs):
|
|
# If examination/modality not specified, try to infer from the first series
|
|
try:
|
|
if (self.examination is None or self.modality is None) and self.case.series.exists():
|
|
first_series = self.case.series.first()
|
|
if self.examination is None and getattr(first_series, 'examination', None) is not None:
|
|
self.examination = first_series.examination
|
|
if self.modality is None and getattr(first_series, 'modality', None) is not None:
|
|
self.modality = first_series.modality
|
|
except Exception:
|
|
# Be defensive: if anything goes wrong whilst inferring, continue and save as-is
|
|
pass
|
|
|
|
super().save(*args, **kwargs)
|
|
|
|
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_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(FindingBase):
|
|
series = models.ForeignKey(
|
|
"Series", related_name="findings", on_delete=models.SET_NULL, null=True
|
|
)
|
|
findings = models.ManyToManyField(Finding, blank=True)
|
|
structures = models.ManyToManyField(Structure, blank=True)
|
|
conditions = models.ManyToManyField(Condition, blank=True)
|
|
|
|
def __str__(self) -> str:
|
|
findings = self.findings.all().values_list("name")
|
|
if self.series is None:
|
|
return f"SeriesFinding: {findings}/{self.description} (no series)"
|
|
return f"SeriesFinding: {self.series.id}/{findings}/{self.description}"
|
|
|
|
@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 __str__(self) -> str:
|
|
if self.description:
|
|
return self.description
|
|
else:
|
|
return f"{self.examination} ({self.plane})"
|
|
|
|
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(
|
|
"<a href='{}'>{}</a>", self.get_absolute_url(), self.get_full_str()
|
|
)
|
|
|
|
# @lru_cache(maxsize=128)
|
|
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):
|
|
series_json = []
|
|
series_json.append(self.get_series_dicom_json())
|
|
|
|
patient_name = ""
|
|
|
|
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,
|
|
}
|
|
]
|
|
}
|
|
|
|
def use_date_as_description(self):
|
|
# get first image
|
|
image = self.images.first()
|
|
|
|
with pydicom.dcmread(image.image) as ds:
|
|
date = ds.get("StudyDate", "No date")
|
|
|
|
self.description = f"{date[:4]}-{date[4:6]}-{date[6:]}"
|
|
|
|
self.save()
|
|
|
|
def get_base_template(self):
|
|
return "atlas/base.html"
|
|
|
|
def get_link_headers(self):
|
|
return "atlas/series_headers.html"
|
|
|
|
def get_related_findings(self):
|
|
"""Returns the related findings for the series, including findings from other series in the same case."""
|
|
# Get findings directly related to this series
|
|
findings = list(self.findings.all())
|
|
|
|
# Get findings from other series in the same case
|
|
if hasattr(self, "case") and self.case.exists():
|
|
for case in self.case.all():
|
|
for other_series in case.series.exclude(pk=self.pk):
|
|
findings.extend(other_series.findings.all())
|
|
|
|
# Remove duplicates (by pk)
|
|
unique_findings = {f.pk: f for f in findings}.values()
|
|
return list(unique_findings)
|
|
|
|
|
|
|
|
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",
|
|
)
|
|
|
|
markers = models.ManyToManyField(
|
|
settings.AUTH_USER_MODEL,
|
|
blank=True,
|
|
help_text="Authorised markers for the exam",
|
|
related_name="casecollection_markers",
|
|
)
|
|
|
|
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.",
|
|
)
|
|
|
|
exam_user_status = GenericRelation(ExamUserStatus, related_query_name="atlas_exams")
|
|
cid_user_exam = GenericRelation("generic.CidUserExam", related_query_name="atlas_exams")
|
|
|
|
feedback_once_collection_complete = models.BooleanField(default=True, help_text="If true feedback is only given once the collection is complete. If false feedback is given after each case.")
|
|
|
|
question_time_limit = models.PositiveIntegerField(
|
|
blank=True,
|
|
null=True,
|
|
help_text="Time limit for answering questions in seconds."
|
|
)
|
|
|
|
# Collections that must be completed before this collection can be taken
|
|
prerequisites = models.ManyToManyField(
|
|
"self",
|
|
blank=True,
|
|
symmetrical=False,
|
|
related_name="dependents",
|
|
help_text="Collections that must be completed before this collection can be taken",
|
|
)
|
|
|
|
class COLLECTION_TYPE_CHOICES(models.TextChoices):
|
|
REVIEW = (
|
|
"REV",
|
|
_("Review"),
|
|
)
|
|
REPORT = (
|
|
"REP",
|
|
_("Report"),
|
|
)
|
|
QUESTION = (
|
|
"QUE",
|
|
_("Question"),
|
|
)
|
|
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_app_name(self):
|
|
return "atlas"
|
|
|
|
def get_base_template(self):
|
|
return "atlas/base.html"
|
|
|
|
def get_link_headers(self):
|
|
return f"{self.get_app_name()}/collection_headers.html"
|
|
|
|
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(self) -> bool:
|
|
return self.viewer_mode in (
|
|
self.VIEWER_MODE_CHOICES.OHIF,
|
|
)
|
|
|
|
def show_ohif_viewer_link(self) -> bool:
|
|
return self.viewer_mode in (
|
|
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_cases(self):
|
|
"""Returns the cases in the collection in order of the CaseDetail sort_order"""
|
|
return self.cases.all().order_by("casedetail__sort_order")
|
|
|
|
def get_next_case(self, case):
|
|
cases = list(self.get_cases())
|
|
|
|
try:
|
|
return cases[cases.index(case) + 1]
|
|
except IndexError:
|
|
return None
|
|
|
|
def get_previous_case(self, case):
|
|
cases = list(self.get_cases())
|
|
|
|
new_index = cases.index(case) - 1
|
|
|
|
if new_index < 0:
|
|
return None
|
|
else:
|
|
return cases[new_index]
|
|
|
|
|
|
def get_index_of_case(self, case, case_count=False):
|
|
cases = list(self.get_cases())
|
|
|
|
if case_count:
|
|
return cases.index(case), len(cases)
|
|
else:
|
|
return cases.index(case)
|
|
|
|
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.get_cases().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.get_cases().prefetch_related())
|
|
|
|
return reverse(
|
|
"atlas:collection_case_view_take_user",
|
|
kwargs={"pk": self.pk, "case_number": cases.index(case)},
|
|
)
|
|
|
|
def check_user_can_take(self, cid, passcode, user=None, active_only=True):
|
|
"""
|
|
Extend base check_user_can_take to also require completion of any
|
|
prerequisite collections.
|
|
"""
|
|
# Perform the normal access checks first
|
|
super().check_user_can_take(cid, passcode, user=user, active_only=active_only)
|
|
|
|
# If there are prerequisites, the user (or CID) must have completed them
|
|
if not self.prerequisites.exists():
|
|
return
|
|
|
|
for prereq in self.prerequisites.all():
|
|
# Look up any existing exam record for this user/cid on the prerequisite
|
|
ct = ContentType.objects.get_for_model(prereq)
|
|
exam_record = None
|
|
if cid is not None:
|
|
# Find CidUser by cid
|
|
try:
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
except Exception:
|
|
cid_user = None
|
|
|
|
if cid_user is None:
|
|
exam_record = None
|
|
else:
|
|
exam_record = CidUserExam.objects.filter(
|
|
content_type=ct, object_id=prereq.pk, cid_user=cid_user
|
|
).first()
|
|
else:
|
|
# Check for a normal user_user exam record
|
|
exam_record = CidUserExam.objects.filter(
|
|
content_type=ct, object_id=prereq.pk, user_user=user
|
|
).first()
|
|
|
|
if exam_record is None or not getattr(exam_record, "completed", False):
|
|
# Not allowed to take this collection until prereq completed
|
|
# Raise the PrerequisiteRequired exception including the prereq object
|
|
raise PrerequisiteRequired(
|
|
f"Collection not available until prerequisite '{prereq.name}' is completed.",
|
|
prereq=prereq,
|
|
)
|
|
|
|
def get_ohif_dicom_json(self, case_title_as_patient_name=True):
|
|
studies = []
|
|
for n, case in enumerate(self.cases.all()):
|
|
series_json = []
|
|
for series in case.series.all():
|
|
series_json.append(series.get_series_dicom_json())
|
|
|
|
patient_name = ""
|
|
if case_title_as_patient_name:
|
|
patient_name = case.title
|
|
|
|
studies.append(
|
|
{
|
|
"StudyInstanceUID": f"1.3.6.1.4.1.14519.5.2.1.6279.6001.{n}",
|
|
"StudyDate": "20000101",
|
|
"StudyTime": "",
|
|
"PatientName": patient_name,
|
|
"PatientID": f"ID{n}",
|
|
"AccessionNumber": "",
|
|
"PatientAge": "",
|
|
"PatientSex": "",
|
|
"series": series_json,
|
|
}
|
|
)
|
|
|
|
return {"studies": studies}
|
|
|
|
def add_case(self, case):
|
|
"""Adds a case to the collection and makes sure order is maintained"""
|
|
# We might be better off adding via the case detail model direct
|
|
# CaseDetail.objects.create(case=case, collection=self)
|
|
self.cases.add(case)
|
|
self.order_cases()
|
|
|
|
def order_cases(self):
|
|
"""Modifies the casedetail sort_order to sequentially order the cases"""
|
|
for n, c in enumerate(self.casedetail_set.all().order_by("sort_order")):
|
|
c.sort_order = n
|
|
c.save()
|
|
|
|
|
|
class SeriesDetail(models.Model):
|
|
case = models.ForeignKey(Case, on_delete=models.CASCADE)
|
|
series = models.ForeignKey(Series, on_delete=models.CASCADE)
|
|
|
|
sort_order = models.IntegerField(default=1000)
|
|
|
|
feedback = models.BooleanField(
|
|
default=False,
|
|
help_text="Set to true if the series should only be shown for feedback purposes.",
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ("sort_order",)
|
|
|
|
class CaseDisplaySet(models.Model, AuthorMixin):
|
|
"""
|
|
This is analogous to a SeriesFinding but for a Case (it has
|
|
access to all the series stacks).
|
|
|
|
"""
|
|
case = models.ForeignKey(Case, on_delete=models.CASCADE, related_name="display_sets")
|
|
|
|
name = models.CharField(
|
|
max_length=255,
|
|
help_text="Name of the display set",
|
|
)
|
|
|
|
description = models.TextField(
|
|
blank=True,
|
|
help_text="Description of the display set",
|
|
)
|
|
|
|
viewerstate = models.JSONField(
|
|
null=True,
|
|
blank=True,
|
|
help_text="Viewer state for the display set",
|
|
)
|
|
|
|
annotations = models.JSONField(
|
|
null=True,
|
|
blank=True,
|
|
help_text="Annotations for the display set",
|
|
)
|
|
|
|
findings = models.ManyToManyField(Finding, blank=True)
|
|
structures = models.ManyToManyField(Structure, blank=True)
|
|
conditions = models.ManyToManyField(Condition, blank=True)
|
|
|
|
def viewerstate_string(self):
|
|
return json.dumps(self.viewerstate) if self.viewerstate else "{}"
|
|
|
|
def annotations_string(self):
|
|
return json.dumps(self.annotations) if self.annotations else "{}"
|
|
|
|
|
|
class CaseDetail(models.Model):
|
|
"""A through table that stores the relationship between a case and a collection.
|
|
|
|
This is what user answers are linked to (as there be different questions for the same case in a different collection)
|
|
|
|
The collection questions and answers are also defined here.
|
|
"""
|
|
case = models.ForeignKey(Case, on_delete=models.CASCADE)
|
|
collection = models.ForeignKey(CaseCollection, on_delete=models.CASCADE)
|
|
|
|
question_schema = models.JSONField(null=True, blank=True)
|
|
question_answers = models.JSONField(null=True, blank=True)
|
|
# TODO add feedback for questions
|
|
#question_feedback = models.JSONField(null=True, blank=True)
|
|
|
|
default_viewerstate = models.JSONField(
|
|
null=True,
|
|
blank=True,
|
|
help_text="Default viewer state for the case"
|
|
)
|
|
|
|
sort_order = models.IntegerField(default=1000)
|
|
|
|
redact_history = models.BooleanField(
|
|
default=False,
|
|
help_text="Set to true if the history should be redacted whilst taking the case."
|
|
)
|
|
|
|
override_history = models.TextField(
|
|
null=True, blank=True,
|
|
help_text="This will override the case history for the purpose of the exam/collection."
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ("sort_order",)
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.case} -> {self.collection}"
|
|
|
|
|
|
def get_case_index(self):
|
|
"""Returns the question index (number) in the collection (0 indexed)"""
|
|
case_details = list(self.collection.casedetail_set.all().order_by("sort_order"))
|
|
|
|
return case_details.index(self)
|
|
|
|
def get_question_schema(self):
|
|
return json.dumps(self.question_schema)
|
|
|
|
def get_question_answers(self):
|
|
"""Returns the correct question answers as a json string"""
|
|
return json.dumps(self.question_answers)
|
|
|
|
def get_user_answers(self, user):
|
|
"""Returns the users answers as a json string"""
|
|
try:
|
|
return UserReportAnswer.objects.get(question=self, user=user)
|
|
except UserReportAnswer.DoesNotExist:
|
|
return None
|
|
|
|
def get_cid_answers(self, cid):
|
|
"""Returns the cid users answers as a json string"""
|
|
try:
|
|
return CidReportAnswer.objects.get(question=self, cid=cid)
|
|
except CidReportAnswer.DoesNotExist:
|
|
return None
|
|
|
|
def default_viewerstate_string(self):
|
|
return json.dumps(self.default_viewerstate) if self.default_viewerstate else "{}"
|
|
|
|
def get_history_pre(self):
|
|
if self.collection.show_history_pre:
|
|
if self.redact_history:
|
|
return "[Redacted]"
|
|
if self.override_history and self.override_history != "":
|
|
return self.override_history
|
|
return self.case.history or "No history provided"
|
|
return ""
|
|
|
|
def get_case_series_nested(self, include_priors=True):
|
|
case_series_images = self.case.get_series_images_nested(as_json=False)
|
|
|
|
if include_priors:
|
|
logger.debug(f"Checking for prior cases for case {self.case}")
|
|
logger.debug(f"Found {self.case.prior_case.count()} prior cases for case {self.case}")
|
|
|
|
for prior in self.caseprior_set.all():
|
|
logger.debug(f"Adding prior case {prior.prior_case} to case {self.case}")
|
|
case_series_images.extend(prior.prior_case.get_series_images_nested(as_json=False))
|
|
|
|
return json.dumps(case_series_images)
|
|
|
|
def get_case_named_stacks(self, include_priors=True):
|
|
def build_stacks_for(case_obj, prefix=None):
|
|
stacks = []
|
|
for series in case_obj.get_ordered_series():
|
|
images = [f"{REMOTE_URL}{img.image.url}" for img in series.images.all()]
|
|
name = f"{prefix}: {series}" if prefix else str(series)
|
|
stacks.append({"name": name, "imageIds": images})
|
|
return stacks
|
|
|
|
results = []
|
|
|
|
# main case entry
|
|
results.append(
|
|
{
|
|
"caseId": f"CASE-{self.case.pk}",
|
|
"studyId": f"Current Case",
|
|
"stacks": build_stacks_for(self.case, prefix=None),
|
|
}
|
|
)
|
|
|
|
# include priors as separate entries
|
|
if include_priors:
|
|
for prior in self.caseprior_set.all():
|
|
prior_case = prior.prior_case
|
|
results.append(
|
|
{
|
|
"caseId": f"CASE-{prior_case.pk}",
|
|
"studyId": f"Prior: {prior.relation_text}",
|
|
"stacks": build_stacks_for(prior_case, prefix="Prior"),
|
|
}
|
|
)
|
|
|
|
return json.dumps(results)
|
|
|
|
def render_example_form(self, request=None):
|
|
"""Build and return a rendered HTML snippet for the example answers form.
|
|
|
|
If `request` is provided it will be passed to the template renderer so
|
|
csrf tokens and other context processors work correctly. The returned
|
|
HTML contains the json-editor widget (the `json_answer` field) and a
|
|
submit button for saving example answers.
|
|
"""
|
|
from django.template.loader import render_to_string
|
|
from django.utils.safestring import mark_safe
|
|
from .forms import JsonAnswerForm
|
|
|
|
post_data = {}
|
|
# Ensure we pass the stored answers into the form so the widget is populated
|
|
post_data["json_answer"] = json.dumps(self.question_answers) if self.question_answers is not None else json.dumps({})
|
|
|
|
form = JsonAnswerForm(post_data, question_schema=self.question_schema)
|
|
|
|
html = render_to_string("atlas/_rendered_example_form.html", {"form": form}, request=request)
|
|
return mark_safe(html)
|
|
|
|
class CasePrior(models.Model):
|
|
casedetail = models.ForeignKey(CaseDetail, on_delete=models.CASCADE)
|
|
prior_case = models.ForeignKey(Case, on_delete=models.CASCADE, related_name="prior_case")
|
|
|
|
relation_text = models.CharField(max_length=255, blank=True, help_text="Text to describe the relationship between the cases")
|
|
|
|
class PriorVisibility(models.TextChoices):
|
|
NONE = "NO", _("None")
|
|
ALWAYS = "AL", _("Always")
|
|
REVIEW = "RE", _("Review")
|
|
|
|
|
|
prior_visibility = models.CharField(
|
|
max_length=2,
|
|
choices=PriorVisibility.choices,
|
|
default=PriorVisibility.ALWAYS,
|
|
help_text="Defines when the prior case is shown to the user",
|
|
)
|
|
|
|
class Meta:
|
|
unique_together = ("casedetail", "prior_case")
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.casedetail.case} -> {self.prior_case}"
|
|
|
|
class BaseReportAnswer(models.Model):
|
|
question = models.ForeignKey(CaseDetail, on_delete=models.CASCADE)
|
|
|
|
answer = models.TextField(blank=True)
|
|
|
|
json_answer = models.JSONField(null=True, 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)]
|
|
)
|
|
|
|
completed = models.BooleanField(default=False)
|
|
# Timestamp when the user first loaded the question (started answering)
|
|
started_at = models.DateTimeField(null=True, blank=True)
|
|
# Timestamp when the answer was submitted/saved
|
|
submitted_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
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")
|
|
|
|
def get_correct_json_answers(self):
|
|
logger.debug(f"Getting correct json answers for question {self.question.pk}")
|
|
if self.question.question_schema is None or not "properties" in self.question.question_schema:
|
|
return []
|
|
answers = []
|
|
for name, value in self.question.question_schema["properties"].items():
|
|
logger.debug(f"Processing question property '{name}' with schema {value}")
|
|
|
|
# Safely retrieve the user's answer and the stored correct answer.
|
|
# json_answer or question_answers may be None (or not a dict) if not set,
|
|
# so check before subscripting to avoid TypeError.
|
|
if isinstance(self.json_answer, dict):
|
|
user_answer = self.json_answer.get(name, "")
|
|
else:
|
|
user_answer = ""
|
|
|
|
qa = getattr(self.question, "question_answers", None)
|
|
if isinstance(qa, dict):
|
|
correct_answer = qa.get(name, "")
|
|
else:
|
|
correct_answer = ""
|
|
|
|
match value:
|
|
case {"type": "string", "enum": _}:
|
|
automark = True
|
|
answer_is_correct = user_answer == correct_answer
|
|
case {"type": "string"}:
|
|
if user_answer == correct_answer:
|
|
automark = True
|
|
answer_is_correct = True
|
|
else:
|
|
automark = False
|
|
answer_is_correct = False
|
|
case _:
|
|
automark = False
|
|
answer_is_correct = user_answer == correct_answer
|
|
answers.append((value, user_answer, correct_answer, answer_is_correct, automark))
|
|
|
|
return answers
|
|
|
|
#def get_marked_answers_html(self):
|
|
# """Returns an HTML representation of the users answers with correct answers highlighted"""
|
|
# html = "<div class='json-answers'>"
|
|
# for value, user_answer, correct_answer, answer_is_correct, automark in self.get_correct_json_answers():
|
|
# if automark:
|
|
# if answer_is_correct:
|
|
# html += f"<div class='answer correct'>{user_answer}</div>"
|
|
# else:
|
|
# html += f"<div class='answer incorrect'>Your answer: {user_answer}<br/>Correct answer: {correct_answer}</div>"
|
|
# else:
|
|
# html += f"<div class='answer unmarked'>Your answer: {user_answer} (Not auto-marked)</div>"
|
|
|
|
# html += "</div>"
|
|
|
|
# return format_html(html)
|
|
|
|
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 __str__(self) -> str:
|
|
date = self.review_date
|
|
|
|
if self.review_update_date is not None:
|
|
date = self.review_update_date
|
|
|
|
return f"Self review: {date:%Y-%m-%d} / {self.user_exam.exam} - {self.case}"
|
|
|
|
def get_absolute_url(self):
|
|
return self.user_exam.exam.get_case_take_url(self.case)
|
|
|
|
def get_display_block(self):
|
|
html = """<div class='self-feedback-block'>
|
|
Comments: <span class="comment">{comments}</span><br/>
|
|
|
|
Findings: <span class="findings">{findings}</span><br/>
|
|
|
|
Interpretation: <span class="interpretation">{interpretation}</span><br/>
|
|
<span class="date">Date: {date}</span><br/><a href="{edit_url}">Edit</a>
|
|
</div>
|
|
|
|
"""
|
|
|
|
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=None|str):
|
|
duplicate = None
|
|
|
|
if image_hash is None:
|
|
image_hash = self.image_blake3_hash
|
|
|
|
if obj := UncategorisedDicom.objects.filter(
|
|
image_blake3_hash=image_hash
|
|
).first():
|
|
# duplicate = obj
|
|
return obj
|
|
|
|
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()
|
|
|
|
try:
|
|
self.series_instance_uid = ds["SeriesInstanceUID"].value
|
|
except AttributeError:
|
|
raise InvalidDicomError("Invalid dicom file")
|
|
|
|
# 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(duplicate)
|
|
|
|
|
|
# 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.dcmread(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.dcmread(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
|
|
|
|
def get_series(self):
|
|
return self.series
|
|
|
|
def get_dicom_url(self):
|
|
return f"{REMOTE_URL}{self.image.url}"
|
|
|
|
|
|
|
|
|
|
class DuplicateDicom(Exception):
|
|
def __init__(self, duplicate):
|
|
self.duplicate = duplicate
|
|
|
|
|
|
class Resource(models.Model, AuthorMixin):
|
|
name = models.CharField(max_length=255)
|
|
description = models.TextField(blank=True, null=True)
|
|
url = models.URLField(blank=True, null=True)
|
|
file = models.FileField(upload_to="atlas/resources/", blank=True, null=True)
|
|
author = models.ManyToManyField(
|
|
settings.AUTH_USER_MODEL,
|
|
blank=True,
|
|
help_text="Author of the resource",
|
|
related_name="resources",
|
|
)
|
|
created_date = models.DateTimeField(default=timezone.now)
|
|
modified_date = models.DateTimeField(auto_now=True)
|
|
# Link resources to one or more Sites (from generic.Site)
|
|
sites = models.ManyToManyField(
|
|
'generic.Site',
|
|
blank=True,
|
|
help_text='The site(s) this resource is associated with (if any).',
|
|
related_name='resources',
|
|
)
|
|
|
|
# Categorise resources by subspecialty
|
|
subspecialty = models.ManyToManyField(
|
|
Subspecialty,
|
|
blank=True,
|
|
help_text='Subspecialty or subspecialties this resource relates to.',
|
|
)
|
|
|
|
open_access = models.BooleanField(
|
|
default=True,
|
|
help_text="If true the resource is available to all users, otherwise only in the context of cases."
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
return self.name
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("atlas:resource_detail", kwargs={"pk": self.pk})
|
|
|
|
def get_link(self):
|
|
return format_html("<a href='{}'>{}</a>", self.get_absolute_url(), self.name)
|
|
|
|
def get_display(self) -> str:
|
|
if self.url:
|
|
html = f"<a target='_blank' href='{self.url}'>{self.name}</a>"
|
|
elif self.file:
|
|
filetype, _ = mimetypes.guess_type(self.file.url)
|
|
|
|
match filetype.split("/"):
|
|
case "image", _:
|
|
html = f"<img src='{self.file.url}'>"
|
|
case "video", _:
|
|
html = f"<video controls><source src='{self.file.url}'></video>"
|
|
|
|
case _:
|
|
html = f"<a target='_blank' href='{self.file.url}'>{self.name}</a>"
|
|
else:
|
|
html = self.name
|
|
|
|
meta = ''
|
|
if self.sites.exists():
|
|
meta += ' Sites: ' + ', '.join([s.short_code or s.full_name for s in self.sites.all()])
|
|
if self.subspecialty.exists():
|
|
meta += ' Subspecialty: ' + ', '.join([str(s) for s in self.subspecialty.all()])
|
|
|
|
return format_html(
|
|
"<details class='resource-block'><summary>{}</summary><div>{}</div><div class='text-muted small'>{}</div></details>",
|
|
self.name,
|
|
mark_safe(html),
|
|
meta,
|
|
)
|
|
|
|
|
|
|
|
class CaseResource(models.Model):
|
|
resource = models.ForeignKey(Resource, on_delete=models.CASCADE)
|
|
case = models.ForeignKey(Case, on_delete=models.CASCADE)
|
|
|
|
pre_review = models.BooleanField(
|
|
default=False,
|
|
help_text="Defines if the resource should be shown pre review (i.e. for pre reading before the case is taken)",
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.resource} - {self.case} (Pre: {self.pre_review})"
|
|
|
|
|
|
class QuestionSchema(models.Model, AuthorMixin):
|
|
name = models.CharField(max_length=255)
|
|
description = models.TextField(blank=True, null=True)
|
|
schema = models.JSONField()
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("atlas:question_schema_detail", kwargs={"pk": self.pk})
|
|
|
|
def get_example_form(self):
|
|
from .forms import JsonAnswerForm
|
|
example_form = JsonAnswerForm(
|
|
question_schema=self.schema
|
|
)
|
|
return example_form
|
|
|
|
def get_schema_as_json(self) -> str:
|
|
return json.dumps(self.schema)
|
|
|
|
def __str__(self) -> str:
|
|
return "{}".format(self.name)
|
|
|
|
|
|
class PrerequisiteRequired(Exception):
|
|
"""Raised when a user attempts to access a collection but has not completed a prerequisite.
|
|
|
|
The exception stores an optional `prereq` attribute pointing to the prerequisite
|
|
CaseCollection instance so views can render a helpful page linking to it.
|
|
"""
|
|
|
|
def __init__(self, message=None, *, prereq=None):
|
|
super().__init__(message or "Prerequisite required")
|
|
self.prereq = prereq |