# 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 auditlog.registry import auditlog
from django.contrib.auth.models import Group
from django.db import models
from django.db.models import Q
from simple_history.models import HistoricalRecords
from simple_history.models import HistoricForeignKey
from nlpmed_portal.users.models import User
[docs]
class Project(models.Model):
CLASSIFICATION = "classification"
NER = "ner"
ANNOTATION_TYPE_CHOICES = [
(CLASSIFICATION, "Classification"),
(NER, "Named-Entity Recognition"),
]
history = HistoricalRecords()
name = models.CharField(max_length=100, unique=True)
description = models.TextField(blank=True, max_length=1000)
disp_range_months_before_idx = models.PositiveIntegerField(null=True, blank=True)
disp_range_months_after_idx = models.PositiveIntegerField(null=True, blank=True)
run_nlp = models.BooleanField(default=True)
annotation_type = models.CharField(
max_length=20,
choices=ANNOTATION_TYPE_CHOICES,
default=CLASSIFICATION,
)
required_patient_annotations = models.PositiveSmallIntegerField(default=2)
required_patient_adjudications = models.PositiveSmallIntegerField(default=1)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
def __str__(self) -> str:
return self.name
[docs]
class LabelCategory(models.Model):
history = HistoricalRecords()
name = models.CharField(max_length=50)
project = HistoricForeignKey(
Project,
on_delete=models.CASCADE,
related_name="label_categories",
)
class Meta:
unique_together = ("name", "project")
def __str__(self) -> str:
return f"{self.name} ({self.project.name})"
[docs]
class LabelValue(models.Model):
history = HistoricalRecords()
name = models.CharField(max_length=50)
category = HistoricForeignKey(
LabelCategory,
on_delete=models.CASCADE,
related_name="label_values",
)
class Meta:
unique_together = ("name", "category")
def __str__(self) -> str:
return f"{self.name} ({self.category.project.name}: {self.category.name})"
[docs]
class Patient(models.Model):
history = HistoricalRecords()
pat_id = models.CharField(max_length=50)
index_date = models.DateField()
project = HistoricForeignKey(
Project,
on_delete=models.CASCADE,
related_name="patients",
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
unique_together = (("project", "pat_id"),)
permissions = (
("import_patient", "Can import patients"),
("delete_all_patients", "Can delete all patients"),
)
def __str__(self) -> str:
return f"Patient {self.pat_id}"
[docs]
class Note(models.Model):
history = HistoricalRecords()
note_id = models.CharField(max_length=50)
note_text = models.TextField()
note_date = models.DateField()
TYPE_CHOICES = [
("progress", "Progress Note"),
("radiology", "Radiology Note"),
("discharge", "Discharge Summary"),
]
note_type = models.CharField(max_length=20, choices=TYPE_CHOICES)
patient = HistoricForeignKey(
Patient,
on_delete=models.CASCADE,
related_name="notes",
)
preprocessed_text = models.TextField(blank=True)
predicted_label = models.CharField(max_length=100, blank=True)
predicted_score = models.FloatField(blank=True, null=True)
important_sentences_count = models.PositiveIntegerField(default=0)
duplicate_sentences_count = models.PositiveIntegerField(default=0)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ["id"]
permissions = (
("import_note", "Can import notes"),
("rerun_nlp", "Can rerun nlp"),
("delete_all_notes", "Can delete all notes"),
)
def __str__(self) -> str:
return f"Note {self.note_id} ({self.get_note_type_display()})"
[docs]
class Lab(models.Model):
history = HistoricalRecords()
lab_id = models.CharField(max_length=50)
patient = HistoricForeignKey(Patient, on_delete=models.CASCADE, related_name="labs")
component = models.CharField(max_length=100)
collection_datetime = models.DateTimeField()
value = models.FloatField()
unit = models.CharField(max_length=20)
abnormal = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
permissions = (
("import_lab", "Can import labs"),
("delete_all_labs", "Can delete all labs"),
)
def __str__(self) -> str:
return f"Lab {self.lab_id} - {self.component}"
[docs]
class Section(models.Model):
history = HistoricalRecords()
text = models.TextField()
start_index = models.PositiveIntegerField()
end_index = models.PositiveIntegerField()
note = HistoricForeignKey(
Note,
on_delete=models.CASCADE,
related_name="note_sections",
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
indexes = [
models.Index(fields=["note"]),
]
def __str__(self) -> str:
return f"Section {self.pk} of Note {self.note.note_id}"
@property
def important_sentences(self):
return self.section_sentences.filter(is_important=True)
@property
def duplicate_sentences(self):
return self.section_sentences.filter(is_duplicate=True)
@property
def expanded_sentences(self):
return self.section_sentences.filter(is_expanded=True)
[docs]
class Sentence(models.Model):
history = HistoricalRecords()
text = models.TextField()
start_index = models.PositiveIntegerField()
end_index = models.PositiveIntegerField()
is_duplicate = models.BooleanField(default=False)
is_important = models.BooleanField(default=False)
is_expanded = models.BooleanField(default=False)
section = HistoricForeignKey(
Section,
on_delete=models.CASCADE,
related_name="section_sentences",
)
class Meta:
indexes = [
models.Index(
fields=["section"],
name="sentence_imp_idx",
condition=Q(is_important=True),
),
models.Index(
fields=["section"],
name="sentence_dup_idx",
condition=Q(is_duplicate=True),
),
]
def __str__(self) -> str:
return f"Sentence {self.pk} in Section {self.section.pk}"
[docs]
class PatientAnnotation(models.Model):
event_date = models.DateField(blank=True, null=True)
event_label_values = models.ManyToManyField(
LabelValue,
related_name="patient_annotations",
blank=True,
)
comment = models.TextField(blank=True, max_length=1000)
patient = HistoricForeignKey(
Patient,
on_delete=models.CASCADE,
related_name="patient_annotations",
)
annotator = HistoricForeignKey(
User,
on_delete=models.CASCADE,
related_name="patient_annotations",
)
notes = models.ManyToManyField(Note, related_name="patient_annotations")
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
history = HistoricalRecords(m2m_fields=[notes])
class Meta:
permissions = (
(
"view_patientannotation_history",
"Can view history of patient annotations",
),
("view_all_patientannotations", "Can view all patient annotations"),
("delete_all_patientannotations", "Can delete all patient annotations"),
)
def __str__(self) -> str:
return f"Annotation {self.pk} by {self.annotator.username}"
[docs]
class Adjudication(models.Model):
history = HistoricalRecords()
patient = HistoricForeignKey(
Patient,
on_delete=models.CASCADE,
related_name="adjudications",
)
event_date = models.DateField()
event_label_values = models.ManyToManyField(
LabelValue,
related_name="adjudications",
blank=True,
)
adjudicator = HistoricForeignKey(
User,
on_delete=models.PROTECT,
related_name="adjudications",
)
notes = models.ManyToManyField(
Note,
related_name="adjudications",
blank=True,
)
source_annotations = models.ManyToManyField(
PatientAnnotation,
related_name="adjudications",
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
permissions = (
(
"view_adjudication_history",
"Can view history of adjudications",
),
("view_all_adjudications", "Can view all adjudications"),
("delete_all_adjudications", "Can delete all adjudications"),
)
def __str__(self) -> str:
return f"Adjudication {self.pk} by {self.adjudicator.username}"
[docs]
class ProjectMembership(models.Model):
history = HistoricalRecords()
assignee = HistoricForeignKey(
User,
on_delete=models.CASCADE,
related_name="project_memberships_assignee",
)
project = HistoricForeignKey(
Project,
on_delete=models.CASCADE,
related_name="project_memberships_project",
)
group = HistoricForeignKey(
Group,
on_delete=models.CASCADE,
related_name="project_memberships_group",
)
assigner = HistoricForeignKey(
User,
null=True,
on_delete=models.SET_NULL,
related_name="project_memberships_assigner",
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
unique_together = ("assignee", "project", "group")
def __str__(self):
return f"{self.assignee} in {self.project} ({self.group})"
[docs]
class Task(models.Model):
STATUS_CHOICES = [
("pending", "Pending"),
("completed", "Completed"),
("closed", "Closed (not needed)"),
]
CLASSIFICATION = "classification"
NER = "ner"
ADJUDICATION = "adjudication"
TYPE_CHOICES = [
(CLASSIFICATION, "Classification"),
(NER, "NER"),
(ADJUDICATION, "Adjudication"),
]
assignee = HistoricForeignKey(
User,
on_delete=models.CASCADE,
related_name="tasks_assignee",
)
patient = HistoricForeignKey(
Patient,
on_delete=models.CASCADE,
related_name="tasks_patient",
)
assigner = HistoricForeignKey(
User,
on_delete=models.CASCADE,
related_name="tasks_assigner",
)
task_type = models.CharField(
max_length=20,
choices=TYPE_CHOICES,
default="classification",
)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="pending")
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
history = HistoricalRecords()
class Meta:
unique_together = ("assignee", "patient", "task_type")
permissions = (
("view_all_tasks", "Can view all tasks"),
("delete_all_tasks", "Can delete all tasks"),
)
def __str__(self):
return f"{self.assignee} → {self.patient} ({self.status})"
[docs]
class ImportJob(models.Model):
STATUS_CHOICES = (
("uploaded", "Uploaded"),
("in_progress", "In Progress"),
("completed", "Completed"),
("failed", "Failed"),
)
project = HistoricForeignKey(
Project,
on_delete=models.CASCADE,
related_name="import_jobs",
)
user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
import_file = models.FileField(upload_to="imports/")
import_format = models.CharField(max_length=10, default="csv")
import_target = models.CharField(
max_length=50,
choices=[
("patient", "Patient"),
("note", "Note"),
("lab", "Lab"),
],
default="patient",
)
import_missing_patients = models.BooleanField(default=False)
column_map = models.JSONField(null=True, blank=True)
date_format_map = models.JSONField(null=True, blank=True)
new_rows_count = models.PositiveIntegerField(default=0)
job_status = models.CharField(
max_length=20,
choices=STATUS_CHOICES,
default="uploaded",
)
job_progress = models.PositiveIntegerField(default=0)
error_message = models.TextField(blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
def __str__(self) -> str:
return f"ImportJob #{self.pk} - {self.import_target} ({self.job_status})"
[docs]
def delete_import_file(self, *, save: bool = True) -> None:
"""Delete the uploaded source file after it is no longer needed."""
if not self.import_file:
return
self.import_file.delete(save=False)
self.import_file = ""
if save and self.pk:
self.save(update_fields=["import_file", "updated_at"])
[docs]
class NERAnnotation(models.Model):
history = HistoricalRecords()
patient = HistoricForeignKey(
Patient,
on_delete=models.CASCADE,
related_name="ner_annotations",
)
annotator = HistoricForeignKey(
User,
on_delete=models.CASCADE,
related_name="ner_annotations",
)
note = models.OneToOneField(
Note,
on_delete=models.CASCADE,
related_name="ner_annotations",
primary_key=True,
)
spans = models.JSONField(
default=list,
blank=True,
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
permissions = (
(
"view_nerannotation_history",
"Can view history of NER annotations",
),
("view_all_nerannotations", "Can view all NER annotations"),
("delete_all_nerannotations", "Can delete all NER annotations"),
)
def __str__(self) -> str:
return f"Annotation {self.pk} by {self.annotator.username}"
[docs]
class ExportAnnotation(models.Model):
pass
auditlog.register(
Project,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
LabelCategory,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
LabelValue,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
Patient,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
Note,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
Lab,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
Section,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
Sentence,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
PatientAnnotation,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
m2m_fields=("notes", "event_label_values"),
)
auditlog.register(
Adjudication,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
m2m_fields=("notes", "event_label_values", "source_annotations"),
)
auditlog.register(
ProjectMembership,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
Task,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
NERAnnotation,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)