# 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/>.
import re
from collections.abc import Iterable
from io import BytesIO
from pathlib import Path
from typing import TYPE_CHECKING
import numpy as np
import pandas as pd
from django.conf import settings
from django.db.models import Prefetch
from django.db.models import QuerySet
from django.utils.html import escape
from nlpmed_portal.annotations.models import Lab
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.models import Section
if TYPE_CHECKING:
from datetime import date
[docs]
def read_dataframe(
file_path: Path,
file_format: str,
nrows: int | None = None,
) -> pd.DataFrame:
if file_format == "csv":
input_df = pd.read_csv(
file_path,
nrows=nrows,
delimiter=",",
encoding="utf-8",
encoding_errors="ignore",
)
elif file_format in {"xlsx", "xls"}:
input_df = pd.read_excel(file_path, nrows=nrows)
elif file_format == "dta":
input_df = pd.read_stata(
file_path,
convert_categoricals=False,
convert_missing=False,
)
elif file_format == "parquet":
input_df = pd.read_parquet(file_path, engine="pyarrow")
elif file_format == "feather":
input_df = pd.read_feather(file_path)
else:
msg = f"Unsupported file format: {file_format}"
raise ValueError(msg)
if nrows is not None:
max_len = 50
input_df = input_df.head(nrows).map(
lambda x: x[:max_len] + "..." if isinstance(x, str) and len(x) > max_len else x,
)
# DRF doesn't support anything other than None
return input_df.astype(object).where(pd.notna(input_df), None)
[docs]
def get_required_fields_for_import(
*,
import_target: str,
import_missing_patients: bool,
) -> list:
if import_target == "patient":
return [
{"name": "pat_id", "type": "char"},
{"name": "index_date", "type": "date"},
]
if import_target == "note":
if import_missing_patients:
return [
{"name": "pat_id", "type": "char"},
{"name": "index_date", "type": "date"},
{"name": "note_id", "type": "char"},
{"name": "note_text", "type": "char"},
{"name": "note_date", "type": "date"},
{"name": "note_type", "type": "char"},
]
return [
{"name": "pat_id", "type": "char"},
{"name": "note_id", "type": "char"},
{"name": "note_text", "type": "char"},
{"name": "note_date", "type": "date"},
{"name": "note_type", "type": "char"},
]
if import_target == "lab":
if import_missing_patients:
return [
{"name": "pat_id", "type": "char"},
{"name": "index_date", "type": "date"},
{"name": "lab_id", "type": "char"},
{"name": "component", "type": "char"},
{"name": "collection_datetime", "type": "datetime"},
{"name": "value", "type": "float"},
{"name": "unit", "type": "char"},
{"name": "abnormal", "type": "bool"},
]
return [
{"name": "pat_id", "type": "char"},
{"name": "lab_id", "type": "char"},
{"name": "component", "type": "char"},
{"name": "collection_datetime", "type": "datetime"},
{"name": "value", "type": "float"},
{"name": "unit", "type": "char"},
{"name": "abnormal", "type": "bool"},
]
return []
[docs]
def pivot_labs(labs_qs: QuerySet[Lab]) -> dict[str, list]:
data = list(
labs_qs.values(
"component",
"collection_datetime",
"value",
"unit",
"abnormal",
),
)
if not data:
return {"columns": [], "data": []}
labs_df = pd.DataFrame(data)
labs_df["collection_datetime"] = pd.to_datetime(
labs_df["collection_datetime"],
utc=True,
errors="coerce",
)
labs_df["collection_datetime"] = labs_df["collection_datetime"].dt.tz_convert(
settings.TIME_ZONE,
)
labs_df["collection_date"] = labs_df["collection_datetime"].dt.date
labs_df = labs_df.sort_values("collection_datetime")
labs_df["datetime_str"] = labs_df["collection_datetime"].dt.strftime(
settings.DATETIME_FORMAT,
)
labs_df["value_unit_str"] = (
labs_df["value"]
.astype(str)
.str.cat(
labs_df["unit"].astype(str),
sep=" ",
)
)
labs_df["value_unit_str"] = labs_df["value_unit_str"].apply(escape)
labs_df["tooltip_attr"] = (
'data-bs-toggle="tooltip" data-bs-title="' + labs_df["datetime_str"] + '"'
)
labs_df["css_class"] = np.where(labs_df["abnormal"], "abnormal_lab", "")
labs_df["value_unit_html"] = (
'<span class="'
+ labs_df["css_class"]
+ '" '
+ labs_df["tooltip_attr"]
+ ">"
+ labs_df["value_unit_str"]
+ "</span>"
)
df_pivot = labs_df.pivot_table(
index="component",
columns="collection_date",
values="value_unit_html",
aggfunc=", ".join,
).fillna("")
df_pivot = df_pivot.sort_index(axis=1)
df_pivot.columns = [col.strftime(settings.DATE_FORMAT) for col in df_pivot.columns]
df_pivot = df_pivot.reset_index()
return {
"columns": df_pivot.columns,
"data": df_pivot.to_dict(orient="records"),
}
[docs]
def highlight_keywords(
text: str,
keywords: list[str] | None = None,
css_class="pink-hl",
) -> str:
if not keywords:
return text
escaped_keywords = map(re.escape, keywords)
joined_keywords = "|".join(escaped_keywords)
keyword_pattern = re.compile(
rf"(?<![a-zA-Z0-9])(?:{joined_keywords})(?![a-zA-Z0-9])",
flags=re.IGNORECASE,
)
return keyword_pattern.sub(
lambda m: f'<span class="{css_class}">{m.group(0)}</span>',
text,
)
[docs]
def highlight_sentences_in_note(
note_text: str,
sections: Iterable[Section],
important_css_class="yellow-hl",
duplicate_css_class="gray-hl",
) -> str:
all_sentences = []
for sec in sections:
for sent in sec.section_sentences.all():
if sent.is_important:
all_sentences.append(
(sent.start_index, sent.end_index, important_css_class),
)
if sent.is_duplicate:
all_sentences.append(
(sent.start_index, sent.end_index, duplicate_css_class),
)
if not all_sentences:
return note_text
all_sentences.sort(key=lambda x: x[0])
highlighted_text = []
last_index = 0
for start, end, css_class in all_sentences:
highlighted_text.append(note_text[last_index:start])
highlighted_text.append(
f'<span class="{css_class}">{note_text[start:end]}</span>',
)
last_index = end
highlighted_text.append(note_text[last_index:])
return "".join(highlighted_text)
[docs]
def build_html_note_text(
note: Note,
sections: Iterable[Section],
keywords: list[str] | None = None,
) -> str:
html_note_text = highlight_sentences_in_note(
note.note_text,
sections,
)
html_note_text = highlight_keywords(html_note_text, keywords)
return html_note_text.strip()
[docs]
def build_html_hover_text(
sections: list[Section],
keywords: list[str] | None = None,
) -> str:
results = []
for sec in sections:
all_sents = sec.section_sentences.all()
important_sents = [sent for sent in all_sents if sent.is_important]
if not important_sents:
continue
results.append(f"<p>{sec.text[:20]}...</p>")
results.append("<ul>")
for sent in important_sents:
html_sent = highlight_keywords(sent.text, keywords)
results.append(f"<li>{html_sent}</li>")
results.append("</ul>")
return "".join(results)
[docs]
def annotations_to_buffer(output, fmt: str, filename: str) -> tuple[BytesIO, str, str]:
buf = BytesIO()
df = pd.DataFrame(output)
for c in df.columns:
if "date" in c:
df[c] = pd.to_datetime(df[c])
if fmt == "csv":
data = df.to_csv(index=False).encode()
buf.write(data)
content_type = "text/csv"
elif fmt == "xlsx":
df.to_excel(buf, index=False)
content_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
elif fmt == "dta":
df.to_stata(buf, write_index=False, version=118)
content_type = "application/x-stata"
elif fmt == "parquet":
df.to_parquet(buf, index=False)
content_type = "application/vnd.apache.parquet"
elif fmt == "feather":
df.to_feather(buf)
content_type = "application/octet-stream"
else:
msg = f"Unknown format: {fmt}"
raise ValueError(msg)
buf.seek(0)
return buf, content_type, filename
SKIP_LABELS = {"none", "historic"}
WINDOW_GROUP_SIZE_DAYS = 7 # inclusive: start -> start+6 days
[docs]
def is_valid_annotation_lv(lv) -> bool:
return bool(
lv.category and (lv.category.name or "").lower() not in SKIP_LABELS,
)
[docs]
def build_annotation_window_start_map(
patient_id: int,
window_size_days: int = WINDOW_GROUP_SIZE_DAYS,
):
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"),
),
)
)
distinct_event_dates = sorted(
{
ann.event_date
for ann in annotations
if ann.event_date
and any(is_valid_annotation_lv(lv) for lv in ann.event_label_values.all())
},
)
event_date_windows: list[list[date]] = []
for d in distinct_event_dates:
if (not event_date_windows) or (d - event_date_windows[-1][0]).days >= window_size_days:
event_date_windows.append([d])
else:
event_date_windows[-1].append(d)
return {date: window[0] for window in event_date_windows for date in window}