Merge pull request '#67_unittests_for_DTOs' (#69) from #67_unittests_for_DTOs into main

Reviewed-on: #69

.gitea/workflows/publish.yaml

@@ -1,9 +1,8 @@
 name: CI Pipeline
 run-name: ${{ gitea.actor }} is running the CI Pipeline
 on:
-  pull_request:
+  release:
-    branches:
+    types: [published]
-      - main
 jobs:
   test:
     name: Test

.gitea/workflows/pull.yaml

@@ -0,0 +1,50 @@
+name: Code Quality Pipeline
+on:
+  pull_request:
+    branches:
+      - main
+jobs:
+  job:
+    name: Check Code
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Code
+        uses: actions/checkout@v4
+      - name: Setup Environment
+        uses: https://github.com/actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          architecture: "x64"
+      - name: Setup poetry
+        env:
+          POETRY_VERSION: 2.1.1
+          POETRY_HOME: /opt/poetry
+          POETRY_NO_INTERACTION: 1
+          POETRY_NO_CACHE: 1
+        run: |
+          curl -sSL https://install.python-poetry.org | python3 -
+          export PATH=$POETRY_HOME/bin:$PATH
+          poetry --version
+      - name: Install Dependencies
+        env:
+          PIP_INDEX_URL: ${{ vars.PIP_INDEX_URL }}
+          PIP_TRUSTED_HOST: ${{ vars.PIP_TRUSTED_HOST }}
+        run: |
+          /opt/poetry/bin/poetry install
+      - name: PEP8 Check
+        run: |
+          /opt/poetry/bin/poetry run flake8 . --benchmark
+      - name: Type Check
+        run: |
+          /opt/poetry/bin/poetry run mypy .
+      - name: Pytest & Calculate Coverage
+        env:
+          MINIO_ENDPOINT: ${{ vars.MINIO_ENDPOINT }}
+          MINIO_ACCESS_KEY: ${{ vars.MINIO_ACCESS_KEY }}
+          MINIO_SECRET_KEY: ${{ secrets.MINIO_SECRET_KEY }}
+          MONGO_ENDPOINT: ${{ vars.MONGO_ENDPOINT }}
+        run: |
+          /opt/poetry/bin/poetry run coverage run -m pytest .
+      - name: Coverage Report
+        run: |
+          /opt/poetry/bin/poetry run coverage report -m

.gitignore

@@ -177,3 +177,6 @@ ipython_config.py
 
 # Remove previous ipynb_checkpoints
 #   git rm -r .ipynb_checkpoints/
+
+# Logging data
+runs/*

.pre-commit-config.yaml

@@ -1,37 +1,33 @@
 repos:
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.5.0
+    rev: v5.0.0
     hooks:
     -   id: trailing-whitespace
     -   id: end-of-file-fixer
     -   id: check-yaml
+    -   id: check-added-large-files
     -   id: debug-statements
     -   id: double-quote-string-fixer
     -   id: name-tests-test
-    -   id: requirements-txt-fixer
 -   repo: https://github.com/asottile/setup-cfg-fmt
-    rev: v2.5.0
+    rev: v2.7.0
     hooks:
     -   id: setup-cfg-fmt
--   repo: https://github.com/asottile/reorder-python-imports
+-   repo: https://github.com/pre-commit/mirrors-isort
-    rev: v3.12.0
+    rev: v5.10.1
     hooks:
-    -   id: reorder-python-imports
+      - id: isort
-        exclude: ^(pre_commit/resources/|testing/resources/python3_hooks_repo/)
+        language_version: python3.10
-        args: [--py39-plus, --add-import, 'from __future__ import annotations']
+        args: [ --tc ]
 -   repo: https://github.com/asottile/add-trailing-comma
     rev: v3.1.0
     hooks:
     -   id: add-trailing-comma
 -   repo: https://github.com/asottile/pyupgrade
-    rev: v3.15.1
+    rev: v3.19.1
     hooks:
     -   id: pyupgrade
         args: [--py39-plus]
--   repo: https://github.com/hhatto/autopep8
-    rev: v2.0.4
-    hooks:
-    -   id: autopep8
 -   repo: https://github.com/PyCQA/flake8
     rev: 7.0.0
     hooks:
@@ -40,8 +36,27 @@ repos:
         additional_dependencies:
           - "pyproject-flake8"
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.8.0
+    rev: v1.15.0
     hooks:
     -   id: mypy
-        additional_dependencies: [types-all]
         exclude: ^testing/resources/
+-   repo: https://github.com/psf/black
+    rev: 25.1.0
+    hooks:
+    - id: black
+      language_version: python3.12
+      args: [ --skip-string-normalization ]
+-   repo: https://github.com/myint/docformatter
+    rev: v1.5.0
+    hooks:
+      - id: docformatter
+        name: docformatter
+        description: 'Formats docstrings to follow PEP 257.'
+        entry: docformatter
+        language: python
+        types: [ python ]
+-     repo: https://github.com/jendrikseipp/vulture
+      rev: 'v2.14'
+      hooks:
+      - id: vulture
+        entry: vulture . --min-confidence 90 --exclude */.venv/*.py,*/tests/*.py

Dockerfile.model

@@ -1,5 +1,5 @@
 # build stage
-FROM python:3.12-slim-bookworm as BUILDER
+FROM python:3.12-slim-bookworm AS builder
 
 # set environment variables
 ENV PYTHONUNBUFFERED=1 \
@@ -44,7 +44,7 @@ COPY --from=builder ${VIRTUAL_ENV} ${VIRTUAL_ENV}
 ENV PATH="${VIRTUAL_ENV}/bin:${PATH}"
 
 # create home directory and app user
-RUN mkdir -p /home/app
+RUN mkdir -p $APP_HOME
 
 # add code while changing ownership
 WORKDIR $APP_HOME

Dockerfile.web_ui

@@ -1,5 +1,5 @@
 # build stage
-FROM python:3.12-slim-bookworm as BUILDER
+FROM python:3.12-slim-bookworm AS builder
 
 # set environment variables
 ENV PYTHONUNBUFFERED=1 \

misc/image_download.py

@@ -5,8 +5,8 @@ from pathlib import Path
 
 from dotenv import load_dotenv
 
-from shared.database import connect
+from shared.docstore import connect_mongodb
-from shared.database.classes import VisualCommunication
+from shared.docstore.classes import VisualCommunication
 
 if __name__ == '__main__':
     # prepare env vars
@@ -15,7 +15,7 @@ if __name__ == '__main__':
     load_dotenv(env_path)
     os.environ['MONGO_HOST'] = 'localhost'
     # connect to database
-    collection, db, client = connect()
+    collection, db, client = connect_mongodb()
     print(client.server_info())
     # download images
     data = None

misc/model_outputs_from_annotation.py

@@ -0,0 +1,9 @@
+from shared.docstore.classes import ModelData
+
+if __name__ == '__main__':
+    # instantiate data object
+    vis_com_list = [ModelData.from_random() for i in range(3)]
+    # generate random predictions
+    [vis_com.from_random() for vis_com in vis_com_list]
+    for vis_com in vis_com_list:
+        print(vis_com)

misc/total_annotated.py

@@ -5,8 +5,7 @@ from pathlib import Path
 
 from dotenv import load_dotenv
 
-from shared.database import connect
+from shared.docstore import connect_mongodb, count_documents
-from shared.database import count_documents
 
 if __name__ == '__main__':
     # prepare env vars
@@ -15,7 +14,7 @@ if __name__ == '__main__':
     load_dotenv(env_path)
     os.environ['MONGO_HOST'] = 'localhost'
     # connect to database
-    collection, db, client = connect()
+    collection, db, client = connect_mongodb()
     # get visual communication
     num_docs = count_documents(
         collection=collection,

misc/total_documents.py

@@ -5,8 +5,7 @@ from pathlib import Path
 
 from dotenv import load_dotenv
 
-from shared.database import connect
+from shared.docstore import connect_mongodb, count_documents
-from shared.database import count_documents
 
 if __name__ == '__main__':
     # prepare env vars
@@ -15,7 +14,7 @@ if __name__ == '__main__':
     load_dotenv(env_path)
     os.environ['MONGO_HOST'] = 'localhost'
     # connect to database
-    collection, db, client = connect()
+    collection, db, client = connect_mongodb()
     # get visual communication
     num_docs = count_documents(
         collection=collection,

model/docker-compose.server.yml

@@ -9,3 +9,8 @@ services:
       - ../server.env
     environment:
       - ENV=TEST
+    deploy:
+      resources:
+        limits:
+          cpus: '2.000'
+          memory: 8G

model/src/configs/resnet18.json

@@ -0,0 +1,14 @@
+{
+    "datasets": {
+        "train": {
+            "filelist": "datasets/train.csv"
+        },
+        "val": {
+            "filelist": "datasets/val.csv"
+        }
+    },
+    "backbone": {
+        "class": "model.src.models.VisualCommunicationModel",
+        "model_name": "pretrained"
+    }
+}

model/src/data_store/__init__.py

@@ -1,5 +0,0 @@
-from __future__ import annotations
-
-from .connect import connect
-from .get import get
-from .put import put

model/src/data_store/connect.py

@@ -1,30 +0,0 @@
-"""Definition of connect function."""
-from __future__ import annotations
-
-import os
-
-from minio import Minio
-
-
-def connect(
-) -> Minio:
-    """Connect to MinIO server"""
-    minio_endpoint = os.getenv('MINIO_ENDPOINT', default=None)
-    assert isinstance(minio_endpoint, str)
-    minio_access_key = os.getenv('MINIO_ACCESS_KEY', default=None)
-    assert isinstance(minio_access_key, str)
-    minio_secret_key = os.getenv('MINIO_SECRET_KEY', default=None)
-    assert isinstance(minio_secret_key, str)
-    minio_bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
-    assert isinstance(minio_bucket_name, str)
-    # connect client
-    client = Minio(
-        endpoint=minio_endpoint,
-        access_key=minio_access_key,
-        secret_key=minio_secret_key,
-        secure=False,
-    )
-    # ensure bucket exists
-    if not client.bucket_exists(bucket_name=minio_bucket_name):
-        client.make_bucket(bucket_name=minio_bucket_name)
-    return client

model/src/data_store/get.py

@@ -1,27 +0,0 @@
-"""Definition of get function."""
-from __future__ import annotations
-
-import os
-from io import BytesIO
-
-from minio import Minio
-
-
-def get(
-    client: Minio,
-    object_name: str,
-) -> BytesIO:
-    """Get buffer from bucket in MinIO."""
-    assert isinstance(client, Minio)
-    assert isinstance(object_name, str)
-    bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
-    assert isinstance(bucket_name, str)
-    # get buffer
-    try:
-        response = client.get_object(bucket_name, object_name)
-        buffer = BytesIO(response.data)
-    finally:
-        response.close()
-        response.release_conn()
-    buffer.seek(0)
-    return buffer

model/src/data_store/put.py

@@ -1,30 +0,0 @@
-"""Definition of put function."""
-from __future__ import annotations
-
-import os
-from io import BytesIO
-
-from minio import Minio
-
-
-def put(
-    client: Minio,
-    buffer: BytesIO,
-    object_name: str,
-) -> None:
-    """Put buffer in bucket in MinIO."""
-    assert isinstance(client, Minio)
-    assert isinstance(buffer, BytesIO)
-    assert isinstance(object_name, str)
-    bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
-    assert isinstance(bucket_name, str)
-    # prepare for saving
-    num_bytes = buffer.tell()
-    buffer.seek(0)
-    # send data to bucket
-    client.put_object(
-        bucket_name=bucket_name,
-        object_name=object_name,
-        length=num_bytes,
-        data=buffer,
-    )

model/src/dataloader.py

@@ -1,146 +0,0 @@
-from __future__ import annotations
-
-import random
-from io import BytesIO
-
-from torch import Tensor
-from torch.utils.data import Dataset
-from torchvision.io import read_image
-from torchvision.transforms import ColorJitter
-from torchvision.transforms import InterpolationMode
-from torchvision.transforms import Normalize
-from torchvision.transforms.functional import hflip
-from torchvision.transforms.functional import pad
-from torchvision.transforms.functional import resize
-from torchvision.transforms.functional import rotate
-
-from shared.database import connect
-
-# resnet18 original normalization values
-RESNET_NORMALIZE_MEAN = [0.485, 0.456, 0.406]
-RESNET_NORMALIZE_STD = [0.229, 0.224, 0.225]
-
-
-class VCDADataset(Dataset):
-    def __init__(
-        self,
-        data_name_list: list[str],
-        do_augment: bool = False,
-        random_annotations: bool = False,
-        normalize_mean: list[float] = RESNET_NORMALIZE_MEAN,
-        normalize_std: list[float] = RESNET_NORMALIZE_STD,
-    ):
-        super().__init__()
-        self.data_name_list = data_name_list
-        self.do_augment = do_augment
-        self.random_annotations = random_annotations
-        self.normalize_mean = normalize_mean
-        self.normalize_std = normalize_std
-        # prepare augmentation functions
-        self.normalize = Normalize(
-            mean=normalize_mean,
-            std=normalize_std,
-        )
-        self.color_jitter = ColorJitter(
-            brightness=1e-1,
-            contrast=8e-2,
-            saturation=8e-2,
-        )
-        # connect to database
-        collection, _, _ = connect()
-        self.collection = collection
-
-    def __len__(self):
-        return len(self.data_name_list)
-
-    def __getitem__(self, idx):
-        # get image from database
-        name = self.data_name_list[idx]
-        query = {
-            'name': name,
-        }
-        projection = {
-            '_id': False,
-            'image': True,
-        }
-        img_bytes = self.collection.find_one(
-            filter=query,
-            projection=projection,
-        )
-        img = self.load_image(img_bytes)
-        if self.do_augment:
-            img = self.augment(img)
-        return img
-
-    def load_image(
-        self,
-        data: bytes,
-    ) -> Tensor:
-        """Load images tensor from bytes."""
-        assert isinstance(data, bytes)
-        img = read_image(BytesIO(data))
-        img /= 255  # normalize 8-bit image
-        img = self.square_pad(img)
-        img = resize(
-            img=img,
-            size=(512, 512),
-            interpolation=InterpolationMode.BICUBIC,
-        )
-        return img
-
-    @staticmethod
-    def square_pad(
-        img: Tensor,
-    ) -> Tensor:
-        """
-        Pads image to a square with side length
-        equal to the largest side of the input image.
-        """
-        assert isinstance(img, Tensor)
-        # B, nc, w, h = img.shape
-        h = img.shape[-2]
-        w = img.shape[-1]
-        if h == w:
-            return img
-        max_wh = max([h, w])
-        hp = int((max_wh - w) / 2)
-        vp = int((max_wh - h) / 2)
-        padding = (hp, vp, hp, vp)
-        return pad(img, padding, 0, 'constant')
-
-    def augment(
-        self,
-        img: Tensor,
-    ) -> Tensor:
-        """
-        Augment image with random horizontal flips,
-        rotations and color jitter.
-        """
-        assert isinstance(img, Tensor)
-        # left-right flip
-        if random.random() >= 0.5:
-            img = hflip(img)
-        # rotation
-        rnd = random.random()
-        if rnd < 0.25:
-            img = rotate(img, angle=90)
-        if rnd < 0.5:
-            img = rotate(img, angle=180)
-        if rnd < 0.75:
-            img = rotate(img, angle=270)
-        # color jitter
-        img = self.color_jitter(img)
-        return img
-
-    def reverse_normalise(
-        self,
-        img: Tensor,
-    ) -> Tensor:
-        """
-        Reverse normalization to get an image
-        that can be interpreted by humans.
-        """
-        assert isinstance(img, Tensor)
-        img *= Tensor(self.normalize_std).reshape((3, 1, 1))
-        img += Tensor(self.normalize_mean).reshape((3, 1, 1))
-        return img

model/src/main.py

@@ -1,45 +1,43 @@
 """Main script to be run by service."""
-from __future__ import annotations
 
-from hashlib import md5
+import json
-from io import BytesIO
+import logging
+from traceback import print_exc
 
 import torch
-from data_store import connect
-from data_store import put
 from models import VisualCommunicationModel
-from torchinfo import summary
+from tqdm import tqdm
-from utils import check_env
+from utils import DEVICE, VCDADataset, load_model
 
 from shared.utils import setup_logging
 
-
-DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-
 if __name__ == '__main__':
-    # check that environment variables are set
-    check_env()
     # setup logging
     setup_logging()
     # instantiate model
-    model = VisualCommunicationModel().to(DEVICE)
+    model: VisualCommunicationModel = load_model()
-    # show model weights
+    model.eval()
-    summary(model)
+    # setup dataset
-    # for param in model.parameters():
+    dataset = VCDADataset(
-    #     logging.info(param.data)
+        data_name_list=[
-    # save model to buffer
+            '02dbaf48d713e4e6d3a6b98fd2dc866e',
-    buffer = BytesIO()
+        ],
-    torch.save(model.state_dict(), buffer)
+        do_augment=False,
-    print(f"buffer size: {len(buffer.getvalue())}")
-    # calculate buffer hash
-    hash_str = md5(buffer.getbuffer()).hexdigest()
-    print(f"hash string: {hash_str}")
-    # connect to minio
-    client = connect()
-    # put buffer in minio bucket
-    put(
-        client=client,
-        buffer=buffer,
-        object_name=hash_str,
     )
-    print('saved to Minio')
+
+    # make prediction
+    with torch.no_grad():
+        for i in tqdm(range(len(dataset))):
+            try:
+                # get image
+                image = dataset[i]
+                image = torch.unsqueeze(image, 0)  # add artificial batch dimension
+                image = image.to(DEVICE)
+                # make prediction
+                pred: dict = model(image)
+            except Exception:
+                print_exc()
+                continue
+            else:
+                print(json.dumps(pred, indent=4))
+    logging.debug('finished')

model/src/model_name.txt

@@ -0,0 +1 @@
+2bbe2be17a663f4299657378ac5e993e

model/src/models/angle.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/contact.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/distance.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/framing.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/fully_connected.py

@@ -1,26 +1,24 @@
-from __future__ import annotations
+from torch import nn
-
-import torch.nn as nn
 
 
 class FullyConnectedModel(nn.Module):
+    """Fully connected layers model template for intrepreting feature space-
+    output from ResNet18 head."""
+
     def __init__(self, num_out_features: int):
         super().__init__()
         # define layers
-        self.fc1 = nn.Linear(in_features=16*16*512, out_features=512)
+        self.fc1 = nn.Linear(in_features=512, out_features=128)
         self.af1 = nn.ReLU()
-        self.fc2 = nn.Linear(in_features=512, out_features=128)
+        self.fc2 = nn.Linear(in_features=128, out_features=32)
         self.af2 = nn.ReLU()
-        self.fc3 = nn.Linear(in_features=128, out_features=32)
+        self.fc3 = nn.Linear(in_features=32, out_features=num_out_features)
-        self.af3 = nn.ReLU()
-        self.fc4 = nn.Linear(in_features=32, out_features=num_out_features)
 
     def forward(self, x):
+        """Pass input through model."""
         x = self.fc1(x)
         x = self.af1(x)
         x = self.fc2(x)
         x = self.af2(x)
         x = self.fc3(x)
-        x = self.af3(x)
-        x = self.fc4(x)
         return x

model/src/models/information_value.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/modality_color.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/modality_depth.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/modality_lighting.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/point_of_view.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/resnet18_head.py

@@ -5,11 +5,15 @@ import torchvision
 
 
 class ResNet18Head(nn.Module):
-    def __init__(self):
+    def __init__(self, download_resnet_weights: bool = False):
         super().__init__()
         # copy out parts from ResNet18 with weights
+        if download_resnet_weights:
+            weights = torchvision.models.ResNet18_Weights.IMAGENET1K_V1
+        else:
+            weights = None
         resnet18 = torchvision.models.resnet18(
-            weights=torchvision.models.ResNet18_Weights.IMAGENET1K_V1,
+            weights=weights,
         )
         # save relevant layers
         self.conv1 = resnet18.conv1
@@ -21,7 +25,7 @@ class ResNet18Head(nn.Module):
         self.layer3 = resnet18.layer3
         self.layer4 = resnet18.layer4
         self.avgpool = resnet18.avgpool
-        self.flat = nn.Flatten()    # size 512
+        self.flat = nn.Flatten()  # size 512
 
     def forward(self, x):
         x = self.conv1(x)

model/src/models/salience.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/models/visual_communication.py

@@ -1,4 +1,5 @@
 """Definition of VisualCommunicationModel class."""
+
 from __future__ import annotations
 
 from torch import nn
@@ -15,26 +16,15 @@ from .point_of_view import PointOfViewTail
 from .resnet18_head import ResNet18Head
 from .salience import SalienceTail
 from .visual_syntax import VisualSyntaxTail
-from shared.dto import AngleData
-from shared.dto import ContactData
-from shared.dto import DistanceData
-from shared.dto import FramingData
-from shared.dto import InformationValueData
-from shared.dto import ModalityColorData
-from shared.dto import ModalityDepthData
-from shared.dto import ModalityLightingData
-from shared.dto import PointOfViewData
-from shared.dto import SalienceData
-from shared.dto import VisualSyntaxData
 
 
 class VisualCommunicationModel(nn.Module):
     """Visual communication model."""
 
-    def __init__(self):
+    def __init__(self, download_resnet_weights: bool = False):
         super().__init__()
         # store other models
-        self.resnet_head = ResNet18Head()
+        self.resnet_head = ResNet18Head(download_resnet_weights)
         self.visual_syntax_tail = VisualSyntaxTail()
         self.contact_tail = ContactTail()
         self.angle_tail = AngleTail()
@@ -47,76 +37,22 @@ class VisualCommunicationModel(nn.Module):
         self.framing_tail = FramingTail()
         self.salience_tail = SalienceTail()
 
-    def forward(self, x):
+    def forward(self, x) -> dict:
         """Calculate model output on data."""
         # generate visual representation
-        vis_rep = self.resnet_head(x)
+        features = self.resnet_head(x)
-        # prepare result map
+        # make predictions
-        results = {}
+        prediction_dict = {
-        # predict visual syntax
+            'visual_syntax': self.visual_syntax_tail(features),
-        results['visual_syntax'] = VisualSyntaxData.from_list(
+            'contact': self.contact_tail(features),
-            self.visual_syntax_tail(
+            'angle': self.angle_tail(features),
-                vis_rep,
+            'point_of_view': self.point_of_view_tail(features),
-            ).cpu(),
+            'distance': self.distance_tail(features),
-        )
+            'modality_lighting': self.modality_lighting_tail(features),
-        # predict contact
+            'modality_color': self.modality_color_tail(features),
-        results['contact'] = ContactData.from_list(
+            'modality_depth': self.modality_depth_tail(features),
-            self.contact_tail(
+            'information_value': self.information_value_tail(features),
-                vis_rep,
+            'framing': self.framing_tail(features),
-            ).cpu(),
+            'salience': self.salience_tail(features),
-        )
+        }
-        # predict angle
+        return prediction_dict
-        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

model/src/models/visual_syntax.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 from .fully_connected import FullyConnectedModel
 
 

model/src/utils/__init__.py

@@ -1,3 +1,3 @@
-from __future__ import annotations
+from .get_class import get_class
-
+from .load_model import DEVICE, load_model
-from .check_env import check_env
+from .vcda_dataset import VCDADataset

model/src/utils/check_env.py

@@ -1,25 +0,0 @@
-"""Definition of check_env function."""
-from __future__ import annotations
-
-import os
-
-
-def check_env() -> None:
-    """Check necessary environment variables are set."""
-    necesasary_var_list = {
-        'MONGO_HOST',
-        'MONGO_DB',
-        'MONGO_COLLECTION',
-        'MONGO_USER',
-        'MONGO_PASSWORD',
-        'MINIO_ENDPOINT',
-        'MINIO_ACCESS_KEY',
-        'MINIO_SECRET_KEY',
-    }
-    for env_var in necesasary_var_list:
-        # ensure env var set
-        assert (
-            env_var in os.environ
-        ), (
-            f"environment variable not set: {env_var}"
-        )

model/src/utils/get_class.py

@@ -0,0 +1,11 @@
+from importlib import import_module
+
+from torch.nn import Module
+
+
+def get_class(path: str) -> Module:
+    parts = path.split('.')
+    module_path = '.'.join(parts[:-1])
+    class_name = parts[-1]
+    module = import_module(module_path)
+    return getattr(module, class_name)

model/src/utils/get_model_name.py

@@ -0,0 +1,19 @@
+"""Definition of get_model_name function."""
+
+import logging
+from pathlib import Path
+
+
+def get_model_name(
+    path: Path,
+) -> str:
+    """Get model name from model_name file."""
+    assert isinstance(path, Path)
+    assert path.exists(), f'{path} does not exist'
+    # read file contents
+    with open(file=path, encoding='utf-8') as fh:
+        model_name = fh.read()
+    # remove newline character
+    model_name = model_name.strip()
+    logging.debug('finished')
+    return model_name

model/src/utils/load_model.py

@@ -0,0 +1,37 @@
+"""Definition of load_model function."""
+
+import logging
+from pathlib import Path
+
+import torch
+
+from model.src.models import VisualCommunicationModel
+from shared.repositories import ModelRepository
+
+from .get_model_name import get_model_name
+
+DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+
+def load_model() -> VisualCommunicationModel:
+    """Instantiate model with weights loaded from latest model saved in
+    MinIO."""
+    # instantiate model
+    model = VisualCommunicationModel()
+    # get model object name
+    model_name_path = Path('model_name.txt')
+    model_object_name = get_model_name(path=model_name_path)
+    logging.info('using model: %s', model_object_name)
+    # load model data
+    with ModelRepository() as repo:
+        model_data = repo.get_data(model_object_name)
+    if model_data is None:
+        raise FileNotFoundError(f'model {model_object_name} not found')
+    model_checkpoint = torch.load(model_data.buffer)
+    model.load_state_dict(model_checkpoint)
+    # clean memory
+    model_checkpoint.clear()
+    # move model to selected device
+    model = model.to(DEVICE)
+    logging.debug('finished')
+    return model

model/src/utils/train.py

@@ -6,7 +6,7 @@ import torch
 import torch.nn as nn
 from dataloader import VCDADataset  # noqa: F401
 
-from .models import VisualCommunicationModel
+from ..models import VisualCommunicationModel
 
 PRE_WARMUP_LR = 1e-10
 POST_WARMUP_LR = 1e-5

model/src/utils/vcda_dataset.py

@@ -0,0 +1,134 @@
+"""Definition of Visual Critial Discourse Analysis dataloader."""
+
+import random
+
+from PIL import Image
+from torch import Tensor
+from torch.utils.data import Dataset
+from torchvision.transforms import ColorJitter, InterpolationMode, Normalize
+from torchvision.transforms.functional import (
+    hflip,
+    pad,
+    resize,
+    rotate,
+    to_pil_image,
+    to_tensor,
+)
+
+from shared.repositories import ImageRepository
+
+# resnet18 original normalization values
+RESNET_NORMALIZE_MEAN = [0.485, 0.456, 0.406]
+RESNET_NORMALIZE_STD = [0.229, 0.224, 0.225]
+
+
+class VCDADataset(Dataset):
+    """VCDA dataset class."""
+
+    def __init__(
+        self,
+        data_name_list: list[str],
+        do_augment: bool = False,
+        random_annotations: bool = False,
+    ):
+        super().__init__()
+        self.data_name_list = data_name_list
+        self.do_augment = do_augment
+        self.random_annotations = random_annotations
+        self.normalize_mean = RESNET_NORMALIZE_MEAN
+        self.normalize_std = RESNET_NORMALIZE_STD
+        # prepare augmentation functions
+        self.normalize = Normalize(
+            mean=self.normalize_mean,
+            std=self.normalize_std,
+        )
+        self.color_jitter = ColorJitter(
+            brightness=1e-1,
+            contrast=8e-2,
+            saturation=8e-2,
+        )
+
+    def __len__(self):
+        return len(self.data_name_list)
+
+    def __getitem__(self, idx):
+        # get image from database
+        image_name = self.data_name_list[idx]
+        with ImageRepository() as repo:
+            image_data = repo.get_data(image_name)
+        assert image_data is not None
+        tensor = self.image_to_tensor(image_data.image)
+        if self.do_augment:
+            tensor = self.augment(tensor)
+        return tensor
+
+    def image_to_tensor(
+        self,
+        image: Image.Image,
+    ) -> Tensor:
+        """Load images tensor from bytes."""
+        tensor = to_tensor(image)
+        tensor /= 255  # normalize 8-bit image
+        tensor = self.square_pad(tensor=tensor)
+        tensor = resize(
+            img=tensor,
+            size=(512, 512),
+            interpolation=InterpolationMode.BICUBIC,
+        )
+        return tensor
+
+    @staticmethod
+    def square_pad(
+        tensor: Tensor,
+    ) -> Tensor:
+        """Pads image to a square with side length equal to the largest side of
+        the input image."""
+        assert isinstance(tensor, Tensor)
+        # B, nc, w, h = img.shape
+        h = tensor.shape[-2]
+        w = tensor.shape[-1]
+        if h == w:
+            return tensor
+        max_wh = max([h, w])
+        hp = int((max_wh - w) / 2)
+        vp = int((max_wh - h) / 2)
+        padding = (hp, vp, hp, vp)
+        tensor = pad(tensor, padding, 0, 'constant')
+        return tensor
+
+    def augment(
+        self,
+        tensor: Tensor,
+    ) -> Tensor:
+        """Augment image with random horizontal flips, rotations and color
+        jitter."""
+        assert isinstance(tensor, Tensor)
+        # left-right flip
+        if random.random() >= 0.5:
+            tensor = hflip(tensor)
+        # rotation
+        rnd = random.random()
+        if rnd < 0.25:
+            tensor = rotate(tensor, angle=90)
+        if rnd < 0.5:
+            tensor = rotate(tensor, angle=180)
+        if rnd < 0.75:
+            tensor = rotate(tensor, angle=270)
+        # color jitter
+        tensor = self.color_jitter(tensor)
+        return tensor
+
+    def reverse_normalise(
+        self,
+        tensor: Tensor,
+    ) -> Image.Image:
+        """Reverse normalization to get an image that can be interpreted by
+        humans."""
+        assert isinstance(tensor, Tensor)
+        tensor *= Tensor(self.normalize_std).reshape((3, 1, 1))
+        tensor += Tensor(self.normalize_mean).reshape((3, 1, 1))
+        image = to_pil_image(
+            pic=tensor,
+            mode='RGB',
+        )
+        return image

model/train.py

@@ -0,0 +1,163 @@
+import json
+import os
+from argparse import ArgumentParser
+from datetime import datetime, timezone
+from pathlib import Path
+
+import torch
+from dotenv import load_dotenv
+from ignite.engine import Events, create_supervised_evaluator, create_supervised_trainer
+from ignite.handlers import global_step_from_engine
+from ignite.handlers.checkpoint import Checkpoint
+from ignite.handlers.param_scheduler import create_lr_scheduler_with_warmup
+from ignite.handlers.tensorboard_logger import TensorboardLogger
+from ignite.handlers.tqdm_logger import ProgressBar
+from ignite.metrics import Average, Loss, RunningAverage
+from torch import nn
+from torch.optim.lr_scheduler import ExponentialLR
+from torch.utils.data import DataLoader
+
+from model.src.utils import VCDADataset, get_class
+
+
+def parse_arguments():
+    parser = ArgumentParser()
+    parser.add_argument('-c', '--config', type=Path, required=True)
+    parser.add_argument('-d', '--device', type=str, default='cpu')
+    parser.add_argument('-b', '--batch-size', type=int, default=32)
+    parser.add_argument('--epochs', type=int, default=50)
+    parser.add_argument('--epoch-length', type=int, default=128)
+    parser.add_argument('--checkpoint', type=Path)
+    parser.add_argument('--lr', type=float, default=1e-5)
+    parser.add_argument('--lr-gamma', type=float, default=0.95)
+    parser.add_argument('--lr-warmup-start', type=float, default=1e-8)
+    parser.add_argument('--lr-warmup-duration', type=int, default=5)
+    parser.add_argument('--train-seed', type=int, default=1312)
+    parser.add_argument('--val-seed', type=int, default=1313)
+    parser.add_argument('--loader-workers', type=int, default=8)
+    return parser.parse_args()
+
+
+args = parse_arguments()
+
+load_dotenv('server.env')
+
+# read model config
+with open(args.config) as fh:
+    config = json.load(fh)
+model_name = args.config.stem
+run_name = datetime.now(timezone.utc).strftime('%Y-%m-%d-%H%M') + '_' + model_name
+device = torch.device(args.device)
+
+# create model
+model_class = get_class(config['backbone']['class'])
+model = model_class().to(device)
+optimizer = torch.optim.Adam(model.parameters(), lr=args.lr)
+loss_fn = nn.CrossEntropyLoss()
+
+# create datasets and loaders
+with open('model/src/dataset/train.csv', encoding='utf-8') as fh:
+    train_data_name_list = fh.read().split('\n')
+train_dataset = VCDADataset(
+    data_name_list=train_data_name_list,
+)
+train_loader = DataLoader(dataset=train_dataset, num_workers=args.loader_workers)
+
+with open('model/src/dataset/val.csv', encoding='utf-8') as fh:
+    val_data_name_list = fh.read().split('\n')
+val_dataset = VCDADataset(data_name_list=val_data_name_list)
+val_loader = DataLoader(dataset=val_dataset, num_workers=args.loader_workers)
+
+# create trainer and evaluator
+trainer = create_supervised_trainer(
+    model=model,
+    optimizer=optimizer,
+    loss_fn=loss_fn,
+    device=device,
+)
+Average(output_transform=lambda x: x).attach(trainer, name='loss')
+RunningAverage(output_transform=lambda x: x, alpha=0.5).attach(
+    trainer,
+    'running_avg_loss',
+)
+ProgressBar(desc='Train', ncols=80).attach(trainer, ['running_avg_loss'])
+
+val_metrics = {
+    'loss': Loss(loss_fn=loss_fn, device=device),
+}
+evaluator = create_supervised_evaluator(
+    model=model,
+    metrics=val_metrics,  # type: ignore
+    device=device,
+)
+ProgressBar(desc='Val', ncols=80).attach(evaluator)
+
+# create lr scheduler
+lr_scheduler = ExponentialLR(
+    optimizer=optimizer,
+    gamma=args.lr_gamma,
+)
+lr_handler = create_lr_scheduler_with_warmup(
+    lr_scheduler=lr_scheduler,
+    warmup_start_value=args.lr_warmup_start,
+    warmup_duration=args.lr_warmup_duration,
+    warmup_end_value=args.lr,
+)
+trainer.add_event_handler(Events.EPOCH_STARTED, lr_handler)
+
+
+# log metrics to TensorBoard
+@trainer.on(Events.EPOCH_COMPLETED)
+def evaluate():
+    evaluator.run(val_loader, epoch_length=args.val_epoch_length)
+
+
+tb_logger = TensorboardLogger(log_dir=f'runs/logs/{run_name}')
+tb_logger.attach_opt_params_handler(
+    engine=trainer,
+    event_name=Events.EPOCH_COMPLETED,
+    optimizer=optimizer,
+)
+for tag, engine in [('train', trainer), ('val', evaluator)]:
+    tb_logger.attach_output_handler(
+        engine,
+        event_name=Events.EPOCH_COMPLETED,
+        tag=tag,
+        metric_names='all',
+        global_step_transform=global_step_from_engine(trainer),
+    )
+
+# Set up checkpoint saving
+to_save = {
+    'model': model,
+    'optimizer': optimizer,
+    'trainer': trainer,
+    'lr_scheduler': lr_scheduler,
+    'lr_handler': lr_handler,
+}
+checkpoint_handler = Checkpoint(
+    to_save,
+    f'runs/checkpoints/{run_name}',
+    n_saved=3,
+    filename_prefix='best',
+    score_function=lambda engine: -engine.state.metrics['loss'],
+    score_name='neg_val_loss',
+    global_step_transform=global_step_from_engine(trainer),
+)
+evaluator.add_event_handler(Events.COMPLETED, checkpoint_handler)
+
+if args.checkpoint:
+    Checkpoint.load_objects(to_load=to_save, checkpoint=str(args.checkpoint))
+
+# save model config
+os.makedirs('runs/configs/', exist_ok=True)
+with open(f'runs/configs/{run_name}.json', 'w', encoding='utf-8') as fh:
+    json.dump(config, fh)
+
+# start training
+trainer.run(
+    train_loader,
+    max_epochs=args.epochs,
+    epoch_length=args.epoch_length,
+)
+tb_logger.close()

other/transfer_images_mongo_minio.py

@@ -0,0 +1,124 @@
+"""Script to move all images from MongoDB to MinIO."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from bson import ObjectId
+from dotenv import load_dotenv
+from pymongo.collection import Collection
+
+from shared.datastore import Datastore
+from shared.docstore import connect_mongodb
+from shared.docstore.src.classes import VisualCommunication
+from shared.utils import check_env, setup_logging
+from web_ui.src.main import NECESSARY_ENV_VAR_LIST
+
+
+def list_mongo_document_ids(
+    collection: Collection,
+) -> list[ObjectId]:
+    """Get list of all VisualCommunication documents in MongoDB."""
+    # prepare query
+    query = collection.find(
+        filter={},
+        projection={'_id': True},
+    )
+    # execute query
+    res_list = list(query)
+    logging.debug('got %s document(s)', len(res_list))
+    # convert result
+    id_list = [elem['_id'] for elem in res_list]
+    return id_list
+
+
+def get_visual_communication(
+    collection: Collection,
+    doc_id: ObjectId,
+) -> VisualCommunication:
+    """Get image from MongoDB."""
+    # prepare query
+    query = collection.find(
+        filter={'_id': doc_id},
+        projection={'_id': False},
+    )
+    # execute query
+    res_list = list(query)
+    logging.debug('got %s document(s)', len(res_list))
+    # convert result
+    vis_com = VisualCommunication.model_validate(res_list[0])
+    return vis_com
+
+
+def update_visual_communication(
+    collection: Collection,
+    vis_com: VisualCommunication,
+) -> None:
+    """Update VisualCommunication document in MongoDB."""
+    try:
+        collection.update_one(
+            filter={
+                'name': vis_com.name,
+            },
+            update={
+                '$set': vis_com.model_dump(),
+            },
+        )
+    except Exception as exc:
+        logging.error('failed saving %s', vis_com.name)
+        logging.debug(exc)
+        raise exc
+
+
+if __name__ == '__main__':
+    # load in env file
+    env_path = Path(__file__).parent.parent / 'server.env'
+    assert env_path.exists()
+    load_dotenv(env_path)
+    # ensure env vars set
+    check_env(NECESSARY_ENV_VAR_LIST)
+    # setup logging
+    setup_logging()
+    # connect to minIO
+    datastore = Datastore()
+    datastore.connect()
+    assert datastore._client is not None
+    # connect to MongoDB
+    collection, db, client = connect_mongodb()
+    # list documents in mongoDB
+    id_list = list_mongo_document_ids(collection)
+    for doc_id in id_list:
+        try:
+            # get image from MongoDB
+            vis_com = get_visual_communication(collection, doc_id)
+        except Exception as exc:
+            logging.debug(exc)
+            logging.error('failed getting image from document: %s', doc_id)
+            continue
+        try:
+            # get image
+            image = vis_com.get_image(
+                minio_client=datastore._client,
+            )
+            # put buffer in minio
+            object_name = datastore.put_image(
+                image=image,
+            )
+        except Exception as exc:
+            logging.debug(exc)
+            logging.error('failed saving image to minio')
+            continue
+        # update visual communication in mongodb
+        try:
+            vis_com.image = None  # type: ignore
+            vis_com.object_name = object_name
+            update_visual_communication(collection, vis_com)
+        except Exception as exc:
+            logging.debug(exc)
+            logging.error(
+                'failed updating visual communication %s',
+                vis_com.name,
+            )
+            continue
+        logging.debug('updated visual communication %s', vis_com.name)

pyproject.toml

@@ -20,6 +20,11 @@ types-requests = "^2.32.0.20240602"
 types-retry = "^0.9.9.4"
 flake8-pyproject = "^1.2.3"
 pandas-stubs = "^2.2.2.240603"
+types-tqdm = "^4.66.0.20240417"
+pytest = "^8.3.3"
+testcontainers = "^4.8.2"
+coverage = "^7.6.4"
+pytest-cov = "^6.0.0"
 
 
 [tool.poetry.group.dev.dependencies]
@@ -27,13 +32,17 @@ pandas = "^2.2.1"
 selenium = "^4.18.1"
 webdriver-manager = "^4.0.1"
 retry = "^0.9.2"
+pre-commit = "^4.1.0"
 
 
 [tool.poetry.group.model.dependencies]
-torch = "^2.2.1"
+torch = "^2.0.0"
 torchvision = "^0.17.1"
 torchinfo = "^1.8.0"
 minio = "^7.2.7"
+tqdm = "^4.66.4"
+pytorch-ignite = "^0.5.1"
+tensorboard = "^2.17.1"
 
 
 [tool.poetry.group.shared.dependencies]
@@ -59,8 +68,13 @@ priority = "primary"
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
 
+[tool.isort]
+profile = "black"
+
 [tool.flake8]
 per-file-ignores = "__init__.py:F401"
+max-line-length = 88
+extend-ignore = "E203"
 
 [tool.mypy]
 exclude = "image_download"
@@ -98,9 +112,13 @@ module = "utils.*"
 ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
-module = "data_store.*"
+module = "shared.datastore.*"
 ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
 module = "models.*"
 ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "shared.docstore.*"
+ignore_missing_imports = true

shared/database/__init__.py

@@ -1,13 +0,0 @@
-"""Database module content."""
-from __future__ import annotations
-
-from .classes.dataset import Dataset
-from .classes.exceptions import NoDocumentFoundException
-from .classes.visual_communication import VisualCommunication
-from .utils.connect 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

shared/database/classes/__init__.py

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

shared/database/classes/dataset.py

@@ -1,71 +0,0 @@
-"""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)

shared/database/classes/exceptions.py

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

shared/database/classes/visual_communication.py

@@ -1,85 +0,0 @@
-"""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 shared.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()

shared/database/utils/__init__.py

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

shared/database/utils/connect.py

@@ -1,33 +0,0 @@
-"""
-Definition of function to connect to database
-using environment variables.
-"""
-from __future__ import annotations
-
-import logging
-import os
-
-from dotenv import load_dotenv
-from pymongo import MongoClient
-
-
-def connect():
-    """Connect to MongoDB using env vars."""
-    # load env vars
-    load_dotenv()
-    necessary_env_vars = [
-        'MONGO_HOST',
-        'MONGO_DB',
-        'MONGO_COLLECTION',
-    ]
-    for env_var in necessary_env_vars:
-        assert env_var in os.environ, f"{env_var} not found"
-    # 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.debug('finished')
-    return collection, db, client

shared/database/utils/count_documents.py

@@ -1,21 +0,0 @@
-"""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)

shared/database/utils/get_dataset.py

@@ -1,13 +0,0 @@
-from __future__ import annotations
-
-from typing import Literal
-
-from pymongo.collection import Collection
-
-
-def get_dataset(
-    collection: Collection,
-    type: Literal['train', 'test', 'validation'],
-) -> list[str]:
-    """Get list of data names for the corresponding type."""
-    return []

shared/database/utils/get_visual_communication.py

@@ -1,39 +0,0 @@
-"""Definition of function to get visual communication from database."""
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from shared.database import NoDocumentFoundException
-from shared.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

shared/database/utils/list_names.py

@@ -1,31 +0,0 @@
-"""
-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

shared/database/utils/save_dataset.py

@@ -1,36 +0,0 @@
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from shared.database import Dataset
-
-
-def save_dataset(
-    collection: Collection,
-    dataset: Dataset,
-) -> None:
-    """Save dataset to database."""
-    res = collection.insert_one(
-        document=dataset.model_dump(),
-    )
-    logging.debug('inserted document: %s', res)
-
-
-if __name__ == '__main__':
-    from dotenv import load_dotenv
-    load_dotenv('local.env')
-    from shared.database import list_names, connect
-    # connect to database
-    collection, db, client = connect()
-    print(client.server_info())
-
-    name_list = list_names(collection=collection, only_with_annotation=True)
-    ds = Dataset.new_from_name_list(name_list=name_list)
-
-    print(ds)
-    # save_dataset(
-    #     collection=collection,
-    #     dataset=ds
-    # )

shared/database/utils/upsert_annotation.py

@@ -1,30 +0,0 @@
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from shared.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')

shared/database/utils/upsert_prediction.py

@@ -1,30 +0,0 @@
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from shared.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')

shared/database/utils/upsert_visual_communication.py

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

shared/dto/__init__.py

@@ -1,15 +0,0 @@
-"""Data transfer objects module content."""
-from __future__ import annotations
-
-from .angle import AngleData
-from .contact import ContactData
-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 .model_data import ModelData
-from .point_of_view import PointOfViewData
-from .salience import SalienceData
-from .visual_syntax import VisualSyntaxData

shared/dto/angle.py

@@ -1,11 +0,0 @@
-"""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

shared/dto/contact.py

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

shared/dto/data_model.py

@@ -1,67 +0,0 @@
-"""Definition of DataModel base class."""
-from __future__ import annotations
-
-import random
-
-from pydantic import BaseModel
-from pydantic import ValidationError
-
-
-class DataModel(BaseModel):
-    """DataModel base class."""
-
-    @classmethod
-    def classname(cls) -> str:
-        """Return classname."""
-        return cls.__name__
-
-    @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):
-        """Instantiate with random numbers."""
-        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)
-
-    @classmethod
-    def from_list(cls, data_list: list[float]):
-        """Instantiate from list of values."""
-        kwargs = {key: val for key, val in zip(cls.list_fields(), data_list)}
-        return cls(**kwargs)
-
-    def __repr__(self) -> str:
-        model_dict = self.model_dump()
-        model_repr_str = f"{self.classname()}("
-        model_repr_str += ', '.join([
-            f"{field}={value:.3f}"
-            for field, value
-            in model_dict.items()
-        ])
-        model_repr_str += ')'
-        return model_repr_str
-
-    def highest_score_field(self) -> str:
-        """Return name of field with highest score."""
-        model_dict = self.model_dump()
-        return max(model_dict, key=lambda k: model_dict[k])
-
-    def highest_score_value(self) -> float:
-        """Return value of field with highest score."""
-        model_dict = self.model_dump()
-        return max(model_dict.values())

shared/dto/distance.py

@@ -1,12 +0,0 @@
-"""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

shared/dto/framing.py

@@ -1,13 +0,0 @@
-"""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
-    form_contrast: float

shared/dto/information_value.py

@@ -1,12 +0,0 @@
-"""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

shared/dto/modality_color.py

@@ -1,12 +0,0 @@
-"""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

shared/dto/modality_depth.py

@@ -1,12 +0,0 @@
-"""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

shared/dto/modality_lighting.py

@@ -1,12 +0,0 @@
-"""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

shared/dto/model_data.py

@@ -1,82 +0,0 @@
-"""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)

shared/dto/point_of_view.py

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

shared/dto/salience.py

@@ -1,14 +0,0 @@
-"""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
-    form: float
-    positioning: float

shared/repositories/__init__.py

@@ -0,0 +1,8 @@
+from .src import ImageRepository, ModelRepository, VisualCommunicationRepository
+from .src.dto import (
+    HexadecimalString,
+    ImageData,
+    ModelData,
+    VisualCommunicationData,
+    VisualCommunicationValues,
+)

shared/repositories/src/__init__.py

@@ -0,0 +1,3 @@
+from .image_repository import ImageRepository
+from .model_repository import ModelRepository
+from .visual_communication_repository import VisualCommunicationRepository

shared/repositories/src/dto/__init__.py

@@ -0,0 +1,7 @@
+from .hexadecimal_string import HexadecimalString
+from .image_data import ImageData
+from .model_data import ModelData
+from .visual_communication_data import (
+    VisualCommunicationData,
+    VisualCommunicationValues,
+)

shared/repositories/src/dto/bytes_io_pydantic_annotation.py

@@ -0,0 +1,37 @@
+"""Definition of BytesIO Pydantic Annotation."""
+
+from io import BytesIO
+from typing import Any
+
+from pydantic.json_schema import JsonSchemaValue
+from pydantic_core import core_schema
+
+
+class BytesIOPydanticAnnotation:
+    """Pydantic annotation that defines input validation, as well as general
+    and json serialization."""
+
+    @classmethod
+    def validate_input(cls, v: Any, handler) -> BytesIO:
+        """Pydantic-related function to validate input on instantiation."""
+        if isinstance(v, BytesIO):
+            return v
+        s = handler(v)
+        return BytesIO(s)
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls,
+        source_type,
+        _handler,
+    ) -> core_schema.CoreSchema:
+        assert source_type is BytesIO
+        return core_schema.no_info_wrap_validator_function(
+            function=cls.validate_input,
+            schema=core_schema.str_schema(),
+            serialization=core_schema.to_string_ser_schema(),
+        )
+
+    @classmethod
+    def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
+        return handler(core_schema.str_schema())

shared/repositories/src/dto/hexadecimal_string.py

@@ -0,0 +1,24 @@
+"""Definition of Checksum DTO."""
+
+from __future__ import annotations
+
+import re
+
+
+class HexadecimalString(str):
+    """Hexadecimal-string class."""
+
+    def __new__(cls, string):
+        # ensure proper input format
+        pattern = r'[0-9-a-fA-F]{32}'
+        match = re.match(pattern, string)
+        if match is None:
+            raise ValueError(f'format does not match a hexadecimal-string: {string}')
+        return super().__new__(cls, string)
+
+    def __repr__(self) -> str:
+        class_name = self.__class__.__name__
+        return f"{class_name}('{self}')"
+
+    def __reduce__(self):
+        return self.__class__, (self,)

shared/repositories/src/dto/hexadecimal_string_pydantic_annotation.py

@@ -0,0 +1,38 @@
+"""Definition of HexadecimalString Pydantic Annotation."""
+
+from typing import Any
+
+from pydantic.json_schema import JsonSchemaValue
+from pydantic_core import core_schema
+
+from .hexadecimal_string import HexadecimalString
+
+
+class HexadecimalStringPydanticAnnotation:
+    """Pydantic annotation that defines input validation, as well as general
+    and json serialization."""
+
+    @classmethod
+    def validate_input(cls, v: Any, handler) -> HexadecimalString:
+        """Pydantic-related function to validate input on instantiation."""
+        if isinstance(v, HexadecimalString):
+            return v
+        s = handler(v)
+        return HexadecimalString(s)
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls,
+        source_type,
+        _handler,
+    ) -> core_schema.CoreSchema:
+        assert source_type is HexadecimalString
+        return core_schema.no_info_wrap_validator_function(
+            function=cls.validate_input,
+            schema=core_schema.str_schema(),
+            serialization=core_schema.to_string_ser_schema(),
+        )
+
+    @classmethod
+    def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
+        return handler(core_schema.str_schema())

shared/repositories/src/dto/image_data.py

@@ -0,0 +1,16 @@
+"""Definition of VisualData DTO."""
+
+from typing import Annotated
+
+from PIL import Image
+from pydantic import Field
+
+from .image_pydantic_annotation import ImagePydanticAnnotation
+from .type_checking_base_model import TypeCheckingBaseModel
+
+
+class ImageData(TypeCheckingBaseModel):
+    """Visual data class."""
+
+    image: Annotated[Image.Image, ImagePydanticAnnotation]
+    name: str = Field(min_length=1)

shared/repositories/src/dto/image_pydantic_annotation.py

@@ -0,0 +1,37 @@
+"""Definition of BytesIO Pydantic Annotation."""
+
+from typing import Any
+
+from PIL import Image
+from pydantic.json_schema import JsonSchemaValue
+from pydantic_core import core_schema
+
+
+class ImagePydanticAnnotation:
+    """Pydantic annotation that defines input validation, as well as general
+    and json serialization."""
+
+    @classmethod
+    def validate_input(cls, v: Any, handler) -> Image.Image:
+        """Pydantic-related function to validate input on instantiation."""
+        if isinstance(v, Image.Image):
+            return v
+        s = handler(v)
+        return Image.open(s)
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls,
+        source_type,
+        _handler,
+    ) -> core_schema.CoreSchema:
+        assert source_type is Image.Image
+        return core_schema.no_info_wrap_validator_function(
+            function=cls.validate_input,
+            schema=core_schema.str_schema(),
+            serialization=core_schema.to_string_ser_schema(),
+        )
+
+    @classmethod
+    def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
+        return handler(core_schema.str_schema())

shared/repositories/src/dto/model_data.py

@@ -0,0 +1,60 @@
+"""Definition of ModelData DTO."""
+
+from __future__ import annotations
+
+from hashlib import md5
+from io import BytesIO
+from typing import Annotated
+
+import torch
+from pydantic import Field
+
+from .bytes_io_pydantic_annotation import BytesIOPydanticAnnotation
+from .hexadecimal_string import HexadecimalString
+from .hexadecimal_string_pydantic_annotation import HexadecimalStringPydanticAnnotation
+from .type_checking_base_model import TypeCheckingBaseModel
+
+
+class ModelData(TypeCheckingBaseModel):
+    """Model Data DTO."""
+
+    buffer: Annotated[BytesIO, BytesIOPydanticAnnotation]
+    buffer_checksum: Annotated[HexadecimalString, HexadecimalStringPydanticAnnotation]
+    class_name: str = Field(
+        min_length=1,
+        description='name of model class to generate data.',
+    )
+
+    @staticmethod
+    def calculate_checksum(buffer: BytesIO) -> HexadecimalString:
+        """Calculate buffer checksum."""
+        checksum = md5(buffer.getbuffer()).hexdigest()
+        return HexadecimalString(checksum)
+
+    @staticmethod
+    def model_to_buffer(model: torch.nn.Module) -> BytesIO:
+        """Save model to buffer."""
+        assert isinstance(model, torch.nn.Module)
+        buffer = BytesIO()
+        torch.save(model.state_dict(), buffer)
+        return buffer
+
+    @classmethod
+    def from_model(cls, model: torch.nn.Module) -> ModelData:
+        """Instantiate from torch module."""
+        assert isinstance(model, torch.nn.Module)
+        # get model name
+        class_name = type(model).__name__
+        # save data to buffer
+        buffer = cls.model_to_buffer(model)
+        buffer = BytesIO()
+        torch.save(model.state_dict(), buffer)
+        # calculate checksum
+        buffer_checksum = cls.calculate_checksum(buffer)
+        # instantiate from buffer
+        data = cls(
+            buffer=buffer,
+            buffer_checksum=buffer_checksum,
+            class_name=class_name,
+        )
+        return data

shared/repositories/src/dto/type_checking_base_model.py

@@ -0,0 +1,11 @@
+"""Definition of TypeCheckingBaseModel class."""
+
+from pydantic import BaseModel, ConfigDict
+
+
+class TypeCheckingBaseModel(BaseModel):
+    """BaseModel with added type checking on input types."""
+
+    model_config = ConfigDict(
+        frozen=True,  # ensure data immutability
+    )

shared/repositories/src/dto/visual_communication_data/__init__.py

@@ -0,0 +1,2 @@
+from .visual_communication_data import VisualCommunicationData
+from .visual_communication_values import VisualCommunicationValues

shared/repositories/src/dto/visual_communication_data/angle_values.py

@@ -0,0 +1,11 @@
+"""Definition of AngleValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class AngleValues(ValuesModel):
+    """Angle values DTO."""
+
+    high: float
+    eye_level: float
+    low: float

shared/repositories/src/dto/visual_communication_data/contact_values.py

@@ -0,0 +1,10 @@
+"""Definition of ContactValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class ContactValues(ValuesModel):
+    """Contact values DTO."""
+
+    offer: float
+    demand: float

shared/repositories/src/dto/visual_communication_data/distance_values.py

@@ -0,0 +1,11 @@
+"""Definition of DistanceValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class DistanceValues(ValuesModel):
+    """Distance values DTO."""
+
+    long: float
+    medium: float
+    close: float

shared/repositories/src/dto/visual_communication_data/framing_values.py

@@ -0,0 +1,12 @@
+"""Definition of FramingValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class FramingValues(ValuesModel):
+    """Framing values DTO."""
+
+    frame_lines: float
+    empty_space: float
+    colour_contrast: float
+    form_contrast: float

shared/repositories/src/dto/visual_communication_data/information_value_values.py

@@ -0,0 +1,11 @@
+"""Definition of InformationValueValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class InformationValueValues(ValuesModel):
+    """Information value values DTO."""
+
+    given_new: float
+    ideal_real: float
+    central_marginal: float

shared/repositories/src/dto/visual_communication_data/modality_color_values.py

@@ -0,0 +1,11 @@
+"""Definition of ModalityColorValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class ModalityColorValues(ValuesModel):
+    """Modality color values DTO."""
+
+    high: float
+    medium: float
+    low: float

shared/repositories/src/dto/visual_communication_data/modality_depth_values.py

@@ -0,0 +1,11 @@
+"""Definition of ModalityDepthValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class ModalityDepthValues(ValuesModel):
+    """Modality depth values DTO."""
+
+    high: float
+    medium: float
+    low: float

shared/repositories/src/dto/visual_communication_data/modality_lighting_values.py

@@ -0,0 +1,11 @@
+"""Definition of ModalityLightingValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class ModalityLightingValues(ValuesModel):
+    """Modality lighting values DTO."""
+
+    high: float
+    medium: float
+    low: float

shared/repositories/src/dto/visual_communication_data/point_of_view_values.py

@@ -0,0 +1,10 @@
+"""Definition of PointOfViewValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class PointOfViewValues(ValuesModel):
+    """Point-of-view values DTO."""
+
+    frontal: float
+    oblique: float

shared/repositories/src/dto/visual_communication_data/salience_values.py

@@ -0,0 +1,13 @@
+"""Definition of SalienceValues DTO."""
+
+from .values_model import ValuesModel
+
+
+class SalienceValues(ValuesModel):
+    """Salience values DTO."""
+
+    size: float
+    colour: float
+    tone: float
+    form: float
+    positioning: float

shared/repositories/src/dto/visual_communication_data/values_model.py

@@ -0,0 +1,60 @@
+"""Definition of ValuesModel base class."""
+
+from __future__ import annotations
+
+import random
+
+from pydantic import BaseModel, ConfigDict
+from torch import Tensor
+
+
+class ValuesModel(BaseModel):
+    """ValuesModel base class."""
+
+    model_config = ConfigDict(
+        validate_assignment=True,  # argument type checking
+        frozen=True,  # ensure data immutability
+    )
+
+    @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):
+        """Instantiate with random numbers."""
+        kwargs = {field: random.random() for field in cls.list_fields()}
+        return cls(**kwargs)
+
+    @classmethod
+    def from_choice(cls, option: str) -> ValuesModel:
+        """Instantiate from choice."""
+        assert isinstance(option, str)
+        assert len(option) > 0
+        allowed_options_list = cls.list_fields()
+        if option not in allowed_options_list:
+            raise ValueError(f'option {option} must be in {allowed_options_list}')
+        # generate field values
+        kwargs = {field: 0 for field in allowed_options_list}
+        # set chosen value to max probability
+        kwargs[option] = 1
+        return cls(**kwargs)
+
+    @classmethod
+    def from_tensor(cls, tensor: Tensor):
+        """Instantiate from list of values."""
+        assert tensor.size(dim=0) == 1, f'tensor batch larger than 1: {tensor}'
+        data_list = [float(t.item()) for t in tensor[0]]
+        kwargs = dict(zip(cls.list_fields(), data_list))
+        return cls(**kwargs)
+
+    def highest_score_field(self) -> str:
+        """Return name of field with highest score."""
+        model_dict = self.model_dump()
+        return max(model_dict, key=lambda k: model_dict[k])
+
+    def highest_score_value(self) -> float:
+        """Return value of field with highest score."""
+        model_dict = self.model_dump()
+        return max(model_dict.values())

shared/repositories/src/dto/visual_communication_data/visual_communication_data.py

@@ -0,0 +1,14 @@
+"""Definition of VisualCommunicationData DTO."""
+
+from pydantic import Field
+
+from ..type_checking_base_model import TypeCheckingBaseModel
+from .visual_communication_values import VisualCommunicationValues
+
+
+class VisualCommunicationData(TypeCheckingBaseModel):
+    """Visual communication data class."""
+
+    name: str = Field(min_length=1)
+    annotation: VisualCommunicationValues | None = None
+    prediction: VisualCommunicationValues | None = None