import gradio as gr import torch from datetime import datetime from dateutil import parser from demo_assets import * import re categories = ['Contact related', 'Gathering additional information', 'Defining problem', 'Treatment goal', 'Drug related', 'Therapeutic procedure related', 'Evaluating test result', 'Deferment', 'Advice and precaution', 'Legal and insurance related'] unicode_symbols = [ "\U0001F91D", # Handshake "\U0001F50D", # Magnifying glass "\U0001F9E9", # Puzzle piece "\U0001F3AF", # Target "\U0001F48A", # Pill "\U00002702", # Surgical scissors "\U0001F9EA", # Test tube "\U000023F0", # Alarm clock "\U000026A0", # Warning sign "\U0001F4C4" # Document ] OTHERS_ID = 18 def postprocess_labels(text, logits, t2c): tags = [None for _ in text] labels = logits.argmax(-1) for i,cat in enumerate(labels): if cat != OTHERS_ID: char_ids = t2c(i) if char_ids is None: continue for idx in range(char_ids.start, char_ids.end): if tags[idx] is None and idx < len(text): tags[idx] = categories[cat // 2] for i in range(len(text)-1): if text[i] == ' ' and (text[i+1] == ' ' or tags[i-1] == tags[i+1]): tags[i] = tags[i-1] return tags def indicators_to_spans(labels, t2c = None): def add_span(c, start, end): if t2c(start) is None or t2c(end) is None: start, end = -1, -1 else: start = t2c(start).start end = t2c(end).end span = (c, start, end) spans.add(span) spans = set() num_tokens = len(labels) num_classes = OTHERS_ID // 2 start = None cls = None for t in range(num_tokens): if start and labels[t] == cls + 1: continue elif start: add_span(cls // 2, start, t - 1) start = None # if not start and labels[t] in [2*x for x in range(num_classes)]: if not start and labels[t] != OTHERS_ID: start = t cls = int(labels[t]) // 2 * 2 return spans def extract_date(text): pattern = r'(?<=Date: )\s*(\[\*\*.*?\*\*\]|\d{1,4}[-/]\d{1,2}[-/]\d{1,4})' match = re.search(pattern, text).group(1) start, end = None, None for i, c in enumerate(match): if start is None and c.isnumeric(): start = i elif c.isnumeric(): end = i + 1 match = match[start:end] return match def run_gradio(model, tokenizer): def predict(text): encoding = tokenizer.encode_plus(text) x = torch.tensor(encoding['input_ids']).unsqueeze(0).to(device) mask = torch.ones_like(x) output = model.generate(x, mask)[0] return output, encoding.token_to_chars def process(text): if text is not None: output, t2c = predict(text) tags = postprocess_labels(text, output, t2c) with open('log.csv', 'a') as f: f.write(f'{datetime.now()},{text}\n') return list(zip(text, tags)) else: return text def process_sum(*inputs): global sum_c dates = {} for i in range(sum_c): text = inputs[i] output, t2c = predict(text) spans = indicators_to_spans(output.argmax(-1), t2c) date = extract_date(text) present_decs = set(cat for cat, _, _ in spans) decs = {k: [] for k in sorted(present_decs)} for c, s, e in spans: decs[c].append(text[s:e]) dates[date] = decs out = "" for date in sorted(dates.keys(), key = lambda x: parser.parse(x)): out += f'## **[{date}]**\n\n' decs = dates[date] for c in decs: out += f'### {unicode_symbols[c]} ***{categories[c]}***\n\n' for dec in decs[c]: out += f'{dec}\n\n' return out global sum_c sum_c = 1 SUM_INPUTS = 20 def update_inputs(inputs): outputs = [] if inputs is None: c = 0 else: inputs = [open(f.name).read() for f in inputs] for i, text in enumerate(inputs): outputs.append(gr.update(value=text, visible=True)) c = len(inputs) n = SUM_INPUTS for i in range(n - c): outputs.append(gr.update(value='', visible=False)) global sum_c; sum_c = c return outputs def add_ex(*inputs): global sum_c new_idx = sum_c if new_idx < SUM_INPUTS: out = inputs[:new_idx] + (gr.update(visible=True),) + inputs[new_idx+1:] sum_c += 1 else: out = inputs return out def sub_ex(*inputs): global sum_c new_idx = sum_c - 1 if new_idx > 0: out = inputs[:new_idx] + (gr.update(visible=False),) + inputs[new_idx+1:] sum_c -= 1 else: out = inputs return out device = model.backbone.device # colors = ['aqua', 'blue', 'fuchsia', 'teal', 'green', 'olive', 'lime', 'silver', 'purple', 'red', # 'yellow', 'navy', 'gray', 'white', 'maroon', 'black'] colors = ['#8dd3c7', '#ffffb3', '#bebada', '#fb8072', '#80b1d3', '#fdb462', '#b3de69', '#fccde5', '#d9d9d9', '#bc80bd'] color_map = {cat: colors[i] for i,cat in enumerate(categories)} det_desc = ['Admit, discharge, follow-up, referral', 'Ordering test, consulting colleague, seeking external information', 'Diagnostic conclusion, evaluation of health state, etiological inference, prognostic judgment', 'Quantitative or qualitative', 'Start, stop, alter, maintain, refrain', 'Start, stop, alter, maintain, refrain', 'Positive, negative, ambiguous test results', 'Transfer responsibility, wait and see, change subject', 'Advice or precaution', 'Sick leave, drug refund, insurance, disability'] desc = '### Zones (categories)\n' desc += '| | |\n| --- | --- |\n' for i,cat in enumerate(categories): desc += f'| {unicode_symbols[i]} **{cat}** | {det_desc[i]}|\n' #colors #markdown labels #legend and desc #css font-size css = '.category-legend {border:1px dashed black;}'\ '.text-sm {font-size: 1.5rem; line-height: 200%;}'\ '.gr-sample-textbox {width: 1000px; white-space: nowrap; overflow: hidden; text-overflow: ellipsis;}'\ '.text-limit label textarea {height: 150px !important; overflow: scroll; }'\ '.text-gray-500 {color: #111827; font-weight: 600; font-size: 1.25em; margin-top: 1.6em; margin-bottom: 0.6em;'\ 'line-height: 1.6;}'\ '#sum-out {border: 2px solid #007bff; padding: 20px; border-radius: 10px; box-shadow: 0 0 10px rgba(0, 0, 0, 0.1);' title='Clinical Decision Zoning' with gr.Blocks(title=title, css=css) as demo: gr.Markdown(f'# {title}') with gr.Tab("Label a Clinical Note"): with gr.Row(): with gr.Column(): gr.Markdown("## Enter a Discharge Summary or Clinical Note"), text_input = gr.Textbox( # value=examples[0], label="", placeholder="Enter text here...") text_btn = gr.Button('Run') with gr.Column(): gr.Markdown("## Labeled Summary or Note"), text_out = gr.Highlight(label="", combine_adjacent=True, show_legend=False, color_map=color_map) gr.Examples(text_examples, inputs=text_input) with gr.Tab("Summarize Patient History"): with gr.Row(): with gr.Column(): sum_inputs = [gr.Text(label='Clinical Note 1', elem_classes='text-limit')] sum_inputs.extend([gr.Text(label='Clinical Note %d'%i, visible=False, elem_classes='text-limit') for i in range(2, SUM_INPUTS + 1)]) sum_btn = gr.Button('Run') with gr.Row(): ex_add = gr.Button("+") ex_sub = gr.Button("-") upload = gr.File(label='Upload clinical notes', file_type='text', file_count='multiple') gr.Examples(sum_examples, inputs=upload, fn = update_inputs, outputs=sum_inputs, run_on_click=True) with gr.Column(): gr.Markdown("## Summarized Clinical Decision History") sum_out = gr.Markdown(elem_id='sum-out') gr.Markdown(desc) # Functions text_input.submit(process, inputs=text_input, outputs=text_out) text_btn.click(process, inputs=text_input, outputs=text_out) upload.change(update_inputs, inputs=upload, outputs=sum_inputs) ex_add.click(add_ex, inputs=sum_inputs, outputs=sum_inputs) ex_sub.click(sub_ex, inputs=sum_inputs, outputs=sum_inputs) sum_btn.click(process_sum, inputs=sum_inputs, outputs=sum_out) # demo = gr.TabbedInterface([text_demo, sum_demo], ["Label a Clinical Note", "Summarize Patient History"]) demo.launch(share=False)