Spaces:
Sleeping
Sleeping
| 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) | |
| 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() | |
| 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() | |
| 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() | |
| 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() | |
| 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() | |
| 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() | |