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 series_truncate_htmx(request, series_id, start, end): """HTMX endpoint to truncate a series' images to the given inclusive start/end bounds (zero-based). Returns an HTML fragment with the result. """ series = get_object_or_404(Series, pk=series_id) if not series.check_user_can_edit(request.user): return HttpResponse("
Permission denied
") images_removed = [] for n, image in enumerate(series.get_images()): if n >= int(start) and n <= int(end): # keep images within bounds continue else: image.removed = True image.image = None image.save() images_removed.append(image.pk) # Recalculate stored series size try: series.get_total_image_size(refresh=True) except Exception: # ignore errors when recalculating pass # Render a simple message fragment for HTMX swap html = render_to_string( "atlas/partials/truncate_result.html", {"images_removed": images_removed, "series": series}, request=request, ) return HttpResponse(html) @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 def case_search_page(request): """Full page case search with quick filters. Supports GET params: - q: query string - field: one of ['all','title','description','history','author'] - my: '1' to limit to cases authored by current user - open: '1' to limit to open_access cases - has_images: '1' to require series images - subspecialty: id of subspecialty to filter - page: pagination page """ q = request.GET.get("q", "").strip() field = request.GET.get("field", "all") my_cases = request.GET.get("my", "0") == "1" open_access = request.GET.get("open", "0") == "1" has_images = request.GET.get("has_images", "0") == "1" subspecialty_id = request.GET.get("subspecialty") qs = get_cases_available_to_user(request.user) if my_cases: qs = qs.filter(author=request.user) if open_access: qs = qs.filter(open_access=True) if subspecialty_id: try: sid = int(subspecialty_id) qs = qs.filter(subspecialty__id=sid) except (ValueError, TypeError): pass if q: if field == "title": qs = qs.filter(title__icontains=q) elif field == "description": qs = qs.filter(description__icontains=q) elif field == "history": qs = qs.filter(history__icontains=q) elif field == "author": qs = qs.filter( Q(author__first_name__icontains=q) | Q(author__last_name__icontains=q) | Q(author__username__icontains=q) ) else: 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) | 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() if has_images: qs = qs.filter(series__images__isnull=False).distinct() qs = qs.order_by("-created_date") # Paginate results paginator = Paginator(qs, 25) page_number = request.GET.get("page") page_obj = paginator.get_page(page_number) # Recent cases (authored by current user) recent_cases = get_cases_available_to_user(request.user).filter(author=request.user).order_by("-created_date")[:5] subspecialty_queryset = Subspecialty.objects.all().order_by("name") return render( request, "atlas/case_search_page.html", { "cases": page_obj, "q": q, "field": field, "my": my_cases, "open": open_access, "has_images": has_images, "subspecialty_queryset": subspecialty_queryset, "selected_subspecialty": subspecialty_id, "recent_cases": recent_cases, }, ) @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= return redirect(f"{reverse('atlas:case_displaysets', args=[case.pk])}?displayset={pk}") @login_required def case_displaysets_modal(request, pk): """Return HTML fragment for a CaseDisplaySet to be shown inside a modal.""" displayset = get_object_or_404(CaseDisplaySet, pk=pk) can_edit = False try: can_edit = displayset.case.check_user_can_edit(request.user) except Exception: can_edit = False return render(request, 'atlas/case_displayset_modal.html', {'ds': displayset, 'can_edit': can_edit}) 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("
  • 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) return render( request, "atlas/case_detail.html", { "case": case, "cimar_sid": request.user.userprofile.cimar_sid, "can_edit": can_edit, # 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: rely on `can_edit` check on the Case model can_edit = case.check_user_can_edit(request.user) if not can_edit: 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_edit = case.check_user_can_edit(request.user) 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_edit": can_edit}, 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: rely on `can_edit` helper on the Case model can_edit = case.check_user_can_edit(request.user) if not can_edit: 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: rely on `can_edit` helper on the Case model can_edit = case.check_user_can_edit(request.user) if not can_edit: 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_edit = case.check_user_can_edit(request.user) 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_edit": can_edit}, request=request, ) toggle_html += "".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 = ( '' ) 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) # Resolve the Case early so templates always receive it when available try: case = Case.objects.get(pk=case_id) except Exception: case = None # If called via HTMX (inline from case view) return a compact fragment is_htmx = getattr(request, "htmx", False) or request.headers.get("HX-Request", "").lower() == "true" if is_htmx: # Provide the 10 most recently updated collections for this user by default user = request.user recent_base = ( CaseCollection.objects.filter(Q(author=user) | Q(markers=user)) .distinct() .order_by("-updated") ) # Exclude collections that already contain the case (if resolved) if case is not None: recent_base = recent_base.exclude(pk__in=case.casecollection_set.values_list("pk", flat=True)) recent_qs = recent_base[:10] html = render_to_string( "atlas/partials/add_collection_inline.html", {"recent_collections": recent_qs, "case": case}, request=request, ) return HttpResponse(html) return render( request, "generic_form.html", { "form": form, "target_url": reverse("atlas:case_collection_form", args=(case_id,)), }, ) @login_required def collection_inline_search(request, case_id): # HTMX endpoint: return recent collections or search results for the user if not request.htmx: return Http404 q = request.GET.get("q", "").strip() user = request.user base_qs = ( CaseCollection.objects.filter(Q(author=user) | Q(markers=user) | Q(valid_user_users=user) | Q(open_access=True)) .distinct() ) try: case = Case.objects.get(pk=case_id) except Exception: case = None # Exclude collections already containing the case if case is not None: base_qs = base_qs.exclude(pk__in=case.casecollection_set.values_list("pk", flat=True)) if q: results = base_qs.filter(name__icontains=q).order_by("-updated")[:30] else: results = base_qs.order_by("-updated")[:10] html = render_to_string( "atlas/partials/collection_inline_list.html", {"collections": results, "case": case}, request=request, ) return HttpResponse(html) @login_required @require_http_methods(["POST"]) def add_collection_to_case_inline(request, case_id, collection_id): # HTMX endpoint to add a single collection to a case and return the new list item if not request.htmx: return Http404 case = get_object_or_404(Case, pk=case_id) collection = get_object_or_404(CaseCollection, pk=collection_id) # Permission: user must be able to edit the case if not case.check_user_can_edit(request.user): return HttpResponse("You do not have permission to add collections to this case.", status=403) # Add the association if collection in case.casecollection_set.all(): return HttpResponse("Already in collection") case.casecollection_set.add(collection) # Render a list item to insert into the case's collection list rendered = render_to_string( "atlas/partials/collection_list_item.html", {"collection": collection, "case": case, "can_edit": case.check_user_can_edit(request.user)}, request=request, ) # Also instruct the client to remove the collection entry from the inline search list remove_script = ( '' ) return HttpResponse(rendered + remove_script) # 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('' % (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 series.check_user_can_edit(request.user): 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("

    Series does not appear to contain dicoms

    ") 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"
    {dict_diff}

    {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"
    {dict_diff}
    ") 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( "

    Series does not appear to contain dicoms (or field InstanceNumber)

    " ) 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( "

    Series does not appear to contain dicoms (or field SeriesInstanceUID)

    " ) 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", # ) #) # Prefetch and annotate aggressively to avoid N+1 queries in the viva view. # - Prefetch `case__series` using the ordering from SeriesDetail # - Prefetch only non-removed images for those series (ordered by position) # - Prefetch case resources and display sets from django.db.models import Prefetch as _Prefetch series_images_qs = SeriesImage.objects.filter(removed=False).order_by("position") series_qs = Series.objects.order_by("seriesdetail__sort_order").prefetch_related( _Prefetch("images", queryset=series_images_qs) ) casedetails = ( CaseDetail.objects .filter(collection=collection) .select_related("case") .order_by("sort_order") .prefetch_related( _Prefetch("case__series", queryset=series_qs), _Prefetch("case__series__modality"), _Prefetch("case__series__examination"), _Prefetch("case__series__plane"), _Prefetch("case__series__contrast"), _Prefetch("case__caseresource_set"), _Prefetch("case__display_sets"), ) .annotate(series_count=Count("case__seriesdetail")) ) 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_number): return collection_case_dicom_json(request, exam_id, case_number, review=True) def collection_case_dicom_json(request, exam_id, case_number, review=False): # Support either case_number (index) 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 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 def resource_search_panel(request): """HTMX endpoint: return the full resource search panel HTML fragment. Expects optional GET param `case_id` to prefill the hidden field. """ case_id = request.GET.get("case_id") return render(request, "atlas/partials/resource_search_panel.html", { "case": (Case.objects.filter(pk=case_id).first() if case_id else None), "site_queryset": Site.objects.all(), "subspecialty_queryset": Subspecialty.objects.all(), }) @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
  • ). 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("Viewer state reset to default.") viewerstate = data.get("viewerstate") if viewerstate is not None: casedetail.default_viewerstate = viewerstate casedetail.save() return HttpResponse("Viewer state saved.") else: return HttpResponse("No viewer state provided.", 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, })