src/web/app.py

@@ -18,6 +18,8 @@ from src.database import get_visual_communication
 from src.database import ModelOutputs
 from src.database import NoDocumentFoundException
 from src.database import upsert_annotations
+from src.database import upsert_visual_communication
+from src.database import VisualCommunication
 
 # setup app
 app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
@@ -46,12 +48,74 @@ def show_alert(
 ):
     if msg is None or msg == '':
         return False, ''
-    logging.info(f"updated alert message: {msg}")
+    logging.info(f'updated alert message: {msg}')
     return True, msg
 
 
 @app.callback(
-    Output('alert-message', 'data'),
+    Output('success-element', 'is_open'),
+    Output('success-element', 'children'),
+    Input('success-message', 'data'),
+)
+def show_success(
+    msg: str | None,
+):
+    if msg is None or msg == '':
+        return False, ''
+    logging.info(f"updated success message: {msg}")
+    return True, msg
+
+
+@app.callback(
+    Output('success-message', 'data', allow_duplicate=True),
+    Output('alert-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,
+):
+    logging.info('began uploading images')
+    global collection
+    # stop early if possible
+    if content_list is None or filename_list is None:
+        logging.warning('content_list is None -> nothing to upload')
+        return None, 'nothing selected to upload'.title()
+    # build list of visual communication
+    visual_communication_list = []
+    for content, filename in zip(content_list, filename_list):
+        try:
+            image = VisualCommunication.decode_image(content)
+            vis_com = VisualCommunication(
+                name=filename,
+                image=image,
+            )
+        except Exception as exc:
+            logging.warning(
+                (
+                    'failed creating VisualCommunication object '
+                    f"from file {filename}"
+                ),
+                exc,
+            )
+        else:
+            visual_communication_list.append(vis_com)
+    # upsert documents
+    success = upsert_visual_communication(
+        collection=collection,
+        visual_communication_list=visual_communication_list,
+    )
+    if success:
+        logging.info(f"succesfully uploaded {len(filename_list)} images")
+        return 'successfully uploaded images'.title(), None
+    logging.warning('failed inserting images into database')
+    return None, 'failed uploading images'.title()
+
+
+@app.callback(
+    Output('alert-message', 'data', allow_duplicate=True),
     Output('vis-com-name', 'data'),
     Output('image-container', 'src'),
     Output({'type': 'annotation', 'index': ALL}, 'value'),