Utility & Helper Modules

# import the module
from sparknlp_jsl.alab import AnnotationLab
alab = AnnotationLab()

alab.get_classification_data(

# required: path to NLP Lab JSON export
input_json_path='alab_demo.json',

# optional: set to True to select ground truth completions, False to select latest completions,
# defaults to False
# ground_truth=False)

alab.get_conll_data(

# required: Spark session with spark-nlp-jsl jar
spark=spark,

# required: path to NLP Lab JSON export
input_json_path="alab_demo.json",

# required: name of the CoNLL file to save
output_name="conll_demo",

# optional: path for CoNLL file saving directory, defaults to 'exported_conll'
# save_dir="exported_conll",

# optional: set to True to select ground truth completions, False to select latest completions,
# defaults to False
# ground_truth=False,

# optional: labels to exclude from CoNLL; these are all assertion labels and irrelevant NER labels,
# defaults to empty list
# excluded_labels=['ABSENT'],

# optional: set a pattern to use regex tokenizer, defaults to regular tokenizer if pattern not defined
# regex_pattern="\\s+|(?=[-.:;*+,$&%\\[\\]])|(?<=[-.:;*+,$&%\\[\\]])"

# optional: list of NLP Lab task titles to exclude from CoNLL, defaults to empty list
# excluded_task_ids = [2, 3]

# optional: list of NLP Lab task titles to exclude from CoNLL, defaults to None
# excluded_task_titles = ['Note 1'])

alab.get_assertion_data(

# required: SparkSession with spark-nlp-jsl jar
spark=spark,

# required: path to NLP Lab JSON export
input_json_path = 'alab_demo.json',

# required: annotated assertion labels to train on
assertion_labels = ['ABSENT'],

# required: relevant NER labels that are assigned assertion labels
relevant_ner_labels = ['PROBLEM', 'TREATMENT'],

# optional: set to True to select ground truth completions, False to select latest completions,
# defaults to False
# ground_truth = False,

# optional: assertion label to assign to entities that have no assertion labels, defaults to None
# unannotated_label = 'PRESENT',

# optional: set a pattern to use regex tokenizer, defaults to regular tokenizer if pattern not defined
# regex_pattern = "\\s+|(?=[-.:;*+,$&%\\[\\]])|(?<=[-.:;*+,$&%\\[\\]])",

# optional: set the strategy to control the number of occurrences of the unannotated assertion label
# in the output dataframe, options are 'weighted' or 'counts', 'weighted' allows to sample using a
# fraction, 'counts' allows to sample using absolute counts, defaults to None
# unannotated_label_strategy = None,

# optional: dictionary in the format {'ENTITY_LABEL': sample_weight_or_counts} to control the number of
# occurrences of the unannotated assertion label in the output dataframe, where 'ENTITY_LABEL' are the
# NER labels that are assigned the unannotated assertion label, and sample_weight_or_counts should be
# between 0 and 1 if `unannotated_label_strategy` is 'weighted' or between 0 and the max number of
# occurrences of that NER label if `unannotated_label_strategy` is 'counts'
# unannotated_label_strategy_dict = {'PROBLEM': 0.5, 'TREATMENT': 0.5},

# optional: list of NLP Lab task IDs to exclude from output dataframe, defaults to None
# excluded_task_ids = [2, 3]

# optional: list of NLP Lab task titles to exclude from output dataframe, defaults to None
# excluded_task_titles = ['Note 1'])

alab.get_relation_extraction_data(

# required: Spark session with spark-nlp-jsl jar
spark=spark,

# required: path to NLP Lab JSON export
input_json_path='alab_demo.json',

# optional: set to True to select ground truth completions, False to select latest completions,
# defaults to False
ground_truth=True,

# optional: set to True to assign a relation label between entities where no relation was annotated,
# defaults to False
negative_relations=True,

# optional: all assertion labels that were annotated in the NLP Lab, defaults to None
assertion_labels=['ABSENT'],

# optional: plausible pairs of entities for relations, separated by a '-', use the same casing as the
# annotations, include only one relation direction, defaults to all possible pairs of annotated entities
relation_pairs=['DATE-PROBLEM','TREATMENT-PROBLEM','TEST-PROBLEM'],

# optional: set the strategy to control the number of occurrences of the negative relation label
# in the output dataframe, options are 'weighted' or 'counts', 'weighted' allows to sample using a
# fraction, 'counts' allows to sample using absolute counts, defaults to None
negative_relation_strategy='weighted',

# optional: dictionary in the format {'ENTITY1-ENTITY2': sample_weight_or_counts} to control the number of
# occurrences of negative relations in the output dataframe for each entity pair, where 'ENTITY1-ENTITY2'
# represent the pairs of entities for relations separated by a `-` (include only one relation direction),
# and sample_weight_or_counts should be between 0 and 1 if `negative_relation_strategy` is 'weighted' or
# between 0 and the max number of occurrences of negative relations if `negative_relation_strategy` is
# 'counts', defaults to None
negative_relation_strategy_dict = {'DATE-PROBLEM': 0.1, 'TREATMENT-PROBLEM': 0.5, 'TEST-PROBLEM': 0.2},

# optional: list of NLP Lab task IDs to exclude from output dataframe, defaults to None
# excluded_task_ids = [2, 3]

# optional: list of NLP Lab task titles to exclude from output dataframe, defaults to None
# excluded_task_titles = ['Note 1'])

pre_annotations, summary = alab.generate_preannotations(

# required: list of results.
all_results = results,

# requied: output column name of 'DocumentAssembler' stage - to get original document string.
document_column = 'document',

# required: column name(s) of ner model(s). Note: multiple NER models can be used, but make sure their results don't overrlap.
# Or use 'ChunkMergeApproach' to combine results from multiple NER models.
ner_columns = ['ner_chunk'],

# optional: column name(s) of assertion model(s). Note: multiple assertion models can be used, but make sure their results don't overrlap.
# assertion_columns = ['assertion_res'],

# optional: column name(s) of relation extraction model(s). Note: multiple relation extraction models can be used, but make sure their results don't overrlap.
# relations_columns = ['relations_clinical', 'relations_pos'],

# optional: This can be defined to identify which pipeline/user/model was used to get predictions.
# Default: 'model'
# user_name = 'model',

# optional: Option to assign custom titles to tasks. By default, tasks will be titled as 'task_#'
# titles_list = [],

# optional: If there are already tasks in project, then this id offset can be used to make sure default titles 'task_#' do not overlap.
# While upload a batch after the first one, this can be set to number of tasks currently present in the project
# This number would be added to each tasks's ID and title.
# id_offset=0)

alab = AnnotationLab()

username=''
password=''
client_secret=''
annotationlab_url=''

alab.set_credentials(

# required: username
username=username,

# required: password
password=password,

# required: secret for you alab instance (every alab installation has a different secret)
client_secret=client_secret,

# required: http(s) url for you NLP lab
annotationlab_url=annotationlab_url)

alab.get_all_projects()

alab.create_project(

# required: unique name of project
project_name = 'alab_demo',

# optional: other details about project. Default: Empty string
project_description='',

# optional: Sampling option of tasks. Default: random
project_sampling='',

# optional: Annotation Guidelines of project
project_instruction='')

alab.delete_project(

# required: unique name of project
project_name = 'alab_demo',

# optional: confirmation for deletion. Default: False - will ask for confirmation. If set to true, will delete directly.
confirm=False)

alab.upload_tasks(

# required: name of project to upload tasks to
project_name='alab_demo',

# required: list of examples / tasks as string (One string is one task).
task_list=task_list,

# optional: Option to assign custom titles to tasks. By default, tasks will be titled as 'task_#'
title_list = [],

# optional: If there are already tasks in project, then this id offset can be used to make sure default titles 'task_#' do not overlap.
# While upload a batch after the first one, this can be set to number of tasks currently present in the project
# This number would be added to each tasks's ID and title.
id_offset=0)

alab.delete_tasks(

# required: name of project to upload tasks to
project_name='alab_demo',

# required: list of ids of tasks.
# note: you can get task ids from the above step. Look for 'task_ids' key.
task_ids=[1, 2],

# optional: confirmation for deletion. Default: False - will ask for confirmation. If set to true, will delete directly.
confirm=False)

alab.upload_preannotations(

# required: name of project to upload annotations to
project_name = 'alab_demo',

# required: preannotation JSON
preannotations = pre_annotations)

from sparknlp_jsl import Deid

deid_implementor= Deid(
# required: Spark session with spark-nlp-jsl jar
spark,

# required: The path of the input file. Default is None. File type must be 'csv' or 'json'.
input_file_path="data.csv",

#optional:  The path of the output file. Default is 'deidentified.csv'. File type must be 'csv' or 'json'.
output_file_path="deidentified.csv",

#optional: The separator of the input csv file. Default is "\t".
separator=",",

#optional: A custom pipeline model to be used for deidentification. If not specified, the default is None.
custom_pipeline=nlpModel,

#optional: Fields to be deidentified and their deidentification modes, by default {"text": "mask"}
fields={"text": "mask", "text_1": "obfuscate"},

#optional:  The masking policy. Default is "entity_labels".
masking_policy="fixed_length_chars",

#optional: The fixed mask length. Default is 4.
fixed_mask_length=4,

#optional: The final chunk column name of the custom pipeline that will be deidentified, if specified. Default is "ner_chunk".
ner_chunk="ner_chunk",

#optional: The corresponding document column name of the custom pipeline, if specified. Default is "document"
document="document",

#optional: The corresponding sentence column name of the custom pipeline, if specified. Default is "sentence"
sentence="sentence",

#optional: The corresponding token column name of the custom pipeline, if specified. Default is "token"
token="token",

#optional: The source of the reference file for obfuscation. Default is "faker".
#obfuscate_ref_source="both",

#optional: The path of the reference file for obfuscation. Default is None.
#obfuscate_ref_file_path="obfuscation.txt",

#optional: Obfuscate date. Default is True.
#obfuscate_date=True,

#optional: The document hash coder column name. Default is "documentHash".
#documentHashCoder_col_name= "documentHash"

#optional: ID column name. Default is "id".
#id_column_name= "ID"

#optional: Date shift column name. Default is "date_shift".
#date_shift_column_name= "date_shift"

#optional: Json file path for multi-mode Deid. Default is NONE.
#multi_mode_file_path= "multi_mode_file_path.json"

#optional: The date tag. Default is "DATE".
#date_tag="DATE"

#optional: Language. Default is "en"
#language="en"

#optional: Region. Default is "us"
#region="us"

#optional: Age group obfuscation. Default is False.
#age_group_obfuscation=True

#optional: Age ranges for obfuscation. Default is [1, 4, 12, 20, 40, 60, 80].
#age_ranges=[1, 4, 12, 20, 40, 60, 80]

#optional:  Shift days. Default is False.
#shift_days=False

#optional: The number of days to shift. Default is None.
#number_of_days=5

#optional: Use unnormalized date. Default is False.
#unnormalized_date=True

#optional: The unnormalized mode. Default is "mask".
#unnormalized_mode="obfuscate"
)

res= deid_implementor.deidentify()

from sparknlp_jsl import Deid

deid_implementor= Deid(
# required: Spark session with spark-nlp-jsl jar
spark,

# required: The path of the input file. Default is None. File type must be 'csv' or 'json'.
input_file_path="data.csv",

#optional:  The path of the output file. Default is 'deidentified.csv'. File type must be 'csv' or 'json'.
output_file_path="deidentified.csv",

#optional: The separator of the input csv file. Default is "\t".
separator=",",

#optional: Fields to be deidentified and their deidentification modes, by default {"text": "mask"}
fields={"text": "mask"},

#optional: The masking policy. Default is "entity_labels".
masking_policy="entity_labels",

#optional: Json file path for multi-mode Deid. Default is NONE.
#multi_mode_file_path= "multi_mode_file_path.json",

#optional: Age group obfuscation. Default is False.
#age_group_obfuscation=True

#optional: Age ranges for obfuscation. Default is [1, 4, 12, 20, 40, 60, 80].
#age_ranges=[1, 4, 12, 20, 40, 60, 80]

#optional:  Shift days. Default is False.
#shift_days=False

#optional: The number of days to shift. Default is None.
#number_of_days=5

#optional: Use unnormalized date. Default is False.
#unnormalized_date=True

#optional: The unnormalized mode. Default is "mask".
#unnormalized_mode="obfuscate"
)

res= deid_implementor.deidentify()

from sparknlp_jsl.structured_deidentification import StructuredDeidentification

obfuscator = StructuredDeidentification(spark,
                                        columns={"PATIENT":"PATIENT","DOB":"DATE","TEL":"PHONE"},
                                        columnsSeed={"PATIENT": 23, "DOB": 23, "TEL": 23},
                                        obfuscateRefSource = "faker",
                                        days=60,
                                        region="eu",
                                        keepYear=True,
                                        keepTextSizeForObfuscation=True
                                        )
obfuscator_df = obfuscator.obfuscateColumns(df, outputAsArray=False, overwrite=False, suffix="_obfuscated")
obfuscator_df.show(truncate=False)

import pandas as pd
from sparknlp_jsl.utils import DateShiftFiller

data = pd.DataFrame(
  {
    "patientID": ["A001", "A002", "A001", "A002", "A003", "A003"],
    "text": [
      "Chris Brown was discharged on 10/02/2022",
      "Mark White was discharged on 03/01/2020",
      "Chris Brown was born on 05/10/1982",
      "Mark White was born on 10/04/2000",
      "John was discharged on 03/15/2022",
      "John Moore was born on 12/31/2002",
    ],
    "dateshift": ["10", "-2", None, None, None, 5],
  }
)

input_df = spark.createDataFrame(data)

filler = DateShiftFiller(spark, seed=42, max_shift_days=60)
result_df = filler.fill_missing_shifts(
  input_df,
  id_col="patientID",
  shift_col="dateshift",
  suffix="_filled"
)

result_df.show(truncate=False)

from johnsnowlabs import medical
# Or: from sparknlp_jsl.compatibility import Compatibility

compatibility = medical.Compatibility()

# Returns a list of dict objects
found_models = compatibility.findVersion('ner_clinical')

import pandas as pd

models_df = pd.DataFrame([dict(x) for x in list(models)])
models_df

compatibility.showVersion('ner_clinical')

medical_ner_models = medical.InternalResourceDownloader.returnPrivateModels("MedicalNerModel")

medical_ner_models[0]

medical.InternalResourceDownloader.showPrivateModels("MedicalNerModel")

# Annotator that transforms a text column from dataframe into an Annotation ready for NLP
documentAssembler = DocumentAssembler()\
    .setInputCol("text")\
    .setOutputCol("document")

sentenceDetector = SentenceDetectorDLModel.pretrained("sentence_detector_dl_healthcare","en","clinical/models")\
    .setInputCols(["document"])\
    .setOutputCol("sentence")

# Tokenizer splits words in a relevant format for NLP
tokenizer = Tokenizer()\
    .setInputCols(["sentence"])\
    .setOutputCol("token")

# Clinical word embeddings trained on PubMED dataset
word_embeddings = WordEmbeddingsModel.pretrained("embeddings_clinical","en","clinical/models")\
    .setInputCols(["sentence","token"])\
    .setOutputCol("embeddings")

# NER model trained on i2b2 (sampled from MIMIC) dataset
clinical_ner = MedicalNerModel.pretrained("ner_clinical_large","en","clinical/models")\
    .setInputCols(["sentence","token","embeddings"])\
    .setOutputCol("ner")\
    .setLabelCasing("upper") #decide if we want to return the tags in upper or lower case

ner_converter = NerConverter()\
    .setInputCols(["sentence","token","ner"])\
    .setOutputCol("ner_chunk")

nlpPipeline = Pipeline(
    stages=[
        documentAssembler,
        sentenceDetector,
        tokenizer,
        word_embeddings,
        clinical_ner,
        ner_converter
        ])


empty_data = spark.createDataFrame([[""]]).toDF("text")

model = nlpPipeline.fit(empty_data)

from sparknlp_jsl.modelTracer import ModelTracer

df = model.transform(empty_data)
tracer_result = ModelTracer().addUidCols(pipeline = nlpPipeline, df = df)
tracer_result.show(truncate=False)

from sparknlp_jsl.pipeline_tracer import PipelineTracer
from sparknlp_jsl.pipeline_output_parser import PipelineOutputParser

PipelineTracer.showAvailableModels(language="en", source="clinical/models")

import sparknlp
import sparknlp_jsl
from sparknlp.pretrained import PretrainedPipeline
from sparknlp_jsl.pipeline_output_parser import PipelineOutputParser

## Load the pipeline and the text

text = [
    "Immunohistochemistry was negative for thyroid transcription factor-1 and napsin A. The test was positive for ER and PR, and negative for HER2.",
    "The had previously undergone a left mastectomy and an axillary lymph node dissection for a left breast cancer twenty years ago." +
    "The tumor was positive for ER and PR. Postoperatively, radiotherapy was administered to the residual breast. The cancer recurred as a right lung metastasis 13 years later." +
    "He underwent a regimen consisting of adriamycin (60 mg/m2) and cyclophosphamide (600 mg/m2) over six courses, as first line therapy."
]

oncology_pipeline = PretrainedPipeline("oncology_biomarker_pipeline", "en", "clinical/models")
annotations = oncology_pipeline.fullAnnotate(text)

## Define the column_maps dictionary

column_maps = {
    'document_identifier': 'XYZ_123',
    'document_text': 'document',
    'entities': ["merged_chunk"],
    'assertions': ["assertion_chunk"],
    'resolver': ["icd10_code"],
    'relations': [],
    'summaries': [],
    "deidentifications": [],
    "classifications": []
}

## Initialized parser from the dictionary

pipeline_parser = PipelineOutputParser(column_maps)

## Run the parser on the output of a Spark NLP pipeline

parsed_result = pipeline_parser.run(annotations)
print(parsed_result)

from sparknlp_jsl.utils.database_deidentification import RelationalDBDeidentification

config = {
    "db_config": {
        "host": "localhost",
        "user": "root",
        "password": "root",
        "database": "healthcare_db"
    },
    "deid_options": {
        "days_to_shift": 10,
        "age_groups": {
            "child": (0, 12),
            "teen": (13, 19),
            "adult": (20, 64),
            "senior": (65, 90)
        },
        "pk_fk_shift_value": 100,
        "use_hipaa": False,
        "output_path": "deidentified_output/"
    },
    "logging": {
        "level": "INFO",
        "file": "deidentification.log"
    }
}

deidentifier = RelationalDBDeidentification(spark, config)
deidentifier.deidentify()

from sparknlp_jsl.utils import apply_exception_handling

from sparknlp.pretrained import PretrainedPipeline

oncology_pipeline = PretrainedPipeline("oncology_biomarker_pipeline", "en", "clinical/models")
stage = oncology_pipeline.model.stages[15]
stage.getDoExceptionHandling()

handled_pretrainedPipeline = apply_exception_handling(oncology_pipeline)

stage = handled_pretrainedPipeline.stages[15]
stage.getDoExceptionHandling()

from sparknlp.base import LightPipeline

light_model = LightPipeline(handled_pretrainedPipeline)
result = light_model.fullAnnotate("The patient is a 55-years old. He has a history of smoking and lung cancer.")

from sparknlp_jsl.training import Annotation2Training
from sparknlp_jsl.annotator import *
from sparknlp.annotator import *
from sparknlp.base import *
from pyspark.ml import Pipeline

# Create a base pipeline to extract sentences and tokens
document_assembler = DocumentAssembler() \
  .setInputCol("text") \
  .setOutputCol("document")
sentence_detector = SentenceDetectorDLModel.pretrained("sentence_detector_dl_healthcare", "en", "clinical/models") \
  .setInputCols(["document"]) \
  .setOutputCol("sentence")
tokenizer = Tokenizer() \
  .setInputCols(["sentence"]) \
  .setOutputCol("token")

pipeline = Pipeline(stages=[
  document_assembler,
  sentence_detector,
  tokenizer,
])

pipeline_model = pipeline.fit(spark.createDataFrame([[""]]).toDF("text"))

# Load the annotation2training utility class
annotation2training = Annotation2Training(spark)

# Convert a json file to a training dataframe
training_df = annotation2training.convertJson2NerDF(
  json_path="path/to/annotations.json",  # Path to GenAI json file
  pipeline_model=pipeline_model,
  repartition=32,                        # Number of partitions to use when creating the DataFrame
  token_output_col="token",
  ner_label_col="label"
)

training_df.show(5)
# Returns a Spark DataFrame to train NER models.
# Ready to go for MedicalNerApproach

list_of_dict = [
  {
    "text": "My name is George, and I was born on 12/11/1995. I have the pleasure of working at John Snow Labs.",
    "chunks": [
      {
        "begin": 11,
        "end": 16, 
        "result": "George",
        "entity": "PERSON",
        "metadata": {"confidence": "1", "ner_source": "llm_output"}
      },
      {
        "begin": 37,
        "end": 46,
        "result": "12/11/1995",
        "entity": "DATE",
        "metadata": {"confidence": "0.9", "ner_source": "llm_output"}
      },
      {
        "begin": 83,
        "end": 96,
        "result": "John Snow labs",
        "entity": "ORG",
        "metadata": {"confidence": "0.87", "ner_source": "llm_output"}
      }
    ],
    "doc_id": "1",
    "document_metadata": {"dateshift": "10"},
    "file_path": "/path/to/file1"
  },
  {
    "text": "I am Bush, and English is my native language. You can reach me at my email: bush@example.com.",
    "chunks": [
      {
        "begin": 5,
        "end": 8,
        "result": "Bush",
        "entity": "PERSON",
        "metadata": {"confidence": "1", "ner_source": "ner_dl"}
      },
      {
        "begin": 15,
        "end": 21,
        "result": "English",
        "entity": "LANGUAGE",
        "metadata": {"confidence": "0.98", "ner_source": "ner_dl"}
      },
      {
        "begin": 76,
        "end": 91,
        "result": "bush@example.com",
        "entity": "EMAIL",
        "metadata": {"confidence": "0.87", "ner_source": "ner_dl"}
      }
    ],
    "doc_id": "2",
    "document_metadata": {"dateshift": "5"},
    "file_path": "/path/to/file2"
  }
]

from sparknlp.annotator import *
from sparknlp_jsl.utils import *

result_df = dict_to_annotation_converter(
  spark, 
  list_of_dict, 
  document_col_name="document", 
  chunk_col_name="chunk", 
  adjust_end=False
)

light_deIdentification = (
  LightDeIdentification()
  .setInputCols(["document", "chunk"])
  .setOutputCol("deidentified")
  .setMode("mask")
)
light_deIdentification.transform(result_df).selectExpr("deidentified.result").show(truncate=False)

df = spark.createDataFrame([
    ("The weather is nice today",),
    ("It will rain tomorrow",),
    ("I love sunny days",)
], ["text"])

result_df = llm_df_preprocessor(
    data_frame=df,
    text_col_name="text",
    prefix_prompt="Analyze the sentiment of this text: ",
    new_text_col_name="prompt"
)
result_df.show(truncate=False)

df = vision_llm_preprocessor(
    spark=spark,
    images_path="/path/to/medical/images/",
    prompt="Analyze this medical image for abnormalities",
    output_col_name="medical_prompt"
)
df.show(5, truncate=False)