Ross
3636 lines
119 KiB
Python
Executable File
3636 lines
119 KiB
Python
Executable File
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.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.views.generic.edit import CreateView, UpdateView, DeleteView, FormView
|
|
from django.views.generic import ListView
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
|
|
from django.urls import reverse_lazy, reverse
|
|
|
|
from django.http import Http404, JsonResponse
|
|
from django.http import HttpResponseRedirect, HttpResponse
|
|
|
|
from generic.models import CidUser, CidUserExam, CimarCase
|
|
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,
|
|
CaseDisplaySetForm,
|
|
CaseForm,
|
|
CaseResourceFormSet,
|
|
CaseUpdateSeriesForm,
|
|
CidQuestionAnswerForm,
|
|
CidReportAnswerForm,
|
|
CidReportAnswerMarkForm,
|
|
ConditionAutocompleteForm,
|
|
ConditionForm,
|
|
ExamGroupsForm,
|
|
FindingForm,
|
|
JsonAnswerForm,
|
|
#ModelPriorCaseForm,
|
|
PriorCaseForm,
|
|
QuestionSchemaForm,
|
|
ResourceForm,
|
|
SelfReviewForm,
|
|
SeriesAuthorForm,
|
|
SeriesForm,
|
|
SeriesImageFormSet,
|
|
SeriesFormSet,
|
|
SeriesFindingForm,
|
|
CaseDifferentialFormSet,
|
|
StructureForm,
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
|
|
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__)
|
|
|
|
|
|
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
|
|
|
|
@login_required
|
|
@user_has_case_view_access
|
|
def case_displaysets(request, pk):
|
|
case = get_object_or_404(Case, pk=pk)
|
|
|
|
can_edit = case.check_user_can_edit(request.user)
|
|
|
|
# Get displayset id from GET parameter
|
|
displayset_id = request.GET.get("displayset")
|
|
edit = request.GET.get("edit", False)
|
|
selected_displayset = None
|
|
if displayset_id:
|
|
try:
|
|
selected_displayset = case.display_sets.get(pk=displayset_id)
|
|
except CaseDisplaySet.DoesNotExist:
|
|
selected_displayset = None
|
|
|
|
return render(
|
|
request,
|
|
"atlas/case_displaysets.html",
|
|
{
|
|
"case": case,
|
|
"edit": edit,
|
|
"can_edit": can_edit,
|
|
"selected_displayset": selected_displayset,
|
|
},
|
|
)
|
|
|
|
@login_required
|
|
@user_has_case_view_access
|
|
@require_http_methods(["GET", "POST"])
|
|
def case_displaysets_add(request, pk):
|
|
case = get_object_or_404(Case, pk=pk)
|
|
form = CaseDisplaySetForm(request.POST or None)
|
|
if request.method == "POST":
|
|
if form.is_valid():
|
|
ds = form.save(commit=False)
|
|
ds.case = case
|
|
ds.author = request.user
|
|
ds.viewerstate = request.POST.get("viewerstate") or None
|
|
ds.annotations = request.POST.get("annotations") or None
|
|
ds.save()
|
|
form.save_m2m()
|
|
# HTMX: return the new display set row HTML fragment
|
|
html = render_to_string("atlas/partials/displayset_row.html", {"ds": ds, "can_edit": True})
|
|
return HttpResponse(html)
|
|
else:
|
|
# HTMX: return the form HTML fragment with errors
|
|
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
|
|
return HttpResponse(html, status=400)
|
|
else:
|
|
# HTMX: return the blank form HTML fragment
|
|
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
|
|
return HttpResponse(html)
|
|
|
|
@login_required
|
|
@require_http_methods(["GET", "POST"])
|
|
def case_displaysets_edit(request, pk):
|
|
ds = get_object_or_404(CaseDisplaySet, pk=pk)
|
|
|
|
case = ds.case
|
|
if not case.check_user_can_edit(request.user):
|
|
return HttpResponse("You do not have permission to delete this display set.")
|
|
form = CaseDisplaySetForm(request.POST or None, instance=ds)
|
|
if request.method == "POST":
|
|
if form.is_valid():
|
|
ds = form.save(commit=False)
|
|
ds.viewerstate = request.POST.get("viewerstate") or ds.viewerstate
|
|
ds.annotations = request.POST.get("annotations") or ds.annotations
|
|
ds.save()
|
|
form.save_m2m()
|
|
# Return updated row HTML
|
|
html = render_to_string("atlas/partials/displayset_row.html", {"ds": ds, "can_edit": True}, request=request)
|
|
return HttpResponse(html)
|
|
else:
|
|
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
|
|
return HttpResponse(html, status=400)
|
|
else:
|
|
html = render_to_string("atlas/partials/displayset_form.html", {"form": form, "case": case}, request=request)
|
|
return HttpResponse(html)
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def case_displaysets_detail(request, pk):
|
|
displayset = get_object_or_404(CaseDisplaySet, pk=pk)
|
|
case = displayset.case
|
|
# Redirect to the case_displaysets page with ?displayset=<pk>
|
|
return redirect(f"{reverse('atlas:case_displaysets', args=[case.pk])}?displayset={pk}")
|
|
|
|
def case_displaysets_delete(request, pk):
|
|
try:
|
|
displayset = CaseDisplaySet.objects.get(pk=pk)
|
|
except CaseDisplaySet.DoesNotExist:
|
|
raise Http404("Display set not found.")
|
|
|
|
if not displayset.case.check_user_can_edit(request.user):
|
|
raise PermissionDenied("You do not have permission to delete this display set.")
|
|
displayset.delete()
|
|
return HttpResponse("<li class='list-group-item' >Display set deleted successfully.")
|
|
|
|
@login_required
|
|
@user_has_case_view_access
|
|
def case_detail(request, pk):
|
|
# Prefetch all related objects needed for the template
|
|
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()]
|
|
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,
|
|
},
|
|
)
|
|
|
|
@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_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 = ConditionAutocompleteForm()
|
|
|
|
can_merge = False
|
|
|
|
if (
|
|
request.user.is_superuser
|
|
or request.user.groups.filter(name="atlas_editor").exists()
|
|
):
|
|
can_merge = True
|
|
|
|
# logging.debug(atlas.subspecialty.first().name.all())
|
|
return render(
|
|
request,
|
|
"atlas/condition_detail.html",
|
|
{"condition": condition, "form": 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})
|
|
|
|
|
|
def index(request):
|
|
return render(request, "atlas/index.html", {})
|
|
|
|
|
|
def user_collections(request):
|
|
collections = request.user.user_casecollection_exams.all()
|
|
|
|
return render(request, "atlas/user_collections.html", {"collections": collections})
|
|
|
|
@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 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)
|
|
return HttpResponse(f"Case added to collection {collection.name} (refresh to see)")
|
|
|
|
# if request.method == "POST":
|
|
|
|
cases = get_cases_available_to_user(request.user)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/add_case_to_collection.html",
|
|
{"cases": cases, "collection": collection},
|
|
)
|
|
|
|
|
|
@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, 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, ListView):
|
|
model = Resource
|
|
# form_class = ResourceForm
|
|
|
|
|
|
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)
|
|
|
|
if 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 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)
|
|
|
|
|
|
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 AtlasCreateBase(RevisionMixin, LoginRequiredMixin):
|
|
model = Case
|
|
form_class = CaseForm
|
|
|
|
def get_form_kwargs(self):
|
|
kwargs = super(AtlasCreateBase, self).get_form_kwargs()
|
|
kwargs.update({"user": self.request.user})
|
|
return kwargs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super(AtlasCreateBase, self).get_context_data(**kwargs)
|
|
if self.request.POST:
|
|
#context["series_formset"] = SeriesFormSet(
|
|
# self.request.POST,
|
|
# self.request.FILES,
|
|
# form_kwargs={"user": self.request.user},
|
|
#)
|
|
context["casedifferential_formset"] = CaseDifferentialFormSet(
|
|
self.request.POST, self.request.FILES
|
|
)
|
|
context["caseresource_formset"] = CaseResourceFormSet(
|
|
self.request.POST,
|
|
self.request.FILES,
|
|
form_kwargs={"user": self.request.user},
|
|
)
|
|
#context["series_formset"].full_clean()
|
|
context["casedifferential_formset"].full_clean()
|
|
context["caseresource_formset"].full_clean()
|
|
else:
|
|
# context["series_formset"] = SeriesFormSet(instance=Case.objects.filter(author__id=self.request.user.id), queryset=Series.objects.filter(author__id=self.request.user.id))
|
|
#context["series_formset"] = SeriesFormSet(
|
|
# form_kwargs={"user": self.request.user}
|
|
#)
|
|
context["casedifferential_formset"] = CaseDifferentialFormSet()
|
|
context["caseresource_formset"] = CaseResourceFormSet(
|
|
form_kwargs={"user": self.request.user}
|
|
)
|
|
return context
|
|
|
|
def form_valid(self, form):
|
|
self.object = form.save(commit=False)
|
|
|
|
# Add exam objects
|
|
|
|
self.object.save()
|
|
|
|
form.instance.author.add(self.request.user.id)
|
|
|
|
# Check Cimar Case
|
|
# CimarAPI().get_study_by_uuid(self.object.cimar_uuid)
|
|
|
|
context = self.get_context_data(form=form)
|
|
#series_formset = context["series_formset"]
|
|
casedifferential_formset = context["casedifferential_formset"]
|
|
caseresource_formset = context["caseresource_formset"]
|
|
if (
|
|
#series_formset.is_valid()
|
|
casedifferential_formset.is_valid()
|
|
and caseresource_formset.is_valid()
|
|
):
|
|
response = super().form_valid(form)
|
|
#series_formset.instance = self.object
|
|
#series_formset.save()
|
|
casedifferential_formset.instance = self.object
|
|
casedifferential_formset.save()
|
|
caseresource_formset.instance = self.object
|
|
caseresource_formset.save()
|
|
# If the normal submit button is pressed we save as normal
|
|
if "submit" in self.request.POST:
|
|
return response
|
|
# else we redirect to the clone url
|
|
else:
|
|
return redirect("atlas:case_clone", pk=self.object.pk)
|
|
|
|
else:
|
|
return super().form_invalid(form)
|
|
|
|
|
|
# @login_required
|
|
# CaseCreate
|
|
class AtlasCreate(AtlasCreateBase, CreateView):
|
|
def get_initial(self):
|
|
if "pk" in self.kwargs:
|
|
initial = super().get_initial()
|
|
exam = get_object_or_404(CaseCollection, pk=self.kwargs["pk"])
|
|
|
|
initial["exams"] = [exam.id]
|
|
|
|
return initial
|
|
## There has to be a better way...
|
|
# try:
|
|
# s = (i.pk for i in self.request.user.atlas_default.site.all())
|
|
# self.initial.update({"site": s})
|
|
# except AttributeError:
|
|
# pass
|
|
# return self.initial
|
|
|
|
|
|
# CaseUpdate
|
|
class AtlasUpdate(
|
|
RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView
|
|
):
|
|
model = Case
|
|
form_class = CaseForm
|
|
|
|
# fields = '__all__'
|
|
# #fields = [ 'condition' ]
|
|
# #initial = {'date_of_death': '05/01/2018'}
|
|
# exclude = [ 'created_date', 'published_date' ]
|
|
|
|
|
|
#def get_object(self, queryset=None):
|
|
# # Prefetch all related objects needed for the form and formsets
|
|
# qs = self.model.objects.select_related(
|
|
# # Add all FK fields here
|
|
# ).prefetch_related(
|
|
# "series__examination",
|
|
# "series__plane",
|
|
# "series__contrast",
|
|
# "series__images",
|
|
# "subspecialty",
|
|
# "condition",
|
|
# "presentation",
|
|
# "pathological_process",
|
|
# "differentialcase",
|
|
# "caseresource_set",
|
|
# )
|
|
# return qs.get(pk=self.kwargs["pk"])
|
|
|
|
def get_form_kwargs(self):
|
|
kwargs = super(AtlasUpdate, self).get_form_kwargs()
|
|
kwargs.update({"user": self.request.user})
|
|
return kwargs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super(AtlasUpdate, self).get_context_data(**kwargs)
|
|
if self.request.POST:
|
|
#context["series_formset"] = SeriesFormSet(
|
|
# self.request.POST,
|
|
# self.request.FILES,
|
|
# instance=self.object,
|
|
# form_kwargs={"user": self.request.user},
|
|
#)
|
|
context["casedifferential_formset"] = CaseDifferentialFormSet(
|
|
self.request.POST, self.request.FILES, instance=self.object
|
|
)
|
|
context["caseresource_formset"] = CaseResourceFormSet(
|
|
self.request.POST,
|
|
self.request.FILES,
|
|
instance=self.object,
|
|
form_kwargs={"user": self.request.user},
|
|
)
|
|
#context["series_formset"].full_clean()
|
|
context["casedifferential_formset"].full_clean()
|
|
context["caseresource_formset"].full_clean()
|
|
else:
|
|
#context["series_formset"] = SeriesFormSet(
|
|
# instance=self.object, form_kwargs={"user": self.request.user}
|
|
#)
|
|
context["casedifferential_formset"] = CaseDifferentialFormSet(
|
|
instance=self.object
|
|
)
|
|
context["caseresource_formset"] = CaseResourceFormSet(
|
|
instance=self.object,
|
|
form_kwargs={"user": self.request.user},
|
|
)
|
|
return context
|
|
|
|
def form_valid(self, form):
|
|
self.object = form.save(commit=False)
|
|
self.object.save()
|
|
|
|
form.instance.author.add(self.request.user.id)
|
|
|
|
context = self.get_context_data(form=form)
|
|
#series_formset = context["series_formset"]
|
|
casedifferential_formset = context["casedifferential_formset"]
|
|
caseresource_formset = context["caseresource_formset"]
|
|
# logger.debug(formset.is_valid())
|
|
if (
|
|
#series_formset.is_valid()
|
|
casedifferential_formset.is_valid()
|
|
and caseresource_formset.is_valid()
|
|
):
|
|
response = super().form_valid(form)
|
|
#series_formset.instance = self.object
|
|
#series_formset.save()
|
|
casedifferential_formset.instance = self.object
|
|
casedifferential_formset.save()
|
|
caseresource_formset.instance = self.object
|
|
caseresource_formset.save()
|
|
|
|
return response
|
|
else:
|
|
return super().form_invalid(form)
|
|
|
|
class CollectionCaseUpdate(
|
|
RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView):
|
|
model = CaseCollection
|
|
form_class = CaseCollectionUpdateCaseForm
|
|
template_name = "atlas/casecollection_update_case.html"
|
|
|
|
def get_form_kwargs(self):
|
|
kwargs = super().get_form_kwargs()
|
|
kwargs.update({"user": self.request.user, "collection": self.object})
|
|
|
|
return kwargs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
if self.request.POST:
|
|
context["case_formset"] = CaseCollectionCaseFormSet(
|
|
self.request.POST,
|
|
self.request.FILES,
|
|
instance=self.object,
|
|
form_kwargs={"user": self.request.user, "collection": self.object},
|
|
)
|
|
context["case_formset"].full_clean()
|
|
else:
|
|
context["case_formset"] = CaseCollectionCaseFormSet(
|
|
instance=self.object,
|
|
form_kwargs={"user": self.request.user, "collection": self.object},
|
|
)
|
|
context["collection"] = context["casecollection"]
|
|
return context
|
|
|
|
def form_valid(self, form):
|
|
context = self.get_context_data(form=form)
|
|
case_formset = context["case_formset"]
|
|
if case_formset.is_valid():
|
|
case_formset.instance = self.object
|
|
case_formset.save()
|
|
return super().form_valid(form)
|
|
else:
|
|
return self.form_invalid(form)
|
|
|
|
|
|
class AtlasSeriesUpdate(
|
|
RevisionMixin, LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView
|
|
):
|
|
model = Case
|
|
form_class = CaseUpdateSeriesForm
|
|
template_name = "atlas/case_update_series.html"
|
|
|
|
def get_form_kwargs(self):
|
|
kwargs = super().get_form_kwargs()
|
|
kwargs.update({"user": self.request.user})
|
|
return kwargs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
is_atlas_editor = self.request.user.groups.filter(name="atlas_editor").exists() or self.request.user.is_superuser
|
|
form_kwargs = {"user": self.request.user, "is_atlas_editor": is_atlas_editor}
|
|
|
|
if not is_atlas_editor:
|
|
queryset = Series.objects.filter(author__id=self.request.user.id)
|
|
else:
|
|
queryset = Series.objects.all()
|
|
queryset = queryset.select_related("modality").prefetch_related(
|
|
"examination", "plane", "contrast",
|
|
Prefetch("images", queryset=SeriesImage.objects.filter(removed=False).order_by("position"))
|
|
|
|
)
|
|
|
|
if self.request.POST:
|
|
context["series_formset"] = SeriesFormSet(
|
|
self.request.POST,
|
|
self.request.FILES,
|
|
queryset=queryset,
|
|
instance=self.object,
|
|
form_kwargs=form_kwargs,
|
|
)
|
|
context["series_formset"].full_clean()
|
|
else:
|
|
context["series_formset"] = SeriesFormSet(
|
|
queryset=queryset,
|
|
instance=self.object,
|
|
form_kwargs=form_kwargs,
|
|
)
|
|
return context
|
|
|
|
def form_valid(self, form):
|
|
context = self.get_context_data(form=form)
|
|
series_formset = context["series_formset"]
|
|
if series_formset.is_valid():
|
|
series_formset.instance = self.object
|
|
series_formset.save()
|
|
return super().form_valid(form)
|
|
else:
|
|
return self.form_invalid(form)
|
|
|
|
def get_object(self, queryset=None):
|
|
qs = Case.objects.select_related().prefetch_related(
|
|
"seriesdetail_set__series__modality",
|
|
"seriesdetail_set__series__examination",
|
|
"seriesdetail_set__series__plane",
|
|
"seriesdetail_set__series__contrast",
|
|
"seriesdetail_set__series__images",
|
|
)
|
|
return qs.get(pk=self.kwargs["pk"])
|
|
|
|
def add_collection_to_case_form(request, case_id):
|
|
if request.method == "POST":
|
|
form = AddCollectionToCaseForm(request.POST, user=request.user)
|
|
if form.is_valid():
|
|
case = get_object_or_404(Case, pk=case_id)
|
|
|
|
case.casecollection_set.add(*form.cleaned_data["casecollection"])
|
|
# do_something_with_form_data(form.cleaned_data)
|
|
return HttpResponse("Done")
|
|
return render(request, "web/htmx_contact_form_confirm.html")
|
|
else:
|
|
form = AddCollectionToCaseForm(user=request.user)
|
|
return render(
|
|
request,
|
|
"generic_form.html",
|
|
{
|
|
"form": form,
|
|
"target_url": reverse("atlas:case_collection_form", args=(case_id,)),
|
|
},
|
|
)
|
|
|
|
|
|
# class AddCollectionToCaseView(FormView):
|
|
# form_class = AddCollectionToCaseForm
|
|
# template_name = "generic_form.html"
|
|
#
|
|
# def form_valid(self, form):
|
|
# form.save()
|
|
# return super(AddCollectionToCaseView, self).form_valid(form)
|
|
#
|
|
# def get_form_kwargs(self):
|
|
# kwargs = super(AddCollectionToCaseView, self).get_form_kwargs()
|
|
# kwargs.update({"user": self.request.user})
|
|
# return kwargs
|
|
#
|
|
# def get_success_url(self) -> str:
|
|
# pk = self.kwargs["pk"]
|
|
# return reverse("atlas:case_detail", kwargs={"pk": pk})
|
|
|
|
|
|
class AtlasClone(AtlasCreateBase, AuthorOrCheckerRequiredMixin, CreateView):
|
|
"""Clones a existing atlas case"""
|
|
|
|
# fields = '__all__'
|
|
# #fields = [ 'condition' ]
|
|
# #initial = {'date_of_death': '05/01/2018'}
|
|
# exclude = [ 'created_date', 'published_date' ]
|
|
def get_initial(self):
|
|
# print(self.request)
|
|
old_object = get_object_or_404(Case, pk=self.kwargs["pk"])
|
|
|
|
if (
|
|
self.request.user.groups.filter(name="atlas_editor").exists()
|
|
or self.request.user.is_superuser
|
|
):
|
|
pass
|
|
else:
|
|
if self.request.user not in old_object.get_author_objects():
|
|
raise PermissionDenied() # or Http404
|
|
else:
|
|
pass
|
|
|
|
initial_data = model_to_dict(old_object, exclude=["id", "previous_case", "history", "discussion", "report"])
|
|
return initial_data
|
|
|
|
|
|
class AtlasCloneNext(AtlasClone):
|
|
"""Extends AtlasClone to set the previous_case field to the case being cloned"""
|
|
|
|
def get_initial(self):
|
|
initial_data = super().get_initial()
|
|
initial_data["previous_case"] = self.kwargs["pk"]
|
|
return initial_data
|
|
|
|
|
|
@login_required
|
|
@user_is_author_or_atlas_editor
|
|
def atlas_scrap(request, pk):
|
|
try:
|
|
atlas = Case.objects.get(pk=pk)
|
|
except Case.DoesNotExist:
|
|
raise Http404("Case does not exist")
|
|
|
|
atlas.scrapped = not atlas.scrapped
|
|
atlas.save()
|
|
return HttpResponseRedirect(reverse("atlas:case_detail", args=(pk,)))
|
|
|
|
|
|
# @login_required
|
|
# def edit_abnormality_popup(request):
|
|
# instance = get_object_or_404(Abnormality, pk = pk)
|
|
# form = AbnormalityForm(request.POST or None)
|
|
# if form.is_valid():
|
|
# instance = form.save()
|
|
# return HttpResponse('<script>opener.closePopup(window, "%s", "%s", "#id_abnormality");</script>' % (instance.pk, instance))
|
|
# return render(request, "atlas/create_simple.html", {'form': form})
|
|
|
|
|
|
@csrf_exempt
|
|
def get_examination_id(request):
|
|
if request.accepts("application/json")():
|
|
examination_name = request.GET["examination_name"]
|
|
examination_id = Examination.objects.get(name=examination_name).id
|
|
data = {
|
|
"examination_id": examination_id,
|
|
}
|
|
return HttpResponse(json.dumps(data), content_type="application/json")
|
|
return HttpResponse("/")
|
|
|
|
|
|
class CaseView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
|
|
model = Case
|
|
table_class = CaseTable
|
|
template_name = "atlas/case_view.html"
|
|
|
|
filterset_class = CaseFilter
|
|
|
|
|
|
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
|
|
|
|
|
|
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
|
|
|
|
|
|
class PresentationView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
|
|
model = Presentation
|
|
table_class = PresentationTable
|
|
template_name = "atlas/view.html"
|
|
|
|
filterset_class = PresentationFilter
|
|
|
|
|
|
class PathologicalProcessView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
|
|
model = PathologicalProcess
|
|
table_class = PathologicalProcessTable
|
|
template_name = "atlas/view.html"
|
|
|
|
filterset_class = PathologicalProcessFilter
|
|
|
|
|
|
class StructureView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
|
|
model = Structure
|
|
table_class = StructureTable
|
|
template_name = "atlas/view.html"
|
|
|
|
filterset_class = StructureFilter
|
|
|
|
|
|
class FindingView(LoginRequiredMixin, UserConfigurablePaginationMixin, SingleTableMixin, FilterView):
|
|
model = Finding
|
|
table_class = FindingTable
|
|
template_name = "atlas/view.html"
|
|
|
|
filterset_class = FindingFilter
|
|
|
|
|
|
@login_required
|
|
def case_order_dicom(request, pk):
|
|
if request.htmx:
|
|
case: Case = get_object_or_404(Case, pk=pk)
|
|
|
|
fail = []
|
|
# TODO: don't try to order non stack series
|
|
for series in case.series.all():
|
|
try:
|
|
if request.user in series.get_author_objects():
|
|
series.order_by_dicom()
|
|
else:
|
|
fail.append(f"{series.pk}: invalid permission")
|
|
except:
|
|
fail.append(f"{series.pk}: error")
|
|
|
|
if fail:
|
|
return HttpResponse(
|
|
f"Unable to order series {','.join([str(i) for i in fail])}"
|
|
)
|
|
else:
|
|
return HttpResponse("Series reordered")
|
|
|
|
raise Http404
|
|
|
|
|
|
@login_required
|
|
@user_is_author_or_atlas_series_checker
|
|
def series_order_dicom(request, pk):
|
|
series = get_object_or_404(Series, pk=pk)
|
|
try:
|
|
series.order_by_dicom()
|
|
except:
|
|
return HttpResponse("<h1>Series does not appear to contain dicoms</h1>")
|
|
|
|
return redirect("atlas:series_detail", pk=pk)
|
|
|
|
|
|
@login_required
|
|
def image_diff(request):
|
|
if request.htmx:
|
|
image_ids = request.POST.getlist("image-id")
|
|
|
|
if len(image_ids) == 2:
|
|
first: SeriesImage = get_object_or_404(SeriesImage, pk=image_ids[0])
|
|
second: SeriesImage = get_object_or_404(SeriesImage, pk=image_ids[1])
|
|
|
|
first_ds = pydicom.dcmread(first.image.file)
|
|
# del first_ds.PixelData
|
|
# j1 = first_ds.to_json_dict()
|
|
f = print_dicom(first_ds, join=False)
|
|
|
|
second_ds = pydicom.dcmread(second.image.file)
|
|
# del second_ds.PixelData
|
|
# j2 = second_ds.to_json_dict()
|
|
s = print_dicom(second_ds, join=False)
|
|
|
|
diff = difflib.HtmlDiff().make_table(f, s)
|
|
|
|
dict_diff = pprint.pformat(compare_dicom_datasets(first_ds, second_ds))
|
|
|
|
# return HttpResponse(diff)
|
|
|
|
return HttpResponse(f"<pre>{dict_diff}</pre><br/>{diff}")
|
|
|
|
else:
|
|
ds_set = []
|
|
for pk in image_ids:
|
|
series_image = get_object_or_404(SeriesImage, pk=pk)
|
|
ds_set.append(pydicom.dcmread(series_image.image.file))
|
|
|
|
dict_diff = pprint.pformat(compare_dicom_datasets(*ds_set))
|
|
return HttpResponse(f"<pre>{dict_diff}</pre>")
|
|
|
|
return HttpResponse("Fail")
|
|
|
|
|
|
@login_required
|
|
def series_image_dicom(request, pk):
|
|
image = get_object_or_404(SeriesImage, pk=pk)
|
|
|
|
return render(request, "atlas/series_image_dicom.html", {"image": image})
|
|
|
|
|
|
@login_required
|
|
@user_is_author_or_atlas_series_checker
|
|
def series_anonymise_dicom(request, pk):
|
|
series = get_object_or_404(Series, pk=pk)
|
|
|
|
series.anonymise_images()
|
|
|
|
return redirect("atlas:series_detail", pk=pk)
|
|
|
|
|
|
@login_required
|
|
@user_is_author_or_atlas_series_checker
|
|
def series_order_dicom_instance(request, pk):
|
|
series = get_object_or_404(Series, pk=pk)
|
|
try:
|
|
series.order_by_dicom("InstanceNumber")
|
|
except:
|
|
return HttpResponse(
|
|
"<h1>Series does not appear to contain dicoms (or field InstanceNumber)</h1>"
|
|
)
|
|
|
|
return redirect("atlas:series_detail", pk=pk)
|
|
|
|
|
|
@login_required
|
|
@user_is_author_or_atlas_series_checker
|
|
def series_order_dicom_SeriesInstanceUID(request, pk):
|
|
series = get_object_or_404(Series, pk=pk)
|
|
try:
|
|
series.order_by_dicom("SeriesInstanceUID")
|
|
except:
|
|
return HttpResponse(
|
|
"<h1>Series does not appear to contain dicoms (or field SeriesInstanceUID)</h1>"
|
|
)
|
|
|
|
return redirect("atlas:series_detail", pk=pk)
|
|
|
|
|
|
@login_required
|
|
@user_is_author_or_atlas_series_checker
|
|
def series_order_upload_filename(request, pk):
|
|
series = get_object_or_404(Series, pk=pk)
|
|
series.order_by_upload_filename()
|
|
|
|
return redirect("atlas:series_detail", pk=pk)
|
|
|
|
|
|
# Vendored in from BaseZipView
|
|
class SeriesImagesZipView(SeriesImagesZipViewBase, SuperuserRequiredMixin):
|
|
series_object = Series
|
|
|
|
|
|
# At the moment only the series owner / author can delete findings but anyone can create
|
|
def create_series_findings(request):
|
|
# posts = Post.objects.all()
|
|
response_data = {}
|
|
|
|
if request.POST.get("action") == "post":
|
|
series_finding_id = int(request.POST.get("series_finding_id"))
|
|
series_id = request.POST.get("series")
|
|
findings_ids = json.loads(request.POST.get("findings"))
|
|
structure_ids = json.loads(request.POST.get("structures"))
|
|
condition_ids = json.loads(request.POST.get("conditions"))
|
|
description = request.POST.get("description")
|
|
#annotation_json = request.POST.get("annotation_json")
|
|
#viewport_json = request.POST.get("viewport_json")
|
|
#current_image_id_index = request.POST.get("current_image_id_index")
|
|
annotation_json_3d = request.POST.get("annotation_json_3d")
|
|
viewer_state_3d = request.POST.get("viewer_state_3d")
|
|
|
|
series = Series.objects.get(pk=series_id)
|
|
findings = Finding.objects.filter(pk__in=findings_ids)
|
|
structures = Structure.objects.filter(pk__in=structure_ids)
|
|
conditions = Condition.objects.filter(pk__in=condition_ids)
|
|
|
|
if series_finding_id > 0:
|
|
sf = get_object_or_404(SeriesFinding, pk=series_finding_id)
|
|
sf.series = series
|
|
sf.description = description
|
|
#sf.annotation_json = annotation_json
|
|
#sf.viewport_json = viewport_json
|
|
else:
|
|
sf = SeriesFinding.objects.create(
|
|
series=series,
|
|
description=description,
|
|
#annotation_json=annotation_json,
|
|
#viewport_json=viewport_json,
|
|
)
|
|
|
|
sf.findings.set(findings)
|
|
sf.structures.set(structures)
|
|
sf.conditions.set(conditions)
|
|
#sf.current_image_id_index = current_image_id_index
|
|
sf.annotation_json_3d = annotation_json_3d
|
|
sf.viewer_state_3d = viewer_state_3d
|
|
sf.save()
|
|
return JsonResponse({"success": True})
|
|
|
|
return JsonResponse({"success": False})
|
|
return render(request, "create_post.html", {"posts": posts})
|
|
|
|
|
|
class ConditionAutocomplete(autocomplete.Select2QuerySetView):
|
|
def get_queryset(self):
|
|
# Don't forget to filter out results depending on the visitor !
|
|
if not self.request.user.is_authenticated:
|
|
return Condition.objects.none()
|
|
|
|
qs = Condition.objects.all()
|
|
|
|
if self.q:
|
|
# This raises a fielderror which breaks creating a new item if not caught
|
|
try:
|
|
qs = qs.filter(name__icontains=self.q)
|
|
except FieldError:
|
|
return Condition.objects.none()
|
|
|
|
return qs
|
|
|
|
|
|
class ConditionAutocompleteRCRCurriculum(autocomplete.Select2QuerySetView):
|
|
def get_queryset(self):
|
|
# Don't forget to filter out results depending on the visitor !
|
|
if not self.request.user.is_authenticated:
|
|
return Condition.objects.none()
|
|
|
|
qs = Condition.objects.filter(rcr_curriculum=True)
|
|
|
|
if self.q:
|
|
# This raises a fielderror which breaks creating a new item if not caught
|
|
try:
|
|
qs = qs.filter(name__icontains=self.q, rcr_curriculum=True)
|
|
except FieldError:
|
|
return Condition.objects.none()
|
|
|
|
return qs
|
|
|
|
|
|
class PresentationAutocomplete(autocomplete.Select2QuerySetView):
|
|
def get_queryset(self):
|
|
# Don't forget to filter out results depending on the visitor !
|
|
if not self.request.user.is_authenticated:
|
|
return Presentation.objects.none()
|
|
|
|
qs = Presentation.objects.all()
|
|
|
|
if self.q:
|
|
# This raises a fielderror which breaks creating a new item if not caught
|
|
try:
|
|
qs = qs.filter(name__icontains=self.q)
|
|
except FieldError:
|
|
return Presentation.objects.none()
|
|
|
|
return qs
|
|
|
|
|
|
class FindingAutocomplete(autocomplete.Select2QuerySetView):
|
|
def get_queryset(self):
|
|
# Don't forget to filter out results depending on the visitor !
|
|
if not self.request.user.is_authenticated:
|
|
return Finding.objects.none()
|
|
|
|
qs = Finding.objects.all()
|
|
|
|
if self.q:
|
|
# This raises a fielderror which breaks creating a new item if not caught
|
|
try:
|
|
qs = qs.filter(name__icontains=self.q)
|
|
except FieldError:
|
|
return Finding.objects.none()
|
|
|
|
return qs
|
|
|
|
|
|
class StructureAutocomplete(autocomplete.Select2QuerySetView):
|
|
def get_queryset(self):
|
|
# Don't forget to filter out results depending on the visitor !
|
|
if not self.request.user.is_authenticated:
|
|
return Structure.objects.none()
|
|
|
|
qs = Structure.objects.all()
|
|
|
|
if self.q:
|
|
# This raises a fielderror which breaks creating a new item if not caught
|
|
try:
|
|
qs = qs.filter(name__icontains=self.q)
|
|
except FieldError:
|
|
return Structure.objects.none()
|
|
|
|
return qs
|
|
|
|
|
|
@login_required
|
|
# @user_is_atlas_editor
|
|
def categories_list(request):
|
|
# condition = get_object_or_404(Condition, pk=pk)
|
|
|
|
# logging.debug(atlas.subspecialty.first().name.all())
|
|
return render(
|
|
request,
|
|
"atlas/categories_list.html",
|
|
# {
|
|
# "condition": condition,
|
|
# },
|
|
)
|
|
|
|
|
|
# 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/view.html"
|
|
|
|
filterset_class = CaseCollectionFilter
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_viva(request, pk):
|
|
collection = get_object_or_404(CaseCollection, pk=pk)
|
|
|
|
#cases = (
|
|
# collection.cases.all()
|
|
# .order_by("casedetail__sort_order")
|
|
# .prefetch_related(
|
|
# "series",
|
|
# "series__images",
|
|
# "series__contrast",
|
|
# "series__plane",
|
|
# "series__examination",
|
|
# )
|
|
#)
|
|
|
|
casedetails = (
|
|
CaseDetail.objects
|
|
.filter(collection=collection)
|
|
.select_related("case")
|
|
.order_by("sort_order")
|
|
.prefetch_related(
|
|
# Prefetch series and their images directly on the case
|
|
"case__series",
|
|
"case__series__modality",
|
|
"case__series__examination",
|
|
"case__series__plane",
|
|
"case__series__contrast",
|
|
"case__series__images",
|
|
)
|
|
)
|
|
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_viva.html",
|
|
{"collection": collection, "casedetails": casedetails, "can_edit": True},
|
|
)
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_viva_case(request, pk, case_number):
|
|
collection = get_object_or_404(CaseCollection, pk=pk)
|
|
|
|
# cases = collection.cases.all().order_by("casedetail__sort_order")
|
|
current_case = collection.get_case_by_index(case_number)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_viva_case.html",
|
|
{"collection": collection, "current_case": current_case},
|
|
)
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_viva_series(request, pk, series_id):
|
|
collection = get_object_or_404(CaseCollection, pk=pk)
|
|
|
|
series = get_object_or_404(Series, pk=series_id)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_viva_series.html",
|
|
{"collection": collection, "current_series": series},
|
|
)
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_detail(request, pk):
|
|
collection = get_object_or_404(CaseCollection, pk=pk)
|
|
|
|
# cases = collection.cases.all().order_by("casedetail__sort_order")
|
|
casedetails = (
|
|
CaseDetail.objects
|
|
.filter(collection=collection)
|
|
.select_related("case")
|
|
.annotate(series_count=Count("case__seriesdetail"))
|
|
.order_by("sort_order")
|
|
)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_detail.html",
|
|
{"collection": collection, "casesdetails": casedetails, "can_edit": True, "exam": collection},
|
|
)
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_history(request, exam_id: int):
|
|
"""An overview page that shows who has attempted / finised the collection"""
|
|
collection = get_object_or_404(CaseCollection, pk=exam_id)
|
|
|
|
casedetails = CaseDetail.objects.filter(collection=collection).order_by(
|
|
"sort_order"
|
|
)
|
|
|
|
userexams = collection.get_cid_user_exams()
|
|
|
|
print(userexams)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_history.html",
|
|
{
|
|
"collection": collection,
|
|
"casesdetails": casedetails,
|
|
"can_edit": True,
|
|
"userexams": userexams,
|
|
},
|
|
)
|
|
|
|
|
|
@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,
|
|
},
|
|
)
|
|
|
|
|
|
|
|
|
|
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_
|
|
"""
|
|
print(f"Collection take start: {pk}, cid: {cid}, passcode: {passcode}")
|
|
collection = get_object_or_404(CaseCollection, pk=pk)
|
|
print(f"Collection: {collection}")
|
|
|
|
try:
|
|
collection.check_user_can_take(cid, passcode, request.user)
|
|
valid_user = True
|
|
except Http404:
|
|
valid_user = False
|
|
#valid_user = collection.check_logged_in_user(request)
|
|
|
|
|
|
|
|
cid_exam = None
|
|
if collection.collection_type == "REV" or valid_user:
|
|
cid_exam = collection.get_cid_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
|
|
)
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_case_priors(request, exam_id, case_id):
|
|
case_detail = CaseDetail.objects.get(case=case_id, collection=exam_id)
|
|
|
|
if request.htmx:
|
|
if "remove" in request.POST:
|
|
p = CasePrior.objects.get(
|
|
case_detail=case_detail, prior_case=request.POST["remove"]
|
|
)
|
|
p.delete()
|
|
return HttpResponse(f"Case removed")
|
|
elif "prior_case_id" in request.POST:
|
|
prior_case = Case.objects.get(pk=request.POST["prior_case_id"])
|
|
p, created = CasePrior.objects.get_or_create(
|
|
case_detail=case_detail, prior_case=prior_case
|
|
)
|
|
p.relation_text = request.POST["relation"]
|
|
|
|
if not p.relation_text:
|
|
return HttpResponse(
|
|
"You need to enter text to describe the relationship between the cases"
|
|
)
|
|
p.prior_visibility = request.POST["prior_visibility"]
|
|
|
|
p.save()
|
|
|
|
return HttpResponse(f"Case added")
|
|
else:
|
|
return HttpResponse("False")
|
|
raise Http404()
|
|
|
|
collection = case_detail.collection
|
|
|
|
added_priors = CasePrior.objects.filter(case_detail=case_detail)
|
|
|
|
added_relation_map = {}
|
|
for prior in added_priors:
|
|
added_relation_map[prior.prior_case] = (prior.relation_text, prior.prior_visibility)
|
|
|
|
case = case_detail.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(
|
|
case_detail.case, case_count=True
|
|
)
|
|
|
|
previous = collection.get_previous_case(case_detail.case)
|
|
next = collection.get_next_case(case_detail.case)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_case_priors.html",
|
|
{
|
|
"case_detail": case_detail,
|
|
"form": form,
|
|
#"model_form": model_form,
|
|
# "example_form": example_form,
|
|
"collection": collection,
|
|
"case": case_detail.case,
|
|
"previous": previous,
|
|
"next": next,
|
|
"collection_length": case_count,
|
|
"case_number": case_number,
|
|
"added_priors": added_priors,
|
|
"available_priors": available_priors,
|
|
},
|
|
)
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_case_details(request, exam_id, case_id):
|
|
case_detail = CaseDetail.objects.get(case=case_id, collection=exam_id)
|
|
|
|
collection = case_detail.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
|
|
case_detail.question_answers = json.loads(request.POST.get("json_answer"))
|
|
case_detail.save()
|
|
example_form = JsonAnswerForm(
|
|
request.POST, question_schema=case_detail.question_schema
|
|
)
|
|
form = CaseDetailForm(instance=case_detail)
|
|
# Called if the user saves the main form
|
|
elif request.POST.get("submit") == "save":
|
|
form = CaseDetailForm(request.POST, instance=case_detail)
|
|
if form.is_valid():
|
|
form.save()
|
|
# Add any additional logic or redirection here
|
|
# example_form = JsonAnswerForm(request.POST, question_schema=case_detail.question_schema)
|
|
post_data = request.POST.copy()
|
|
if (
|
|
case_detail.question_schema is not None
|
|
and case_detail.question_answers is not None
|
|
):
|
|
question_keys = set(case_detail.question_schema["properties"].keys())
|
|
answer_keys = set(case_detail.question_answers.keys())
|
|
|
|
answers = {}
|
|
for key, value in case_detail.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)
|
|
case_detail.question_answers = answers
|
|
case_detail.save()
|
|
example_form = JsonAnswerForm(
|
|
post_data, question_schema=case_detail.question_schema
|
|
)
|
|
|
|
form = CaseDetailForm(instance=case_detail)
|
|
# This shouldn't happen
|
|
else:
|
|
assert False
|
|
form = CaseDetailForm(request.POST, instance=case_detail)
|
|
|
|
else:
|
|
post_data = request.POST.copy()
|
|
post_data["json_answer"] = json.dumps(case_detail.question_answers)
|
|
example_form = JsonAnswerForm(
|
|
post_data, question_schema=case_detail.question_schema
|
|
)
|
|
|
|
form = CaseDetailForm(instance=case_detail)
|
|
pass
|
|
# post_data = None
|
|
|
|
# if case_detail.question_schema is not None and case_detail.question_answers is not None:
|
|
# question_keys = set(case_detail.question_schema["properties"].keys())
|
|
# answer_keys = set(case_detail.question_answers.keys())
|
|
|
|
# answers = {}
|
|
# for key, value in case_detail.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)
|
|
# case_detail.question_answers = answers
|
|
# example_form = JsonAnswerForm(post_data, question_schema=case_detail.question_schema)
|
|
|
|
# form = CaseDetailForm(instance=case_detail)
|
|
|
|
# blank_form = JsonAnswerForm(question_schema=case_detail.question_schema)
|
|
case_number, case_count = collection.get_index_of_case(
|
|
case_detail.case, case_count=True
|
|
)
|
|
|
|
previous = collection.get_previous_case(case_detail.case)
|
|
next = collection.get_next_case(case_detail.case)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_case_detail.html",
|
|
{
|
|
"case_detail": case_detail,
|
|
"form": form,
|
|
"example_form": example_form,
|
|
"collection": collection,
|
|
"case": case_detail.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()
|
|
|
|
case_details = CaseDetail.objects.filter(
|
|
case__in=cases, collection=collection
|
|
).prefetch_related("case")
|
|
|
|
answers = CidReportAnswer.objects.filter(question__in=case_details)
|
|
|
|
case_map = {}
|
|
for cd in case_details:
|
|
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,
|
|
"case_details": case_details,
|
|
"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)
|
|
|
|
case_detail = CaseDetail.objects.filter(case=case, collection=collection)
|
|
|
|
answers = CidReportAnswer.objects.filter(question__in=case_detail)
|
|
|
|
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 = case_detail
|
|
# 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,
|
|
"case_detail": case_detail,
|
|
"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)
|
|
|
|
collection.check_user_can_take(cid, passcode, request.user)
|
|
|
|
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()
|
|
|
|
case_details = CaseDetail.objects.filter(case__in=cases, collection=collection)
|
|
|
|
if cid is not None:
|
|
answers = CidReportAnswer.objects.filter(cid=cid, question__in=case_details)
|
|
else:
|
|
answers = UserReportAnswer.objects.filter(
|
|
user=request.user, question__in=case_details
|
|
)
|
|
|
|
answer_question_map = {}
|
|
for ans in answers:
|
|
answer_question_map[ans.question] = ans
|
|
|
|
question_answer_tuples = []
|
|
answer_count = 0
|
|
for cd in case_details:
|
|
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)
|
|
|
|
collection.check_user_can_take(cid, passcode, request.user)
|
|
|
|
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
|
|
)
|
|
|
|
case_detail = CaseDetail.objects.get(case=case, collection=collection)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_case_view_take_answers.html",
|
|
{
|
|
"collection": collection,
|
|
"case": case,
|
|
"case_detail": case_detail,
|
|
},
|
|
)
|
|
|
|
|
|
def collection_case_view_take_user(request, pk: int, case_number: int):
|
|
return collection_case_view_take(request, pk, case_number)
|
|
|
|
|
|
def collection_case_view_take(
|
|
request, pk: int, case_number: int, cid=None, passcode=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)
|
|
|
|
form = None
|
|
answer: None | CidReportAnswer | UserReportAnswer = None
|
|
|
|
collection.check_user_can_take(cid, passcode, request.user)
|
|
|
|
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()
|
|
|
|
case_detail = CaseDetail.objects.get(case=case, collection=collection)
|
|
if not collection.review_only():
|
|
if collection.collection_type == "REP":
|
|
if cid is not None:
|
|
answer = case_detail.cidreportanswer_set.filter(cid=cid).first()
|
|
ReportAnswerForm = CidReportAnswerForm
|
|
else:
|
|
answer = case_detail.userreportanswer_set.filter(
|
|
user=request.user
|
|
).first()
|
|
ReportAnswerForm = UserReportAnswerForm
|
|
elif collection.collection_type == "QUE":
|
|
if cid is not None:
|
|
answer = case_detail.cidreportanswer_set.filter(cid=cid).first()
|
|
ReportAnswerForm = CidQuestionAnswerForm
|
|
else:
|
|
answer = case_detail.userreportanswer_set.filter(
|
|
user=request.user
|
|
).first()
|
|
ReportAnswerForm = UserQuestionAnswerForm
|
|
|
|
if request.method == "POST":
|
|
if collection.collection_type in ("REP", "QUE"):
|
|
if not collection.publish_results:
|
|
if answer:
|
|
form = ReportAnswerForm(
|
|
request.POST, instance=answer, case_detail=case_detail
|
|
)
|
|
else:
|
|
form = ReportAnswerForm(request.POST, case_detail=case_detail)
|
|
|
|
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 = case_detail
|
|
# 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
|
|
|
|
answer.save()
|
|
|
|
cid_user_exam.end_time = timezone.now()
|
|
cid_user_exam.save()
|
|
|
|
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, case_detail=case_detail)
|
|
|
|
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 = case_detail.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
|
|
|
|
return render(
|
|
request,
|
|
"atlas/collection_case_view_take.html",
|
|
{
|
|
"form": form,
|
|
"collection": collection,
|
|
"case": case,
|
|
"series_list": series_list,
|
|
"series_to_load": series_to_load,
|
|
"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,
|
|
},
|
|
)
|
|
|
|
|
|
@user_is_collection_author_or_atlas_editor
|
|
def collection_case_view(request, pk, case_number):
|
|
"""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)
|
|
|
|
form = None
|
|
answer = None
|
|
|
|
case, case_count = collection.get_case_by_index(case_number, case_count=True)
|
|
|
|
series_list = case.series.all().prefetch_related("images", "examination", "plane")
|
|
|
|
previous = case_number > 0
|
|
next = case_number < (case_count - 1)
|
|
|
|
return render(
|
|
request,
|
|
"atlas/case_detail.html",
|
|
{
|
|
"form": form,
|
|
"collection": collection,
|
|
"collection_length": case_count,
|
|
"case": case,
|
|
"series_list": series_list,
|
|
"case_number": case_number,
|
|
"previous": previous,
|
|
"next": next,
|
|
"answer": answer,
|
|
"cimar_sid": request.user.userprofile.cimar_sid,
|
|
},
|
|
)
|
|
|
|
|
|
def collection_dicom_json(request, pk):
|
|
collection = get_object_or_404(CaseCollection, pk=pk)
|
|
|
|
return JsonResponse(collection.get_ohif_dicom_json())
|
|
|
|
|
|
@user_is_author_or_atlas_editor
|
|
def use_dates_as_descriptions(request, pk):
|
|
case = get_object_or_404(Case, pk=pk)
|
|
|
|
series: Series
|
|
for series in case.series.all():
|
|
series.use_date_as_description()
|
|
|
|
return HttpResponse("Done")
|
|
|
|
def collection_case_dicom_json_review(request, exam_id, case_id):
|
|
return collection_case_dicom_json(request, exam_id, case_id, review=True)
|
|
|
|
def collection_case_dicom_json(request, exam_id, case_id, review=False):
|
|
case_detail = CaseDetail.objects.get(case=case_id, collection=exam_id)
|
|
|
|
if review:
|
|
priors = case_detail.caseprior_set.exclude(prior_visibility="NO")
|
|
else:
|
|
priors = case_detail.caseprior_set.exclude(prior_visibility__in=["NO", "RE"])
|
|
|
|
return JsonResponse(case_detail.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())
|
|
|
|
|
|
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()
|
|
|
|
case_details = CaseDetail.objects.filter(
|
|
case__in=cases, collection=collection
|
|
).prefetch_related("case")
|
|
|
|
cid_user_answers = CidReportAnswer.objects.filter(
|
|
question__in=case_details, 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()
|
|
|
|
case_details = CaseDetail.objects.filter(
|
|
case__in=cases, collection=collection
|
|
).prefetch_related("case")
|
|
|
|
cid_user_answers = CidReportAnswer.objects.filter(question__in=case_details)
|
|
|
|
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()
|
|
|
|
case_details = CaseDetail.objects.filter(
|
|
case__in=cases, collection=collection
|
|
).prefetch_related("case")
|
|
|
|
cid_user_answers = CidReportAnswer.objects.filter(question__in=case_details)
|
|
|
|
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()
|
|
|
|
case_details = (
|
|
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,
|
|
"case_details": case_details,
|
|
},
|
|
)
|
|
|
|
|
|
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")
|
|
|
|
|
|
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)
|
|
case_details = collection.casedetail_set.all().prefetch_related()
|
|
|
|
# Delete all answers
|
|
for case in case_details:
|
|
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")
|
|
|
|
|
|
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)
|
|
case_details = collection.casedetail_set.all().prefetch_related()
|
|
|
|
# Delete all answers
|
|
for case in case_details:
|
|
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)
|
|
|
|
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)
|
|
|
|
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)
|
|
|
|
# Associate the series with the case
|
|
series.case.remove(case)
|
|
|
|
return HttpResponse(f"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_id):
|
|
"""
|
|
View to set up the display options for a specific case in a collection.
|
|
This can include setting which series to display, annotations, etc.
|
|
"""
|
|
case_detail = get_object_or_404(CaseDetail, case__pk=case_id, collection__pk=collection_id)
|
|
|
|
if request.method == "POST":
|
|
data = json.loads(request.body.decode("utf-8"))
|
|
if data.get("reset_viewerstate"):
|
|
case_detail.default_viewerstate = None
|
|
case_detail.save()
|
|
return HttpResponse("<span class='text-success'>Viewer state reset to default.</span>")
|
|
viewerstate = data.get("viewerstate")
|
|
if viewerstate is not None:
|
|
case_detail.default_viewerstate = viewerstate
|
|
case_detail.save()
|
|
return HttpResponse("<span class='text-success'>Viewer state saved.</span>")
|
|
else:
|
|
return HttpResponse("<span class='text-danger'>No viewer state provided.</span>", status=400)
|
|
|
|
return render(request, "atlas/collection_case_displaysetup.html", {
|
|
"case_detail": case_detail
|
|
})
|