applied new functions and fixed bugs

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:

src/database/__init__.py

@@ -7,5 +7,7 @@ from .database import connect
 from .utils import (
     total_documents,
     total_annotated,
-    get_visual_communication
+    get_visual_communication,
+    upsert_annotations,
+    upsert_predictions,
 )

src/database/classes.py

@@ -4,8 +4,12 @@ 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,
@@ -22,7 +26,7 @@ from src.model_textual import (
 )
 
 class ModelOutputs(BaseModel):
-    experiential: ExperientialModelOutput
+    visual_syntax: VisualSyntaxModelOutput
     contact: ContactModelOutput
     angle: AngleModelOutput
     point_of_view: PointOfViewModelOutput
@@ -34,6 +38,52 @@ 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
@@ -81,8 +131,13 @@ class VisualCommunication(BaseModel):
         buffer = BytesIO()
         self.image.save(buffer, format="png")
         img_enc = b64encode(buffer.getvalue()).decode("utf-8")
-        return img_enc
+        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/utils.py

@@ -1,8 +1,10 @@
 from pymongo.collection import Collection
+import logging
 
 from .classes import (
     VisualCommunication,
-    NoDocumentFoundException
+    NoDocumentFoundException,
+    ModelOutputs
 )
 
 
@@ -25,8 +27,7 @@ def total_annotated(
 
 def get_visual_communication(
         collection: Collection,
-        with_annotation: bool = False,
+        with_annotation: bool = False
-        with_prediction: bool = False
     ) -> VisualCommunication:
     """Get a random visual communication from the database."""
     query = {}
@@ -34,15 +35,56 @@ def get_visual_communication(
         query["annotation"] = {"$ne": None}
     else:
         query["annotation"] = None
-    if with_prediction:
-        query["prediction"] = {"$ne": None}
-    else:
-        query["prediction"] = 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:
-        raise NoDocumentFoundException
+        logging.error("failed getting visual communication")
-    return VisualCommunication.model_validate(data[0])
+        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/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,12 +1,16 @@
-from dash import Dash, Input, Output
+from dash import Dash, Input, Output, State, ALL
 import dash_bootstrap_components as dbc
 import logging
+from typing import List
+from pydantic import ValidationError
 
 from .layout import app_layout
 from src.database import (
     connect,
     get_visual_communication,
-    NoDocumentFoundException
+    NoDocumentFoundException,
+    upsert_annotations,
+    ModelOutputs
 )
 
 # setup app
@@ -27,39 +31,111 @@ collection, db, client = connect()
 def show_alert(
     msg: str | None
 ):
-    if msg is 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("visual-communication-name", "data"),
+    Output("vis-com-name", "data"),
     Output("image-container", "src"),
+    Output({"type": "annotation", "index": ALL}, "value"),
     Input("next-button", "n_clicks"),
-    prevent_initial_call=True
+    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 load_unannotated_visual_communication_data(
+def cycle_visual_communication_data(
-    n_clicks: int
+    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
         )
-    except NoDocumentFoundException:
+        # set variables
-        return (
-            "Did not find any unannotated data in database",
-            None,
-            ""
-        )
-    # prepare return values
-    alert_message = None
         vis_com_name = vis_com.name
-    img_src = f"data:image/png;base64, {vis_com.webencoded_image()}"
+        image_src = vis_com.webencoded_image()
-    logging.info("updated visual communication")
+        if vis_com.prediction is not None:
-    return (
+            # TODO: update to use optional predictions
-        alert_message,
+            pass
-        vis_com_name,
+        else:
-        img_src
+            # 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/layout/image.py

@@ -7,6 +7,8 @@ from base64 import b64encode
 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(
@@ -14,6 +16,6 @@ image_element = dmc.Center(
             "width": "100%",
         },
         id="image-container",
-        src=f"data:image/png;base64, {init_img_enc}"
+        src=init_img_src
     )
 )

src/web/layout/inputs.py

@@ -5,6 +5,7 @@ from .labels import labels_element
 next_button = dmc.Button(
     "next".title(),
     id="next-button",
+    n_clicks=0,
     fullWidth=True,
     color="lime",
     radius="sm",

src/web/layout/labels.py

@@ -2,7 +2,9 @@ from dash import html, dcc
 import dash_mantine_components as dmc
 from typing import List
 
-from src.model_experiential import ExperientialModelOutput
+from src.model_experiential import (
+    VisualSyntaxModelOutput
+)
 from src.model_interpersonal import (
     ContactModelOutput,
     AngleModelOutput,
@@ -26,46 +28,53 @@ def generate_option_labels(model) -> List[str]:
     ]
     return labels
 
-def generate_experiential_options_map():
+def generate_visual_syntax_options_map():
-    """Generate map of titles and options for experiential labels."""
+    """Generate map of titles and options for visual syntax labels."""
     options_map = {}
     # add experiential labels
-    options_map["experiential".title()] = generate_option_labels(ExperientialModelOutput)
+    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".title()] = generate_option_labels(ContactModelOutput)
+    options_map["contact"] = generate_option_labels(ContactModelOutput)
-    options_map["angle".title()] = generate_option_labels(AngleModelOutput)
+    options_map["angle"] = generate_option_labels(AngleModelOutput)
-    options_map["point of view".title()] = generate_option_labels(PointOfViewModelOutput)
+    options_map["point of view"] = generate_option_labels(PointOfViewModelOutput)
-    options_map["distance".title()] = generate_option_labels(DistanceModelOutput)
+    options_map["distance"] = generate_option_labels(DistanceModelOutput)
-    options_map["modality lighting".title()] = generate_option_labels(ModalityLightingModelOutput)
+    options_map["modality lighting"] = generate_option_labels(ModalityLightingModelOutput)
-    options_map["modality color".title()] = generate_option_labels(ModalityColorModelOutput)
+    options_map["modality color"] = generate_option_labels(ModalityColorModelOutput)
-    options_map["modality depth".title()] = generate_option_labels(ModalityDepthModelOutput)
+    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".title()] = generate_option_labels(InformationValueModelOutput)
+    options_map["information value"] = generate_option_labels(InformationValueModelOutput)
-    options_map["framing".title()] = generate_option_labels(FramingModelOutput)
+    options_map["framing"] = generate_option_labels(FramingModelOutput)
-    options_map["salience".title()] = generate_option_labels(SalienceModelOutput)
+    options_map["salience"] = generate_option_labels(SalienceModelOutput)
     return options_map
 
 # prepare experiential container
-experiential_map = generate_experiential_options_map()
+experiential_map = generate_visual_syntax_options_map()
 experiential_container = dmc.Col(
     children=[
-        dmc.Container([
+        html.H4("experiential".title()),
-            html.H4(list(experiential_map.keys())[0]),
-            html.B("visual syntax".title()),
-            dcc.RadioItems(options=list(experiential_map.values())[0]),
-        ])
     ], 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(
@@ -74,10 +83,14 @@ interpersonal_container = dmc.Col(
     ], 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),
+            html.B(title.title()),
-            dcc.RadioItems(options)
+            dcc.RadioItems(
+                options=options,
+                id=id_dict,
+            ),
         ])
     )
 # prepare textual container
@@ -88,10 +101,14 @@ textual_container = dmc.Col(
     ], 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)
+            dcc.RadioItems(
+                options=options,
+                id=id_dict,
+            ),
         ])
     )
 

src/web/layout/stores.py

@@ -1,9 +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="session"),
+        dcc.Store(id="alert-message", storage_type=storage_type, data=""),
-        dcc.Store(id="visual-communication-name", storage_type="session"),
+        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

@@ -18,4 +18,4 @@ if __name__ == "__main__":
     print(client.server_info())
     # get visual communication
     vis_com = get_visual_communication(collection)
-    print(vis_com.image)
+    print(vis_com)

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