ross/penracourses
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
4628f39a4947fc7ae46dae18e7d1c9a4b85be062
penracourses/atlas/views.py
T

5179 lines
182 KiB
Python
Executable File
Raw Blame History

import json
import pprint
from typing import Any
from dal import autocomplete
from django.db.models.base import Model as Model
from django.db.models.query import QuerySet
from django.db.models import Prefetch
from django.db.models import Q
from django.shortcuts import render, get_object_or_404, redirect
from django import forms
from django.utils import timezone
from django.views.decorators.http import require_POST
from django.views.decorators.http import require_http_methods
from django.template.loader import render_to_string
from django.db.models import Count
# from django.contrib.auth.models import User
from django.contrib.auth.decorators import login_required, user_passes_test
from django.contrib.auth.models import User
from django.core.exceptions import (
FieldError,
ObjectDoesNotExist,
PermissionDenied,
ViewDoesNotExist,
)
from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin
from django.views.generic import View
from django.views.generic.detail import DetailView
import pydicom
from pydicom.uid import generate_uid
from generic.mixins import SuperuserRequiredMixin, UserConfigurablePaginationMixin
from django.contrib.admin.views.decorators import staff_member_required
from django.core.paginator import Paginator
from django.views.generic.edit import CreateView, UpdateView, DeleteView, FormView
from django.views.generic import ListView
from django_filters.views import FilterView
from django_tables2 import SingleTableMixin
from .filters import ResourceFilter
from .tables import ResourceTable
from django.views.decorators.csrf import csrf_exempt
from django.urls import reverse_lazy, reverse
from django.urls.exceptions import NoReverseMatch
from django.http import Http404, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from django.contrib.contenttypes.models import ContentType
from generic.models import CidUser, CidUserExam, CimarCase, Site
from atlas.helpers import get_cases_available_to_user
from rad.settings import REMOTE_URL, CIMAR_USERNAME, CIMAR_PASSWORD
from .forms import (
AddCollectionToCaseForm,
CaseAuthorForm,
CaseCollectionAuthorForm,
CaseCollectionCaseFormSet,
CaseCollectionForm,
CaseCollectionUpdateCaseForm,
CaseDetailForm,
CaseQuestionForm,
CaseDisplaySetForm,
CaseForm,
CaseResourceFormSet,
CaseUpdateSeriesForm,
CidQuestionAnswerForm,
CidReportAnswerForm,
CidReportAnswerMarkForm,
ConditionAutocompleteForm,
ConditionForm,
ExamGroupsForm,
FindingForm,
JsonAnswerForm,
#ModelPriorCaseForm,
PriorCaseForm,
QuestionSchemaForm,
ResourceForm,
SelfReviewForm,
SeriesAuthorForm,
SeriesForm,
SeriesImageFormSet,
SeriesFormSet,
SeriesFindingForm,
CaseDifferentialFormSet,
StructureForm,
SubspecialtyForm,
PresentationForm,
PathologicalProcessForm,
UserQuestionAnswerForm,
UserReportAnswerForm,
)
from .models import (
Case,
CaseCollection,
CaseDetail,
CaseDisplaySet,
CasePrior,
CaseResource,
CidReportAnswer,
Condition,
Differential,
PathologicalProcess,
Presentation,
QuestionSchema,
Resource,
SelfReview,
Series,
Examination,
Finding,
SeriesDetail,
Structure,
Subspecialty,
SeriesFinding,
SeriesImage,
UncategorisedDicom,
UserReportAnswer,
PrerequisiteRequired
)
from .models import NormalCase
from .tables import (
CaseCollectionTable,
CaseTable,
ConditionTable,
FindingTable,
PathologicalProcessTable,
PresentationTable,
QuestionSchemaTable,
SeriesTable,
StructureTable,
SubspecialtyTable,
)
from .filters import (
CaseCollectionFilter,
CaseFilter,
ConditionFilter,
FindingFilter,
PathologicalProcessFilter,
PresentationFilter,
QuestionSchemaFilter,
SeriesFilter,
StructureFilter,
SubspecialtyFilter,
NormalCaseFilter,
)
from .tasks import push_case_to_cimar_task
from django_tables2 import SingleTableView, SingleTableMixin
from django_filters.views import FilterView
# from autocomplete import HTMXAutoComplete
from .decorators import (
user_is_author_or_atlas_editor,
user_has_case_view_access,
user_is_author_or_atlas_series_checker_or_atlas_marker_or_open_access,
user_is_atlas_editor,
user_is_author_or_atlas_series_checker,
user_is_atlas_marker,
user_is_collection_author_or_atlas_editor,
)
from collections import defaultdict
import json
import statistics
import plotly.express as px
from django.core.cache import cache
from django.conf import settings
from helpers.images import compare_dicom_datasets, image_as_base64, print_dicom
import logging
from copy import deepcopy
from django.forms.models import model_to_dict
from generic.views import (
AuthorRequiredMixin,
ExamCloneMixin,
ExamGroupsUpdateBase,
ExamViews,
SeriesImagesZipViewBase,
)
from reversion.views import RevisionMixin
import reversion
import os
import difflib
from helpers.cimar import CimarAPI
logger = logging.getLogger(__name__)
@login_required
def case_search_partial(request):
"""Return an HTMX partial listing cases matching the query parameter `q`.
This view returns a small list-group fragment that can be swapped into
the index page. It respects the user's available cases via
get_cases_available_to_user.
"""
# NOTE (search migration):
#
# This view currently performs simple `__icontains` lookups across Case
# fields and related models. For better relevance and performance you can
# migrate this to use PostgreSQL full-text search (FTS). Recommended steps:
#
# 1) Enable optional Postgres extensions (if you want trigram/unaccent):
# - Add migrations: `TrigramExtension()` and optionally `UnaccentExtension()`
# from `django.contrib.postgres.operations`.
#
# 2) Create a tsvector expression index (recommended) or add a
# SearchVectorField on the Case model. Example expression index SQL:
# CREATE INDEX CONCURRENTLY idx_case_search ON atlas_case
# USING GIN (to_tsvector('english', coalesce(title,'') || ' ' ||
# coalesce(description,'') || ' ' || coalesce(history,'')));
#
# Alternatively use a Django migration with `GinIndex` on an
# `SearchVectorField` or a `Func` expression.
#
# 3) Query using django.contrib.postgres.search utilities. Example:
# from django.contrib.postgres.search import (
# SearchVector, SearchQuery, SearchRank
# )
# sv = (SearchVector('title', weight='A') +
# SearchVector('description', weight='B') +
# SearchVector('history', weight='C'))
# sq = SearchQuery(q)
# qs = Case.objects.annotate(search=sv).filter(search=sq)
# .annotate(rank=SearchRank(F('search'), sq)).order_by('-rank')
#
# 4) For interactive/fuzzy typing consider pg_trgm (TrigramExtension)
# and `TrigramSimilarity` as a fallback when FTS returns few results.
#
# 5) Keep in mind to re-index (or recreate indexes) when you change
# the fields included in the tsvector. Use concurrent index creation
# for minimal downtime.
#
# If you want, I can prepare the migration, model/index changes and a
# Django-code example to wire FTS into this view and tune weights.
q = request.GET.get('q', '') or request.GET.get('case-search-input', '')
q = q.strip()
# Start with cases available to the user
qs = get_cases_available_to_user(request.user)
if q:
from django.db.models import Q
# Search across multiple Case fields and related names (conditions,
# presentations, subspecialties, series fields and authors).
qs = qs.filter(
Q(title__icontains=q)
| Q(description__icontains=q)
| Q(history__icontains=q)
| Q(discussion__icontains=q)
| Q(report__icontains=q)
| Q(condition__name__icontains=q)
| Q(presentation__name__icontains=q)
| Q(subspecialty__name__icontains=q)
| Q(series__description__icontains=q)
# Series.modality is a FK to generic.Modality; search its text field
| Q(series__modality__modality__icontains=q)
| Q(series__examination__examination__icontains=q)
| Q(author__first_name__icontains=q)
| Q(author__last_name__icontains=q)
| Q(author__username__icontains=q)
).distinct()
qs = qs.order_by('-created_date')[:20]
html = render_to_string('atlas/partials/_case_search_results.html', {'cases': qs}, request=request)
return HttpResponse(html)
class AuthorOrCheckerRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
if (
self.request.user.groups.filter(name="atlas_editor").exists()
or self.request.user.is_superuser
):
return obj
if self.request.user not in obj.get_author_objects():
raise PermissionDenied() # or Http404
return obj
class AtlasEditorRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
if (
self.request.user.groups.filter(name="atlas_editor").exists()
or self.request.user.is_superuser
):
return obj
raise PermissionDenied("You must be an atlas editor to do this.") # or Http404
class NormalCaseList(LoginRequiredMixin, FilterView):
model = NormalCase
template_name = "atlas/normals_list.html"
filterset_class = NormalCaseFilter
paginate_by = 25
def get_queryset(self):
qs = NormalCase.objects.select_related("case", "examination", "modality", "added_by")
# Restrict similar to CaseFilter: non-editors only see authored or open_access
if not self.request.user.groups.filter(name="atlas_editor").exists():
qs = qs.filter((Q(case__author__id=self.request.user.id) | Q(case__open_access=True)))
return qs.order_by("-added_date")
@login_required
@user_has_case_view_access
def case_displaysets(request, pk):
case = get_object_or_404(Case, pk=pk)
can_edit = case.check_user_can_edit(request.user)
# Get displayset id from GET parameter
displayset_id = request.GET.get("displayset")
edit = request.GET.get("edit", False)
selected_displayset = None
if displayset_id:
try:
selected_displayset = case.display_sets.get(pk=displayset_id)
except CaseDisplaySet.DoesNotExist:
selected_displayset = None
return render(
request,
"atlas/case_displaysets.html",
{
"case": case,
"edit": edit,
"can_edit": can_edit,
"selected_displayset": selected_displayset,
},
)
@login_required
@user_has_case_view_access
@require_http_methods(["GET", "POST"])
def case_displaysets_add(request, pk):
case = get_object_or_404(Case, pk=pk)
form = CaseDisplaySetForm(request.POST or None)
if request.method == "POST":
if form.is_valid():
ds = form.save(commit=False)
ds.case = case
ds.author = request.user
ds.viewerstate = request.POST.get("viewerstate") or None
ds.annotations = request.POST.get("annotations") or None
ds.save()
form.save_m2m()
# HTMX: return the new display set row HTML fragment
html = render_to_string("atlas/partials/displayset_row.html", {"ds": ds, "can_edit": True})
return HttpResponse(html)
else:
# HTMX: return the form HTML fragment with errors
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
return HttpResponse(html, status=400)
else:
# HTMX: return the blank form HTML fragment
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
return HttpResponse(html)
@login_required
@require_http_methods(["GET", "POST"])
def case_displaysets_edit(request, pk):
ds = get_object_or_404(CaseDisplaySet, pk=pk)
case = ds.case
if not case.check_user_can_edit(request.user):
return HttpResponse("You do not have permission to delete this display set.")
form = CaseDisplaySetForm(request.POST or None, instance=ds)
if request.method == "POST":
if form.is_valid():
ds = form.save(commit=False)
ds.viewerstate = request.POST.get("viewerstate") or ds.viewerstate
ds.annotations = request.POST.get("annotations") or ds.annotations
ds.save()
form.save_m2m()
# Return updated row HTML
html = render_to_string("atlas/partials/displayset_row.html", {"ds": ds, "can_edit": True}, request=request)
return HttpResponse(html)
else:
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
return HttpResponse(html, status=400)
else:
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
return HttpResponse(html)
@require_http_methods(["GET", "POST"])
def case_displaysets_detail(request, pk):
displayset = get_object_or_404(CaseDisplaySet, pk=pk)
case = displayset.case
# Redirect to the case_displaysets page with ?displayset=<pk>
return redirect(f"{reverse('atlas:case_displaysets', args=[case.pk])}?displayset={pk}")
def case_displaysets_delete(request, pk):
try:
displayset = CaseDisplaySet.objects.get(pk=pk)
except CaseDisplaySet.DoesNotExist:
raise Http404("Display set not found.")
if not displayset.case.check_user_can_edit(request.user):
raise PermissionDenied("You do not have permission to delete this display set.")
displayset.delete()
return HttpResponse("<li class='list-group-item' >Display set deleted successfully.")
def get_case_for_case_detail(pk: int) -> Case:
try:
case = (
Case.objects
.select_related() # Add any FK fields if needed
.prefetch_related(
Prefetch(
"seriesdetail_set",
queryset=SeriesDetail.objects.select_related("series").order_by("sort_order").prefetch_related(
Prefetch(
"series",
queryset=Series.objects.select_related("modality", "examination", "plane", "contrast")
.prefetch_related(
Prefetch(
"findings",
queryset=SeriesFinding.objects.prefetch_related(
"findings", "conditions", "structures"
)
),
Prefetch("images", queryset=SeriesImage.objects.filter(removed=False).order_by("position")),
)
),
),
),
Prefetch("subspecialty", queryset=Subspecialty.objects.all()),
Prefetch("condition", queryset=Condition.objects.all()),
Prefetch("presentation", queryset=Presentation.objects.all()),
Prefetch("pathological_process", queryset=PathologicalProcess.objects.all()),
Prefetch("caseresource_set", queryset=CaseResource.objects.all()),
Prefetch("casecollection_set", queryset=CaseCollection.objects.all()),
Prefetch("display_sets", queryset=CaseDisplaySet.objects.prefetch_related(
Prefetch("findings", queryset=Finding.objects.all()),
Prefetch("structures", queryset=Structure.objects.all()),
Prefetch("conditions", queryset=Condition.objects.all()),
)),
)
.get(pk=pk)
)
case.ordered_series = [sd.series for sd in case.seriesdetail_set.all()]
return case
except Case.DoesNotExist:
raise Http404("Case not found.")
@login_required
@user_has_case_view_access
def case_detail(request, pk):
case = get_case_for_case_detail(pk)
can_edit = case.check_user_can_edit(request.user)
is_atlas_editor = request.user.is_superuser or request.user.groups.filter(name="atlas_editor").exists()
is_case_author = case.author.filter(id=request.user.id).exists()
can_mark_normal = is_atlas_editor or is_case_author or request.user.is_superuser
return render(
request,
"atlas/case_detail.html",
{
"case": case,
"cimar_sid": request.user.userprofile.cimar_sid,
"can_edit": can_edit,
"is_atlas_editor": is_atlas_editor,
"is_case_author": is_case_author,
"can_mark_normal": can_mark_normal,
# Provide helper lists for resource search filters
"site_queryset": Site.objects.all(),
"subspecialty_queryset": Subspecialty.objects.all(),
},
)
@login_required
@require_POST
def toggle_case_normal(request, pk):
"""HTMX endpoint to toggle the NormalCase mark for a Case.
- Only users in the 'atlas_editor' group or superusers may toggle.
- On create, attempt to auto-extract age_years from DICOM metadata.
- Returns an HTML fragment that replaces the toggle block.
"""
case = get_case_for_case_detail(pk)
# Permission: atlas editors, case authors, or superusers
is_atlas_editor = request.user.is_superuser or request.user.groups.filter(name="atlas_editor").exists()
is_case_author = case.author.filter(id=request.user.id).exists()
can_mark_normal = is_atlas_editor or is_case_author or request.user.is_superuser
if not can_mark_normal:
return HttpResponse(status=403)
# If already marked normal, unmark (delete the NormalCase)
try:
normal = case.normal_case
except Exception:
normal = None
if normal is not None:
normal.delete()
# Remove any cached relation on the case and reload so the template
# sees the updated state (no normal_case).
try:
if hasattr(case, "normal_case"):
delattr(case, "normal_case")
except Exception:
pass
case = get_case_for_case_detail(pk)
else:
# Attempt to auto-extract age from DICOM
age = None
try:
from datetime import datetime
import re
# Iterate series then images to find first usable DICOM
for series in case.series.all():
img = series.images.filter(removed=False).first()
if not img:
continue
ds = img.get_dicom_data()
if not ds:
continue
# Try PatientAge (format often like '045Y' or '45Y')
try:
pa = getattr(ds, "PatientAge", None)
except Exception:
pa = None
if pa:
s = str(pa)
m = re.search(r"(\d+)", s)
if m:
age = int(m.group(1))
break
# Try birthdate + studydate
try:
pbd = getattr(ds, "PatientBirthDate", None)
sdate = getattr(ds, "StudyDate", None) or getattr(ds, "AcquisitionDate", None) or getattr(ds, "SeriesDate", None)
except Exception:
pbd = None
sdate = None
if pbd and sdate:
try:
bd = datetime.strptime(str(pbd), "%Y%m%d")
sd = datetime.strptime(str(sdate), "%Y%m%d")
years = round((sd - bd).days / 365.25)
age = int(years)
break
except Exception:
pass
except Exception:
age = None
# Create NormalCase
from .models import NormalCase
# Convert inferred integer years -> canonical days (approx)
if age is not None:
age_days = NormalCase.to_days(age, NormalCase.AGE_UNIT_YEARS)
else:
age_days = None
NormalCase.objects.create(case=case, age_days=age_days, added_by=request.user)
# Render the updated fragment
# Recompute flags for the fragment render
is_atlas_editor = request.user.is_superuser or request.user.groups.filter(name="atlas_editor").exists()
is_case_author = case.author.filter(id=request.user.id).exists()
can_mark_normal = is_atlas_editor or is_case_author or request.user.is_superuser
html = render_to_string(
"atlas/partials/_normal_toggle.html",
{"case": case, "user": request.user, "is_atlas_editor": is_atlas_editor, "is_case_author": is_case_author, "can_mark_normal": can_mark_normal},
request=request,
)
return HttpResponse(html)
@login_required
def case_normal_form(request, pk):
"""Return an HTMX modal with a NormalCaseForm prepopulated with auto-extracted values."""
case = get_case_for_case_detail(pk)
# Permission check: allow atlas editors, case authors, or superusers
is_atlas_editor = request.user.is_superuser or request.user.groups.filter(name="atlas_editor").exists()
is_case_author = case.author.filter(id=request.user.id).exists()
can_mark_normal = is_atlas_editor or is_case_author or request.user.is_superuser
if not can_mark_normal:
return HttpResponse(status=403)
# Extract defaults
age = None
inferred_examination = None
inferred_modality = None
try:
for series in case.series.all():
img = series.images.filter(removed=False).first()
if not img:
continue
ds = img.get_dicom_data()
if not ds:
continue
pa = getattr(ds, "PatientAge", None)
if pa:
import re
m = re.search(r"(\d+)", str(pa))
if m:
age = int(m.group(1))
break
pbd = getattr(ds, "PatientBirthDate", None)
sdate = getattr(ds, "StudyDate", None) or getattr(ds, "AcquisitionDate", None) or getattr(ds, "SeriesDate", None)
if pbd and sdate:
from datetime import datetime
try:
bd = datetime.strptime(str(pbd), "%Y%m%d")
sd = datetime.strptime(str(sdate), "%Y%m%d")
years = round((sd - bd).days / 365.25)
age = int(years)
break
except Exception:
pass
# Try inferring exam/modality from first series
first_series = case.series.first()
if first_series is not None:
inferred_examination = getattr(first_series, "examination", None)
inferred_modality = getattr(first_series, "modality", None)
except Exception:
pass
# Build form with initial values
from .forms import NormalCaseForm
# Map inferred age (years) into age_value/unit defaults. If we inferred
# a fractional year (from birthdate), prefer months for ages < 2 years.
from .models import NormalCase
inferred_age_value = None
inferred_age_unit = None
try:
if age is not None:
# age currently holds integer years; for small ages prefer months
if age < 2:
inferred_age_value = int(age * 12)
inferred_age_unit = NormalCase.AGE_UNIT_MONTHS
else:
inferred_age_value = int(age)
inferred_age_unit = NormalCase.AGE_UNIT_YEARS
except Exception:
inferred_age_value = None
inferred_age_unit = None
form = NormalCaseForm(initial={
"age_value": inferred_age_value,
"age_unit": inferred_age_unit,
"examination": inferred_examination,
"modality": inferred_modality,
})
# Log the rendered form HTML so we can verify inputs/names reach the client
from loguru import logger
try:
form_html = form.as_p()
logger.debug("case_normal_form rendered form HTML: {}", form_html)
except Exception as e:
logger.exception("Failed to render form.as_p(): {}", e)
html = render_to_string("atlas/partials/_normal_form.html", {"form": form, "case": case}, request=request)
logger.debug("case_normal_form returning HTML length: {}", len(html))
return HttpResponse(html)
@login_required
@require_POST
def create_case_normal(request, pk):
case = get_case_for_case_detail(pk)
# Permission check: allow atlas editors, case authors, or superusers
is_atlas_editor = request.user.is_superuser or request.user.groups.filter(name="atlas_editor").exists()
is_case_author = case.author.filter(id=request.user.id).exists()
can_mark_normal = is_atlas_editor or is_case_author or request.user.is_superuser
if not can_mark_normal:
return HttpResponse(status=403)
from .forms import NormalCaseForm
from loguru import logger
form = NormalCaseForm(request.POST)
logger.debug("create_case_normal POST data: {}", dict(request.POST))
if form.is_valid():
logger.debug("NormalCaseForm.cleaned_data: {}", form.cleaned_data)
nc = form.save(commit=False)
nc.case = case
nc.added_by = request.user
nc.save()
# Ensure the case reflects the newly created NormalCase when rendering
try:
if hasattr(case, "normal_case"):
delattr(case, "normal_case")
except Exception:
pass
case = get_case_for_case_detail(pk)
# Return the updated toggle fragment and close modal via inline script
is_atlas_editor = request.user.is_superuser or request.user.groups.filter(name="atlas_editor").exists()
is_case_author = case.author.filter(id=request.user.id).exists()
can_mark_normal = is_atlas_editor or is_case_author or request.user.is_superuser
toggle_html = render_to_string(
"atlas/partials/_normal_toggle.html",
{"case": case, "user": request.user, "is_atlas_editor": is_atlas_editor, "is_case_author": is_case_author, "can_mark_normal": can_mark_normal},
request=request,
)
toggle_html += "<script>var m = bootstrap.Modal.getOrCreateInstance(document.getElementById('normal-modal-{}')); if(m) m.hide();</script>".format(case.pk)
return HttpResponse(toggle_html)
else:
# On form errors, re-render the modal with errors
html = render_to_string("atlas/partials/_normal_form.html", {"form": form, "case": case}, request=request)
return HttpResponse(html, status=400)
@login_required
@user_is_author_or_atlas_series_checker_or_atlas_marker_or_open_access
def series_thumbnail_fail(request, pk):
return series_thumbnail(request, pk=pk, fail_loudly=True)
@login_required
@user_is_author_or_atlas_series_checker_or_atlas_marker_or_open_access
def series_thumbnail(request, pk, fail_loudly=False):
series = get_object_or_404(Series, pk=pk)
thumbnail, _ = series.get_thumbnail(recreate=True, fail_loudly=fail_loudly)
return HttpResponse(thumbnail)
@login_required
@user_is_author_or_atlas_series_checker_or_atlas_marker_or_open_access
def series_viewer(request, pk, finding_pk=None):
series = get_object_or_404(Series, pk=pk)
context = {
"series": series,
}
return render(
request,
"atlas/series_viewer_test.html",
context,
)
@login_required
@user_is_author_or_atlas_series_checker_or_atlas_marker_or_open_access
def series_detail(request, pk, finding_pk=None):
series = get_object_or_404(Series, pk=pk)
can_edit = series.check_user_can_edit(request.user)
editing_finding = -1
if finding_pk is not None:
finding = get_object_or_404(SeriesFinding, pk=finding_pk)
series_finding_form = SeriesFindingForm(None, instance=finding)
editing_finding = finding.pk
else:
series_finding_form = SeriesFindingForm(series_id=series.id)
context = {
"series": series,
"series_finding_form": series_finding_form,
"editing_finding": editing_finding,
"can_edit": can_edit,
}
if can_edit:
context["available_cases"] = Case.objects.filter(author__id=request.user.id) # Adjust the queryset as needed
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/series.html",
context,
)
@login_required
@user_is_author_or_atlas_series_checker_or_atlas_marker_or_open_access
def series_images_partial(request, pk):
"""Return a small fragment that lists images for a series. Intended for HTMX."""
series = get_object_or_404(Series, pk=pk)
logger.debug(series)
images = series.images.filter(removed=False).order_by("pk")
logger.debug(images)
return render(request, "atlas/partials/series_images.html", {"series": series, "images": images})
@login_required
# @user_is_atlas_editor
def question_schema_detail(request, pk: int):
question_schema = get_object_or_404(QuestionSchema, pk=pk)
example_form = question_schema.get_example_form()
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/question_schema_detail.html",
{"question_schema": question_schema, "example_form": example_form},
)
@login_required
# @user_is_atlas_editor
def question_schema_schemas(request):
question_schemas = QuestionSchema.objects.all()
return render(
request,
"atlas/question_schemas_preset.html",
{
"question_schemas": question_schemas,
},
)
@login_required
# @user_is_atlas_editor
def condition_detail(request, pk):
condition = get_object_or_404(Condition, pk=pk)
# form used for the merge UI
merge_form = ConditionAutocompleteForm()
# form used to add a synonym (can post either an existing condition via autocomplete
# or provide a plain name in 'synonym_name')
synonym_form = ConditionAutocompleteForm()
can_merge = False
if (
request.user.is_superuser
or request.user.groups.filter(name="atlas_editor").exists()
):
can_merge = True
# Handle POST to add a synonym
if request.method == "POST" and request.POST.get("action") == "add_synonym":
# only allow staff/editors
if not (
request.user.is_superuser
or request.user.groups.filter(name="atlas_editor").exists()
):
return HttpResponse("Forbidden", status=403)
# Try autocomplete-selected condition first
form = ConditionAutocompleteForm(request.POST)
other = None
if form.is_valid():
other = form.cleaned_data.get("condition")
# keep the form (possibly with errors) to render back in HTMX responses
synonym_form = form
# If no autocomplete selection, accept a free-text name
synonym_name = request.POST.get("synonym_name")
if not other and synonym_name:
other, created = Condition.objects.get_or_create(name=synonym_name)
if created:
# created condition: treat as alias by setting canonical below
pass
if other and other != condition:
# Option B: treat synonym as alias/canonical relationship.
# Determine canonical/master for the current condition
master = condition.canonical if condition.canonical else condition
# If `other` belongs to a different canonical group, move its group
# to point at our master. If `other` has no canonical, set it.
if other.canonical:
# move the whole other group (canonical + aliases) to our master
group_master = other.canonical
members = Condition.objects.filter(Q(canonical=group_master) | Q(pk=group_master.pk))
for m in members:
if m.pk != master.pk:
m.canonical = master
m.save()
else:
# other is standalone; make it an alias of our master
other.canonical = master
other.save()
# Master is indicated by having no `canonical` pointer. Ensure
# it remains canonical (no save required unless changed elsewhere).
# If this was an HTMX request return the updated synonyms partial so the
# page can be updated without a full redirect. Otherwise perform the
# existing redirect back to the detail page.
if request.htmx:
html = render_to_string(
"atlas/partials/_synonyms_list.html",
{"condition": condition, "synonym_form": synonym_form, "can_merge": can_merge},
request=request,
)
return HttpResponse(html)
return redirect("atlas:condition_detail", pk=condition.pk)
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/condition_detail.html",
{
"condition": condition,
"form": merge_form,
"synonym_form": synonym_form,
"can_merge": can_merge,
},
)
# class GetConditionAutocomplete(HTMXAutoComplete):
# name = "conditions"
# multiselect = False
#
# class Meta:
# model = Condition
@login_required
@user_is_atlas_editor
def condition_merge(request, pk_to_merge):
if not request.htmx:
raise Http404
condition_to_merge = get_object_or_404(Condition, pk=pk_to_merge)
# condition_to_merge_into = get_object_or_404(Condition, pk=pk_to_merge_into)
print(condition_to_merge.case_set)
if condition_to_merge.rcr_curriculum:
return HttpResponse("You cannot merge rcr curriculum items")
condition_to_merge_into = request.POST.get("conditions")
if not condition_to_merge_into:
return HttpResponse("You need to select a condition to merge into.")
condition_to_merge_into = get_object_or_404(Condition, pk=condition_to_merge_into)
if condition_to_merge == condition_to_merge_into:
return HttpResponse("You cannot merge into the same condition.")
# move all relevant relationships
cases = condition_to_merge.case_set.all().values_list("id", flat=True)
condition_to_merge_into.case_set.add(cases)
# cases = Case.objects.filter(condition=condition_to_merge)
# cases.update(condition=condition_to_merge_into)
differentials = Differential.objects.filter(condition=condition_to_merge)
differentials.update(condition=condition_to_merge_into)
return HttpResponse("Merged")
# conditions = Condition.objects.all()
# return render(request, "atlas/condition_options.html", {"conditions": conditions})
@login_required
# @user_is_atlas_editor
def subspecialty_detail(request, pk):
subspecialty = get_object_or_404(Subspecialty, pk=pk)
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/subspecialty_detail.html",
{
"subspecialty": subspecialty,
},
)
@login_required
# @user_is_atlas_editor
def presentation_detail(request, pk):
presentation = get_object_or_404(Presentation, pk=pk)
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/presentation_detail.html",
{
"presentation": presentation,
},
)
@login_required
# @user_is_atlas_editor
def pathological_process_detail(request, pk):
pathological_process = get_object_or_404(PathologicalProcess, pk=pk)
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/pathological_process_detail.html",
{
"pathological_process": pathological_process,
},
)
@login_required
# @user_is_atlas_editor
def structure_detail(request, pk):
structure = get_object_or_404(Structure, pk=pk)
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/structure_detail.html",
{
"structure": structure,
},
)
@login_required
# @user_is_atlas_editor
def finding_detail(request, pk):
finding = get_object_or_404(Finding, pk=pk)
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"atlas/finding_detail.html",
{
"finding": finding,
},
)
@login_required
def resource_detail(request, pk):
resource = get_object_or_404(Resource, pk=pk)
return render(request, "atlas/resource_detail.html", {"resource": resource})
@login_required
def resource_view(request, pk):
resource = get_object_or_404(Resource, pk=pk)
return HttpResponse(resource.get_display())
@login_required
def author_detail(request, pk):
# logging.debug(Author.objects.all())
# author = get_object_or_404(Author, pk=pk)
author = User.objects.get(pk=pk)
if request.user.pk == pk:
atlas = Case.objects.filter(author=pk)
else:
atlas = Case.objects.filter(author=pk, open_access=True)
return render(
request, "atlas/category_detail.html", {"category": author, "atlas": atlas}
)
@user_is_author_or_atlas_editor
def author_list(request):
authors = User.objects.all()
return render(request, "atlas/author_list.html", {"authors": authors})
@login_required
def index(request):
return render(request, "atlas/index.html", {})
@login_required
def user_collections(request):
# Collections the user is explicitly allowed to take (via CaseCollection.valid_user_users)
available_collections = request.user.user_casecollection_exams.all()
# Collections the user has already taken (recorded in generic.CidUserExam)
in_progress_ids = []
finished_ids = []
try:
ct = ContentType.objects.get_for_model(CaseCollection)
taken_exams = CidUserExam.objects.filter(user_user=request.user, content_type=ct)
in_progress_ids = list(
taken_exams.filter(completed=False).values_list("object_id", flat=True).distinct()
)
finished_ids = list(
taken_exams.filter(completed=True).values_list("object_id", flat=True).distinct()
)
except Exception:
# If anything goes wrong determining taken/finished exams, treat as none
in_progress_ids = []
finished_ids = []
# Build querysets for the three groups
taken_all_ids = set(in_progress_ids) | set(finished_ids)
# Available to start: explicitly allowed collections that the user hasn't taken at all
available_qs = available_collections.exclude(pk__in=taken_all_ids).order_by("name")
# In-progress (taken but not completed)
in_progress_qs = CaseCollection.objects.filter(pk__in=list(in_progress_ids)).order_by("name")
# Finished (completed)
finished_qs = CaseCollection.objects.filter(pk__in=list(finished_ids)).order_by("name")
context = {
"available_collections": available_qs,
"in_progress_collections": in_progress_qs,
"finished_collections": finished_qs,
}
# If called via HTMX return a small fragment suitable for swapping
# into the index page. Otherwise render the full page.
is_htmx = getattr(request, "htmx", False) or request.headers.get("HX-Request", "").lower() == "true"
if is_htmx:
html = render_to_string("atlas/partials/_user_collections.html", context, request=request)
return HttpResponse(html)
return render(request, "atlas/user_collections.html", context)
# Note: keep the above full-page render for normal requests. If this
# endpoint is called by HTMX we prefer returning a small fragment so it
# can be swapped into the index page without duplicating layout.
@login_required
def user_related_collections(request):
"""Return up to 10 most recently modified collections where the user
is an author or a marker. If called via HTMX return a fragment suitable
for swapping into the index page.
"""
user = request.user
qs = (
CaseCollection.objects.filter(Q(author=user) | Q(markers=user))
.distinct()
.order_by("-updated")[:10]
)
related = []
for c in qs:
related.append(
{
"collection": c,
"is_author": c.author.filter(id=user.id).exists(),
"is_marker": c.markers.filter(id=user.id).exists(),
}
)
context = {"related_collections": related}
is_htmx = getattr(request, "htmx", False) or request.headers.get("HX-Request", "").lower() == "true"
if is_htmx:
html = render_to_string("atlas/partials/_user_related_collections.html", context, request=request)
return HttpResponse(html)
return render(request, "atlas/user_collections.html", context)
@login_required
def collection_options(request):
if not request.htmx:
return Http404
collections = CaseCollection.objects.filter(author=request.user).order_by("name")
html = render_to_string("atlas/partials/collection_options.html", {"collections": collections}, request=request)
return HttpResponse(html)
@login_required
def collection_case_jump_partial(request, pk: int):
"""Return an HTMX partial containing a jump list for cases in the collection.
Expects optional GET params:
- current: the current case index (0-based) to mark active
- cid, passcode: optional candidate identifiers so links include them when present
"""
# Allow non-HTMX callers to still get the fragment (useful for testing)
collection = get_object_or_404(CaseCollection, pk=pk)
casedetails = (
CaseDetail.objects.filter(collection=collection)
.select_related('case')
.order_by('sort_order')
)
try:
current = int(request.GET.get('current', 0) or 0)
except Exception:
current = 0
cid = request.GET.get('cid')
passcode = request.GET.get('passcode')
html = render_to_string(
'atlas/partials/_collection_case_jump.html',
{
'collection': collection,
'casedetails': casedetails,
'current': current,
'cid': cid,
'passcode': passcode,
},
request=request,
)
return HttpResponse(html)
@login_required
def remove_cases_from_collection(request):
if not request.htmx:
return Http404
case_ids = request.POST.getlist('selection')
collection_id = request.POST.get('collection_id')
if not case_ids or not collection_id:
return HttpResponse("No cases selected or collection ID provided.")
collection = get_object_or_404(CaseCollection, pk=collection_id)
if request.user not in collection.author.all():
return HttpResponse("You do not have permission to remove cases from this collection.")
cases = Case.objects.filter(pk__in=case_ids)
for case_ in cases:
collection.cases.remove(case_)
return HttpResponse(f"Removed {len(cases)} cases from collection {collection.name} (refresh to see)")
@login_required
def add_cases_to_collection(request):
if not request.htmx:
return Http404
case_ids = request.POST.getlist('selection')
collection_id = request.POST.get('collection_id')
if not case_ids or not collection_id:
return HttpResponse("No cases selected or collection ID provided.")
collection = get_object_or_404(CaseCollection, pk=collection_id)
if request.user not in collection.author.all():
return HttpResponse("You do not have permission to add cases to this collection.")
cases = Case.objects.filter(pk__in=case_ids)
for case_ in cases:
collection.cases.add(case_)
return HttpResponse(f"Added {len(cases)} cases to collection {collection.name} (refresh to see)")
@login_required
def set_open_access(request):
if not request.htmx:
return Http404
case_ids = request.POST.getlist("selection")
if not case_ids:
return HttpResponse("No case ID(s) provided.")
cases = Case.objects.filter(pk__in=case_ids)
for case in cases:
# This will not let give accurate permission errors if the user does not have permission to edit any of the cases
# but it is simpler to just check the first case.
if not case.check_user_can_edit(request.user):
return HttpResponse(f"You do not have permission to set case {case.name} as open access.")
case.open_access = True
case.save()
return HttpResponse(f"Cases {', '.join([case.title for case in cases])} set to open access (refresh to see changes).")
@login_required
def remove_open_access(request):
if not request.htmx:
return Http404
case_ids = request.POST.getlist("selection")
if not case_ids:
return HttpResponse("No case ID(s) provided.")
cases = Case.objects.filter(pk__in=case_ids)
for case in cases:
# This will not let give accurate permission errors if the user does not have permission to edit any of the cases
# but it is simpler to just check the first case.
if not case.check_user_can_edit(request.user):
return HttpResponse(f"You do not have permission to remove open access from case {case.name}.")
case.open_access = False
case.save()
return HttpResponse(f"Cases {', '.join([case.title for case in cases])} removed from open access (refresh to see changes).")
def add_case_to_collection(request, collection_id):
if not request.htmx:
return Http404
collection = get_object_or_404(CaseCollection, pk=collection_id)
if request.method == "POST":
case = get_object_or_404(Case, pk=request.POST["case"])
if request.user not in collection.author.all():
return HttpResponse("You do not have permission to add cases to this collection.")
if case in collection.cases.all():
return HttpResponse("Case already in collection")
print(case)
collection.add_case(case)
# Render the new case list item and return it so HTMX can append it
try:
casedetail = collection.casedetail_set.get(case=case)
except Exception:
return HttpResponse("Case added, but could not render item.")
total = collection.casedetail_set.count()
can_edit = request.user.is_superuser or collection.author.filter(pk=request.user.pk).exists()
rendered = render_to_string(
"atlas/partials/case_list_item.html",
{
"casedetail": casedetail,
"collection": collection,
"forloop": {"counter": total, "counter0": max(0, total - 1)},
"can_edit": can_edit,
},
request=request,
)
# Also instruct the client to remove this case from any visible search/recent lists
remove_script = (
'<script>'
'(function(){'
'try {'
"var pk = '%s';" % case.pk +
"document.querySelectorAll('#case-search-results [data-case-pk=\"' + pk + '\"]').forEach(function(el){ el.remove(); });" +
"document.querySelectorAll('#case-search-results input[name=\"case\"][value=\"' + pk + '\"]').forEach(function(inp){ var li = inp.closest('.list-group-item'); if (li) li.remove(); });" +
'} catch(e) { console.error("remove added case from search error", e); }'
'})();'
'</script>'
)
return HttpResponse(rendered + remove_script)
# if request.method == "POST":
# don't show search results on initial load; the search partial will render only when a query exists
cases = Case.objects.none()
# Also compute recent cases for initial display (authored by current user and not in collection)
recent_qs = get_cases_available_to_user(request.user).filter(author=request.user)
recent_qs = recent_qs.exclude(pk__in=collection.cases.values_list("pk", flat=True))
recent_cases = recent_qs.order_by("-created_date")[:5]
return render(
request,
"atlas/add_case_to_collection.html",
{"cases": cases, "collection": collection, "recent_cases": recent_cases, "q": ""},
)
@login_required
def case_search(request):
# HTMX endpoint for searching cases available to the user
if not request.htmx:
return Http404
q = request.GET.get("q", "").strip()
collection_id = request.GET.get("collection")
collection = None
if collection_id:
try:
collection = CaseCollection.objects.get(pk=collection_id)
except CaseCollection.DoesNotExist:
collection = None
qs = get_cases_available_to_user(request.user, collection=collection)
if q:
qs = qs.filter(title__icontains=q)
# Recent cases: last 5 created that are available to the user and not already in the collection
# Restrict recent cases to those authored by the current user and not already in the collection
recent_qs = get_cases_available_to_user(request.user).filter(author=request.user)
if collection is not None:
recent_qs = recent_qs.exclude(pk__in=collection.cases.values_list("pk", flat=True))
recent_cases = recent_qs.order_by("-created_date")[:5]
# Exclude recent cases from the main search results to avoid duplication
recent_pks = [c.pk for c in recent_cases]
# Build the base queryset for results, apply excludes first (before slicing)
cases_qs = qs.exclude(pk__in=recent_pks)
# Also exclude any cases that are already present in the collection
if collection is not None:
existing_pks = list(collection.cases.values_list("pk", flat=True))
if existing_pks:
cases_qs = cases_qs.exclude(pk__in=existing_pks)
cases = cases_qs.order_by("title")[:30]
return render(
request,
"atlas/partials/case_search_results.html",
{"cases": cases, "collection": collection, "recent_cases": recent_cases, "q": q},
)
@login_required
def user_uploads_series(request, series_instance_uid: str):
# We don't restrict to teh uploading user here. Should we?
dicoms = UncategorisedDicom.objects.filter(series_instance_uid=series_instance_uid)
if not dicoms:
raise Http404("Series not found")
return render(
request,
"atlas/user_upload_viewer.html",
{"dicoms": dicoms, "series_id": series_instance_uid},
)
@login_required
def new_uploads(request):
return render(request, "atlas/new_uploads.html", {})
@login_required
@user_passes_test(lambda u: u.is_superuser)
def all_uploads(request, case_id: int | None = None):
return user_uploads(request, case_id, user_only=False)
@user_passes_test(lambda u: u.is_superuser)
def other_user_uploads(request, user_pk: int, case_id: int | None = None):
return user_uploads(request, case_id, user_only=False, user_pk=user_pk)
@login_required
def user_uploads(
request,
case_id: int | None = None,
user_only: bool = True,
user_pk: int | None = None,
):
user = request.user
if user_only:
dicoms = UncategorisedDicom.objects.filter(user=request.user)
elif user_pk is not None:
user = get_object_or_404(User, pk=user_pk)
dicoms = UncategorisedDicom.objects.filter(user=user)
else:
dicoms = UncategorisedDicom.objects.filter()
data = defaultdict(list)
for d in dicoms:
tags = d.basic_dicom_tags
if tags is None:
d.save()
tags = d.basic_dicom_tags
data[tags["SeriesInstanceUID"]].append((tags, d.created_date))
series_list = []
for series in data:
tags, date = data[series][0]
series_list.append((series, len(data[series]), tags, date))
series_list.sort(key=lambda l: l[-1])
case = False
if case_id is not None:
case = get_object_or_404(Case, pk=case_id)
return render(
request,
"atlas/user_uploads.html",
{"series_list": series_list, "case": case, "user": user},
)
class CaseDelete(RevisionMixin, AuthorOrCheckerRequiredMixin, DeleteView):
model = Case
success_url = reverse_lazy("atlas:case_view")
template_name = "confirm_delete.html"
# TODO fix permissions
class SeriesFindingDelete(RevisionMixin, PermissionRequiredMixin, DeleteView):
permission_required = "atlas.delete_seriesfinding"
model = SeriesFinding
template_name = "confirm_delete.html"
# success_url = reverse_lazy("atlas:case_view")
def get_object(self):
series_finding = super().get_object()
series = series_finding.series
if series.check_user_can_edit(self.request.user):
return series_finding
else:
raise PermissionDenied
def get_success_url(self):
pk = self.kwargs["pk"]
series_id = get_object_or_404(SeriesFinding, pk=pk).series.id
return reverse("atlas:series_detail", kwargs={"pk": series_id})
class SeriesDelete(RevisionMixin, AtlasEditorRequiredMixin, DeleteView):
model = Series
template_name = "confirm_delete.html"
success_url = reverse_lazy("atlas:series_view")
class CaseCollectionDelete(RevisionMixin, AtlasEditorRequiredMixin, DeleteView):
model = CaseCollection
template_name = "confirm_delete.html"
success_url = reverse_lazy("atlas:collection_view")
class ConditionDelete(RevisionMixin, AtlasEditorRequiredMixin, DeleteView):
model = Condition
template_name = "confirm_delete.html"
success_url = reverse_lazy("atlas:condition_view")
class FindingDelete(RevisionMixin, AtlasEditorRequiredMixin, DeleteView):
model = Finding
template_name = "confirm_delete.html"
success_url = reverse_lazy("atlas:finding_view")
class StructureDelete(RevisionMixin, AtlasEditorRequiredMixin, DeleteView):
model = Structure
template_name = "confirm_delete.html"
success_url = reverse_lazy("atlas:structure_view")
class ResourceDelete(LoginRequiredMixin, AuthorOrCheckerRequiredMixin, DeleteView):
model = Resource
template_name = "confirm_delete.html"
success_url = reverse_lazy("atlas:resource_view")
class ResourceCreate(LoginRequiredMixin, CreateView):
model = Resource
form_class = ResourceForm
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
return super().form_valid(form)
class ResourceView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Resource
table_class = ResourceTable
template_name = "atlas/resource_list.html"
filterset_class = ResourceFilter
def get_queryset(self):
# Base queryset
qs = Resource.objects.all()
# Atlas editors and superusers see everything
if self.request.user.is_superuser or self.request.user.groups.filter(name="atlas_editor").exists():
return qs
# Otherwise only show open access resources or those authored by the user
return qs.filter(Q(open_access=True) | Q(author__id=self.request.user.id)).distinct()
class CaseCollectionCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = CaseCollection
form_class = CaseCollectionForm
template_name = "atlas/casecollection_form.html"
def get_initial(self):
# print(self.request)
if "pk" in self.kwargs:
initial = super(CaseCollectionCreate, self).get_initial()
# case = get_object_or_404(Case, pk=self.kwargs["pk"])
# initial["case"] = case.id
return initial
def get_form_kwargs(self):
kwargs = super(CaseCollectionCreate, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(CaseCollectionCreate, self).get_context_data(**kwargs)
# Only bind the case formset when the POST actually contains formset data.
# We don't require adding cases on create, so if the template doesn't
# include the casedetail management fields we should skip validation.
if self.request.POST and "casedetail_set-TOTAL_FORMS" in self.request.POST:
context["case_formset"] = CaseCollectionCaseFormSet(
self.request.POST,
self.request.FILES,
form_kwargs={"user": self.request.user},
)
context["case_formset"].full_clean()
else:
context["case_formset"] = CaseCollectionCaseFormSet(
form_kwargs={"user": self.request.user}
)
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
case_formset = context["case_formset"]
# If the POST included case formset data, validate and save it. Otherwise
# skip formset processing so creating a collection without cases redirects
# normally to the detail view.
if self.request.POST and "casedetail_set-TOTAL_FORMS" in self.request.POST:
if case_formset.is_valid():
response = super().form_valid(form)
case_formset.instance = self.object
case_formset.save()
return response
else:
return super().form_invalid(form)
# No formset data present: proceed with normal create redirect.
return super().form_valid(form)
def get_success_url(self):
"""Redirect to the collection detail page after successful creation."""
return reverse("atlas:collection_detail", kwargs={"pk": self.object.pk})
class SeriesCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = Series
form_class = SeriesForm
def get_initial(self):
# print(self.request)
if "pk" in self.kwargs:
initial = super().get_initial()
case = get_object_or_404(Case, pk=self.kwargs["pk"])
initial["case"] = case.id
return initial
def get_form_kwargs(self):
kwargs = super(SeriesCreate, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(SeriesCreate, self).get_context_data(**kwargs)
if self.request.POST:
context["image_formset"] = SeriesImageFormSet(
self.request.POST, self.request.FILES
)
context["image_formset"].full_clean()
else:
context["image_formset"] = SeriesImageFormSet()
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
series_formset = context["image_formset"]
if series_formset.is_valid():
response = super().form_valid(form)
series_formset.instance = self.object
series_formset.save()
return response
else:
return super().form_invalid(form)
class CaseCollectionUpdate(RevisionMixin, AuthorOrCheckerRequiredMixin, UpdateView):
model = CaseCollection
form_class = CaseCollectionForm
template_name = "atlas/casecollection_form.html"
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
def get_form_kwargs(self):
kwargs = super(CaseCollectionUpdate, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(CaseCollectionUpdate, self).get_context_data(**kwargs)
#if self.request.POST:
# context["case_formset"] = CaseCollectionCaseFormSet(
# self.request.POST,
# self.request.FILES,
# instance=self.object,
# form_kwargs={"user": self.request.user, "collection": self.object},
# )
# context["case_formset"].full_clean()
#else:
# context["case_formset"] = CaseCollectionCaseFormSet(
# instance=self.object,
# form_kwargs={"user": self.request.user, "collection": self.object},
# )
context["collection"] = context["casecollection"]
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
return super().form_valid(form)
context = self.get_context_data(form=form)
case_formset = context["case_formset"]
# logger.debug(formset.is_valid())
if case_formset.is_valid():
response = super().form_valid(form)
case_formset.instance = self.object
case_formset.save()
# restore exam orders
# for exam in self.object.exams.all():
# if exam in exam_orders and self.object in exam_orders[exam]:
# print(exam_orders[exam])
# exam.exam_questions.set(exam_orders[exam])
# exam.save()
return response
else:
return super().form_invalid(form)
class SeriesUpdate(
RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView
):
model = Series
form_class = SeriesForm
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
def get_context_data(self, **kwargs):
context = super(SeriesUpdate, self).get_context_data(**kwargs)
if self.request.POST:
context["image_formset"] = SeriesImageFormSet(
self.request.POST, self.request.FILES, instance=self.object
)
context["image_formset"].full_clean()
else:
context["image_formset"] = SeriesImageFormSet(instance=self.object)
return context
def get_form_kwargs(self):
kwargs = super(SeriesUpdate, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
image_formset = context["image_formset"]
# logger.debug(formset.is_valid())
if image_formset.is_valid():
response = super().form_valid(form)
image_formset.instance = self.object
image_formset.save()
return response
else:
return super().form_invalid(form)
class ResourceUpdate(LoginRequiredMixin, UpdateView):
model = Resource
form_class = ResourceForm
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
return super().form_valid(form)
class ConditionUpdate(RevisionMixin, LoginRequiredMixin, UpdateView):
model = Condition
form_class = ConditionForm
class ConditionCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = Condition
form_class = ConditionForm
class QuestionSchemaCreate(LoginRequiredMixin, CreateView):
model = QuestionSchema
form_class = QuestionSchemaForm
class QuestionSchemaUpdate(AuthorOrCheckerRequiredMixin, UpdateView):
model = QuestionSchema
form_class = QuestionSchemaForm
class QuestionSchemaDelete(AuthorOrCheckerRequiredMixin, DeleteView):
model = QuestionSchema
template_name = "confirm_delete.html"
class FindingUpdate(RevisionMixin, LoginRequiredMixin, UpdateView):
model = Finding
form_class = FindingForm
class FindingCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = Finding
form_class = FindingForm
class StructureUpdate(RevisionMixin, LoginRequiredMixin, UpdateView):
model = Structure
form_class = StructureForm
class StructureCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = Structure
form_class = StructureForm
class SubspecialtyCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = Subspecialty
form_class = SubspecialtyForm
template_name = "atlas/subspecialty_form.html"
class PresentationCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = Presentation
form_class = PresentationForm
template_name = "atlas/presentation_form.html"
class PathologicalProcessCreate(RevisionMixin, LoginRequiredMixin, CreateView):
model = PathologicalProcess
form_class = PathologicalProcessForm
template_name = "atlas/pathological_process_form.html"
class AtlasCreateBase(RevisionMixin, LoginRequiredMixin):
model = Case
form_class = CaseForm
def get_form_kwargs(self):
kwargs = super(AtlasCreateBase, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(AtlasCreateBase, self).get_context_data(**kwargs)
if self.request.POST:
#context["series_formset"] = SeriesFormSet(
# self.request.POST,
# self.request.FILES,
# form_kwargs={"user": self.request.user},
#)
context["casedifferential_formset"] = CaseDifferentialFormSet(
self.request.POST, self.request.FILES
)
context["caseresource_formset"] = CaseResourceFormSet(
self.request.POST,
self.request.FILES,
form_kwargs={"user": self.request.user},
)
#context["series_formset"].full_clean()
context["casedifferential_formset"].full_clean()
context["caseresource_formset"].full_clean()
else:
# context["series_formset"] = SeriesFormSet(instance=Case.objects.filter(author__id=self.request.user.id), queryset=Series.objects.filter(author__id=self.request.user.id))
#context["series_formset"] = SeriesFormSet(
# form_kwargs={"user": self.request.user}
#)
context["casedifferential_formset"] = CaseDifferentialFormSet()
context["caseresource_formset"] = CaseResourceFormSet(
form_kwargs={"user": self.request.user}
)
return context
def form_valid(self, form):
self.object = form.save(commit=False)
# Add exam objects
self.object.save()
form.instance.author.add(self.request.user.id)
# Check Cimar Case
# CimarAPI().get_study_by_uuid(self.object.cimar_uuid)
context = self.get_context_data(form=form)
#series_formset = context["series_formset"]
casedifferential_formset = context["casedifferential_formset"]
caseresource_formset = context["caseresource_formset"]
if (
#series_formset.is_valid()
casedifferential_formset.is_valid()
and caseresource_formset.is_valid()
):
response = super().form_valid(form)
#series_formset.instance = self.object
#series_formset.save()
casedifferential_formset.instance = self.object
casedifferential_formset.save()
caseresource_formset.instance = self.object
caseresource_formset.save()
# If the normal submit button is pressed we save as normal
if "submit" in self.request.POST:
return response
# else we redirect to the clone url
else:
return redirect("atlas:case_clone", pk=self.object.pk)
else:
return super().form_invalid(form)
# @login_required
# CaseCreate
class AtlasCreate(AtlasCreateBase, CreateView):
def get_initial(self):
if "pk" in self.kwargs:
initial = super().get_initial()
exam = get_object_or_404(CaseCollection, pk=self.kwargs["pk"])
initial["exams"] = [exam.id]
return initial
## There has to be a better way...
# try:
# s = (i.pk for i in self.request.user.atlas_default.site.all())
# self.initial.update({"site": s})
# except AttributeError:
# pass
# return self.initial
# CaseUpdate
class AtlasUpdate(
RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView
):
model = Case
form_class = CaseForm
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
#def get_object(self, queryset=None):
# # Prefetch all related objects needed for the form and formsets
# qs = self.model.objects.select_related(
# # Add all FK fields here
# ).prefetch_related(
# "series__examination",
# "series__plane",
# "series__contrast",
# "series__images",
# "subspecialty",
# "condition",
# "presentation",
# "pathological_process",
# "differentialcase",
# "caseresource_set",
# )
# return qs.get(pk=self.kwargs["pk"])
def get_form_kwargs(self):
kwargs = super(AtlasUpdate, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(AtlasUpdate, self).get_context_data(**kwargs)
if self.request.POST:
#context["series_formset"] = SeriesFormSet(
# self.request.POST,
# self.request.FILES,
# instance=self.object,
# form_kwargs={"user": self.request.user},
#)
context["casedifferential_formset"] = CaseDifferentialFormSet(
self.request.POST, self.request.FILES, instance=self.object
)
context["caseresource_formset"] = CaseResourceFormSet(
self.request.POST,
self.request.FILES,
instance=self.object,
form_kwargs={"user": self.request.user},
)
#context["series_formset"].full_clean()
context["casedifferential_formset"].full_clean()
context["caseresource_formset"].full_clean()
else:
#context["series_formset"] = SeriesFormSet(
# instance=self.object, form_kwargs={"user": self.request.user}
#)
context["casedifferential_formset"] = CaseDifferentialFormSet(
instance=self.object
)
context["caseresource_formset"] = CaseResourceFormSet(
instance=self.object,
form_kwargs={"user": self.request.user},
)
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
#series_formset = context["series_formset"]
casedifferential_formset = context["casedifferential_formset"]
caseresource_formset = context["caseresource_formset"]
# logger.debug(formset.is_valid())
if (
#series_formset.is_valid()
casedifferential_formset.is_valid()
and caseresource_formset.is_valid()
):
response = super().form_valid(form)
#series_formset.instance = self.object
#series_formset.save()
casedifferential_formset.instance = self.object
casedifferential_formset.save()
caseresource_formset.instance = self.object
caseresource_formset.save()
return response
else:
return super().form_invalid(form)
class CollectionCaseUpdate(
RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView):
model = CaseCollection
form_class = CaseCollectionUpdateCaseForm
template_name = "atlas/casecollection_update_case.html"
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
kwargs.update({"user": self.request.user, "collection": self.object})
return kwargs
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
if self.request.POST:
context["case_formset"] = CaseCollectionCaseFormSet(
self.request.POST,
self.request.FILES,
instance=self.object,
form_kwargs={"user": self.request.user, "collection": self.object},
)
context["case_formset"].full_clean()
else:
context["case_formset"] = CaseCollectionCaseFormSet(
instance=self.object,
form_kwargs={"user": self.request.user, "collection": self.object},
)
context["collection"] = context["casecollection"]
return context
def form_valid(self, form):
context = self.get_context_data(form=form)
case_formset = context["case_formset"]
if case_formset.is_valid():
case_formset.instance = self.object
case_formset.save()
return super().form_valid(form)
else:
return self.form_invalid(form)
class AtlasSeriesUpdate(
RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView
):
model = Case
form_class = CaseUpdateSeriesForm
template_name = "atlas/case_update_series.html"
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
is_atlas_editor = self.request.user.groups.filter(name="atlas_editor").exists() or self.request.user.is_superuser
form_kwargs = {"user": self.request.user, "is_atlas_editor": is_atlas_editor}
if not is_atlas_editor:
queryset = Series.objects.filter(author__id=self.request.user.id)
else:
queryset = Series.objects.all()
queryset = queryset.select_related("modality").prefetch_related(
"examination", "plane", "contrast",
Prefetch("images", queryset=SeriesImage.objects.filter(removed=False).order_by("position"))
)
if self.request.POST:
context["series_formset"] = SeriesFormSet(
self.request.POST,
self.request.FILES,
queryset=queryset,
instance=self.object,
form_kwargs=form_kwargs,
)
context["series_formset"].full_clean()
else:
context["series_formset"] = SeriesFormSet(
queryset=queryset,
instance=self.object,
form_kwargs=form_kwargs,
)
return context
def form_valid(self, form):
context = self.get_context_data(form=form)
series_formset = context["series_formset"]
if series_formset.is_valid():
series_formset.instance = self.object
series_formset.save()
return super().form_valid(form)
else:
return self.form_invalid(form)
def get_object(self, queryset=None):
qs = Case.objects.select_related().prefetch_related(
"seriesdetail_set__series__modality",
"seriesdetail_set__series__examination",
"seriesdetail_set__series__plane",
"seriesdetail_set__series__contrast",
"seriesdetail_set__series__images",
)
return qs.get(pk=self.kwargs["pk"])
def add_collection_to_case_form(request, case_id):
if request.method == "POST":
form = AddCollectionToCaseForm(request.POST, user=request.user)
if form.is_valid():
case = get_object_or_404(Case, pk=case_id)
case.casecollection_set.add(*form.cleaned_data["casecollection"])
# do_something_with_form_data(form.cleaned_data)
return HttpResponse("Done")
return render(request, "web/htmx_contact_form_confirm.html")
else:
form = AddCollectionToCaseForm(user=request.user)
return render(
request,
"generic_form.html",
{
"form": form,
"target_url": reverse("atlas:case_collection_form", args=(case_id,)),
},
)
# class AddCollectionToCaseView(FormView):
# form_class = AddCollectionToCaseForm
# template_name = "generic_form.html"
#
# def form_valid(self, form):
# form.save()
# return super(AddCollectionToCaseView, self).form_valid(form)
#
# def get_form_kwargs(self):
# kwargs = super(AddCollectionToCaseView, self).get_form_kwargs()
# kwargs.update({"user": self.request.user})
# return kwargs
#
# def get_success_url(self) -> str:
# pk = self.kwargs["pk"]
# return reverse("atlas:case_detail", kwargs={"pk": pk})
class AtlasClone(AtlasCreateBase, AuthorOrCheckerRequiredMixin, CreateView):
"""Clones a existing atlas case"""
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
def get_initial(self):
# print(self.request)
old_object = get_object_or_404(Case, pk=self.kwargs["pk"])
if (
self.request.user.groups.filter(name="atlas_editor").exists()
or self.request.user.is_superuser
):
pass
else:
if self.request.user not in old_object.get_author_objects():
raise PermissionDenied() # or Http404
else:
pass
initial_data = model_to_dict(old_object, exclude=["id", "previous_case", "history", "discussion", "report"])
return initial_data
class AtlasCloneNext(AtlasClone):
"""Extends AtlasClone to set the previous_case field to the case being cloned"""
def get_initial(self):
initial_data = super().get_initial()
initial_data["previous_case"] = self.kwargs["pk"]
return initial_data
@login_required
@user_is_author_or_atlas_editor
def atlas_scrap(request, pk):
try:
atlas = Case.objects.get(pk=pk)
except Case.DoesNotExist:
raise Http404("Case does not exist")
atlas.scrapped = not atlas.scrapped
atlas.save()
return HttpResponseRedirect(reverse("atlas:case_detail", args=(pk,)))
# @login_required
# def edit_abnormality_popup(request):
# instance = get_object_or_404(Abnormality, pk = pk)
# form = AbnormalityForm(request.POST or None)
# if form.is_valid():
# instance = form.save()
# return HttpResponse('<script>opener.closePopup(window, "%s", "%s", "#id_abnormality");</script>' % (instance.pk, instance))
# return render(request, "atlas/create_simple.html", {'form': form})
@csrf_exempt
def get_examination_id(request):
if request.accepts("application/json")():
examination_name = request.GET["examination_name"]
examination_id = Examination.objects.get(name=examination_name).id
data = {
"examination_id": examination_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
class CaseView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Case
table_class = CaseTable
template_name = "atlas/case_view.html"
filterset_class = CaseFilter
@staff_member_required
def cases_by_size(request):
"""Admin view: list cases ordered by total images size to find large cases.
Shows case id, title, number of series, number of images and total size (bytes).
Optional GET parameter `min_size` (bytes) to filter cases larger than that.
"""
min_size = request.GET.get("min_size")
try:
min_size = int(min_size) if min_size else 0
except Exception:
min_size = 0
cases_qs = Case.objects.prefetch_related("series__images")
def human(n):
# simple human readable bytes
for unit in ("B", "KB", "MB", "GB", "TB"):
if n < 1024.0:
return f"{n:.1f}{unit}"
n /= 1024.0
return f"{n:.1f}PB"
rows = []
for c in cases_qs:
try:
total = c.get_total_series_images_size()
except Exception:
total = 0
series_count = c.series.count()
image_count = sum(s.images.filter(removed=False).count() for s in c.series.all())
if total >= min_size:
rows.append({
"case": c,
"total": total,
"series_count": series_count,
"image_count": image_count,
"human_total": human(total),
})
rows.sort(key=lambda r: r["total"], reverse=True)
rows.sort(key=lambda r: r["total"], reverse=True)
paginator = Paginator(rows, 50)
page_num = request.GET.get("page")
page = paginator.get_page(page_num)
def human(n):
# simple human readable bytes
for unit in ("B", "KB", "MB", "GB", "TB"):
if n < 1024.0:
return f"{n:.1f}{unit}"
n /= 1024.0
return f"{n:.1f}PB"
context = {"page": page, "human": human, "min_size": min_size}
return render(request, "atlas/cases_by_size.html", context)
class SeriesView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Series
table_class = SeriesTable
template_name = "atlas/series_view.html"
filterset_class = SeriesFilter
class ConditionView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Condition
table_class = ConditionTable
template_name = "atlas/view.html"
filterset_class = ConditionFilter
def get_context_data(self, **kwargs):
ctx = super().get_context_data(**kwargs)
try:
ctx["create_url"] = reverse(f"atlas:{self.model._meta.model_name}_create")
except NoReverseMatch:
ctx["create_url"] = None
# Provide verbose names to templates without accessing _meta in templates
ctx["model_verbose_name"] = getattr(self.model._meta, "verbose_name", None)
ctx["model_verbose_name_plural"] = getattr(self.model._meta, "verbose_name_plural", None)
return ctx
class QuestionSchemaView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = QuestionSchema
table_class = QuestionSchemaTable
template_name = "atlas/schema_table_view.html"
filterset_class = QuestionSchemaFilter
class SubspecialtyView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Subspecialty
table_class = SubspecialtyTable
template_name = "atlas/view.html"
filterset_class = SubspecialtyFilter
def get_context_data(self, **kwargs):
ctx = super().get_context_data(**kwargs)
try:
ctx["create_url"] = reverse(f"atlas:{self.model._meta.model_name}_create")
except NoReverseMatch:
ctx["create_url"] = None
ctx["model_verbose_name"] = getattr(self.model._meta, "verbose_name", None)
ctx["model_verbose_name_plural"] = getattr(self.model._meta, "verbose_name_plural", None)
return ctx
class PresentationView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Presentation
table_class = PresentationTable
template_name = "atlas/view.html"
filterset_class = PresentationFilter
def get_context_data(self, **kwargs):
ctx = super().get_context_data(**kwargs)
try:
ctx["create_url"] = reverse(f"atlas:{self.model._meta.model_name}_create")
except NoReverseMatch:
ctx["create_url"] = None
ctx["model_verbose_name"] = getattr(self.model._meta, "verbose_name", None)
ctx["model_verbose_name_plural"] = getattr(self.model._meta, "verbose_name_plural", None)
return ctx
class PathologicalProcessView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = PathologicalProcess
table_class = PathologicalProcessTable
template_name = "atlas/view.html"
filterset_class = PathologicalProcessFilter
def get_context_data(self, **kwargs):
ctx = super().get_context_data(**kwargs)
try:
ctx["create_url"] = reverse(f"atlas:{self.model._meta.model_name}_create")
except NoReverseMatch:
ctx["create_url"] = None
ctx["model_verbose_name"] = getattr(self.model._meta, "verbose_name", None)
ctx["model_verbose_name_plural"] = getattr(self.model._meta, "verbose_name_plural", None)
return ctx
class StructureView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Structure
table_class = StructureTable
template_name = "atlas/view.html"
filterset_class = StructureFilter
def get_context_data(self, **kwargs):
ctx = super().get_context_data(**kwargs)
try:
ctx["create_url"] = reverse(f"atlas:{self.model._meta.model_name}_create")
except NoReverseMatch:
ctx["create_url"] = None
ctx["model_verbose_name"] = getattr(self.model._meta, "verbose_name", None)
ctx["model_verbose_name_plural"] = getattr(self.model._meta, "verbose_name_plural", None)
return ctx
class FindingView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
model = Finding
table_class = FindingTable
template_name = "atlas/view.html"
filterset_class = FindingFilter
def get_context_data(self, **kwargs):
ctx = super().get_context_data(**kwargs)
try:
ctx["create_url"] = reverse(f"atlas:{self.model._meta.model_name}_create")
except NoReverseMatch:
ctx["create_url"] = None
return ctx
@login_required
def case_order_dicom(request, pk):
if request.htmx:
case: Case = get_object_or_404(Case, pk=pk)
fail = []
# TODO: don't try to order non stack series
for series in case.series.all():
try:
if request.user in series.get_author_objects():
series.order_by_dicom()
else:
fail.append(f"{series.pk}: invalid permission")
except:
fail.append(f"{series.pk}: error")
if fail:
return HttpResponse(
f"Unable to order series {','.join([str(i) for i in fail])}"
)
else:
return HttpResponse("Series reordered")
raise Http404
@login_required
@user_is_author_or_atlas_series_checker
def series_order_dicom(request, pk):
series = get_object_or_404(Series, pk=pk)
try:
series.order_by_dicom()
except:
return HttpResponse("<h1>Series does not appear to contain dicoms</h1>")
return redirect("atlas:series_detail", pk=pk)
@login_required
def image_diff(request):
if request.htmx:
image_ids = request.POST.getlist("image-id")
if len(image_ids) == 2:
first: SeriesImage = get_object_or_404(SeriesImage, pk=image_ids[0])
second: SeriesImage = get_object_or_404(SeriesImage, pk=image_ids[1])
first_ds = pydicom.dcmread(first.image.file)
# del first_ds.PixelData
# j1 = first_ds.to_json_dict()
f = print_dicom(first_ds, join=False)
second_ds = pydicom.dcmread(second.image.file)
# del second_ds.PixelData
# j2 = second_ds.to_json_dict()
s = print_dicom(second_ds, join=False)
diff = difflib.HtmlDiff().make_table(f, s)
dict_diff = pprint.pformat(compare_dicom_datasets(first_ds, second_ds))
# return HttpResponse(diff)
return HttpResponse(f"<pre>{dict_diff}</pre><br/>{diff}")
else:
ds_set = []
for pk in image_ids:
series_image = get_object_or_404(SeriesImage, pk=pk)
ds_set.append(pydicom.dcmread(series_image.image.file))
dict_diff = pprint.pformat(compare_dicom_datasets(*ds_set))
return HttpResponse(f"<pre>{dict_diff}</pre>")
return HttpResponse("Fail")
@login_required
def series_image_dicom(request, pk):
image = get_object_or_404(SeriesImage, pk=pk)
return render(request, "atlas/series_image_dicom.html", {"image": image})
@login_required
@user_is_author_or_atlas_series_checker
def series_anonymise_dicom(request, pk):
series = get_object_or_404(Series, pk=pk)
series.anonymise_images()
return redirect("atlas:series_detail", pk=pk)
@login_required
@user_is_author_or_atlas_series_checker
def series_order_dicom_instance(request, pk):
series = get_object_or_404(Series, pk=pk)
try:
series.order_by_dicom("InstanceNumber")
except:
return HttpResponse(
"<h1>Series does not appear to contain dicoms (or field InstanceNumber)</h1>"
)
return redirect("atlas:series_detail", pk=pk)
@login_required
@user_is_author_or_atlas_series_checker
def series_order_dicom_SeriesInstanceUID(request, pk):
series = get_object_or_404(Series, pk=pk)
try:
series.order_by_dicom("SeriesInstanceUID")
except:
return HttpResponse(
"<h1>Series does not appear to contain dicoms (or field SeriesInstanceUID)</h1>"
)
return redirect("atlas:series_detail", pk=pk)
@login_required
@user_is_author_or_atlas_series_checker
def series_order_upload_filename(request, pk):
series = get_object_or_404(Series, pk=pk)
series.order_by_upload_filename()
return redirect("atlas:series_detail", pk=pk)
# Vendored in from BaseZipView
class SeriesImagesZipView(SeriesImagesZipViewBase, SuperuserRequiredMixin):
series_object = Series
# At the moment only the series owner / author can delete findings but anyone can create
def create_series_findings(request):
# posts = Post.objects.all()
response_data = {}
if request.POST.get("action") == "post":
series_finding_id = int(request.POST.get("series_finding_id"))
series_id = request.POST.get("series")
findings_ids = json.loads(request.POST.get("findings"))
structure_ids = json.loads(request.POST.get("structures"))
condition_ids = json.loads(request.POST.get("conditions"))
description = request.POST.get("description")
#annotation_json = request.POST.get("annotation_json")
#viewport_json = request.POST.get("viewport_json")
#current_image_id_index = request.POST.get("current_image_id_index")
annotation_json_3d = request.POST.get("annotation_json_3d")
viewer_state_3d = request.POST.get("viewer_state_3d")
series = Series.objects.get(pk=series_id)
findings = Finding.objects.filter(pk__in=findings_ids)
structures = Structure.objects.filter(pk__in=structure_ids)
conditions = Condition.objects.filter(pk__in=condition_ids)
if series_finding_id > 0:
sf = get_object_or_404(SeriesFinding, pk=series_finding_id)
sf.series = series
sf.description = description
#sf.annotation_json = annotation_json
#sf.viewport_json = viewport_json
else:
sf = SeriesFinding.objects.create(
series=series,
description=description,
#annotation_json=annotation_json,
#viewport_json=viewport_json,
)
sf.findings.set(findings)
sf.structures.set(structures)
sf.conditions.set(conditions)
#sf.current_image_id_index = current_image_id_index
sf.annotation_json_3d = annotation_json_3d
sf.viewer_state_3d = viewer_state_3d
sf.save()
return JsonResponse({"success": True})
return JsonResponse({"success": False})
return render(request, "create_post.html", {"posts": posts})
class ConditionAutocomplete(autocomplete.Select2QuerySetView):
def get_queryset(self):
# Don't forget to filter out results depending on the visitor !
if not self.request.user.is_authenticated:
return Condition.objects.none()
qs = Condition.objects.all()
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(name__icontains=self.q)
except FieldError:
return Condition.objects.none()
return qs
class ConditionAutocompleteRCRCurriculum(autocomplete.Select2QuerySetView):
def get_queryset(self):
# Don't forget to filter out results depending on the visitor !
if not self.request.user.is_authenticated:
return Condition.objects.none()
qs = Condition.objects.filter(rcr_curriculum=True)
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(name__icontains=self.q, rcr_curriculum=True)
except FieldError:
return Condition.objects.none()
return qs
class PresentationAutocomplete(autocomplete.Select2QuerySetView):
def get_queryset(self):
# Don't forget to filter out results depending on the visitor !
if not self.request.user.is_authenticated:
return Presentation.objects.none()
qs = Presentation.objects.all()
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(name__icontains=self.q)
except FieldError:
return Presentation.objects.none()
return qs
class FindingAutocomplete(autocomplete.Select2QuerySetView):
def get_queryset(self):
# Don't forget to filter out results depending on the visitor !
if not self.request.user.is_authenticated:
return Finding.objects.none()
qs = Finding.objects.all()
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(name__icontains=self.q)
except FieldError:
return Finding.objects.none()
return qs
class StructureAutocomplete(autocomplete.Select2QuerySetView):
def get_queryset(self):
# Don't forget to filter out results depending on the visitor !
if not self.request.user.is_authenticated:
return Structure.objects.none()
qs = Structure.objects.all()
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(name__icontains=self.q)
except FieldError:
return Structure.objects.none()
return qs
class SubspecialtyAutocomplete(autocomplete.Select2QuerySetView):
def get_queryset(self):
# Don't forget to filter out results depending on the visitor !
if not self.request.user.is_authenticated:
return Subspecialty.objects.none()
qs = Subspecialty.objects.all()
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(name__icontains=self.q)
except FieldError:
return Subspecialty.objects.none()
return qs
@login_required
# @user_is_atlas_editor
def categories_list(request):
# Allow an optional global search across category models via GET `q`.
query = request.GET.get("q", "").strip()
# Optional `types` GET parameter (can be repeated) to restrict which category
# models to search. Expected values: 'conditions','findings','structures',
# 'subspecialties','presentations'. If omitted, search all.
types = request.GET.getlist("types")
results = None
if query:
from .models import Condition, Finding, Structure, Subspecialty, Presentation
# Helper to conditionally populate each result list only when the
# corresponding type is requested (or when no types filter provided).
def include(name: str) -> bool:
return not types or name in types
results = {
"conditions": Condition.objects.filter(name__icontains=query).order_by("name") if include("conditions") else (),
"findings": Finding.objects.filter(name__icontains=query).order_by("name") if include("findings") else (),
"structures": Structure.objects.filter(name__icontains=query).order_by("name") if include("structures") else (),
"subspecialties": Subspecialty.objects.filter(name__icontains=query).order_by("name") if include("subspecialties") else (),
"presentations": Presentation.objects.filter(name__icontains=query).order_by("name") if include("presentations") else (),
}
return render(request, "atlas/categories_list.html", {"query": query, "results": results})
@login_required
def categories_search_partial(request):
"""Return only the search results partial for HTMX requests.
Expects GET param `q`.
"""
query = request.GET.get("q", "").strip()
types = request.GET.getlist("types")
results = {"conditions": (), "findings": (), "structures": (), "subspecialties": (), "presentations": ()}
if query:
from .models import Condition, Finding, Structure, Subspecialty, Presentation
def include(name: str) -> bool:
return not types or name in types
results = {
"conditions": Condition.objects.filter(name__icontains=query).order_by("name") if include("conditions") else (),
"findings": Finding.objects.filter(name__icontains=query).order_by("name") if include("findings") else (),
"structures": Structure.objects.filter(name__icontains=query).order_by("name") if include("structures") else (),
"subspecialties": Subspecialty.objects.filter(name__icontains=query).order_by("name") if include("subspecialties") else (),
"presentations": Presentation.objects.filter(name__icontains=query).order_by("name") if include("presentations") else (),
}
return render(request, "atlas/_categories_search_results.html", {"query": query, "results": results})
# def collection_view(request):
# collections = CaseCollection.objects.all()
#
# return render(
# request, "atlas/collection_view.html", {"collections": collections}
# )
class CollectionView(LoginRequiredMixin, SingleTableMixin, FilterView):
model = CaseCollection
table_class = CaseCollectionTable
template_name = "atlas/collection_view.html"
filterset_class = CaseCollectionFilter
@user_is_collection_author_or_atlas_editor
def collection_viva(request, pk):
collection = get_object_or_404(CaseCollection, pk=pk)
#cases = (
# collection.cases.all()
# .order_by("casedetail__sort_order")
# .prefetch_related(
# "series",
# "series__images",
# "series__contrast",
# "series__plane",
# "series__examination",
# )
#)
casedetails = (
CaseDetail.objects
.filter(collection=collection)
.select_related("case")
.order_by("sort_order")
.prefetch_related(
# Prefetch series and their images directly on the case
"case__series",
"case__series__modality",
"case__series__examination",
"case__series__plane",
"case__series__contrast",
"case__series__images",
)
)
return render(
request,
"atlas/collection_viva.html",
{"collection": collection, "casedetails": casedetails, "can_edit": True},
)
@user_is_collection_author_or_atlas_editor
def collection_viva_case(request, pk, case_number):
collection = get_object_or_404(CaseCollection, pk=pk)
# cases = collection.cases.all().order_by("casedetail__sort_order")
current_case = collection.get_case_by_index(case_number)
return render(
request,
"atlas/collection_viva_case.html",
{"collection": collection, "current_case": current_case},
)
@user_is_collection_author_or_atlas_editor
def collection_viva_series(request, pk, series_id):
collection = get_object_or_404(CaseCollection, pk=pk)
series = get_object_or_404(Series, pk=series_id)
return render(
request,
"atlas/collection_viva_series.html",
{"collection": collection, "current_series": series},
)
@user_is_collection_author_or_atlas_editor
def collection_detail(request, pk):
collection = get_object_or_404(CaseCollection, pk=pk)
# cases = collection.cases.all().order_by("casedetail__sort_order")
casedetails = (
CaseDetail.objects
.filter(collection=collection)
.select_related("case")
.annotate(series_count=Count("case__seriesdetail"))
.order_by("sort_order")
)
return render(
request,
"atlas/collection_detail.html",
{"collection": collection, "casesdetails": casedetails, "can_edit": True, "exam": collection},
)
@user_is_collection_author_or_atlas_editor
def collection_history(request, exam_id: int):
"""An overview page that shows who has attempted / finised the collection"""
collection = get_object_or_404(CaseCollection, pk=exam_id)
casedetails = CaseDetail.objects.filter(collection=collection).order_by(
"sort_order"
)
userexams = collection.get_user_exams()
cidexams = collection.get_cid_exams()
print("useruserexams:", userexams)
print("cidexams:", cidexams)
return render(
request,
"atlas/collection_history.html",
{
"collection": collection,
"casesdetails": casedetails,
"can_edit": True,
"userexams": userexams,
"cidexams": cidexams,
},
)
@user_is_collection_author_or_atlas_editor
def collection_history_user(request, exam_id: int, user_pk: int):
collection = get_object_or_404(CaseCollection, pk=exam_id)
user = get_object_or_404(User, pk=user_pk)
casedetails = CaseDetail.objects.filter(
collection=collection,
).order_by("sort_order")
print(user)
print(casedetails)
user_answers = [
(casedetail, casedetail.get_user_answers(user)) for casedetail in casedetails
]
return render(
request,
"atlas/collection_history_user.html",
{
"collection": collection,
"user": user,
"casedetails": casedetails,
"user_answers": user_answers,
},
)
@user_is_collection_author_or_atlas_editor
def collection_history_ciduser(request, exam_id: int, cid: int):
collection = get_object_or_404(CaseCollection, pk=exam_id)
casedetails = CaseDetail.objects.filter(
collection=collection,
).order_by("sort_order")
print(casedetails)
user_answers = [
(casedetail, casedetail.get_cid_answers(cid)) for casedetail in casedetails
]
return render(
request,
"atlas/collection_history_user.html",
{
"collection": collection,
"user": cid,
"casedetails": casedetails,
"user_answers": user_answers,
},
)
def collection_take_start(request, pk, cid=None, passcode=None):
"""The starting page for taking / reviewing a case collection
Args:
request (_type_): _description_
pk (_type_): _description_
Returns:
_type_: _description_
"""
collection = get_object_or_404(CaseCollection, pk=pk)
try:
collection.check_user_can_take(cid, passcode, request.user)
valid_user = True
except Http404:
valid_user = False
except PrerequisiteRequired as e:
# Show a friendly page linking to the required collection rather than 404
prereq = getattr(e, 'prereq', None)
return render(
request,
'atlas/prerequisite_required.html',
{'message': str(e), 'prereq': prereq, 'cid': cid, 'passcode': passcode},
)
#valid_user = collection.check_logged_in_user(request)
cid_exam = None
if collection.collection_type == "REV" or valid_user:
cid_exam = collection.get_cid_and_user_exams(user_user=request.user).first()
template_variables = {
"collection": collection,
"valid_user": valid_user,
"cid_exam": cid_exam,
}
match collection.collection_type:
case x if x in ("REP", "QUE"):
return render(
request, "atlas/collection_take_start.html", template_variables
)
case "REV":
return render(
request, "atlas/collection_review_start.html", template_variables
)
def _case_number_for_collection_and_caseid(collection_pk, case_id):
"""Helper: given a collection PK and a case PK, return the case index (case_number).
Raises Http404 if the case is not found in the collection.
"""
collection = get_object_or_404(CaseCollection, pk=collection_pk)
try:
# attempt to resolve the case object (may raise Http404)
case_obj = get_case_for_case_detail(case_id)
except Http404:
raise
try:
return collection.get_index_of_case(case_obj)
except ValueError:
raise Http404(f"Case {case_id} not in collection {collection_pk}")
@login_required
def redirect_collection_case_details_by_id(request, exam_id, case_id):
case_number = _case_number_for_collection_and_caseid(exam_id, case_id)
return redirect("atlas:collection_case_details", exam_id=exam_id, case_number=case_number)
@login_required
def redirect_collection_case_questions_by_id(request, exam_id, case_id):
case_number = _case_number_for_collection_and_caseid(exam_id, case_id)
return redirect("atlas:collection_case_questions", exam_id=exam_id, case_number=case_number)
@login_required
def redirect_collection_case_priors_by_id(request, exam_id, case_id):
case_number = _case_number_for_collection_and_caseid(exam_id, case_id)
return redirect("atlas:collection_case_priors", exam_id=exam_id, case_number=case_number)
@login_required
def redirect_collection_case_dicom_json_by_id(request, exam_id, case_id):
case_number = _case_number_for_collection_and_caseid(exam_id, case_id)
return redirect("atlas:collection_case_dicom_json", exam_id=exam_id, case_number=case_number)
@login_required
def redirect_collection_case_dicom_json_review_by_id(request, exam_id, case_id):
case_number = _case_number_for_collection_and_caseid(exam_id, case_id)
return redirect("atlas:collection_case_dicom_json_review", exam_id=exam_id, case_number=case_number)
@login_required
def redirect_collection_case_view_by_id(request, pk, case_id):
case_number = _case_number_for_collection_and_caseid(pk, case_id)
return redirect("atlas:collection_case_view", pk=pk, case_number=case_number)
@login_required
def redirect_collection_case_displaysetup_by_id(request, collection_id, case_id):
case_number = _case_number_for_collection_and_caseid(collection_id, case_id)
return redirect("atlas:collection_case_displaysetup", collection_id=collection_id, case_number=case_number)
@login_required
def redirect_collection_case_view_take_by_id(request, pk, case_id, cid, passcode):
case_number = _case_number_for_collection_and_caseid(pk, case_id)
return redirect("atlas:collection_case_view_take", pk=pk, case_number=case_number, cid=cid, passcode=passcode)
@login_required
def redirect_collection_case_view_take_user_by_id(request, pk, case_id):
case_number = _case_number_for_collection_and_caseid(pk, case_id)
return redirect("atlas:collection_case_view_take_user", pk=pk, case_number=case_number)
@login_required
def redirect_collection_case_view_take_user_answers_by_id(request, pk, case_id):
case_number = _case_number_for_collection_and_caseid(pk, case_id)
return redirect("atlas:collection_case_view_take_user_answers", pk=pk, case_number=case_number)
@user_is_collection_author_or_atlas_editor
def collection_case_priors(request, exam_id, case_number):
collection = get_object_or_404(CaseCollection, pk=exam_id)
try:
case_obj = collection.get_case_by_index(case_number)
casedetail = CaseDetail.objects.get(case=case_obj, collection=collection)
except Exception:
raise Http404("Case not found in collection")
if request.htmx:
# Ensure we can render the updated single-card partial and return it so HTMX
# can swap the card on the client side.
if "remove" in request.POST:
prior_pk = request.POST["remove"]
try:
p = CasePrior.objects.get(casedetail=casedetail, prior_case=prior_pk)
p.delete()
except CasePrior.DoesNotExist:
# Already removed; continue and render the not-added card
pass
prior_case = Case.objects.get(pk=prior_pk)
added = False
relation = ""
visibility = "AL"
html = render_to_string(
"atlas/partials/_prior_card.html",
{
"case": prior_case,
"added": added,
"relation": relation,
"visibility": visibility,
"casedetail": casedetail,
"collection": collection,
},
request=request,
)
return HttpResponse(html)
elif "prior_case_id" in request.POST:
if not request.POST.get("relation"):
# Return the card partial with an inline error so HTMX will replace
# the card and display the validation message rather than navigating
# to a new page or showing a plain text response.
prior_case = Case.objects.get(pk=request.POST.get("prior_case_id"))
added = False
relation = request.POST.get("relation", "")
visibility = request.POST.get("prior_visibility", "AL")
html = render_to_string(
"atlas/partials/_prior_card.html",
{
"case": prior_case,
"added": added,
"relation": relation,
"visibility": visibility,
"casedetail": casedetail,
"collection": collection,
"error": "You need to enter text to describe the relationship between the cases",
},
request=request,
)
return HttpResponse(html)
prior_case = Case.objects.get(pk=request.POST["prior_case_id"])
p, created = CasePrior.objects.get_or_create(
casedetail=casedetail, prior_case=prior_case
)
p.relation_text = request.POST.get("relation", "")
p.prior_visibility = request.POST.get("prior_visibility", "AL")
p.save()
added = True
relation = p.relation_text
visibility = p.prior_visibility
html = render_to_string(
"atlas/partials/_prior_card.html",
{
"case": prior_case,
"added": added,
"relation": relation,
"visibility": visibility,
"casedetail": casedetail,
"collection": collection,
},
request=request,
)
return HttpResponse(html)
else:
return HttpResponse("False", status=400)
collection = casedetail.collection
added_priors = CasePrior.objects.filter(casedetail=casedetail)
added_relation_map = {}
for prior in added_priors:
added_relation_map[prior.prior_case] = (prior.relation_text, prior.prior_visibility)
case = casedetail.case
available_cases = case.get_all_prior_cases()
available_priors = []
for case in available_cases:
if case in added_relation_map:
available_priors.append((case, True, added_relation_map[case][0], added_relation_map[case][1]))
else:
available_priors.append((case, False, "", "AL"))
form = PriorCaseForm()
case_number, case_count = collection.get_index_of_case(
casedetail.case, case_count=True
)
previous = collection.get_previous_case(casedetail.case)
next = collection.get_next_case(casedetail.case)
# Determine whether the current user can edit this collection/case so
# included fragments which rely on `can_edit` have the value available.
can_edit = request.user.is_superuser or collection.author.filter(pk=request.user.pk).exists()
return render(
request,
"atlas/collection_case_priors.html",
{
"casedetail": casedetail,
"form": form,
#"model_form": model_form,
# "example_form": example_form,
"collection": collection,
"case": casedetail.case,
"previous": previous,
"next": next,
"collection_length": case_count,
"case_number": case_number,
"added_priors": added_priors,
"available_priors": available_priors,
"can_edit": can_edit,
# Provide a default `error` value so `{% if error %}` checks in included
# partials won't raise during variable resolution when the key is absent.
"error": None,
},
)
@user_is_collection_author_or_atlas_editor
def collection_case_questions(request, exam_id, case_number):
collection = get_object_or_404(CaseCollection, pk=exam_id)
try:
case_obj = collection.get_case_by_index(case_number)
casedetail = CaseDetail.objects.get(case=case_obj, collection=collection)
except Exception:
raise Http404("Case not found in collection")
if not collection.collection_type == "QUE":
raise Http404("Collection not in question mode")
if request.method == "POST":
# Called if the user saves the correct answer form
if request.POST.get("submit") == "answer":
# answers = request.POST
casedetail.question_answers = json.loads(request.POST.get("json_answer"))
casedetail.save()
example_form = JsonAnswerForm(
request.POST, question_schema=casedetail.question_schema
)
form = CaseQuestionForm(instance=casedetail)
# Called if the user saves the main form
elif request.POST.get("submit") == "save":
form = CaseQuestionForm(request.POST, instance=casedetail)
if form.is_valid():
form.save()
# Add any additional logic or redirection here
# example_form = JsonAnswerForm(request.POST, question_schema=casedetail.question_schema)
post_data = request.POST.copy()
if (
casedetail.question_schema is not None
and casedetail.question_answers is not None
):
question_keys = set(casedetail.question_schema["properties"].keys())
answer_keys = set(casedetail.question_answers.keys())
answers = {}
for key, value in casedetail.question_answers.items():
if key in question_keys:
answers[key] = value
diff = question_keys - answer_keys
for key in diff:
answers[key] = ""
post_data["json_answer"] = json.dumps(answers)
casedetail.question_answers = answers
casedetail.save()
example_form = JsonAnswerForm(
post_data, question_schema=casedetail.question_schema
)
form = CaseQuestionForm(instance=casedetail)
# This shouldn't happen
else:
assert False
form = CaseQuestionForm(request.POST, instance=casedetail)
else:
post_data = request.POST.copy()
post_data["json_answer"] = json.dumps(casedetail.question_answers)
example_form = JsonAnswerForm(
post_data, question_schema=casedetail.question_schema
)
form = CaseQuestionForm(instance=casedetail)
pass
# post_data = None
# if casedetail.question_schema is not None and casedetail.question_answers is not None:
# question_keys = set(casedetail.question_schema["properties"].keys())
# answer_keys = set(casedetail.question_answers.keys())
# answers = {}
# for key, value in casedetail.question_answers.items():
# if key in question_keys:
# answers[key] = value
# diff = question_keys - answer_keys
# for key in diff:
# answers[key] = ""
# post_data = request.POST.copy()
# post_data["json_answer"]= json.dumps(answers)
# casedetail.question_answers = answers
# example_form = JsonAnswerForm(post_data, question_schema=casedetail.question_schema)
# form = CaseQuestionForm(instance=casedetail)
# blank_form = JsonAnswerForm(question_schema=casedetail.question_schema)
case_number, case_count = collection.get_index_of_case(
casedetail.case, case_count=True
)
previous = collection.get_previous_case(casedetail.case)
next = collection.get_next_case(casedetail.case)
return render(
request,
"atlas/collection_case_questions.html",
{
"casedetail": casedetail,
"form": form,
"example_form": example_form,
"collection": collection,
"case": casedetail.case,
"previous": previous,
"next": next,
"collection_length": case_count,
"case_number": case_number,
},
)
@user_is_collection_author_or_atlas_editor
def collection_case_details(request, exam_id, case_number):
# Accept either case_number (index into collection) or case_id (case PK).
collection = get_object_or_404(CaseCollection, pk=exam_id)
try:
case_obj = collection.get_case_by_index(int(case_number))
casedetail = CaseDetail.objects.get(case=case_obj, collection=collection)
except Exception:
raise Http404("Case not found in collection")
if request.method == "POST":
form = CaseDetailForm(request.POST, instance=casedetail)
if form.is_valid():
form.save()
if request.htmx:
return HttpResponse("Saved")
return redirect(request.path)
else:
if request.htmx:
return HttpResponse("Invalid form", status=400)
else:
form = CaseDetailForm(instance=casedetail)
case_number, case_count = collection.get_index_of_case(
casedetail.case, case_count=True
)
previous = collection.get_previous_case(casedetail.case)
next = collection.get_next_case(casedetail.case)
return render(
request,
"atlas/collection_case_details.html",
{
"casedetail": casedetail,
"form": form,
"collection": collection,
"case": casedetail.case,
"previous": previous,
"next": next,
"collection_length": case_count,
"case_number": case_number,
},
)
@user_is_collection_author_or_atlas_editor
def collection_mark_overview(request, pk):
collection = get_object_or_404(CaseCollection, pk=pk)
if collection.collection_type == "REV":
raise Http404("Collection in review mode")
cases = collection.cases.all().order_by("casedetail__sort_order").prefetch_related()
casedetails = CaseDetail.objects.filter(
case__in=cases, collection=collection
).prefetch_related("case")
answers = CidReportAnswer.objects.filter(question__in=casedetails)
case_map = {}
for cd in casedetails:
case_map[cd] = []
for answer in answers:
case_map[answer.question].append(answer)
return render(
request,
"atlas/collection_mark_overview.html",
{
"collection": collection,
"answers": answers,
"casedetails": casedetails,
"case_map": case_map,
},
)
@user_is_collection_author_or_atlas_editor
def collection_mark_question(request, pk, case_number):
collection = get_object_or_404(CaseCollection, pk=pk)
if collection.collection_type == "REV":
raise Http404("Collection in review mode")
case, case_count = collection.get_case_by_index(case_number, case_count=True)
previous = case_number > 0
next = case_number < (case_count - 1)
casedetail = CaseDetail.objects.filter(case=case, collection=collection)
answers = CidReportAnswer.objects.filter(question__in=casedetail)
answer_forms = []
if request.method == "POST":
for answer in answers:
if answer:
form = CidReportAnswerMarkForm(
request.POST, instance=answer, prefix=answer.id
)
else:
form = CidReportAnswerMarkForm(request.POST, prefix=answer.id)
if form.is_valid():
answer = form.save(commit=False)
# answer.cid = cid
# answer.question = casedetail
# answer.published_date = timezone.now()
answer.save()
kwargs = {"pk": pk}
if "save" in request.POST:
return redirect(
"atlas:collection_mark_question", case_number=case_number, **kwargs
)
if "next" in request.POST:
return redirect(
"atlas:collection_mark_question", case_number=case_number + 1, **kwargs
)
elif "previous" in request.POST:
return redirect(
"atlas:collection_mark_question", case_number=case_number - 1, **kwargs
)
elif "goto" in request.POST:
return redirect(
"atlas:collection_mark_question",
case_number=int(request.POST.get("goto")),
**kwargs,
)
else:
for answer in answers:
form = CidReportAnswerMarkForm(instance=answer, prefix=answer.pk)
answer_forms.append((answer, form))
series_list = case.series.all().prefetch_related("images", "examination", "plane")
return render(
request,
"atlas/collection_mark_question.html",
{
"collection": collection,
"series_list": series_list,
"answers": answers,
"casedetail": casedetail,
"case": case,
"case_number": case_number,
"answer_forms": answer_forms,
"previous": previous,
"next": next,
},
)
@user_is_collection_author_or_atlas_editor
def exam_cids(request, pk):
collection = get_object_or_404(CaseCollection, pk=pk)
if (
request.user in collection.author.all()
or request.user.groups.filter(name="cid_user_manager").exists()
):
cid_users = collection.valid_cid_users.all()
cid_user_count = cid_users.count()
return render(
request,
"atlas/exam_cids.html",
{
"collection": collection,
"cid_users": cid_users,
"cid_user_count": cid_user_count,
},
)
raise PermissionDenied
def collection_take_overview_user(request, pk):
return collection_take_overview(request, pk)
def collection_take_overview(
request, pk: int, cid: int | None = None, passcode: str | None = None
):
"""
Summary page that shows which questions a user has attempted
(with / without answers)
Args:
request (_type_): _description_
pk (int): _description_
cid (int | None, optional): _description_. Defaults to None.
passcode (str | None, optional): _description_. Defaults to None.
Returns:
_type_: _description_
"""
collection = get_object_or_404(CaseCollection, pk=pk)
try:
collection.check_user_can_take(cid, passcode, request.user)
except PrerequisiteRequired as e:
prereq = getattr(e, 'prereq', None)
return render(request, 'atlas/prerequisite_required.html', {'message': str(e), 'prereq': prereq, 'cid': cid, 'passcode': passcode})
cid_user_exam = collection.get_or_create_cid_user_exam(
cid=cid, user_user=request.user
)
if request.htmx:
if "finish" in request.POST:
cid_user_exam.completed = True
cid_user_exam.save()
collection.exam_user_status.create(
cid_user_exam=cid_user_exam,
status="submitted",
extra="manual submission",
)
return HttpResponse("True")
else:
raise Http404()
cases = collection.cases.all().order_by("casedetail__sort_order").prefetch_related()
casedetails = CaseDetail.objects.filter(case__in=cases, collection=collection)
if cid is not None:
answers = CidReportAnswer.objects.filter(cid=cid, question__in=casedetails)
else:
answers = UserReportAnswer.objects.filter(
user=request.user, question__in=casedetails
)
answer_question_map = {}
for ans in answers:
answer_question_map[ans.question] = ans
question_answer_tuples = []
answer_count = 0
for cd in casedetails:
self_review = cid_user_exam.selfreview_set.filter(case=cd.case)
if cd in answer_question_map and (
answer_question_map[cd].answer or answer_question_map[cd].json_answer
):
question_answer_tuples.append((cd, answer_question_map[cd], self_review))
answer_count += 1
else:
question_answer_tuples.append((cd, None, self_review))
return render(
request,
"atlas/collection_take_overview.html",
{
"collection": collection,
"cid": cid,
"question_answer_tuples": question_answer_tuples,
"answer_count": answer_count,
"collection_length": len(cases),
"passcode": passcode,
"cid_user_exam": cid_user_exam,
},
)
def collection_case_view_take_user_answers(request, pk: int, case_number: int):
return collection_case_view_take_answers(request, pk, case_number)
def collection_case_view_take_answers(
request, pk: int, case_number: int, cid=None, passcode=None
):
""" """
collection = get_object_or_404(CaseCollection, pk=pk)
try:
collection.check_user_can_take(cid, passcode, request.user)
except PrerequisiteRequired as e:
prereq = getattr(e, 'prereq', None)
return render(request, 'atlas/prerequisite_required.html', {'message': str(e), 'prereq': prereq, 'cid': cid, 'passcode': passcode})
case, case_count = collection.get_case_by_index(case_number, case_count=True)
cid_user_exam = collection.get_or_create_cid_user_exam(
cid=cid, user_user=request.user
)
casedetail = CaseDetail.objects.get(case=case, collection=collection)
return render(
request,
"atlas/collection_case_view_take_answers.html",
{
"collection": collection,
"case": case,
"casedetail": casedetail,
},
)
def collection_case_view_take_user(request, pk, case_number=None, case_id=None):
# Accept either case_number (index) or case_id (case PK) and forward both to
# the handler. This allows URLs that use either name to keep working.
return collection_case_view_take(request, pk, case_number=case_number, cid=None, passcode=None, case_id=case_id)
def collection_case_view_take(
request, pk, case_number=None, cid=None, passcode=None, case_id=None
):
"""
Displays the page from which the user takes a case in the collection
Args:
request (_type_): _description_
pk (int): _description_
case_number (int): _description_
cid (_type_, optional): _description_. Defaults to None.
passcode (_type_, optional): _description_. Defaults to None.
Raises:
Http404: If user does not have access
Returns:
_type_: _description_
"""
collection = get_object_or_404(CaseCollection, pk=pk)
# If caller provided a case_id (case PK) instead of a case_number (index),
# translate it into the index used by collection.get_case_by_index().
if case_id is not None and case_number is None:
try:
case_obj = get_case_for_case_detail(case_id)
except Http404:
raise
try:
case_number = collection.get_index_of_case(case_obj)
except ValueError:
raise Http404(f"Case {case_id} not in collection {pk}")
# Ensure we have a case_number to work with
if case_number is None:
raise Http404("No case specified")
form = None
answer: None | CidReportAnswer | UserReportAnswer = None
try:
collection.check_user_can_take(cid, passcode, request.user)
except PrerequisiteRequired as e:
prereq = getattr(e, 'prereq', None)
return render(request, 'atlas/prerequisite_required.html', {'message': str(e), 'prereq': prereq, 'cid': cid, 'passcode': passcode})
case, case_count = collection.get_case_by_index(case_number, case_count=True)
cid_user_exam = collection.get_or_create_cid_user_exam(
cid=cid, user_user=request.user
)
if cid_user_exam.start_time is None:
cid_user_exam.start_time = timezone.now()
cid_user_exam.save()
casedetail = CaseDetail.objects.get(case=case, collection=collection)
if not collection.review_only():
if collection.collection_type == "REP":
if cid is not None:
answer = casedetail.cidreportanswer_set.filter(cid=cid).first()
ReportAnswerForm = CidReportAnswerForm
else:
answer = casedetail.userreportanswer_set.filter(
user=request.user
).first()
ReportAnswerForm = UserReportAnswerForm
elif collection.collection_type == "QUE":
if cid is not None:
answer = casedetail.cidreportanswer_set.filter(cid=cid).first()
ReportAnswerForm = CidQuestionAnswerForm
else:
answer = casedetail.userreportanswer_set.filter(
user=request.user
).first()
ReportAnswerForm = UserQuestionAnswerForm
# If we have an answer object but it doesn't have a started_at, set it now.
if answer and getattr(answer, 'started_at', None) is None:
try:
answer.started_at = timezone.now()
answer.save()
except Exception:
# Non-fatal: if saving fails, continue without blocking the user
logger.exception("Failed to set started_at on answer", exc_info=True)
# If no answer exists yet, create a placeholder so we have a started_at timestamp
if not answer and collection.collection_type in ("REP", "QUE"):
try:
if cid is not None:
answer = CidReportAnswer(question=casedetail, cid=cid)
else:
answer = UserReportAnswer(question=casedetail, user=request.user)
answer.started_at = timezone.now()
answer.save()
except Exception:
# If creation fails, leave answer as None; the normal flow will handle form creation
answer = None
logger.exception("Failed to create initial answer object", exc_info=True)
if request.method == "POST":
if collection.collection_type in ("REP", "QUE"):
if not collection.publish_results:
if answer:
form = ReportAnswerForm(
request.POST, instance=answer, casedetail=casedetail
)
else:
form = ReportAnswerForm(request.POST, casedetail=casedetail)
if not cid_user_exam.completed and not (
answer is not None and answer.completed
):
if form.is_valid():
answer = form.save(commit=False)
answer.set_cid_or_user(cid=cid, user=request.user)
answer.question = casedetail
# Record submission timestamp
answer.submitted_at = timezone.now()
# answer.published_date = timezone.now()
if "complete_case" in request.POST:
if not collection.self_review:
raise Http404("Self review not enabled")
answer.completed = True
# If this was a timed-out submission, mark as completed
if request.POST.get('timed_out') == '1':
answer.completed = True
## Also mark the overall exam as completed for this user/CID
#try:
# cid_user_exam.completed = True
# cid_user_exam.save()
# # Record an exam status entry
# collection.exam_user_status.create(
# cid_user_exam=cid_user_exam,
# status="submitted",
# extra="autosubmit",
# )
#except Exception:
# logger.exception("Failed to mark cid_user_exam completed on autosubmit")
answer.save()
cid_user_exam.end_time = timezone.now()
cid_user_exam.save()
# If this was an AJAX/HTMX request, return JSON so the client
# can remain on the same page and update the UI without redirect.
# Detect HTMX/XHR requests robustly. HTMX sets the HX-Request header.
is_ajax = (
request.headers.get('HX-Request', '').lower() == 'true'
or request.headers.get('x-requested-with') == 'XMLHttpRequest'
or getattr(request, 'htmx', False)
)
if is_ajax:
return JsonResponse(
{
'status': 'ok',
'locked': answer.completed,
'submitted_at': answer.submitted_at.isoformat() if getattr(answer, 'submitted_at', None) else None,
}
)
if cid is not None:
kwargs = {"pk": pk, "cid": cid, "passcode": passcode}
redirect_url = "atlas:collection_case_view_take"
overview_url = "atlas:collection_take_overview"
else:
kwargs = {"pk": pk}
redirect_url = "atlas:collection_case_view_take_user"
overview_url = "atlas:collection_take_overview_user"
if "next" in request.POST:
return redirect(redirect_url, case_number=case_number + 1, **kwargs)
elif "previous" in request.POST:
return redirect(redirect_url, case_number=case_number - 1, **kwargs)
elif "finish" in request.POST:
return redirect(overview_url, **kwargs)
elif "goto" in request.POST:
return redirect(
redirect_url,
case_number=int(request.POST.get("goto")),
**kwargs,
)
else:
if collection.collection_type in ("REP", "QUE"):
form = ReportAnswerForm(instance=answer, casedetail=casedetail)
series_list = case.series.all().prefetch_related("images", "examination", "plane")
previous = case_number > 0
next = case_number < (case_count - 1)
# Set what details a user is / is not able to view when taking the case
# this can be configured via the collection edit menu
if (
collection.publish_results
or cid_user_exam.completed
or (
not collection.feedback_once_collection_complete
and answer is not None
and answer.completed
)
):
question_completed = True
show_title = collection.show_title_post
show_history = collection.show_history_post
show_description = collection.show_description_post
show_discussion = collection.show_discussion_post
show_report = collection.show_report_post
self_review = cid_user_exam.selfreview_set.filter(case=case)
resources = case.caseresource_set.all()
else:
question_completed = False
show_title = collection.show_title_pre
show_history = collection.show_history_pre
show_description = collection.show_description_pre
show_discussion = collection.show_discussion_pre
show_report = collection.show_report_pre
self_review = []
resources = case.caseresource_set.filter(pre_review=True)
prior_cases = casedetail.caseprior_set.all()
series_to_load = []
for series in series_list:
series_to_load.append((series, False, ""))
for prior in prior_cases:
match prior.prior_visibility:
case "AL":
for series in prior.prior_case.get_series():
series_to_load.append((series, True, prior.relation_text))
case "RE":
if question_completed:
for series in prior.prior_case.get_series():
series_to_load.append((series, True, prior.relation_text))
case _:
continue
# Determine whether any prior series will be shown to the candidate.
has_priors = any(prior_flag for (_, prior_flag, _) in series_to_load)
prior_count = prior_cases.count()
return render(
request,
"atlas/collection_case_view_take.html",
{
"form": form,
"collection": collection,
"case": case,
"casedetail": casedetail,
"series_list": series_list,
"series_to_load": series_to_load,
"has_priors": has_priors,
"prior_count": prior_count,
"case_number": case_number,
"previous": previous,
"next": next,
"cid": cid,
"passcode": passcode,
"answer": answer,
"show_title": show_title,
"show_history": show_history,
"show_description": show_description,
"show_discussion": show_discussion,
"show_report": show_report,
"resources": resources,
"cid_user_exam": cid_user_exam,
"question_completed": question_completed,
"self_review": self_review,
"answer_started_at_iso": answer.started_at.isoformat() if answer and getattr(answer, 'started_at', None) else None,
},
)
@user_is_collection_author_or_atlas_editor
def collection_case_view(request, pk, case_number=None, case_id=None):
"""Edit/management overview of the case (as part of the collection)
Args:
request (_type_): _description_
pk (_type_): _description_
case_number (_type_): _description_
Raises:
Http404: _description_
Returns:
_type_: _description_
"""
collection = get_object_or_404(CaseCollection, pk=pk)
# Support either case_number (index) or case_id (case PK) in the URL/kwargs.
if case_id is not None and case_number is None:
try:
case_obj = get_case_for_case_detail(case_id)
except Http404:
raise
try:
case_number = collection.get_index_of_case(case_obj)
except ValueError:
raise Http404(f"Case {case_id} not in collection {pk}")
if case_number is None:
raise Http404("No case specified")
form = None
answer = None
case, case_count = collection.get_case_by_index(case_number, case_count=True)
case = get_case_for_case_detail(case.pk)
casedetail = CaseDetail.objects.get(case=case, collection=collection)
series_list = case.series.all().prefetch_related("images", "examination", "plane")
previous = collection.get_previous_case(casedetail.case)
next = collection.get_next_case(casedetail.case)
return render(
request,
"atlas/case_detail.html",
{
"form": form,
"collection": collection,
"collection_length": case_count,
"case": case,
"casedetail": casedetail,
"series_list": series_list,
"case_number": case_number,
"previous": previous,
"next": next,
"answer": answer,
"cimar_sid": request.user.userprofile.cimar_sid,
},
)
def collection_dicom_json(request, pk):
collection = get_object_or_404(CaseCollection, pk=pk)
return JsonResponse(collection.get_ohif_dicom_json())
@user_is_author_or_atlas_editor
def use_dates_as_descriptions(request, pk):
case = get_object_or_404(Case, pk=pk)
series: Series
for series in case.series.all():
series.use_date_as_description()
return HttpResponse("Done")
def collection_case_dicom_json_review(request, exam_id, case_id):
return collection_case_dicom_json(request, exam_id, case_id, review=True)
def collection_case_dicom_json(request, exam_id, case_id, review=False):
# Support either case_number (index) or case_id (case PK)
try:
casedetail = CaseDetail.objects.get(case=case_id, collection=exam_id)
except Exception:
collection = get_object_or_404(CaseCollection, pk=exam_id)
try:
case_obj = collection.get_case_by_index(int(case_id))
casedetail = CaseDetail.objects.get(case=case_obj, collection=collection)
except Exception:
raise Http404("Case not found in collection")
if review:
priors = casedetail.caseprior_set.exclude(prior_visibility="NO")
else:
priors = casedetail.caseprior_set.exclude(prior_visibility__in=["NO", "RE"])
return JsonResponse(casedetail.case.get_case_dicom_json(priors=priors))
def case_dicom_json(request, pk):
case = get_object_or_404(Case, pk=pk)
return JsonResponse(case.get_case_dicom_json())
@login_required
def resource_search_widget(request):
"""HTMX endpoint: search existing `Resource` objects with simple filters.
GET params:
- q: query string (name/description)
- site: site id (optional)
- subspecialty: subspecialty id (optional)
- open_access: '1' or '0' (optional)
Returns rendered partial `atlas/partials/resource_search_widget_results.html`.
"""
q = request.GET.get("q", "").strip()
site = request.GET.get("site")
subspecialty = request.GET.get("subspecialty")
open_access = request.GET.get("open_access")
qs = Resource.objects.all()
if q:
qs = qs.filter(Q(name__icontains=q) | Q(description__icontains=q) | Q(url__icontains=q))
if site:
try:
qs = qs.filter(sites__id=int(site))
except Exception:
pass
if subspecialty:
try:
qs = qs.filter(subspecialty__id=int(subspecialty))
except Exception:
pass
if open_access in ["0", "1"]:
qs = qs.filter(open_access=(open_access == "1"))
# Restrict to what the user may see: open_access or authored (unless editor)
if not (request.user.is_superuser or request.user.groups.filter(name="atlas_editor").exists()):
qs = qs.filter(Q(open_access=True) | Q(author__id=request.user.id)).distinct()
results = qs.order_by("name")[:50]
field = request.GET.get("field", "resource_search")
case_id = request.GET.get("case_id")
return render(request, "atlas/partials/resource_search_widget_results.html", {"results": results, "field": field, "case_id": case_id})
@login_required
@require_POST
def case_attach_resource(request, pk):
"""Attach an existing resource to the case (HTMX POST).
POST params:
- resource_id: integer
- pre_review: '1' or '0' (optional)
Returns the updated `atlas/partials/case_resources_list.html` fragment.
"""
case = get_case_for_case_detail(pk)
if not case.check_user_can_edit(request.user):
return HttpResponse(status=403)
resource_id = request.POST.get("resource_id")
pre_review = request.POST.get("pre_review", "0") == "1"
try:
resource = Resource.objects.get(pk=int(resource_id))
except Exception:
return HttpResponse(status=400)
# create if not exists
cr, created = CaseResource.objects.get_or_create(case=case, resource=resource, defaults={"pre_review": pre_review})
if not created:
# update flag
cr.pre_review = pre_review
cr.save()
# Refresh case from DB to ensure related sets reflect changes and render updated list
try:
case = Case.objects.get(pk=case.pk)
except Exception:
pass
html = render_to_string("atlas/partials/case_resources_list.html", {"case": case}, request=request)
resp = HttpResponse(html)
# Notify any HTMX listeners that the case resources have changed
resp["HX-Trigger"] = json.dumps({"caseResourcesChanged": {"case_id": case.pk}})
return resp
@login_required
@require_POST
def case_detach_resource(request, pk):
case = get_case_for_case_detail(pk)
if not case.check_user_can_edit(request.user):
return HttpResponse(status=403)
resource_id = request.POST.get("resource_id")
try:
cr = CaseResource.objects.filter(case=case, resource__id=int(resource_id))
cr.delete()
except Exception:
pass
# Refresh case from DB and return updated partial
try:
case = Case.objects.get(pk=case.pk)
except Exception:
pass
html = render_to_string("atlas/partials/case_resources_list.html", {"case": case}, request=request)
resp = HttpResponse(html)
resp["HX-Trigger"] = json.dumps({"caseResourcesChanged": {"case_id": case.pk}})
return resp
@login_required
@require_POST
def case_toggle_resource_pre(request, pk):
"""Toggle the pre_review flag on a CaseResource (HTMX POST).
POST params:
- resource_id: integer
Returns the updated `atlas/partials/case_resources_list.html` fragment.
"""
case = get_case_for_case_detail(pk)
if not case.check_user_can_edit(request.user):
return HttpResponse(status=403)
resource_id = request.POST.get("resource_id")
try:
cr = CaseResource.objects.get(case=case, resource__id=int(resource_id))
except Exception:
return HttpResponse(status=400)
cr.pre_review = not bool(cr.pre_review)
cr.save()
# Refresh case from DB to ensure related sets reflect changes and render updated list
try:
case = Case.objects.get(pk=case.pk)
except Exception:
pass
html = render_to_string("atlas/partials/case_resources_list.html", {"case": case}, request=request)
resp = HttpResponse(html)
resp["HX-Trigger"] = json.dumps({"caseResourcesChanged": {"case_id": case.pk}})
return resp
def series_dicom_json(request, pk):
series = get_object_or_404(Series, pk=pk)
return JsonResponse(series.get_ohif_dicom_json())
@user_is_collection_author_or_atlas_editor
def delete_collection_cid_answers(request, exam_id, cid):
collection = get_object_or_404(CaseCollection, pk=exam_id)
cases = collection.cases.all().order_by("casedetail__sort_order").prefetch_related()
casedetails = CaseDetail.objects.filter(
case__in=cases, collection=collection
).prefetch_related("case")
cid_user_answers = CidReportAnswer.objects.filter(
question__in=casedetails, cid=cid
)
temp = {}
for ans in cid_user_answers:
temp[ans.pk] = (ans.question.pk, ans.score, ans.cid, ans.answer)
ans.delete()
return JsonResponse(temp)
@user_is_collection_author_or_atlas_editor
def delete_collection_answers(request, exam_id):
collection = get_object_or_404(CaseCollection, pk=exam_id)
cases = collection.cases.all().order_by("casedetail__sort_order").prefetch_related()
casedetails = CaseDetail.objects.filter(
case__in=cases, collection=collection
).prefetch_related("case")
cid_user_answers = CidReportAnswer.objects.filter(question__in=casedetails)
temp = {}
for ans in cid_user_answers:
temp[ans.pk] = (ans.question.pk, ans.score, ans.cid, ans.answer)
ans.delete()
return JsonResponse(temp)
# TODO: this needs a major cleanup
@user_is_collection_author_or_atlas_editor
def collection_scores_cid(request, pk):
collection = get_object_or_404(CaseCollection, pk=pk)
if not collection.exam_mode:
raise Http404("Collection not in exam mode")
user_answers_and_marks = defaultdict(list)
user_answers_marks = defaultdict(list)
user_answers = defaultdict(list)
user_names = {}
# cid_passcodes = {}
by_question = defaultdict(dict)
unmarked = set()
cases = collection.cases.all().order_by("casedetail__sort_order").prefetch_related()
casedetails = CaseDetail.objects.filter(
case__in=cases, collection=collection
).prefetch_related("case")
cid_user_answers = CidReportAnswer.objects.filter(question__in=casedetails)
cids = set()
case_list = list(cases)
# Loop through all candidates
for cid_user_answer in cid_user_answers:
# Convoluted (probably...)
cid = cid_user_answer.cid
# cid_passcodes[cid] = cid_user_answer.passcode
cids.add(cid)
s = cid_user_answer
user_names[cid] = cid
q = cid_user_answer.question.case
# if not s:
# # skip if no answer
# user_answers_marks[cid].append(0)
# user_answers[cid].append("")
# by_question[q].append(("", 0))
# continue
ans = s.answer
answer_score = s.get_answer_score()
if answer_score == "unmarked" or answer_score == None:
index = case_list.index(q)
unmarked.add(index)
user_answers[cid].append(ans)
user_answers_marks[cid].append(answer_score)
user_answers_and_marks[cid].append((ans, answer_score))
by_question[q][cid] = (ans, answer_score, s.feedback)
user_scores = {}
user_scores_normalised = {}
for user in user_answers_marks:
user_scores[user] = sum(
[i for i in user_answers_marks[user] if (i != "unmarked" and i != None)]
)
user_scores_normalised[user] = user_scores[user]
# ignore scores of 0 for stats
user_scores_list = [i for i in user_scores.values() if i > 0]
max_score = len(cases)
if len(user_scores_list) < 1:
mean = 0
median = 0
mode = 0
fig_html = ""
else:
mean = statistics.mean(user_scores_list)
median = statistics.median(user_scores_list)
try:
mode = statistics.mode(user_scores_list)
except statistics.StatisticsError:
mode = "No unique mode"
df = user_scores_list
fig = px.histogram(
df,
x=0,
title="{}: distribution of scores".format(collection),
labels={"0": "Score"},
height=400,
width=600,
)
fig_html = fig.to_html()
collection.stats_mean = mean
collection.stats_median = median
collection.stats_mode = mode
collection.stats_candidates = len(user_scores_list)
collection.stats_max_possible = max_score
collection.stats_min = min(user_scores_list)
collection.stats_max = max(user_scores_list)
# collection.stats_graph = fig_html
collection.user_scores = user_scores
collection.save()
return render(
request,
f"atlas/collection_scores.html",
{
"cids": sorted(cids),
# "cid_passcodes": cid_passcodes,
"exam": collection,
"collection": collection,
"unmarked": unmarked,
"questions": cases,
"by_question": by_question,
"user_answers": dict(user_answers),
"user_answers_marks": dict(user_answers_marks),
"user_scores": user_scores,
"user_scores_normalised": user_scores_normalised,
"user_scores_list": user_scores_list,
"user_names": user_names,
"user_answers_and_marks": user_answers_and_marks,
"max_score": max_score,
"mean": mean,
"median": median,
"mode": mode,
"plot": fig_html,
},
)
GenericExamViews = ExamViews(
CaseCollection, Case, None, CidReportAnswer, "atlas", "casecollection"
)
class CollectionCloneMixin:
def get_initial(self):
old_object = get_object_or_404(self.model, pk=self.kwargs["exam_id"])
initial_data = model_to_dict(old_object, exclude=["id"])
# We manually transfer the forign keys / m2m relationships
questions = old_object.cases.all().values_list("id", flat=True)
authors = old_object.author.all().values_list("id", flat=True)
self.cases = list(questions)
self.author = list(authors)
return initial_data
def form_valid(self, form):
object = form.save()
# Reapply these otherwise they get lost?
object.cases.set(self.cases)
object.author.set(self.author)
object.save()
return HttpResponseRedirect(object.get_absolute_url())
class CollectionClone(CollectionCloneMixin, CaseCollectionCreate):
"""Clone exam view"""
class SelfReviewDelete(DeleteView):
model = SelfReview
template_name = "confirm_delete.html"
# success_url = reverse_lazy("atlas:finding_view")
def get_initial(self):
user_exam_id = self.kwargs["user_exam_id"]
case_id = self.kwargs["case_id"]
review_id = self.kwargs["pk"]
self.user_exam = get_object_or_404(CidUserExam, pk=user_exam_id)
self.case = get_object_or_404(Case, pk=case_id)
self.review = get_object_or_404(SelfReview, pk=review_id)
return {"user_exam": user_exam_id, "case": case_id, "review": review_id}
def get_context_data(self, **kwargs):
# This can also be accessed via view.**** in the template
context = super(SelfReviewDelete, self).get_context_data(**kwargs)
context["user_exam"] = self.user_exam
context["case"] = self.case
context["review"] = self.review
return context
def get_success_url(self) -> str:
return self.user_exam.exam.get_case_take_url(self.case)
class SelfReviewUpdate(UpdateView):
model = SelfReview
form_class = SelfReviewForm
def get_initial(self):
user_exam_id = self.kwargs["user_exam_id"]
case_id = self.kwargs["case_id"]
review_id = self.kwargs["pk"]
self.user_exam = get_object_or_404(CidUserExam, pk=user_exam_id)
self.case = get_object_or_404(Case, pk=case_id)
self.review = get_object_or_404(SelfReview, pk=review_id)
return {"user_exam": user_exam_id, "case": case_id, "review": review_id}
def get_context_data(self, **kwargs):
# This can also be accessed via view.**** in the template
context = super(SelfReviewUpdate, self).get_context_data(**kwargs)
context["user_exam"] = self.user_exam
context["case"] = self.case
context["review"] = self.review
return context
def get_success_url(self) -> str:
return self.user_exam.exam.get_case_take_url(self.case)
# def get_object(self):
# return self.review
# return
# def get_initial(self):
# user_exam_id = self.kwargs["user_exam_id"]
# case_id = self.kwargs["case_id"]
# self.user_exam = get_object_or_404(CidUserExam, pk=user_exam_id)
# self.case = get_object_or_404(Case, pk=case_id)
# return {"user_exam": user_exam_id, "case": case_id}
class AddSelfReview(CreateView):
model = SelfReview
form_class = SelfReviewForm
# fields = ("author", "note_type", "note")
def get_form_kwargs(self):
kwargs = super(AddSelfReview, self).get_form_kwargs()
# update the kwargs for the form init method with yours
kwargs.update(self.kwargs) # self.kwargs contains all url conf params
# kwargs.update({"user": self.request.user})
return kwargs
def get_initial(self):
user_exam_id = self.kwargs["user_exam_id"]
case_id = self.kwargs["case_id"]
self.user_exam = get_object_or_404(CidUserExam, pk=user_exam_id)
self.case = get_object_or_404(Case, pk=case_id)
return {"user_exam": user_exam_id, "case": case_id}
def get_context_data(self, **kwargs):
# This can also be accessed via view.**** in the template
context = super(AddSelfReview, self).get_context_data(**kwargs)
context["user_exam"] = self.user_exam
context["case"] = self.case
return context
def get_success_url(self) -> str:
return self.user_exam.exam.get_case_take_url(self.case)
# return reverse("atlas:collection_take_overview_user", kwargs={"pk": self.user_exam.collection.id })
# def form_valid(self, form):
# model = form.save(commit=False)
#
# model.save()
#
# response = super().form_valid(form)
#
# return response
@login_required
def uncategorised_dicoms(request):
dicoms = UncategorisedDicom.objects.filter(user=request.user)
data = []
for d in dicoms:
data.append((d.image.name, d.get_basic_dicom_tags()))
return JsonResponse(data, safe=False)
@login_required
def combine_series(request):
"""Combines a list of series into the first series in the list
NOTE: this will only migrate images
it will not migrate other models such as SeriesFinding or other details
"""
series_ids = request.POST.getlist("series-ids")
# series = get_object_or_404(Series, pk=series_id)
if len(series_ids) < 2:
return HttpResponse("Not enough series to combine")
base_series = None
series_to_add = []
for n, series_id in enumerate(series_ids):
series = get_object_or_404(Series, pk=series_id)
if not series.check_user_can_edit(request.user):
return HttpResponse("You do not have permission to edit one of the series")
if n == 0:
base_series = series
else:
series_to_add.append(series)
for series in series_to_add:
for image in series.images.all():
image.series = base_series
image.save()
series.delete()
return HttpResponse(f"Series {series_ids} combines")
# return HttpResponse("Fail")
class CaseCollectionAuthorUpdate(RevisionMixin, AuthorRequiredMixin, UpdateView):
model = CaseCollection
form_class = CaseCollectionAuthorForm
template_name = "author_form.html"
def get_context_data(self, **kwargs):
context = super(CaseCollectionAuthorUpdate, self).get_context_data(**kwargs)
context["collection"] = context["object"]
return context
class ExamGroupsUpdate(ExamGroupsUpdateBase):
model = CaseCollection
form_class = ExamGroupsForm
def get_context_data(self, **kwargs):
context = super(ExamGroupsUpdate, self).get_context_data(**kwargs)
context["exam"] = context["object"]
return context
class CaseAuthorUpdate(RevisionMixin, AuthorRequiredMixin, UpdateView):
model = Case
form_class = CaseAuthorForm
template_name = "author_form.html"
def get_context_data(self, **kwargs):
context = super(CaseAuthorUpdate, self).get_context_data(**kwargs)
context["collection"] = context["object"]
return context
class SeriesAuthorUpdate(RevisionMixin, AuthorRequiredMixin, UpdateView):
model = Series
form_class = SeriesAuthorForm
template_name = "author_form.html"
def get_context_data(self, **kwargs):
context = super(SeriesAuthorUpdate, self).get_context_data(**kwargs)
context["collection"] = context["object"]
return context
def question_schemas_preset(request):
schemas = [
json.dumps(
{
"type": "object",
"title": "Questions",
"required": ["laterality", "diagnosis"],
"properties": {
"test": {
"type": "string",
"title": "test string",
"description": "What side is the lesion on?",
},
"diagnosis": {
"enum": [
"Meningioma",
"Schwannoma",
"Neurofibroma",
"Ependymoma",
],
"type": "string",
"title": "Diagnosis",
"description": "What is the diagnosis?",
},
"laterality": {
"enum": ["left", "right"],
"type": "string",
"title": "Laterality",
"widget": "radio",
"description": "What side is the lesion on?",
},
"laterality (copy 2)": {
"enum": ["left", "right"],
"type": "string",
"title": "Lateralityaoue",
"widget": "radio",
"description": "What side is the lesion on?",
},
},
}
),
]
return render(
request,
"atlas/question_schemas_preset.html",
{
"schemas": schemas,
},
)
def collection_question_schemas(request, exam_id: int):
collection = get_object_or_404(CaseCollection, pk=exam_id)
questions = collection.cases.all().prefetch_related()
casedetails = (
CaseDetail.objects.filter(case__in=questions, collection=collection)
.prefetch_related("case")
.order_by("sort_order")
)
return render(
request,
"atlas/collection_question_schemas.html",
{
"collection": collection,
"casedetails": casedetails,
},
)
@user_is_collection_author_or_atlas_editor
def collection_sync_prerequisite_users(request, pk: int):
"""Copy valid cid_users and user_users from all prerequisite collections into this collection.
This endpoint is intended to be called via HTMX from the collection detail page.
"""
if not request.htmx:
raise Http404("Invalid request")
collection = get_object_or_404(CaseCollection, pk=pk)
# Collect the closure of collections: the target collection and all prerequisites (recursively)
to_visit = [collection]
visited = {}
while to_visit:
col = to_visit.pop()
if col.pk in visited:
continue
visited[col.pk] = col
for p in col.prerequisites.all():
if p.pk not in visited:
to_visit.append(p)
collections = list(visited.values())
# Compute the union of all cid users and user users across the set
union_cid_users = set()
union_user_users = set()
union_cid_groups = set()
union_user_groups = set()
for col in collections:
union_cid_users.update(list(col.valid_cid_users.all()))
union_user_users.update(list(col.valid_user_users.all()))
union_cid_groups.update(list(col.cid_user_groups.all()))
union_user_groups.update(list(col.user_user_groups.all()))
# Apply the union to each collection and count additions
total_added_cid = 0
total_added_user = 0
total_added_cid_groups = 0
total_added_user_groups = 0
per_collection_added = []
for col in collections:
added_cid = 0
added_user = 0
added_cid_groups = 0
added_user_groups = 0
for cid_user in union_cid_users:
if not col.valid_cid_users.filter(pk=cid_user.pk).exists():
col.valid_cid_users.add(cid_user)
added_cid += 1
for user in union_user_users:
if not col.valid_user_users.filter(pk=user.pk).exists():
col.valid_user_users.add(user)
added_user += 1
# Sync groups as well
for gid in union_cid_groups:
if not col.cid_user_groups.filter(pk=gid.pk).exists():
col.cid_user_groups.add(gid)
added_cid_groups += 1
for ug in union_user_groups:
if not col.user_user_groups.filter(pk=ug.pk).exists():
col.user_user_groups.add(ug)
added_user_groups += 1
added_user += added_user_groups
added_cid += added_cid_groups
if added_cid or added_user:
col.save()
total_added_cid += added_cid
total_added_user += added_user
total_added_cid_groups += added_cid_groups
total_added_user_groups += added_user_groups
per_collection_added.append((col.pk, added_cid, added_user, added_cid_groups, added_user_groups))
html = render_to_string(
"atlas/partials/sync_prereq_result.html",
{
"total_added_cid": total_added_cid,
"total_added_user": total_added_user,
"total_added_cid_groups": total_added_cid_groups,
"total_added_user_groups": total_added_user_groups,
"per_collection_added": per_collection_added,
},
request=request,
)
return HttpResponse(html)
def collection_reset_answers_user_list(request, exam_id: int):
if request.htmx:
collection = get_object_or_404(CaseCollection, pk=exam_id)
cid_users = collection.cid_users.all()
if cid_users:
return render(
request,
"atlas/collection_reset_answers_user_list.html",
{"collection": collection, "cid_users": cid_users},
)
else:
return HttpResponse("No user answers.")
else:
raise Http404("Invalid request")
@user_is_collection_author_or_atlas_editor
def collection_reset_answers_user(request, exam_id: int, user_id: int):
if request.htmx:
collection = get_object_or_404(CaseCollection, pk=exam_id)
# Select all case details (answers are linked to these)
casedetails = collection.casedetail_set.all().prefetch_related()
# Delete all answers
for case in casedetails:
user_answers = case.userreportanswer_set.filter(user_id=user_id)
user_answers.delete()
collection.cid_users.filter(user_user_id=user_id).delete()
collection.exam_user_status.filter(cid_user_exam__user_user_id=user_id).delete()
return HttpResponse(f"User [{user_id}] answers deleted")
else:
raise Http404("Invalid request")
@user_is_collection_author_or_atlas_editor
def collection_reset_answers_ciduser(request, exam_id: int, cid: int):
if request.htmx:
collection = get_object_or_404(CaseCollection, pk=exam_id)
# Select all case details (answers are linked to these)
casedetails = collection.casedetail_set.all().prefetch_related()
# Delete all answers
for case in casedetails:
cid_answers = case.cidreportanswer_set.filter(cid=cid)
cid_answers.delete()
collection.cid_users.filter(cid=cid).delete()
collection.exam_user_status.filter(cid_user_exam__cid__pk=cid).delete()
return HttpResponse(f"CID [{cid}] answers deleted")
else:
raise Http404("Invalid request")
def collection_reset_answers(request, exam_id: int):
if request.htmx:
collection = get_object_or_404(CaseCollection, pk=exam_id)
# Select all case details (answers are linked to these)
casedetails = collection.casedetail_set.all().prefetch_related()
# Delete all answers
for case in casedetails:
cid_answers = case.cidreportanswer_set.all()
cid_answers.delete()
user_answers = case.userreportanswer_set.all()
user_answers.delete()
# User statuses
collection.exam_user_status.all().delete()
# CidUserExams
collection.cid_users.all().delete()
return HttpResponse("Success")
else:
raise Http404("Invalid request")
def push_case_to_cimar(request, case_id):
push_case_to_cimar_task.delay(case_id=case_id)
return HttpResponse("Push started")
case = get_object_or_404(Case, pk=case_id)
api = CimarAPI()
api.login(username=CIMAR_USERNAME, password=CIMAR_PASSWORD)
# We use the same study_uid for all images in a case
study_uid = generate_uid()
for series in case.get_series():
for image in series.images.all():
data = api.upload_dicom(image.image.path, study_uid=study_uid)
cimar_uuid = api.get_study_by_study_uid(data["study_uid"])["uuid"]
case.cimar_uuid = cimar_uuid
case.save()
cimar_case, created = CimarCase.objects.get_or_create(uuid=cimar_uuid)
cimar_case.refresh_study()
return HttpResponse("Success")
@require_POST
def remove_case_from_collection(request, case_pk, collection_pk):
case = get_object_or_404(Case, pk=case_pk)
collection = get_object_or_404(CaseCollection, pk=collection_pk)
# Remove the case from the collection
collection.cases.remove(case)
# If this was an HTMX request, return an empty response so the caller
# can swap out (remove) the surrounding element (e.g., the <li>).
if request.htmx:
return HttpResponse("")
return HttpResponse(f"Case removed from collection: {collection.name}")
@require_POST
def remove_selected_series_from_case(request, case_pk):
ids = request.POST.getlist('series-ids')
case = Case.objects.get(pk=case_pk)
if not case.check_user_can_edit(request.user):
return HttpResponse("You do not have permission to edit this case.", status=403)
if not ids:
return HttpResponse("No series selected for removal.")
# Remove the association (for a through model)
SeriesDetail.objects.filter(case=case, series_id__in=ids).delete()
return HttpResponse(f"Series removed from case.")
@require_POST
def reorder_series(request, case_pk):
case = get_object_or_404(Case, pk=case_pk)
if not case.check_user_can_edit(request.user):
return HttpResponse("You do not have permission to edit this case.", status=403)
# Get new order from POST
order = []
for key in sorted([k for k in request.POST if k.startswith("series_order_")]):
order.append(int(request.POST[key]))
# Update sort_order in SeriesDetail
for idx, series_id in enumerate(order):
SeriesDetail.objects.filter(case=case, series_id=series_id).update(sort_order=idx)
return redirect(request.META.get("HTTP_REFERER", "/"))
def linked_cases_overview(request, case_id):
"""
View to display an overview of linked cases via the previous_case field.
"""
# Get the current case
case = get_object_or_404(Case, pk=case_id)
# Traverse backward to get all previous cases
previous_cases = []
try:
current_case = case.previous_case
while current_case:
previous_cases.insert(0, current_case) # Insert at the beginning to maintain order
current_case = current_case.previous_case
except ObjectDoesNotExist:
pass
# Traverse forward to get all next cases
next_cases = []
try:
current_case = case.next_case
while current_case:
next_cases.append(current_case)
current_case = current_case.next_case
except ObjectDoesNotExist:
pass
# Render the template with the linked cases
return render(
request,
"atlas/linked_cases_overview.html",
{
"current_case": case,
"previous_cases": previous_cases,
"next_cases": next_cases,
},
)
@login_required
def add_series_to_case(request, series_pk):
if request.method == "POST":
case_id = request.POST.get("case_id")
case = get_object_or_404(Case, pk=case_id)
if not case.check_user_can_edit(request.user):
raise PermissionDenied("You do not have permission to edit this case.")
series = get_object_or_404(Series, pk=series_pk)
# Associate the series with the case
series.case.add(case)
# If this is an HTMX request, return the updated cases list partial so the
# client can replace the #cases-list innerHTML. Otherwise return a simple message.
if request.headers.get("Hx-Request") or request.META.get("HTTP_HX_REQUEST"):
return render(request, 'atlas/partials/series_cases_list.html', {'series': series, 'can_edit': case.check_user_can_edit(request.user)})
return HttpResponse(f"Series {series_pk} added to case {case.title} ({case_id}).")
return HttpResponse("Invalid request.")
@login_required
def remove_series_from_case(request, series_pk, case_pk):
if request.method == "POST":
case = get_object_or_404(Case, pk=case_pk)
if not case.check_user_can_edit(request.user):
raise PermissionDenied("You do not have permission to edit this case.")
series = get_object_or_404(Series, pk=series_pk)
# Remove the association
series.case.remove(case)
# If this is an HTMX request, return the updated cases list partial so the
# client can replace the #cases-list innerHTML. Otherwise return a simple message.
if request.headers.get("Hx-Request") or request.META.get("HTTP_HX_REQUEST"):
return render(request, 'atlas/partials/series_cases_list.html', {'series': series, 'can_edit': case.check_user_can_edit(request.user)})
return HttpResponse("Removed")
return HttpResponse("Invalid request.")
@login_required
def series_finding_details(request, finding_id):
finding = get_object_or_404(SeriesFinding, pk=finding_id)
return render(request, 'atlas/series_finding_details.html', {'finding': finding})
@login_required
def series_finding_related(request, series_id):
series = get_object_or_404(Series, pk=series_id)
findings = series.get_related_findings()
return render(request, 'atlas/series_finding_related.html', {'series': series, 'findings': findings})
def seriesfinding_migration_status(request):
"""
View to list all SeriesFinding objects and their migration status.
Shows whether annotation_json_3d and viewer_state_3d are populated.
"""
findings = SeriesFinding.objects.all().select_related()
findings_status = []
for finding in findings:
findings_status.append({
"id": finding.id,
"str": str(finding),
"annotation_json_3d": bool(finding.annotation_json_3d),
"viewer_state_3d": bool(finding.viewer_state_3d),
"series": finding.series
})
return render(request, "atlas/seriesfinding_migration_status.html", {
"findings_status": findings_status
})
def series_bulk_delete(request):
if request.method == "POST":
ids = request.POST.getlist("series_ids")
# Only delete series the user has permission to delete
series_to_delete = Series.objects.filter(pk__in=ids)
deleted_ids = []
for series in series_to_delete:
if series.check_user_can_edit(request.user):
series.delete()
deleted_ids.append(series.pk)
# Optionally, you could add a message here about which were deleted
return redirect('atlas:series_view')
@user_is_collection_author_or_atlas_editor
@csrf_exempt # Only if you have CSRF issues; otherwise, keep CSRF protection
def collection_case_displaysetup(request, collection_id, case_number=None, case_id=None):
"""
View to set up the display options for a specific case in a collection.
This can include setting which series to display, annotations, etc.
"""
# Accept either case_number (index into collection) or case_id (case PK).
casedetail = None
try:
# If a case_id was passed explicitly, try that first
if case_id is not None:
casedetail = get_object_or_404(CaseDetail, case__pk=case_id, collection__pk=collection_id)
else:
# Otherwise the route most likely supplied case_number; try resolving by index
collection = get_object_or_404(CaseCollection, pk=collection_id)
case_obj = collection.get_case_by_index(int(case_number))
casedetail = CaseDetail.objects.get(case=case_obj, collection=collection)
except Http404:
raise
except Exception:
raise Http404("Case not found in collection")
if request.method == "POST":
data = json.loads(request.body.decode("utf-8"))
if data.get("reset_viewerstate"):
casedetail.default_viewerstate = None
casedetail.save()
return HttpResponse("<span class='text-success'>Viewer state reset to default.</span>")
viewerstate = data.get("viewerstate")
if viewerstate is not None:
casedetail.default_viewerstate = viewerstate
casedetail.save()
return HttpResponse("<span class='text-success'>Viewer state saved.</span>")
else:
return HttpResponse("<span class='text-danger'>No viewer state provided.</span>", status=400)
collection = casedetail.collection
case_number, case_count = collection.get_index_of_case(
casedetail.case, case_count=True
)
previous = collection.get_previous_case(casedetail.case)
next = collection.get_next_case(casedetail.case)
return render(request, "atlas/collection_case_displaysetup.html", {
"casedetail": casedetail,
"collection": collection,
"case_number": case_number,
"case_count": case_count,
"previous": previous,
"next": next,
"collection_length": case_count,
})
Reference in New Issue View Git Blame Copy Permalink