# Copyright (c) 2024
# 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/>.
"""Models for the nlp application."""
from auditlog.registry import auditlog
from django.db import models
from simple_history.models import HistoricalRecords
from simple_history.models import HistoricForeignKey
from simple_history.models import HistoricOneToOneField
from nlpmed_portal.annotations.models import Project
from nlpmed_portal.nlp.constants import VTE_KEYWORD_EXC_LIST
from nlpmed_portal.nlp.constants import VTE_KEYWORD_INC_LIST
from nlpmed_portal.nlp.constants import VTE_SECTION_EXC_LIST
from nlpmed_portal.nlp.constants import VTE_SECTION_INC_LIST
from nlpmed_portal.users.models import User
[docs]
def vte_section_inc_list():
return VTE_SECTION_INC_LIST
[docs]
def vte_section_exc_list():
return VTE_SECTION_EXC_LIST
[docs]
def vte_keyword_inc_list():
return VTE_KEYWORD_INC_LIST
[docs]
def vte_keyword_exc_list():
return VTE_KEYWORD_EXC_LIST
[docs]
class ComponentBase(models.Model):
"""Base model for component status."""
STATUS_CHOICES = [
("enabled", "Enabled"),
("disabled", "Disabled"),
("excluded", "Excluded"),
]
status = models.CharField(max_length=10, choices=STATUS_CHOICES, default="enabled")
[docs]
class NLPSetting(models.Model):
history = HistoricalRecords()
debug = models.BooleanField(default=True)
project = HistoricOneToOneField(
Project,
on_delete=models.CASCADE,
related_name="nlp_setting",
)
def __str__(self) -> str:
return f"{self.project.name}"
[docs]
class EncodingFixer(ComponentBase):
"""Model for encoding fixer component status."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="encoding_fixer",
)
[docs]
class PatternReplacer(ComponentBase):
"""Model for pattern replacer component with pattern and target replacements."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="pattern_replacer",
)
pattern = models.CharField(
max_length=30,
blank=True,
default=r"(?:\s*\n\s*){2,}",
help_text="Regex pattern to replace in the text.",
)
target = models.CharField(
max_length=30,
blank=True,
default=r"\n\n",
help_text="Target string to replace matched pattern.",
)
[docs]
class WordMasker(ComponentBase):
"""Model for word masker component with mask settings."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="word_masker",
)
words_to_mask = models.JSONField(
blank=True,
null=True,
default=vte_keyword_exc_list,
help_text="List of words to mask in the text.",
)
mask_char = models.CharField(
max_length=5,
blank=True,
default="*",
help_text="Character used for masking.",
)
[docs]
class NoteFilter(ComponentBase):
"""Model for filtering notes based on keywords."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="note_filter",
)
words_to_search = models.JSONField(
blank=True,
null=True,
default=vte_keyword_inc_list,
help_text="Keywords to search in the notes.",
)
[docs]
class SectionSplitter(ComponentBase):
"""Model for splitting sections using a delimiter."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="section_splitter",
)
delimiter = models.CharField(
max_length=30,
blank=True,
default=r"\n\n",
help_text="Delimiter used to split sections.",
)
[docs]
class SectionFilter(ComponentBase):
"""Model for filtering sections based on include and exclude keywords."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="section_filter",
)
section_inc_list = models.JSONField(
blank=True,
null=True,
default=vte_section_inc_list,
help_text="Keywords for including sections.",
)
section_exc_list = models.JSONField(
blank=True,
null=True,
default=vte_section_exc_list,
help_text="Keywords for excluding sections.",
)
fallback = models.BooleanField(
default=True,
help_text="Enable fallback behavior if no sections match.",
)
[docs]
class SentenceSegmenter(ComponentBase):
"""Model for sentence segmentation settings."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="sentence_segmenter",
)
model_name = models.CharField(
max_length=50,
choices=[
("en_core_sci_lg", "en_core_sci_lg"),
],
default="en_core_sci_lg",
help_text="Name of the model used for sentence segmentation.",
)
batch_size = models.IntegerField(
blank=True,
null=True,
default=10,
help_text="Batch size for processing.",
)
[docs]
class DuplicateChecker(ComponentBase):
"""Model for duplicate checking configuration."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="duplicate_checker",
)
num_perm = models.IntegerField(
blank=True,
null=True,
default=256,
help_text="Number of permutations for MinHash.",
)
sim_threshold = models.FloatField(
blank=True,
null=True,
default=0.9,
help_text="Similarity threshold for duplicates.",
)
length_threshold = models.IntegerField(
blank=True,
null=True,
default=50,
help_text="Length threshold for checking duplicates.",
)
[docs]
class SentenceFilter(ComponentBase):
"""Model for filtering sentences based on keywords."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="sentence_filter",
)
words_to_search = models.JSONField(
blank=True,
null=True,
default=vte_keyword_inc_list,
help_text="Keywords to filter sentences.",
)
[docs]
class SentenceExpander(ComponentBase):
"""Model for expanding short sentences."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="sentence_expander",
)
length_threshold = models.IntegerField(
blank=True,
null=True,
default=50,
help_text="Threshold length for expanding short sentences.",
)
[docs]
class Joiner(ComponentBase):
"""Model for joining sentences and sections."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="joiner",
)
sentence_delimiter = models.CharField(
max_length=30,
blank=True,
default=r"\n",
help_text="Delimiter for joining sentences.",
)
section_delimiter = models.CharField(
max_length=30,
blank=True,
default=r"\n\n",
help_text="Delimiter for joining sections.",
)
[docs]
class MLInference(ComponentBase):
"""Model for machine learning inference settings."""
history = HistoricalRecords()
nlp_setting = HistoricOneToOneField(
NLPSetting,
on_delete=models.CASCADE,
related_name="ml_inference",
)
model_name = models.CharField(
max_length=50,
choices=[
("VTE_MULTICLASS", "vte_multiclass"),
("BLEED_BINARY", "bleed_binary"),
],
default="VTE_MULTICLASS",
help_text="Name of the model used for ML inference.",
)
use_preped_text = models.BooleanField(
default=True,
help_text="Use preprocessed text for inference.",
)
[docs]
class NLP(models.Model):
pass
[docs]
class NlpProcessJob(models.Model):
STATUS_CHOICES = (
("pending", "Pending"),
("in_progress", "In Progress"),
("completed", "Completed"),
("failed", "Failed"),
)
project = HistoricForeignKey(
Project,
on_delete=models.CASCADE,
related_name="nlp_jobs",
)
user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
patient_pks = models.JSONField(default=list)
job_status = models.CharField(
max_length=20,
choices=STATUS_CHOICES,
default="pending",
)
total_count = models.PositiveIntegerField(default=0)
completed_count = 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"NlpProcessJob #{self.pk} ({self.job_status})"
@property
def job_progress(self):
if self.total_count == 0:
return 0
return int((self.completed_count / self.total_count) * 100)
[docs]
def mark_in_progress(self):
self.job_status = "in_progress"
self.save(update_fields=["job_status", "updated_at"])
[docs]
def mark_completed(self):
self.job_status = "completed"
self.completed_count = self.total_count
self.save(update_fields=["job_status", "completed_count", "updated_at"])
[docs]
def mark_failed(self, error=""):
self.job_status = "failed"
self.error_message = error
self.save(update_fields=["job_status", "error_message", "updated_at"])
auditlog.register(
NLPSetting,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
EncodingFixer,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
PatternReplacer,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
WordMasker,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
NoteFilter,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
SectionSplitter,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
SectionFilter,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
SentenceSegmenter,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
DuplicateChecker,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
SentenceFilter,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
SentenceExpander,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
Joiner,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)
auditlog.register(
MLInference,
exclude_fields=(
"created_at",
"updated_at",
"history",
"last_seen_at",
"last_update",
),
mask_fields=("password",),
)