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

Reviewed-on: #69

.gitea/workflows/pull.yaml

@@ -40,7 +40,7 @@ jobs:
       - name: Pytest & Calculate Coverage
         env:
           MINIO_ENDPOINT: ${{ vars.MINIO_ENDPOINT }}
-          MINIO_ACCESS_KEY: ${{ secrets.MINIO_ACCESS_KEY }}
+          MINIO_ACCESS_KEY: ${{ vars.MINIO_ACCESS_KEY }}
           MINIO_SECRET_KEY: ${{ secrets.MINIO_SECRET_KEY }}
           MONGO_ENDPOINT: ${{ vars.MONGO_ENDPOINT }}
         run: |

misc/generate_random_prediction.py

@@ -1,42 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-
-from shared.datastore import Datastore
-from shared.docstore import connect_mongodb
-from shared.docstore.src.classes import VisualCommunication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-if __name__ == '__main__':
-    # load in env file
-    env_path = Path(__file__).parent.parent / 'server.env'
-    assert env_path.exists()
-    load_dotenv(env_path)
-    # ensure env vars set
-    check_env(NECESSARY_ENV_VAR_LIST)
-    # setup logging
-    setup_logging()
-    # connect to minIO
-    datastore = Datastore()
-    datastore.connect()
-    assert datastore._client is not None
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # 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=datastore._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,31 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-
-from shared.datastore import Datastore
-from shared.docstore 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
-    datastore = Datastore()
-    datastore.connect()
-    assert datastore._client is not None
-    # 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=datastore._client)
-    image.show()

misc/image_upload.py

@@ -1,48 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-from pymongo.errors import DuplicateKeyError
-
-from shared.datastore import Datastore
-from shared.docstore import connect_mongodb
-from shared.docstore.src.classes import VisualCommunication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-if __name__ == '__main__':
-    # load in env file
-    env_path = Path(__file__).parent.parent / 'server.env'
-    assert env_path.exists()
-    load_dotenv(env_path)
-    # ensure env vars set
-    check_env(NECESSARY_ENV_VAR_LIST)
-    # setup logging
-    setup_logging()
-    # connect to minIO
-    datastore = Datastore()
-    datastore.connect()
-    assert datastore._client is not None
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # 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=datastore._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,53 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-from pymongo.errors import DuplicateKeyError
-
-from shared.datastore import Datastore
-from shared.docstore import connect_mongodb
-from shared.docstore.src.classes import VisualCommunication
-from shared.utils import check_env, setup_logging
-from web_ui.src.main import NECESSARY_ENV_VAR_LIST
-
-if __name__ == '__main__':
-    # load in env file
-    env_path = Path(__file__).parent.parent / 'server.env'
-    assert env_path.exists()
-    load_dotenv(env_path)
-    # ensure env vars set
-    check_env(NECESSARY_ENV_VAR_LIST)
-    # setup logging
-    setup_logging()
-    # connect to minIO
-    datastore = Datastore()
-    datastore.connect()
-    assert datastore._client is not None
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # 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=datastore._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 Datastore
-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
-    datastore = Datastore()
-    datastore.connect()
-    # instantiate model
-    model = VisualCommunicationModel().to(DEVICE)
-    # show model weights
-    summary(model)
-    # put buffer in minio bucket
-    hash_str = datastore.put_model(
-        model=model,
-    )
-    print(f"hash string: {hash_str}")
-    print('saved to Minio')

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 Datastore
-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
-    datastore = Datastore()
-    datastore.connect()
-    # instantiate model
-    model = VisualCommunicationModel(download_resnet_weights=True)
-    # show model weights
-    summary(model)
-    # put buffer in minio bucket
-    hash_str = datastore.put_model(
-        model=model,
-    )
-    print(f"hash string: {hash_str}")
-    print('saved to Minio')

misc/prediction_upload.py

@@ -1,48 +0,0 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-from dotenv import load_dotenv
-
-from shared.datastore import Datastore
-from shared.docstore import connect_mongodb, upsert_prediction
-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
-
-if __name__ == '__main__':
-    # load in env file
-    env_path = Path(__file__).parent.parent / 'server.env'
-    assert env_path.exists()
-    load_dotenv(env_path)
-    # ensure env vars set
-    check_env(NECESSARY_ENV_VAR_LIST)
-    # setup logging
-    setup_logging()
-    # connect to minIO
-    datastore = Datastore()
-    datastore.connect()
-    assert datastore._client is not None
-    # connect to MongoDB
-    collection, db, client = connect_mongodb()
-    # 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=datastore._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,
-        )

model/src/main.py

@@ -9,22 +9,16 @@ from models import VisualCommunicationModel
 from tqdm import tqdm
 from utils import DEVICE, VCDADataset, load_model
 
-from shared.datastore import Datastore
-from shared.docstore.classes import ModelData
 from shared.utils import setup_logging
 
 if __name__ == '__main__':
     # setup logging
     setup_logging()
-    # connect to minio
-    datastore = Datastore()
-    datastore.connect()
     # instantiate model
-    model: VisualCommunicationModel = load_model(client=datastore._client)
+    model: VisualCommunicationModel = load_model()
     model.eval()
     # setup dataset
     dataset = VCDADataset(
-        minio_client=datastore._client,
         data_name_list=[
             '02dbaf48d713e4e6d3a6b98fd2dc866e',
         ],
@@ -40,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.docstore.src.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

@@ -6,29 +6,28 @@ from pathlib import Path
 import torch
 
 from model.src.models import VisualCommunicationModel
-from shared.datastore import Datastore
+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:
-    datastore: Datastore,
-) -> VisualCommunicationModel:
     """Instantiate model with weights loaded from latest model saved in
     MinIO."""
-    assert isinstance(datastore, Datastore)
     # 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 = datastore.get_model(
+    with ModelRepository() as repo:
-        object_name=model_object_name,
+        model_data = repo.get_data(model_object_name)
-    )
+    if model_data is None:
+        raise FileNotFoundError(f'model {model_object_name} not found')
+    model_checkpoint = torch.load(model_data.buffer)
     model.load_state_dict(model_checkpoint)
     # clean memory
     model_checkpoint.clear()

model/src/utils/vcda_dataset.py

@@ -15,7 +15,7 @@ from torchvision.transforms.functional import (
     to_tensor,
 )
 
-from shared.datastore import Datastore
+from shared.repositories import ImageRepository
 
 # resnet18 original normalization values
 RESNET_NORMALIZE_MEAN = [0.485, 0.456, 0.406]
@@ -27,13 +27,11 @@ class VCDADataset(Dataset):
 
     def __init__(
         self,
-        datastore: Datastore,
         data_name_list: list[str],
         do_augment: bool = False,
         random_annotations: bool = False,
     ):
         super().__init__()
-        self.datastore = datastore
         self.data_name_list = data_name_list
         self.do_augment = do_augment
         self.random_annotations = random_annotations
@@ -55,11 +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 = self.datastore.get_image(
+        with ImageRepository() as repo:
-            object_name=object_name,
+            image_data = repo.get_data(image_name)
-        )
+        assert image_data is not None
-        tensor = self.image_to_tensor(image)
+        tensor = self.image_to_tensor(image_data.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 Datastore
 
 
 def parse_arguments():
@@ -57,19 +56,16 @@ optimizer = torch.optim.Adam(model.parameters(), lr=args.lr)
 loss_fn = nn.CrossEntropyLoss()
 
 # create datasets and loaders
-datastore = Datastore()
-datastore.connect()
 with open('model/src/dataset/train.csv', encoding='utf-8') as fh:
     train_data_name_list = fh.read().split('\n')
 train_dataset = VCDADataset(
-    datastore=datastore,
     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(datastore=datastore, 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
@@ -141,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'],
@@ -155,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,42 +0,0 @@
-"""Script to move minio images to subfolder."""
-
-import os
-from pathlib import Path
-
-from dotenv import load_dotenv
-from PIL import Image
-
-from shared.datastore import Datastore
-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
-    datastore = Datastore()
-    datastore.connect()
-    assert datastore._client is not None
-    # list images in bucket
-    BUCKET_NAME = os.getenv(
-        'MINIO_BUCKET_NAME',
-        default='visual-critical-discourse-analysis',
-    )
-    obj_list = datastore._client.list_objects(
-        bucket_name=BUCKET_NAME,
-    )
-    # begin moving images
-    for obj in obj_list:
-        # get image from minio
-        buffer = datastore._get(
-            object_name=obj.object_name,
-        )
-        # convert data to image
-        image = Image.open(buffer)
-        # put image into minio subfolder
-        _ = datastore.put_image(
-            image=image,
-        )

pyproject.toml

@@ -112,7 +112,7 @@ module = "utils.*"
 ignore_missing_imports = true
 
 [[tool.mypy.overrides]]
-module = "datastore.*"
+module = "shared.datastore.*"
 ignore_missing_imports = true
 
 [[tool.mypy.overrides]]

shared/datastore/__init__.py

@@ -1 +0,0 @@
-from .src import Datastore

shared/datastore/src/__init__.py

@@ -1 +0,0 @@
-from .datastore_minio import DatastoreMinio as Datastore

shared/datastore/src/datastore_interface.py

@@ -1,69 +0,0 @@
-"""Definition of datastore interface."""
-
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from collections import OrderedDict
-
-from PIL import Image
-from torch.nn import Module
-
-
-class DatastoreInterface(ABC):
-    """Datastore interface class."""
-
-    @abstractmethod
-    def connect(
-        self,
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def close(
-        self,
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def __enter__(
-        self,
-    ) -> DatastoreInterface:
-        self.connect()
-        return self
-
-    @abstractmethod
-    def __exit__(
-        self,
-        exc_type,
-        exc_val,
-        exc_tb,
-    ) -> None:
-        self.close()
-
-    @abstractmethod
-    def put_image(
-        self,
-        image: Image.Image,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def get_image(
-        self,
-        object_name: str,
-    ) -> Image.Image:
-        pass
-
-    @abstractmethod
-    def put_model(
-        self,
-        model: Module,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def get_model(
-        self,
-        object_name: str,
-    ) -> OrderedDict:
-        pass

shared/datastore/src/datastore_minio.py

@@ -1,241 +0,0 @@
-"""Definition of datastore minio implementation."""
-
-from __future__ import annotations
-
-import logging
-import os
-from collections import OrderedDict
-from hashlib import md5
-from io import BytesIO
-from traceback import format_exc
-
-import torch
-from minio import Minio
-from PIL import Image
-
-from shared.utils import check_env
-
-from .datastore_interface import DatastoreInterface
-
-
-class DatastoreMinio(DatastoreInterface):
-    """Datastore interface."""
-
-    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) -> None:
-        """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)
-        logging.debug('finished')
-        # 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 __enter__(self) -> DatastoreMinio:
-        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,
-        object_name: str,
-        buffer: BytesIO,
-    ) -> None:
-        """Save in-memory buffer as object in Minio."""
-        assert isinstance(object_name, str)
-        assert len(object_name) > 0
-        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=object_name,
-                length=num_bytes,
-                data=buffer,
-            )
-            logging.debug('saved data to %s', object_name)
-        except Exception as exc:
-            logging.error('failed saving data to MinIO')
-            raise exc
-
-    def _get(
-        self,
-        object_name: str,
-    ) -> BytesIO:
-        """Get object from Minio as in-memory buffer."""
-        assert isinstance(object_name, str)
-        assert len(object_name) > 0
-        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=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')
-            logging.debug(format_exc())
-            raise exc
-        finally:
-            # close connection if established
-            if 'response' in locals():
-                response.close()
-                response.release_conn()
-
-    def _delete(
-        self,
-        object_name: str,
-    ) -> None:
-        """Delete object from Minio."""
-        assert isinstance(object_name, str)
-        assert len(object_name) > 0
-        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=object_name,
-            )
-            logging.debug('deleted %s', object_name)
-        except Exception as exc:
-            logging.error('failed deleting %s', object_name)
-            logging.debug(format_exc())
-            raise exc
-
-    def put_image(
-        self,
-        image: Image.Image,
-    ) -> str:
-        """Put image in Minio."""
-        assert isinstance(image, Image.Image)
-        # save data to buffer
-        buffer = BytesIO()
-        image.save(buffer, 'png')
-        # get md5 of buffer
-        checksum = md5(buffer.getbuffer()).hexdigest()
-        # build object path
-        object_path = f'images/{checksum}'
-        # send data to bucket
-        self._put(
-            object_name=object_path,
-            buffer=buffer,
-        )
-        logging.debug('saved data to %s', object_path)
-        return checksum
-
-    def get_image(
-        self,
-        object_name: str,
-    ) -> Image.Image:
-        """Get image from Minio."""
-        assert isinstance(object_name, str)
-        assert len(object_name) > 0
-        # build object path
-        object_path = f'images/{object_name}'
-        # get object from bucket
-        buffer = self._get(
-            object_name=object_path,
-        )
-        # convert data to image
-        image = Image.open(buffer)
-        logging.debug('got data from %s', object_path)
-        return image
-
-    def put_model(
-        self,
-        model: torch.nn.Module,
-    ) -> str:
-        """Put model in Minio."""
-        assert isinstance(model, torch.nn.Module)
-        # save data to buffer
-        buffer = BytesIO()
-        torch.save(model.state_dict(), buffer)
-        # get md5 of image
-        checksum = md5(buffer.getbuffer()).hexdigest()
-        # build object path
-        object_path = f'models/{checksum}'
-        # send data to bucket
-        self._put(
-            object_name=object_path,
-            buffer=buffer,
-        )
-        logging.debug('saved data to %s', object_path)
-        return checksum
-
-    def get_model(
-        self,
-        object_name: str,
-    ) -> OrderedDict:
-        """Get model data from Minio."""
-        assert isinstance(object_name, str)
-        assert len(object_name) > 0
-        # build object path
-        object_path = f'models/{object_name}'
-        # get object from bucket
-        buffer = self._get(
-            object_name=object_path,
-        )
-        # convert data to model checkpoint
-        model_content = torch.load(buffer)
-        logging.debug('got data from %s', object_path)
-        return model_content

shared/datastore/tests/integration/base_crud_test.py

@@ -1,112 +0,0 @@
-"""Integration tests related to base CRUD functions."""
-
-import logging
-import os
-from io import BytesIO
-
-import minio
-import pytest
-
-from shared.datastore import Datastore
-
-
-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
-    if len(data_a_bytes) != len(data_b_bytes):
-        logging.error(
-            'data has different length: %s and %s',
-            len(data_a_bytes),
-            len(data_b_bytes),
-        )
-        return False
-    # compare content
-    if data_a_bytes != data_b_bytes:
-        logging.error('data has different bytes')
-        return False
-    return True
-
-
-def test_should_get_data(
-    datastore: Datastore,
-    data_in_minio,
-):
-    # ARRANGE
-    data, _, object_name = data_in_minio
-    # ACT
-    received_data = datastore._get(
-        object_name=object_name,
-    )
-    # ASSERT
-    assert isinstance(data, BytesIO)
-    assert same_data(data, received_data)
-
-
-def test_should_delete_data(
-    datastore: Datastore,
-    data_in_minio,
-):
-    # ARRANGE
-    _, _, object_name = data_in_minio
-    # ACT
-    datastore._delete(
-        object_name=object_name,
-    )
-    # ASSERT
-    with pytest.raises(minio.error.S3Error):
-        _ = datastore._get(
-            object_name=object_name,
-        )
-
-
-def test_should_put_data(
-    datastore: Datastore,
-    data,
-):
-    # ARRANGE
-    minio_object_name = str(os.getenv('MINIO_OBJECT_NAME'))
-    buffer = BytesIO(data)
-    # ACT
-    datastore._put(
-        object_name=minio_object_name,
-        buffer=buffer,
-    )
-    received_data = datastore._get(
-        object_name=minio_object_name,
-    )
-    # ASSERT
-    assert isinstance(received_data, BytesIO)
-    assert same_data(received_data, buffer)
-
-
-def test_should_update_data(
-    datastore: Datastore,
-    data_in_minio,
-):
-    # ARRANGE
-    buffer, _, object_name = data_in_minio
-    # ACT
-    datastore._put(
-        object_name=object_name,
-        buffer=buffer,
-    )
-    received_data = datastore._get(
-        object_name=object_name,
-    )
-    # ASSERT
-    assert same_data(received_data, buffer)
-
-
-if __name__ == '__main__':
-    pytest.main()

shared/datastore/tests/integration/conftest.py

@@ -1,196 +0,0 @@
-"""Integration test configurations."""
-
-import os
-import random
-from collections.abc import Iterator
-from hashlib import md5
-from io import BytesIO
-from pathlib import Path
-
-import pytest
-import torch
-from dotenv import load_dotenv
-from PIL import Image
-
-from model.src.models import VisualCommunicationModel
-from shared.datastore import Datastore
-
-env_var_map = {
-    'MINIO_BUCKET_NAME': 'test-bucket',
-    'MINIO_OBJECT_NAME': '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX',
-}
-
-
-@pytest.fixture(scope='session', autouse=True)
-def populate_env(
-    request: pytest.FixtureRequest,
-) -> None:
-    """Populate environment with variables used for testing."""
-    # read env-file for local testing
-    load_dotenv(
-        dotenv_path=Path(__file__).parent.parent.parent.parent.parent / 'server.env',
-    )
-    # 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 datastore(
-    populate_env,
-) -> Iterator[Datastore]:
-    # prepare arguments
-    minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    # connect to minio
-    datastore_client = Datastore()
-    datastore_client.connect()
-    assert datastore_client._client is not None
-    # ensure bucket exists
-    if not datastore_client._client.bucket_exists(bucket_name=minio_bucket_name):
-        datastore_client._client.make_bucket(bucket_name=minio_bucket_name)
-    # expose client
-    yield datastore_client
-    # remove objects left behind by tests
-    for obj in datastore_client._client.list_objects(
-        bucket_name=minio_bucket_name,
-        recursive=True,
-    ):
-        datastore_client._client.remove_object(
-            bucket_name=obj.bucket_name,
-            object_name=obj.object_name,
-        )
-    # remove bucket
-    datastore_client._client.remove_bucket(bucket_name=minio_bucket_name)
-    assert not datastore_client._client.bucket_exists(bucket_name=minio_bucket_name)
-    # disconnect from minio
-    datastore_client.close()
-
-
-@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(
-    datastore: Datastore,
-    data: bytes,
-) -> Iterator[tuple[BytesIO, str, str]]:
-    assert datastore._client is not None
-    # 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 = len(buffer.getvalue())
-    buffer.seek(0)
-    # send data to bucket
-    datastore._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
-    datastore._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(
-    datastore: Datastore,
-    image: Image.Image,
-) -> Iterator[tuple[Image.Image, str]]:
-    assert datastore._client is not None
-    # prepare arguments
-    minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
-    # save data to buffer
-    buffer = BytesIO()
-    image.save(buffer, 'png')
-    # get md5 of buffer
-    checksum = md5(buffer.getbuffer()).hexdigest()
-    # build object path
-    object_path = f'images/{checksum}'
-    # prepare for saving
-    num_bytes = len(buffer.getvalue())
-    buffer.seek(0)
-    # send data to bucket
-    datastore._client.put_object(
-        bucket_name=minio_bucket_name,
-        object_name=object_path,
-        length=num_bytes,
-        data=buffer,
-    )
-    # expose image and object name
-    yield image, checksum
-    # cleanup
-    datastore._client.remove_object(
-        bucket_name=minio_bucket_name,
-        object_name=object_path,
-    )
-
-
-@pytest.fixture
-def model() -> Iterator[torch.nn.Module]:
-    # generate model
-    model = VisualCommunicationModel().to('cpu')
-    # expose model
-    yield model
-
-
-@pytest.fixture
-def model_in_minio(
-    datastore: Datastore,
-    model: torch.nn.Module,
-) -> Iterator[tuple[torch.nn.Module, str]]:
-    assert datastore._client is not None
-    # prepare arguments
-    minio_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()
-    # build object path
-    object_path = f'models/{checksum}'
-    # prepare for saving
-    num_bytes = len(buffer.getvalue())
-    buffer.seek(0)
-    # send data to bucket
-    datastore._client.put_object(
-        bucket_name=minio_bucket_name,
-        object_name=object_path,
-        length=num_bytes,
-        data=buffer,
-    )
-    # expose model and object name
-    yield model, checksum
-    # cleanup
-    datastore._client.remove_object(
-        bucket_name=minio_bucket_name,
-        object_name=object_path,
-    )

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 Datastore
-
-
-def test_should_return_correct_type():
-    with Datastore() as ds:
-        assert isinstance(ds._client, Minio)

shared/datastore/tests/integration/image_crud_test.py

@@ -1,82 +0,0 @@
-"""Integration tests related to image CRUD."""
-
-import numpy as np
-import pytest
-from PIL import Image
-
-from shared.datastore import Datastore
-
-
-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(
-    datastore: Datastore,
-    image_in_minio: tuple[Image.Image, str],
-):
-    # ARRANGE
-    image, object_name = image_in_minio
-    # ACT
-    received_image = datastore.get_image(
-        object_name=object_name,
-    )
-    # ASSERT
-    assert isinstance(image, Image.Image)
-    assert same_image(image, received_image)
-
-
-def test_should_put_image(
-    datastore: Datastore,
-    image: Image.Image,
-):
-    # ARRANGE
-    object_name = datastore.put_image(
-        image=image,
-    )
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # ACT
-    received_image = datastore.get_image(
-        object_name=object_name,
-    )
-    # ASSERT
-    assert same_image(image, received_image)
-
-
-def test_should_update_image(
-    datastore: Datastore,
-    image_in_minio: tuple[Image.Image, str],
-):
-    # ARRANGE
-    image, object_name = image_in_minio
-    object_name = datastore.put_image(
-        image=image,
-    )
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # ACT
-    received_image = datastore.get_image(
-        object_name=object_name,
-    )
-    # ASSERT
-    assert same_image(image, received_image)
-
-
-if __name__ == '__main__':
-    pytest.main()

shared/datastore/tests/integration/model_crud_test.py

@@ -1,90 +0,0 @@
-"""Integration tests related to model CRUD."""
-
-from collections import OrderedDict
-
-from torch.nn import Module
-
-from model.src.models import VisualCommunicationModel
-from shared.datastore import Datastore
-
-
-def same_model(
-    model_a: Module,
-    model_b: Module,
-) -> bool:
-    """Check if two models are the same class, have the same number of
-    parameters and contain the same weights."""
-    assert isinstance(model_a, Module)
-    assert isinstance(model_b, Module)
-    # compare model classes
-    assert type(model_a) is type(model_b)
-    # compare number of parameters
-    params_a = list(model_a.parameters())
-    params_b = list(model_b.parameters())
-    if len(params_a) != len(params_b):
-        return False
-    # compare model weights
-    for p_a, p_b in zip(params_a, params_b):
-        if p_a.data.ne(p_b.data).sum() > 0:
-            return False
-    return True
-
-
-def test_should_get_model(
-    datastore: Datastore,
-    model_in_minio: tuple[Module, str],
-):
-    # ARRANGE
-    model, object_name = model_in_minio
-    # ACT
-    model_data = datastore.get_model(
-        object_name=object_name,
-    )
-    assert isinstance(model_data, OrderedDict)
-    received_model = VisualCommunicationModel().to('cpu')
-    received_model.load_state_dict(model_data)
-    # ASSERT
-    assert same_model(model, received_model)
-
-
-def test_should_put_model(
-    datastore: Datastore,
-    model: Module,
-):
-    # ARRANGE
-    object_name = datastore.put_model(
-        model=model,
-    )
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # ACT
-    model_data = datastore.get_model(
-        object_name=object_name,
-    )
-    assert isinstance(model_data, OrderedDict)
-    received_model = VisualCommunicationModel().to('cpu')
-    received_model.load_state_dict(model_data)
-    # ASSERT
-    assert same_model(model, received_model)
-
-
-def test_should_update_model(
-    datastore: Datastore,
-    model_in_minio: tuple[Module, str],
-):
-    # ARRANGE
-    model, object_name = model_in_minio
-    object_name = datastore.put_model(
-        model=model,
-    )
-    assert isinstance(object_name, str)
-    assert len(object_name) > 0
-    # ACT
-    model_data = datastore.get_model(
-        object_name=object_name,
-    )
-    assert isinstance(model_data, OrderedDict)
-    received_model = VisualCommunicationModel().to('cpu')
-    received_model.load_state_dict(model_data)
-    # ASSERT
-    assert same_model(model, received_model)

shared/datastore/tests/unit/datastore_minio_test.py

@@ -1,230 +0,0 @@
-"""Definition of unittests for Datastore Minio implementation."""
-
-import os
-from hashlib import md5
-from io import BytesIO
-from unittest import TestCase
-from unittest.mock import ANY, MagicMock, patch
-
-from minio import Minio
-from PIL import Image
-from urllib3 import BaseHTTPResponse
-
-from shared.datastore.src.datastore_minio import DatastoreMinio
-
-
-class TestDatastoreMinioImplementation(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
-        # set other variables
-        self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
-        self.image = Image.new(mode='RGB', size=(480, 480))
-        self.buffer = BytesIO()
-        self.image.save(self.buffer, 'png')
-        self.num_bytes = len(self.buffer.getvalue())
-        self.checksum = md5(self.buffer.getbuffer()).hexdigest()
-
-    def tearDown(self):
-        # clear env vars
-        for key in self.env_var_map:
-            _ = os.environ.pop(key, default=None)
-
-    def test_instantiation_should_fail_when_env_not_set(self):
-        # ensure env not set
-        self.tearDown()
-        # run test
-        with self.assertRaises(OSError):
-            _ = DatastoreMinio()
-
-    @patch('shared.datastore.src.datastore_minio.Minio')
-    def test_connect_should_call_Minio_with_env_vars(self, minio_mock):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        minio_mock().bucket_exists.return_value = False
-        # ACT
-        datastore.connect()
-        # ASSERT
-        minio_mock.assert_called_with(
-            endpoint=self.env_var_map['MINIO_ENDPOINT'],
-            access_key=self.env_var_map['MINIO_ACCESS_KEY'],
-            secret_key=self.env_var_map['MINIO_SECRET_KEY'],
-            secure=False,
-        )
-        minio_mock().bucket_exists.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-        )
-        minio_mock().make_bucket.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-        )
-
-    def test_close_should_overwrite_private_variables(self):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        datastore._client = MagicMock()
-        datastore._bucket_name = MagicMock()
-        assert isinstance(datastore._client, MagicMock)
-        assert isinstance(datastore._bucket_name, MagicMock)
-        # ACT
-        datastore.close()
-        # ASSERT
-        assert datastore._client is None
-        assert datastore._bucket_name is None
-
-    @patch('shared.datastore.src.datastore_minio.DatastoreMinio.connect')
-    @patch('shared.datastore.src.datastore_minio.DatastoreMinio.close')
-    def test_context_management_implemented(
-        self,
-        mocked_close_method,
-        mocked_connect_method,
-    ):
-        # ARRANGE, ACT and ASSERT
-        with DatastoreMinio() as ds:
-            mocked_connect_method.assert_called_once()
-            assert isinstance(ds, DatastoreMinio)
-        mocked_close_method.assert_called_once()
-
-    def test_should_call_put_object_with_arguments(self):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        datastore._client = MagicMock(Minio)
-        datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
-        self.buffer.seek(0)
-        # ACT
-        datastore._put(
-            object_name=self.object_name,
-            buffer=self.buffer,
-        )
-        # ASSERT
-        datastore._client.put_object.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-            object_name=self.object_name,
-            length=self.num_bytes,
-            data=self.buffer,
-        )
-
-    def test_should_call_get_object_with_arguments(self):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        datastore._client = MagicMock(Minio)
-        datastore._client.get_object.return_value = MagicMock(
-            BaseHTTPResponse,
-            status=200,
-        )
-        # datastore._client.get_object.read.side_effect = [b'random ', b'test', b'text']
-        datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
-        # ACT
-        with self.assertRaises(TypeError):  # dont care to mock even more...
-            datastore._get(
-                object_name=self.object_name,
-            )
-        # ASSERT
-        datastore._client.get_object.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-            object_name=self.object_name,
-        )
-
-    def test_should_call_remove_object_with_arguments(self):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        datastore._client = MagicMock(Minio)
-        datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
-        # ACT
-        datastore._delete(
-            object_name=self.object_name,
-        )
-        # ASSERT
-        datastore._client.remove_object.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-            object_name=self.object_name,
-        )
-
-    def test_put_image_should_call_put_object_with_arguments(self):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        datastore._client = MagicMock(Minio)
-        datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
-        object_path = f'images/{self.checksum}'
-        # ACT
-        datastore.put_image(
-            image=self.image,
-        )
-        # ASSERT
-        datastore._client.put_object.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-            object_name=object_path,
-            length=self.num_bytes,
-            data=ANY,  # saved to different buffer when converting image
-        )
-
-    def test_get_image_should_call_get_object_with_arguments(self):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        datastore._client = MagicMock(Minio)
-        datastore._client.get_object.return_value = MagicMock(
-            BaseHTTPResponse,
-            status=200,
-        )
-        datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
-        object_path = f'images/{self.checksum}'
-        # ACT
-        with self.assertRaises(TypeError):  # dont care to mock even more...
-            datastore.get_image(
-                object_name=self.checksum,
-            )
-        # ASSERT
-        datastore._client.get_object.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-            object_name=object_path,
-        )
-
-    # @patch('shared.datastore.src.datastore_minio.torch.serialization.save')
-    # def test_put_model_should_call_put_object_with_arguments(self, mocked_torch_fn):
-    #     # ARRANGE
-    #     datastore = DatastoreMinio()
-    #     datastore._client = MagicMock(Minio)
-    #     datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
-    #     object_path = f'images/{self.checksum}'
-    #     mocked_torch_fn.return_value = None
-    #     model = MagicMock(torch.nn.Module)
-    #     # ACT
-    #     datastore.put_model(
-    #         model=model,
-    #     )
-    #     # ASSERT
-    #     datastore._client.put_object.assert_called_with(
-    #         bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-    #         object_name=object_path,
-    #         length=self.num_bytes,
-    #         data=ANY,  # saved to different buffer when converting data
-    #     )
-
-    def test_get_model_should_call_get_object_with_arguments(self):
-        # ARRANGE
-        datastore = DatastoreMinio()
-        datastore._client = MagicMock(Minio)
-        datastore._client.get_object.return_value = MagicMock(
-            BaseHTTPResponse,
-            status=200,
-        )
-        datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
-        object_path = f'models/{self.checksum}'
-        # ACT
-        with self.assertRaises(TypeError):  # dont care to mock even more...
-            datastore.get_model(
-                object_name=self.checksum,
-            )
-        # ASSERT
-        datastore._client.get_object.assert_called_with(
-            bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
-            object_name=object_path,
-        )

shared/docstore/__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/docstore/src/__init__.py

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

shared/docstore/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/docstore/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/docstore/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/docstore/src/classes/data_model.py

@@ -1,67 +0,0 @@
-"""Definition of DataModel base class."""
-
-import random
-
-from pydantic import BaseModel, ValidationError
-from torch import Tensor
-
-
-class DataModel(BaseModel):
-    """DataModel base class."""
-
-    @classmethod
-    def classname(cls) -> str:
-        """Return classname."""
-        return cls.__name__
-
-    @classmethod
-    def list_fields(cls) -> list[str]:
-        """List options that are stored as attributes."""
-        return list(cls.model_fields.keys())
-
-    @classmethod
-    def from_random(cls):
-        """Instantiate with random numbers."""
-        kwargs = {field: random.random() for field in cls.list_fields()}
-        return cls(**kwargs)
-
-    @classmethod
-    def from_choice(cls, option: str):
-        """Instantiate from choice."""
-        if option is None:
-            raise ValidationError()
-        assert isinstance(option, str), 'option is not a string'
-        allowed_options_list = cls.list_fields()
-        assert (
-            option in allowed_options_list
-        ), f"{option} is not among allowed fields {allowed_options_list}"
-        kwargs = {field: 0 for field in cls.list_fields()}
-        kwargs[option] = 1
-        return cls(**kwargs)
-
-    @classmethod
-    def from_tensor(cls, tensor: Tensor):
-        """Instantiate from list of values."""
-        assert tensor.size(dim=0) == 1, f'tensor batch larger than 1: {tensor}'
-        data_list = [float(t.item()) for t in tensor[0]]
-        kwargs = dict(zip(cls.list_fields(), data_list))
-        return cls(**kwargs)
-
-    def __repr__(self) -> str:
-        model_dict = self.model_dump()
-        model_repr_str = f'{self.classname()}('
-        model_repr_str += ', '.join(
-            [f'{field}={value:.3f}' for field, value in model_dict.items()],
-        )
-        model_repr_str += ')'
-        return model_repr_str
-
-    def highest_score_field(self) -> str:
-        """Return name of field with highest score."""
-        model_dict = self.model_dump()
-        return max(model_dict, key=lambda k: model_dict[k])
-
-    def highest_score_value(self) -> float:
-        """Return value of field with highest score."""
-        model_dict = self.model_dump()
-        return max(model_dict.values())

shared/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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 Datastore
-from shared.docstore.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)
-        with Datastore() as ds:
-            object_name = ds.put_image(
-                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
-        with Datastore() as ds:
-            image = ds.get_image(
-                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/docstore/src/classes/visual_syntax.py

@@ -1,28 +0,0 @@
-"""Definition of VisualSyntaxData data model."""
-
-from __future__ import annotations
-
-from .data_model import DataModel
-
-
-class VisualSyntaxData(DataModel):
-    """VisualSyntaxData data model."""
-
-    non_transactional_action: float
-    non_transactional_reaction: float
-    unidirectional_transactional_action: float
-    unidirectional_transactional_reaction: float
-    bidirectional_transactional_action: float
-    bidirectional_transactional_reaction: float
-    conversion: float
-    speech_process: float
-    classification_overt_taxonomy: float
-    analytical_exhaustive: float
-    analytical_disarranged: float
-    analytical_temporal: float
-    analytical_distributed: float
-    analytical_topological: float
-    analytical_exploded: float
-    analytical_inclusive: float
-    symbolic_suggestive: float
-    symbolic_attributive: float

shared/docstore/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/docstore/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/docstore/src/docstore_interface.py

@@ -1,83 +0,0 @@
-"""Definition of docstore interface."""
-
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-
-from shared.docstore.src.classes import ModelData, VisualCommunication
-
-
-class DocstoreInterface(ABC):
-    """Docstore interface class."""
-
-    @abstractmethod
-    def connect(
-        self,
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def close(
-        self,
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def __enter__(
-        self,
-    ) -> DocstoreInterface:
-        pass
-
-    @abstractmethod
-    def __exit__(
-        self,
-        exc_type,
-        exc_val,
-        exc_tb,
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def count_documents(
-        self,
-        only_with_annotation: bool,
-    ) -> int:
-        pass
-
-    @abstractmethod
-    def list_names(
-        self,
-        only_with_annotation: bool,
-    ) -> list[str]:
-        pass
-
-    @abstractmethod
-    def upsert_visual_comminucations(
-        self,
-        visual_communication_list: list[VisualCommunication],
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def upsert_annotations(
-        self,
-        visual_communication_name: str,
-        annotations: ModelData,
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def upsert_prediction(
-        self,
-        visual_communication_name: str,
-        predictions: ModelData,
-    ) -> None:
-        pass
-
-    @abstractmethod
-    def get_visual_communication(
-        self,
-        with_annotation: bool,
-        name: str | None = None,
-    ) -> VisualCommunication:
-        pass

shared/docstore/src/docstore_mongo.py

@@ -1,241 +0,0 @@
-"""Definition of docstore mongodb implementation."""
-
-from __future__ import annotations
-
-import logging
-import os
-from typing import Any
-
-from pymongo import MongoClient
-from pymongo.collection import Collection
-from pymongo.database import Database
-
-from shared.utils import check_env
-
-from .classes import ModelData, VisualCommunication
-from .docstore_interface import DocstoreInterface
-from .exceptions import NoDocumentFoundException
-
-
-class DocstoreMongo(DocstoreInterface):
-    """Docstore interface."""
-
-    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 __enter__(self) -> DocstoreMongo:
-        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 count_documents(
-        self,
-        only_with_annotation: bool = False,
-    ):
-        """Get the total number of Visual Communication documents matching the
-        filter."""
-        assert isinstance(only_with_annotation, bool)
-        # build query
-        query = {}
-        if only_with_annotation:
-            query['annotation'] = {'$ne': None}
-        # execute query
-        num_docs = self._collection.count_documents(filter=query)
-        return num_docs
-
-    def list_names(
-        self,
-        only_with_annotation: bool,
-    ) -> list[str]:
-        """List names of Visual Communication documents that match the
-        filters."""
-        assert isinstance(only_with_annotation, bool)
-        assert isinstance(self._collection, Collection)
-        # build query
-        query = {}
-        if only_with_annotation:
-            query['annotation'] = {'$ne': None}
-        # execute query
-        res_list = self._collection.find(
-            filter=query,
-            projection={
-                '_id': False,  # don't get id
-                'name': True,  # include document name
-            },
-        )
-        # extract info
-        name_list = [elem['name'] for elem in res_list]
-        return name_list
-
-    def upsert_visual_comminucations(
-        self,
-        visual_communication_list: list[VisualCommunication],
-    ) -> None:
-        """Upsert Visual Communication document in the database."""
-        assert isinstance(visual_communication_list, list)
-        assert all(
-            isinstance(vis_com, VisualCommunication)
-            for vis_com in visual_communication_list
-        )
-        assert isinstance(self._collection, Collection)
-        # convert to dict
-        doc_list = [vis_com.model_dump() for vis_com in visual_communication_list]
-        # insert documents
-        response = self._collection.insert_many(documents=doc_list)
-        # check response
-        if not response.acknowledged:
-            raise OSError('failed inserting documents')
-
-    def upsert_annotations(
-        self,
-        visual_communication_name: str,
-        annotations: ModelData,
-    ) -> None:
-        """Upsert annotation for the document with matching name."""
-        assert isinstance(visual_communication_name, str)
-        assert len(visual_communication_name) > 0
-        assert isinstance(annotations, ModelData)
-        assert isinstance(self._collection, Collection)
-        # convert to dict
-        doc = annotations.model_dump()
-        # build query
-        query = {
-            'name': visual_communication_name,
-        }
-        update = {
-            '$set': {
-                'annotation': doc,
-            },
-        }
-        # execute query
-        res = self._collection.update_one(
-            filter=query,
-            update=update,
-            upsert=True,
-        )
-        # check response
-        if not res.acknowledged:
-            raise OSError(
-                f'failed upserting annotations for {visual_communication_name}',
-            )
-
-    def upsert_prediction(
-        self,
-        visual_communication_name: str,
-        predictions: ModelData,
-    ) -> None:
-        """Upsert prediction for the document with matching name."""
-        assert isinstance(visual_communication_name, str)
-        assert len(visual_communication_name) > 0
-        assert isinstance(annotations, ModelData)
-        assert isinstance(self._collection, Collection)
-        # convert to dict
-        doc = predictions.model_dump()
-        # build query
-        query = {
-            'name': visual_communication_name,
-        }
-        update = {
-            '$set': {
-                'prediction': doc,
-            },
-        }
-        # execute query
-        res = self._collection.update_one(
-            filter=query,
-            update=update,
-            upsert=True,
-        )
-        # check response
-        if not res.acknowledged:
-            raise OSError(
-                f'failed upserting predictions for {visual_communication_name}',
-            )
-
-    def get_visual_communication(
-        self,
-        with_annotation: bool = False,
-        name: str | None = None,
-    ) -> VisualCommunication:
-        """Get a random Visual Communication document that matches annotation
-        filter.
-
-        If name is not specified, a random document matching filter is
-        returned.
-        """
-        assert isinstance(with_annotation, bool)
-        if name is not None:
-            assert isinstance(name, str)
-            assert len(name) > 0
-        # build query
-        query: dict[str, Any] = {}
-        if with_annotation:
-            query['annotation'] = {'$ne': None}
-        else:
-            query['annotation'] = {'$eq': None}
-        if name is not None:
-            query['name'] = {'$eq': name}
-        # execute query
-        res_list = self._collection.aggregate(
-            pipeline=[
-                {
-                    '$match': query,  # find using filters
-                },
-                {
-                    '$sample': {
-                        'size': 1,  # get one random
-                    },
-                },
-            ],
-        )
-        doc_list = list(res_list)
-        # check result
-        if len(doc_list) == 0:
-            raise NoDocumentFoundException()
-        # convert
-        vis_com = VisualCommunication.model_validate(doc_list[0])
-        return vis_com

shared/docstore/src/exceptions/__init__.py

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

shared/docstore/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/docstore/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/docstore/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.docstore.src.classes import VisualCommunication
-from shared.docstore.src.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/docstore/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/docstore/src/upsert_annotation.py

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

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

@@ -1,19 +0,0 @@
-from __future__ import annotations
-
-from pymongo.collection import Collection
-
-from shared.docstore.src.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/docstore/tests/unit/docstore_mongo_test.py

@@ -1,45 +0,0 @@
-"""Definition of unittests for Docstore MongoDB implementation."""
-
-import os
-from unittest import TestCase
-from unittest.mock import MagicMock
-
-from pymongo import MongoClient
-from pymongo.collection import Collection
-from pymongo.database import Database
-
-from shared.docstore.src.docstore_mongo import DocstoreMongo
-
-
-class TestDocstoreMongoImplementation(TestCase):
-
-    def setUp(self):
-        # define relevant env vars
-        self.env_var_map = {
-            'MONGO_ENDPOINT': '192.168.1.2:27017',
-            'MONGO_DB': 'visual_critical_discourse_analysis',
-            'MONGO_COLLECTION': 'test-collection',
-        }
-        # set env vars
-        for key, val in self.env_var_map.items():
-            os.environ[key] = val
-        # set other variables
-        self.mongo_client_mock = MagicMock(MongoClient)
-        self.mongo_database_mock = MagicMock(Database)
-        self.mongo_collection_mock = MagicMock(Collection)
-
-    def tearDown(self):
-        # clear env vars
-        for key in self.env_var_map:
-            _ = os.environ.pop(key, default=None)
-        # reset reuseable mocks
-        self.mongo_client_mock.reset_mock()
-        self.mongo_database_mock.reset_mock()
-        self.mongo_collection_mock.reset_mock()
-
-    def test_instantiation_should_fail_when_env_not_set(self):
-        # ensure env not set
-        self.tearDown()
-        # run test
-        with self.assertRaises(OSError):
-            _ = DocstoreMongo()

shared/repositories/src/dto/type_checking_base_model.py

@@ -7,6 +7,5 @@ class TypeCheckingBaseModel(BaseModel):
     """BaseModel with added type checking on input types."""
 
     model_config = ConfigDict(
-        validate_assignment=True,  # argument type checking
         frozen=True,  # ensure data immutability
     )

shared/repositories/tests/integration/conftest.py

@@ -31,12 +31,13 @@ from shared.repositories.src.implementations import (
 # set random seed for reproducibility
 random.seed(13)
 
-# load local test environment variables
-test_env_path = Path(__file__).parent.parent.parent.parent.parent / 'test.env'
-if test_env_path.exists():
-    load_dotenv(test_env_path.as_posix())
-
 # define test environment variables
+necessary_env_vars = {
+    'MINIO_ENDPOINT',
+    'MINIO_ACCESS_KEY',
+    'MINIO_SECRET_KEY',
+    'MONGO_ENDPOINT',
+}
 env_var_map = {
     'MINIO_BUCKET_NAME': 'test-bucket',
     'MINIO_OBJECT_NAME': 'test-object',
@@ -47,15 +48,18 @@ env_var_map = {
 
 
 @pytest.fixture(scope='session', autouse=True)
-def populate_env(
+def setup_env(
     request: pytest.FixtureRequest,
 ) -> None:
     """Populate environment with variables used for testing."""
-    # update env
+    # load in optional local test environment variables
+    test_env_path = Path(__file__).parent.parent.parent.parent.parent / 'test.env'
+    load_dotenv(test_env_path)
+    # check if necessary env vars are set
+    for key in necessary_env_vars:
+        assert key in os.environ, f'{key} not set'
+    # set env vars unique to this test
     for key, val in env_var_map.items():
-        # skip if already set by CI pipeline
-        if key in os.environ:
-            continue
         # set env var
         os.environ[key] = val
 
@@ -69,7 +73,7 @@ def populate_env(
 
 @pytest.fixture(scope='session')
 def raw_minio_client(
-    populate_env,
+    setup_env,
 ) -> Iterator[minio.Minio]:
     """Raw Minio client fixture."""
     # prepare arguments
@@ -101,7 +105,7 @@ def raw_minio_client(
 
 @pytest.fixture
 def minio_client(
-    populate_env,
+    setup_env,
 ) -> Iterator[MinioImplementation]:
     """MinioImplementation fixture."""
     # instantiate and connect client
@@ -202,9 +206,14 @@ def image_data_in_minio(
 
 @pytest.fixture(scope='session')
 def image_repo(
-    populate_env,
+    setup_env,
 ) -> Iterator[ImageRepository]:
     """Image repository fixture."""
+    # define test environment variables
+    assert 'MINIO_ENDPOINT' in os.environ, 'MINIO_ENDPOINT not set'
+    assert 'MINIO_ACCESS_KEY' in os.environ, 'MINIO_ACCESS_KEY not set'
+    assert 'MINIO_SECRET_KEY' in os.environ, 'MINIO_SECRET_KEY not set'
+    assert 'MONGO_ENDPOINT' in os.environ, 'MONGO_ENDPOINT not set'
     repo = ImageRepository()
     repo.connect()
     yield repo
@@ -262,7 +271,7 @@ def model_data_in_minio(
 
 @pytest.fixture(scope='session')
 def model_repo(
-    populate_env,
+    setup_env,
 ) -> Iterator[ModelRepository]:
     """Model repository fixture."""
     repo = ModelRepository()
@@ -273,7 +282,7 @@ def model_repo(
 
 @pytest.fixture
 def raw_mongo_client(
-    populate_env,
+    setup_env,
 ) -> Iterator[MongoClient]:
     """Raw mongo client fixture."""
     # prepare arguments
@@ -290,7 +299,7 @@ def raw_mongo_client(
 
 @pytest.fixture
 def mongo_client(
-    populate_env,
+    setup_env,
 ) -> Iterator[MongoImplementation]:
     """MongoImplementation fixture."""
     # instantiate and connect client
@@ -385,7 +394,7 @@ def visual_communication_data_in_mongo(
 
 @pytest.fixture(scope='session')
 def visual_communication_repo(
-    populate_env,
+    setup_env,
 ) -> Iterator[VisualCommunicationRepository]:
     """Visual communication repository fixture."""
     repo = VisualCommunicationRepository()

shared/repositories/tests/unit/type_checking_base_model_test.py

@@ -0,0 +1,46 @@
+"""Definition of unittests for TypeCheckingBaseModel class."""
+
+import unittest
+
+import pytest
+from pydantic import ValidationError
+
+from shared.repositories.src.dto.type_checking_base_model import TypeCheckingBaseModel
+
+
+class UUT(TypeCheckingBaseModel):
+    """Test class for TypeCheckingBaseModel."""
+
+    string_field: str
+    int_field: int
+
+
+class TestTypeCheckingBaseModel(unittest.TestCase):
+    """Test class for TypeCheckingBaseModel."""
+
+    def setUp(self):
+        """Set up the test case."""
+        self.uut = UUT
+        self.string_field = 'test'
+        self.int_field = 123
+
+    def test_type_checking_on_instantiation(self):
+        """Test type checking on instantiation."""
+        with pytest.raises(ValidationError):
+            _ = self.uut(
+                string_field=self.int_field,
+                int_field=self.string_field,
+            )
+
+    def test_attributes_frozen(self):
+        """Test that attributes cannot be updated."""
+        uut_instance = self.uut(
+            string_field=self.string_field,
+            int_field=self.int_field,
+        )
+        with pytest.raises(ValidationError):
+            uut_instance.string_field = self.int_field
+
+
+if __name__ == '__main__':
+    pytest.main(['-s', '-v', __file__])

shared/utils/check_env.py

@@ -14,5 +14,5 @@ def check_env(
     for env_var in var_list:
         if env_var not in os.environ:
             missing_vars.append(env_var)
-    if len(missing_vars) > 0:
+    if missing_vars:
         raise OSError(f"environment variable(s) not set: {missing_vars}")

web_ui/src/app/init_app.py

@@ -4,29 +4,28 @@ from __future__ import annotations
 
 import logging
 import os
+import random
+from base64 import b64decode, b64encode
+from io import BytesIO
 
 import dash_bootstrap_components as dbc
 from dash import ALL, Dash, Input, Output, State
 from dash_auth import BasicAuth
+from PIL import Image
 from pydantic import ValidationError
-from pymongo.collection import Collection
 
-from shared.datastore import Datastore
+from shared.repositories import (
-from shared.docstore import count_documents, get_visual_communication, upsert_annotation
+    ImageData,
-from shared.docstore.src.classes import ModelData, VisualCommunication
+    ImageRepository,
-from shared.docstore.src.exceptions import NoDocumentFoundException
+    VisualCommunicationData,
+    VisualCommunicationRepository,
+)
 
 from .layout import app_layout
 
 
-def init_app(
+def init_app() -> Dash:
-    mongo_collection: Collection,
-    datastore: Datastore,
-) -> Dash:
     """Initialise web UI application."""
-    assert isinstance(mongo_collection, Collection)
-    assert isinstance(datastore, Datastore)
-    assert datastore._client is not None
     # setup app
     app = Dash(
         name='visual_critical_discourse_analysis_web_ui',
@@ -90,14 +89,10 @@ def init_app(
         if filename_list is not None:
             assert isinstance(filename_list, list)
         # get values from database
-        num_total = count_documents(
+        with VisualCommunicationRepository() as repo:
-            collection=mongo_collection,
+            name_list = repo.list_names()
-            only_with_annotation=False,
+        num_total = len(name_list)
-        )
+        num_handled = 0
-        num_handled = count_documents(
-            collection=mongo_collection,
-            only_with_annotation=True,
-        )
         limit = int(num_total / 20)
         label_str = f'{num_handled}/{num_total}' if num_handled >= limit else ''
         return num_handled, num_total, label_str
@@ -125,38 +120,37 @@ def init_app(
             for content, filename in zip(content_list, filename_list):
                 try:
                     # decode image content
-                    image = VisualCommunication.decode_image(content=content)
+                    content_data = content.split(',')[-1]
+                    image = Image.open(BytesIO(b64decode(content_data)))
+                    # instantiate ImageData object
+                    image_data = ImageData(
+                        name=filename,
+                        image=image,
+                    )
+                    # instantiate VisualCommunicationData object
+                    vis_com_data = VisualCommunicationData(name=filename)
                 except Exception:
                     logging.debug('failed decoding %s', filename)
                     failed_filename_list.append(filename)
                     continue
                 try:
-                    assert datastore._client is not None
+                    # save ImageData
-                    # instantiate to upload image to minio
+                    with ImageRepository() as repo:
-                    vis_com = VisualCommunication.from_name_and_image(
+                        repo.put_data(image_data)
-                        name=filename,
+                    # save VisualCommunicationData
-                        image=image,
+                    with VisualCommunicationRepository() as repo:
-                        minio_client=datastore._client,
+                        repo.put_data(vis_com_data)
-                    )
                 except Exception as exc:
                     logging.debug(exc)
                     failed_filename_list.append(filename)
-                    continue
+                    # cleanup failed uploads
-                try:
+                    with ImageRepository() as repo:
-                    # save to mongodb
+                        repo.remove_data(image_data.name)
-                    vis_com.save_to_mongo(collection=mongo_collection)
+                    with VisualCommunicationRepository() as repo:
-                except Exception as exc:
+                        repo.remove_data(vis_com_data.name)
-                    logging.debug(exc)
+            if len(failed_filename_list) == 0:
-                    failed_filename_list.append(filename)
+                err_msg = 'failed uploading:' + '\n'.join(failed_filename_list)
-                    # remove document from minio
+                raise FileNotFoundError(err_msg)
-                    datastore._delete(
-                        object_name=vis_com.object_name,
-                    )
-            assert (
-                len(failed_filename_list) == 0
-            ), f"failed uploading:{
-                '\n'.join(failed_filename_list)
-            }"
         except Exception as exc:
             logging.debug(exc)
             return None, str(exc).title()
@@ -208,7 +202,7 @@ def init_app(
                 logging.info(annotation_keys)
                 for option, value in zip(annotation_keys, annotation_values):
                     if value is None:
-                        raise ValueError(f"{option} is not set")
+                        raise ValueError(f'{option} is not set')
                 # prepare data to save
                 annotation_keys = [elem.replace(' ', '_') for elem in annotation_keys]
                 annotation_values = [
@@ -217,16 +211,21 @@ def init_app(
                 annotation_map = {
                     key: value for key, value in zip(annotation_keys, annotation_values)
                 }
-                # instantiate ModelOutputs object
+                # get data
-                annotations = ModelData.from_annotations(**annotation_map)
+                with VisualCommunicationRepository() as repo:
-                # save data to database
+                    vis_com_data = repo.get_data(vis_com_name)
-                upsert_annotation(
+                    # data already present, as it was loaded earlier to get here
-                    collection=mongo_collection,
+                    assert vis_com_data is not None
-                    vis_com_name=vis_com_name,
+                    # update annotations
-                    annotations=annotations,
+                    vis_com_data_updated = vis_com_data.model_copy(
+                        update={
+                            'annotations': annotation_map,
+                        },
                     )
+                    # save updated data
+                    repo.put_data(vis_com_data_updated)
             except (ValueError, ValidationError) as exc:
-                msg = f"failed saving annotation: {exc}"
+                msg = f'failed saving annotation: {exc}'
                 logging.warning(msg)
                 response[0] = msg
                 return tuple(response)
@@ -234,21 +233,30 @@ def init_app(
         logging.info('trying to get new visual communication')
         try:
             # get data
-            vis_com = get_visual_communication(
+            with VisualCommunicationRepository() as repo:
-                collection=mongo_collection,
+                name_list = repo.list_names()
-                with_annotation=False,
+                random_name = random.choice(name_list)
-            )
+                vis_com = repo.get_data(random_name)
+            # could not get here if data doesn't exist
+            assert vis_com is not None
             # set variables
             vis_com_name = vis_com.name
-            assert datastore._client is not None
+            with ImageRepository() as repo:
-            image_src = vis_com.webencoded_image(minio_client=datastore._client)
+                image_data = repo.get_data(vis_com_name)
+            # could not get here if data doesn't exist
+            assert image_data is not None
+            buffer = BytesIO()
+            image_data.image.save(buffer, format='PNG')
+            image_src = 'data:image/png;base64,' + b64encode(buffer.getvalue()).decode(
+                'utf-8',
+            )
             if vis_com.prediction is not None:
                 # TODO: update to use optional predictions
                 pass
             else:
                 # reset annotations
                 annotation_values = [None for elem in annotation_values]
-        except NoDocumentFoundException:
+        except Exception:
             msg = 'no unannotated data in database'
             logging.warning(msg)
             response[0] = msg

web_ui/src/main.py

@@ -4,15 +4,13 @@ from __future__ import annotations
 
 import os
 
-from shared.datastore import Datastore
-from shared.docstore import connect_mongodb
 from shared.utils import check_env, setup_logging
 
 from .app import init_app
 
 # ensure env vars set
 NECESSARY_ENV_VAR_LIST = {
-    'MONGO_HOST',
+    'MONGO_ENDPOINT',
     'MONGO_DB',
     'MONGO_COLLECTION',
     'MONGO_USER',
@@ -23,24 +21,13 @@ NECESSARY_ENV_VAR_LIST = {
     'MINIO_ACCESS_KEY',
     'MINIO_SECRET_KEY',
     'MINIO_BUCKET_NAME',
-    'MINIO_BUCKET_NAME_MODELS',
 }
 check_env(NECESSARY_ENV_VAR_LIST)
 
 # setup logging stream handler
 setup_logging()
 
-# connect to database
-collection, db, client = connect_mongodb()
-
-# connect to minio
-datastore = Datastore()
-datastore.connect()
-
 # initialise application
-app = init_app(
+app = init_app()
-    mongo_collection=collection,
-    datastore=datastore,
-)
 server = app.server
 server.config.update(SECRET_KEY=os.urandom(24))