Source code for nlpmed_portal.annotations.api.viewsets.projects
# SPDX-FileCopyrightText: Copyright (C) 2026 Omid Jafari <omidjafari.com>
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from collections import defaultdict
from django.core.cache import cache
from django.db.models import Case
from django.db.models import CharField
from django.db.models import Count
from django.db.models import Exists
from django.db.models import F
from django.db.models import IntegerField
from django.db.models import OuterRef
from django.db.models import Prefetch
from django.db.models import Q
from django.db.models import Value
from django.db.models import When
from django.db.models.functions import Coalesce
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import ValidationError
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from nlpmed_portal.annotations.api.serializers import PatientSerializer
from nlpmed_portal.annotations.api.serializers import ProjectDetailSerializer
from nlpmed_portal.annotations.api.serializers import ProjectListSerializer
from nlpmed_portal.annotations.api.viewsets.base import BaseViewSet
from nlpmed_portal.annotations.constants import PROJECT_META_CACHE_KEY
from nlpmed_portal.annotations.models import Lab
from nlpmed_portal.annotations.models import LabelCategory
from nlpmed_portal.annotations.models import LabelValue
from nlpmed_portal.annotations.models import NERAnnotation
from nlpmed_portal.annotations.models import Note
from nlpmed_portal.annotations.models import Patient
from nlpmed_portal.annotations.models import Project
from nlpmed_portal.annotations.permissions import MethodBasePermission
from nlpmed_portal.annotations.permissions import has_permission
[docs]
class ProjectViewSet(BaseViewSet):
queryset = Project.objects.all()
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_project",
"HEAD": "view_project",
"OPTIONS": "view_project",
"POST": "add_project",
"PUT": "change_project",
"PATCH": "change_project",
"DELETE": "delete_project",
}
[docs]
def get_serializer_class(self):
if self.action == "list":
return ProjectListSerializer
return ProjectDetailSerializer
[docs]
def get_object(self, *args, **kwargs):
if not hasattr(self, "_cached_obj"):
self._cached_obj = super().get_object(*args, **kwargs)
return self._cached_obj
[docs]
def get_queryset(self):
required_perm = self.required_perms.get("GET")
qs = self.filter_queryset_by_scope(super().get_queryset(), required_perm)
if self.action in {"list", "retrieve"}:
qs = qs.prefetch_related(
Prefetch(
"label_categories",
queryset=LabelCategory.objects.prefetch_related(
Prefetch(
"label_values",
queryset=LabelValue.objects.all(),
),
),
),
)
if self.action == "retrieve":
return qs.select_related(
"nlp_setting",
"nlp_setting__encoding_fixer",
"nlp_setting__pattern_replacer",
"nlp_setting__word_masker",
"nlp_setting__note_filter",
"nlp_setting__section_splitter",
"nlp_setting__section_filter",
"nlp_setting__sentence_segmenter",
"nlp_setting__duplicate_checker",
"nlp_setting__sentence_filter",
"nlp_setting__sentence_expander",
"nlp_setting__joiner",
"nlp_setting__ml_inference",
)
return qs
[docs]
@action(detail=True, methods=["GET"])
def daterange(self, request, pk=None):
proj = self.get_object()
return Response(
{
"disp_range_months_before_idx": proj.disp_range_months_before_idx,
"disp_range_months_after_idx": proj.disp_range_months_after_idx,
},
)
[docs]
@action(detail=False, methods=["GET"])
def meta(self, request):
data = cache.get(PROJECT_META_CACHE_KEY)
if data is not None:
return Response(data)
project_ids = list(self.get_queryset().values_list("id", flat=True))
total_patients = Patient.objects.filter(project_id__in=project_ids).distinct().count()
total_notes = Note.objects.filter(patient__project_id__in=project_ids).distinct().count()
total_labs = Lab.objects.filter(patient__project_id__in=project_ids).distinct().count()
patients_qs = (
Patient.objects
.filter(project_id__in=project_ids)
.select_related("project")
.annotate(
ann_cnt=Count("patient_annotations__annotator", distinct=True),
adj_cnt=Count("adjudications", distinct=True),
has_adjudicable_ann=Exists(
LabelValue.objects
.filter(
patient_annotations__patient_id=OuterRef("pk"),
category__isnull=False,
)
.exclude(category__name__iexact="none")
.exclude(category__name__iexact="historic"),
),
)
.annotate(
ann_status=Case(
When(ann_cnt=0, then=Value("not_started")),
When(
ann_cnt__lt=F("project__required_patient_annotations"),
then=Value("in_progress"),
),
default=Value("complete"),
output_field=CharField(),
),
adj_status=Case(
When(has_adjudicable_ann=False, then=Value("skipped")),
When(adj_cnt=0, then=Value("not_started")),
When(
adj_cnt__lt=F("project__required_patient_adjudications"),
then=Value("in_progress"),
),
default=Value("complete"),
output_field=CharField(),
),
)
)
patients_agg = patients_qs.aggregate(
ann_not_started=Count("id", filter=Q(ann_status="not_started")),
ann_in_progress=Count("id", filter=Q(ann_status="in_progress")),
ann_complete=Count("id", filter=Q(ann_status="complete")),
adj_skipped=Count("id", filter=Q(adj_status="skipped")),
adj_not_started=Count("id", filter=Q(adj_status="not_started")),
adj_in_progress=Count("id", filter=Q(adj_status="in_progress")),
adj_complete=Count("id", filter=Q(adj_status="complete")),
)
data = {
"patients_cnt": total_patients,
"notes_cnt": total_notes,
"labs_cnt": total_labs,
"annotations": {
"not_started": patients_agg["ann_not_started"],
"in_progress": patients_agg["ann_in_progress"],
"complete": patients_agg["ann_complete"],
},
"adjudications": {
"skipped": patients_agg["adj_skipped"],
"not_started": patients_agg["adj_not_started"],
"in_progress": patients_agg["adj_in_progress"],
"complete": patients_agg["adj_complete"],
},
}
cache.set(PROJECT_META_CACHE_KEY, data, timeout=600)
return Response(data)
[docs]
class PatientViewSet(BaseViewSet):
serializer_class = PatientSerializer
queryset = Patient.objects.select_related("project").all()
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_patient",
"HEAD": "view_patient",
"OPTIONS": "view_patient",
"POST": "add_patient",
"PUT": "change_patient",
"PATCH": "change_patient",
"DELETE": "delete_patient",
}
[docs]
def get_queryset(self):
required_perm = self.required_perms.get("GET")
queryset = super().get_queryset()
project_id = self.request.query_params.get("project_id")
if project_id:
if not project_id.isdigit():
raise ValidationError(
{"project_id": "Invalid project_id. Must be an integer."},
)
queryset = queryset.filter(project_id=project_id)
queryset = queryset.annotate(
num_distinct_annotators=Coalesce(
Count("patient_annotations__annotator", distinct=True),
Value(0),
output_field=IntegerField(),
)
+ Coalesce(
Count("ner_annotations__annotator", distinct=True),
Value(0),
output_field=IntegerField(),
),
num_distinct_adjudicators=Count(
"adjudications__adjudicator",
distinct=True,
),
)
return self.filter_queryset_by_scope(
queryset,
required_perm,
project_field="project",
)
[docs]
@action(detail=False, methods=["DELETE"])
def delete_all(self, request, *args, **kwargs):
patient_id = request.query_params.get("patient_id")
project_id = request.query_params.get("project_id")
if not patient_id and not project_id:
return Response(
{
"error": "Either patient_id or project_id \
must be provided for delete all operation.",
},
status=status.HTTP_400_BAD_REQUEST,
)
queryset = super().get_queryset()
project = None
if patient_id:
if not patient_id.isdigit():
raise ValidationError(
{"patient_id": "Invalid patient_id. Must be an integer."},
)
queryset = queryset.filter(id=patient_id)
project = Patient.objects.get(id=patient_id).project
elif project_id:
if not project_id.isdigit():
raise ValidationError(
{"project_id": "Invalid project_id. Must be an integer."},
)
queryset = queryset.filter(project_id=project_id)
project = Project.objects.get(id=project_id)
if not has_permission(
user=request.user,
codename="delete_all_patients",
project=project,
):
return Response(
{"error": "No permission to delete all patients."},
status=status.HTTP_403_FORBIDDEN,
)
_, deleted_details = queryset.delete()
model_label = Patient._meta.label # ruff: ignore[private-member-access]
model_deleted_count = deleted_details.get(model_label, 0)
return Response(
{"deleteCount": model_deleted_count},
status=status.HTTP_200_OK,
)
[docs]
@action(detail=False, methods=["GET"])
def meta(self, request): # ruff: ignore[too-many-locals]
patients = self.get_queryset().order_by()
total_patients = patients.count()
total_notes = Note.objects.filter(patient__in=patients).order_by().distinct().count()
total_labs = Lab.objects.filter(patient__in=patients).order_by().distinct().count()
patients_qs = (
patients
.select_related("project")
.annotate(
ann_cnt=Count("patient_annotations__annotator", distinct=True),
adj_cnt=Count("adjudications", distinct=True),
has_adjudicable_ann=Exists(
LabelValue.objects
.filter(
patient_annotations__patient_id=OuterRef("pk"),
category__isnull=False,
)
.exclude(category__name__iexact="none")
.exclude(category__name__iexact="historic"),
),
)
.annotate(
ann_status=Case(
When(ann_cnt=0, then=Value("not_started")),
When(
ann_cnt__lt=F("project__required_patient_annotations"),
then=Value("in_progress"),
),
default=Value("complete"),
output_field=CharField(),
),
adj_status=Case(
When(has_adjudicable_ann=False, then=Value("skipped")),
When(adj_cnt=0, then=Value("not_started")),
When(
adj_cnt__lt=F("project__required_patient_adjudications"),
then=Value("in_progress"),
),
default=Value("complete"),
output_field=CharField(),
),
)
)
patients_agg = patients_qs.aggregate(
ann_not_started=Count("id", filter=Q(ann_status="not_started")),
ann_in_progress=Count("id", filter=Q(ann_status="in_progress")),
ann_complete=Count("id", filter=Q(ann_status="complete")),
adj_skipped=Count("id", filter=Q(adj_status="skipped")),
adj_not_started=Count("id", filter=Q(adj_status="not_started")),
adj_in_progress=Count("id", filter=Q(adj_status="in_progress")),
adj_complete=Count("id", filter=Q(adj_status="complete")),
)
project_ids = patients.values_list("project_id", flat=True).distinct()
classification_categories = LabelCategory.objects.filter(
project_id__in=project_ids,
project__annotation_type="classification",
)
# classification counts
notes_by_classification_category = (
Note.objects
.filter(
patient__in=patients,
patient_annotations__isnull=False,
)
.order_by()
.values(
cat_id=F("patient_annotations__event_label_values__category_id"),
)
.annotate(
notes_count=Count("id", distinct=True),
)
)
notes_classification_map = {
row["cat_id"]: row["notes_count"] for row in notes_by_classification_category
}
label_category_counts = [
{
"id": cat.id,
"name": cat.name,
"notes_count": notes_classification_map.get(cat.id, 0),
}
for cat in classification_categories
]
# ner counts
ner_anns = list(
NERAnnotation.objects.filter(note__patient__in=patients).values(
"note_id",
"spans",
),
)
ner_label_to_notes = defaultdict(set)
for ann in ner_anns:
note_id = ann["note_id"]
for span in ann["spans"]:
for label_id in span.get("labels", []):
ner_label_to_notes[label_id].add(note_id)
ner_values = list(
LabelValue.objects.filter(id__in=ner_label_to_notes.keys()).values(
"id",
"name",
),
)
ner_label_stats = [
{
"id": lv["id"],
"name": lv["name"],
"notes_count": len(ner_label_to_notes[lv["id"]]),
}
for lv in ner_values
]
return Response(
{
"patients_cnt": total_patients,
"notes_cnt": total_notes,
"labs_cnt": total_labs,
"annotations": {
"not_started": patients_agg["ann_not_started"],
"in_progress": patients_agg["ann_in_progress"],
"complete": patients_agg["ann_complete"],
},
"adjudications": {
"skipped": patients_agg["adj_skipped"],
"not_started": patients_agg["adj_not_started"],
"in_progress": patients_agg["adj_in_progress"],
"complete": patients_agg["adj_complete"],
},
"label_category_counts": label_category_counts,
"ner_label_counts": ner_label_stats,
},
)