"""Definition of init_app function.""" from __future__ import annotations import logging import os import dash_bootstrap_components as dbc from dash import ALL, Dash, Input, Output, State from dash_auth import BasicAuth from minio import Minio from pydantic import ValidationError from pymongo.collection import Collection from shared.datastore import delete as delete_from_minio from shared.mongodb import count_documents, get_visual_communication, upsert_annotation from shared.mongodb.classes import ModelData, VisualCommunication from shared.mongodb.exceptions import NoDocumentFoundException from .layout import app_layout def init_app( mongo_collection: Collection, minio_client: Minio, ) -> Dash: """Initialise web UI application.""" # setup app app = Dash( name='visual_critical_discourse_analysis_web_ui', external_stylesheets=[ dbc.themes.BOOTSTRAP, ], ) app.title = 'visual critical discourse analysis'.title() app.layout = app_layout # setup authentication auth_dict = { os.getenv('DASH_AUTH_USERNAME'): os.getenv('DASH_AUTH_PASSWORD'), } BasicAuth(app, auth_dict) # define callback: show warning @app.callback( Output('warning-element', 'is_open'), Output('warning-element', 'children'), Input('warning-message', 'data'), ) def show_warning( msg: str | None, ) -> tuple[bool, str]: """Show warning message in web ui.""" if msg is None or msg == '': return False, '' logging.debug('updated warning message: %s', msg) return True, msg # define callback: show info @app.callback( Output('info-element', 'is_open'), Output('info-element', 'children'), Input('info-message', 'data'), ) def show_info( msg: str | None, ) -> tuple[bool, str]: """Show info message in web ui.""" if msg is None or msg == '': return False, '' logging.debug('updated info message: %s', msg) return True, msg # define callback: update progress bar @app.callback( Output('progress-bar', 'value'), Output('progress-bar', 'max'), Output('progress-bar', 'label'), Input('next-button', 'n_clicks'), Input('upload-data', 'filename'), ) def update_progress_bar( n_clicks: int, filename_list: list[str] | None, ) -> tuple[int, int, str]: """Update progress bar in web ui.""" assert isinstance(n_clicks, int) if filename_list is not None: assert isinstance(filename_list, list) # get values from database num_total = count_documents( collection=mongo_collection, only_with_annotation=False, ) num_handled = count_documents( collection=mongo_collection, only_with_annotation=True, ) limit = int(num_total / 20) label_str = f'{num_handled}/{num_total}' if num_handled >= limit else '' return num_handled, num_total, label_str # define callback: upload image @app.callback( Output('info-message', 'data', allow_duplicate=True), Output('warning-message', 'data', allow_duplicate=True), Input('upload-data', 'contents'), State('upload-data', 'filename'), prevent_initial_call=True, ) def upload_images( content_list: list[str] | None, filename_list: list[str] | None, ) -> tuple[str | None, str | None]: """Upload image to database through web ui.""" try: # stop if no input assert (content_list is not None) and ( filename_list is not None ), 'nothing to upload' # handle input failed_filename_list = [] for content, filename in zip(content_list, filename_list): try: # decode image content image = VisualCommunication.decode_image(content=content) except Exception: logging.debug('failed decoding %s', filename) failed_filename_list.append(filename) continue try: # instantiate to upload image to minio vis_com = VisualCommunication.from_name_and_image( name=filename, image=image, minio_client=minio_client, ) except Exception as exc: logging.debug(exc) failed_filename_list.append(filename) continue try: # save to mongodb vis_com.save_to_mongo(collection=mongo_collection) except Exception as exc: logging.debug(exc) failed_filename_list.append(filename) # remove document from minio delete_from_minio( client=minio_client, object_name=vis_com.object_name, ) assert ( len(failed_filename_list) == 0 ), f"failed uploading:{ '\n'.join(failed_filename_list) }" except Exception as exc: logging.debug(exc) return None, str(exc).title() return 'successfully uploaded images'.title(), None # define callback: cycle visual communication data @app.callback( Output('warning-message', 'data', allow_duplicate=True), Output('vis-com-name', 'data'), Output('image-container', 'src'), Output({'type': 'annotation', 'index': ALL}, 'value'), Input('next-button', 'n_clicks'), State('vis-com-name', 'data'), State('image-container', 'src'), State({'type': 'annotation', 'index': ALL}, 'id'), State({'type': 'annotation', 'index': ALL}, 'value'), prevent_initial_call=True, ) def cycle_vis_com_data( n_clicks: int, vis_com_name: str, image_src: str, annotation_keys: list, annotation_values: list, ) -> tuple: """Cycle visual communcation data.""" assert isinstance(n_clicks, int) assert isinstance(vis_com_name, str) assert isinstance(image_src, str) assert isinstance(annotation_keys, list) assert isinstance(annotation_values, list) # prepare default response response = [ '', vis_com_name, image_src, annotation_values, ] # stop early if possible if n_clicks == 0: return tuple(response) # check if visual communication name is set if len(vis_com_name) > 0: logging.info('saving annotations to database: %s', vis_com_name) try: # extract option keys annotation_keys = [elem['index'] for elem in annotation_keys] # ensure all options are set logging.info(annotation_keys) for option, value in zip(annotation_keys, annotation_values): if value is None: raise ValueError(f"{option} is not set") # prepare data to save annotation_keys = [elem.replace(' ', '_') for elem in annotation_keys] annotation_values = [ elem.replace(' ', '_').lower() for elem in annotation_values ] annotation_map = { key: value for key, value in zip(annotation_keys, annotation_values) } # instantiate ModelOutputs object annotations = ModelData.from_annotations(**annotation_map) # save data to database upsert_annotation( collection=mongo_collection, vis_com_name=vis_com_name, annotations=annotations, ) except (ValueError, ValidationError) as exc: msg = f"failed saving annotation: {exc}" logging.warning(msg) response[0] = msg return tuple(response) # get new visual communication logging.info('trying to get new visual communication') try: # get data vis_com = get_visual_communication( collection=mongo_collection, with_annotation=False, ) # set variables vis_com_name = vis_com.name image_src = vis_com.webencoded_image(minio_client=minio_client) if vis_com.prediction is not None: # TODO: update to use optional predictions pass else: # reset annotations annotation_values = [None for elem in annotation_values] except NoDocumentFoundException: msg = 'no unannotated data in database' logging.warning(msg) response[0] = msg return tuple(response) else: response[1] = vis_com_name response[2] = image_src response[3] = annotation_values logging.info( 'finished getting visual communication: %s', vis_com_name, ) return tuple(response) logging.info('initialised app') return app