# 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 OrderedDict
from collections import defaultdict
from datetime import timedelta
from typing import Any
from django.conf import settings
from django.db.models import Prefetch
from django.db.models import QuerySet
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 AdjudicationSerializer
from nlpmed_portal.annotations.api.viewsets.base import BaseViewSet
from nlpmed_portal.annotations.models import Adjudication
from nlpmed_portal.annotations.models import LabelValue
from nlpmed_portal.annotations.models import Note
from nlpmed_portal.annotations.models import PatientAnnotation
from nlpmed_portal.annotations.permissions import MethodBasePermission
from nlpmed_portal.annotations.utils import build_annotation_window_start_map
from nlpmed_portal.annotations.utils import is_valid_annotation_lv
[docs]
class AdjudicationViewSet(BaseViewSet):
queryset = Adjudication.objects.select_related(
"patient",
"adjudicator",
).prefetch_related(
Prefetch(
"event_label_values",
queryset=LabelValue.objects.select_related("category"),
),
Prefetch(
"notes",
queryset=Note.objects.all(),
),
Prefetch(
"source_annotations",
queryset=PatientAnnotation.objects.all(),
),
)
serializer_class = AdjudicationSerializer
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_adjudication",
"HEAD": "view_adjudication",
"OPTIONS": "view_adjudication",
"POST": "add_adjudication",
"PUT": "change_adjudication",
"PATCH": "change_adjudication",
"DELETE": "delete_adjudication",
}
[docs]
def get_queryset(self):
required_perm = self.required_perms.get("GET")
queryset = super().get_queryset()
patient_id = self.request.query_params.get("patient_id")
if patient_id:
if not patient_id.isdigit():
raise ValidationError(
{"patient_id": "Invalid patient_id. Must be an integer."},
)
queryset = queryset.filter(patient_id=patient_id)
return self.filter_queryset_by_scope(
queryset,
required_perm,
project_field="patient__project",
)
[docs]
@action(detail=False, methods=["GET"])
def pivot(self, request, *args, **kwargs): # ruff: ignore[complex-structure, too-many-branches, too-many-locals, too-many-statements]
patient_id = self.request.query_params.get("patient_id")
if not patient_id:
raise ValidationError(
{"patient_id": "patient_id query parameter is required."},
)
window_size_days = 7 # inclusive: start -> start+6 days
def fmt(d):
return d.strftime(settings.DATE_FORMAT)
# Annotations
annotations: QuerySet[PatientAnnotation] = (
PatientAnnotation.objects
.select_related("annotator")
.filter(patient_id=patient_id)
.order_by("event_date", "annotator__username")
.prefetch_related(
Prefetch(
"event_label_values",
queryset=LabelValue.objects.select_related("category"),
),
)
)
event_date_to_window = build_annotation_window_start_map(
int(patient_id),
window_size_days=window_size_days,
)
# get window_start for given date, if not, use date itself
def window_start_for(d):
return event_date_to_window.get(d, d)
# grouped windows
grouped: dict[str, dict[str, Any]] = {}
annotator_usernames = {} # uid -> username
lv_meta = {} # lv_id -> (cat_name, val_name)
for ann in annotations:
if not ann.event_date:
continue
valid_lvs = [lv for lv in ann.event_label_values.all() if is_valid_annotation_lv(lv)]
if not valid_lvs:
continue
win_start = window_start_for(ann.event_date)
win_start_str = fmt(win_start)
grp = grouped.get(win_start_str)
if grp is None:
grp = {
"window_start": win_start_str,
"window_end": fmt(win_start + timedelta(days=6)),
"by_annotator": defaultdict(
lambda: {"annotation_ids": set(), "labels": defaultdict(set)},
),
}
grouped[win_start_str] = grp
uid = ann.annotator_id
uname = ann.annotator.username
annotator_usernames[uid] = uname
cell = grp["by_annotator"][uid]
cell["annotation_ids"].add(ann.pk)
ann_date_str = fmt(ann.event_date)
for lv in valid_lvs:
lv_meta[lv.pk] = (lv.category.name, lv.name)
cell["labels"][ann_date_str, lv.pk].add((ann.pk, uname))
annotators = sorted(
annotator_usernames.items(),
key=lambda x: (x[1] or "").lower(),
)
# Adjudications
adjs = self.get_queryset().filter(patient_id=patient_id)
adj_by_window: dict[str, Adjudication] = {}
for adj in adjs:
anchor_ann = next(
(ann for ann in adj.source_annotations.all() if ann.event_date),
None,
)
if anchor_ann is None:
continue
win_start = window_start_for(anchor_ann.event_date)
win_start_str = fmt(win_start)
# keep latest by updated_at then id
existing = adj_by_window.get(win_start_str)
if existing is None or (adj.updated_at, adj.id) > (
existing.updated_at,
existing.id,
):
adj_by_window[win_start_str] = adj
# Build output
rows = []
for wkey in sorted(grouped.keys()):
grp = grouped[wkey]
row = OrderedDict(
[
("window_start", grp["window_start"]),
("window_end", grp["window_end"]),
],
)
by_annotator = grp["by_annotator"]
for idx, (uid, _uname) in enumerate(annotators, start=1):
cell = by_annotator.get(uid)
if not cell:
row[f"annotator{idx}_annotation_ids"] = []
row[f"annotator{idx}_labels"] = None
continue
row[f"annotator{idx}_annotation_ids"] = sorted(cell["annotation_ids"])
if not cell["labels"]:
row[f"annotator{idx}_labels"] = None
continue
def label_sort_key(k):
ann_date_str, lv_id = k
cat_name, val_name = lv_meta.get(lv_id, ("", ""))
return (
ann_date_str,
(cat_name or "").lower(),
(val_name or "").lower(),
lv_id,
)
labels_out = []
for ann_date_str, lv_id in sorted(
cell["labels"].keys(),
key=label_sort_key,
):
cat_name, val_name = lv_meta.get(lv_id, ("", ""))
srcs = sorted(
cell["labels"][ann_date_str, lv_id],
key=lambda t: ((t[1] or "").lower(), t[0]),
)
labels_out.append(
{
"date": ann_date_str,
"category": cat_name,
"value": val_name,
"label_value_id": lv_id,
"sources": [
{"annotation_id": ann_id, "username": username}
for (ann_id, username) in srcs
],
},
)
row[f"annotator{idx}_labels"] = labels_out if labels_out else None
# adjudicator fields for this window
adj = adj_by_window.get(wkey)
if adj:
row["adjudication_id"] = adj.id
row["adjudicator_date"] = fmt(adj.event_date) if adj.event_date else None
row["adjudication_source_annotation_ids"] = sorted(
adj.source_annotations.values_list("id", flat=True),
)
adj_labels = [
{
"category": lv.category.name,
"value": lv.name,
"label_value_id": lv.pk,
"sources": [
{
"adjudication_id": adj.id,
"username": adj.adjudicator.username,
},
],
}
for lv in adj.event_label_values.all()
]
adj_labels.sort(
key=lambda o: (
(o["category"] or "").lower(),
(o["value"] or "").lower(),
o["label_value_id"],
),
)
row["adjudicator_labels"] = adj_labels if adj_labels else None
else:
row["adjudication_id"] = None
row["adjudicator_date"] = None
row["adjudicator_labels"] = None
row["adjudication_source_annotation_ids"] = []
rows.append(row)
return Response(rows, status=status.HTTP_200_OK)
[docs]
@action(detail=False, methods=["GET"])
def timeline(self, request, *args, **kwargs): # ruff: ignore[complex-structure]
patient_id = self.request.query_params.get("patient_id")
if not patient_id:
raise ValidationError(
{"patient_id": "patient_id query parameter is required."},
)
annotations: QuerySet[PatientAnnotation] = (
PatientAnnotation.objects
.select_related("annotator")
.filter(patient_id=patient_id)
.order_by("event_date")
.prefetch_related(
Prefetch(
"event_label_values",
queryset=LabelValue.objects.select_related("category"),
),
)
)
adjs = self.get_queryset().filter(patient_id=patient_id)
excluded_cats = {"none", "historic"}
def build_content_from_lvs(lvs):
cat_to_vals = defaultdict(list)
for lv in lvs:
cat_to_vals[lv.category.name].append(lv.name)
blocks = []
for cat in sorted(cat_to_vals.keys(), key=lambda s: (s or "").lower()):
vals = sorted(set(cat_to_vals[cat]), key=lambda s: (s or "").lower())
blocks.append(f"{cat}<br/>{', '.join(vals)}")
return "<br/><br/>".join(blocks)
usernames = set()
items = []
all_dates = []
for ann in annotations:
if not ann.event_date:
continue
valid_lvs = [
lv
for lv in ann.event_label_values.all()
if lv.category and (lv.category.name or "").lower() not in excluded_cats
]
if not valid_lvs:
continue
uname = ann.annotator.username
usernames.add(uname)
items.append(
{
"id": f"ann:{ann.pk}",
"kind": "annotation",
"group": f"user:{uname}",
"start": ann.event_date.isoformat(),
"content": build_content_from_lvs(valid_lvs),
"title": ann.event_date.strftime(settings.DATE_FORMAT),
},
)
all_dates.append(ann.event_date)
for adj in adjs:
if not adj.event_date:
continue
valid_lvs = [
lv
for lv in adj.event_label_values.all()
if lv.category and (lv.category.name or "").lower() not in excluded_cats
]
if not valid_lvs:
continue
uname = adj.adjudicator.username
usernames.add(uname)
items.append(
{
"id": f"adj:{adj.pk}",
"kind": "adjudication",
"group": f"user:{uname}",
"start": adj.event_date.isoformat(),
"content": build_content_from_lvs(valid_lvs),
"title": adj.event_date.strftime(settings.DATE_FORMAT),
},
)
all_dates.append(adj.event_date)
groups = [
{"id": f"user:{u}", "content": u}
for u in sorted(usernames, key=lambda s: (s or "").lower())
]
range_obj = None
if all_dates:
dmin = min(all_dates)
dmax = max(all_dates)
range_obj = {
"start": (dmin - timedelta(days=2)).isoformat(),
"end": (dmax + timedelta(days=2)).isoformat(),
}
return Response(
{
"groups": groups,
"items": items,
"range": range_obj,
},
status=status.HTTP_200_OK,
)