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

Reviewed-on: #69

.gitea/workflows/pull.yaml

@@ -9,28 +9,42 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout Code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
       - name: Setup Environment
-        uses: https://github.com/actions/setup-python@v3
+        uses: https://github.com/actions/setup-python@v5
         with:
-          python-verison: "3.12"
+          python-version: "3.12"
           architecture: "x64"
-      - name: Install Packages
+      - name: Setup poetry
         env:
-          PIP_INDEX_URL: http://192.168.1.2:5001/index/
+          POETRY_VERSION: 2.1.1
-          PIP_TRUSTED_HOST: 192.168.1.2
+          POETRY_HOME: /opt/poetry
+          POETRY_NO_INTERACTION: 1
+          POETRY_NO_CACHE: 1
         run: |
-          pip install poetry
+          curl -sSL https://install.python-poetry.org | python3 -
-          poetry install
+          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: |
-          poetry run flake8 . --benchmark
+          /opt/poetry/bin/poetry run flake8 . --benchmark
       - name: Type Check
         run: |
-          poetry run mypy .
+          /opt/poetry/bin/poetry run mypy .
-      - name: Pytest
+      - 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: |
-          poetry run pytest .
+          /opt/poetry/bin/poetry run coverage run -m pytest .
       - name: Coverage Report
         run: |
-          poetry run coverage report -m
+          /opt/poetry/bin/poetry run coverage report -m

.pre-commit-config.yaml

@@ -1,15 +1,16 @@
 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
 -   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/pre-commit/mirrors-isort
@@ -23,7 +24,7 @@ repos:
     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]
@@ -35,12 +36,12 @@ repos:
         additional_dependencies:
           - "pyproject-flake8"
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.8.0
+    rev: v1.15.0
     hooks:
     -   id: mypy
         exclude: ^testing/resources/
 -   repo: https://github.com/psf/black
-    rev: 24.4.2
+    rev: 25.1.0
     hooks:
     - id: black
       language_version: python3.12
@@ -55,7 +56,7 @@ repos:
         language: python
         types: [ python ]
 -     repo: https://github.com/jendrikseipp/vulture
-      rev: 'v2.6'
+      rev: 'v2.14'
       hooks:
       - id: vulture
         entry: vulture . --min-confidence 90 --exclude */.venv/*.py,*/tests/*.py

misc/generate_random_prediction.py

@@ -1,40 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-
-from shared.datastore import connect_minio
-from shared.mongodb import connect_mongodb
-from shared.mongodb.classes import VisualCommunication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-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
-    minio_client = connect_minio()
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # get list of image paths
-    test_dir = Path(__file__).parent
-    img_dir = test_dir / 'imgs'
-    img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
-    print(img_path_list)
-    # instantiate data object
-    vis_com_list = [
-        VisualCommunication.from_file(path, minio_client=minio_client)
-        for path in img_path_list
-    ]
-    # generate random predictions
-    for vis_com in vis_com_list:
-        vis_com.generate_random_prediction()
-    for vis_com in vis_com_list:
-        print(vis_com)

misc/get_visual_communication.py

@@ -1,29 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-
-from shared.datastore import connect_minio
-from shared.mongodb import connect_mongodb, get_visual_communication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-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
-    minio_client = connect_minio()
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # get visual communication
-    vis_com = get_visual_communication(collection)
-    print(repr(vis_com))
-    image = vis_com.get_image(minio_client=minio_client)
-    image.show()

misc/image_download.py

@@ -5,8 +5,8 @@ from pathlib import Path
 
 from dotenv import load_dotenv
 
-from shared.mongodb import connect_mongodb
+from shared.docstore import connect_mongodb
-from shared.mongodb.classes import VisualCommunication
+from shared.docstore.classes import VisualCommunication
 
 if __name__ == '__main__':
     # prepare env vars

misc/image_upload.py

@@ -1,46 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-from pymongo.errors import DuplicateKeyError
-
-from shared.datastore import connect_minio
-from shared.mongodb import connect_mongodb
-from shared.mongodb.classes import VisualCommunication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-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
-    minio_client = connect_minio()
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # get list of image paths
-    test_dir = Path(__file__).parent
-    img_dir = test_dir / 'imgs'
-    img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
-    print(img_path_list)
-    # instantiate data object
-    vis_com_list = [
-        VisualCommunication.from_file(path, minio_client=minio_client)
-        for path in img_path_list
-    ]
-    for vis_com in vis_com_list:
-        print(repr(vis_com))
-    # upload images
-    for vis_com in vis_com_list:
-        try:
-            result = collection.insert_one(vis_com.model_dump())
-        except DuplicateKeyError as exc:
-            print('ignoring:\n', exc)
-        else:
-            print(f"inserted document: {result}")

misc/image_upload_to_server.py

@@ -1,51 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-from pymongo.errors import DuplicateKeyError
-
-from shared.datastore import connect_minio
-from shared.mongodb import connect_mongodb
-from shared.mongodb.classes import VisualCommunication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-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
-    minio_client = connect_minio()
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # get list of image paths
-    ext_img_dir = Path('/Volumes/BW-PSSD/Mixed Methods/')
-    assert ext_img_dir.exists()
-    img_path_list = [
-        path
-        for path in ext_img_dir.glob(
-            '*.jpg',
-        )
-        if path.is_file()
-    ]
-    print(f"found {len(img_path_list)} images")
-    # create visual communication objects
-    vis_com_list = [
-        VisualCommunication.from_file(path, minio_client=minio_client)
-        for path in img_path_list
-    ]
-    print(f"created {len(vis_com_list)} visual communication objects")
-    # upload images
-    for vis_com in vis_com_list:
-        try:
-            result = collection.insert_one(vis_com.model_dump())
-        except DuplicateKeyError as exc:
-            print('ignoring:\n', exc)
-        else:
-            print(f"inserted document: {result}")

misc/model_io.py

@@ -1,34 +0,0 @@
-"""Test that a model can be saved and loaded again."""
-
-from pathlib import Path
-
-import torch
-from dotenv import load_dotenv
-from torchinfo import summary
-
-from model.src.models import VisualCommunicationModel
-from shared.datastore import connect_minio, put_model
-from shared.utils import setup_logging
-
-DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-
-if __name__ == '__main__':
-    # load in env file
-    env_path = Path(__file__).parent.parent / 'server.env'
-    assert env_path.exists()
-    load_dotenv(env_path)
-    # setup logging
-    setup_logging()
-    # connect to minio
-    client = connect_minio()
-    # instantiate model
-    model = VisualCommunicationModel().to(DEVICE)
-    # show model weights
-    summary(model)
-    # put buffer in minio bucket
-    hash_str = put_model(
-        client=client,
-        model=model,
-    )
-    print(f"hash string: {hash_str}")
-    print('saved to Minio')

misc/model_outputs_from_annotation.py

@@ -1,4 +1,4 @@
-from shared.mongodb.classes import ModelData
+from shared.docstore.classes import ModelData
 
 if __name__ == '__main__':
     # instantiate data object

misc/new_model_to_minio.py

@@ -1,32 +0,0 @@
-"""Definition of function to generate a new randomly initialized model and save
-it in Minio datastore."""
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-from torchinfo import summary
-
-from model.src.models import VisualCommunicationModel
-from shared.datastore import connect_minio, put_model
-from shared.utils import setup_logging
-
-if __name__ == '__main__':
-    # load in env file
-    env_path = Path(__file__).parent.parent / 'server.env'
-    assert env_path.exists()
-    load_dotenv(env_path)
-    # setup logging
-    setup_logging()
-    # connect to minio
-    client = connect_minio()
-    # instantiate model
-    model = VisualCommunicationModel(download_resnet_weights=True)
-    # show model weights
-    summary(model)
-    # put buffer in minio bucket
-    hash_str = put_model(
-        client=client,
-        model=model,
-    )
-    print(f"hash string: {hash_str}")
-    print('saved to Minio')

misc/prediction_upload.py

@@ -1,46 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-
-from shared.datastore import connect_minio
-from shared.mongodb import connect_mongodb, upsert_prediction
-from shared.mongodb.classes import VisualCommunication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-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
-    minio_client = connect_minio()
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # get list of image paths
-    test_dir = Path(__file__).parent
-    img_dir = test_dir / 'imgs'
-    img_path_list = [path for path in img_dir.glob('*.jpeg') if path.is_file()]
-    # instantiate data object
-    vis_com_list = [
-        VisualCommunication.from_file(path, minio_client=minio_client)
-        for path in img_path_list
-    ]
-    # generate random predictions
-    for vis_com in vis_com_list:
-        vis_com.generate_random_prediction()
-    # upload visual communication
-    for vis_com in vis_com_list:
-        if vis_com.prediction is None:
-            continue
-        upsert_prediction(
-            collection=collection,
-            vis_com_name=vis_com.name,
-            predictions=vis_com.prediction,
-        )

misc/total_annotated.py

@@ -5,7 +5,7 @@ from pathlib import Path
 
 from dotenv import load_dotenv
 
-from shared.mongodb import connect_mongodb, count_documents
+from shared.docstore import connect_mongodb, count_documents
 
 if __name__ == '__main__':
     # prepare env vars

misc/total_documents.py

@@ -5,7 +5,7 @@ from pathlib import Path
 
 from dotenv import load_dotenv
 
-from shared.mongodb import connect_mongodb, count_documents
+from shared.docstore import connect_mongodb, count_documents
 
 if __name__ == '__main__':
     # prepare env vars

model/src/main.py

@@ -9,21 +9,16 @@ from models import VisualCommunicationModel
 from tqdm import tqdm
 from utils import DEVICE, VCDADataset, load_model
 
-from shared.datastore import connect_minio
-from shared.mongodb.classes import ModelData
 from shared.utils import setup_logging
 
 if __name__ == '__main__':
     # setup logging
     setup_logging()
-    # connect to minio
-    minio_client = connect_minio()
     # instantiate model
-    model: VisualCommunicationModel = load_model(client=minio_client)
+    model: VisualCommunicationModel = load_model()
     model.eval()
     # setup dataset
     dataset = VCDADataset(
-        minio_client=minio_client,
         data_name_list=[
             '02dbaf48d713e4e6d3a6b98fd2dc866e',
         ],
@@ -39,10 +34,10 @@ if __name__ == '__main__':
                 image = torch.unsqueeze(image, 0)  # add artificial batch dimension
                 image = image.to(DEVICE)
                 # make prediction
-                pred: ModelData = model(image)
+                pred: dict = model(image)
             except Exception:
                 print_exc()
                 continue
             else:
-                print(json.dumps(pred.model_dump(), indent=4))
+                print(json.dumps(pred, indent=4))
     logging.debug('finished')

model/src/models/visual_communication.py

@@ -4,8 +4,6 @@ from __future__ import annotations
 
 from torch import nn
 
-from shared.mongodb.classes import ModelData
-
 from .angle import AngleTail
 from .contact import ContactTail
 from .distance import DistanceTail
@@ -39,7 +37,7 @@ class VisualCommunicationModel(nn.Module):
         self.framing_tail = FramingTail()
         self.salience_tail = SalienceTail()
 
-    def forward(self, x) -> ModelData:
+    def forward(self, x) -> dict:
         """Calculate model output on data."""
         # generate visual representation
         features = self.resnet_head(x)
@@ -57,6 +55,4 @@ class VisualCommunicationModel(nn.Module):
             'framing': self.framing_tail(features),
             'salience': self.salience_tail(features),
         }
-        # convert to respective classes
+        return prediction_dict
-        data = ModelData.from_prediction_dict(prediction_dict)
-        return data

model/src/utils/load_model.py

@@ -4,33 +4,30 @@ import logging
 from pathlib import Path
 
 import torch
-from minio import Minio
 
 from model.src.models import VisualCommunicationModel
-from shared.datastore import get_model
+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(
+def load_model() -> VisualCommunicationModel:
-    client: Minio,
-) -> VisualCommunicationModel:
     """Instantiate model with weights loaded from latest model saved in
     MinIO."""
-    assert isinstance(client, 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 from minio
+    # load model data
-    model_checkpoint = get_model(
+    with ModelRepository() as repo:
-        client=client,
+        model_data = repo.get_data(model_object_name)
-        object_name=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()

model/src/utils/vcda_dataset.py

@@ -2,7 +2,6 @@
 
 import random
 
-from minio import Minio
 from PIL import Image
 from torch import Tensor
 from torch.utils.data import Dataset
@@ -16,7 +15,7 @@ from torchvision.transforms.functional import (
     to_tensor,
 )
 
-from shared.datastore import get_image
+from shared.repositories import ImageRepository
 
 # resnet18 original normalization values
 RESNET_NORMALIZE_MEAN = [0.485, 0.456, 0.406]
@@ -28,13 +27,11 @@ class VCDADataset(Dataset):
 
     def __init__(
         self,
-        minio_client: Minio,
         data_name_list: list[str],
         do_augment: bool = False,
         random_annotations: bool = False,
     ):
         super().__init__()
-        self.minio_client = minio_client
         self.data_name_list = data_name_list
         self.do_augment = do_augment
         self.random_annotations = random_annotations
@@ -56,12 +53,11 @@ class VCDADataset(Dataset):
 
     def __getitem__(self, idx):
         # get image from database
-        object_name = self.data_name_list[idx]
+        image_name = self.data_name_list[idx]
-        image = get_image(
+        with ImageRepository() as repo:
-            client=self.minio_client,
+            image_data = repo.get_data(image_name)
-            object_name=object_name,
+        assert image_data is not None
-        )
+        tensor = self.image_to_tensor(image_data.image)
-        tensor = self.image_to_tensor(image)
         if self.do_augment:
             tensor = self.augment(tensor)
         return tensor

model/train.py

@@ -18,7 +18,6 @@ from torch.optim.lr_scheduler import ExponentialLR
 from torch.utils.data import DataLoader
 
 from model.src.utils import VCDADataset, get_class
-from shared.datastore import connect_minio
 
 
 def parse_arguments():
@@ -57,18 +56,16 @@ optimizer = torch.optim.Adam(model.parameters(), lr=args.lr)
 loss_fn = nn.CrossEntropyLoss()
 
 # create datasets and loaders
-minio_client = connect_minio()
 with open('model/src/dataset/train.csv', encoding='utf-8') as fh:
     train_data_name_list = fh.read().split('\n')
 train_dataset = VCDADataset(
-    minio_client=minio_client,
     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(minio_client=minio_client, data_name_list=val_data_name_list)
+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
@@ -140,7 +137,7 @@ to_save = {
 }
 checkpoint_handler = Checkpoint(
     to_save,
-    f"runs/checkpoints/{run_name}",
+    f'runs/checkpoints/{run_name}',
     n_saved=3,
     filename_prefix='best',
     score_function=lambda engine: -engine.state.metrics['loss'],
@@ -154,7 +151,7 @@ if 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:
+with open(f'runs/configs/{run_name}.json', 'w', encoding='utf-8') as fh:
     json.dump(config, fh)
 
 # start training

other/transfer_images_minio_subfolder.py

@@ -1,39 +0,0 @@
-"""Script to move minio images to subfolder."""
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-from PIL import Image
-
-from shared.datastore import connect_minio, get, put_image
-from shared.utils import setup_logging
-
-if __name__ == '__main__':
-    # load in env file
-    env_path = Path(__file__).parent.parent / 'server.env'
-    assert env_path.exists()
-    load_dotenv(env_path)
-    # setup logging
-    setup_logging()
-    # connect to minio
-    minio_client = connect_minio()
-    # list images in bucket
-    BUCKET_NAME = 'visual-critical-discourse-analysis'
-    obj_list = minio_client.list_objects(
-        bucket_name=BUCKET_NAME,
-    )
-    # begin moving images
-    for obj in obj_list:
-        # get image from minio
-        buffer = get(
-            client=minio_client,
-            bucket_name=BUCKET_NAME,
-            object_name=obj.object_name,
-        )
-        # convert data to image
-        image = Image.open(buffer)
-        # put image into minio subfolder
-        put_image(
-            client=minio_client,
-            image=image,
-        )

other/transfer_images_mongo_minio.py

@@ -9,9 +9,9 @@ from bson import ObjectId
 from dotenv import load_dotenv
 from pymongo.collection import Collection
 
-from shared.datastore import connect_minio, put_image
+from shared.datastore import Datastore
-from shared.mongodb import connect_mongodb
+from shared.docstore import connect_mongodb
-from shared.mongodb.src.classes import VisualCommunication
+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
 
@@ -81,7 +81,9 @@ if __name__ == '__main__':
     # setup logging
     setup_logging()
     # connect to minIO
-    minio_client = connect_minio()
+    datastore = Datastore()
+    datastore.connect()
+    assert datastore._client is not None
     # connect to MongoDB
     collection, db, client = connect_mongodb()
     # list documents in mongoDB
@@ -97,11 +99,10 @@ if __name__ == '__main__':
         try:
             # get image
             image = vis_com.get_image(
-                minio_client=minio_client,
+                minio_client=datastore._client,
             )
             # put buffer in minio
-            object_name = put_image(
+            object_name = datastore.put_image(
-                client=minio_client,
                 image=image,
             )
         except Exception as exc:

pyproject.toml

@@ -24,6 +24,7 @@ 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]
@@ -31,10 +32,11 @@ 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"
@@ -110,7 +112,7 @@ module = "utils.*"
 ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
-module = "datastore.*"
+module = "shared.datastore.*"
 ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
@@ -118,9 +120,5 @@ module = "models.*"
 ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
-module = "shared.mongodb.*"
+module = "shared.docstore.*"
-ignore_missing_imports = true
-
-[[tool.mypy.overrides]]
-module = "testcontainers.*"
 ignore_missing_imports = true

shared/datastore/__init__.py

@@ -1,10 +0,0 @@
-from .src import (
-    connect_minio,
-    delete,
-    get,
-    get_image,
-    get_model,
-    put,
-    put_image,
-    put_model,
-)

shared/datastore/src/__init__.py

@@ -1,8 +0,0 @@
-from .connect_minio import connect_minio
-from .delete import delete
-from .get import get
-from .get_image import get_image
-from .get_model import get_model
-from .put import put
-from .put_image import put_image
-from .put_model import put_model

shared/datastore/src/connect_minio.py

@@ -1,38 +0,0 @@
-"""Definition of connect function."""
-
-import logging
-import os
-
-from minio import Minio
-
-from shared.utils import check_env
-
-
-def connect_minio() -> Minio:
-    """Connect to MinIO server."""
-    # ensure necessary env vars available
-    var_list = {
-        'MINIO_ENDPOINT',
-        'MINIO_ACCESS_KEY',
-        'MINIO_SECRET_KEY',
-        'MINIO_BUCKET_NAME',
-    }
-    check_env(var_list)
-    # prepare arguments
-    minio_endpoint = str(os.getenv('MINIO_ENDPOINT'))
-    minio_access_key = str(os.getenv('MINIO_ACCESS_KEY'))
-    minio_secret_key = str(os.getenv('MINIO_SECRET_KEY'))
-    minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    # 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):
-        logging.info('creating bucket: %s', minio_bucket_name)
-        client.make_bucket(bucket_name=minio_bucket_name)
-    logging.debug('finished')
-    return client

shared/datastore/src/delete.py

@@ -1,31 +0,0 @@
-"""Definition of delete function."""
-
-import logging
-from traceback import print_exc
-
-from minio import Minio
-
-
-def delete(
-    client: Minio,
-    bucket_name: str,
-    object_name: str,
-) -> None:
-    """Delete object from MinIO."""
-    assert isinstance(client, Minio)
-    assert isinstance(bucket_name, str)
-    assert len(bucket_name) > 0
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # remove object
-    try:
-        client.remove_object(
-            bucket_name=bucket_name,
-            object_name=object_name,
-        )
-    except Exception as exc:
-        logging.error('failed deleting %s', object_name)
-        print_exc()
-        raise exc
-    else:
-        logging.debug('deleted %s', object_name)

shared/datastore/src/get.py

@@ -1,42 +0,0 @@
-"""Definition of get function."""
-
-import logging
-from io import BytesIO
-from traceback import print_exc
-
-from minio import Minio
-
-
-def get(
-    client: Minio,
-    bucket_name: str,
-    object_name: str,
-) -> BytesIO:
-    """Get buffer from bucket in MinIO."""
-    assert isinstance(client, Minio)
-    assert isinstance(bucket_name, str)
-    assert len(bucket_name) > 0
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    try:
-        # make request
-        response = client.get_object(
-            bucket_name=bucket_name,
-            object_name=object_name,
-        )
-        assert response.status == 200
-        # get buffer
-        buffer = BytesIO()
-        chunk_size = 2**14
-        while chunk := response.read(chunk_size):
-            buffer.write(chunk)
-        buffer.seek(0)
-        logging.debug('got %s', object_name)
-        return buffer
-    except Exception as exc:
-        logging.error('failed getting data from MinIO')
-        print_exc()
-        raise exc
-    finally:
-        response.close()
-        response.release_conn()

shared/datastore/src/get_image.py

@@ -1,35 +0,0 @@
-"""Definition of get_image function."""
-
-import logging
-import os
-
-from minio import Minio
-from PIL import Image
-
-from shared.utils import check_env
-
-from .get import get
-
-
-def get_image(
-    client: Minio,
-    object_name: str,
-) -> Image.Image:
-    """Get image from image subfolder in bucket in Minio."""
-    assert isinstance(client, Minio)
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # prepare arguments
-    check_env({'MINIO_BUCKET_NAME'})
-    bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    object_name = f'images/{object_name}'
-    # get object from bucket
-    buffer = get(
-        client=client,
-        bucket_name=bucket_name,
-        object_name=object_name,
-    )
-    # convert data to image
-    image = Image.open(buffer)
-    logging.debug('got data from %s', object_name)
-    return image

shared/datastore/src/get_model.py

@@ -1,36 +0,0 @@
-"""Definition of get_model function."""
-
-import logging
-import os
-from collections import OrderedDict
-
-import torch
-from minio import Minio
-
-from shared.utils import check_env
-
-from .get import get
-
-
-def get_model(
-    client: Minio,
-    object_name: str,
-) -> OrderedDict:
-    """Get model from model subfolder in bucket in Minio."""
-    assert isinstance(client, Minio)
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # prepare arguments
-    check_env({'MINIO_BUCKET_NAME'})
-    bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    object_name = f'models/{object_name}'
-    # get object from bucket
-    buffer = get(
-        client=client,
-        bucket_name=bucket_name,
-        object_name=object_name,
-    )
-    # convert data to model checkpoint
-    model_content = torch.load(buffer)
-    logging.debug('finished')
-    return model_content

shared/datastore/src/put.py

@@ -1,36 +0,0 @@
-"""Definition of put function."""
-
-import logging
-from io import BytesIO
-
-from minio import Minio
-
-
-def put(
-    client: Minio,
-    buffer: BytesIO,
-    bucket_name: str,
-    object_name: str,
-) -> None:
-    """Put buffer in bucket in MinIO and return MD5 checksum as object name."""
-    assert isinstance(client, Minio)
-    assert isinstance(buffer, BytesIO)
-    assert isinstance(bucket_name, str)
-    assert len(bucket_name) > 0
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # prepare for saving
-    num_bytes = buffer.tell()
-    buffer.seek(0)
-    # send data to bucket
-    try:
-        client.put_object(
-            bucket_name=bucket_name,
-            object_name=object_name,
-            length=num_bytes,
-            data=buffer,
-        )
-    except Exception as exc:
-        logging.error('failed saving data to MinIO')
-        raise exc
-    logging.debug('saved data to %s', object_name)

shared/datastore/src/put_image.py

@@ -1,40 +0,0 @@
-"""Definition of put_image function."""
-
-import logging
-import os
-from hashlib import md5
-from io import BytesIO
-
-from minio import Minio
-from PIL import Image
-
-from .put import put
-
-
-def put_image(
-    client: Minio,
-    image: Image.Image,
-) -> str:
-    """Put image in image subfolder in bucket in Minio and return MD5 checksum
-    used as object name."""
-    assert isinstance(client, Minio)
-    assert isinstance(image, Image.Image)
-    # get bucket name from env
-    bucket_name = os.getenv('MINIO_BUCKET_NAME', default='')
-    assert len(bucket_name) > 0
-    # save data to buffer
-    buffer = BytesIO()
-    image.save(buffer, 'png')
-    # get md5 of buffer
-    checksum = md5(buffer.getbuffer()).hexdigest()
-    # set object name
-    object_name = f'images/{checksum}'
-    # send data to bucket
-    put(
-        client=client,
-        buffer=buffer,
-        bucket_name=bucket_name,
-        object_name=object_name,
-    )
-    logging.debug('finished')
-    return checksum

shared/datastore/src/put_model.py

@@ -1,43 +0,0 @@
-"""Definition of put_model function."""
-
-import logging
-import os
-from hashlib import md5
-from io import BytesIO
-
-import torch
-from minio import Minio
-from torch.nn import Module
-
-from shared.utils import check_env
-
-from .put import put
-
-
-def put_model(
-    client: Minio,
-    model: Module,
-) -> str:
-    """Put model in model subfolder in bucket in Minio and return MD5 checksum
-    used as object name."""
-    assert isinstance(client, Minio)
-    assert isinstance(model, Module)
-    # get bucket name from env
-    check_env({'MINIO_BUCKET_NAME'})
-    bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    # save data to buffer
-    buffer = BytesIO()
-    torch.save(model.state_dict(), buffer)
-    # get md5 of image
-    checksum = md5(buffer.getbuffer()).hexdigest()
-    # set object name
-    object_name = f'models/{checksum}'
-    # send data to bucket
-    put(
-        client=client,
-        buffer=buffer,
-        bucket_name=bucket_name,
-        object_name=object_name,
-    )
-    logging.debug('finished')
-    return checksum

shared/datastore/tests/integration/base_crud_test.py

@@ -1,106 +0,0 @@
-"""Integration tests related to base CRUD functions."""
-
-import logging
-from io import BytesIO
-
-import pytest
-
-# from shared.datastore import connect_minio, delete, get, put
-
-
-def same_data(
-    data_a: BytesIO,
-    data_b: BytesIO,
-) -> bool:
-    """Check if two BytesIO-objects contain the same data."""
-    assert isinstance(data_a, BytesIO)
-    assert isinstance(data_b, BytesIO)
-    # prepare for being read
-    data_a.seek(0)
-    data_b.seek(0)
-    # convert to bytes
-    data_a_bytes = data_a.read()
-    data_b_bytes = data_b.read()
-    # compare size
-    logging.error(len(data_a_bytes))
-    logging.error(len(data_b_bytes))
-    if len(data_a_bytes) != len(data_b_bytes):
-        return False
-    # compare content
-    if data_a_bytes != data_b_bytes:
-        return False
-    return True
-
-
-# def test_should_get_data(
-#     minio_client,
-#     data_in_minio,
-# ):
-#     data, bucket_name, object_name = data_in_minio
-#     received_data = get(
-#         client=minio_client,
-#         bucket_name=bucket_name,
-#         object_name=object_name,
-#     )
-
-#     assert isinstance(data, BytesIO)
-#     assert same_data(data, received_data)
-
-
-# def test_should_delete_data(
-#     data_in_minio,
-# ):
-#     _, bucket_name, object_name = data_in_minio
-#     client = connect_minio()
-#     delete(
-#         client=client,
-#         bucket_name=bucket_name,
-#         object_name=object_name,
-#     )
-#     with pytest.raises(ValueError):
-#         _ = get(
-#             client=client,
-#             bucket_name=bucket_name,
-#             object_name=object_name,
-#         )
-
-
-# def test_should_put_data(
-#     data,
-# ):
-#     buffer, bucket_name, object_name = data
-#     client = connect_minio()
-#     put(
-#         client=client,
-#         buffer=buffer,
-#         bucket_name=bucket_name,
-#         object_name=object_name,
-#     )
-#     received_data = get(
-#         client=client,
-#         bucket_name=bucket_name,
-#         object_name=object_name,
-#     )
-#     assert received_data == data
-
-
-# def test_should_update_data(
-#     data_in_minio,
-# ):
-#     buffer, bucket_name, object_name = data_in_minio
-#     client = connect_minio()
-#     put(
-#         client=client,
-#         buffer=buffer,
-#         bucket_name=bucket_name,
-#         object_name=object_name,
-#     )
-#     received_data = get(
-#         client=client,
-#         bucket_name=bucket_name,
-#         object_name=object_name,
-#     )
-#     assert received_data == buffer
-
-if __name__ == '__main__':
-    pytest.main()

shared/datastore/tests/integration/conftest.py

@@ -1,161 +0,0 @@
-"""Integration test configurations."""
-
-import logging
-import os
-import random
-from collections.abc import Iterator
-from io import BytesIO
-from pathlib import Path
-
-import pytest
-from minio import Minio
-from PIL import Image
-from testcontainers.core.container import inside_container
-from testcontainers.minio import MinioContainer
-
-
-class FixedMinioContainer(MinioContainer):
-    """Fixed Minio Container to allow running inside CI pipeline."""
-
-    def get_container_host_ip(self) -> str:
-        if inside_container() and Path('/var/run/docker.sock').exists():
-            return self.get_docker_client().gateway_ip(self._container.id)
-        return super().get_container_host_ip()
-
-
-env_var_map = {
-    'MINIO_ENDPOINT': 'localhost:9000',
-    'MINIO_ACCESS_KEY': 'test-access-key',
-    'MINIO_SECRET_KEY': 'test-secret-key',
-    'MINIO_BUCKET_NAME': 'test-bucket',
-    'MINIO_OBJECT_NAME': '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX',
-}
-container_map = {
-    'minio': FixedMinioContainer(
-        port=env_var_map['MINIO_ENDPOINT'].rsplit(':', maxsplit=1)[-1],
-        access_key=env_var_map['MINIO_ACCESS_KEY'],
-        secret_key=env_var_map['MINIO_SECRET_KEY'],
-    ),
-}
-
-
-@pytest.fixture(scope='session', autouse=True)
-def setup_infrastructure(request: pytest.FixtureRequest) -> None:
-    """Prepare infrastructure for integration test."""
-    # prepare infrastructure
-    for container in container_map.values():
-        container.start()
-    # update env var map
-    minio_host = container_map['minio'].get_container_host_ip()
-    minio_port = container_map['minio'].get_exposed_port(
-        port=env_var_map['MINIO_ENDPOINT'].rsplit(':', maxsplit=1)[-1],
-    )
-    env_var_map['MINIO_ENDPOINT'] = f'{minio_host}:{minio_port}'
-    logging.error('MINIO_ENDPOINT: %s', env_var_map['MINIO_ENDPOINT'])
-
-    # ensure cleanup
-    def cleanup_infrastructure():
-        for container in container_map.values():
-            container.stop()
-
-    request.addfinalizer(cleanup_infrastructure)
-
-
-@pytest.fixture(scope='session', autouse=True)
-def populate_env(
-    request: pytest.FixtureRequest,
-    setup_infrastructure,
-) -> None:
-    """Populate environment with variables used for testing."""
-    # update env
-    for key, val in env_var_map.items():
-        os.environ[key] = val
-
-    # ensure cleanup
-    def cleanup_env():
-        for key in env_var_map:
-            _ = os.environ.pop(key, default=None)
-
-    request.addfinalizer(cleanup_env)
-
-
-@pytest.fixture(scope='session')
-def minio_client(
-    setup_infrastructure,
-    populate_env,
-) -> Iterator[Minio]:
-    # prepare arguments
-    minio_endpoint = str(os.getenv('MINIO_ENDPOINT'))
-    minio_access_key = str(os.getenv('MINIO_ACCESS_KEY'))
-    minio_secret_key = str(os.getenv('MINIO_SECRET_KEY'))
-    minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    # connect to minio
-    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)
-    # expose client
-    yield client
-
-
-@pytest.fixture
-def data() -> Iterator[bytes]:
-    # generate random data
-    num_bytes = 2**21  # 2 MB
-    data = random.randbytes(n=num_bytes)
-    # expose data
-    yield data
-
-
-@pytest.fixture
-def data_in_minio(
-    minio_client,
-    data,
-) -> Iterator[tuple[BytesIO, str, str]]:
-    # prepare arguments
-    minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    minio_object_name = str(os.getenv('MINIO_OBJECT_NAME'))
-    # convert data
-    buffer = BytesIO(data)
-    # prepare for saving
-    num_bytes = buffer.tell()
-    buffer.seek(0)
-    # send data to bucket
-    minio_client.put_object(
-        bucket_name=minio_bucket_name,
-        object_name=minio_object_name,
-        length=num_bytes,
-        data=buffer,
-    )
-    # expose data
-    yield buffer, minio_bucket_name, minio_object_name
-    # clean up
-    minio_client.remove_object(
-        bucket_name=minio_bucket_name,
-        object_name=minio_object_name,
-    )
-
-
-@pytest.fixture
-def image() -> Iterator[Image.Image]:
-    # generate image
-    image = Image.new(mode='RGB', size=(480, 480))
-    # expose image
-    yield image
-
-
-# @pytest.fixture
-# def image_in_minio(
-#     image: Image.Image,
-# ) -> tuple(Image.Image, str):
-#     #
-
-
-#     # expose image and object name
-#     yield image, object_name
-#     # cleanup

shared/datastore/tests/integration/connect_test.py

@@ -1,10 +0,0 @@
-"""Integration test for connect_minio function."""
-
-from minio import Minio
-
-from shared.datastore import connect_minio
-
-
-def test_should_return_correct_type():
-    client = connect_minio()
-    assert isinstance(client, Minio)

shared/datastore/tests/integration/image_crud_test.py

@@ -1,73 +0,0 @@
-"""Integration tests related to image CRUD."""
-
-import numpy as np
-from PIL import Image
-
-# from shared.datastore import connect_minio, get_image, put_image
-
-
-def same_image(
-    img_a: Image.Image,
-    img_b: Image.Image,
-) -> bool:
-    """Check if two images contain the same data."""
-    assert isinstance(img_a, Image.Image)
-    assert isinstance(img_b, Image.Image)
-    # check if images have a comparable number of channels
-    if img_a.getbands() != img_b.getbands():
-        return False
-    # calculate pixel difference between images
-    img_a_arr = np.asarray(img_a)
-    img_b_arr = np.asarray(img_b)
-    diff = np.subtract(img_a_arr, img_b_arr)
-    if np.sum(diff) != 0:
-        return False
-    return True
-
-
-# def test_should_get_image(
-#     image_in_minio,
-# ):
-#     image, object_name = image_in_minio
-#     client = connect_minio()
-#     received_image = get_image(
-#         client=client,
-#         object_name=object_name,
-#     )
-#     assert isinstance(image, Image.Image)
-#     assert received_image == image
-
-
-# def test_should_put_image(
-#     image,
-# ):
-#     client = connect_minio()
-#     object_name = put_image(
-#         client=client,
-#         image=image,
-#     )
-#     assert isinstance(object_name, str)
-#     assert len(object_name) > 0
-#     received_image = get_image(
-#         client=client,
-#         object_name=object_name,
-#     )
-#     assert same_image(received_image, image)
-
-
-# def test_should_update_image(
-#     image_in_minio,
-# ):
-#     image, object_name = image_in_minio
-#     client = connect_minio()
-#     object_name = put_image(
-#         client=client,
-#         image=image,
-#     )
-#     assert isinstance(object_name, str)
-#     assert len(object_name) > 0
-#     received_image = get_image(
-#         client=client,
-#         object_name=object_name,
-#     )
-#     assert received_image == image

shared/datastore/tests/integration/model_crud_test.py

@@ -1,52 +0,0 @@
-"""Integration tests related to model CRUD."""
-
-# from torch.nn import Module
-
-# from shared.datastore import connect_minio, get_model, put_model
-
-# def test_should_get_model(
-#     model_in_minio,
-# ):
-#     model, object_name = model_in_minio
-#     client = connect_minio()
-#     received_model = get_model(
-#         client=client,
-#         object_name=object_name,
-#     )
-#     assert isinstance(model, Module)
-#     assert received_model == model
-
-
-# def test_should_put_model(
-#     model,
-# ):
-#     client = connect_minio()
-#     object_name = put_model(
-#         client=client,
-#         model=model,
-#     )
-#     assert isinstance(object_name, str)
-#     assert len(object_name) > 0
-#     received_model = get_model(
-#         client=client,
-#         object_name=object_name,
-#     )
-#     assert received_model == model
-
-
-# def test_should_update_model(
-#     model_in_minio,
-# ):
-#     model, object_name = model_in_minio
-#     client = connect_minio()
-#     object_name = put_model(
-#         client=client,
-#         model=model,
-#     )
-#     assert isinstance(object_name, str)
-#     assert len(object_name) > 0
-#     received_model = get_model(
-#         client=client,
-#         object_name=object_name,
-#     )
-#     assert received_model == model

shared/datastore/tests/unit/connect_minio_test.py

@@ -1,33 +0,0 @@
-"""Definition of unittests for connect_minio function."""
-
-import os
-import unittest
-
-from shared.datastore import connect_minio
-
-
-class TestConnectMinio(unittest.TestCase):
-
-    def setUp(self):
-        # define relevant env vars
-        self.env_var_map = {
-            'MINIO_ENDPOINT': '192.168.1.2',
-            'MINIO_ACCESS_KEY': 'randomAccess_key',
-            'MINIO_SECRET_KEY': 'randomSecret_key',
-            'MINIO_BUCKET_NAME': 'test-bucket-name',
-        }
-        # set env vars
-        for key, val in self.env_var_map.items():
-            os.environ[key] = val
-
-    def tearDown(self):
-        # clear env vars
-        for key in self.env_var_map:
-            _ = os.environ.pop(key, default=None)
-
-    def test_should_fail_when_env_not_set(self):
-        # ensure env not set
-        self.tearDown()
-        # run test
-        with self.assertRaises(AssertionError):
-            _ = connect_minio()

shared/datastore/tests/unit/delete_test.py

@@ -1,61 +0,0 @@
-"""Definition of unittests for delete function."""
-
-import unittest
-from unittest.mock import Mock
-
-from minio import Minio
-
-from shared.datastore import delete
-
-
-class TestDelete(unittest.TestCase):
-
-    def setUp(self):
-        # set relevant variables
-        self.client = Mock(spec=Minio)
-        self.bucket_name = 'test-bucket'
-        self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
-        # set bad arguments
-        self.bad_client = 'not-minio-type'
-        self.bad_string = float(0.0)
-        self.len_0_string = ''
-
-    def test_should_fail_on_wrong_input_type_client(self):
-        with self.assertRaises(AssertionError):
-            delete(
-                client=self.bad_client,
-                bucket_name=self.bucket_name,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_bucket_name(self):
-        with self.assertRaises(AssertionError):
-            delete(
-                client=self.client,
-                bucket_name=self.bad_string,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_length_bucket_name(self):
-        with self.assertRaises(AssertionError):
-            delete(
-                client=self.client,
-                bucket_name=self.len_0_string,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_object_name(self):
-        with self.assertRaises(AssertionError):
-            delete(
-                client=self.client,
-                bucket_name=self.bucket_name,
-                object_name=self.bad_string,
-            )
-
-    def test_should_fail_on_wrong_input_length_object_name(self):
-        with self.assertRaises(AssertionError):
-            delete(
-                client=self.client,
-                bucket_name=self.bucket_name,
-                object_name=self.len_0_string,
-            )

shared/datastore/tests/unit/get_image_test.py

@@ -1,63 +0,0 @@
-"""Definition of unittests for get_image function."""
-
-import os
-import unittest
-from unittest.mock import MagicMock
-
-from minio import Minio
-
-from shared.datastore import get_image
-
-
-class TestGetImage(unittest.TestCase):
-
-    def setUp(self):
-        # set relevant variables
-        self.client = MagicMock(spec=Minio)
-        self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
-        self.env_var_map = {
-            'MINIO_BUCKET_NAME': 'test-bucket',
-        }
-        # set bad arguments
-        self.bad_client = 'not-minio-type'
-        self.bad_string = float(0.0)
-        self.len_0_string = ''
-        # populate env
-        for key, val in self.env_var_map.items():
-            os.environ[key] = val
-
-    def tearDown(self):
-        # clean env
-        for key in self.env_var_map:
-            _ = os.environ.pop(key, default=None)
-
-    def test_should_fail_when_env_not_set(self):
-        # ensure env not set
-        self.tearDown()
-        # run test
-        with self.assertRaises(AssertionError):
-            get_image(
-                client=self.client,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_client(self):
-        with self.assertRaises(AssertionError):
-            get_image(
-                client=self.bad_client,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_object_name(self):
-        with self.assertRaises(AssertionError):
-            get_image(
-                client=self.client,
-                object_name=self.bad_string,
-            )
-
-    def test_should_fail_on_wrong_input_length_object_name(self):
-        with self.assertRaises(AssertionError):
-            get_image(
-                client=self.client,
-                object_name=self.len_0_string,
-            )

shared/datastore/tests/unit/get_model_test.py

@@ -1,63 +0,0 @@
-"""Definition of unittest for get_model function."""
-
-import os
-import unittest
-from unittest.mock import MagicMock
-
-from minio import Minio
-
-from shared.datastore import get_model
-
-
-class TestGetModel(unittest.TestCase):
-
-    def setUp(self):
-        # set relevant variables
-        self.client = MagicMock(spec=Minio)
-        self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
-        self.env_var_map = {
-            'MINIO_BUCKET_NAME': 'test-bucket',
-        }
-        # set bad arguments
-        self.bad_client = 'not-minio-type'
-        self.bad_string = float(0.0)
-        self.len_0_string = ''
-        # populate env
-        for key, val in self.env_var_map.items():
-            os.environ[key] = val
-
-    def tearDown(self):
-        # clean env
-        for key in self.env_var_map:
-            _ = os.environ.pop(key, default=None)
-
-    def test_should_fail_when_env_not_set(self):
-        # ensure env not set
-        self.tearDown()
-        # run test
-        with self.assertRaises(AssertionError):
-            get_model(
-                client=self.client,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_client(self):
-        with self.assertRaises(AssertionError):
-            get_model(
-                client=self.bad_client,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_object_name(self):
-        with self.assertRaises(AssertionError):
-            get_model(
-                client=self.client,
-                object_name=self.bad_string,
-            )
-
-    def test_should_fail_on_wrong_input_length_object_name(self):
-        with self.assertRaises(AssertionError):
-            get_model(
-                client=self.client,
-                object_name=self.len_0_string,
-            )

shared/datastore/tests/unit/get_test.py

@@ -1,60 +0,0 @@
-"""Definition of unittests for get function."""
-
-import unittest
-from unittest.mock import MagicMock
-
-from minio import Minio
-
-from shared.datastore import get
-
-
-class TestGet(unittest.TestCase):
-    def setUp(self):
-        # set relevant variables
-        self.client = MagicMock(spec=Minio)
-        self.bucket_name = 'test-bucket'
-        self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
-        # set bad arguments
-        self.bad_client = 'not-minio-type'
-        self.bad_string = float(0.0)
-        self.len_0_string = ''
-
-    def test_should_fail_on_wrong_input_type_client(self):
-        with self.assertRaises(AssertionError):
-            get(
-                client=self.bad_client,
-                bucket_name=self.bucket_name,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_bucket_name(self):
-        with self.assertRaises(AssertionError):
-            get(
-                client=self.client,
-                bucket_name=self.bad_string,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_length_bucket_name(self):
-        with self.assertRaises(AssertionError):
-            get(
-                client=self.client,
-                bucket_name=self.len_0_string,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_object_name(self):
-        with self.assertRaises(AssertionError):
-            get(
-                client=self.client,
-                bucket_name=self.bucket_name,
-                object_name=self.bad_string,
-            )
-
-    def test_should_fail_on_wrong_input_length_object_name(self):
-        with self.assertRaises(AssertionError):
-            get(
-                client=self.client,
-                bucket_name=self.bucket_name,
-                object_name=self.len_0_string,
-            )

shared/datastore/tests/unit/put_image_test.py

@@ -1,41 +0,0 @@
-"""Definition of unittests for put_image function."""
-
-import os
-import unittest
-from unittest.mock import MagicMock
-
-from minio import Minio
-from PIL import Image
-
-from shared.datastore import put_image
-
-
-class TestPutImage(unittest.TestCase):
-
-    def setUp(self):
-        # set relevant variables
-        self.client = MagicMock(spec=Minio)
-        self.image = Image.new(mode='RGB', size=(480, 480))
-        self.env_var_map = {
-            'MINIO_BUCKET_NAME': 'test-bucket',
-        }
-        # set bad arguments
-        self.bad_client = 'not-minio-type'
-        self.bad_image = 'not-image-type'
-        # populate env
-        for key, val in self.env_var_map.items():
-            os.environ[key] = val
-
-    def test_should_fail_on_wrong_input_type_client(self):
-        with self.assertRaises(AssertionError):
-            put_image(
-                client=self.bad_client,
-                image=self.image,
-            )
-
-    def test_should_fail_on_wrong_input_type_image(self):
-        with self.assertRaises(AssertionError):
-            put_image(
-                client=self.client,
-                image=self.bad_image,
-            )

shared/datastore/tests/unit/put_model_test.py

@@ -1,40 +0,0 @@
-"""Definition of unittests for put_model function."""
-
-import os
-import unittest
-from unittest.mock import MagicMock
-
-from minio import Minio
-from torch.nn import Module
-
-from shared.datastore import put_model
-
-
-class TestPutModel(unittest.TestCase):
-    def setUp(self):
-        # set relevant variables
-        self.client = MagicMock(spec=Minio)
-        self.model = MagicMock(spec=Module)
-        self.env_var_map = {
-            'MINIO_BUCKET_NAME': 'test-bucket',
-        }
-        # set bad arguments
-        self.bad_client = 'not-minio-type'
-        self.bad_model = 'not-image-type'
-        # populate env
-        for key, val in self.env_var_map.items():
-            os.environ[key] = val
-
-    def test_should_fail_on_wrong_input_type_client(self):
-        with self.assertRaises(AssertionError):
-            put_model(
-                client=self.bad_client,
-                model=self.model,
-            )
-
-    def test_should_fail_on_wrong_input_type_model(self):
-        with self.assertRaises(AssertionError):
-            put_model(
-                client=self.client,
-                model=self.bad_model,
-            )

shared/datastore/tests/unit/put_test.py

@@ -1,80 +0,0 @@
-"""Definition of unittests for put function."""
-
-import unittest
-from io import BytesIO
-from unittest.mock import MagicMock
-
-from minio import Minio
-from PIL import Image
-
-from shared.datastore import put
-
-
-class TestPut(unittest.TestCase):
-    def setUp(self):
-        # set relevant variables
-        self.client = MagicMock(spec=Minio)
-        self.bucket_name = 'test-bucket'
-        self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
-        self.image = Image.new(mode='RGB', size=(480, 480))
-        self.buffer = BytesIO()
-        self.image.save(self.buffer, 'png')
-        # set bad arguments
-        self.bad_client = 'not-minio-type'
-        self.bad_string = float(0.0)
-        self.bad_buffer = 'not-buffer-type'
-        self.len_0_string = ''
-
-    def test_should_fail_on_wrong_input_type_client(self):
-        with self.assertRaises(AssertionError):
-            put(
-                client=self.bad_client,
-                buffer=self.buffer,
-                bucket_name=self.bucket_name,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_buffer(self):
-        with self.assertRaises(AssertionError):
-            put(
-                client=self.client,
-                buffer=self.bad_buffer,
-                bucket_name=self.bucket_name,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_bucket_name(self):
-        with self.assertRaises(AssertionError):
-            put(
-                client=self.client,
-                buffer=self.buffer,
-                bucket_name=self.bad_string,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_length_bucket_name(self):
-        with self.assertRaises(AssertionError):
-            put(
-                client=self.client,
-                buffer=self.buffer,
-                bucket_name=self.len_0_string,
-                object_name=self.object_name,
-            )
-
-    def test_should_fail_on_wrong_input_type_object_name(self):
-        with self.assertRaises(AssertionError):
-            put(
-                client=self.client,
-                buffer=self.buffer,
-                bucket_name=self.bucket_name,
-                object_name=self.bad_string,
-            )
-
-    def test_should_fail_on_wrong_input_length_object_name(self):
-        with self.assertRaises(AssertionError):
-            put(
-                client=self.client,
-                buffer=self.buffer,
-                bucket_name=self.bucket_name,
-                object_name=self.len_0_string,
-            )

shared/mongodb/__init__.py

@@ -1,8 +0,0 @@
-from .src import classes, exceptions
-from .src.connect_mongodb import connect_mongodb
-from .src.count_documents import count_documents
-from .src.get_visual_communication import get_visual_communication
-from .src.list_names import list_names
-from .src.upsert_annotation import upsert_annotation
-from .src.upsert_prediction import upsert_prediction
-from .src.upsert_visual_communication import upsert_visual_communication

shared/mongodb/src/__init__.py

@@ -1,3 +0,0 @@
-"""Database utils module content."""
-
-from .connect_mongodb import connect_mongodb

shared/mongodb/src/classes/__init__.py

@@ -1,13 +0,0 @@
-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_communication import VisualCommunication
-from .visual_syntax import VisualSyntaxData

shared/mongodb/src/classes/angle.py

@@ -1,13 +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/mongodb/src/classes/contact.py

@@ -1,12 +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/mongodb/src/classes/distance.py

@@ -1,13 +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/mongodb/src/classes/framing.py

@@ -1,14 +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/mongodb/src/classes/information_value.py

@@ -1,13 +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/mongodb/src/classes/modality_color.py

@@ -1,13 +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/mongodb/src/classes/modality_depth.py

@@ -1,13 +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/mongodb/src/classes/modality_lighting.py

@@ -1,13 +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/mongodb/src/classes/model_data.py

@@ -1,115 +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_prediction_dict(
-        cls,
-        prediction_dict: dict,
-    ) -> ModelData:
-        """Instantiate from prediction dictionary."""
-        kwargs = {
-            'visual_syntax': VisualSyntaxData.from_tensor(
-                prediction_dict['visual_syntax'],
-            ),
-            'contact': ContactData.from_tensor(
-                prediction_dict['contact'],
-            ),
-            'angle': AngleData.from_tensor(
-                prediction_dict['angle'],
-            ),
-            'point_of_view': PointOfViewData.from_tensor(
-                prediction_dict['point_of_view'],
-            ),
-            'distance': DistanceData.from_tensor(
-                prediction_dict['distance'],
-            ),
-            'modality_lighting': ModalityLightingData.from_tensor(
-                prediction_dict['modality_lighting'],
-            ),
-            'modality_color': ModalityColorData.from_tensor(
-                prediction_dict['modality_color'],
-            ),
-            'modality_depth': ModalityDepthData.from_tensor(
-                prediction_dict['modality_depth'],
-            ),
-            'information_value': InformationValueData.from_tensor(
-                prediction_dict['information_value'],
-            ),
-            'framing': FramingData.from_tensor(
-                prediction_dict['framing'],
-            ),
-            'salience': SalienceData.from_tensor(
-                prediction_dict['salience'],
-            ),
-        }
-        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/mongodb/src/classes/point_of_view.py

@@ -1,12 +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/mongodb/src/classes/salience.py

@@ -1,15 +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/mongodb/src/classes/visual_communication.py

@@ -1,122 +0,0 @@
-"""Definition of VisualCommunication model."""
-
-from __future__ import annotations
-
-import logging
-from base64 import b64decode, b64encode
-from io import BytesIO
-from pathlib import Path
-
-from minio import Minio
-from PIL import Image
-from pydantic import BaseModel, ConfigDict
-from pymongo.collection import Collection
-
-from shared.datastore import get_image, put_image
-from shared.mongodb.src.classes import ModelData
-
-
-class VisualCommunication(BaseModel):
-    """Visual communication model."""
-
-    name: str
-    object_name: str
-    annotation: ModelData | None = None
-    prediction: ModelData | None = None
-    model_config = ConfigDict(arbitrary_types_allowed=True)
-
-    @classmethod
-    def classname(cls) -> str:
-        """Return classname."""
-        return cls.__name__
-
-    @classmethod
-    def upload_image_to_minio(
-        cls,
-        image: Image.Image,
-        minio_client: Minio,
-    ) -> str:
-        """Upload image to MinIO and return MD5 checksum of hashed image."""
-        assert isinstance(image, Image.Image)
-        assert isinstance(minio_client, Minio)
-        object_name = put_image(
-            client=minio_client,
-            image=image,
-        )
-        return object_name
-
-    @classmethod
-    def from_name_and_image(
-        cls,
-        name: str,
-        image: Image.Image,
-        minio_client: Minio,
-    ) -> VisualCommunication:
-        """Instantiate from filename and image that is automatically uploaded
-        to MinIO."""
-        assert isinstance(name, str)
-        assert isinstance(image, Image.Image)
-        assert isinstance(minio_client, Minio)
-        # upload file to minio
-        object_name = VisualCommunication.upload_image_to_minio(
-            image=image,
-            minio_client=minio_client,
-        )
-        return VisualCommunication(name=name, object_name=object_name)
-
-    @classmethod
-    def from_file(cls, path: Path, minio_client: Minio) -> VisualCommunication:
-        """Instantiate from file."""
-        assert isinstance(path, Path)
-        assert isinstance(minio_client, Minio)
-        # determine name
-        name = path.stem
-        # open image
-        image = Image.open(path)
-        image.load()
-        # instantiate object
-        return VisualCommunication.from_name_and_image(
-            name=name,
-            image=image,
-            minio_client=minio_client,
-        )
-
-    @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)))
-
-    def get_image(self, minio_client: Minio) -> Image.Image:
-        """Load image data from minio."""
-        assert isinstance(minio_client, Minio)
-        # get image from minio
-        image = get_image(
-            client=minio_client,
-            object_name=self.object_name,
-        )
-        return image
-
-    def save_to_mongo(self, collection: Collection) -> None:
-        """Save self as document in MongoDB."""
-        res = collection.insert_one(
-            document=self.model_dump(),
-        )
-        assert res.acknowledged
-
-    def webencoded_image(self, minio_client: Minio) -> str:
-        """Convert image to be displayed on webpage."""
-        assert isinstance(minio_client, Minio)
-        # get image from minio
-        image = self.get_image(minio_client)
-        # convert images to bytes string
-        buffer = BytesIO()
-        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/mongodb/src/connect_mongodb.py

@@ -1,30 +0,0 @@
-"""Definition of function to connect to database using environment
-variables."""
-
-import logging
-import os
-
-from dotenv import load_dotenv
-from pymongo import MongoClient
-
-
-def connect_mongodb():
-    """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/mongodb/src/count_documents.py

@@ -1,20 +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/mongodb/src/exceptions/__init__.py

@@ -1 +0,0 @@
-from .no_document_found import NoDocumentFoundException

shared/mongodb/src/exceptions/no_document_found.py

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

shared/mongodb/src/get_dataset.py

@@ -1,16 +0,0 @@
-"""Definition of get_dataset function."""
-
-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."""
-    assert isinstance(collection, Collection)
-    assert isinstance(type, str)
-    assert type in ['train', 'test', 'validation']
-    return []

shared/mongodb/src/get_visual_communication.py

@@ -1,40 +0,0 @@
-"""Definition of function to get visual communication from database."""
-
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from shared.mongodb.classes import VisualCommunication
-from shared.mongodb.exceptions import NoDocumentFoundException
-
-
-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/mongodb/src/list_names.py

@@ -1,30 +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/mongodb/src/upsert_annotation.py

@@ -1,30 +0,0 @@
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from shared.mongodb.classes 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/mongodb/src/upsert_prediction.py

@@ -1,30 +0,0 @@
-from __future__ import annotations
-
-import logging
-
-from pymongo.collection import Collection
-
-from shared.mongodb.classes 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/mongodb/src/upsert_visual_communication.py

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

@@ -1,18 +1,20 @@
-"""Definition of DataModel base class."""
+"""Definition of ValuesModel base class."""
+
+from __future__ import annotations
 
 import random
 
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel, ConfigDict
 from torch import Tensor
 
 
-class DataModel(BaseModel):
+class ValuesModel(BaseModel):
-    """DataModel base class."""
+    """ValuesModel base class."""
 
-    @classmethod
+    model_config = ConfigDict(
-    def classname(cls) -> str:
+        validate_assignment=True,  # argument type checking
-        """Return classname."""
+        frozen=True,  # ensure data immutability
-        return cls.__name__
+    )
 
     @classmethod
     def list_fields(cls) -> list[str]:
@@ -26,16 +28,16 @@ class DataModel(BaseModel):
         return cls(**kwargs)
 
     @classmethod
-    def from_choice(cls, option: str):
+    def from_choice(cls, option: str) -> ValuesModel:
         """Instantiate from choice."""
-        if option is None:
+        assert isinstance(option, str)
-            raise ValidationError()
+        assert len(option) > 0
-        assert isinstance(option, str), 'option is not a string'
         allowed_options_list = cls.list_fields()
-        assert (
+        if option not in allowed_options_list:
-            option in allowed_options_list
+            raise ValueError(f'option {option} must be in {allowed_options_list}')
-        ), f"{option} is not among allowed fields {allowed_options_list}"
+        # generate field values
-        kwargs = {field: 0 for field in cls.list_fields()}
+        kwargs = {field: 0 for field in allowed_options_list}
+        # set chosen value to max probability
         kwargs[option] = 1
         return cls(**kwargs)
 
@@ -47,15 +49,6 @@ class DataModel(BaseModel):
         kwargs = dict(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()

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

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

@@ -0,0 +1,49 @@
+"""Definition of VisualCommunicationValues DTO."""
+
+from __future__ import annotations
+
+from ..type_checking_base_model import TypeCheckingBaseModel
+from .angle_values import AngleValues
+from .contact_values import ContactValues
+from .distance_values import DistanceValues
+from .framing_values import FramingValues
+from .information_value_values import InformationValueValues
+from .modality_color_values import ModalityColorValues
+from .modality_depth_values import ModalityDepthValues
+from .modality_lighting_values import ModalityLightingValues
+from .point_of_view_values import PointOfViewValues
+from .salience_values import SalienceValues
+from .visual_syntax_values import VisualSyntaxValues
+
+
+class VisualCommunicationValues(TypeCheckingBaseModel):
+    """Visual communication values class."""
+
+    visual_syntax: VisualSyntaxValues
+    contact: ContactValues
+    angle: AngleValues
+    point_of_view: PointOfViewValues
+    distance: DistanceValues
+    modality_lighting: ModalityLightingValues
+    modality_color: ModalityColorValues
+    modality_depth: ModalityDepthValues
+    information_value: InformationValueValues
+    framing: FramingValues
+    salience: SalienceValues
+
+    @classmethod
+    def from_random(cls) -> VisualCommunicationValues:
+        """Create a random instance."""
+        return cls(
+            visual_syntax=VisualSyntaxValues.from_random(),
+            contact=ContactValues.from_random(),
+            angle=AngleValues.from_random(),
+            point_of_view=PointOfViewValues.from_random(),
+            distance=DistanceValues.from_random(),
+            modality_lighting=ModalityLightingValues.from_random(),
+            modality_color=ModalityColorValues.from_random(),
+            modality_depth=ModalityDepthValues.from_random(),
+            information_value=InformationValueValues.from_random(),
+            framing=FramingValues.from_random(),
+            salience=SalienceValues.from_random(),
+        )

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

@@ -1,12 +1,10 @@
-"""Definition of VisualSyntaxData data model."""
+"""Definition of VisualSyntaxValues DTO."""
 
-from __future__ import annotations
+from .values_model import ValuesModel
-
-from .data_model import DataModel
 
 
-class VisualSyntaxData(DataModel):
+class VisualSyntaxValues(ValuesModel):
-    """VisualSyntaxData data model."""
+    """Visual syntax values DTO."""
 
     non_transactional_action: float
     non_transactional_reaction: float

shared/repositories/src/image_repository.py

@@ -0,0 +1,74 @@
+"""Definition of ImageRepository class."""
+
+from __future__ import annotations
+
+from io import BytesIO
+from pathlib import Path
+
+from PIL import Image
+
+from .dto import ImageData
+from .implementations import MinioImplementation
+from .interfaces import ImageInterface
+
+
+class ImageRepository(ImageInterface, MinioImplementation):
+    """Image repository class that handles CRUD functionality for
+    VisualData."""
+
+    def __enter__(self) -> ImageRepository:
+        self.connect()
+        return self
+
+    @staticmethod
+    def _build_path(image_name: str) -> Path:
+        """Build object path."""
+        assert isinstance(image_name, str)
+        path = Path('images') / image_name
+        return path
+
+    def get_data(self, image_name: str) -> ImageData | None:
+        """Get Visual data."""
+        assert isinstance(image_name, str)
+        assert len(image_name) > 0
+        # build path
+        path = self._build_path(image_name)
+        # get object from bucket
+        buffer = self._get(path)
+        # handle if no data found
+        if not buffer:
+            return None
+        # convert data
+        image = Image.open(buffer)
+        data = ImageData(image=image, name=image_name)
+        return data
+
+    def put_data(self, data: ImageData) -> None:
+        """Put visual data."""
+        assert isinstance(data, ImageData)
+        # build path
+        path = self._build_path(data.name)
+        # save image to buffer
+        buffer = BytesIO()
+        data.image.save(buffer, 'png')
+        # put object in bucket
+        self._put(path, buffer)
+
+    def remove_data(self, image_name: str) -> None:
+        """Remove visual data."""
+        assert isinstance(image_name, str)
+        assert len(image_name) > 0
+        # build path
+        path = self._build_path(image_name)
+        # remove object
+        self._delete(path)
+
+    def list_names(self) -> list[str]:
+        """List names of all images."""
+        # build path
+        path = self._build_path('')
+        # list object paths
+        obj_path_list = self._list_objects(path)
+        # strip prefix
+        name_list = [obj_path.split('/')[-1] for obj_path in obj_path_list]
+        return name_list

shared/repositories/src/implementations/__init__.py

@@ -0,0 +1,2 @@
+from .minio_implementation import MinioImplementation
+from .mongo_implementation import MongoImplementation

shared/repositories/src/implementations/minio_implementation.py

@@ -0,0 +1,186 @@
+"""MinIO implementation."""
+
+from __future__ import annotations
+
+import logging
+import os
+from io import BytesIO
+from pathlib import Path
+from traceback import format_exc
+
+from minio import Minio
+
+from shared.utils import check_env
+
+from ..interfaces import DatabaseInterface
+
+
+class MinioImplementation(DatabaseInterface):
+    """MinIO basic CRUD implementation."""
+
+    def __init__(self):
+        # ensure necessary env vars available
+        var_list = {
+            'MINIO_ENDPOINT',
+            'MINIO_ACCESS_KEY',
+            'MINIO_SECRET_KEY',
+            'MINIO_BUCKET_NAME',
+        }
+        check_env(var_list)
+        # prepare internal variables
+        self._client: Minio | None = None
+        self._bucket_name: str | None = None
+
+    def connect(self):
+        """Connect to MinIO server."""
+        # prepare arguments
+        minio_endpoint = str(os.getenv('MINIO_ENDPOINT'))
+        minio_access_key = str(os.getenv('MINIO_ACCESS_KEY'))
+        minio_secret_key = str(os.getenv('MINIO_SECRET_KEY'))
+        minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
+        # 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):
+            logging.debug('creating bucket: %s', minio_bucket_name)
+            client.make_bucket(bucket_name=minio_bucket_name)
+        # persist state
+        self._client = client
+        self._bucket_name = minio_bucket_name
+
+    def close(self) -> None:
+        """Close connection to MinIO server.
+
+        N.B. MinIO connection cannot be closed manually.
+        """
+        self._client = None
+        self._bucket_name = None
+
+    def connected(self):
+        """Check connection to Minio."""
+        res = isinstance(self._client, Minio)
+        logging.debug(res)
+        return res
+
+    def __enter__(self) -> MinioImplementation:
+        self.connect()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        if any(
+            (
+                exc_type is not None,
+                exc_val is not None,
+                exc_tb is not None,
+            ),
+        ):
+            logging.error('error while exiting context')
+        self.close()
+
+    def _put(
+        self,
+        path: Path,
+        buffer: BytesIO,
+    ) -> None:
+        """Save in-memory buffer as object in MinIO."""
+        assert isinstance(path, Path)
+        assert isinstance(buffer, BytesIO)
+        assert isinstance(self._client, Minio)
+        assert isinstance(self._bucket_name, str)
+        # prepare for saving
+        num_bytes = len(buffer.getvalue())
+        buffer.seek(0)
+        # send data to bucket
+        try:
+            self._client.put_object(
+                bucket_name=self._bucket_name,
+                object_name=path.as_posix(),
+                length=num_bytes,
+                data=buffer,
+            )
+            logging.debug('saved data to %s', path)
+        except Exception as exc:
+            logging.error('failed saving data to MinIO')
+            raise exc
+
+    def _get(
+        self,
+        path: Path,
+    ) -> BytesIO | None:
+        """Get object from MinIO as in-memory buffer."""
+        assert isinstance(path, Path)
+        assert isinstance(self._client, Minio)
+        assert isinstance(self._bucket_name, str)
+        try:
+            # make request
+            response = self._client.get_object(
+                bucket_name=self._bucket_name,
+                object_name=path.as_posix(),
+            )
+            assert response.status == 200
+            # get buffer
+            buffer = BytesIO()
+            chunk_size = 2**14
+            while chunk := response.read(chunk_size):
+                buffer.write(chunk)
+            buffer.seek(0)
+            logging.debug('got %s', path)
+            return buffer
+        except Exception:
+            logging.error('failed getting data from MinIO')
+            logging.debug(format_exc())
+            return None
+        finally:
+            # close connection if established
+            if 'response' in locals():
+                response.close()
+                response.release_conn()
+
+    def _delete(
+        self,
+        path: Path,
+    ) -> None:
+        """Delete object from MinIO."""
+        assert isinstance(path, Path)
+        assert isinstance(self._client, Minio)
+        assert isinstance(self._bucket_name, str)
+        # remove object
+        try:
+            self._client.remove_object(
+                bucket_name=self._bucket_name,
+                object_name=path.as_posix(),
+            )
+            logging.debug('deleted %s', path)
+        except Exception as exc:
+            logging.error('failed deleting %s', path)
+            logging.debug(format_exc())
+            raise exc
+
+    def _list_objects(
+        self,
+        path: Path,
+    ) -> list[str]:
+        """List objects in bucket under path."""
+        assert isinstance(path, Path)
+        assert isinstance(self._client, Minio)
+        assert isinstance(self._bucket_name, str)
+        try:
+            # list objects
+            obj_list = self._client.list_objects(
+                bucket_name=self._bucket_name,
+                prefix=path.as_posix(),
+                recursive=True,
+            )
+            # extract info
+            name_list = [obj.object_name for obj in obj_list]
+            logging.debug('got %s objects matching %s', len(name_list), path)
+            return name_list
+        except Exception as exc:
+            logging.error('failed listing objects under %s', path)
+            logging.debug(format_exc())
+            raise exc

shared/repositories/src/implementations/mongo_implementation.py

@@ -0,0 +1,145 @@
+"""Mongo implementation."""
+
+from __future__ import annotations
+
+import logging
+import os
+import traceback
+
+from pymongo import MongoClient
+from pymongo.collection import Collection
+from pymongo.database import Database
+from pymongo.errors import ServerSelectionTimeoutError
+
+from shared.utils import check_env
+
+from ..interfaces import DatabaseInterface
+
+
+class MongoImplementation(DatabaseInterface):
+    """MongoDB basic CRUD implementation."""
+
+    def __init__(self):
+        # ensure necessary env vars available
+        var_list = {
+            'MONGO_ENDPOINT',
+            'MONGO_DB',
+            'MONGO_COLLECTION',
+        }
+        check_env(var_list)
+        # prepare internal variables
+        self.client: MongoClient | None = None
+        self.db: Database | None = None
+        self.collection: Collection | None = None
+
+    def connect(self) -> None:
+        """Connect to Mongo server."""
+        # prepare arguments
+        mongo_endpoint = str(os.getenv('MONGO_ENDPOINT'))
+        mongo_database = str(os.getenv('MONGO_DB'))
+        mongo_collection = str(os.getenv('MONGO_COLLECTION'))
+        # connect client
+        client: MongoClient = MongoClient(mongo_endpoint)
+        database = client[mongo_database]
+        collection = database[mongo_collection]
+        # set unique index on 'name'
+        collection.create_index(keys='name', unique=True)
+        # persist state
+        self._client = client
+        self._database = database
+        self._collection = collection
+
+    def close(self) -> None:
+        """Close connection to Mongo server."""
+        self._client.close()
+        self._client = None  # type: ignore
+        self._database = None  # type: ignore
+        self._collection = None  # type: ignore
+
+    def connected(self) -> bool:
+        """Check connection to Mongo."""
+        if self._client is None:
+            return False
+        try:
+            # trigger fetch data
+            _ = self._client.server_info()
+            res = True
+        except ServerSelectionTimeoutError:
+            res = False
+        logging.debug(res)
+        return res
+
+    def __enter__(self) -> MongoImplementation:
+        self.connect()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        if any(
+            (
+                exc_type is not None,
+                exc_val is not None,
+                exc_tb is not None,
+            ),
+        ):
+            logging.error('error while exiting context')
+            traceback.print_exception(exc_type, exc_val, exc_tb)
+        self.close()
+
+    def _save(
+        self,
+        data: dict,
+        query: dict,
+    ) -> None:
+        """Save document in Mongo."""
+        assert isinstance(data, dict)
+        assert isinstance(query, dict)
+        assert 'name' in data
+        assert self.connected()
+        self._collection.update_one(
+            filter=query,
+            update={
+                '$set': data.copy(),
+            },
+            upsert=True,
+        )
+        logging.debug('Save %s', data)
+
+    def _get(
+        self,
+        query: dict,
+    ) -> dict | None:
+        """Get document from Mongo."""
+        assert isinstance(query, dict)
+        assert self.connected()
+        doc = self._collection.find_one(query, projection={'_id': False})
+        logging.debug('Found %s', doc)
+        return doc
+
+    def _delete(
+        self,
+        query: dict,
+    ) -> None:
+        """Remove document from Mongo."""
+        assert isinstance(query, dict)
+        assert self.connected()
+        doc = self._collection.delete_one(query)
+        logging.debug('Deleted %s', doc)
+
+    def _list_documents(self, key='name') -> list[str]:
+        """List documents in Mongo."""
+        assert isinstance(key, str)
+        assert len(key) > 0
+        # build query
+        doc_list = list(
+            self._collection.find(
+                filter={},
+                projection={
+                    '_id': False,
+                    key: True,
+                },
+            ),
+        )
+        # extract values
+        value_list = [doc[key] for doc in doc_list]
+        logging.debug('Got %s document(s)', len(doc_list))
+        return value_list

shared/repositories/src/interfaces/__init__.py

@@ -0,0 +1,4 @@
+from .database_interface import DatabaseInterface
+from .image_interface import ImageInterface
+from .model_interface import ModelInterface
+from .visual_communication_interface import VisualCommunicationInterface