from collections import defaultdict from typing import List from django.shortcuts import get_object_or_404 from ninja import ModelSchema, Router, Schema, Field # from .decorators import user_is_author_or_rapid_checker from django.core.exceptions import PermissionDenied from generic.decorators import check_user_in_group from generic.constants import Group from ninja.security import django_auth from ninja import NinjaAPI, File from ninja.files import UploadedFile from generic.models import Modality from .models import Case, DuplicateDicom, Series, SeriesImage, UncategorisedDicom router = Router() class SeriesSchema(ModelSchema): case_id: List[int] = [] class Config: model = Series model_fields = ["id", "modality", "examination", "series_instance_uid", "author"] @staticmethod def resolve_case_id(obj): return [i.id for i in obj.case.all()] @router.post("/upload_dicom", auth=django_auth) def upload_dicom(request, files: List[UploadedFile] = File(...)): uploaded = [] duplicate = [] for file in files: # data = file.read() try: ud = UncategorisedDicom(image=file, user=request.user) ud.save() uploaded.append(ud) except DuplicateDicom: duplicate.append(ud) pass return {"uploaded": len(uploaded), "duplicates": len(duplicate)} @router.get("/clear_dicoms", auth=django_auth) def clear_dicoms(request): dicoms = UncategorisedDicom.objects.filter(user=request.user) dicoms.delete() return True @router.get("/uncategorised_dicoms", auth=django_auth) def uncategorised_dicoms(request): dicoms = UncategorisedDicom.objects.filter(user=request.user) data = defaultdict(list) for d in dicoms: tags = d.get_basic_dicom_tags() data[tags["SeriesInstanceUID"]].append(tags) return data @router.get("/import_dicoms", auth=django_auth) def import_dicoms(request): dicoms = UncategorisedDicom.objects.filter(user=request.user) data = defaultdict(list) for d in dicoms: tags = d.get_basic_dicom_tags() data[tags["SeriesInstanceUID"]].append((d, tags)) for series_uid in data: modality = get_object_or_404(Modality, short_code=tags["Modality"]) tags = data[series_uid][0][1] # Check if series with the id already exists (in which case we just add to htat) try: series = Series.objects.get(series_instance_uid=tags["SeriesInstanceUID"]) except Series.DoesNotExist: series = Series( modality=modality, description=tags["SeriesDescription"], series_instance_uid=tags["SeriesInstanceUID"], ) series.save() # We might only want to add the author during creation.... series.author.add(request.user) for dicom, dicom_tags in data[series_uid]: series_image = SeriesImage(image=dicom.image, series=series) series_image.save() dicom.delete() return {} @router.get("/orphan_series", auth=django_auth, response=List[SeriesSchema]) def orphan_series(request): #print(request.user.series.filter(case=None)) return request.user.series.filter(case=None) @router.get("/series_remove_duplicate_images", auth=django_auth) def series_remove_duplicate_images(request, series_id: int): series = get_object_or_404(Series, pk=series_id) img_ids = set() dupes = set() for series_image in series.images.all(): if series_image.image_md5_hash in img_ids: dupes.add(series_image.id) img_ids.add(series_image.image_md5_hash) if dupes: SeriesImage.objects.filter(id__in=dupes).delete() return len(dupes)