Ross
2454 lines
80 KiB
Python
2454 lines
80 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, db_index=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, FileNotFoundError, OSError):
|
||
return {}
|
||
|
||
def get_full_url(self):
|
||
return f"{REMOTE_URL}{self.image.url}"
|
||
|
||
def get_series(self):
|
||
return self.series
|
||
|
||
def get_dicom_json(self):
|
||
try:
|
||
json = pydicom.dcmread(self.image).to_json()
|
||
except (pydicom.errors.InvalidDicomError, FileNotFoundError, OSError):
|
||
return {}
|
||
return json
|
||
|
||
def get_dicom_info(self):
|
||
try:
|
||
info = pretty_print_dicom(pydicom.dcmread(self.image))
|
||
except (pydicom.errors.InvalidDicomError, FileNotFoundError, OSError):
|
||
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, AuthorMixin):
|
||
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_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
|
||
|
||
"""
|
||
class InactiveException(Exception):
|
||
pass
|
||
|
||
# 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")
|
||
|
||
created = models.DateTimeField(auto_now_add=True, help_text="Creation timestamp")
|
||
updated = models.DateTimeField(auto_now=True, help_text="Last updated timestamp")
|
||
|
||
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,
|
||
)
|
||
|
||
@property
|
||
def content_type_id(self) -> int:
|
||
from django.contrib.contenttypes.models import ContentType
|
||
return ContentType.objects.get_for_model(self).id
|
||
|
||
|
||
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.")
|
||
|
||
pre_survey = models.ForeignKey(
|
||
"survey.Survey",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="%(app_label)s_%(class)s_pre_surveys",
|
||
help_text="Survey to be filled out BEFORE taking this collection/exam."
|
||
)
|
||
post_survey = models.ForeignKey(
|
||
"survey.Survey",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="%(app_label)s_%(class)s_post_surveys",
|
||
help_text="Survey to be filled out AFTER completing this collection/exam."
|
||
)
|
||
pre_survey_mandatory = models.BooleanField(
|
||
default=False,
|
||
help_text="If checked, candidates must complete the pre-take survey to start."
|
||
)
|
||
post_survey_mandatory = models.BooleanField(
|
||
default=False,
|
||
help_text="If checked, candidates must complete the post-take survey to view results."
|
||
)
|
||
|
||
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
|
||
"""
|
||
logger.debug(f"Checking user can take: {cid=}, passcode=****, user={user}, active_only={active_only}")
|
||
if user is not None and self.is_author(user):
|
||
logger.debug("User is author, allowing access")
|
||
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 with an explanatory message so the frontend can
|
||
# surface the reason to the end user.
|
||
logger.debug("Exam is outside of allowed dates, denying access")
|
||
raise Http404("Exam not currently available (outside allowed dates)")
|
||
# If it is not active no one can take
|
||
elif active_only and not self.active:
|
||
logger.debug("Exam is not active, denying access")
|
||
raise self.InactiveException("Exam currently inactive")
|
||
|
||
# If candidates only check they have access
|
||
if self.candidates_only:
|
||
logger.debug("Exam is candidates only, checking CID/user access")
|
||
if not self.check_cid_user(cid, passcode, user):
|
||
logger.debug("CID/user access check failed, denying access")
|
||
raise Http404("Error accessing exam")
|
||
return
|
||
|
||
# Otherwise check that they are a valid user.
|
||
if user is None:
|
||
logger.debug("No user provided, denying access")
|
||
raise Http404("Error accessing exam")
|
||
|
||
if not user.is_active:
|
||
logger.debug("User is not active, denying access")
|
||
raise Http404("Error accessing exam")
|
||
|
||
logger.debug("User passed all checks, allowing access")
|
||
|
||
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"""
|
||
logger.debug(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():
|
||
logger.debug("No valid CID or user users found")
|
||
return False
|
||
|
||
# Start by checking if the logged in user can access
|
||
logger.debug(f"Checking user access for user_id={user_id}")
|
||
if user_id is not None:
|
||
if self.valid_user_users.filter(pk=user_id).exists():
|
||
return True
|
||
else:
|
||
logger.debug("No valid user users found for user_id={user_id}")
|
||
|
||
# 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 = [f.name for f in self._meta.many_to_many]
|
||
FK_fields = ("examcollection",)
|
||
|
||
model_dict = model_to_dict(self)
|
||
|
||
m2m_to_update = {}
|
||
for field in m2m_fields:
|
||
if field in model_dict:
|
||
m2m_to_update[field] = model_dict[field]
|
||
del model_dict[field]
|
||
fk_to_update = {}
|
||
for field in FK_fields:
|
||
if field in model_dict:
|
||
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)
|
||
time_answered = models.DateTimeField(
|
||
null=True,
|
||
blank=True,
|
||
help_text="Time the question was answered (client-side)",
|
||
)
|
||
|
||
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
|
||
)
|
||
|
||
notes = models.TextField(blank=True, help_text="Internal notes about the candidate")
|
||
|
||
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."
|
||
|
||
# Build the message payload. Exposed as a separate method so callers
|
||
# can preview the email without actually sending it.
|
||
msg = self.generate_email_details_msg()
|
||
|
||
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 generate_email_details_msg(self) -> str:
|
||
"""Generate the candidate details email body without sending it.
|
||
|
||
Returns:
|
||
str: The plaintext email body that would be 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
|
||
"""
|
||
|
||
return msg
|
||
|
||
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")
|
||
|
||
shared_with_users = models.ManyToManyField(
|
||
settings.AUTH_USER_MODEL,
|
||
blank=True,
|
||
related_name="shared_cid_user_exams",
|
||
help_text="Users with whom this attempt has been manually shared.",
|
||
)
|
||
|
||
def get_sharing_info(self):
|
||
"""Return a structured dict describing who can view this attempt and why.
|
||
|
||
Returns a dict with keys:
|
||
always_visible_to – list of User objects (authors + markers of the exam)
|
||
supervisor – User or None (the supervisor's Django account, if applicable)
|
||
supervisor_enabled – bool (CidUserExam.share_with_supervisor value)
|
||
exam_supervisor_visible – bool (exam-level results_supervisor_visible flag)
|
||
manual_shares – QuerySet of User objects (shared_with_users)
|
||
"""
|
||
exam = self.exam
|
||
|
||
always_visible = []
|
||
seen_pks = set()
|
||
for attr in ("author", "markers"):
|
||
if hasattr(exam, attr):
|
||
for u in getattr(exam, attr).all():
|
||
if u.pk not in seen_pks:
|
||
seen_pks.add(u.pk)
|
||
always_visible.append(u)
|
||
|
||
exam_supervisor_visible = getattr(exam, "results_supervisor_visible", False)
|
||
supervisor_user = None
|
||
if self.share_with_supervisor or exam_supervisor_visible:
|
||
if self.user_user_id:
|
||
try:
|
||
sup = self.user_user.userprofile.supervisor
|
||
if sup and sup.user_id:
|
||
supervisor_user = sup.user
|
||
except Exception:
|
||
pass
|
||
elif self.cid_user_id:
|
||
sup = self.cid_user.supervisor
|
||
if sup and sup.user_id:
|
||
supervisor_user = sup.user
|
||
|
||
return {
|
||
"always_visible_to": always_visible,
|
||
"supervisor": supervisor_user,
|
||
"supervisor_enabled": self.share_with_supervisor,
|
||
"exam_supervisor_visible": exam_supervisor_visible,
|
||
"manual_shares": self.shared_with_users.all(),
|
||
}
|
||
|
||
def can_user_view(self, user):
|
||
"""Check whether a given authenticated User may view this attempt.
|
||
|
||
Permitted if the user is:
|
||
- The attempt owner (user_user)
|
||
- An author or marker of the related exam/collection
|
||
- The supervisor of the attempt owner (when share_with_supervisor=True or
|
||
the exam-level results_supervisor_visible flag is set)
|
||
- Manually listed in shared_with_users
|
||
"""
|
||
if not user or not user.is_authenticated:
|
||
return False
|
||
if self.user_user_id and self.user_user_id == user.pk:
|
||
return True
|
||
exam = self.exam
|
||
for attr in ("author", "markers"):
|
||
if hasattr(exam, attr) and getattr(exam, attr).filter(pk=user.pk).exists():
|
||
return True
|
||
exam_supervisor_visible = getattr(exam, "results_supervisor_visible", False)
|
||
if self.share_with_supervisor or exam_supervisor_visible:
|
||
if self.user_user_id:
|
||
try:
|
||
sup = self.user_user.userprofile.supervisor
|
||
if sup and sup.user_id == user.pk:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
elif self.cid_user_id:
|
||
sup = self.cid_user.supervisor
|
||
if sup and sup.user_id == user.pk:
|
||
return True
|
||
if self.shared_with_users.filter(pk=user.pk).exists():
|
||
return True
|
||
return False
|
||
|
||
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.manual_submission = True
|
||
self.save()
|
||
|
||
|
||
class Flag(models.Model):
|
||
"""A reusable flag model that can be attached to any object via a
|
||
GenericForeignKey. Useful for marking questions, items or other objects
|
||
across apps.
|
||
"""
|
||
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
||
object_id = models.PositiveIntegerField()
|
||
content_object = GenericForeignKey("content_type", "object_id")
|
||
|
||
# Optional actor who created the flag (either a user or a CID user)
|
||
user = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.CASCADE,
|
||
null=True,
|
||
blank=True,
|
||
related_name="flags",
|
||
)
|
||
cid_user = models.ForeignKey(
|
||
"CidUser",
|
||
on_delete=models.CASCADE,
|
||
null=True,
|
||
blank=True,
|
||
related_name="flags",
|
||
)
|
||
|
||
note = models.TextField(blank=True)
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
class Meta:
|
||
ordering = ("-created_at",)
|
||
|
||
def __str__(self) -> str:
|
||
actor = None
|
||
if self.user:
|
||
actor = f"user:{self.user.username}"
|
||
elif self.cid_user:
|
||
actor = f"cid:{self.cid_user.cid}"
|
||
else:
|
||
actor = "anon"
|
||
return f"Flag [{actor}] -> {self.content_type}({self.object_id})"
|
||
|
||
|
||
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
|
||
|
||
def bulk_add_user_user_groups(self, groups, exam_types: None | list[str] = None) -> int:
|
||
"""Bulk-add one or more UserUserGroup objects (or PKs) to all exams in this
|
||
collection.
|
||
|
||
Args:
|
||
groups: iterable of UserUserGroup instances or an iterable of PKs (ints),
|
||
or a Django QuerySet of UserUserGroup.
|
||
exam_types: optional list of exam type keys to limit the operation. Keys
|
||
are the same as returned by `get_exams()` (e.g. 'physics', 'anatomy').
|
||
|
||
Returns:
|
||
Number of exams that were updated (int).
|
||
|
||
Note: permission checks are not performed here; callers should ensure the
|
||
current user is allowed to modify these exams.
|
||
"""
|
||
# import locally to avoid circular import surprises when module is imported
|
||
UserUserGroup = globals().get("UserUserGroup")
|
||
if UserUserGroup is None:
|
||
raise RuntimeError("UserUserGroup model not available")
|
||
|
||
# Debug: log what we received
|
||
try:
|
||
group_pks = [g.pk for g in groups] if groups is not None else []
|
||
except Exception:
|
||
group_pks = []
|
||
logger.debug("bulk_add_user_user_groups called on collection=%s with groups=%s exam_types=%s", self.pk, group_pks, exam_types)
|
||
|
||
# Normalize groups to a list of model instances
|
||
if groups is None:
|
||
return 0
|
||
|
||
# If groups is a queryset or list of model instances, materialize it
|
||
try:
|
||
# QuerySet-ish
|
||
groups_qs = list(groups)
|
||
except TypeError:
|
||
# single value provided
|
||
groups_qs = [groups]
|
||
|
||
# If provided values were ints (PKs), resolve them to model instances
|
||
if groups_qs and all(isinstance(g, int) for g in groups_qs):
|
||
groups_qs = list(UserUserGroup.objects.filter(pk__in=groups_qs))
|
||
|
||
if not groups_qs:
|
||
return 0
|
||
|
||
updated = 0
|
||
for exam_type, exams in self.get_exams().items():
|
||
if exam_types is not None and exam_type not in exam_types:
|
||
continue
|
||
|
||
for exam in exams:
|
||
# Each exam model exposes a `user_user_groups` ManyToManyField
|
||
try:
|
||
before_pks = set(exam.user_user_groups.values_list('pk', flat=True))
|
||
exam.user_user_groups.add(*groups_qs)
|
||
# Ensure any downstream signals / caches see the change
|
||
exam.save()
|
||
after_pks = set(exam.user_user_groups.values_list('pk', flat=True))
|
||
# log if expected group pks are not present after add
|
||
try:
|
||
expected_pks = {g.pk for g in groups_qs}
|
||
except Exception:
|
||
expected_pks = set()
|
||
if not expected_pks.issubset(after_pks):
|
||
logger.warning(
|
||
"After add, exam %s missing expected user_user_groups; before=%s after=%s expected=%s",
|
||
getattr(exam, 'pk', exam), sorted(before_pks), sorted(after_pks), sorted(expected_pks),
|
||
)
|
||
updated += 1
|
||
except Exception as e:
|
||
logger.exception("Failed to add user_user_groups to exam %s: %s", getattr(exam, 'pk', exam), e)
|
||
|
||
return updated
|
||
|
||
def bulk_add_cid_user_groups(self, groups, exam_types: None | list[str] = None) -> int:
|
||
"""Bulk-add one or more CidUserGroup objects (or PKs) to all exams in this
|
||
collection. Returns number of exams updated.
|
||
"""
|
||
CidUserGroup = globals().get("CidUserGroup")
|
||
if CidUserGroup is None:
|
||
raise RuntimeError("CidUserGroup model not available")
|
||
|
||
# Debug: log what we received
|
||
try:
|
||
group_pks = [g.pk for g in groups] if groups is not None else []
|
||
except Exception:
|
||
group_pks = []
|
||
logger.debug("bulk_add_cid_user_groups called on collection=%s with groups=%s exam_types=%s", self.pk, group_pks, exam_types)
|
||
|
||
if groups is None:
|
||
return 0
|
||
|
||
try:
|
||
groups_qs = list(groups)
|
||
except TypeError:
|
||
groups_qs = [groups]
|
||
|
||
if groups_qs and all(isinstance(g, int) for g in groups_qs):
|
||
groups_qs = list(CidUserGroup.objects.filter(pk__in=groups_qs))
|
||
|
||
if not groups_qs:
|
||
return 0
|
||
|
||
updated = 0
|
||
for exam_type, exams in self.get_exams().items():
|
||
if exam_types is not None and exam_type not in exam_types:
|
||
continue
|
||
|
||
for exam in exams:
|
||
# Each exam model exposes a `cid_user_groups` ManyToManyField
|
||
try:
|
||
before_pks = set(exam.cid_user_groups.values_list('pk', flat=True))
|
||
exam.cid_user_groups.add(*groups_qs)
|
||
exam.save()
|
||
after_pks = set(exam.cid_user_groups.values_list('pk', flat=True))
|
||
try:
|
||
expected_pks = {g.pk for g in groups_qs}
|
||
except Exception:
|
||
expected_pks = set()
|
||
if not expected_pks.issubset(after_pks):
|
||
logger.warning(
|
||
"After add, exam %s missing expected cid_user_groups; before=%s after=%s expected=%s",
|
||
getattr(exam, 'pk', exam), sorted(before_pks), sorted(after_pks), sorted(expected_pks),
|
||
)
|
||
updated += 1
|
||
except Exception as e:
|
||
logger.exception("Failed to add cid_user_groups to exam %s: %s", getattr(exam, 'pk', exam), e)
|
||
|
||
return updated
|
||
|
||
|
||
#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):
|
||
if not self.annotation_json:
|
||
return ""
|
||
try:
|
||
return get_pretty_json(json.loads(self.annotation_json))
|
||
except Exception:
|
||
return ""
|
||
|
||
def get_pretty_viewport_json(self):
|
||
if not self.viewport_json:
|
||
return ""
|
||
try:
|
||
return get_pretty_json(json.loads(self.viewport_json))
|
||
except Exception:
|
||
return ""
|
||
|
||
class UserHiddenItem(models.Model):
|
||
"""Model to track which exams or collections have been hidden by a user.
|
||
"""
|
||
user = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.CASCADE,
|
||
related_name="hidden_items",
|
||
)
|
||
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
||
object_id = models.PositiveIntegerField()
|
||
content_object = GenericForeignKey("content_type", "object_id")
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
class Meta:
|
||
constraints = [
|
||
models.UniqueConstraint(
|
||
fields=["user", "content_type", "object_id"],
|
||
name="unique_user_hidden_item",
|
||
)
|
||
]
|
||
|
||
def __str__(self):
|
||
return f"{self.user} hid {self.content_object}"
|
||
|