app.py

import os
import tiger
import pandas as pd
import streamlit as st

ENTRY_METHODS = dict(
    manual='Manual entry of single transcript',
    fasta="Fasta file upload (supports multiple transcripts if they have unique ID's)"
)


@st.cache_data
def convert_df(df):
    # IMPORTANT: Cache the conversion to prevent computation on every rerun
    return df.to_csv().encode('utf-8')


def mode_change_callback():
    if st.session_state.mode in {tiger.RUN_MODES['all'], tiger.RUN_MODES['titration']}:  # TODO: support titration
        st.session_state.check_off_targets = False
        st.session_state.disable_off_target_checkbox = True
    else:
        st.session_state.disable_off_target_checkbox = False


def progress_update(update_text, percent_complete):
    with progress.container():
        st.write(update_text)
        st.progress(percent_complete / 100)


def initiate_run():

    # initialize state variables
    st.session_state.transcripts = None
    st.session_state.input_error = None
    st.session_state.on_target = None
    st.session_state.titration = None
    st.session_state.off_target = None

    # initialize transcript DataFrame
    transcripts = pd.DataFrame(columns=[tiger.ID_COL, tiger.SEQ_COL])

    # manual entry
    if st.session_state.entry_method == ENTRY_METHODS['manual']:
        transcripts = pd.DataFrame({
            tiger.ID_COL: ['ManualEntry'],
            tiger.SEQ_COL: [st.session_state.manual_entry]
        }).set_index(tiger.ID_COL)

    # fasta file upload
    elif st.session_state.entry_method == ENTRY_METHODS['fasta']:
        if st.session_state.fasta_entry is not None:
            fasta_path = st.session_state.fasta_entry.name
            with open(fasta_path, 'w') as f:
                f.write(st.session_state.fasta_entry.getvalue().decode('utf-8'))
            transcripts = tiger.load_transcripts([fasta_path], enforce_unique_ids=False)
            os.remove(fasta_path)

    # convert to upper case as used by tokenizer
    transcripts[tiger.SEQ_COL] = transcripts[tiger.SEQ_COL].apply(lambda s: s.upper())

    # ensure all transcripts have unique identifiers
    if transcripts.index.has_duplicates:
        st.session_state.input_error = "Duplicate transcript ID's detected in fasta file"

    # ensure all transcripts only contain nucleotides A, C, G, T, and wildcard N
    elif not all(transcripts[tiger.SEQ_COL].apply(lambda s: set(s).issubset(tiger.NUCLEOTIDE_TOKENS.keys()))):
        st.session_state.input_error = 'Transcript(s) must only contain upper or lower case A, C, G, and Ts'

    # ensure all transcripts satisfy length requirements
    elif any(transcripts[tiger.SEQ_COL].apply(lambda s: len(s) < tiger.TARGET_LEN)):
        st.session_state.input_error = 'Transcript(s) must be at least {:d} bases.'.format(tiger.TARGET_LEN)

    # run model if we have any transcripts
    elif len(transcripts) > 0:
        st.session_state.transcripts = transcripts


if __name__ == '__main__':

    # app initialization
    if 'mode' not in st.session_state:
        st.session_state.mode = tiger.RUN_MODES['all']
        st.session_state.disable_off_target_checkbox = True
    if 'entry_method' not in st.session_state:
        st.session_state.entry_method = ENTRY_METHODS['manual']
    if 'transcripts' not in st.session_state:
        st.session_state.transcripts = None
    if 'input_error' not in st.session_state:
        st.session_state.input_error = None
    if 'on_target' not in st.session_state:
        st.session_state.on_target = None
    if 'titration' not in st.session_state:
        st.session_state.titration = None
    if 'off_target' not in st.session_state:
        st.session_state.off_target = None

    # title and documentation
    st.title('TIGER Cas13 Efficacy Prediction')

    # mode selection
    col1, col2 = st.columns([0.65, 0.35])
    with col1:
        st.radio(
            label='What do you want to predict?',
            options=tuple(tiger.RUN_MODES.values()),
            key='mode',
            on_change=mode_change_callback,
            disabled=st.session_state.transcripts is not None,
        )
    with col2:
        st.checkbox(
            label='Find off-target effects (slow)',
            key='check_off_targets',
            disabled=st.session_state.disable_off_target_checkbox or st.session_state.transcripts is not None
        )

    # transcript entry
    st.selectbox(
        label='How would you like to provide transcript(s) of interest?',
        options=ENTRY_METHODS.values(),
        key='entry_method',
        disabled=st.session_state.transcripts is not None
    )
    if st.session_state.entry_method == ENTRY_METHODS['manual']:
        st.text_input(
            label='Enter a target transcript:',
            key='manual_entry',
            placeholder='Upper or lower case',
            disabled=st.session_state.transcripts is not None
        )
    elif st.session_state.entry_method == ENTRY_METHODS['fasta']:
        st.file_uploader(
            label='Upload a fasta file:',
            key='fasta_entry',
            disabled=st.session_state.transcripts is not None
        )

    # let's go!
    st.button(label='Get predictions!', on_click=initiate_run, disabled=st.session_state.transcripts is not None)
    progress = st.empty()

    # input error
    error = st.empty()
    if st.session_state.input_error is not None:
        error.error(st.session_state.input_error, icon="🚨")
    else:
        error.empty()

    # on-target results
    on_target_results = st.empty()
    if st.session_state.on_target is not None:
        with on_target_results.container():
            st.write('On-target predictions:', st.session_state.on_target)
            st.download_button(
                label='Download on-target predictions',
                data=convert_df(st.session_state.on_target),
                file_name='on_target.csv',
                mime='text/csv'
            )
    else:
        on_target_results.empty()

    # titration results
    titration_results = st.empty()
    if st.session_state.titration is not None:
        with titration_results.container():
            st.write('Titration predictions:', st.session_state.titration)
            st.download_button(
                label='Download titration predictions',
                data=convert_df(st.session_state.titration),
                file_name='titration.csv',
                mime='text/csv'
            )
    else:
        titration_results.empty()

    # off-target results
    off_target_results = st.empty()
    if st.session_state.off_target is not None:
        with off_target_results.container():
            if len(st.session_state.off_target) > 0:
                st.write('Off-target predictions:', st.session_state.off_target)
                st.download_button(
                    label='Download off-target predictions',
                    data=convert_df(st.session_state.off_target),
                    file_name='off_target.csv',
                    mime='text/csv'
                )
            else:
                st.write('We did not find any off-target effects!')
    else:
        off_target_results.empty()

    # keep trying to run model until we clear inputs (streamlit UI changes can induce race-condition reruns)
    if st.session_state.transcripts is not None:
        st.session_state.on_target, st.session_state.titration, st.session_state.off_target = tiger.tiger_exhibit(
            transcripts=st.session_state.transcripts,
            mode={v: k for k, v in tiger.RUN_MODES.items()}[st.session_state.mode],
            check_off_targets=st.session_state.check_off_targets,
            status_update_fn=progress_update
        )
        st.session_state.transcripts = None
        st.experimental_rerun()