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' ]): 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 Terms of entity ," + myEntityGroup + ", with term ," + eterm + ", LOINC codes of ," + g11 + ", and LOINC questions of ," + g12 + ", 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 Panels of entity ," + myEntityGroup + ", with term ," + eterm + ", LOINC codes of ," + g21 + ", and LOINC name of ," + g22 + ", and Parent codes of ," + g23 + ", with Parent names of ," + g24 + ", 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 Terms of entity ," + myEntityGroup + ", with term ," + eterm + ", SNOMED concepts of ," + g31 + ", and SNOMED terms of ," + g32 + ", 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 Terms of entity ," + myEntityGroup + ", with term ," + eterm + ", Omaha codes of ," + g41 + ", and SNOMED concepts of ," + g42 + ", and SNOMED codes of ," + g43 + ", and OMS problem of ," + g44 + ", and OMS Sign Symptom of ," + g45) 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 matches of entity ," + myEntityGroup + ", with term ," + eterm + ", ICD10 codes of ," + g51 + ", and descriptions of ," + g52 + ", 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 #print(sAll) #return out; #break; # out["total"] = sum(out.values()) # return out 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 ], } #grouped = group_by_entity(raw) outputFile = group_by_entity(raw) #figure = plot_to_figure(grouped) label = EXAMPLES.get(text, "Unknown") #meta = { # "entity_counts": grouped, # "entities": len(set(grouped.keys())), # "counts": sum(grouped.values()), # } #return (ner_content, meta, label, figure) outputDataframe = pd.read_csv(outputFile) #outputFile = outputFile.replace(os.path.dirname(__file__) + "\\","") # Just filename for File download UI output element #return (ner_content, meta, label, figure, outputDataframe, outputFile) return (ner_content, outputDataframe, outputFile) # New way = Gradio Blocks: demo = gr.Blocks() with demo: gr.Markdown( """ # 🩺⚕️NLP Clinical Ontology Biomedical NER """ ) input = gr.Textbox(label="Note text", value="") #output=[ # gr.HighlightedText(label="NER", combine_adjacent=True) #] 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"): #output=[ # gr.Textbox(placeholder="CT Match Results", lines=10) #] 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") #output=[ # gr.HighlightedText(label="NER", combine_adjacent=True), # gr.File(label="File"), # add download link here # gr.Dataframe(label="Dataframe", headers=["LOINC", "Panels", "SNOMED", "OMS", "ICD10"]), # add harmonised output for input corpus here as a dataframe to UI # gr.Textbox(placeholder="CT Match Results", lines=10) # add matched text scratchpad here #] #textCT = gr.Textbox(placeholder="CT Match Results", lines=10) #btnLOINC.click(loadFile, inputs=["LOINCTerms.txt"], outputs=output) #btnPanels.click(loadFile, "LOINCPanelsandForms.txt", output) #btnSNOMED.click(loadFile, "SNOMEDTerms.txt", output) #btnOMS.click(loadFile, "OMSTerms.txt", output) #btnICD10.click(loadFile, "ICD10Terms.txt", output) examples=list(EXAMPLES.keys()) gr.Examples(examples, inputs=input) input.change(fn=ner, inputs=input, outputs=output) #with gr.Tab("Examples Page 1"): # gr.Examples(["a", "b", "c"], inputs=input) #with gr.Tab("Examples Page 2"): # gr.Examples(["d", "e", "f"], inputs=input) #with gr.Tab("Examples Page 2"): # gr.Examples(["g", "h", "i"], inputs=input) demo.launch(debug=True) # Old Way - Interface Load #interface = gr.Interface( # ner, # inputs=gr.Textbox(label="Note text", value=""), # outputs=[ # gr.HighlightedText(label="NER", combine_adjacent=True), # gr.JSON(label="Entity Counts"), # gr.Label(label="Rating"), # gr.Plot(label="Bar"), # ], # examples=list(EXAMPLES.keys()), # allow_flagging="never", #) #interface.launch()