Merge pull request 'fe_load_image' (#14) from fe_load_image into main

Reviewed-on: http://192.168.1.2:3000/brian/visual_critical_discourse_analysis/pulls/14

docker-compose.local.yml

@@ -8,6 +8,8 @@ services:
       dockerfile: Dockerfile
     env_file:
       - local.env
+    environment:
+      - ENV=DEV
     ports:
       - 8050:8050
     networks:
@@ -25,6 +27,7 @@ services:
       - backend
   mongo-express:
     image: mongo-express
+    container_name: mongo_express
     ports:
       - 8081:8081
     env_file:

poetry.lock

@@ -251,23 +251,6 @@ files = [
     {file = "dash_table-5.0.0.tar.gz", hash = "sha256:18624d693d4c8ef2ddec99a6f167593437a7ea0bf153aa20f318c170c5bc7308"},
 ]
 
-[[package]]
-name = "discord-webhook"
-version = "1.3.1"
-description = "Easily send Discord webhooks with Python"
-optional = false
-python-versions = ">=3.10,<4.0"
-files = [
-    {file = "discord_webhook-1.3.1-py3-none-any.whl", hash = "sha256:ede07028316de76d24eb811836e2b818b2017510da786777adcb0d5970e7af79"},
-    {file = "discord_webhook-1.3.1.tar.gz", hash = "sha256:ee3e0f3ea4f3dc8dc42be91f75b894a01624c6c13fea28e23ebcf9a6c9a304f7"},
-]
-
-[package.dependencies]
-requests = ">=2.28.1,<3.0.0"
-
-[package.extras]
-async = ["httpx (>=0.23.0,<0.24.0)"]
-
 [[package]]
 name = "dnspython"
 version = "2.6.1"
@@ -806,23 +789,6 @@ files = [
 [package.extras]
 cli = ["click (>=5.0)"]
 
-[[package]]
-name = "python-logging-discord-handler"
-version = "0.1.4"
-description = "Discord handler for Python logging framework"
-optional = false
-python-versions = ">=3.8,<4.0"
-files = [
-    {file = "python_logging_discord_handler-0.1.4-py3-none-any.whl", hash = "sha256:b804b48e3f5af8c9c781a9afe8243c806f01521662e38a60fcda2c3631d27f4f"},
-    {file = "python_logging_discord_handler-0.1.4.tar.gz", hash = "sha256:8bfa839b6503b3b87e5851dd13bc5ff80bf2fadb496ac22c338ac10bd926f75a"},
-]
-
-[package.dependencies]
-discord-webhook = ">=1.0.0,<2.0.0"
-
-[package.extras]
-docs = ["Sphinx (>=4.4.0,<5.0.0)", "sphinx-autodoc-typehints[docs] (>=1.16.0,<2.0.0)", "sphinx-rtd-theme (>=1.0.0,<2.0.0)", "sphinx-sitemap (>=2.2.0,<3.0.0)"]
-
 [[package]]
 name = "requests"
 version = "2.31.0"
@@ -962,4 +928,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.12"
-content-hash = "e4aacea5a98281d935411e0d96152d1d24680f6c1e5288e9a1be913a0536b78e"
+content-hash = "9146cd32f0af25ddc99e5950dbf6ebc5a249f16238834b9a5db8d1b45fa9efb9"

pyproject.toml

@@ -11,7 +11,6 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.12"
 gunicorn = "^21.2.0"
-python-logging-discord-handler = "^0.1.4"
 python-dotenv = "^1.0.1"
 dash = "^2.15.0"
 dash-bootstrap-components = "^1.5.0"

src/database/__init__.py

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

src/database/classes.py

@@ -3,8 +3,13 @@ from pydantic import BaseModel, field_validator, field_serializer
 from PIL import Image
 from io import BytesIO
 from pathlib import Path
+from base64 import b64encode
+import logging
+from typing import List
 
-from src.model_experiential import ExperientialModelOutput
+from src.model_experiential import (
+    VisualSyntaxModelOutput
+)
 from src.model_interpersonal import (
     ContactModelOutput,
     AngleModelOutput,
@@ -21,7 +26,7 @@ from src.model_textual import (
 )
 
 class ModelOutputs(BaseModel):
-    experiential: ExperientialModelOutput
+    visual_syntax: VisualSyntaxModelOutput
     contact: ContactModelOutput
     angle: AngleModelOutput
     point_of_view: PointOfViewModelOutput
@@ -33,10 +38,56 @@ class ModelOutputs(BaseModel):
     framing: FramingModelOutput
     salience: SalienceModelOutput
 
+    @classmethod
+    def list_fields(cls) -> List[str]:
+        """List options that are stored as attributes."""
+        return list(cls.model_fields.keys())
+
+    @classmethod
+    def from_random(cls) -> ModelOutputs:
+        """Instantiate with random numbers."""
+        kwargs = {
+            field: field_info.annotation.from_random()
+            for field, field_info
+            in cls.model_fields.items()
+        }
+        return cls(**kwargs)
+    
+    @classmethod
+    def from_annotations(
+        cls,
+        visual_syntax: str,
+        contact: str,
+        angle: str,
+        point_of_view: str,
+        distance: str,
+        modality_lighting: str,
+        modality_color: str,
+        modality_depth: str,
+        information_value: str,
+        framing: str,
+        salience: str
+    ) -> ModelOutputs:
+        """Instantiate from annotation."""
+        kwargs = {
+            "visual_syntax": VisualSyntaxModelOutput.from_choice(visual_syntax),
+            "contact": ContactModelOutput.from_choice(contact),
+            "angle": AngleModelOutput.from_choice(angle),
+            "point_of_view": PointOfViewModelOutput.from_choice(point_of_view),
+            "distance": DistanceModelOutput.from_choice(distance),
+            "modality_lighting": ModalityLightingModelOutput.from_choice(modality_lighting),
+            "modality_color": ModalityColorModelOutput.from_choice(modality_color),
+            "modality_depth": ModalityDepthModelOutput.from_choice(modality_depth),
+            "information_value": InformationValueModelOutput.from_choice(information_value),
+            "framing": FramingModelOutput.from_choice(framing),
+            "salience": SalienceModelOutput.from_choice(salience)
+        }
+        return cls(**kwargs)
+
 
 class VisualCommunication(BaseModel):
     name: str
-    image: Image.Image | BytesIO | bytes
+    image: Image.Image
     annotation: ModelOutputs | None = None
     prediction: ModelOutputs | None = None
 
@@ -73,3 +124,20 @@ class VisualCommunication(BaseModel):
 
     def __repr__(self) -> str:
         return f"{self.classname()}(name='{self.name}')"
+    
+    def webencoded_image(self) -> str:
+        """Convert image to be displayed on webpage."""
+        # convert images to bytes string
+        buffer = BytesIO()
+        self.image.save(buffer, format="png")
+        img_enc = b64encode(buffer.getvalue()).decode("utf-8")
+        return f"data:image/png;base64, {img_enc}"
+
+    def generate_random_prediction(self, force: bool = False) -> None:
+        """Generate random prediction values."""
+        if not force and self.prediction is not None:
+            logging.warning("set force=True to overwrite existing values.")
+        self.prediction = ModelOutputs.from_random()
+
+class NoDocumentFoundException(Exception):
+    pass

src/database/database.py

@@ -1,7 +1,9 @@
 from pymongo import MongoClient
 from dotenv import load_dotenv
+import logging
 import os
 
+
 def connect():
     """Connect to MongoDB."""
     # load env vars
@@ -18,4 +20,5 @@ def connect():
     db = client[os.getenv("MONGO_DB")]
     collection = db[os.getenv("MONGO_COLLECTION")]
     collection.create_index("name", unique=True)
+    logging.info("connected to database")
     return collection, db, client

src/database/utils.py

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

src/defaults.ini

@@ -1,6 +0,0 @@
-[main]
-logger_level=debug
-
-[discord]
-service_name=visual_critical_discourse_analysis
-logger_level=warning

src/main.py

@@ -1,99 +1,48 @@
 import logging
-from discord_logging.handler import DiscordHandler
 from dotenv import load_dotenv
-from configparser import ConfigParser
 from pathlib import Path
 import logging
 import os
 
-from src.web import server
+# prepare optional local setup
+env_path = Path(__file__).parent.parent / "local.env"
+load_dotenv(env_path)
 
-# load default values
+# ensure env vars set
-config = ConfigParser()
+necesasary_var_list = {
-config.read(Path(__file__).parent / 'defaults.ini')
+    "MONGO_HOST",
-LOGGER_LEVEL = config.get('main', 'logger_level')
+    "MONGO_DB",
-DISCORD_SERVICE_NAME = config.get('discord', 'service_name')
+    "MONGO_COLLECTION",
-DISCORD_LOGGER_LEVEL = config.get('discord', 'logger_level')
+    "MONGO_USER",
-
+    "MONGO_PASSWORD",
-
+}
-def setup_logging(
+for env_var in necesasary_var_list:
-    discord_webhook_url: str,
+    # ensure env var set
-    discord_service_name: str = DISCORD_SERVICE_NAME,
+    assert (
-    discord_logger_level: str = DISCORD_LOGGER_LEVEL,
+        env_var in os.environ
-    logger_level: str = LOGGER_LEVEL,
+    ), (
-) -> None:
+        f"environment variable not set: {env_var}"
-    # setup stream handler
-    fmt = (
-        '%(asctime)s | '
-        '%(levelname)s | '
-        '%(filename)s | '
-        '%(funcName)s | '
-        '%(message)s'
     )
-    datefmt = '%Y-%m-%d %H:%M:%S'
-    level = getattr(logging, logger_level.upper())
-    logging.basicConfig(format=fmt, datefmt=datefmt, level=level)
-    logger = logging.getLogger()
-    # add discord handler
-    discord_handler = DiscordHandler(
-        service_name=discord_service_name,
-        webhook_url=discord_webhook_url,
-    )
-    discord_handler.setFormatter(logging.Formatter('%(message)s'))
-    level = getattr(logging, discord_logger_level.upper())
-    discord_handler.setLevel(level=level)
-    logger.addHandler(discord_handler)
-    logging.debug('finished')
 
+# setup logging stream handler
+fmt = (
+    '%(asctime)s | '
+    '%(levelname)s | '
+    '%(filename)s | '
+    '%(funcName)s | '
+    '%(message)s'
+)
+datefmt = '%Y-%m-%d %H:%M:%S'
+logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO)
 
-def initialise_app() -> None:
+logging.info("initialized app")
-    """
-    Ensure all necessary environment variables are provided
-    """
-    # load env vars
-    load_dotenv()
-    # ensure env vars set
-    necesasary_var_map = {
-        'LOGGER_LEVEL': False,
-        'DISCORD_SERVICE_NAME': False,
-        'DISCORD_WEBHOOK_URL': True,
-        'DISCORD_LOGGER_LEVEL': False,
-        'DATABASE_HOST': True,
-        'DATABASE_PORT': True,
-        'DATABASE_ENGINE': True,
-        'DATABASE_DATABASE': True,
-        'DATABASE_USERNAME': True,
-        'DATABASE_PASSWORD': True,
-    }
-    for env_var, must_be_set in necesasary_var_map.items():
-        if must_be_set:
-            # ensure env var set
-            assert (
-                env_var in os.environ
-            ), (
-                f"environment variable not set: {env_var}"
-            )
-            # set variable from env var
-            locals()[env_var.lower()] = os.getenv(key=env_var)
-        else:
-            # ensure default value set
-            assert env_var in globals(), f"default variable not set: {env_var}"
-            # set variable from env var with backup from default value
-            locals()[env_var.lower()] = os.getenv(
-                key=env_var,
-                default=globals()[env_var]
-            )
-    setup_logging(
-        discord_webhook_url=locals()['discord_webhook_url'],
-        discord_service_name=locals()['discord_service_name'],
-        discord_logger_level=locals()['discord_logger_level'],
-        logger_level=locals()['logger_level'],
-    )
-    logging.debug('finished')
 
+from src.web import app
+server = app.server
 
 if __name__ == "__main__":
-    from src.web import app
+    # prepare local env vars
-    # initialise_app()
+    os.environ["MONGO_HOST"] = "localhost"
+    # run app
     app.run(debug=True)
     logging.info("started app")

src/model_experiential/__init__.py

@@ -1,4 +1,4 @@
-from .classes import ExperientialModelOutput
+from .classes import VisualSyntaxModelOutput
 
 
 

src/model_experiential/classes.py

@@ -1,8 +1,11 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from typing import List
 import random
 
 
+class OptionNotSetException(Exception):
+    pass
+
 class ModelOutput(BaseModel):
 
     @classmethod
@@ -21,6 +24,18 @@ class ModelOutput(BaseModel):
         kwargs = {field: random.random() for field in cls.list_fields()}
         return cls(**kwargs)
     
+    @classmethod
+    def from_choice(cls, option: str):
+        """Instantiate from choice."""
+        if option is None:
+            raise ValidationError()
+        assert isinstance(option, str), "option is not a string"
+        allowed_options_list = cls.list_fields()
+        assert option in allowed_options_list, f"{option} is not among allowed fields {allowed_options_list}"
+        kwargs = {field: 0 for field in cls.list_fields()}
+        kwargs[option] = 1
+        return cls(**kwargs)
+
     def __repr__(self) -> str:
         model_dict = self.model_dump()
         model_repr_str = f"{self.classname()}("
@@ -39,7 +54,7 @@ class ModelOutput(BaseModel):
         return max(model_dict.values())
 
 
-class ExperientialModelOutput(ModelOutput):
+class VisualSyntaxModelOutput(ModelOutput):
     non_transactional_action: float
     non_transactional_reaction: float
     unidirectional_transactional_action: float
@@ -61,7 +76,7 @@ class ExperientialModelOutput(ModelOutput):
 
 
 if __name__ == '__main__':
-    m = ExperientialModelOutput.from_random()
+    m = VisualSyntaxModelOutput.from_random()
     print(m)
     print(repr(m))
     print(m.highest_score_field())

src/model_interpersonal/classes.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from typing import List
 import random
 
@@ -21,6 +21,18 @@ class ModelOutput(BaseModel):
         kwargs = {field: random.random() for field in cls.list_fields()}
         return cls(**kwargs)
     
+    @classmethod
+    def from_choice(cls, option: str):
+        """Instantiate from choice."""
+        if option is None:
+            raise ValidationError()
+        assert isinstance(option, str)
+        allowed_options_list = cls.list_fields()
+        assert option in allowed_options_list, f"{option} is not among allowed fields {allowed_options_list}"
+        kwargs = {field: 0 for field in cls.list_fields()}
+        kwargs[option] = 1
+        return cls(**kwargs)
+
     def __repr__(self) -> str:
         model_dict = self.model_dump()
         model_repr_str = f"{self.classname()}("

src/model_textual/classes.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from typing import List
 import random
 
@@ -21,6 +21,18 @@ class ModelOutput(BaseModel):
         kwargs = {field: random.random() for field in cls.list_fields()}
         return cls(**kwargs)
     
+    @classmethod
+    def from_choice(cls, option: str):
+        """Instantiate from choice."""
+        if option is None:
+            raise ValidationError()
+        assert isinstance(option, str)
+        allowed_options_list = cls.list_fields()
+        assert option in allowed_options_list, f"{option} is not among allowed fields {allowed_options_list}"
+        kwargs = {field: 0 for field in cls.list_fields()}
+        kwargs[option] = 1
+        return cls(**kwargs)
+
     def __repr__(self) -> str:
         model_dict = self.model_dump()
         model_repr_str = f"{self.classname()}("

src/web/app.py

@@ -1,28 +1,141 @@
-from dash import Dash, html, Input, Output, State, page_container, page_registry
+from dash import Dash, Input, Output, State, ALL
 import dash_bootstrap_components as dbc
-import dash_mantine_components as dmc
+import logging
+from typing import List
+from pydantic import ValidationError
 
-from .header import generate_header
+from .layout import app_layout
-from .body import generate_body
+from src.database import (
-
+    connect,
-
+    get_visual_communication,
-app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
+    NoDocumentFoundException,
-server = app.server
+    upsert_annotations,
-
+    ModelOutputs
-app.layout = dmc.MantineProvider(
-    theme={
-        'fontFamily': '"Inter", sans-serif',
-        "components": {
-            "NavLink":{'styles':{'label':{'color':'#c2c7d0'}}}
-            },
-        },
-    children=[
-        dmc.Container(
-            [
-                generate_header(),
-                generate_body(),
-            ], fluid=True
-        ),
-    ],
 )
 
+# setup app
+app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
+app.title = "visual critical discourse analysis".title()
+app.layout = app_layout
+server = app.server
+
+# connect to database
+collection, db, client = connect()
+
+# define callbacks
+@app.callback(
+    Output("alert-element", "is_open"),
+    Output("alert-element", "children"),
+    Input("alert-message", "data")
+)
+def show_alert(
+    msg: str | None
+):
+    if msg is None or msg == "":
+        return False, ""
+    logging.info(f"updated alert message: {msg}")
+    return True, msg
+
+@app.callback(
+    Output("alert-message", "data"),
+    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_visual_communication_data(
+    n_clicks: int,
+    vis_com_name: str,
+    image_src: str,
+    annotation_keys: List,
+    annotation_values: List,
+):
+    logging.info("began cycling visual communication data")
+    global collection
+    # prepare default response
+    response = [
+        "",
+        vis_com_name,
+        image_src,
+        annotation_values
+    ]
+    # check if next-button clicked
+    if n_clicks  == 0:
+        logging.info("stopping early: next-button has not yet been clicked")
+        return 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
+            ]
+            annotations = {
+                key: value 
+                for key, value 
+                in zip(annotation_keys, annotation_values)
+            }
+            # instantiate ModelOutputs object
+            annotations = ModelOutputs.from_annotations(**annotations)
+            # save data to 
+            upsert_annotations(
+                collection=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=collection,
+            with_annotation=False
+        )
+        # set variables
+        vis_com_name = vis_com.name
+        image_src = vis_com.webencoded_image()
+        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 = f"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)

src/web/body.py

@@ -1,143 +0,0 @@
-import dash_mantine_components as dmc
-from dash import dcc, html
-from typing import List
-
-from src.model_experiential import ExperientialModelOutput
-from src.model_interpersonal import (
-    ContactModelOutput,
-    AngleModelOutput,
-    PointOfViewModelOutput,
-    DistanceModelOutput,
-    ModalityLightingModelOutput,
-    ModalityColorModelOutput,
-    ModalityDepthModelOutput
-)
-from src.model_textual import (
-    InformationValueModelOutput,
-    FramingModelOutput,
-    SalienceModelOutput
-)
-
-def generate_option_labels(model) -> List[str]:
-    """Generate presentable list of attributes from an OutputModel."""
-    labels = [
-        label.replace('_', ' ').title()
-        for label in model.list_fields()
-    ]
-    return labels
-
-def generate_experiential_options_map():
-    """Generate map of titles and options for experiential labels."""
-    options_map = {}
-    # add experiential labels
-    options_map["experiential".title()] = generate_option_labels(ExperientialModelOutput)
-    return options_map
-
-def generate_interpersonal_options_map():
-    """Generate map of titles and options for interpersonal labels."""
-    options_map = {}
-    # add interpersonal labels
-    options_map["contact".title()] = generate_option_labels(ContactModelOutput)
-    options_map["angle".title()] = generate_option_labels(AngleModelOutput)
-    options_map["point of view".title()] = generate_option_labels(PointOfViewModelOutput)
-    options_map["distance".title()] = generate_option_labels(DistanceModelOutput)
-    options_map["modality lighting".title()] = generate_option_labels(ModalityLightingModelOutput)
-    options_map["modality color".title()] = generate_option_labels(ModalityColorModelOutput)
-    options_map["modality depth".title()] = generate_option_labels(ModalityDepthModelOutput)
-    return options_map
-
-def generate_textual_options_map():
-    """Generate map of titles and options for textual labels."""
-    options_map = {}
-    # add textual labels
-    options_map["information value".title()] = generate_option_labels(InformationValueModelOutput)
-    options_map["framing".title()] = generate_option_labels(FramingModelOutput)
-    options_map["salience".title()] = generate_option_labels(SalienceModelOutput)
-    return options_map
-
-def generate_body():
-    image_container = dmc.Image(
-        width=600,
-        height=600,
-        withPlaceholder=True,
-        placeholder=[dmc.Loader(color="gray", size="md")],
-    )
-    # prepare experiential container
-    experiential_map = generate_experiential_options_map()
-    experiential_container = dmc.Col(
-        children=[
-            dmc.Container([
-                html.H4(list(experiential_map.keys())[0]),
-                html.B("visual syntax".title()),
-                dcc.RadioItems(options=list(experiential_map.values())[0]),
-            ])
-        ], span=4
-    )
-    # prepare interpersonal container
-    interpersonal_map = generate_interpersonal_options_map()
-    interpersonal_container = dmc.Col(
-        children=[
-            html.H4("interpersonal".title()),
-        ], span=4
-    )
-    for title, options in interpersonal_map.items():
-        interpersonal_container.children.append(
-            dmc.Container([
-                html.B(title),
-                dcc.RadioItems(options)
-            ])
-        )
-    # prepare textual container
-    textual_map = generate_textual_options_map()
-    textual_container = dmc.Col(
-        children=[
-            html.H4("textual".title()),
-        ], span=4
-    )
-    for title, options in textual_map.items():
-        textual_container.children.append(
-            dmc.Container([
-                html.B(title),
-                dcc.RadioItems(options)
-            ])
-        )
-    # prepare labels container
-    label_container = dmc.Grid(
-        children=[
-            experiential_container,
-            interpersonal_container,
-            textual_container,
-        ],
-    )
-    # build the full body container
-    body_container = dmc.Container(
-        dmc.Grid(
-            children=[
-                dmc.Col(
-                    dmc.Center(
-                    image_container,
-                    ),
-                    span=5,
-                ),
-                dmc.Col(
-                    # radio buttons part
-                    children = [
-                        label_container,
-                        dmc.Button(
-                            "confirm",
-                            id="submit-button",
-                            fullWidth=True,
-                            color="lime",
-                            radius="sm",
-                            size="md",
-                            style={
-                                "height": "50px"
-                            }
-                        ),
-                    ], span=7,
-                ),
-                # dmc.Col(span=1),
-            ], grow=True
-        ), fluid=True
-    )
-    return body_container

src/web/header.py

@@ -1,15 +0,0 @@
-import dash_mantine_components as dmc
-from dash import html
-
-def generate_header():
-    header = dmc.Header(
-        height=80,
-        children=[
-            dmc.Container(
-                children=[
-                    html.H2(children="Visual Critical Discourse Analysis Tool", style={"margin-left": "20px", "padding-top": "-5px"}),
-                ], size="xl", px="xl",
-            ),
-        ]
-    )
-    return header

src/web/layout/__init__.py

@@ -0,0 +1 @@
+from .layout import app_layout

src/web/layout/alerts.py

@@ -0,0 +1,15 @@
+from dash import html, dcc
+import dash_bootstrap_components as dbc
+
+
+alerts_element = html.Div(
+    children=[
+        dbc.Alert(
+            children="",
+            id="alert-element",
+            dismissable=True,
+            fade=False,
+            is_open=False,
+        )
+    ]
+)

src/web/layout/body.py

@@ -0,0 +1,20 @@
+import dash_mantine_components as dmc
+from dash import dcc, html
+from typing import List
+
+from .image import image_element
+from .inputs import inputs_element
+
+
+body_element = dmc.Container(
+    fluid=True,
+    children=[
+        dmc.Grid(
+            grow=True,
+            children=[
+                dmc.Col([image_element], span=5),
+                dmc.Col([inputs_element], span=7)
+            ],
+        )
+    ],
+)

src/web/layout/header.py

@@ -0,0 +1,21 @@
+import dash_mantine_components as dmc
+from dash import html
+
+
+header_element = dmc.Header(
+    height=80,
+    children=[
+        dmc.Container(
+            children=[
+                html.H2(
+                    children="Visual Critical Discourse Analysis Tool",
+                    style={
+                        "margin-left": "20px",
+                        "padding-top": "-5px"
+                    },
+                ),
+            ],
+            size="xl", px="xl",
+        ),
+    ]
+)

src/web/layout/image.py

@@ -0,0 +1,21 @@
+import dash_mantine_components as dmc
+from dash import html
+from pathlib import Path
+from base64 import b64encode
+
+# read init img
+init_img_path = Path(__file__).parent / "init_img.png"
+with open(init_img_path.absolute(), "rb") as fh:
+    init_img_enc = b64encode(fh.read()).decode("utf-8")
+# generate init img string
+init_img_src = f"data:image/png;base64, {init_img_enc}"
+
+image_element = dmc.Center(
+    html.Img(
+        style={
+            "width": "100%",
+        },
+        id="image-container",
+        src=init_img_src
+    )
+)

src/web/layout/inputs.py

@@ -0,0 +1,23 @@
+import dash_mantine_components as dmc
+
+from .labels import labels_element
+
+next_button = dmc.Button(
+    "next".title(),
+    id="next-button",
+    n_clicks=0,
+    fullWidth=True,
+    color="lime",
+    radius="sm",
+    size="md",
+    style={
+        "height": "50px"
+    }
+)
+
+inputs_element = dmc.SimpleGrid(
+    children=[
+        labels_element,
+        next_button
+    ]
+)

src/web/layout/labels.py

@@ -0,0 +1,121 @@
+from dash import html, dcc
+import dash_mantine_components as dmc
+from typing import List
+
+from src.model_experiential import (
+    VisualSyntaxModelOutput
+)
+from src.model_interpersonal import (
+    ContactModelOutput,
+    AngleModelOutput,
+    PointOfViewModelOutput,
+    DistanceModelOutput,
+    ModalityLightingModelOutput,
+    ModalityColorModelOutput,
+    ModalityDepthModelOutput
+)
+from src.model_textual import (
+    InformationValueModelOutput,
+    FramingModelOutput,
+    SalienceModelOutput
+)
+
+def generate_option_labels(model) -> List[str]:
+    """Generate presentable list of attributes from an OutputModel."""
+    labels = [
+        label.replace('_', ' ').title()
+        for label in model.list_fields()
+    ]
+    return labels
+
+def generate_visual_syntax_options_map():
+    """Generate map of titles and options for visual syntax labels."""
+    options_map = {}
+    # add experiential labels
+    options_map["visual syntax"] = generate_option_labels(VisualSyntaxModelOutput)
+    return options_map
+
+def generate_interpersonal_options_map():
+    """Generate map of titles and options for interpersonal labels."""
+    options_map = {}
+    # add interpersonal labels
+    options_map["contact"] = generate_option_labels(ContactModelOutput)
+    options_map["angle"] = generate_option_labels(AngleModelOutput)
+    options_map["point of view"] = generate_option_labels(PointOfViewModelOutput)
+    options_map["distance"] = generate_option_labels(DistanceModelOutput)
+    options_map["modality lighting"] = generate_option_labels(ModalityLightingModelOutput)
+    options_map["modality color"] = generate_option_labels(ModalityColorModelOutput)
+    options_map["modality depth"] = generate_option_labels(ModalityDepthModelOutput)
+    return options_map
+
+def generate_textual_options_map():
+    """Generate map of titles and options for textual labels."""
+    options_map = {}
+    # add textual labels
+    options_map["information value"] = generate_option_labels(InformationValueModelOutput)
+    options_map["framing"] = generate_option_labels(FramingModelOutput)
+    options_map["salience"] = generate_option_labels(SalienceModelOutput)
+    return options_map
+
+# prepare experiential container
+experiential_map = generate_visual_syntax_options_map()
+experiential_container = dmc.Col(
+    children=[
+        html.H4("experiential".title()),
+    ], span=5
+)
+for title, options in experiential_map.items():
+    id_dict = {"type": "annotation", "index": title.replace('_', '-')}
+    experiential_container.children.append(
+        dmc.Container([
+            html.B(title.title()),
+            dcc.RadioItems(
+                options=options,
+                id=id_dict,
+            ),
+        ])
+    )
+# prepare interpersonal container
+interpersonal_map = generate_interpersonal_options_map()
+interpersonal_container = dmc.Col(
+    children=[
+        html.H4("interpersonal".title()),
+    ], span=3
+)
+for title, options in interpersonal_map.items():
+    id_dict = {"type": "annotation", "index": title.replace('_', '-')}
+    interpersonal_container.children.append(
+        dmc.Container([
+            html.B(title.title()),
+            dcc.RadioItems(
+                options=options,
+                id=id_dict,
+            ),
+        ])
+    )
+# prepare textual container
+textual_map = generate_textual_options_map()
+textual_container = dmc.Col(
+    children=[
+        html.H4("textual".title()),
+    ], span=4
+)
+for title, options in textual_map.items():
+    id_dict = {"type": "annotation", "index": title.replace('_', '-')}
+    textual_container.children.append(
+        dmc.Container([
+            html.B(title),
+            dcc.RadioItems(
+                options=options,
+                id=id_dict,
+            ),
+        ])
+    )
+
+labels_element = dmc.Grid(
+    children=[
+        experiential_container,
+        interpersonal_container,
+        textual_container,
+    ]
+)

src/web/layout/layout.py

@@ -0,0 +1,35 @@
+from dash import dcc
+import dash_mantine_components as dmc
+
+from .stores import stores_element
+from .alerts import alerts_element
+from .header import header_element
+from .body import body_element
+
+
+app_layout = dmc.MantineProvider(
+    theme={
+        "fontFamily": '"Inter", sans-serif',
+        "components": {
+            "NavLink": {
+                "styles": {
+                    "label": {
+                        "color": "#c2c7d0"
+                    }
+                }
+            }
+        },
+    },
+    children=[
+        stores_element,
+        alerts_element,
+        dmc.Container(
+            children=[
+                header_element,
+                body_element,
+            ],
+            fluid=True
+        ),
+
+    ]
+)

src/web/layout/stores.py

@@ -0,0 +1,16 @@
+from dash import html, dcc
+import logging
+import os
+
+storage_type = "session"
+if "ENV" in os.environ and os.getenv("ENV") == "DEV":
+    storage_type = "memory"
+    logging.info(f"ENV=DEV -> dcc.Stores changed to storage_type={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=""),
+    ]
+)

tests/test_generate_random_prediction.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+
+from src.database import VisualCommunication
+
+
+if __name__ == "__main__":
+    # get list of image paths
+    test_dir = Path(__file__).parent
+    img_dir = test_dir / "imgs"
+    img_path_list = [path for path in img_dir.glob("*.jpeg") if path.is_file()]
+    print(img_path_list)
+    # instantiate data object
+    vis_com_list = [VisualCommunication.from_file(path) for path in img_path_list]
+    # generate random predictions
+    [vis_com.generate_random_prediction() for vis_com in vis_com_list]
+    for vis_com in vis_com_list:
+        print(vis_com)

tests/test_get_visual_communication.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+from dotenv import load_dotenv
+import os
+
+from src.database import (
+    connect,
+    get_visual_communication
+)
+
+if __name__ == "__main__":
+    # prepare env vars
+    env_path = Path(__file__).parent.parent / "local.env"
+    assert env_path.exists()
+    load_dotenv(env_path)
+    os.environ["MONGO_HOST"] = "localhost"
+    # connect to database
+    collection, db, client = connect()
+    print(client.server_info())
+    # get visual communication
+    vis_com = get_visual_communication(collection)
+    print(vis_com)

tests/test_image_download.py

@@ -1,7 +1,5 @@
 from pathlib import Path
 from dotenv import load_dotenv
-from pymongo import MongoClient
-from typing import List
 import os
 
 from src.database import VisualCommunication, connect
@@ -25,3 +23,4 @@ if __name__ == "__main__":
         print(repr(vis_com))
     if data is not None:
         print(vis_com.image)
+        print(vis_com.model_dump())

tests/test_model_outputs_from_annotation.py

@@ -0,0 +1,13 @@
+from src.database import ModelOutputs
+
+
+if __name__ == "__main__":
+    # instantiate data object
+    annotation = {
+        
+    }
+    vis_com_list = [ModelOutputs.from_annotation(path) for path in img_path_list]
+    # generate random predictions
+    [vis_com.generate_random_prediction() for vis_com in vis_com_list]
+    for vis_com in vis_com_list:
+        print(vis_com)

tests/test_prediction_upload.py

@@ -0,0 +1,44 @@
+from pathlib import Path
+from dotenv import load_dotenv
+import os
+import logging
+
+from src.database import (
+    VisualCommunication,
+    connect,
+    upsert_predictions
+)
+
+if __name__ == "__main__":
+    # setup logging
+    fmt = (
+        '%(asctime)s | '
+        '%(levelname)s | '
+        '%(filename)s | '
+        '%(funcName)s | '
+        '%(message)s'
+    )
+    datefmt = '%Y-%m-%d %H:%M:%S'
+    logging.basicConfig(format=fmt, datefmt=datefmt, level=logging.INFO)
+    # get list of image paths
+    test_dir = Path(__file__).parent
+    img_dir = test_dir / "imgs"
+    img_path_list = [path for path in img_dir.glob("*.jpeg") if path.is_file()]
+    # instantiate data object
+    vis_com_list = [VisualCommunication.from_file(path) for path in img_path_list]
+    # generate random predictions
+    [vis_com.generate_random_prediction() for vis_com in vis_com_list]
+    # prepare env vars
+    env_path = test_dir.parent / "local.env"
+    assert env_path.exists()
+    load_dotenv(env_path)
+    os.environ["MONGO_HOST"] = "localhost"
+    # connect to database
+    collection, db, client = connect()
+    # upload visual communication
+    for vis_com in vis_com_list:
+        upsert_predictions(
+            collection=collection,
+            vis_com_name=vis_com.name,
+            predictions=vis_com.prediction,
+        )

tests/test_total_annotated.py

@@ -0,0 +1,20 @@
+from pathlib import Path
+from dotenv import load_dotenv
+import os
+
+from src.database import (
+    connect,
+    total_annotated
+)
+
+if __name__ == "__main__":
+    # prepare env vars
+    env_path = Path(__file__).parent.parent / "local.env"
+    assert env_path.exists()
+    load_dotenv(env_path)
+    os.environ["MONGO_HOST"] = "localhost"
+    # connect to database
+    collection, db, client = connect()
+    # get visual communication
+    num_docs = total_annotated(collection)
+    print(f"number of annotated  documents in database: {num_docs}")

tests/test_total_documents.py

@@ -0,0 +1,20 @@
+from pathlib import Path
+from dotenv import load_dotenv
+import os
+
+from src.database import (
+    connect,
+    total_documents
+)
+
+if __name__ == "__main__":
+    # prepare env vars
+    env_path = Path(__file__).parent.parent / "local.env"
+    assert env_path.exists()
+    load_dotenv(env_path)
+    os.environ["MONGO_HOST"] = "localhost"
+    # connect to database
+    collection, db, client = connect()
+    # get visual communication
+    num_docs = total_documents(collection)
+    print(f"total number of documents in database: {num_docs}")