src/database/__init__.py

@@ -1,13 +1,12 @@
-from .classes import (
-    ModelOutputs,
-    VisualCommunication,
-    NoDocumentFoundException
-)
+from __future__ import annotations
+
+from .classes import ModelOutputs
+from .classes import NoDocumentFoundException
+from .classes import VisualCommunication
 from .database import connect
-from .utils import (
-    total_documents,
-    total_annotated,
-    get_visual_communication,
-    upsert_annotations,
-    upsert_predictions,
-)
+from .utils import get_visual_communication
+from .utils import total_annotated
+from .utils import total_documents
+from .utils import upsert_annotations
+from .utils import upsert_predictions
+from .utils import upsert_visual_communication

src/database/classes.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from base64 import b64decode
 from base64 import b64encode
 from io import BytesIO
 from pathlib import Path
@@ -122,6 +123,12 @@ class VisualCommunication(BaseModel):
         image.load()
         return VisualCommunication(name=name, image=image)
 
+    @classmethod
+    def decode_image(cls, content: str) -> Image.Image:
+        """Decode image."""
+        _, content_data = content.split(',')
+        return Image.open(BytesIO(b64decode(content_data)))
+
     @field_serializer('image')
     def serialize_image(image: Image.Image) -> bytes:   # type: ignore
         buffer = BytesIO()

src/database/utils.py

@@ -1,57 +1,58 @@
-from pymongo.collection import Collection
+from __future__ import annotations
+
 import logging
 
-from .classes import (
-    VisualCommunication,
-    NoDocumentFoundException,
-    ModelOutputs
-)
+from pymongo.collection import Collection
+
+from .classes import ModelOutputs
+from .classes import NoDocumentFoundException
+from .classes import VisualCommunication
 
 
 def total_documents(
-    collection: Collection
+    collection: Collection,
 ) -> int:
     """Get total number of documents in database."""
     return collection.count_documents(filter={})
 
 
 def total_annotated(
-    collection: Collection
+    collection: Collection,
 ) -> int:
     """Get total number of annotated documents in database."""
     query = {
-        "annotation": {
-            "$ne": None
-        }
+        'annotation': {
+            '$ne': None,
+        },
     }
     return collection.count_documents(filter=query)
 
 
 def get_visual_communication(
     collection: Collection,
-    with_annotation: bool = False
+    with_annotation: bool = False,
 ) -> VisualCommunication:
     """Get a random visual communication from the database."""
     query = {}
     if with_annotation:
-        query["annotation"] = {"$ne": None}
+        query['annotation'] = {'$ne': None}
     else:
-        query["annotation"] = {"$eq": None}
+        query['annotation'] = {'$eq': None}
     data = collection.aggregate([
         {
-            "$match": query     # find using filters
+            '$match': query,     # find using filters
         },
         {
-            "$sample": {
-                "size": 1   # get one random
-            }
-        }
+            '$sample': {
+                'size': 1,   # get one random
+            },
+        },
     ])
     data_list = list(data)   # read data from cursor object
     if len(data_list) == 0:
-        logging.error("failed getting visual communication")
+        logging.error('failed getting visual communication')
         raise NoDocumentFoundException()
-    logging.info("finished")
+    logging.info('finished')
     return VisualCommunication.model_validate(data_list[0])
 
 
@@ -62,20 +63,20 @@ def upsert_predictions(
 ) -> None:
     """Upsert prediction data in the database."""
     query = {
-        "name": vis_com_name
+        'name': vis_com_name,
     }
     update = {
-        "$set": {
-            "prediction": predictions.model_dump()
-        }
+        '$set': {
+            'prediction': predictions.model_dump(),
+        },
     }
     res = collection.update_one(
         filter=query,
         update=update,
         upsert=True,
     )
-    logging.debug("upserted document: %s", res)
-    logging.info("finished")
+    logging.debug('upserted document: %s', res)
+    logging.info('finished')
 
 
 def upsert_annotations(
@@ -85,17 +86,35 @@ def upsert_annotations(
 ) -> None:
     """Upserts annotation data in the database."""
     query = {
-        "name": vis_com_name
+        'name': vis_com_name,
     }
     update = {
-        "$set": {
-            "annotation": annotations.model_dump()
-        }
+        '$set': {
+            'annotation': annotations.model_dump(),
+        },
     }
     res = collection.update_one(
         filter=query,
         update=update,
         upsert=True,
     )
-    logging.info("upserted document: %s", res)
-    logging.info("finished")
+    logging.info('upserted document: %s', res)
+    logging.info('finished')
+
+
+def upsert_visual_communication(
+    collection: Collection,
+    visual_communication_list: list[VisualCommunication],
+) -> bool:
+    """
+    Upsert VisualCommunication object in the database.
+    Returns bool stating success.
+    """
+    response = collection.insert_many(
+        [
+            vis_com.model_dump()
+            for vis_com
+            in visual_communication_list
+        ],
+    )
+    return response.acknowledged

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'),

src/web/layout/alerts.py

@@ -1,15 +1,23 @@
-from dash import html
+from __future__ import annotations
+
 import dash_bootstrap_components as dbc
+from dash import html
 
 
 alerts_element = html.Div(
     children=[
         dbc.Alert(
-            children="",
-            id="alert-element",
+            children='',
+            id='alert-element',
             dismissable=True,
             fade=False,
             is_open=False,
-        )
-    ]
+        ),
+        dbc.Alert(
+            children='',
+            id='success-element',
+            is_open=False,
+            duration=1000,
+        ),
+    ],
 )

src/web/layout/header.py

@@ -1,21 +1,49 @@
+from __future__ import annotations
+
 import dash_mantine_components as dmc
+from dash import dcc
 from dash import html
 
+title_element = dmc.Center(
+    children=[
+        html.H2(
+            children=[
+                'Visual Critical Discourse Analysis Tool',
+            ],
+        ),
+    ],
+)
+
+upload_button_element = dcc.Upload(
+    children=[
+        dmc.Button(
+            'upload images'.title(),
+            id='upload-button',
+            color='lime',
+            n_clicks=0,
+            variant='outline',
+            radius='sm',
+            size='md',
+        ),
+    ],
+    multiple=True,
+    id='upload-data',
+    accept='image/png,image/jpeg,image/jpg',
+)
 
 header_element = dmc.Header(
     height=80,
     children=[
-        dmc.Container(
+        dmc.Group(
             children=[
-                html.H2(
-                    children="Visual Critical Discourse Analysis Tool",
-                    style={
-                        "margin-left": "20px",
-                        "padding-top": "-5px"
-                    },
-                ),
+                title_element,
+                upload_button_element,
             ],
-            size="xl", px="xl",
+            position='apart',
+            style={
+                'margin': '10px',
+                'padding': '10px',
+            },
         ),
-    ]
+    ],
 )

src/web/layout/stores.py

@@ -1,18 +1,23 @@
-from dash import html, dcc
+from __future__ import annotations
+
 import logging
 import os
 
-storage_type = "session"
-if "ENV" in os.environ and os.getenv("ENV") == "DEV":
-    storage_type = "memory"
+from dash import dcc
+from dash import html
+
+storage_type = 'session'
+if 'ENV' in os.environ and os.getenv('ENV') == 'DEV':
+    storage_type = 'memory'
     logging.info(
-        "ENV=DEV -> dcc.Stores changed to storage_type=%s",
-        storage_type
+        'ENV=DEV -> dcc.Stores changed to storage_type=%s',
+        storage_type,
     )
 
 stores_element = html.Div(
     children=[
-        dcc.Store(id="alert-message", storage_type=storage_type, data=""),
-        dcc.Store(id="vis-com-name", storage_type=storage_type, data=""),
-    ]
+        dcc.Store(id='alert-message', storage_type=storage_type, data=''),
+        dcc.Store(id='success-message', storage_type=storage_type, data=''),
+        dcc.Store(id='vis-com-name', storage_type=storage_type, data=''),
+    ],
 )