import gradio as gr import pandas as pd import json from collections import defaultdict # Create tokenizer for biomed model from transformers import pipeline, AutoTokenizer, AutoModelForTokenClassification tokenizer = AutoTokenizer.from_pretrained("d4data/biomedical-ner-all") # https://huggingface.co/d4data/biomedical-ner-all?text=asthma model = AutoModelForTokenClassification.from_pretrained("d4data/biomedical-ner-all") pipe = pipeline("ner", model=model, tokenizer=tokenizer, aggregation_strategy="simple") # Matplotlib for entity graph import matplotlib.pyplot as plt plt.switch_backend("Agg") # Load examples from JSON import os # Load terminology datasets: basedir = os.path.dirname(__file__) #dataLOINC = pd.read_csv(basedir + "\\" + f'LoincTableCore.csv') #dataPanels = pd.read_csv(basedir + "\\" + f'PanelsAndForms-ACW1208Labeled.csv') #dataSNOMED = pd.read_csv(basedir + "\\" + f'sct2_TextDefinition_Full-en_US1000124_20220901.txt',sep='\t') #dataOMS = pd.read_csv(basedir + "\\" + f'SnomedOMS.csv') #dataICD10 = pd.read_csv(basedir + "\\" + f'ICD10Diagnosis.csv') dataLOINC = pd.read_csv(f'LoincTableCore.csv') dataPanels = pd.read_csv(f'PanelsAndForms-ACW1208Labeled.csv') dataSNOMED = pd.read_csv(f'sct2_TextDefinition_Full-en_US1000124_20220901.txt',sep='\t') dataOMS = pd.read_csv(f'SnomedOMS.csv') dataICD10 = pd.read_csv(f'ICD10Diagnosis.csv') dir_path = os.path.dirname(os.path.realpath(__file__)) EXAMPLES = {} #with open(dir_path + "\\" + "examples.json", "r") as f: with open("examples.json", "r") as f: example_json = json.load(f) EXAMPLES = {x["text"]: x["label"] for x in example_json} def MatchLOINC(name): #basedir = os.path.dirname(__file__) pd.set_option("display.max_rows", None) #data = pd.read_csv(basedir + "\\" + f'LoincTableCore.csv') data = dataLOINC swith=data.loc[data['COMPONENT'].str.contains(name, case=False, na=False)] return swith def MatchLOINCPanelsandForms(name): #basedir = os.path.dirname(__file__) #data = pd.read_csv(basedir + "\\" + f'PanelsAndForms-ACW1208Labeled.csv') data = dataPanels # Assessment Name: #swith=data.loc[data['ParentName'].str.contains(name, case=False, na=False)] # Assessment Question: swith=data.loc[data['LoincName'].str.contains(name, case=False, na=False)] return swith def MatchSNOMED(name): #basedir = os.path.dirname(__file__) #data = pd.read_csv(basedir + "\\" + f'sct2_TextDefinition_Full-en_US1000124_20220901.txt',sep='\t') data = dataSNOMED swith=data.loc[data['term'].str.contains(name, case=False, na=False)] return swith def MatchOMS(name): #basedir = os.path.dirname(__file__) #data = pd.read_csv(basedir + "\\" + f'SnomedOMS.csv') data = dataOMS swith=data.loc[data['SNOMED CT'].str.contains(name, case=False, na=False)] return swith def MatchICD10(name): #basedir = os.path.dirname(__file__) #data = pd.read_csv(basedir + "\\" + f'ICD10Diagnosis.csv') data = dataICD10 swith=data.loc[data['Description'].str.contains(name, case=False, na=False)] return swith def SaveResult(text, outputfileName): #try: basedir = os.path.dirname(__file__) savePath = outputfileName print("Saving: " + text + " to " + savePath) from os.path import exists file_exists = exists(savePath) if file_exists: with open(outputfileName, "a") as f: #append #for line in text: f.write(str(text.replace("\n"," "))) f.write('\n') else: with open(outputfileName, "w") as f: #write #for line in text: f.write(str(text.replace("\n"," "))) f.write('\n') #except ValueError as err: # raise ValueError("File Save Error in SaveResult \n" + format_tb(err.__traceback__)[0] + err.args[0] + "\nEnd of error message.") from None return def loadFile(filename): try: basedir = os.path.dirname(__file__) loadPath = basedir + "\\" + filename print("Loading: " + loadPath) from os.path import exists file_exists = exists(loadPath) if file_exists: with open(loadPath, "r") as f: #read contents = f.read() print(contents) return contents except ValueError as err: raise ValueError("File Save Error in SaveResult \n" + format_tb(err.__traceback__)[0] + err.args[0] + "\nEnd of error message.") from None return "" def get_today_filename(): from datetime import datetime date = datetime.now().strftime("%Y_%m_%d-%I.%M.%S.%p") #print(f"filename_{date}") 'filename_2023_01_12-03-29-22_AM' return f"MedNER_{date}.csv" def get_base(filename): basedir = os.path.dirname(__file__) loadPath = basedir + "\\" + filename #print("Loading: " + loadPath) return loadPath def group_by_entity(raw): outputFile = get_base(get_today_filename()) out = defaultdict(int) for ent in raw: out[ent["entity_group"]] += 1 myEntityGroup = ent["entity_group"] print("Found entity group type: " + myEntityGroup) # if (myEntityGroup in ['Sign_symptom', 'Detailed_description', 'History', 'Activity', 'Medication', 'DISEASE_DISORDER' ]): if (myEntityGroup not in ['Match All']): eterm = ent["word"].replace('#','') minlength = 3 if len(eterm) > minlength: print("Found eterm: " + eterm) eterm.replace("#","") g1=MatchLOINC(eterm) g2=MatchLOINCPanelsandForms(eterm) g3=MatchSNOMED(eterm) g4=MatchOMS(eterm) g5=MatchICD10(eterm) sAll = "" print("Saving to output file " + outputFile) # Create harmonisation output format of input to output code, name, Text try: # 18 fields, output to labeled CSV dataset for results teaching on scored regret changes to action plan with data inputs col = " 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19" #LOINC g11 = g1['LOINC_NUM'].to_string().replace(","," ").replace("\n"," ") g12 = g1['COMPONENT'].to_string().replace(","," ").replace("\n"," ") s1 = ("LOINC," + myEntityGroup + "," + eterm + ",questions of ," + g12 + "," + g11 + ", Label,Value, Label,Value, Label,Value ") if g11 != 'Series([] )': SaveResult(s1, outputFile) #LOINC Panels g21 = g2['Loinc'].to_string().replace(","," ").replace("\n"," ") g22 = g2['LoincName'].to_string().replace(","," ").replace("\n"," ") g23 = g2['ParentLoinc'].to_string().replace(","," ").replace("\n"," ") g24 = g2['ParentName'].to_string().replace(","," ").replace("\n"," ") # s2 = ("LOINC Panel," + myEntityGroup + "," + eterm + ",name of ," + g22 + "," + g21 + ", and Parent codes of ," + g23 + ", with Parent names of ," + g24 + ", Label,Value ") s2 = ("LOINC Panel," + myEntityGroup + "," + eterm + ",name of ," + g22 + "," + g21 + "," + g24 + ", and Parent codes of ," + g23 + "," + ", Label,Value ") if g21 != 'Series([] )': SaveResult(s2, outputFile) #SNOMED g31 = g3['conceptId'].to_string().replace(","," ").replace("\n"," ").replace("\l"," ").replace("\r"," ") g32 = g3['term'].to_string().replace(","," ").replace("\n"," ").replace("\l"," ").replace("\r"," ") s3 = ("SNOMED Concept," + myEntityGroup + "," + eterm + ",terms of ," + g32 + "," + g31 + ", Label,Value, Label,Value, Label,Value ") if g31 != 'Series([] )': SaveResult(s3, outputFile) #OMS g41 = g4['Omaha Code'].to_string().replace(","," ").replace("\n"," ") g42 = g4['SNOMED CT concept ID'].to_string().replace(","," ").replace("\n"," ") g43 = g4['SNOMED CT'].to_string().replace(","," ").replace("\n"," ") g44 = g4['PR'].to_string().replace(","," ").replace("\n"," ") g45 = g4['S&S'].to_string().replace(","," ").replace("\n"," ") s4 = ("OMS," + myEntityGroup + "," + eterm + ",concepts of ," + g44 + "," + g45 + ", and SNOMED codes of ," + g43 + ", and OMS problem of ," + g42 + ", and OMS Sign Symptom of ," + g41) if g41 != 'Series([] )': SaveResult(s4, outputFile) #ICD10 g51 = g5['Code'].to_string().replace(","," ").replace("\n"," ") g52 = g5['Description'].to_string().replace(","," ").replace("\n"," ") s5 = ("ICD10," + myEntityGroup + "," + eterm + ",descriptions of ," + g52 + "," + g51 + ", Label,Value, Label,Value, Label,Value ") if g51 != 'Series([] )': SaveResult(s5, outputFile) except ValueError as err: raise ValueError("Error in group by entity \n" + format_tb(err.__traceback__)[0] + err.args[0] + "\nEnd of error message.") from None return outputFile def plot_to_figure(grouped): fig = plt.figure() plt.bar(x=list(grouped.keys()), height=list(grouped.values())) plt.margins(0.2) plt.subplots_adjust(bottom=0.4) plt.xticks(rotation=90) return fig def ner(text): raw = pipe(text) ner_content = { "text": text, "entities": [ { "entity": x["entity_group"], "word": x["word"], "score": x["score"], "start": x["start"], "end": x["end"], } for x in raw ], } outputFile = group_by_entity(raw) label = EXAMPLES.get(text, "Unknown") outputDataframe = pd.read_csv(outputFile) return (ner_content, outputDataframe, outputFile) demo = gr.Blocks() with demo: gr.Markdown( """ # 🩺⚕️NLP Clinical Ontology Biomedical NER """ ) input = gr.Textbox(label="Note text", value="") with gr.Tab("Biomedical Entity Recognition"): output=[ gr.HighlightedText(label="NER", combine_adjacent=True), #gr.JSON(label="Entity Counts"), #gr.Label(label="Rating"), #gr.Plot(label="Bar"), gr.Dataframe(label="Dataframe"), gr.File(label="File"), ] examples=list(EXAMPLES.keys()) gr.Examples(examples, inputs=input) input.change(fn=ner, inputs=input, outputs=output) with gr.Tab("Clinical Terminology Resolution"): with gr.Row(variant="compact"): btnLOINC = gr.Button("LOINC") btnPanels = gr.Button("Panels") btnSNOMED = gr.Button("SNOMED") btnOMS = gr.Button("OMS") btnICD10 = gr.Button("ICD10") examples=list(EXAMPLES.keys()) gr.Examples(examples, inputs=input) input.change(fn=ner, inputs=input, outputs=output) #layout="vertical" demo.launch(debug=True)