penracourses/atlas/api.py

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


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_md5_hash))
        except DuplicateDicom:
            print(ud.check_for_duplicates(ud.image_md5_hash))
            duplicate.append((file.name, ud.image_md5_hash))
            pass

    return {"uploaded": uploaded, "duplicates": 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)
        data["image_hash"].append(d.image_md5_hash)

    return data


def import_dicoms_helper(request, case_id: int | None = None):
    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))

    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_md5_hash=dicom.image_md5_hash,
            )
            series_image.save()
            # dicom.image.delete()

            dicom.delete()
        series_list.append(series)

    return series_list


@router.get("/import_dicoms", auth=django_auth, response=List[SeriesSchema])
def import_dicoms(request):
    return import_dicoms_helper(request)


@router.get("/import_dicoms/{case_id}", auth=django_auth, response=List[SeriesSchema])
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_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)


@router.get("/get_cases_user", auth=django_auth, response=List[CaseSchema])
def get_cases_user(request):
    print("HELLO")
    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_md5_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.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_md5_hash = ""

        series_image.save()

        print(series_image)