demo.py

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)