added copying in core code

Dockerfile

@@ -57,8 +57,9 @@ RUN mkdir -p /home/app && \
 # add code while changing ownership
 WORKDIR $APP_HOME
 COPY --chown=app:app ./src ./src
+COPY --chown=app:app ./core ./core
 
 # change to the app user
 USER app
 
-ENTRYPOINT [ "gunicorn", "src.main:server", "-b", "0.0.0.0:8050" ]
+ENTRYPOINT [ "gunicorn", "src.main:server", "-b", "0.0.0.0:8050" ]

core/database/__init__.py

@@ -1,13 +1,13 @@
+"""Database module content."""
 from __future__ import annotations
 
-from .classes import ModelOutputs
+from .classes import Dataset
 from .classes import NoDocumentFoundException
 from .classes import VisualCommunication
-from .database import connect
-from .utils import get_visual_communication
-from .utils import list_names
-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
+from .utils import connect
+from .utils.count_documents import count_documents
+from .utils.get_visual_communication import get_visual_communication
+from .utils.list_names import list_names
+from .utils.upsert_annotation import upsert_annotation
+from .utils.upsert_prediction import upsert_prediction
+from .utils.upsert_visual_communication import upsert_visual_communication

core/database/classes.py

@@ -1,165 +0,0 @@
-from __future__ import annotations
-
-import logging
-from base64 import b64decode
-from base64 import b64encode
-from io import BytesIO
-from pathlib import Path
-
-from PIL import Image
-from pydantic import BaseModel
-from pydantic import field_serializer
-from pydantic import field_validator
-
-from src.model_experiential import (
-    VisualSyntaxModelOutput,
-)
-from src.model_interpersonal import AngleModelOutput
-from src.model_interpersonal import ContactModelOutput
-from src.model_interpersonal import DistanceModelOutput
-from src.model_interpersonal import ModalityColorModelOutput
-from src.model_interpersonal import ModalityDepthModelOutput
-from src.model_interpersonal import ModalityLightingModelOutput
-from src.model_interpersonal import PointOfViewModelOutput
-from src.model_textual import FramingModelOutput
-from src.model_textual import InformationValueModelOutput
-from src.model_textual import SalienceModelOutput
-
-
-class NoDocumentFoundException(Exception):
-    pass
-
-
-class ModelOutputs(BaseModel):
-    visual_syntax: VisualSyntaxModelOutput
-    contact: ContactModelOutput
-    angle: AngleModelOutput
-    point_of_view: PointOfViewModelOutput
-    distance: DistanceModelOutput
-    modality_lighting: ModalityLightingModelOutput
-    modality_color: ModalityColorModelOutput
-    modality_depth: ModalityDepthModelOutput
-    information_value: InformationValueModelOutput
-    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()  # type: ignore
-            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
-    annotation: ModelOutputs | None = None
-    prediction: ModelOutputs | None = None
-
-    class Config:
-        arbitrary_types_allowed = True
-
-    @classmethod
-    def classname(cls) -> str:
-        """Return classname."""
-        return cls.__name__
-
-    @classmethod
-    def from_file(cls, path: Path) -> VisualCommunication:
-        """Instantiate from file."""
-        name = path.stem
-        image = Image.open(path)
-        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()
-        image.save(buffer, format='JPEG')
-        return buffer.getvalue()
-
-    @field_validator('image', mode='before')
-    @classmethod
-    def convert_to_image(
-        cls,
-        image: Image.Image | BytesIO | bytes,
-    ) -> Image.Image:
-        if isinstance(image, bytes):
-            image = BytesIO(image)
-        if isinstance(image, BytesIO):
-            image = Image.open(image)
-        return image
-
-    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()

core/database/classes/__init__.py

@@ -0,0 +1,6 @@
+"""Database classes module content."""
+from __future__ import annotations
+
+from .dataset import Dataset
+from .exceptions import NoDocumentFoundException
+from .visual_communication import VisualCommunication

core/database/classes/dataset.py

@@ -0,0 +1,71 @@
+"""Definition of database Dataset class."""
+from __future__ import annotations
+
+import logging
+import random
+from datetime import datetime
+from datetime import UTC
+
+from pydantic import BaseModel
+from pydantic import Field
+from pymongo.collection import Collection
+
+
+class Dataset(BaseModel):
+    """Database Dataset model."""
+    create_time: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    train_names: list[str]
+    test_names: list[str]
+    validation_names: list[str]
+
+    @classmethod
+    def fraction_map(cls) -> dict[str, float]:
+        """Dict with train, test and validation fractions."""
+        # define map
+        split_map = {
+            'train': 0.7,
+            'test': 0.2,
+            'validation': 0.1,
+        }
+        # sanity check
+        assert sum(split_map.values()) == 1.0
+        return split_map
+
+    @classmethod
+    def new_from_name_list(
+        cls,
+        name_list: list[str],
+    ) -> Dataset:
+        """Generate new dataset from list of filenames."""
+        # calculate split fractions
+        fraction_map = cls.fraction_map()
+        num_total = len(name_list)
+        num_validation = round(num_total * fraction_map['validation'])
+        num_test = round(num_total * fraction_map['test'])
+        # split data
+        validation_name_list = random.choices(name_list, k=num_validation)
+        name_list = [
+            name for name in name_list if name not in validation_name_list
+        ]
+        test_name_list = random.choices(name_list, k=num_test)
+        train_name_list = [
+            name for name in name_list if name not in test_name_list
+        ]
+        # instantiate object
+        dataset = Dataset(
+            train_names=train_name_list,
+            test_names=test_name_list,
+            validation_names=validation_name_list,
+        )
+        logging.debug('finished')
+        return dataset
+
+    def save(
+        self,
+        collection: Collection,
+    ) -> None:
+        """Save dataset to database."""
+        res = collection.insert_one(
+            document=self.model_dump(),
+        )
+        logging.debug('inserted document: %s', res)

core/database/classes/exceptions.py

@@ -0,0 +1,6 @@
+"""Definition of database exception."""
+from __future__ import annotations
+
+
+class NoDocumentFoundException(Exception):
+    """Database exception for when no documents are found."""

core/database/classes/visual_communication.py

@@ -0,0 +1,85 @@
+"""Definition of VisualCommunication model."""
+from __future__ import annotations
+
+import logging
+from base64 import b64decode
+from base64 import b64encode
+from io import BytesIO
+from pathlib import Path
+
+from PIL import Image
+from pydantic import BaseModel
+from pydantic import field_serializer
+from pydantic import field_validator
+
+from core.dto import ModelData
+
+
+class VisualCommunication(BaseModel):
+    """Visual communication model."""
+
+    name: str
+    image: Image.Image
+    annotation: ModelData | None = None
+    prediction: ModelData | None = None
+
+    class Config:
+        """BaseModel configuration."""
+        arbitrary_types_allowed = True
+
+    @classmethod
+    def classname(cls) -> str:
+        """Return classname."""
+        return cls.__name__
+
+    @classmethod
+    def from_file(cls, path: Path) -> VisualCommunication:
+        """Instantiate from file."""
+        name = path.stem
+        image = Image.open(path)
+        image.load()
+        return VisualCommunication(name=name, image=image)
+
+    @classmethod
+    def decode_image(cls, content: str) -> Image.Image:
+        """Extract image from webencoded content."""
+        _, content_data = content.split(',')
+        return Image.open(BytesIO(b64decode(content_data)))
+
+    @field_serializer('image')
+    @classmethod
+    def serialize_image(cls, image: Image.Image) -> bytes:   # type: ignore
+        """Convert image to bytes for storage in database."""
+        buffer = BytesIO()
+        image.save(buffer, format='JPEG')
+        return buffer.getvalue()
+
+    @field_validator('image', mode='before')
+    @classmethod
+    def convert_to_image(
+        cls,
+        image: Image.Image | BytesIO | bytes,
+    ) -> Image.Image:
+        """Convert bytes input from database into image."""
+        if isinstance(image, bytes):
+            image = BytesIO(image)
+        if isinstance(image, BytesIO):
+            image = Image.open(image)
+        return image
+
+    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 = ModelData.from_random()

core/database/utils.py

@@ -1,158 +0,0 @@
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from .classes import ModelOutputs
-from .classes import NoDocumentFoundException
-from .classes import VisualCommunication
-
-
-def count_documents(
-    collection: Collection,
-    has_annotation: bool = False,
-) -> int:
-    """
-    Get the total number of documents
-    in database that matches the filters.
-    """
-    assert isinstance(collection, Collection)
-    assert isinstance(has_annotation, bool)
-    # build query
-    query = {}
-    if has_annotation:
-        query['annotation'] = {'$ne': None}
-    return collection.count_documents(filter=query)
-
-
-def list_names(
-    collection: Collection,
-    has_annotation: bool = True,
-) -> list[str]:
-    """List the names of entries that match the filters."""
-    assert isinstance(collection, Collection)
-    assert isinstance(has_annotation, bool)
-    # build query
-    query = {}
-    if has_annotation:
-        query['annotation'] = {'$ne': None}
-    res_list = collection.find(
-        filter=query,
-        projection={
-            '_id': False,
-            'name': True,
-        },
-    )
-    return list(res_list)
-
-
-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'] = {'$eq': None}
-    data = collection.aggregate([
-        {
-            '$match': query,     # find using filters
-        },
-        {
-            '$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')
-        raise NoDocumentFoundException()
-    logging.info('finished')
-    return VisualCommunication.model_validate(data_list[0])
-
-
-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')
-
-
-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

core/database/utils/__init__.py

@@ -0,0 +1,4 @@
+"""Database utils module content."""
+from __future__ import annotations
+
+from .connect import connect

core/database/utils/connect.py

@@ -1,3 +1,7 @@
+"""
+Definition of function to connect to database
+using environment variables.
+"""
 from __future__ import annotations
 
 import logging
@@ -8,7 +12,7 @@ from pymongo import MongoClient
 
 
 def connect():
-    """Connect to MongoDB."""
+    """Connect to MongoDB using env vars."""
     # load env vars
     load_dotenv()
     necessary_env_vars = [
@@ -21,7 +25,9 @@ def connect():
     # connect to database
     client = MongoClient(os.getenv('MONGO_HOST'))
     db = client[os.getenv('MONGO_DB')]
+    # extract collection
     collection = db[os.getenv('MONGO_COLLECTION')]
+    # set unique index on "name"
     collection.create_index('name', unique=True)
-    logging.info('connected to database')
+    logging.debug('finished')
     return collection, db, client

core/database/utils/count_documents.py

@@ -0,0 +1,21 @@
+"""Definition of function to count documents in database."""
+from __future__ import annotations
+
+from pymongo.collection import Collection
+
+
+def count_documents(
+    collection: Collection,
+    only_with_annotation: bool = False,
+) -> int:
+    """
+    Get the total number of documents
+    in database that matches the filters.
+    """
+    assert isinstance(collection, Collection)
+    assert isinstance(only_with_annotation, bool)
+    # build query
+    query = {}
+    if only_with_annotation:
+        query['annotation'] = {'$ne': None}
+    return collection.count_documents(filter=query)

core/database/utils/get_visual_communication.py

@@ -0,0 +1,39 @@
+"""Definition of function to get visual communication from database."""
+from __future__ import annotations
+
+import logging
+
+from pymongo.collection import Collection
+
+from core.database import NoDocumentFoundException
+from core.database import VisualCommunication
+
+
+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'] = {'$eq': None}
+    data = collection.aggregate(
+        pipeline=[
+            {
+                '$match': query,     # find using filters
+            },
+            {
+                '$sample': {
+                    'size': 1,   # get one random
+                },
+            },
+        ],
+    )
+    data_list = list(data)   # read data from cursor object
+    if len(data_list) == 0:
+        raise NoDocumentFoundException()
+    vis_com = VisualCommunication.model_validate(data_list[0])
+    logging.debug('finished')
+    return vis_com

core/database/utils/list_names.py

@@ -0,0 +1,31 @@
+"""
+Definition of function to list names
+of all visual communication documents in database.
+"""
+from __future__ import annotations
+
+from pymongo.collection import Collection
+
+
+def list_names(
+    collection: Collection,
+    only_with_annotation: bool = True,
+) -> list[str]:
+    """List the names of entries that match the filters."""
+    assert isinstance(collection, Collection)
+    assert isinstance(only_with_annotation, bool)
+    # build query
+    query = {}
+    if only_with_annotation:
+        query['annotation'] = {'$ne': None}
+    # execute query
+    res_list = collection.find(
+        filter=query,
+        projection={
+            '_id': False,
+            'name': True,
+        },
+    )
+    # extract information
+    name_list = [elem['name'] for elem in res_list]
+    return name_list

core/database/utils/upsert_annotation.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import logging
+
+from pymongo.collection import Collection
+
+from core.dto import ModelData
+
+
+def upsert_annotation(
+    collection: Collection,
+    vis_com_name: str,
+    annotations: ModelData,
+) -> 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')

core/database/utils/upsert_prediction.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import logging
+
+from pymongo.collection import Collection
+
+from core.dto import ModelData
+
+
+def upsert_prediction(
+    collection: Collection,
+    vis_com_name: str,
+    predictions: ModelData,
+) -> 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')

core/database/utils/upsert_visual_communication.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from pymongo.collection import Collection
+
+from core.database import VisualCommunication
+
+
+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

core/dto/__init__.py

@@ -1,3 +1,4 @@
+"""Data transfer objects module content."""
 from __future__ import annotations
 
 from .angle import AngleData
@@ -8,6 +9,7 @@ from .information_value import InformationValueData
 from .modality_color import ModalityColorData
 from .modality_depth import ModalityDepthData
 from .modality_lighting import ModalityLightingData
+from .model_data import ModelData
 from .point_of_view import PointOfViewData
 from .salience import SalienceData
 from .visual_syntax import VisualSyntaxData

core/dto/angle.py

@@ -1,9 +1,11 @@
+"""Definition of Angle data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class AngleData(DataModel):
+    """Angle data model."""
     high: float
     eye_level: float
     low: float

core/dto/contact.py

@@ -1,8 +1,11 @@
+"""Definition of ContactData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class ContactData(DataModel):
+    """ContactData data model."""
+
     offer: float
     demand: float

core/dto/data_model.py

@@ -1,3 +1,4 @@
+"""Definition of DataModel base class."""
 from __future__ import annotations
 
 import random
@@ -7,6 +8,7 @@ from pydantic import ValidationError
 
 
 class DataModel(BaseModel):
+    """DataModel base class."""
 
     @classmethod
     def classname(cls) -> str:

core/dto/distance.py

@@ -1,9 +1,12 @@
+"""Definition of DistanceData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class DistanceData(DataModel):
+    """DistanceData data model."""
+
     long: float
     medium: float
     close: float

core/dto/framing.py

@@ -1,9 +1,12 @@
+"""Definition of FramingData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class FramingData(DataModel):
+    """FramingData data model."""
+
     frame_lines: float
     empty_space: float
     colour_contrast: float

core/dto/information_value.py

@@ -1,9 +1,12 @@
+"""Definition of InformationValueData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class InformationValueData(DataModel):
+    """InformationValueData data model."""
+
     given_new: float
     ideal_real: float
     central_marginal: float

core/dto/modality_color.py

@@ -1,9 +1,12 @@
+"""Definition of ModalityColorData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class ModalityColorData(DataModel):
+    """ModalityColorData data model."""
+
     high: float
     medium: float
     low: float

core/dto/modality_depth.py

@@ -1,9 +1,12 @@
+"""Definition of ModalityDepthData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class ModalityDepthData(DataModel):
+    """ModalityDepthData data model."""
+
     high: float
     medium: float
     low: float

core/dto/modality_lighting.py

@@ -1,9 +1,12 @@
+"""Definition of ModalityLightingData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class ModalityLightingData(DataModel):
+    """ModalityLightingData data model."""
+
     high: float
     medium: float
     low: float

core/dto/model_data.py

@@ -0,0 +1,82 @@
+"""Definition of ModelData data model."""
+from __future__ import annotations
+
+from .angle import AngleData
+from .contact import ContactData
+from .data_model import DataModel
+from .distance import DistanceData
+from .framing import FramingData
+from .information_value import InformationValueData
+from .modality_color import ModalityColorData
+from .modality_depth import ModalityDepthData
+from .modality_lighting import ModalityLightingData
+from .point_of_view import PointOfViewData
+from .salience import SalienceData
+from .visual_syntax import VisualSyntaxData
+
+
+class ModelData(DataModel):
+    """ModelData model for data IO with combined ML model."""
+    visual_syntax: VisualSyntaxData
+    contact: ContactData
+    angle: AngleData
+    point_of_view: PointOfViewData
+    distance: DistanceData
+    modality_lighting: ModalityLightingData
+    modality_color: ModalityColorData
+    modality_depth: ModalityDepthData
+    information_value: InformationValueData
+    framing: FramingData
+    salience: SalienceData
+
+    @classmethod
+    def from_random(cls) -> ModelData:
+        """Instantiate with random numbers."""
+        kwargs = {
+            field: field_info.annotation.from_random()  # type: ignore
+            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,
+    ) -> ModelData:
+        """Instantiate from annotation."""
+        kwargs = {
+            'visual_syntax': VisualSyntaxData
+            .from_choice(visual_syntax),
+            'contact': ContactData
+            .from_choice(contact),
+            'angle': AngleData
+            .from_choice(angle),
+            'point_of_view': PointOfViewData
+            .from_choice(point_of_view),
+            'distance': DistanceData
+            .from_choice(distance),
+            'modality_lighting': ModalityLightingData
+            .from_choice(modality_lighting),
+            'modality_color': ModalityColorData
+            .from_choice(modality_color),
+            'modality_depth': ModalityDepthData
+            .from_choice(modality_depth),
+            'information_value': InformationValueData
+            .from_choice(information_value),
+            'framing': FramingData
+            .from_choice(framing),
+            'salience': SalienceData
+            .from_choice(salience),
+        }
+        return cls(**kwargs)

core/dto/point_of_view.py

@@ -1,8 +1,11 @@
+"""Definition of PointOfViewData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class PointOfViewData(DataModel):
+    """PointOfViewData data model."""
+
     frontal: float
     oblique: float

core/dto/salience.py

@@ -1,9 +1,12 @@
+"""Definition of SalienceData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class SalienceData(DataModel):
+    """SalienceData data model."""
+
     size: float
     colour: float
     tone: float

core/dto/visual_syntax.py

@@ -1,9 +1,12 @@
+"""Definition of VisualSyntaxData data model."""
 from __future__ import annotations
 
 from .data_model import DataModel
 
 
 class VisualSyntaxData(DataModel):
+    """VisualSyntaxData data model."""
+
     non_transactional_action: float
     non_transactional_reaction: float
     unidirectional_transactional_action: float

docker-compose.local.yml

@@ -1,21 +1,21 @@
 version: '3.7'
 services:
-  app:
-    image: visual_critical_discourse_analysis:dev
-    container_name: visual_critical_discourse_analysis
-    build:
-      context: .
-      dockerfile: Dockerfile
-    env_file:
-      - local.env
-    environment:
-      - ENV=DEV
-    ports:
-      - 8050:8050
-    networks:
-      - backend
-    depends_on:
-      - mongo
+  # app:
+  #   image: visual_critical_discourse_analysis:dev
+  #   container_name: visual_critical_discourse_analysis
+  #   build:
+  #     context: .
+  #     dockerfile: Dockerfile
+  #   env_file:
+  #     - local.env
+  #   environment:
+  #     - ENV=DEV
+  #   ports:
+  #     - 8050:8050
+  #   networks:
+  #     - backend
+  #   depends_on:
+  #     - mongo
   mongo:
     image: mongo:latest
     container_name: mongo
@@ -41,4 +41,4 @@ services:
 
 networks:
   backend:
-    driver: bridge
+    driver: bridge

model/src/models/__init__.py

@@ -0,0 +1,3 @@
+from __future__ import annotations
+
+from .visual_communication import VisualCommunicationModel

model/src/models/visual_communication.py

@@ -2,9 +2,29 @@ from __future__ import annotations
 
 import torch.nn as nn
 
+from .angle import AngleTail
+from .contact import ContactTail
+from .distance import DistanceTail
+from .framing import FramingTail
+from .information_value import InformationValueTail
+from .modality_color import ModalityColorTail
+from .modality_depth import ModalityDepthTail
+from .modality_lighting import ModalityLightingTail
+from .point_of_view import PointOfViewTail
 from .resnet18_head import ResNet18Head
+from .salience import SalienceTail
 from .visual_syntax import VisualSyntaxTail
-from core.data_models import VisualSyntaxData
+from core.dto import AngleData
+from core.dto import ContactData
+from core.dto import DistanceData
+from core.dto import FramingData
+from core.dto import InformationValueData
+from core.dto import ModalityColorData
+from core.dto import ModalityDepthData
+from core.dto import ModalityLightingData
+from core.dto import PointOfViewData
+from core.dto import SalienceData
+from core.dto import VisualSyntaxData
 
 
 class VisualCommunicationModel(nn.Module):
@@ -13,6 +33,16 @@ class VisualCommunicationModel(nn.Module):
         # store other models
         self.resnet_head = ResNet18Head()
         self.visual_syntax_tail = VisualSyntaxTail()
+        self.contact_tail = ContactTail()
+        self.angle_tail = AngleTail()
+        self.point_of_view_tail = PointOfViewTail()
+        self.distance_tail = DistanceTail()
+        self.modality_lighting_tail = ModalityLightingTail()
+        self.modality_color_tail = ModalityColorTail()
+        self.modality_depth_tail = ModalityDepthTail()
+        self.information_value_tail = InformationValueTail()
+        self.framing_tail = FramingTail()
+        self.salience_tail = SalienceTail()
 
     def forward(self, x):
         # generate visual representation
@@ -20,10 +50,69 @@ class VisualCommunicationModel(nn.Module):
         # prepare result map
         results = {}
         # predict visual syntax
-        visual_syntax_pred = self.visual_syntax_tail(vis_rep).cpu()
-        # visual_syntax_pred = float()
         results['visual_syntax'] = VisualSyntaxData.from_list(
-            visual_syntax_pred,
+            self.visual_syntax_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict contact
+        results['contact'] = ContactData.from_list(
+            self.contact_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict angle
+        results['angle'] = AngleData.from_list(
+            self.angle_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict point of view
+        results['point_of_view'] = PointOfViewData.from_list(
+            self.point_of_view_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict distance
+        results['distance'] = DistanceData.from_list(
+            self.distance_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict modality lighting
+        results['modality_lighting'] = ModalityLightingData.from_list(
+            self.modality_lighting_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict modality color
+        results['modality_color'] = ModalityColorData.from_list(
+            self.modality_color_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict modality depth
+        results['modality_depth'] = ModalityDepthData.from_list(
+            self.modality_depth_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict information value
+        results['information_value'] = InformationValueData.from_list(
+            self.information_value_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict framing
+        results['framing'] = FramingData.from_list(
+            self.framing_tail(
+                vis_rep,
+            ).cpu(),
+        )
+        # predict salience
+        results['salience'] = SalienceData.from_list(
+            self.salience_tail(
+                vis_rep,
+            ).cpu(),
         )
-
         return results

src/web/app.py

@@ -13,15 +13,14 @@ from dash_auth import BasicAuth
 from pydantic import ValidationError
 
 from .layout import app_layout
-from src.database import connect
-from src.database import get_visual_communication
-from src.database import ModelOutputs
-from src.database import NoDocumentFoundException
-from src.database import total_annotated
-from src.database import total_documents
-from src.database import upsert_annotations
-from src.database import upsert_visual_communication
-from src.database import VisualCommunication
+from core.database import connect
+from core.database import count_documents
+from core.database import get_visual_communication
+from core.database import NoDocumentFoundException
+from core.database import upsert_annotation
+from core.database import upsert_visual_communication
+from core.database import VisualCommunication
+from core.dto import ModelData
 
 # setup app
 app = Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP])
@@ -82,8 +81,14 @@ def update_progress_bar(
     logging.info('began updating progress bar')
     global collection
     # get values from database
-    num_total = total_documents(collection=collection)
-    num_handled = total_annotated(collection=collection)
+    num_total = count_documents(
+        collection=collection,
+        only_with_annotation=False,
+    )
+    num_handled = count_documents(
+        collection=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
@@ -200,9 +205,9 @@ def cycle_visual_communication_data(
                 in zip(annotation_keys, annotation_values)
             }
             # instantiate ModelOutputs object
-            annotations = ModelOutputs.from_annotations(**annotation_map)
+            annotations = ModelData.from_annotations(**annotation_map)
             # save data to database
-            upsert_annotations(
+            upsert_annotation(
                 collection=collection,
                 vis_com_name=vis_com_name,
                 annotations=annotations,

src/web/layout/labels.py

@@ -4,35 +4,24 @@ import dash_mantine_components as dmc
 from dash import dcc
 from dash import html
 
-from src.model_experiential import VisualSyntaxModelOutput
-from src.model_interpersonal import AngleModelOutput
-from src.model_interpersonal import ContactModelOutput
-from src.model_interpersonal import DistanceModelOutput
-from src.model_interpersonal import ModalityColorModelOutput
-from src.model_interpersonal import ModalityDepthModelOutput
-from src.model_interpersonal import ModalityLightingModelOutput
-from src.model_interpersonal import PointOfViewModelOutput
-from src.model_textual import FramingModelOutput
-from src.model_textual import InformationValueModelOutput
-from src.model_textual import 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
+from core.dto import AngleData
+from core.dto import ContactData
+from core.dto import DistanceData
+from core.dto import FramingData
+from core.dto import InformationValueData
+from core.dto import ModalityColorData
+from core.dto import ModalityDepthData
+from core.dto import ModalityLightingData
+from core.dto import PointOfViewData
+from core.dto import SalienceData
+from core.dto import VisualSyntaxData
 
 
 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,
-    )
+    options_map['visual syntax'] = VisualSyntaxData.list_fields()
     return options_map
 
 
@@ -40,21 +29,13 @@ 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,
-    )
+    options_map['contact'] = ContactData.list_fields()
+    options_map['angle'] = AngleData.list_fields()
+    options_map['point of view'] = PointOfViewData.list_fields()
+    options_map['distance'] = DistanceData.list_fields()
+    options_map['modality lighting'] = ModalityLightingData.list_fields()
+    options_map['modality color'] = ModalityColorData.list_fields()
+    options_map['modality depth'] = ModalityDepthData.list_fields()
     return options_map
 
 
@@ -62,11 +43,9 @@ 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)
+    options_map['information value'] = InformationValueData.list_fields()
+    options_map['framing'] = FramingData.list_fields()
+    options_map['salience'] = SalienceData.list_fields()
     return options_map
 
 
@@ -83,7 +62,7 @@ for title, options in experiential_map.items():
         dmc.Container([
             html.B(title.title()),
             dcc.RadioItems(
-                options=options,
+                options=[text.replace('_', ' ') for text in options],
                 id=id_dict,
             ),
         ]),
@@ -101,7 +80,7 @@ for title, options in interpersonal_map.items():
         dmc.Container([
             html.B(title.title()),
             dcc.RadioItems(
-                options=options,
+                options=[text.replace('_', ' ') for text in options],
                 id=id_dict,
             ),
         ]),
@@ -117,9 +96,9 @@ for title, options in textual_map.items():
     id_dict = {'type': 'annotation', 'index': title.replace('_', '-')}
     textual_container.children.append(
         dmc.Container([
-            html.B(title),
+            html.B(title.title()),
             dcc.RadioItems(
-                options=options,
+                options=[text.replace('_', ' ') for text in options],
                 id=id_dict,
             ),
         ]),

tests/generate_random_prediction_test.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from pathlib import Path
 
-from database import VisualCommunication
+from core.database.classes import VisualCommunication
 
 
 if __name__ == '__main__':

tests/get_visual_communication_test.py

@@ -3,10 +3,11 @@ from __future__ import annotations
 import os
 from pathlib import Path
 
-from database import connect
-from database import get_visual_communication
 from dotenv import load_dotenv
 
+from core.database import connect
+from core.database import get_visual_communication
+
 if __name__ == '__main__':
     # prepare env vars
     env_path = Path(__file__).parent.parent / 'local.env'

tests/image_download_test.py

@@ -3,10 +3,11 @@ from __future__ import annotations
 import os
 from pathlib import Path
 
-from database import connect
-from database import VisualCommunication
 from dotenv import load_dotenv
 
+from core.database import connect
+from core.database.classes import VisualCommunication
+
 if __name__ == '__main__':
     # prepare env vars
     env_path = Path(__file__).parent.parent / 'local.env'

tests/image_upload_test.py

@@ -3,11 +3,12 @@ from __future__ import annotations
 import os
 from pathlib import Path
 
-from database import connect
-from database import VisualCommunication
 from dotenv import load_dotenv
 from pymongo.errors import DuplicateKeyError
 
+from core.database import connect
+from core.database.classes import VisualCommunication
+
 if __name__ == '__main__':
     # get list of image paths
     test_dir = Path(__file__).parent

tests/image_upload_to_server_test.py

@@ -5,8 +5,8 @@ from pathlib import Path
 from dotenv import load_dotenv
 from pymongo.errors import DuplicateKeyError
 
-from src.database import connect
-from src.database import VisualCommunication
+from core.database import connect
+from core.database import VisualCommunication
 
 if __name__ == '__main__':
     # get list of image paths

tests/model_outputs_from_annotation_test.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from database import ModelOutputs
+from core.dto import ModelData
 
 
 if __name__ == '__main__':
     # instantiate data object
     vis_com_list = [
-        ModelOutputs.from_random()
+        ModelData.from_random()
         for i
         in range(3)
     ]

tests/prediction_upload_test.py

@@ -4,11 +4,12 @@ import logging
 import os
 from pathlib import Path
 
-from database import connect
-from database import upsert_predictions
-from database import VisualCommunication
 from dotenv import load_dotenv
 
+from core.database import connect
+from core.database import upsert_predictions
+from core.database.classes import VisualCommunication
+
 if __name__ == '__main__':
     # setup logging
     fmt = (

tests/total_annotated_test.py

@@ -3,10 +3,11 @@ from __future__ import annotations
 import os
 from pathlib import Path
 
-from database import connect
-from database import total_annotated
 from dotenv import load_dotenv
 
+from core.database import connect
+from core.database import total_documents
+
 if __name__ == '__main__':
     # prepare env vars
     env_path = Path(__file__).parent.parent / 'local.env'
@@ -16,5 +17,7 @@ if __name__ == '__main__':
     # connect to database
     collection, db, client = connect()
     # get visual communication
-    num_docs = total_annotated(collection)
+    num_docs = total_documents(
+        collection=collection,
+    )
     print(f"number of annotated  documents in database: {num_docs}")

tests/total_documents_test.py

@@ -3,10 +3,11 @@ from __future__ import annotations
 import os
 from pathlib import Path
 
-from database import connect
-from database import total_documents
 from dotenv import load_dotenv
 
+from core.database import connect
+from core.database import total_documents
+
 if __name__ == '__main__':
     # prepare env vars
     env_path = Path(__file__).parent.parent / 'local.env'
@@ -16,5 +17,7 @@ if __name__ == '__main__':
     # connect to database
     collection, db, client = connect()
     # get visual communication
-    num_docs = total_documents(collection)
+    num_docs = total_documents(
+        collection=collection,
+    )
     print(f"total number of documents in database: {num_docs}")