penracourses/atlas/api.py

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 django.db import transaction

# 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
from loguru import logger


router = Router()

logger.add("api.log", format="{time} {level} {message}", level="DEBUG")


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("/generate_image_hash", auth=django_auth)
def generate_image_hash(request, id:int):
    s = SeriesImage.objects.get(pk=id)
    s.generate_hashes()

    return {"md5": s.image_md5_hash, "blake3": s.image_blake3_hash, "is_dicom": s.is_dicom}

@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

@logger.catch()
def import_dicoms_helper(request, case_id: int | None = None):
    logger.debug(f"IMPORT {case_id}")
    #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))

    logger.debug("IMPORT2")
    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)
        if Series.objects.filter(series_instance_uid=tags["SeriesInstanceUID"]).exists():
            series = Series.objects.get(series_instance_uid=tags["SeriesInstanceUID"])
        else:
            series = Series(
                modality=modality,
                description=tags["SeriesDescription"],
                series_instance_uid=tags["SeriesInstanceUID"],
            )
            series.save()

        logger.debug(f"IMPORT pre author {series_uid}")
        # We might only want to add the author during creation....
        series.author.add(request.user)

        logger.debug(f"IMPORT pre case add {series_uid}")
        if case_id is not None:
            logger.debug(series.case.all())
            logger.debug(series.case.filter(id=case_id))
            if not series.case.filter(id=case_id).exists():
                series.case.add(case_id)

        logger.debug(f"IMPORT pre dicom loop {series_uid}")
        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,
            )
            logger.debug(f"pre save {series_uid}")
            series_image.save()
            # dicom.image.delete()
            logger.debug(f"pre delete {series_uid}")

            dicom.delete()
            logger.debug(f"post delete {series_uid}")
        logger.debug("IMPORT3B")
        series_list.append((series, series.get_link()))

    logger.debug("IMPORT4")

    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:
            # TOOD also check against uncategorised dicoms?
            with transaction.atomic():
                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 {image}"]

    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}"]