Ross
2100 lines
66 KiB
Python
2100 lines
66 KiB
Python
from collections import defaultdict
|
|
import datetime
|
|
import json
|
|
import os
|
|
from typing import Optional, Self, Tuple
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.db import models
|
|
from django.db.models import UniqueConstraint
|
|
from django.db.models.functions import Lower
|
|
from django.forms import ValidationError
|
|
from django.http import Http404, HttpRequest
|
|
from django.shortcuts import render
|
|
from django.utils import timezone
|
|
from django.core.validators import MaxValueValidator, MinValueValidator
|
|
from smtplib import SMTPException
|
|
|
|
from django.urls import reverse
|
|
from django.views.generic.edit import CreateView
|
|
from reversion.views import RevisionMixin
|
|
|
|
from django.conf import settings
|
|
|
|
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from django.core.mail import send_mail
|
|
from django.core.mail import EmailMessage
|
|
|
|
from django.contrib.auth.models import User
|
|
|
|
from django.db.models.signals import post_save
|
|
from django.dispatch import receiver
|
|
from generic.mixins import AuthorMixin, QuestionMixin
|
|
from helpers.images import get_image_hash, pretty_print_dicom
|
|
from helpers.cimar import CimarAPI, STORAGE_API_URL
|
|
from rad.settings import REMOTE_URL, CIMAR_USERNAME, CIMAR_PASSWORD
|
|
from django.utils.html import format_html
|
|
|
|
from easy_thumbnails.files import get_thumbnailer
|
|
from easy_thumbnails.exceptions import InvalidImageFormatError
|
|
|
|
|
|
import pydicom
|
|
import pydicom.errors
|
|
import dicognito.anonymizer
|
|
from django.contrib.auth.models import User
|
|
from django.forms.models import model_to_dict
|
|
from django.forms.utils import from_current_timezone, to_current_timezone
|
|
|
|
from loguru import logger
|
|
|
|
from pygments import highlight
|
|
from pygments.lexers import JsonLexer
|
|
from pygments.formatters import HtmlFormatter
|
|
|
|
from django.contrib import admin
|
|
from django.utils.safestring import mark_safe
|
|
from django.template.loader import render_to_string
|
|
import requests
|
|
from django.core.files.base import ContentFile
|
|
|
|
|
|
USER_EXAM_TYPES = (
|
|
("Physics", "user_physics_exams"),
|
|
("Rapids", "user_rapid_exams"),
|
|
("Shorts", "user_shorts_exams"),
|
|
("SBAs", "user_sba_exams"),
|
|
("Anatomy", "user_anatomy_exams"),
|
|
("Longs", "user_longs_exams"),
|
|
# ("CaseCollection", "user_casecollection_exams"),
|
|
)
|
|
|
|
|
|
def get_pretty_json(data):
|
|
"""Function to display pretty version of our data"""
|
|
if not data:
|
|
return "No data"
|
|
|
|
# Convert the data to sorted, indented JSON
|
|
response = json.dumps(data, sort_keys=True, indent=2)
|
|
|
|
# Get the Pygments formatter
|
|
formatter = HtmlFormatter(style='colorful')
|
|
|
|
# Highlight the data
|
|
response = highlight(response, JsonLexer(), formatter)
|
|
|
|
# Get the stylesheet
|
|
style = "<style>" + formatter.get_style_defs() + "</style><br>"
|
|
|
|
# Safe the output
|
|
return mark_safe(style + response)
|
|
|
|
def findMiddle(input_list):
|
|
"""Returns the middle element of a list"""
|
|
middle = float(len(input_list)) / 2
|
|
if middle % 2 != 0:
|
|
return input_list[int(middle - 0.5)]
|
|
else:
|
|
return input_list[int(middle)]
|
|
|
|
|
|
class Modality(models.Model):
|
|
modality = models.CharField(max_length=200, unique=True)
|
|
short_code = models.CharField(max_length=5, unique=True, null=True)
|
|
|
|
def __str__(self):
|
|
return f"{self.modality} ({self.short_code})"
|
|
|
|
|
|
class Plane(models.Model):
|
|
plane = models.CharField(max_length=200, unique=True)
|
|
|
|
def __str__(self):
|
|
return self.plane
|
|
|
|
class Meta:
|
|
ordering = ("plane",)
|
|
|
|
|
|
class Contrast(models.Model):
|
|
contrast = models.CharField(max_length=200, unique=True)
|
|
|
|
def __str__(self):
|
|
return self.contrast
|
|
|
|
class Meta:
|
|
ordering = ("contrast",)
|
|
|
|
|
|
class Examination(models.Model):
|
|
"""
|
|
|
|
Currently used for
|
|
- Atlas series
|
|
- Longs series
|
|
- Anatomy questions
|
|
|
|
Not used for rapids - yet??
|
|
"""
|
|
|
|
examination = models.CharField(max_length=200, unique=True)
|
|
|
|
modality = models.ForeignKey(Modality, on_delete=models.SET_NULL, null=True)
|
|
|
|
class Meta:
|
|
ordering = ("examination",)
|
|
|
|
def __str__(self):
|
|
return self.examination
|
|
|
|
def merge_into(self, examination_to_merge_into: Self, delete: bool = True) -> bool:
|
|
"""Merges the current examination into another
|
|
|
|
To do so it replaces all associations (many to many and foreign key relations)
|
|
|
|
Returns True if successful
|
|
"""
|
|
# Get all atlas series with the examination
|
|
atlas_series = self.atlas_series_examination.all()
|
|
longs_series = self.series_examination.all()
|
|
anatomy_questions = self.anatomyquestion_set.all()
|
|
|
|
# Reassign them to the new object
|
|
for series in atlas_series:
|
|
series.examination = examination_to_merge_into
|
|
series.save()
|
|
|
|
for series in longs_series:
|
|
series.examination = examination_to_merge_into
|
|
series.save()
|
|
|
|
for question in anatomy_questions:
|
|
question.examination = examination_to_merge_into
|
|
question.save()
|
|
|
|
# quick check to make sure nothing has gone wrong and we have hanging
|
|
# associations
|
|
querysets = (
|
|
self.atlas_series_examination.all(),
|
|
self.series_examination.all(),
|
|
self.anatomyquestion_set.all(),
|
|
)
|
|
|
|
for q in querysets:
|
|
if q.count() > 0:
|
|
return False
|
|
|
|
if delete:
|
|
self.delete()
|
|
return True
|
|
|
|
|
|
class Site(models.Model):
|
|
"""Model to hold site details"""
|
|
|
|
full_name = models.CharField(
|
|
max_length=255, blank=True, help_text="Name of the site"
|
|
)
|
|
short_code = models.CharField(
|
|
max_length=255, blank=True, help_text="Shortcode/name of the site"
|
|
)
|
|
|
|
def __str__(self):
|
|
return self.short_code
|
|
|
|
|
|
class QuestionBase(models.Model, AuthorMixin, QuestionMixin):
|
|
authors_only = models.BooleanField(
|
|
help_text="If true only question authors will be able to view.",
|
|
default=False,
|
|
)
|
|
|
|
created_date = models.DateTimeField(default=timezone.now)
|
|
|
|
open_access = models.BooleanField(
|
|
help_text="If a question should be freely available to browse", default=True
|
|
)
|
|
|
|
feedback = models.TextField(null=True, blank=True, help_text="Question Feedback")
|
|
|
|
notes = GenericRelation("generic.QuestionNote")
|
|
|
|
reviews = GenericRelation("generic.QuestionReview")
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def get_unanswered_mark_and_text(self) -> tuple[int, str]:
|
|
"""
|
|
Override in models if needed
|
|
"""
|
|
return (0, "Not answered")
|
|
|
|
def get_primary_answer(self):
|
|
"""If this makes sense in the question/answer context override"""
|
|
return None
|
|
|
|
def can_edit(self, user: User) -> bool:
|
|
"""Returns True if the user can edit the question"""
|
|
if user.is_superuser:
|
|
return True
|
|
|
|
if user in self.get_author_objects():
|
|
return True
|
|
|
|
return False
|
|
|
|
def get_thumbnail(self, recreate=False, fail_loudly=False):
|
|
images = self.images.all()
|
|
|
|
if len(images) < 1:
|
|
return "No images", 0
|
|
|
|
img = findMiddle(images).image
|
|
try:
|
|
thumbnailer = get_thumbnailer(img)
|
|
if recreate:
|
|
thumbnailer.delete_thumbnails()
|
|
thumbnail = thumbnailer["exam-list"]
|
|
|
|
except InvalidImageFormatError as e:
|
|
if fail_loudly:
|
|
raise e
|
|
return format_html(
|
|
'<img title="{}" src="/static/not-found-image.jpg" />', img
|
|
), len(images)
|
|
return format_html('<span style="border: 1px solid;"><img src="/media/{}" /></span>', thumbnail), len(images)
|
|
|
|
|
|
|
|
|
|
class SeriesImageBase(models.Model):
|
|
"""
|
|
Series Images may be modified (usually downsampled to save space), in this case the
|
|
(larger) original file will be deleted but the model object will be persisted so that
|
|
the duplicate check continues to function.
|
|
"""
|
|
|
|
position = models.IntegerField(default=0)
|
|
upload_filename = models.CharField(max_length=255, blank=True)
|
|
series = models.ForeignKey(
|
|
"Series", related_name="images", on_delete=models.SET_NULL, null=True
|
|
)
|
|
dicom_tags_ohif = models.JSONField(
|
|
help_text="Holds the dicom tags required for the OHIF viewer",
|
|
blank=True,
|
|
null=True,
|
|
)
|
|
|
|
# We store two hashes of the dicom data to uniquely identify the file
|
|
# and allow duplicate detection
|
|
|
|
# The first is a md5 hash of the pixel array data
|
|
# this is used because I can't get the direct pixel data to match from cornerstone
|
|
# (and struggle to get blake3 working in the browser)
|
|
image_md5_hash = models.CharField(max_length=64, null=True, blank=True)
|
|
|
|
# The second is a blake3 hash of the direct pixel data
|
|
# This is much faster when using pydicom
|
|
image_blake3_hash = models.CharField(max_length=64, null=True, blank=True)
|
|
|
|
is_dicom = models.BooleanField(default=False)
|
|
|
|
removed = models.BooleanField(
|
|
default=False,
|
|
help_text="Set to true if the file this object refers to has been removed (either from series truncation or merging)",
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ["position"]
|
|
abstract = True
|
|
|
|
def get_file_size(self):
|
|
try:
|
|
return self.image.size
|
|
|
|
# We catch this for when the image does not exist
|
|
except FileNotFoundError:
|
|
return 0
|
|
|
|
def get_dicom_data(self):
|
|
try:
|
|
with pydicom.dcmread(self.image) as d:
|
|
return d
|
|
except pydicom.errors.InvalidDicomError:
|
|
return {}
|
|
|
|
def get_series(self):
|
|
return self.series
|
|
|
|
def get_dicom_json(self):
|
|
try:
|
|
json = pydicom.dcmread(self.image).to_json()
|
|
except pydicom.errors.InvalidDicomError:
|
|
return {}
|
|
return json
|
|
|
|
def get_dicom_info(self):
|
|
try:
|
|
info = pretty_print_dicom(pydicom.dcmread(self.image))
|
|
except pydicom.errors.InvalidDicomError:
|
|
info = "File is not a dicom."
|
|
return info
|
|
|
|
def generate_md5_hash(self):
|
|
if not self.removed:
|
|
image_hash, is_dicom = get_image_hash(
|
|
self.image, hash_type="md5", direct_pixel_data=False
|
|
)
|
|
|
|
self.is_dicom = is_dicom
|
|
self.image_md5_hash = image_hash
|
|
|
|
self.save()
|
|
|
|
def generate_blake3_hash(self):
|
|
if not self.removed:
|
|
image_blake3_hash, is_dicom = get_image_hash(
|
|
self.image, hash_type="blake3", direct_pixel_data=True
|
|
)
|
|
self.is_dicom = is_dicom
|
|
self.image_blake3_hash = image_blake3_hash
|
|
|
|
self.save()
|
|
|
|
def save(self, *args, **kwargs):
|
|
"""Override save method to add image hash"""
|
|
if self.image:
|
|
# if not self.image_md5_hash:
|
|
# image_hash, is_dicom = get_image_hash(self.image, hash_type="md5", direct_pixel_data=False)
|
|
|
|
# self.is_dicom = is_dicom
|
|
# self.image_md5_hash = image_hash
|
|
|
|
if not self.image_blake3_hash:
|
|
image_blake3_hash, is_dicom = get_image_hash(
|
|
self.image, hash_type="blake3", direct_pixel_data=True
|
|
)
|
|
self.is_dicom = is_dicom
|
|
self.image_blake3_hash = image_blake3_hash
|
|
|
|
## Hack for tests
|
|
# if image_hash != "12345ABCD":
|
|
# super().save(*args, **kwargs) # Call the "real" save() method.
|
|
|
|
super().save(*args, **kwargs) # Call the "real" save() method.
|
|
|
|
|
|
class SeriesBase(models.Model):
|
|
info = models.TextField(
|
|
blank=True,
|
|
help_text="Description of stack, for admin organisation, will not be visible when taking",
|
|
)
|
|
|
|
description = models.CharField(
|
|
null=True,
|
|
blank=True,
|
|
max_length=255,
|
|
help_text="Description of the series. This is usually visable to the user (as the name of the stack)",
|
|
)
|
|
|
|
open_access = models.BooleanField(
|
|
help_text="If a series should be freely available to browse", default=True
|
|
)
|
|
|
|
created_date = models.DateTimeField(auto_now_add=True)
|
|
modified_date = models.DateTimeField(auto_now=True)
|
|
|
|
class SeriesModifiedChocies(models.TextChoices):
|
|
NO = (
|
|
"NO",
|
|
"Not modified",
|
|
)
|
|
TR = (
|
|
"TR",
|
|
"Truncated",
|
|
)
|
|
RE = (
|
|
"RE",
|
|
"Resliced/resampled",
|
|
)
|
|
|
|
modified = models.CharField(
|
|
max_length=2,
|
|
default=SeriesModifiedChocies.NO,
|
|
choices=SeriesModifiedChocies.choices,
|
|
)
|
|
|
|
total_image_size = models.BigIntegerField(null=True, blank=True, help_text="Stores a (cached) total size of images within the series")
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def __str__(self):
|
|
return f"{self.pk}:{self.description}"
|
|
|
|
def check_user_can_edit(self, user: User):
|
|
if user.is_superuser:
|
|
return True
|
|
|
|
if user in self.get_author_objects():
|
|
return True
|
|
|
|
return False
|
|
|
|
def get_author_objects(self):
|
|
"""Returns a list of authors"""
|
|
if self.author:
|
|
return self.author.all()
|
|
else:
|
|
return ["None"]
|
|
|
|
def get_author_display(self):
|
|
return ", ".join([i.username for i in self.get_author_objects()])
|
|
|
|
def get_examination(self):
|
|
"""Returns a comma seperated text list of regions"""
|
|
return str(self.examination)
|
|
|
|
def get_examination_full(self):
|
|
examination = ""
|
|
plane = ""
|
|
contrast = ""
|
|
|
|
if self.examination:
|
|
examination = self.examination
|
|
if self.plane:
|
|
plane = " {}".format(self.plane)
|
|
if self.contrast:
|
|
contrast = " {}".format(self.contrast)
|
|
return "{}{}{}".format(examination, plane, contrast)
|
|
|
|
def get_images(self, include_removed=False):
|
|
if include_removed:
|
|
return self.images.all()
|
|
else:
|
|
return self.images.filter(removed=False)
|
|
|
|
def get_image_count(self):
|
|
return self.images.filter(removed=False).count()
|
|
|
|
def get_image_urls(self):
|
|
images = [
|
|
f"{REMOTE_URL}{i.image.url}" for i in self.images.filter(removed=False)
|
|
]
|
|
|
|
return ",".join(images)
|
|
|
|
def get_image_url_array_not_json(self):
|
|
return self.get_image_url_array(json_output=False)
|
|
|
|
def get_image_url_array_and_count(self, json_output=True):
|
|
return self.get_image_url_array(json_output=json_output, count=True)
|
|
|
|
def get_image_url_array(self, json_output=True, count=False):
|
|
images = [
|
|
f"{REMOTE_URL}{i.image.url}" for i in self.images.filter(removed=False)
|
|
]
|
|
|
|
if json_output:
|
|
if count:
|
|
return (json.dumps(images), len(images))
|
|
else:
|
|
return json.dumps(images)
|
|
else:
|
|
# This is a mess...
|
|
if count:
|
|
return (format_html('", "'.join(images)), len(images))
|
|
else:
|
|
return format_html('", "'.join(images))
|
|
|
|
def get_thumbnail(self, recreate=False, fail_loudly=False):
|
|
# Use prefetched images if available
|
|
images = getattr(self, '_prefetched_objects_cache', {}).get('images')
|
|
if images is not None:
|
|
images = [img for img in images if not img.removed]
|
|
else:
|
|
images = self.images.filter(removed=False)
|
|
|
|
if len(images) < 1:
|
|
return "No images", 0
|
|
|
|
img = findMiddle(images).image
|
|
try:
|
|
thumbnailer = get_thumbnailer(img)
|
|
if recreate:
|
|
thumbnailer.delete_thumbnails()
|
|
thumbnail = thumbnailer["exam-list"]
|
|
except InvalidImageFormatError as e:
|
|
if fail_loudly:
|
|
raise e
|
|
return format_html(
|
|
'<img title="{}" src="/static/not-found-image.jpg" />', img
|
|
), len(images)
|
|
return format_html('<img src="/media/{}" />', thumbnail), len(images)
|
|
|
|
def get_thumbnail_link(self):
|
|
return format_html(
|
|
"<a href='{}'>{}<a/>", self.get_absolute_url(), self.get_thumbnail()[0]
|
|
)
|
|
|
|
def get_block(self, series_number: None | int = None):
|
|
examination = self.get_examination_full()
|
|
thumb, image_number = self.get_thumbnail()
|
|
|
|
series_html = ""
|
|
if series_number is not None:
|
|
series_html = format_html(
|
|
"<span class='series-block-series-number'>Series {}</span>{}<br>",
|
|
series_number,
|
|
)
|
|
description = ""
|
|
if self.description:
|
|
description = format_html("{}<br/>", self.description)
|
|
|
|
return format_html(
|
|
"""{}{}
|
|
<span>
|
|
<span class='series-block-examination'>{}</span><br/>
|
|
<span class='series-block-thumbnail'>{}</span><br/>
|
|
<span class='series-block-image-number'>Images: <span class='series-block-image-number-count'>{}</span></span>
|
|
</span>""",
|
|
series_html,
|
|
description,
|
|
examination,
|
|
thumb,
|
|
image_number,
|
|
)
|
|
|
|
def order_by_upload_filename(self):
|
|
images = self.images.filter(removed=False)
|
|
|
|
filenames = []
|
|
map = {}
|
|
|
|
for i in images:
|
|
filenames.append(i.upload_filename)
|
|
map[i.upload_filename] = i
|
|
|
|
filenames = sorted(filenames)
|
|
|
|
n = 1
|
|
for f in filenames:
|
|
i = map[f]
|
|
i.position = n
|
|
i.save()
|
|
n = n + 1
|
|
|
|
def order_by_dicom(self, field="SliceLocation"):
|
|
"""
|
|
Orders images in the series by the given DICOM field (default: SliceLocation).
|
|
Only keeps minimal data in memory and uses bulk_update for efficiency.
|
|
"""
|
|
images = self.images.filter(removed=False)
|
|
sortable = []
|
|
|
|
for img in images:
|
|
try:
|
|
ds = pydicom.dcmread(img.image.path, stop_before_pixels=True)
|
|
value = getattr(ds, field, None)
|
|
if value is not None:
|
|
sortable.append((value, img))
|
|
except Exception:
|
|
continue # Skip unreadable or invalid DICOMs
|
|
|
|
if not sortable:
|
|
return
|
|
|
|
# Sort by the DICOM field value
|
|
sortable.sort(key=lambda x: x[0])
|
|
|
|
# Assign new positions and collect for bulk update
|
|
for idx, (value, img) in enumerate(sortable, start=1):
|
|
img.position = idx
|
|
|
|
# Bulk update all changed images
|
|
SeriesImage = self.images.model
|
|
SeriesImage.objects.bulk_update([img for _, img in sortable], ['position'])
|
|
|
|
def get_total_image_size(self, refresh=False):
|
|
"""Returns the total size of all images in the series.
|
|
|
|
If refresh is set to True the size will be recalculated and saved to the database
|
|
(in the field 'total_image_size')."""
|
|
if not refresh and self.total_image_size:
|
|
return self.total_image_size
|
|
|
|
images = self.images.filter(removed=False)
|
|
|
|
size = 0
|
|
|
|
for i in images:
|
|
size += i.get_file_size()
|
|
|
|
self.total_image_size = size
|
|
self.save()
|
|
|
|
return size
|
|
|
|
def anonymise_images(self):
|
|
# NOTE: this will not maintain the correct hashed filename
|
|
# but that doesn't matter as we will never get the same anonymisation
|
|
# even with the same dicom...
|
|
anonymizer = dicognito.anonymizer.Anonymizer()
|
|
|
|
for series_image in self.images.filter(removed=False):
|
|
file_path = os.path.join(settings.MEDIA_ROOT, series_image.image.name)
|
|
|
|
try:
|
|
with pydicom.dcmread(file_path) as dataset:
|
|
anonymizer.anonymize(dataset)
|
|
dataset.save_as(file_path)
|
|
except pydicom.errors.InvalidDicomError:
|
|
pass
|
|
|
|
|
|
class ExamOrCollectionGenericBase(models.Model, AuthorMixin):
|
|
"""Holds functions that relate to both case and other exams
|
|
|
|
e.g. user management
|
|
|
|
"""
|
|
|
|
# Is this actually used? It is but should be replaced on the inherited class
|
|
cid_users = GenericRelation("generic.CidUserExam")
|
|
|
|
name = models.CharField(max_length=200, help_text="Name of the exam/collection")
|
|
|
|
start_date = models.DateTimeField(
|
|
help_text="Date (and time) the exam/collection starts", null=True, blank=True
|
|
)
|
|
end_date = models.DateTimeField(
|
|
help_text="Date (and time) the exam/collection ends", null=True, blank=True
|
|
)
|
|
restrict_to_dates = models.BooleanField(
|
|
default=False,
|
|
help_text="If the exam/collection should only be taken between the start and end date times",
|
|
)
|
|
|
|
active = models.BooleanField(
|
|
help_text="If an exam/collection should be available to take", default=False
|
|
)
|
|
publish_results = models.BooleanField(
|
|
help_text="If an exam/collections results should be available", default=False
|
|
)
|
|
|
|
archive = models.BooleanField(
|
|
help_text="Archived exams/collections will remain on the test system but will not be displayed by default",
|
|
default=False,
|
|
)
|
|
|
|
open_access = models.BooleanField(
|
|
help_text="If the exam/collection is freely accessible (to view and edit on the test system)",
|
|
default=False,
|
|
)
|
|
|
|
candidates_only = models.BooleanField(
|
|
help_text="If the exam/collection is only available to candidates who have been added",
|
|
default=True,
|
|
)
|
|
|
|
authors_only = models.BooleanField(
|
|
help_text="If true only exam/collection authors will be able to view.",
|
|
default=False,
|
|
)
|
|
|
|
results_supervisor_visible = models.BooleanField(
|
|
help_text="If true the results will be visible to supervisors without candidate approval.",
|
|
default=False,
|
|
)
|
|
|
|
exam_open_access = models.BooleanField(default=False, help_text="Set to true if you want any registered user to be able to take the exam.")
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def save(self, *args, **kwargs):
|
|
self.full_clean()
|
|
return super().save(*args, **kwargs)
|
|
|
|
def clean(self, *args, **kwargs):
|
|
if self.restrict_to_dates:
|
|
if self.start_date is None:
|
|
raise ValidationError(
|
|
"If restrict to dates is set, a start date must be set"
|
|
)
|
|
|
|
if self.end_date is not None and self.end_date <= self.start_date:
|
|
raise ValidationError({"end_date": "End date must be after start date"})
|
|
|
|
return super().clean(*args, **kwargs)
|
|
|
|
def get_app_name(self):
|
|
raise NotImplementedError(
|
|
"You must implement get_app_name in the derived class"
|
|
)
|
|
|
|
def get_base_template(self):
|
|
return f"{self.get_app_name()}/exams.html"
|
|
|
|
def get_link_headers(self):
|
|
return "generic/exam_link_headers.html"
|
|
|
|
def check_user_can_edit(self, user: User):
|
|
if user.is_superuser:
|
|
return True
|
|
|
|
if user in self.get_author_objects():
|
|
return True
|
|
|
|
return False
|
|
|
|
def check_logged_in_user(self, request: HttpRequest):
|
|
"""Helper to check if the logged in user can access the exam"""
|
|
return self.check_cid_user(user=request.user, user_id=request.user.id)
|
|
|
|
def check_user_can_review(
|
|
self, cid, passcode, request: HttpRequest, active_only=False
|
|
):
|
|
return self.check_user_can_take(cid, passcode, request.user, active_only)
|
|
|
|
def check_user_can_take(self, cid, passcode, user=None, active_only=True):
|
|
"""
|
|
Helper to check if a user is allowed to access an exam/collection
|
|
|
|
Args:
|
|
cid (_type_): _description_
|
|
passcode (_type_): _description_
|
|
request (_type_): _description_
|
|
|
|
Raises:
|
|
Http404: If user does not have access
|
|
"""
|
|
if user is not None and self.is_author(user):
|
|
return
|
|
|
|
# If we are limiting by dates check that the current date is within the range
|
|
if self.restrict_to_dates:
|
|
if self.start_date >= timezone.now() or (
|
|
self.end_date is not None and self.end_date <= timezone.now()
|
|
):
|
|
raise Http404("Exam not found")
|
|
# If it is not active no one can take
|
|
elif active_only and not self.active:
|
|
raise Http404("Exam not found")
|
|
|
|
# If candidates only check they have access
|
|
if self.candidates_only:
|
|
if not self.check_cid_user(cid, passcode, user):
|
|
raise Http404("Error accessing exam")
|
|
return
|
|
|
|
# Otherwise check that they are a valid user.
|
|
if user is None:
|
|
raise Http404("Error accessing exam")
|
|
|
|
if not user.is_active:
|
|
raise Http404("Error accessing exam")
|
|
|
|
return
|
|
|
|
def check_cid_user(
|
|
self,
|
|
cid: int | None = None,
|
|
passcode: str | None = None,
|
|
user: User | None = None,
|
|
# request: HttpRequest | None = None,
|
|
user_id: int | None = None,
|
|
allow_authors: bool = True,
|
|
):
|
|
"""Checks if a user (cid or otherwise) is allowed to access (and take) the exam"""
|
|
print(f"Check cid user: {cid=}, {passcode=}, exam_id={self.pk}, {user=}, {user_id=}, {allow_authors=}")
|
|
if user is not None and user.is_superuser:
|
|
return True
|
|
|
|
if user_id is not None:
|
|
if allow_authors:
|
|
if self.author.filter(pk=user_id).exists():
|
|
return True
|
|
|
|
if cid is None and user_id is None:
|
|
if user.is_anonymous:
|
|
return False
|
|
if user is not None:
|
|
user_id = user.id
|
|
else:
|
|
return False
|
|
|
|
if self.exam_open_access and user_id is not None:
|
|
return True
|
|
|
|
if not self.valid_cid_users.exists() and not self.valid_user_users.exists():
|
|
return False
|
|
|
|
|
|
|
|
# Start by checking if the logged in user can access
|
|
if user_id is not None:
|
|
if self.valid_user_users.filter(pk=user_id).exists():
|
|
return True
|
|
|
|
# Then test CID data
|
|
if self.valid_cid_users.exists():
|
|
cid_user = self.valid_cid_users.filter(cid=cid).first()
|
|
|
|
if cid_user and cid_user.passcode == passcode:
|
|
return True
|
|
|
|
return False
|
|
|
|
return False
|
|
|
|
def get_or_create_cid_user_exam(
|
|
self,
|
|
cid: int | None = None,
|
|
cid_user: "CidUser | None" = None,
|
|
user_user: User | None = None,
|
|
start_time=None,
|
|
):
|
|
content_type = ContentType.objects.get_for_model(self)
|
|
|
|
if cid is not None:
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
|
|
if cid_user is not None:
|
|
c = CidUserExam.objects.filter(
|
|
content_type=content_type, object_id=self.pk, cid_user=cid_user
|
|
).first()
|
|
if c:
|
|
return c
|
|
|
|
elif user_user is not None:
|
|
c = CidUserExam.objects.filter(
|
|
content_type=content_type, object_id=self.pk, user_user=user_user
|
|
).first()
|
|
if c:
|
|
return c
|
|
else:
|
|
raise ValueError("Invalid cid / cid_user / user_user")
|
|
|
|
if start_time is None:
|
|
start_time = timezone.now()
|
|
|
|
if cid_user is not None:
|
|
new = CidUserExam(
|
|
content_type=content_type,
|
|
object_id=self.pk,
|
|
cid_user=cid_user,
|
|
start_time=start_time,
|
|
)
|
|
|
|
elif user_user is not None:
|
|
new = CidUserExam(
|
|
content_type=content_type,
|
|
object_id=self.pk,
|
|
user_user=user_user,
|
|
start_time=start_time,
|
|
)
|
|
else:
|
|
raise ValueError("Invalid cid / cid_user / user_user")
|
|
|
|
new.save()
|
|
return new
|
|
|
|
def get_question_cid_user_answer(self, question_index, cid):
|
|
raise NotImplementedError
|
|
|
|
def get_question_user_user_answer(self, question_index, user):
|
|
raise NotImplementedError
|
|
|
|
def get_cid_and_user_exams(
|
|
self, cid_user: Optional["CidUser"] = None, user_user: User | None = None
|
|
) -> "CidUserExam":
|
|
"""Returns a queryset of CidUserExam for this exam/collection"""
|
|
content_type = ContentType.objects.get_for_model(self)
|
|
if cid_user is None and user_user is None:
|
|
return CidUserExam.objects.filter(
|
|
content_type=content_type,
|
|
object_id=self.pk,
|
|
)
|
|
else:
|
|
if cid_user is not None:
|
|
return CidUserExam.objects.filter(
|
|
content_type=content_type, object_id=self.pk, cid_user=cid_user
|
|
)
|
|
else:
|
|
return CidUserExam.objects.filter(
|
|
content_type=content_type, object_id=self.pk, user_user=user_user
|
|
)
|
|
|
|
def get_cid_exams(self, cid_user: Optional["CidUser"]= None) -> "CidUserExam":
|
|
"""Returns a queryset of CidUserExam for this exam/collection"""
|
|
content_type = ContentType.objects.get_for_model(self)
|
|
|
|
if cid_user is None:
|
|
# Only return pure CID entries (cid_user set, user_user null)
|
|
return CidUserExam.objects.filter(
|
|
content_type=content_type,
|
|
object_id=self.pk,
|
|
cid_user__isnull=False,
|
|
user_user__isnull=True,
|
|
)
|
|
return CidUserExam.objects.filter(
|
|
content_type=content_type, object_id=self.pk, cid_user=cid_user
|
|
)
|
|
|
|
def get_user_exams(self, user_user: Optional[User] = None) -> "CidUserExam":
|
|
"""Returns a queryset of CidUserExam for this exam/collection"""
|
|
content_type = ContentType.objects.get_for_model(self)
|
|
if user_user is None:
|
|
return CidUserExam.objects.filter(
|
|
content_type=content_type, object_id=self.pk, user_user__isnull=False, cid_user__isnull=True
|
|
)
|
|
return CidUserExam.objects.filter(
|
|
content_type=content_type, object_id=self.pk, user_user=user_user
|
|
)
|
|
|
|
|
|
|
|
def clone_model(self):
|
|
M2M_fields = (
|
|
"exam_questions",
|
|
"author",
|
|
"valid_cid_users",
|
|
"valid_user_users",
|
|
"cid_user_groups",
|
|
"user_user_groups",
|
|
)
|
|
FK_fields = ("examcollection",)
|
|
|
|
model_dict = model_to_dict(self)
|
|
|
|
m2m_to_update = {}
|
|
for field in M2M_fields:
|
|
m2m_to_update[field] = model_dict[field]
|
|
del model_dict[field]
|
|
fk_to_update = {}
|
|
for field in FK_fields:
|
|
fk_to_update[field] = model_dict[field]
|
|
del model_dict[field]
|
|
|
|
cloned_model = self.__class__(**model_dict)
|
|
cloned_model.pk = None
|
|
|
|
for field in fk_to_update:
|
|
setattr(cloned_model, f"{field}_id", fk_to_update[field])
|
|
|
|
cloned_model.save()
|
|
|
|
for field in m2m_to_update:
|
|
rel = getattr(cloned_model, field)
|
|
rel.clear()
|
|
for item in m2m_to_update[field]:
|
|
rel.add(item)
|
|
|
|
return cloned_model
|
|
|
|
def add_markers(self, users):
|
|
self.markers.add(*users)
|
|
|
|
def add_marker(self, user):
|
|
self.markers.add(user)
|
|
|
|
def get_candidate_count(self):
|
|
return self.valid_cid_users.count() + self.valid_user_users.count()
|
|
|
|
|
|
class ExamBase(ExamOrCollectionGenericBase):
|
|
exam_mode = models.BooleanField(
|
|
help_text="If an exam should be taken in exam mode (users results will be submited to the server for marking)",
|
|
default=False,
|
|
)
|
|
|
|
# randomise_question_order = models.BooleanField(
|
|
# help_text="If an exam should randomise the order of questions in RTS.",
|
|
# )
|
|
|
|
include_history = models.BooleanField(
|
|
help_text="If an exam should include history when taking",
|
|
default=False,
|
|
)
|
|
|
|
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)
|
|
exam_json_id = models.IntegerField(
|
|
default=1, help_text="auto incrementing field when json recreated"
|
|
)
|
|
|
|
stats_mean = models.FloatField(default=0)
|
|
stats_mode = models.CharField(default=0, max_length=25)
|
|
stats_median = models.FloatField(default=0)
|
|
|
|
stats_candidates = models.FloatField(default=0)
|
|
|
|
stats_min = models.FloatField(default=0)
|
|
stats_max = models.FloatField(default=0)
|
|
|
|
stats_max_possible = models.FloatField(default=0)
|
|
|
|
stats_graph = models.TextField(default=0)
|
|
|
|
user_scores = models.JSONField(default=dict, blank=True)
|
|
|
|
exam_results_emailed = models.DateTimeField(default=None, null=True, blank=True)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def __str__(self):
|
|
if self.start_date and self.start_date is not None:
|
|
print(self.start_date)
|
|
print(to_current_timezone(self.start_date))
|
|
return f"{self.name} ({to_current_timezone(self.start_date):%d %B %Y})"
|
|
else:
|
|
return self.name
|
|
|
|
def save(self, *args, recreate_json=True, **kwargs):
|
|
self.recreate_json = recreate_json
|
|
|
|
super().save(*args, **kwargs)
|
|
|
|
# def clean(self, *args, **kwargs):
|
|
# if self.user_scores is None:
|
|
# self.user_scores = {}
|
|
# return super().clean(*args, **kwargs)
|
|
|
|
def get_questions(self):
|
|
"""Returns a list of questions in the exam in the correct order
|
|
|
|
This should be used in preference to exam_questions.all() as it
|
|
will order the questions correctly"""
|
|
return self.exam_questions.all().order_by("examquestiondetail__sort_order")
|
|
|
|
def order_questions(self):
|
|
"""Modifies the examquestiondetail sort_order to sequentially order the cases"""
|
|
for n, c in enumerate(self.examquestiondetail_set.all().order_by("sort_order")):
|
|
c.sort_order = n
|
|
c.save()
|
|
|
|
def get_exam_stats(self, name=True):
|
|
if self.stats_candidates < int(4):
|
|
return f"- Candidates: {int(self.stats_candidates)} (too few to generate stats)"
|
|
|
|
text = f"""- Candidates: {int(self.stats_candidates)}
|
|
- Mean: {self.stats_mean:.2f} Median: {self.stats_median} Mode: {self.stats_mode} [Min: {self.stats_min} / Max: {self.stats_max}]
|
|
"""
|
|
if name:
|
|
text = f"""{self.name}
|
|
{"-"*len(self.name)}
|
|
{text}
|
|
"""
|
|
return text
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("{}:exam_overview".format(self.app_name), kwargs={"pk": self.pk})
|
|
|
|
def get_cid_edit_url(self):
|
|
return reverse(
|
|
"{}:exam_cids_edit".format(self.app_name), kwargs={"exam_id": self.pk}
|
|
)
|
|
|
|
def get_user_edit_url(self):
|
|
return reverse(
|
|
"{}:exam_users_edit".format(self.app_name), kwargs={"exam_id": self.pk}
|
|
)
|
|
|
|
def get_take_url(self):
|
|
return f"{settings.REMOTE_URL}/rts"
|
|
|
|
def get_exam_name(self):
|
|
return str(self)
|
|
|
|
def get_json_url(self, cid=None, passcode=None):
|
|
if cid is None:
|
|
return reverse("{}:exam_json".format(self.app_name), args=(self.pk,))
|
|
else:
|
|
return reverse(
|
|
"{}:exam_json_cid".format(self.app_name), args=(self.pk, cid, passcode)
|
|
)
|
|
|
|
def get_time_limit(self):
|
|
"""Returns a human readable time limit"""
|
|
if self.time_limit is None:
|
|
return "No time limit set"
|
|
h, m, s = str(datetime.timedelta(seconds=self.time_limit)).split(":")
|
|
|
|
time_limit = ""
|
|
if s != "0" and s != "00":
|
|
time_limit = f" {s} second(s)"
|
|
if m != "0":
|
|
time_limit = f"{m} minute(s) {time_limit}"
|
|
if h != "0":
|
|
time_limit = f"{h} hour(s) {time_limit}"
|
|
|
|
return time_limit.strip()
|
|
|
|
def get_question_index(self, question):
|
|
return list(self.get_questions()).index(question)
|
|
|
|
def get_question_by_index(self, index: int):
|
|
# There must be a better way to do this
|
|
return list(self.get_questions())[index]
|
|
|
|
def get_cid_user_score(self, cid_user):
|
|
c = "c/" + str(cid_user)
|
|
if c in self.user_scores:
|
|
return self.user_scores[c]
|
|
else:
|
|
return False
|
|
|
|
def get_user_users_with_scores(self):
|
|
user_ids = [i[2:] for i in self.user_scores if i.startswith("u/")]
|
|
|
|
return User.objects.filter(pk__in=user_ids)
|
|
|
|
def generate_user_report(self, user):
|
|
exam_text = [f"Candidate {user.first_name} [{user.email}]"]
|
|
|
|
exam_name = self.name
|
|
|
|
exam_text.append(f"{'='*len(exam_name)}\n{exam_name}\n{'='*len(exam_name)}")
|
|
|
|
try:
|
|
user_score = self.user_scores["u/" + str(user.pk)]["score"]
|
|
|
|
if self.app_name == "rapids":
|
|
callstate = self.user_scores["u/" + str(user.pk)]["callstates"]
|
|
except KeyError:
|
|
user_score = "Score not generated"
|
|
|
|
try:
|
|
percentage = f" ({user_score / int(self.stats_max_possible) * 100:.2f}%)"
|
|
except TypeError:
|
|
percentage = ""
|
|
|
|
exam_text.append(
|
|
f"""Results\n{'-'*len('Results')}
|
|
# Score: {user_score} / {int(self.stats_max_possible)}{percentage}
|
|
"""
|
|
)
|
|
stats = self.get_exam_stats(name=False)
|
|
exam_text.append(f"Stats\n{'-'*len('Stats')}\n{stats}")
|
|
|
|
rapids_extra = ""
|
|
if self.app_name == "rapids":
|
|
rapids_extra = f"Answers breakdown: {callstate}"
|
|
exam_text.append(rapids_extra)
|
|
|
|
msg = "\n\n".join(exam_text)
|
|
|
|
html_msg = f"""
|
|
<p>Candidate: {user.first_name} [{user.email}]</p>
|
|
<h2>{exam_name}</h2>
|
|
|
|
<h3>Results</h3>
|
|
Score: {user_score} / {int(self.stats_max_possible)} {percentage}
|
|
|
|
<h3>Stats</h3>
|
|
{stats}<br/>
|
|
<a href="{settings.REMOTE_URL}{reverse("{}:exam_stats".format(self.app_name), args=(self.pk,))}">View all stats (including graph) here</a>
|
|
|
|
"""
|
|
if self.app_name == "rapids":
|
|
html_msg = f"{html_msg}<br/>{rapids_extra}"
|
|
|
|
return msg, html_msg
|
|
|
|
def email_user_results(self, user, resend=False, additional_emails=None):
|
|
# if self.results_email_sent and not resend:
|
|
# return False, "Already sent."
|
|
|
|
# Get a list of taken exams
|
|
msg, html_msg = self.generate_user_report(user)
|
|
|
|
if not user.email:
|
|
return [False, "User has no email"]
|
|
|
|
emails = [user.email]
|
|
|
|
extra = ""
|
|
|
|
if user.userprofile.supervisor is not None:
|
|
supervisor_email = user.userprofile.supervisor.email
|
|
emails.append(supervisor_email)
|
|
else:
|
|
extra = "No supervisor"
|
|
|
|
if additional_emails is not None:
|
|
emails.extend(additional_emails)
|
|
|
|
try:
|
|
send_mail(
|
|
f"{self.name} results",
|
|
msg,
|
|
"no-reply@penracourses.org.uk",
|
|
emails,
|
|
fail_silently=False,
|
|
html_message=html_msg,
|
|
)
|
|
except SMTPException as e:
|
|
return [False, e]
|
|
|
|
# self.results_email_sent = True
|
|
# self.save()
|
|
return [True, extra]
|
|
|
|
|
|
class ExamUserStatus(models.Model):
|
|
datetime = models.DateTimeField(auto_now_add=True)
|
|
# cid_user: "CidUser" = models.ForeignKey(
|
|
# "CidUser", blank=True, null=True, on_delete=models.SET_NULL
|
|
# )
|
|
# user_user = models.ForeignKey(
|
|
# settings.AUTH_USER_MODEL, blank=True, null=True, on_delete=models.SET_NULL
|
|
# )
|
|
status = models.CharField(max_length=255)
|
|
extra = models.CharField(max_length=255)
|
|
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
exam = GenericForeignKey("content_type", "object_id")
|
|
|
|
cid_user_exam = models.ForeignKey(
|
|
"CidUserExam", blank=True, null=True, on_delete=models.SET_NULL
|
|
)
|
|
|
|
def __str__(self):
|
|
self.cid_user_exam: "CidUserExam"
|
|
if self.cid_user_exam.cid_user:
|
|
user = self.cid_user_exam.cid_user.cid
|
|
email = self.cid_user_exam.cid_user.email
|
|
return f"{self.datetime:%Y-%m-%d %H:%M:%S}: [CID] {user} ({email}) - {self.status} ({self.extra})"
|
|
else:
|
|
user = self.cid_user_exam.user_user.username
|
|
return f"{self.datetime:%Y-%m-%d %H:%M:%S}: [USER] {user} - {self.status} ({self.extra})"
|
|
|
|
|
|
class UserAnswerBase(models.Model):
|
|
cid = models.BigIntegerField(
|
|
blank=True,
|
|
null=True,
|
|
help_text="Candidate ID (limitied by BigIntegerField size)",
|
|
)
|
|
|
|
created = models.DateTimeField(auto_now_add=True)
|
|
updated = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
ordering = ["cid"]
|
|
|
|
def __str__(self) -> str:
|
|
return "{}/{}/{}".format(self.get_candidate_name(), self.exam, self.question.pk)
|
|
|
|
def get_candidate_name(self) -> str:
|
|
if self.cid is not None:
|
|
return str(self.cid)
|
|
else:
|
|
return self.user.username
|
|
|
|
def get_candidate_masked(self) -> str:
|
|
if self.cid is not None:
|
|
return str(self.cid)
|
|
else:
|
|
return str(self.user.id)
|
|
|
|
def get_absolute_url(self):
|
|
return reverse(f"{self.app_name}:user_answer_view", kwargs={"pk": self.pk})
|
|
|
|
def get_answer(self):
|
|
"""might need overriding"""
|
|
return self.answer
|
|
|
|
|
|
class NoteType(models.Model):
|
|
note_type = models.CharField(max_length=200)
|
|
|
|
def __str__(self):
|
|
return self.note_type
|
|
|
|
class QuestionReview(models.Model):
|
|
class StatusChoices(models.TextChoices):
|
|
ACCEPTED = "AC", "Accepted"
|
|
OUTDATED = "OD", "Outdated"
|
|
ERROR = "ER", "Error"
|
|
REJECTED = "RJ", "Rejected"
|
|
IN_PROGRESS = "IP", "In Progress"
|
|
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
question = GenericForeignKey("content_type", "object_id")
|
|
|
|
|
|
author = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True
|
|
)
|
|
|
|
|
|
comment = models.TextField(blank=True)
|
|
status = models.CharField(max_length=2, choices=StatusChoices.choices, default=StatusChoices.ACCEPTED)
|
|
created_on = models.DateTimeField(auto_now_add=True)
|
|
|
|
def __str__(self):
|
|
return "{}: {} [{}] {}".format(
|
|
self.content_type,
|
|
self.get_author_str(),
|
|
self.created_on,
|
|
self.comment,
|
|
)
|
|
|
|
def get_author_str(self):
|
|
if self.author is not None:
|
|
return self.author.username
|
|
else:
|
|
return "Unknown"
|
|
|
|
|
|
|
|
class QuestionNote(models.Model):
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
question = GenericForeignKey("content_type", "object_id")
|
|
|
|
author = models.CharField(max_length=200, blank=True)
|
|
|
|
user_author = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True
|
|
)
|
|
|
|
note = models.TextField(blank=True)
|
|
created_on = models.DateTimeField(auto_now_add=True)
|
|
|
|
complete = models.BooleanField(default=False)
|
|
|
|
note_type = models.ForeignKey(
|
|
NoteType, on_delete=models.CASCADE, null=True, blank=True
|
|
)
|
|
|
|
# def get_absolute_url(self):
|
|
# self.pk = self.rapid_id
|
|
# return reverse('rapids:question_detail', kwargs={'pk': self.pk})
|
|
|
|
def __str__(self):
|
|
return "{}: {} [{}] {} / {}".format(
|
|
self.content_type,
|
|
self.get_author_str(),
|
|
self.created_on,
|
|
self.note_type,
|
|
self.note,
|
|
)
|
|
|
|
def get_absolute_url(self):
|
|
# TODO: return to question if logged in
|
|
return "/"
|
|
|
|
def get_object_url(self):
|
|
return self.question.get_absolute_url()
|
|
|
|
def get_author_str(self):
|
|
if self.user_author is not None:
|
|
return self.user_author.username
|
|
else:
|
|
return self.author
|
|
|
|
def get_short_str(self):
|
|
return f"{self.note_type} - {self.note}"
|
|
|
|
|
|
EXAM_TYPES = (
|
|
("physics", "physics_exams"),
|
|
("rapids", "rapid_exams"),
|
|
("sbas", "sba_exams"),
|
|
("anatomy", "anatomy_exams"),
|
|
("longs", "longs_exams"),
|
|
("casecollection", "casecollection_exams"),
|
|
)
|
|
|
|
|
|
class CidUser(models.Model):
|
|
cid = models.IntegerField(blank=True, null=True, unique=True)
|
|
passcode = models.CharField(blank=True, max_length=50)
|
|
active = models.BooleanField(default=True)
|
|
internal_candidate = models.BooleanField(default=False)
|
|
name = models.CharField(blank=True, max_length=255)
|
|
email = models.EmailField(blank=True)
|
|
supervisor = models.ForeignKey(
|
|
"Supervisor",
|
|
on_delete=models.SET_NULL,
|
|
blank=True,
|
|
null=True,
|
|
related_name="cid_user",
|
|
)
|
|
|
|
login_email_sent = models.BooleanField(default=False)
|
|
results_email_sent = models.BooleanField(default=False)
|
|
|
|
group = models.ForeignKey(
|
|
"CidUserGroup", on_delete=models.SET_NULL, null=True, blank=True
|
|
)
|
|
|
|
class Meta:
|
|
indexes = [models.Index(fields=["cid"])]
|
|
|
|
def __str__(self) -> str:
|
|
return str(self.cid)
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("generic:manage_cids")
|
|
|
|
def check_passcode(self, passcode) -> bool:
|
|
return self.passcode == passcode
|
|
|
|
def get_cid_exams(self, include_case_collections=True):
|
|
available_exams = []
|
|
|
|
for n, t in EXAM_TYPES:
|
|
if not include_case_collections and n == "casecollection":
|
|
continue
|
|
exam_rel = getattr(self, t)
|
|
if exam_rel.exists():
|
|
exams = exam_rel.filter(exam_mode=True, archive=False).order_by("name")
|
|
available_exams.append((n, exams))
|
|
|
|
return available_exams
|
|
|
|
def generate_exam_report(self):
|
|
exam_list = self.get_cid_exams()
|
|
|
|
name = ""
|
|
if self.name:
|
|
name = f"{self.name} "
|
|
|
|
exam_text = [f"Candidate {self.cid} {name}[{self.email}]"]
|
|
|
|
for exam_type, exams in exam_list:
|
|
exam_text.append(f"{'='*len(exam_type)}\n{exam_type}\n{'='*len(exam_type)}")
|
|
|
|
for exam in exams:
|
|
try:
|
|
user_score = exam.user_scores["c/" + str(self.cid)]
|
|
except KeyError:
|
|
user_score = "Score not generated"
|
|
t = exam.get_exam_stats()
|
|
|
|
try:
|
|
percentage = (
|
|
f" ({user_score / int(exam.stats_max_possible) * 100:.2f}%)"
|
|
)
|
|
except TypeError:
|
|
percentage = ""
|
|
|
|
exam_text.append(
|
|
f"""{t}
|
|
# Score: {user_score} / {int(exam.stats_max_possible)}{percentage}
|
|
"""
|
|
)
|
|
|
|
msg = "\n\n".join(exam_text)
|
|
|
|
return msg
|
|
|
|
html_msg = f""""""
|
|
|
|
def email_results(self, resend=False, additional_emails=[]):
|
|
"""Emails a CID candidate results (internal candidates)
|
|
|
|
***DEPRECATED***
|
|
"""
|
|
if not self.internal_candidate:
|
|
return False, "Not internal candidate."
|
|
|
|
if self.results_email_sent and not resend:
|
|
return False, "Already sent."
|
|
|
|
# Get a list of taken exams
|
|
msg = self.generate_exam_report()
|
|
|
|
emails = [self.email, self.supervisor.email]
|
|
|
|
if additional_emails:
|
|
emails.extend(additional_emails)
|
|
|
|
try:
|
|
send_mail(
|
|
f"{self.group} results",
|
|
msg,
|
|
"test@xkjq.uk",
|
|
emails,
|
|
fail_silently=False,
|
|
# html_message=html_msg,
|
|
)
|
|
except SMTPException as e:
|
|
return False, e
|
|
|
|
self.results_email_sent = True
|
|
self.save()
|
|
return True, ""
|
|
|
|
def email_details(self, resend: bool = False) -> Tuple[bool, str]:
|
|
"""Email candidate details to the CID user
|
|
|
|
Args:
|
|
resend (bool, optional): Force sending details (even if sent previously). Defaults to False.
|
|
|
|
Returns:
|
|
Tuple[bool, str]: Tuple containing the email status and any error message
|
|
"""
|
|
if self.login_email_sent and not resend:
|
|
return False, "Already sent."
|
|
|
|
login_url = reverse("cid_selector")
|
|
|
|
msg = f"""
|
|
See below for your details for: {self.group}
|
|
|
|
---------
|
|
CID: {self.cid}
|
|
Passcode: {self.passcode}
|
|
---------
|
|
|
|
You will need these to access to the test system.
|
|
|
|
Direct link to login:
|
|
{settings.REMOTE_URL}{login_url}?cid={self.cid}&passcode={self.passcode}
|
|
|
|
The above details are unique to you. Please do not share as otherwise others will be able to access and submit your results.
|
|
|
|
|
|
|
|
Please note this inbox is not monitored. Direct all enquiries to plh-tr.radiology-academy@nhs.net
|
|
"""
|
|
|
|
try:
|
|
# send_mail(
|
|
# "Candidate Details",
|
|
# msg,
|
|
# "test@xkjq.uk",
|
|
# [self.email],
|
|
# fail_silently=False,
|
|
# )
|
|
email = EmailMessage(
|
|
"Candidate Details",
|
|
msg,
|
|
"no-reply@penracourses.org.uk",
|
|
[self.email],
|
|
# ["ross.kruger@nhs.net"],
|
|
)
|
|
email.send(fail_silently=False)
|
|
except SMTPException as e:
|
|
return False, e
|
|
|
|
self.login_email_sent = True
|
|
self.save()
|
|
return True, ""
|
|
|
|
def get_email_details_url(self):
|
|
return reverse("generic:candidate_email_details", kwargs={"cid": self.cid})
|
|
|
|
def get_email_results_url(self):
|
|
return reverse("generic:candidate_email_results", kwargs={"cid": self.cid})
|
|
|
|
def get_email_results_resend_url(self):
|
|
return reverse(
|
|
"generic:candidate_email_results_resend", kwargs={"cid": self.cid}
|
|
)
|
|
|
|
|
|
def get_next_cid():
|
|
"""Returns the next available CID"""
|
|
return CidUser.objects.order_by("-cid").first().cid + 1
|
|
|
|
|
|
class CidUserExam(models.Model):
|
|
"""This holds the current status of the exam for a candidate
|
|
|
|
Contrast with ExamUserStatus which holds the status changes for a candidate
|
|
(and should be merged into here)
|
|
"""
|
|
|
|
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
|
object_id = models.PositiveIntegerField()
|
|
exam = GenericForeignKey("content_type", "object_id")
|
|
|
|
start_time = models.DateTimeField(blank=True, null=True)
|
|
end_time = models.DateTimeField(blank=True, null=True)
|
|
|
|
completed = models.BooleanField(
|
|
default=False,
|
|
help_text="If a exam has been completed. This will usually lock the exam to further reponses.",
|
|
)
|
|
|
|
cid_user = models.ForeignKey(
|
|
CidUser,
|
|
on_delete=models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
related_name="cid_userexam",
|
|
)
|
|
user_user = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
related_name="user_userexam",
|
|
)
|
|
|
|
manual_submission = models.BooleanField(default=False)
|
|
|
|
# TODO switch to json field?
|
|
results_emailed_status = models.CharField(max_length=255, blank=True)
|
|
|
|
share_with_supervisor = models.BooleanField(default=False, help_text="If true the exam status and results will be available to a users supervisor")
|
|
|
|
def __str__(self) -> str:
|
|
if self.cid_user is None:
|
|
try:
|
|
user = self.user_user.username
|
|
except AttributeError:
|
|
user = "None"
|
|
else:
|
|
user = self.cid_user.cid
|
|
|
|
if self.start_time is None:
|
|
start_time = ""
|
|
else:
|
|
start_time = f"{ self.start_time:%Y-%m-%d %H:%M }"
|
|
|
|
if self.end_time is None:
|
|
end_time = ""
|
|
else:
|
|
end_time = f"{ self.end_time:%Y-%m-%d %H:%M }"
|
|
|
|
return f"{self.exam} / {user}: {start_time} {end_time}"
|
|
|
|
def get_user_name(self) -> str:
|
|
if self.cid_user is None:
|
|
return self.user_user.username
|
|
else:
|
|
return "CID" + str(self.cid_user.cid)
|
|
|
|
def complete_exam(self):
|
|
# TODO add examuserstatus?
|
|
self.end_time = timezone.now()
|
|
self.completed = True
|
|
self.save()
|
|
|
|
|
|
CID_GROUP_EXAMS = (
|
|
("SBAs", "sba_cid_user_groups"),
|
|
("Physics", "physics_cid_user_groups"),
|
|
("Anatomy", "anatomy_cid_user_groups"),
|
|
("Rapids", "rapid_cid_user_groups"),
|
|
("Shorts", "shorts_cid_user_groups"),
|
|
("Longs", "longs_cid_user_groups"),
|
|
)
|
|
|
|
|
|
class BaseUserGroup(models.Model):
|
|
name = models.CharField(blank=True, max_length=50, help_text="Name of the Group")
|
|
archive = models.BooleanField(
|
|
default=False,
|
|
help_text="Archived groups remain on the test system but are not displayed by default",
|
|
)
|
|
|
|
open_access = models.BooleanField(
|
|
default=False, help_text="If the group is freely accessible to use"
|
|
)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
|
|
class CidUserGroup(BaseUserGroup):
|
|
def __str__(self) -> str:
|
|
return str(self.name)
|
|
|
|
def GetGroupUsers(self):
|
|
cid_users = self.ciduser_set.all()
|
|
|
|
s = ", ".join([f"{user.id} - {user.name}" for user in cid_users])
|
|
return s
|
|
|
|
def GetGroupExams(self):
|
|
exams = defaultdict(list)
|
|
for t, rel in CID_GROUP_EXAMS:
|
|
exam_rel = getattr(self, rel)
|
|
if exam_rel.exists():
|
|
exams[t].extend(exam_rel.all())
|
|
|
|
return dict(exams)
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("generic:cid_group_update", kwargs={"pk": self.pk})
|
|
|
|
|
|
USER_GROUP_EXAMS = (
|
|
("SBAs", "sba_user_user_groups"),
|
|
("Physics", "physics_user_user_groups"),
|
|
("Anatomy", "anatomy_user_user_groups"),
|
|
("Rapids", "rapid_user_user_groups"),
|
|
("Shorts", "shorts_user_user_groups"),
|
|
("Longs", "longs_user_user_groups"),
|
|
)
|
|
|
|
|
|
class UserUserGroup(BaseUserGroup):
|
|
users = models.ManyToManyField(
|
|
settings.AUTH_USER_MODEL,
|
|
blank=True,
|
|
related_name="user_groups",
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
return str(self.name)
|
|
|
|
def GetGroupUsers(self):
|
|
users = self.users.all()
|
|
|
|
s = ", ".join([f"{user.id} - {user.username} ({user.email})" for user in users])
|
|
return s
|
|
|
|
def GetGroupExams(self):
|
|
exams = defaultdict(list)
|
|
for t, rel in USER_GROUP_EXAMS:
|
|
exam_rel = getattr(self, rel)
|
|
if exam_rel.exists():
|
|
exams[t].extend(exam_rel.all())
|
|
|
|
return dict(exams)
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("generic:user_group_update", kwargs={"pk": self.pk})
|
|
|
|
|
|
class UserGrades(models.Model):
|
|
name = models.CharField(max_length=100)
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
|
|
class UserProfile(models.Model):
|
|
user = models.OneToOneField(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
|
|
supervisor = models.ForeignKey(
|
|
"Supervisor",
|
|
on_delete=models.SET_NULL,
|
|
blank=True,
|
|
null=True,
|
|
related_name="trainee",
|
|
)
|
|
registration_number = models.CharField(max_length=25, blank=True)
|
|
grade = models.ForeignKey(
|
|
UserGrades,
|
|
null=True,
|
|
blank=True,
|
|
help_text="User grade",
|
|
on_delete=models.CASCADE,
|
|
)
|
|
peninsula_trainee = models.BooleanField(default=False)
|
|
site = models.ForeignKey(
|
|
"Site",
|
|
on_delete=models.SET_NULL,
|
|
blank=True,
|
|
null=True,
|
|
help_text="Primary site / rotation location",
|
|
)
|
|
|
|
def getusername(self):
|
|
return self.user.username
|
|
|
|
def __str__(self):
|
|
return f"Userprofile {self.user}"
|
|
|
|
username = property(getusername)
|
|
|
|
cimar_sid = models.CharField(max_length=100, blank=True)
|
|
|
|
def get_exams(self):
|
|
available_exams = defaultdict(list)
|
|
|
|
for n, t in USER_EXAM_TYPES:
|
|
exam_rel = getattr(self.user, t)
|
|
if exam_rel.exists():
|
|
exams = exam_rel.filter(exam_mode=True, archive=False).order_by("name")
|
|
available_exams[n].extend(exams)
|
|
|
|
return dict(available_exams)
|
|
|
|
|
|
class Supervisor(models.Model):
|
|
"""Model to hold individual supervisor details (email and name)
|
|
|
|
This can be linked with a user account.
|
|
"""
|
|
|
|
email = models.EmailField(
|
|
unique=True, help_text="The (nhs.net) email address of the supervisor"
|
|
)
|
|
name = models.CharField(max_length=255, help_text="Name of the supervisor")
|
|
user = models.OneToOneField(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.CASCADE,
|
|
null=True,
|
|
blank=True,
|
|
help_text="If the supervisor has an account on the test system it can be associated here",
|
|
)
|
|
site = models.ForeignKey(
|
|
"Site",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
help_text="Hospital site at which the supervisor is based",
|
|
)
|
|
active = models.BooleanField(default=True)
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.name} ({self.email})"
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("generic:supervisor_detail", kwargs={"pk": self.pk})
|
|
|
|
def clean(self):
|
|
if self.email:
|
|
# normalize email: strip and lowercase for consistency
|
|
self.email = self.email.strip().lower()
|
|
if self.name:
|
|
self.name = self.name.strip()
|
|
|
|
class Meta:
|
|
constraints = [
|
|
UniqueConstraint(Lower('email'), name='unique_supervisor_email_ci')
|
|
]
|
|
|
|
|
|
@receiver(post_save, sender=User)
|
|
def create_user_profile(sender, instance, created, **kwargs):
|
|
if created:
|
|
UserProfile.objects.create(user=instance)
|
|
|
|
|
|
@receiver(post_save, sender=User)
|
|
def save_user_profile(sender, instance, **kwargs):
|
|
instance.userprofile.save()
|
|
|
|
|
|
class ExamCollection(models.Model, AuthorMixin):
|
|
name = models.CharField(max_length=255)
|
|
|
|
date = models.DateField(blank=True, null=True)
|
|
|
|
archive = models.BooleanField(default=False)
|
|
|
|
author = models.ManyToManyField(
|
|
settings.AUTH_USER_MODEL,
|
|
blank=True,
|
|
help_text="Author/Manager(s) of the exam collection",
|
|
)
|
|
|
|
def __str__(self):
|
|
if self.date is not None and self.date != "":
|
|
return f"{self.name} [{self.date}]"
|
|
else:
|
|
return f"{self.name}"
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("generic:examcollection_detail", kwargs={"pk": self.pk})
|
|
|
|
def get_exams(self):
|
|
return {
|
|
"anatomy": self.anatomy_exams.filter(archive=False),
|
|
"longs": self.longs_exams.filter(archive=False),
|
|
"physics": self.physics_exams.filter(archive=False),
|
|
"rapids": self.rapids_exams.filter(archive=False),
|
|
"sbas": self.sbas_exams.filter(archive=False),
|
|
}
|
|
|
|
def add_author_to_exams(self, users, exam_types: None | list[str] =None):
|
|
|
|
for exam_type, exams in self.get_exams().items():
|
|
if exam_types is None:
|
|
pass
|
|
elif exam_type not in exam_types:
|
|
continue
|
|
else:
|
|
raise ValueError(f"Invalid exam type {exam_type}")
|
|
|
|
for exam in exams:
|
|
exam.add_authors(users)
|
|
#return True
|
|
|
|
def add_marker_to_exams(self, users, exam_types: None | list[str] =None):
|
|
|
|
for exam_type, exams in self.get_exams().items():
|
|
if exam_types is None:
|
|
pass
|
|
elif exam_type not in exam_types:
|
|
continue
|
|
else:
|
|
pass
|
|
|
|
for exam in exams:
|
|
exam.add_markers(users)
|
|
#return True
|
|
|
|
|
|
#class Presentation(models.Model, AuthorMixin):
|
|
#
|
|
# name = models.CharField(max_length=255, help_text="Name of the presentation")
|
|
#
|
|
# date_created = models.DateField(help_text="Date the presentation was created", auto_now_add=True)
|
|
#
|
|
# date_updated = models.DateField(help_text="Date the presentation was last updated", auto_now=True)
|
|
#
|
|
# markdown = models.TextField(help_text="Markdown text for the presentation")
|
|
#
|
|
# output_html = models.TextField(help_text="HTML output of the presentation")
|
|
#
|
|
# author = models.ManyToManyField(
|
|
# settings.AUTH_USER_MODEL,
|
|
# blank=True,
|
|
# help_text="Author/Manager(s) of the exam collection",
|
|
# )
|
|
|
|
class CimarCase(models.Model):
|
|
uuid = models.CharField(max_length=255, unique=True)
|
|
|
|
viewer_link = models.URLField(blank=True)
|
|
|
|
study_details = models.JSONField(blank=True, default=dict)
|
|
|
|
study_schema = models.JSONField(blank=True, default=dict)
|
|
|
|
series_data = models.JSONField(blank=True, default=dict)
|
|
|
|
valid_uuid = models.BooleanField(default=False)
|
|
|
|
def load_series_data(self, api=None):
|
|
if api is None:
|
|
api = CimarAPI()
|
|
api.login(username=CIMAR_USERNAME, password=CIMAR_PASSWORD)
|
|
|
|
|
|
|
|
|
|
def refresh_study(self):
|
|
api = CimarAPI()
|
|
api.login(username=CIMAR_USERNAME, password=CIMAR_PASSWORD)
|
|
|
|
self.study_details = api.get_study_by_uuid(self.uuid)
|
|
self.study_schema = api.get_study_schema_by_uuid(self.uuid)
|
|
|
|
self.viewer_link = api.get_study_viewer_url_by_uuid(self.uuid)
|
|
|
|
series_data = []
|
|
|
|
for series in self.study_schema["series"]:
|
|
# Cache the thumbnail
|
|
#thumbnail_url = f"{self.get_series_image_thumbnail_link(series)}?{api.sid}"
|
|
#response = requests.get(thumbnail_url)
|
|
|
|
#thumbnail = CimarSeriesThumbnail(series_id=series["id"])
|
|
#thumbnail.image.save(f"{series['id']}.jpg", ContentFile(response.content))
|
|
#thumbnail.save()
|
|
|
|
|
|
series_data.append({
|
|
"series_id" : series["id"],
|
|
"series_description" : series["description"],
|
|
"image_count": len(series["images"]),
|
|
"thumbnail": self.get_series_image_thumbnail_link(series)
|
|
|
|
}
|
|
)
|
|
|
|
|
|
self.series_data = {}
|
|
self.series_data["series"] = series_data
|
|
|
|
self.valid_uuid = True
|
|
self.save()
|
|
|
|
def __str__(self):
|
|
return f"{self.uuid} ({self.valid_uuid})"
|
|
|
|
def get_user_viewer_link(self, user):
|
|
return self.viewer_link + f"?sid={user.userprofile.cimar_sid}"
|
|
|
|
|
|
def get_pretty_study_details(self):
|
|
"""Function to display pretty version of our data"""
|
|
return get_pretty_json(self.study_details)
|
|
|
|
def get_pretty_study_schema(self):
|
|
"""Function to display pretty version of our data"""
|
|
return get_pretty_json(self.study_schema)
|
|
|
|
def get_pretty_study_series(self):
|
|
"""Function to display pretty version of our data"""
|
|
return get_pretty_json(self.series_data)
|
|
|
|
#def get_series_details_block(self, cimar_sid=None):
|
|
# if "series" in self.series_data:
|
|
# return render_to_string("cimar_series.html", {"series_data": self.series_data["series"], cimar_sid: cimar_sid})
|
|
#
|
|
# return "No series data available"
|
|
|
|
# return render_to_string("cimar_series.html", {"series_data": self.series_data["series"]})
|
|
|
|
|
|
|
|
def get_series_image_thumbnail_link(self,series):
|
|
|
|
middle_point = len(series["images"]) // 2
|
|
|
|
image = series["images"][middle_point]
|
|
|
|
return STORAGE_API_URL + f"/study/{self.study_details["storage_namespace"]}/{self.study_details["study_uid"]}/image/{image["id"]}/version/{image["version"]}/frame/0/thumbnail"
|
|
|
|
|
|
class CimarSeriesThumbnail(models.Model):
|
|
series_id = models.CharField(max_length=255)
|
|
image = models.ImageField(upload_to="cimar_series_thumbnails")
|
|
|
|
def __str__(self):
|
|
return f"{self.series_id}"
|
|
|
|
|
|
class FindingBase(models.Model):
|
|
description = models.TextField(
|
|
null=True, blank=True, help_text="Findings on the question"
|
|
)
|
|
annotation_json = models.TextField(null=True, blank=True)
|
|
viewport_json = models.TextField(null=True, blank=True)
|
|
# This has been migrated to install the image id (not index)
|
|
current_image_id_index = models.TextField(null=True, blank=True)
|
|
|
|
annotation_json_3d = models.JSONField(default=dict)
|
|
viewer_state_3d = models.JSONField(default=dict)
|
|
|
|
# TODO add tags
|
|
# tags = models.ManyToManyField(Tag, blank=True)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def __str__(self):
|
|
return self.description
|
|
|
|
def annotation_as_string(self) -> str:
|
|
return json.dumps(self.annotation_json)
|
|
|
|
def get_pretty_annotation_json(self):
|
|
return get_pretty_json(json.loads(self.annotation_json))
|
|
|
|
def get_pretty_viewport_json(self):
|
|
return get_pretty_json(json.loads(self.viewport_json)) |