app.py

import logging
from random import randint

from h2o_wave import Q, main, app, copy_expando, handle_on, on

import cards
import constants

# Set up logging
logging.basicConfig(format='%(levelname)s:\t[%(asctime)s]\t%(message)s', level=logging.INFO)


@app('/')
async def serve(q: Q):
    """
    Main entry point. All queries pass through this function.
    """

    try:
        # Initialize the app if not already
        if not q.app.initialized:
            await initialize_app(q)

        # Initialize the client (browser tab) if not already
        if not q.client.initialized:
            await initialize_client(q)

        # Update theme if toggled
        elif q.args.theme_dark is not None and q.args.theme_dark != q.client.theme_dark:
            await update_theme(q)

        # Delegate query to query handlers
        elif await handle_on(q):
            pass

        # Adding this condition to help in identifying bugs (instead of seeing a blank page in the browser)
        else:
            await handle_fallback(q)

    except Exception as error:
        await show_error(q, error=str(error))


async def initialize_app(q: Q):
    """
    Initialize the app.
    """

    logging.info('Initializing app')

    # Set initial argument values
    q.app.cards = ['ner_entities', 'ner_annotator', 'error']

    q.app.initialized = True


async def initialize_client(q: Q):
    """
    Initialize the client (browser tab).
    """

    logging.info('Initializing client')

    # Set initial argument values
    q.client.theme_dark = True

    q.client.ner_tags = constants.NER_TAGS
    q.client.ner_data = constants.NER_DATA
    q.client.ner_index = 0
    q.client.disable_next = False
    q.client.disable_previous = True

    # Add layouts, header and footer
    q.page['meta'] = cards.meta
    q.page['header'] = cards.header
    q.page['footer'] = cards.footer

    # Add cards for main page
    q.page['ner_entities'] = cards.ner_entities(ner_tags=q.client.ner_tags)
    q.page['ner_annotator'] = cards.ner_annotator(
        ner_tags=q.client.ner_tags,
        ner_items=q.client.ner_data[q.client.ner_index],
        disable_next=q.client.disable_next,
        disable_previous=q.client.disable_previous
    )

    q.client.initialized = True

    await q.page.save()


async def update_theme(q: Q):
    """
    Update theme of app.
    """

    # Copying argument values to client
    copy_expando(q.args, q.client)

    if q.client.theme_dark:
        logging.info('Updating theme to dark mode')

        # Update theme from light to dark mode
        q.page['meta'].theme = 'h2o-dark'
        q.page['header'].icon_color = 'black'
    else:
        logging.info('Updating theme to light mode')

        # Update theme from dark to light mode
        q.page['meta'].theme = 'light'
        q.page['header'].icon_color = '#FEC924'

    await q.page.save()


@on('next')
async def show_next_text(q: Q):
    """
    Show next NER data.
    """

    logging.info('Showing the next NER data')

    # Save annotation
    copy_expando(q.args, q.client)
    q.client.ner_data[q.client.ner_index] = q.client.ner_annotator

    # Move to next text
    q.client.ner_index += 1
    q.client.disable_previous = False

    # Disable 'Next' if last text
    if q.client.ner_index == len(q.client.ner_data) - 1:
        q.client.disable_next = True

    # Display data
    q.page['ner_annotator'] = cards.ner_annotator(
        ner_tags=q.client.ner_tags,
        ner_items=q.client.ner_data[q.client.ner_index],
        disable_next=q.client.disable_next,
        disable_previous=q.client.disable_previous
    )

    await q.page.save()


@on('previous')
async def show_previous_text(q: Q):
    """
    Show previous NER data.
    """

    logging.info('Showing the previous NER data')

    # Save annotation
    copy_expando(q.args, q.client)
    q.client.ner_data[q.client.ner_index] = q.client.ner_annotator

    # Move to previous text
    q.client.ner_index -= 1
    q.client.disable_next = False

    # Disable 'Previous' if first text
    if q.client.ner_index == 0:
        q.client.disable_previous = True

    # Display data
    q.page['ner_annotator'] = cards.ner_annotator(
        ner_tags=q.client.ner_tags,
        ner_items=q.client.ner_data[q.client.ner_index],
        disable_next=q.client.disable_next,
        disable_previous=q.client.disable_previous
    )

    await q.page.save()


@on('add')
async def add_entity(q: Q):
    """
    Add a new entity to NER tags.
    """

    logging.info('Adding a new entity')

    # Save annotation
    copy_expando(q.args, q.client)
    q.client.ner_data[q.client.ner_index] = q.client.ner_annotator

    # Add new entity
    if len(q.client.new_entity_name) > 0:
        q.client.ner_tags.append({
            'name': q.client.new_entity_name.lower(),
            'label': q.client.new_entity_name,
            'color': '#{:02x}{:02x}{:02x}'.format(randint(0, 255), randint(0, 255), randint(0, 255))
        })

    # Refresh data with new entity
    q.page['ner_entities'] = cards.ner_entities(ner_tags=q.client.ner_tags)
    q.page['ner_annotator'] = cards.ner_annotator(
        ner_tags=q.client.ner_tags,
        ner_items=q.client.ner_data[q.client.ner_index],
        disable_next=q.client.disable_next,
        disable_previous=q.client.disable_previous
    )

    await q.page.save()


@on('delete')
async def delete_entity(q: Q):
    """
    Delete an entity from NER tags.
    """

    logging.info('Deleting an entity')

    # Save annotation
    copy_expando(q.args, q.client)
    q.client.ner_data[q.client.ner_index] = q.client.ner_annotator

    # Delete entity and it's tags
    if len(q.client.ner_tags) > 1:
        q.client.ner_tags = [tag for tag in q.client.ner_tags if tag['name'] != q.client.delete_entity_name]
        for text in q.client.ner_data:
            for item in text:
                if 'tag' in item.keys():
                    if item['tag'] == q.client.delete_entity_name:
                        item.pop('tag')
    else:
        logging.info('No entities deleted since annotator requires at least one entity available')

    # Refresh data with remaining entities
    q.page['ner_entities'] = cards.ner_entities(ner_tags=q.client.ner_tags)
    q.page['ner_annotator'] = cards.ner_annotator(
        ner_tags=q.client.ner_tags,
        ner_items=q.client.ner_data[q.client.ner_index],
        disable_next=q.client.disable_next,
        disable_previous=q.client.disable_previous
    )

    await q.page.save()


def clear_cards(q: Q, card_names: list):
    """
    Clear cards from the page.
    """

    logging.info('Clearing cards')

    # Delete cards from the page
    for card_name in card_names:
        del q.page[card_name]


async def show_error(q: Q, error: str):
    """
    Displays errors.
    """

    logging.error(error)

    # Clear all cards
    clear_cards(q, q.app.cards)

    # Format and display the error
    q.page['error'] = cards.crash_report(q)

    await q.page.save()


@on('reload')
async def reload_client(q: Q):
    """
    Reset the client (browser tab).
    This function is called when the user clicks "Reload" on the crash report.
    """

    logging.info('Reloading client')

    # Clear all cards
    clear_cards(q, q.app.cards)

    # Reload the client
    await initialize_client(q)


async def handle_fallback(q: Q):
    """
    Handle fallback cases.
    This function should never get called unless there is a bug in our code or query handling logic.
    """

    logging.info('Adding fallback page')

    q.page['fallback'] = cards.fallback

    await q.page.save()