Ross
321 lines
9.3 KiB
Python
321 lines
9.3 KiB
Python
from collections import defaultdict
|
|
from typing import List, Tuple
|
|
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
|
|
import pydicom
|
|
|
|
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()]
|
|
|
|
|
|
class CaseSchema(ModelSchema):
|
|
class Config:
|
|
model = Case
|
|
model_fields = ["id", "title"]
|
|
|
|
|
|
@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((file.name, ud.image_blake3_hash))
|
|
except DuplicateDicom:
|
|
print(ud.check_for_duplicates(ud.image_blake3_hash))
|
|
duplicate.append((file.name, ud.image_blake3_hash))
|
|
pass
|
|
|
|
return {"uploaded": uploaded, "duplicates": duplicate}
|
|
|
|
|
|
@router.post("/clear_dicoms", auth=django_auth)
|
|
def clear_dicoms(request):
|
|
|
|
if "selection" in request.POST:
|
|
dicoms = UncategorisedDicom.objects.filter(series_instance_uid__in=request.POST.getlist("selection"))
|
|
|
|
else:
|
|
dicoms = UncategorisedDicom.objects.filter(user=request.user)
|
|
|
|
dicoms.delete()
|
|
print(dicoms)
|
|
|
|
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)
|
|
data["image_hash"].append(d.image_blake3_hash)
|
|
|
|
return data
|
|
|
|
|
|
def import_dicoms_helper(request, case_id: int | None = None):
|
|
#dicoms = UncategorisedDicom.objects.filter(user=request.user)
|
|
if "selection" in request.POST:
|
|
dicoms = UncategorisedDicom.objects.filter(series_instance_uid__in=request.POST.getlist("selection"))
|
|
|
|
else:
|
|
dicoms = UncategorisedDicom.objects.filter(user=request.user)
|
|
|
|
data = defaultdict(list)
|
|
for d in dicoms:
|
|
tags = d.basic_dicom_tags
|
|
|
|
data[tags["SeriesInstanceUID"]].append((d, tags))
|
|
|
|
series_list = []
|
|
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)
|
|
|
|
if case_id is not None:
|
|
series.case.add(case_id)
|
|
|
|
for dicom, dicom_tags in data[series_uid]:
|
|
series_image = SeriesImage(
|
|
image=dicom.image,
|
|
series=series,
|
|
is_dicom=True,
|
|
image_blake3_hash=dicom.image_blake3_hash,
|
|
)
|
|
series_image.save()
|
|
# dicom.image.delete()
|
|
|
|
dicom.delete()
|
|
series_list.append((series, series.get_link()))
|
|
|
|
return series_list
|
|
|
|
|
|
@router.post("/import_dicoms", auth=django_auth, response=List[Tuple[ SeriesSchema, str ]])
|
|
def import_dicoms(request):
|
|
return import_dicoms_helper(request)
|
|
|
|
|
|
@router.post("/import_dicoms/{case_id}", auth=django_auth, response=List[Tuple[ SeriesSchema, str ]])
|
|
def import_dicoms_case(request, case_id: int):
|
|
return import_dicoms_helper(request, case_id=case_id)
|
|
|
|
|
|
@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_blake3_hash in img_ids:
|
|
dupes.add(series_image.id)
|
|
img_ids.add(series_image.image_blake3_hash)
|
|
|
|
if dupes:
|
|
SeriesImage.objects.filter(id__in=dupes).delete()
|
|
|
|
return len(dupes)
|
|
|
|
@router.get("/series_truncate/{series_id}/{start}/{end}/", auth=django_auth)
|
|
def series_truncate(request, series_id: int, start: int, end:int):
|
|
print(start, end)
|
|
series = get_object_or_404(Series, pk=series_id)
|
|
|
|
if not series.check_user_can_edit(request.user):
|
|
return {"status": "permission denied"}
|
|
|
|
images_removed = []
|
|
for n, image in enumerate(series.get_images()):
|
|
if n >= start and n <=end:
|
|
continue
|
|
else:
|
|
print(n, image)
|
|
image.removed=True
|
|
image.image = None
|
|
image.save()
|
|
images_removed.append(image.pk)
|
|
return {"status": "success", "images_removed": images_removed}
|
|
|
|
|
|
@router.get("/get_cases_user", auth=django_auth, response=List[CaseSchema])
|
|
def get_cases_user(request):
|
|
return Case.objects.filter(author=request.user)
|
|
|
|
|
|
@router.get("/check_image_hash/{hash}", auth=django_auth)
|
|
def check_image_hash(request, hash: str):
|
|
try:
|
|
print(hash)
|
|
series_image = SeriesImage.objects.get(image_blake3_hash=hash)
|
|
data = {
|
|
"status": "success",
|
|
"id": series_image.pk,
|
|
"url": series_image.series.get_absolute_url(),
|
|
}
|
|
return data
|
|
except SeriesImage.DoesNotExist:
|
|
data = {"status": "success", "id": False}
|
|
return data
|
|
|
|
@router.post("/check_image_hashes/", auth=django_auth)
|
|
def check_images_hashes(request, hashes: List[str]):
|
|
"""Checks a list of image hashes and returns the series id / url if found
|
|
|
|
Return format
|
|
{ "hash_id": {"id": "series_id|false", "url": "series_url|false"}, ...}
|
|
"""
|
|
|
|
hash_status = {}
|
|
|
|
for hash in hashes:
|
|
try:
|
|
print(hash)
|
|
series_image = SeriesImage.objects.get(image_blake3_hash=hash)
|
|
data = {
|
|
"id": series_image.pk,
|
|
"url": series_image.series.get_absolute_url(),
|
|
}
|
|
except SeriesImage.DoesNotExist:
|
|
data = { "id": False, "url": False}
|
|
|
|
hash_status[hash] = data
|
|
|
|
return hash_status
|
|
|
|
@router.get("/generate_image_hash/{id}", auth=django_auth)
|
|
def generate_image_hash(request, id: int):
|
|
series_image = SeriesImage.objects.get(pk=id)
|
|
|
|
series_image.image_blake3_hash = ""
|
|
|
|
series_image.save()
|
|
|
|
print(series_image)
|
|
|
|
@router.get("/view_dicom_tags/{hash}", auth=django_auth)
|
|
def view_dicom_tags(request, hash: str):
|
|
|
|
item = SeriesImage.objects.get(image_blake3_hash=hash)
|
|
return item.get_dicom_json()
|
|
|
|
@router.get("/series_split_by_dicom_tag/{series_id}/{dicom_tag}", auth=django_auth)
|
|
def series_split_by_tag(request, series_id: int, dicom_tag:str):
|
|
series = get_object_or_404(Series, pk=series_id)
|
|
|
|
if not series.check_user_can_edit(request.user):
|
|
return {"status": "permission denied"}
|
|
|
|
if dicom_tag.startswith("(") and dicom_tag.endswith(")"):
|
|
dicom_tag = tuple([hex(int(i.strip(),16)) for i in dicom_tag[1:-1].split(",")])
|
|
|
|
image_map = defaultdict(list)
|
|
|
|
# Split teh associated images
|
|
for image in series.images.all():
|
|
print(image)
|
|
ds = image.get_dicom_data()
|
|
|
|
print("2a", ds[dicom_tag])
|
|
|
|
if dicom_tag in ds:
|
|
val = ds[dicom_tag].value
|
|
|
|
if type(val) == pydicom.multival.MultiValue:
|
|
image_map[",".join((val))].append(image)
|
|
else:
|
|
image_map[val].append(image)
|
|
else:
|
|
return [f"{dicom_tag} not found"]
|
|
|
|
new_series = []
|
|
case = series.case.all()
|
|
authors = series.author.all()
|
|
|
|
for n, option in enumerate(image_map):
|
|
if n == 0:
|
|
series.images.set(image_map[option])
|
|
|
|
else:
|
|
series.pk = None
|
|
series.save()
|
|
series.case.set(case)
|
|
series.author.set(authors)
|
|
series.images.set(image_map[option])
|
|
|
|
new_series.append(series.pk)
|
|
|
|
|
|
return new_series
|
|
|
|
|
|
@router.get("/split_order_by_dicom_tag/{series_id}/{dicom_tag}", auth=django_auth)
|
|
def series_order_by_tag(request, series_id: int, dicom_tag:str):
|
|
series = get_object_or_404(Series, pk=series_id)
|
|
|
|
series.order_by_dicom(dicom_tag)
|
|
|
|
return [f"{series} ordered by {dicom_tag}"] |