brian/visual_critical_discourse_analysis
1
0
Fork 0
Code Issues 11 Pull Requests Actions Projects 1 Releases Activity

Merge pull request 'object_based_docstore_approach' (#68) from object_based_docstore_approach into main

Browse Source 
Reviewed-on: #68
This commit was merged in pull request #68.
This commit is contained in:
Brian Bjarke Jensen
2025-04-16 00:03:23 +02:00
parent 8214bd2669 a2b818bbcc
commit 7ce7a368c7
103 changed files with 3840 additions and 3193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitea/workflows/pull.yaml
+21 -11
View File
@@ -9,32 +9,42 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v3
uses: actions/checkout@v4
- name: Setup Environment
uses: https://github.com/actions/setup-python@v3
uses: https://github.com/actions/setup-python@v5
with:
python-verison: "3.12"
python-version: "3.12"
architecture: "x64"
- name: Install Packages
- name: Setup poetry
env:
POETRY_VERSION: 2.1.1
POETRY_HOME: /opt/poetry
POETRY_NO_INTERACTION: 1
POETRY_NO_CACHE: 1
run: |
curl -sSL https://install.python-poetry.org | python3 -
export PATH=$POETRY_HOME/bin:$PATH
poetry --version
- name: Install Dependencies
env:
PIP_INDEX_URL: ${{ vars.PIP_INDEX_URL }}
PIP_TRUSTED_HOST: ${{ vars.PIP_TRUSTED_HOST }}
run: |
pip install poetry
poetry install
/opt/poetry/bin/poetry install
- name: PEP8 Check
run: |
poetry run flake8 . --benchmark
/opt/poetry/bin/poetry run flake8 . --benchmark
- name: Type Check
run: |
poetry run mypy .
/opt/poetry/bin/poetry run mypy .
- name: Pytest & 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: |
poetry run coverage run -m pytest .
/opt/poetry/bin/poetry run coverage run -m pytest .
- name: Coverage Report
run: |
poetry run coverage report -m
/opt/poetry/bin/poetry run coverage report -m
.pre-commit-config.yaml
+7 -6
View File
@@ -1,15 +1,16 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
rev: v5.0.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-files
- id: debug-statements
- id: double-quote-string-fixer
- id: name-tests-test
- repo: https://github.com/asottile/setup-cfg-fmt
rev: v2.5.0
rev: v2.7.0
hooks:
- id: setup-cfg-fmt
- repo: https://github.com/pre-commit/mirrors-isort
@@ -23,7 +24,7 @@ repos:
hooks:
- id: add-trailing-comma
- repo: https://github.com/asottile/pyupgrade
rev: v3.15.1
rev: v3.19.1
hooks:
- id: pyupgrade
args: [--py39-plus]
@@ -35,12 +36,12 @@ repos:
additional_dependencies:
- "pyproject-flake8"
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.8.0
rev: v1.15.0
hooks:
- id: mypy
exclude: ^testing/resources/
- repo: https://github.com/psf/black
rev: 24.4.2
rev: 25.1.0
hooks:
- id: black
language_version: python3.12
@@ -55,7 +56,7 @@ repos:
language: python
types: [ python ]
- repo: https://github.com/jendrikseipp/vulture
rev: 'v2.6'
rev: 'v2.14'
hooks:
- id: vulture
entry: vulture . --min-confidence 90 --exclude */.venv/*.py,*/tests/*.py
misc/generate_random_prediction.py
-42
View File
@@ -1,42 +0,0 @@
from __future__ import annotations
from pathlib import Path
from dotenv import load_dotenv
from shared.datastore import Datastore
from shared.mongodb import connect_mongodb
from shared.mongodb.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
-31
View File
@@ -1,31 +0,0 @@
from __future__ import annotations
from pathlib import Path
from dotenv import load_dotenv
from shared.datastore import Datastore
from shared.mongodb import connect_mongodb, get_visual_communication
from shared.utils import check_env, setup_logging
from web_ui.src.main import NECESSARY_ENV_VAR_LIST
if __name__ == '__main__':
# load in env file
env_path = Path(__file__).parent.parent / 'server.env'
assert env_path.exists()
load_dotenv(env_path)
# ensure env vars set
check_env(NECESSARY_ENV_VAR_LIST)
# setup logging
setup_logging()
# connect to minIO
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_download.py
+2 -2
View File
@@ -5,8 +5,8 @@ from pathlib import Path
from dotenv import load_dotenv
from shared.mongodb import connect_mongodb
from shared.mongodb.classes import VisualCommunication
from shared.docstore import connect_mongodb
from shared.docstore.classes import VisualCommunication
if __name__ == '__main__':
# prepare env vars
misc/image_upload.py
-48
View File
@@ -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.mongodb import connect_mongodb
from shared.mongodb.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
-53
View File
@@ -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.mongodb import connect_mongodb
from shared.mongodb.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
-34
View File
@@ -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/model_outputs_from_annotation.py
+1 -1
View File
@@ -1,4 +1,4 @@
from shared.mongodb.classes import ModelData
from shared.docstore.classes import ModelData
if __name__ == '__main__':
# instantiate data object
misc/new_model_to_minio.py
-32
View File
@@ -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
-48
View File
@@ -1,48 +0,0 @@
from __future__ import annotations
from pathlib import Path
from dotenv import load_dotenv
from shared.datastore import Datastore
from shared.mongodb import connect_mongodb, upsert_prediction
from shared.mongodb.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,
)
misc/total_annotated.py
+1 -1
View File
@@ -5,7 +5,7 @@ from pathlib import Path
from dotenv import load_dotenv
from shared.mongodb import connect_mongodb, count_documents
from shared.docstore import connect_mongodb, count_documents
if __name__ == '__main__':
# prepare env vars
misc/total_documents.py
+1 -1
View File
@@ -5,7 +5,7 @@ from pathlib import Path
from dotenv import load_dotenv
from shared.mongodb import connect_mongodb, count_documents
from shared.docstore import connect_mongodb, count_documents
if __name__ == '__main__':
# prepare env vars
model/src/main.py
+3 -9
View File
@@ -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.mongodb.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
+2 -6
View File
@@ -4,8 +4,6 @@ from __future__ import annotations
from torch import nn
from shared.mongodb.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
data = ModelData.from_prediction_dict(prediction_dict)
return data
return prediction_dict
model/src/utils/load_model.py
+8 -9
View File
@@ -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(
datastore: Datastore,
) -> VisualCommunicationModel:
def load_model() -> 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
model_checkpoint = datastore.get_model(
object_name=model_object_name,
)
# load model data
with ModelRepository() as repo:
model_data = repo.get_data(model_object_name)
if model_data is None:
raise FileNotFoundError(f'model {model_object_name} not found')
model_checkpoint = torch.load(model_data.buffer)
model.load_state_dict(model_checkpoint)
# clean memory
model_checkpoint.clear()
model/src/utils/vcda_dataset.py
+6 -8
View File
@@ -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 = self.datastore.get_image(
object_name=object_name,
)
tensor = self.image_to_tensor(image)
image_name = self.data_name_list[idx]
with ImageRepository() as repo:
image_data = repo.get_data(image_name)
assert image_data is not None
tensor = self.image_to_tensor(image_data.image)
if self.do_augment:
tensor = self.augment(tensor)
return tensor
model/train.py
+3 -7
View File
@@ -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
-42
View File
@@ -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,
)
other/transfer_images_mongo_minio.py
+2 -2
View File
@@ -10,8 +10,8 @@ from dotenv import load_dotenv
from pymongo.collection import Collection
from shared.datastore import Datastore
from shared.mongodb import connect_mongodb
from shared.mongodb.src.classes import VisualCommunication
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
poetry.lock
Generated
+1547 -1039
View File
File diff suppressed because it is too large Load Diff
pyproject.toml
+4 -7
View File
@@ -32,10 +32,11 @@ pandas = "^2.2.1"
selenium = "^4.18.1"
webdriver-manager = "^4.0.1"
retry = "^0.9.2"
pre-commit = "^4.1.0"
[tool.poetry.group.model.dependencies]
torch = "^2.2.1"
torch = "^2.0.0"
torchvision = "^0.17.1"
torchinfo = "^1.8.0"
minio = "^7.2.7"
@@ -111,7 +112,7 @@ module = "utils.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = "datastore.*"
module = "shared.datastore.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
@@ -119,9 +120,5 @@ module = "models.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = "shared.mongodb.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = "testcontainers.*"
module = "shared.docstore.*"
ignore_missing_imports = true
shared/datastore/__init__.py
-1
View File
@@ -1 +0,0 @@
from .src import Datastore
shared/datastore/src/__init__.py
-1
View File
@@ -1 +0,0 @@
from .datastore_minio import DatastoreMinio as Datastore
shared/datastore/src/datastore_interface.py
-69
View File
@@ -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
-241
View File
@@ -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 interval 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
-112
View File
@@ -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
-196
View File
@@ -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
-10
View File
@@ -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
-82
View File
@@ -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
-90
View File
@@ -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
-230
View File
@@ -1,230 +0,0 @@
"""Definition of unittests for Datastore instantiation."""
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 TestDatastoreMinioInstantiation(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/mongodb/__init__.py
-8
View File
@@ -1,8 +0,0 @@
from .src import classes, exceptions
from .src.connect_mongodb import connect_mongodb
from .src.count_documents import count_documents
from .src.get_visual_communication import get_visual_communication
from .src.list_names import list_names
from .src.upsert_annotation import upsert_annotation
from .src.upsert_prediction import upsert_prediction
from .src.upsert_visual_communication import upsert_visual_communication
shared/mongodb/src/__init__.py
-3
View File
@@ -1,3 +0,0 @@
"""Database utils module content."""
from .connect_mongodb import connect_mongodb
shared/mongodb/src/classes/__init__.py
-13
View File
@@ -1,13 +0,0 @@
from .angle import AngleData
from .contact import ContactData
from .distance import DistanceData
from .framing import FramingData
from .information_value import InformationValueData
from .modality_color import ModalityColorData
from .modality_depth import ModalityDepthData
from .modality_lighting import ModalityLightingData
from .model_data import ModelData
from .point_of_view import PointOfViewData
from .salience import SalienceData
from .visual_communication import VisualCommunication
from .visual_syntax import VisualSyntaxData
shared/mongodb/src/classes/angle.py
-13
View File
@@ -1,13 +0,0 @@
"""Definition of Angle data model."""
from __future__ import annotations
from .data_model import DataModel
class AngleData(DataModel):
"""Angle data model."""
high: float
eye_level: float
low: float
shared/mongodb/src/classes/contact.py
-12
View File
@@ -1,12 +0,0 @@
"""Definition of ContactData data model."""
from __future__ import annotations
from .data_model import DataModel
class ContactData(DataModel):
"""ContactData data model."""
offer: float
demand: float
shared/mongodb/src/classes/distance.py
-13
View File
@@ -1,13 +0,0 @@
"""Definition of DistanceData data model."""
from __future__ import annotations
from .data_model import DataModel
class DistanceData(DataModel):
"""DistanceData data model."""
long: float
medium: float
close: float
shared/mongodb/src/classes/framing.py
-14
View File
@@ -1,14 +0,0 @@
"""Definition of FramingData data model."""
from __future__ import annotations
from .data_model import DataModel
class FramingData(DataModel):
"""FramingData data model."""
frame_lines: float
empty_space: float
colour_contrast: float
form_contrast: float
shared/mongodb/src/classes/information_value.py
-13
View File
@@ -1,13 +0,0 @@
"""Definition of InformationValueData data model."""
from __future__ import annotations
from .data_model import DataModel
class InformationValueData(DataModel):
"""InformationValueData data model."""
given_new: float
ideal_real: float
central_marginal: float
shared/mongodb/src/classes/modality_color.py
-13
View File
@@ -1,13 +0,0 @@
"""Definition of ModalityColorData data model."""
from __future__ import annotations
from .data_model import DataModel
class ModalityColorData(DataModel):
"""ModalityColorData data model."""
high: float
medium: float
low: float
shared/mongodb/src/classes/modality_depth.py
-13
View File
@@ -1,13 +0,0 @@
"""Definition of ModalityDepthData data model."""
from __future__ import annotations
from .data_model import DataModel
class ModalityDepthData(DataModel):
"""ModalityDepthData data model."""
high: float
medium: float
low: float
shared/mongodb/src/classes/modality_lighting.py
-13
View File
@@ -1,13 +0,0 @@
"""Definition of ModalityLightingData data model."""
from __future__ import annotations
from .data_model import DataModel
class ModalityLightingData(DataModel):
"""ModalityLightingData data model."""
high: float
medium: float
low: float
shared/mongodb/src/classes/model_data.py
-115
View File
@@ -1,115 +0,0 @@
"""Definition of ModelData data model."""
from __future__ import annotations
from .angle import AngleData
from .contact import ContactData
from .data_model import DataModel
from .distance import DistanceData
from .framing import FramingData
from .information_value import InformationValueData
from .modality_color import ModalityColorData
from .modality_depth import ModalityDepthData
from .modality_lighting import ModalityLightingData
from .point_of_view import PointOfViewData
from .salience import SalienceData
from .visual_syntax import VisualSyntaxData
class ModelData(DataModel):
"""ModelData model for data IO with combined ML model."""
visual_syntax: VisualSyntaxData
contact: ContactData
angle: AngleData
point_of_view: PointOfViewData
distance: DistanceData
modality_lighting: ModalityLightingData
modality_color: ModalityColorData
modality_depth: ModalityDepthData
information_value: InformationValueData
framing: FramingData
salience: SalienceData
@classmethod
def from_random(cls) -> ModelData:
"""Instantiate with random numbers."""
kwargs = {
field: field_info.annotation.from_random() # type: ignore
for field, field_info in cls.model_fields.items()
}
return cls(**kwargs)
@classmethod
def from_prediction_dict(
cls,
prediction_dict: dict,
) -> ModelData:
"""Instantiate from prediction dictionary."""
kwargs = {
'visual_syntax': VisualSyntaxData.from_tensor(
prediction_dict['visual_syntax'],
),
'contact': ContactData.from_tensor(
prediction_dict['contact'],
),
'angle': AngleData.from_tensor(
prediction_dict['angle'],
),
'point_of_view': PointOfViewData.from_tensor(
prediction_dict['point_of_view'],
),
'distance': DistanceData.from_tensor(
prediction_dict['distance'],
),
'modality_lighting': ModalityLightingData.from_tensor(
prediction_dict['modality_lighting'],
),
'modality_color': ModalityColorData.from_tensor(
prediction_dict['modality_color'],
),
'modality_depth': ModalityDepthData.from_tensor(
prediction_dict['modality_depth'],
),
'information_value': InformationValueData.from_tensor(
prediction_dict['information_value'],
),
'framing': FramingData.from_tensor(
prediction_dict['framing'],
),
'salience': SalienceData.from_tensor(
prediction_dict['salience'],
),
}
return cls(**kwargs)
@classmethod
def from_annotations(
cls,
visual_syntax: str,
contact: str,
angle: str,
point_of_view: str,
distance: str,
modality_lighting: str,
modality_color: str,
modality_depth: str,
information_value: str,
framing: str,
salience: str,
) -> ModelData:
"""Instantiate from annotation."""
kwargs = {
'visual_syntax': VisualSyntaxData.from_choice(visual_syntax),
'contact': ContactData.from_choice(contact),
'angle': AngleData.from_choice(angle),
'point_of_view': PointOfViewData.from_choice(point_of_view),
'distance': DistanceData.from_choice(distance),
'modality_lighting': ModalityLightingData.from_choice(modality_lighting),
'modality_color': ModalityColorData.from_choice(modality_color),
'modality_depth': ModalityDepthData.from_choice(modality_depth),
'information_value': InformationValueData.from_choice(information_value),
'framing': FramingData.from_choice(framing),
'salience': SalienceData.from_choice(salience),
}
return cls(**kwargs)
shared/mongodb/src/classes/point_of_view.py
-12
View File
@@ -1,12 +0,0 @@
"""Definition of PointOfViewData data model."""
from __future__ import annotations
from .data_model import DataModel
class PointOfViewData(DataModel):
"""PointOfViewData data model."""
frontal: float
oblique: float
shared/mongodb/src/classes/salience.py
-15
View File
@@ -1,15 +0,0 @@
"""Definition of SalienceData data model."""
from __future__ import annotations
from .data_model import DataModel
class SalienceData(DataModel):
"""SalienceData data model."""
size: float
colour: float
tone: float
form: float
positioning: float
shared/mongodb/src/classes/visual_communication.py
-122
View File
@@ -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.mongodb.src.classes import ModelData
class VisualCommunication(BaseModel):
"""Visual communication model."""
name: str
object_name: str
annotation: ModelData | None = None
prediction: ModelData | None = None
model_config = ConfigDict(arbitrary_types_allowed=True)
@classmethod
def classname(cls) -> str:
"""Return classname."""
return cls.__name__
@classmethod
def upload_image_to_minio(
cls,
image: Image.Image,
minio_client: Minio,
) -> str:
"""Upload image to MinIO and return MD5 checksum of hashed image."""
assert isinstance(image, Image.Image)
assert isinstance(minio_client, Minio)
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/mongodb/src/connect_mongodb.py
-30
View File
@@ -1,30 +0,0 @@
"""Definition of function to connect to database using environment
variables."""
import logging
import os
from dotenv import load_dotenv
from pymongo import MongoClient
def connect_mongodb():
"""Connect to MongoDB using env vars."""
# load env vars
load_dotenv()
necessary_env_vars = [
'MONGO_HOST',
'MONGO_DB',
'MONGO_COLLECTION',
]
for env_var in necessary_env_vars:
assert env_var in os.environ, f"{env_var} not found"
# connect to database
client = MongoClient(os.getenv('MONGO_HOST'))
db = client[os.getenv('MONGO_DB')]
# extract collection
collection = db[os.getenv('MONGO_COLLECTION')]
# set unique index on "name"
collection.create_index('name', unique=True)
logging.debug('finished')
return collection, db, client
shared/mongodb/src/count_documents.py
-20
View File
@@ -1,20 +0,0 @@
"""Definition of function to count documents in database."""
from __future__ import annotations
from pymongo.collection import Collection
def count_documents(
collection: Collection,
only_with_annotation: bool = False,
) -> int:
"""Get the total number of documents in database that matches the
filters."""
assert isinstance(collection, Collection)
assert isinstance(only_with_annotation, bool)
# build query
query = {}
if only_with_annotation:
query['annotation'] = {'$ne': None}
return collection.count_documents(filter=query)
shared/mongodb/src/exceptions/__init__.py
-1
View File
@@ -1 +0,0 @@
from .no_document_found import NoDocumentFoundException
shared/mongodb/src/exceptions/no_document_found.py
-5
View File
@@ -1,5 +0,0 @@
"""Definition of database exception."""
class NoDocumentFoundException(Exception):
"""Database exception for when no documents are found."""
shared/mongodb/src/get_dataset.py
-16
View File
@@ -1,16 +0,0 @@
"""Definition of get_dataset function."""
from typing import Literal
from pymongo.collection import Collection
def get_dataset(
collection: Collection,
type: Literal['train', 'test', 'validation'],
) -> list[str]:
"""Get list of data names for the corresponding type."""
assert isinstance(collection, Collection)
assert isinstance(type, str)
assert type in ['train', 'test', 'validation']
return []
shared/mongodb/src/get_visual_communication.py
-40
View File
@@ -1,40 +0,0 @@
"""Definition of function to get visual communication from database."""
from __future__ import annotations
import logging
from pymongo.collection import Collection
from shared.mongodb.src.classes import VisualCommunication
from shared.mongodb.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/mongodb/src/list_names.py
-30
View File
@@ -1,30 +0,0 @@
"""Definition of function to list names of all visual communication documents
in database."""
from __future__ import annotations
from pymongo.collection import Collection
def list_names(
collection: Collection,
only_with_annotation: bool = True,
) -> list[str]:
"""List the names of entries that match the filters."""
assert isinstance(collection, Collection)
assert isinstance(only_with_annotation, bool)
# build query
query = {}
if only_with_annotation:
query['annotation'] = {'$ne': None}
# execute query
res_list = collection.find(
filter=query,
projection={
'_id': False,
'name': True,
},
)
# extract information
name_list = [elem['name'] for elem in res_list]
return name_list
shared/mongodb/src/upsert_annotation.py
-30
View File
@@ -1,30 +0,0 @@
from __future__ import annotations
import logging
from pymongo.collection import Collection
from shared.mongodb.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/mongodb/src/upsert_prediction.py
-30
View File
@@ -1,30 +0,0 @@
from __future__ import annotations
import logging
from pymongo.collection import Collection
from shared.mongodb.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/mongodb/src/upsert_visual_communication.py
-19
View File
@@ -1,19 +0,0 @@
from __future__ import annotations
from pymongo.collection import Collection
from shared.mongodb.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/repositories/__init__.py
+8
View File
@@ -0,0 +1,8 @@
from .src import ImageRepository, ModelRepository, VisualCommunicationRepository
from .src.dto import (
HexadecimalString,
ImageData,
ModelData,
VisualCommunicationData,
VisualCommunicationValues,
)
shared/repositories/src/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from .image_repository import ImageRepository
from .model_repository import ModelRepository
from .visual_communication_repository import VisualCommunicationRepository
shared/repositories/src/dto/__init__.py
+7
View File
@@ -0,0 +1,7 @@
from .hexadecimal_string import HexadecimalString
from .image_data import ImageData
from .model_data import ModelData
from .visual_communication_data import (
VisualCommunicationData,
VisualCommunicationValues,
)
shared/repositories/src/dto/bytes_io_pydantic_annotation.py
+37
View File
@@ -0,0 +1,37 @@
"""Definition of BytesIO Pydantic Annotation."""
from io import BytesIO
from typing import Any
from pydantic.json_schema import JsonSchemaValue
from pydantic_core import core_schema
class BytesIOPydanticAnnotation:
"""Pydantic annotation that defines input validation, as well as general
and json serialization."""
@classmethod
def validate_input(cls, v: Any, handler) -> BytesIO:
"""Pydantic-related function to validate input on instantiation."""
if isinstance(v, BytesIO):
return v
s = handler(v)
return BytesIO(s)
@classmethod
def __get_pydantic_core_schema__(
cls,
source_type,
_handler,
) -> core_schema.CoreSchema:
assert source_type is BytesIO
return core_schema.no_info_wrap_validator_function(
function=cls.validate_input,
schema=core_schema.str_schema(),
serialization=core_schema.to_string_ser_schema(),
)
@classmethod
def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
return handler(core_schema.str_schema())
shared/repositories/src/dto/hexadecimal_string.py
+24
View File
@@ -0,0 +1,24 @@
"""Definition of Checksum DTO."""
from __future__ import annotations
import re
class HexadecimalString(str):
"""Hexadecimal-string class."""
def __new__(cls, string):
# ensure proper input format
pattern = r'[0-9-a-fA-F]{32}'
match = re.match(pattern, string)
if match is None:
raise ValueError(f'format does not match a hexadecimal-string: {string}')
return super().__new__(cls, string)
def __repr__(self) -> str:
class_name = self.__class__.__name__
return f"{class_name}('{self}')"
def __reduce__(self):
return self.__class__, (self,)
shared/repositories/src/dto/hexadecimal_string_pydantic_annotation.py
+38
View File
@@ -0,0 +1,38 @@
"""Definition of HexadecimalString Pydantic Annotation."""
from typing import Any
from pydantic.json_schema import JsonSchemaValue
from pydantic_core import core_schema
from .hexadecimal_string import HexadecimalString
class HexadecimalStringPydanticAnnotation:
"""Pydantic annotation that defines input validation, as well as general
and json serialization."""
@classmethod
def validate_input(cls, v: Any, handler) -> HexadecimalString:
"""Pydantic-related function to validate input on instantiation."""
if isinstance(v, HexadecimalString):
return v
s = handler(v)
return HexadecimalString(s)
@classmethod
def __get_pydantic_core_schema__(
cls,
source_type,
_handler,
) -> core_schema.CoreSchema:
assert source_type is HexadecimalString
return core_schema.no_info_wrap_validator_function(
function=cls.validate_input,
schema=core_schema.str_schema(),
serialization=core_schema.to_string_ser_schema(),
)
@classmethod
def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
return handler(core_schema.str_schema())
shared/repositories/src/dto/image_data.py
+16
View File
@@ -0,0 +1,16 @@
"""Definition of VisualData DTO."""
from typing import Annotated
from PIL import Image
from pydantic import Field
from .image_pydantic_annotation import ImagePydanticAnnotation
from .type_checking_base_model import TypeCheckingBaseModel
class ImageData(TypeCheckingBaseModel):
"""Visual data class."""
image: Annotated[Image.Image, ImagePydanticAnnotation]
name: str = Field(min_length=1)
shared/repositories/src/dto/image_pydantic_annotation.py
+37
View File
@@ -0,0 +1,37 @@
"""Definition of BytesIO Pydantic Annotation."""
from typing import Any
from PIL import Image
from pydantic.json_schema import JsonSchemaValue
from pydantic_core import core_schema
class ImagePydanticAnnotation:
"""Pydantic annotation that defines input validation, as well as general
and json serialization."""
@classmethod
def validate_input(cls, v: Any, handler) -> Image.Image:
"""Pydantic-related function to validate input on instantiation."""
if isinstance(v, Image.Image):
return v
s = handler(v)
return Image.open(s)
@classmethod
def __get_pydantic_core_schema__(
cls,
source_type,
_handler,
) -> core_schema.CoreSchema:
assert source_type is Image.Image
return core_schema.no_info_wrap_validator_function(
function=cls.validate_input,
schema=core_schema.str_schema(),
serialization=core_schema.to_string_ser_schema(),
)
@classmethod
def __get_pydantic_json_schema__(cls, _core_schema, handler) -> JsonSchemaValue:
return handler(core_schema.str_schema())
shared/repositories/src/dto/model_data.py
+60
View File
@@ -0,0 +1,60 @@
"""Definition of ModelData DTO."""
from __future__ import annotations
from hashlib import md5
from io import BytesIO
from typing import Annotated
import torch
from pydantic import Field
from .bytes_io_pydantic_annotation import BytesIOPydanticAnnotation
from .hexadecimal_string import HexadecimalString
from .hexadecimal_string_pydantic_annotation import HexadecimalStringPydanticAnnotation
from .type_checking_base_model import TypeCheckingBaseModel
class ModelData(TypeCheckingBaseModel):
"""Model Data DTO."""
buffer: Annotated[BytesIO, BytesIOPydanticAnnotation]
buffer_checksum: Annotated[HexadecimalString, HexadecimalStringPydanticAnnotation]
class_name: str = Field(
min_length=1,
description='name of model class to generate data.',
)
@staticmethod
def calculate_checksum(buffer: BytesIO) -> HexadecimalString:
"""Calculate buffer checksum."""
checksum = md5(buffer.getbuffer()).hexdigest()
return HexadecimalString(checksum)
@staticmethod
def model_to_buffer(model: torch.nn.Module) -> BytesIO:
"""Save model to buffer."""
assert isinstance(model, torch.nn.Module)
buffer = BytesIO()
torch.save(model.state_dict(), buffer)
return buffer
@classmethod
def from_model(cls, model: torch.nn.Module) -> ModelData:
"""Instantiate from torch module."""
assert isinstance(model, torch.nn.Module)
# get model name
class_name = type(model).__name__
# save data to buffer
buffer = cls.model_to_buffer(model)
buffer = BytesIO()
torch.save(model.state_dict(), buffer)
# calculate checksum
buffer_checksum = cls.calculate_checksum(buffer)
# instantiate from buffer
data = cls(
buffer=buffer,
buffer_checksum=buffer_checksum,
class_name=class_name,
)
return data
shared/repositories/src/dto/type_checking_base_model.py
+12
View File
@@ -0,0 +1,12 @@
"""Definition of TypeCheckingBaseModel class."""
from pydantic import BaseModel, ConfigDict
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/src/dto/visual_communication_data/__init__.py
+2
View File
@@ -0,0 +1,2 @@
from .visual_communication_data import VisualCommunicationData
from .visual_communication_values import VisualCommunicationValues
shared/repositories/src/dto/visual_communication_data/angle_values.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of AngleValues DTO."""
from .values_model import ValuesModel
class AngleValues(ValuesModel):
"""Angle values DTO."""
high: float
eye_level: float
low: float
shared/repositories/src/dto/visual_communication_data/contact_values.py
+10
View File
@@ -0,0 +1,10 @@
"""Definition of ContactValues DTO."""
from .values_model import ValuesModel
class ContactValues(ValuesModel):
"""Contact values DTO."""
offer: float
demand: float
shared/repositories/src/dto/visual_communication_data/distance_values.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of DistanceValues DTO."""
from .values_model import ValuesModel
class DistanceValues(ValuesModel):
"""Distance values DTO."""
long: float
medium: float
close: float
shared/repositories/src/dto/visual_communication_data/framing_values.py
+12
View File
@@ -0,0 +1,12 @@
"""Definition of FramingValues DTO."""
from .values_model import ValuesModel
class FramingValues(ValuesModel):
"""Framing values DTO."""
frame_lines: float
empty_space: float
colour_contrast: float
form_contrast: float
shared/repositories/src/dto/visual_communication_data/information_value_values.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of InformationValueValues DTO."""
from .values_model import ValuesModel
class InformationValueValues(ValuesModel):
"""Information value values DTO."""
given_new: float
ideal_real: float
central_marginal: float
shared/repositories/src/dto/visual_communication_data/modality_color_values.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of ModalityColorValues DTO."""
from .values_model import ValuesModel
class ModalityColorValues(ValuesModel):
"""Modality color values DTO."""
high: float
medium: float
low: float
shared/repositories/src/dto/visual_communication_data/modality_depth_values.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of ModalityDepthValues DTO."""
from .values_model import ValuesModel
class ModalityDepthValues(ValuesModel):
"""Modality depth values DTO."""
high: float
medium: float
low: float
shared/repositories/src/dto/visual_communication_data/modality_lighting_values.py
+11
View File
@@ -0,0 +1,11 @@
"""Definition of ModalityLightingValues DTO."""
from .values_model import ValuesModel
class ModalityLightingValues(ValuesModel):
"""Modality lighting values DTO."""
high: float
medium: float
low: float
shared/repositories/src/dto/visual_communication_data/point_of_view_values.py
+10
View File
@@ -0,0 +1,10 @@
"""Definition of PointOfViewValues DTO."""
from .values_model import ValuesModel
class PointOfViewValues(ValuesModel):
"""Point-of-view values DTO."""
frontal: float
oblique: float
shared/repositories/src/dto/visual_communication_data/salience_values.py
+13
View File
@@ -0,0 +1,13 @@
"""Definition of SalienceValues DTO."""
from .values_model import ValuesModel
class SalienceValues(ValuesModel):
"""Salience values DTO."""
size: float
colour: float
tone: float
form: float
positioning: float
shared/mongodb/src/classes/data_model.py → shared/repositories/src/dto/visual_communication_data/values_model.py
+18 -25
View File
@@ -1,18 +1,20 @@
"""Definition of DataModel base class."""
"""Definition of ValuesModel base class."""
from __future__ import annotations
import random
from pydantic import BaseModel, ValidationError
from pydantic import BaseModel, ConfigDict
from torch import Tensor
class DataModel(BaseModel):
"""DataModel base class."""
class ValuesModel(BaseModel):
"""ValuesModel base class."""
@classmethod
def classname(cls) -> str:
"""Return classname."""
return cls.__name__
model_config = ConfigDict(
validate_assignment=True, # argument type checking
frozen=True, # ensure data immutability
)
@classmethod
def list_fields(cls) -> list[str]:
@@ -26,16 +28,16 @@ class DataModel(BaseModel):
return cls(**kwargs)
@classmethod
def from_choice(cls, option: str):
def from_choice(cls, option: str) -> ValuesModel:
"""Instantiate from choice."""
if option is None:
raise ValidationError()
assert isinstance(option, str), 'option is not a string'
assert isinstance(option, str)
assert len(option) > 0
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()}
if option not in allowed_options_list:
raise ValueError(f'option {option} must be in {allowed_options_list}')
# generate field values
kwargs = {field: 0 for field in allowed_options_list}
# set chosen value to max probability
kwargs[option] = 1
return cls(**kwargs)
@@ -47,15 +49,6 @@ class DataModel(BaseModel):
kwargs = dict(zip(cls.list_fields(), data_list))
return cls(**kwargs)
def __repr__(self) -> str:
model_dict = self.model_dump()
model_repr_str = f'{self.classname()}('
model_repr_str += ', '.join(
[f'{field}={value:.3f}' for field, value in model_dict.items()],
)
model_repr_str += ')'
return model_repr_str
def highest_score_field(self) -> str:
"""Return name of field with highest score."""
model_dict = self.model_dump()
shared/repositories/src/dto/visual_communication_data/visual_communication_data.py
+14
View File
@@ -0,0 +1,14 @@
"""Definition of VisualCommunicationData DTO."""
from pydantic import Field
from ..type_checking_base_model import TypeCheckingBaseModel
from .visual_communication_values import VisualCommunicationValues
class VisualCommunicationData(TypeCheckingBaseModel):
"""Visual communication data class."""
name: str = Field(min_length=1)
annotation: VisualCommunicationValues | None = None
prediction: VisualCommunicationValues | None = None
shared/repositories/src/dto/visual_communication_data/visual_communication_values.py
+49
View File
@@ -0,0 +1,49 @@
"""Definition of VisualCommunicationValues DTO."""
from __future__ import annotations
from ..type_checking_base_model import TypeCheckingBaseModel
from .angle_values import AngleValues
from .contact_values import ContactValues
from .distance_values import DistanceValues
from .framing_values import FramingValues
from .information_value_values import InformationValueValues
from .modality_color_values import ModalityColorValues
from .modality_depth_values import ModalityDepthValues
from .modality_lighting_values import ModalityLightingValues
from .point_of_view_values import PointOfViewValues
from .salience_values import SalienceValues
from .visual_syntax_values import VisualSyntaxValues
class VisualCommunicationValues(TypeCheckingBaseModel):
"""Visual communication values class."""
visual_syntax: VisualSyntaxValues
contact: ContactValues
angle: AngleValues
point_of_view: PointOfViewValues
distance: DistanceValues
modality_lighting: ModalityLightingValues
modality_color: ModalityColorValues
modality_depth: ModalityDepthValues
information_value: InformationValueValues
framing: FramingValues
salience: SalienceValues
@classmethod
def from_random(cls) -> VisualCommunicationValues:
"""Create a random instance."""
return cls(
visual_syntax=VisualSyntaxValues.from_random(),
contact=ContactValues.from_random(),
angle=AngleValues.from_random(),
point_of_view=PointOfViewValues.from_random(),
distance=DistanceValues.from_random(),
modality_lighting=ModalityLightingValues.from_random(),
modality_color=ModalityColorValues.from_random(),
modality_depth=ModalityDepthValues.from_random(),
information_value=InformationValueValues.from_random(),
framing=FramingValues.from_random(),
salience=SalienceValues.from_random(),
)
shared/mongodb/src/classes/visual_syntax.py → shared/repositories/src/dto/visual_communication_data/visual_syntax_values.py
+4 -6
View File
@@ -1,12 +1,10 @@
"""Definition of VisualSyntaxData data model."""
"""Definition of VisualSyntaxValues DTO."""
from __future__ import annotations
from .data_model import DataModel
from .values_model import ValuesModel
class VisualSyntaxData(DataModel):
"""VisualSyntaxData data model."""
class VisualSyntaxValues(ValuesModel):
"""Visual syntax values DTO."""
non_transactional_action: float
non_transactional_reaction: float
shared/repositories/src/image_repository.py
+74
View File
@@ -0,0 +1,74 @@
"""Definition of ImageRepository class."""
from __future__ import annotations
from io import BytesIO
from pathlib import Path
from PIL import Image
from .dto import ImageData
from .implementations import MinioImplementation
from .interfaces import ImageInterface
class ImageRepository(ImageInterface, MinioImplementation):
"""Image repository class that handles CRUD functionality for
VisualData."""
def __enter__(self) -> ImageRepository:
self.connect()
return self
@staticmethod
def _build_path(image_name: str) -> Path:
"""Build object path."""
assert isinstance(image_name, str)
path = Path('images') / image_name
return path
def get_data(self, image_name: str) -> ImageData | None:
"""Get Visual data."""
assert isinstance(image_name, str)
assert len(image_name) > 0
# build path
path = self._build_path(image_name)
# get object from bucket
buffer = self._get(path)
# handle if no data found
if not buffer:
return None
# convert data
image = Image.open(buffer)
data = ImageData(image=image, name=image_name)
return data
def put_data(self, data: ImageData) -> None:
"""Put visual data."""
assert isinstance(data, ImageData)
# build path
path = self._build_path(data.name)
# save image to buffer
buffer = BytesIO()
data.image.save(buffer, 'png')
# put object in bucket
self._put(path, buffer)
def remove_data(self, image_name: str) -> None:
"""Remove visual data."""
assert isinstance(image_name, str)
assert len(image_name) > 0
# build path
path = self._build_path(image_name)
# remove object
self._delete(path)
def list_names(self) -> list[str]:
"""List names of all images."""
# build path
path = self._build_path('')
# list object paths
obj_path_list = self._list_objects(path)
# strip prefix
name_list = [obj_path.split('/')[-1] for obj_path in obj_path_list]
return name_list
shared/repositories/src/implementations/__init__.py
+2
View File
@@ -0,0 +1,2 @@
from .minio_implementation import MinioImplementation
from .mongo_implementation import MongoImplementation
shared/repositories/src/implementations/minio_implementation.py
+186
View File
@@ -0,0 +1,186 @@
"""MinIO implementation."""
from __future__ import annotations
import logging
import os
from io import BytesIO
from pathlib import Path
from traceback import format_exc
from minio import Minio
from shared.utils import check_env
from ..interfaces import DatabaseInterface
class MinioImplementation(DatabaseInterface):
"""MinIO basic CRUD implementation."""
def __init__(self):
# ensure necessary env vars available
var_list = {
'MINIO_ENDPOINT',
'MINIO_ACCESS_KEY',
'MINIO_SECRET_KEY',
'MINIO_BUCKET_NAME',
}
check_env(var_list)
# prepare internal variables
self._client: Minio | None = None
self._bucket_name: str | None = None
def connect(self):
"""Connect to MinIO server."""
# prepare arguments
minio_endpoint = str(os.getenv('MINIO_ENDPOINT'))
minio_access_key = str(os.getenv('MINIO_ACCESS_KEY'))
minio_secret_key = str(os.getenv('MINIO_SECRET_KEY'))
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
# connect client
client = Minio(
endpoint=minio_endpoint,
access_key=minio_access_key,
secret_key=minio_secret_key,
secure=False,
)
# ensure bucket exists
if not client.bucket_exists(bucket_name=minio_bucket_name):
logging.debug('creating bucket: %s', minio_bucket_name)
client.make_bucket(bucket_name=minio_bucket_name)
# persist state
self._client = client
self._bucket_name = minio_bucket_name
def close(self) -> None:
"""Close connection to MinIO server.
N.B. MinIO connection cannot be closed manually.
"""
self._client = None
self._bucket_name = None
def connected(self):
"""Check connection to Minio."""
res = isinstance(self._client, Minio)
logging.debug(res)
return res
def __enter__(self) -> MinioImplementation:
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
if any(
(
exc_type is not None,
exc_val is not None,
exc_tb is not None,
),
):
logging.error('error while exiting context')
self.close()
def _put(
self,
path: Path,
buffer: BytesIO,
) -> None:
"""Save in-memory buffer as object in MinIO."""
assert isinstance(path, Path)
assert isinstance(buffer, BytesIO)
assert isinstance(self._client, Minio)
assert isinstance(self._bucket_name, str)
# prepare for saving
num_bytes = len(buffer.getvalue())
buffer.seek(0)
# send data to bucket
try:
self._client.put_object(
bucket_name=self._bucket_name,
object_name=path.as_posix(),
length=num_bytes,
data=buffer,
)
logging.debug('saved data to %s', path)
except Exception as exc:
logging.error('failed saving data to MinIO')
raise exc
def _get(
self,
path: Path,
) -> BytesIO | None:
"""Get object from MinIO as in-memory buffer."""
assert isinstance(path, Path)
assert isinstance(self._client, Minio)
assert isinstance(self._bucket_name, str)
try:
# make request
response = self._client.get_object(
bucket_name=self._bucket_name,
object_name=path.as_posix(),
)
assert response.status == 200
# get buffer
buffer = BytesIO()
chunk_size = 2**14
while chunk := response.read(chunk_size):
buffer.write(chunk)
buffer.seek(0)
logging.debug('got %s', path)
return buffer
except Exception:
logging.error('failed getting data from MinIO')
logging.debug(format_exc())
return None
finally:
# close connection if established
if 'response' in locals():
response.close()
response.release_conn()
def _delete(
self,
path: Path,
) -> None:
"""Delete object from MinIO."""
assert isinstance(path, Path)
assert isinstance(self._client, Minio)
assert isinstance(self._bucket_name, str)
# remove object
try:
self._client.remove_object(
bucket_name=self._bucket_name,
object_name=path.as_posix(),
)
logging.debug('deleted %s', path)
except Exception as exc:
logging.error('failed deleting %s', path)
logging.debug(format_exc())
raise exc
def _list_objects(
self,
path: Path,
) -> list[str]:
"""List objects in bucket under path."""
assert isinstance(path, Path)
assert isinstance(self._client, Minio)
assert isinstance(self._bucket_name, str)
try:
# list objects
obj_list = self._client.list_objects(
bucket_name=self._bucket_name,
prefix=path.as_posix(),
recursive=True,
)
# extract info
name_list = [obj.object_name for obj in obj_list]
logging.debug('got %s objects matching %s', len(name_list), path)
return name_list
except Exception as exc:
logging.error('failed listing objects under %s', path)
logging.debug(format_exc())
raise exc
shared/repositories/src/implementations/mongo_implementation.py
+145
View File
@@ -0,0 +1,145 @@
"""Mongo implementation."""
from __future__ import annotations
import logging
import os
import traceback
from pymongo import MongoClient
from pymongo.collection import Collection
from pymongo.database import Database
from pymongo.errors import ServerSelectionTimeoutError
from shared.utils import check_env
from ..interfaces import DatabaseInterface
class MongoImplementation(DatabaseInterface):
"""MongoDB basic CRUD implementation."""
def __init__(self):
# ensure necessary env vars available
var_list = {
'MONGO_ENDPOINT',
'MONGO_DB',
'MONGO_COLLECTION',
}
check_env(var_list)
# prepare internal variables
self.client: MongoClient | None = None
self.db: Database | None = None
self.collection: Collection | None = None
def connect(self) -> None:
"""Connect to Mongo server."""
# prepare arguments
mongo_endpoint = str(os.getenv('MONGO_ENDPOINT'))
mongo_database = str(os.getenv('MONGO_DB'))
mongo_collection = str(os.getenv('MONGO_COLLECTION'))
# connect client
client: MongoClient = MongoClient(mongo_endpoint)
database = client[mongo_database]
collection = database[mongo_collection]
# set unique index on 'name'
collection.create_index(keys='name', unique=True)
# persist state
self._client = client
self._database = database
self._collection = collection
def close(self) -> None:
"""Close connection to Mongo server."""
self._client.close()
self._client = None # type: ignore
self._database = None # type: ignore
self._collection = None # type: ignore
def connected(self) -> bool:
"""Check connection to Mongo."""
if self._client is None:
return False
try:
# trigger fetch data
_ = self._client.server_info()
res = True
except ServerSelectionTimeoutError:
res = False
logging.debug(res)
return res
def __enter__(self) -> MongoImplementation:
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
if any(
(
exc_type is not None,
exc_val is not None,
exc_tb is not None,
),
):
logging.error('error while exiting context')
traceback.print_exception(exc_type, exc_val, exc_tb)
self.close()
def _save(
self,
data: dict,
query: dict,
) -> None:
"""Save document in Mongo."""
assert isinstance(data, dict)
assert isinstance(query, dict)
assert 'name' in data
assert self.connected()
self._collection.update_one(
filter=query,
update={
'$set': data.copy(),
},
upsert=True,
)
logging.debug('Save %s', data)
def _get(
self,
query: dict,
) -> dict | None:
"""Get document from Mongo."""
assert isinstance(query, dict)
assert self.connected()
doc = self._collection.find_one(query, projection={'_id': False})
logging.debug('Found %s', doc)
return doc
def _delete(
self,
query: dict,
) -> None:
"""Remove document from Mongo."""
assert isinstance(query, dict)
assert self.connected()
doc = self._collection.delete_one(query)
logging.debug('Deleted %s', doc)
def _list_documents(self, key='name') -> list[str]:
"""List documents in Mongo."""
assert isinstance(key, str)
assert len(key) > 0
# build query
doc_list = list(
self._collection.find(
filter={},
projection={
'_id': False,
key: True,
},
),
)
# extract values
value_list = [doc[key] for doc in doc_list]
logging.debug('Got %s document(s)', len(doc_list))
return value_list
shared/repositories/src/interfaces/__init__.py
+4
View File
@@ -0,0 +1,4 @@
from .database_interface import DatabaseInterface
from .image_interface import ImageInterface
from .model_interface import ModelInterface
from .visual_communication_interface import VisualCommunicationInterface
shared/repositories/src/interfaces/database_interface.py
+28
View File
@@ -0,0 +1,28 @@
"""Definition of DatabaseInterface class."""
from abc import ABC, abstractmethod
class DatabaseInterface(ABC):
"""Interface base class adding 'connect', 'close' and context
functionalities."""
@abstractmethod
def connect(self):
raise NotImplementedError()
@abstractmethod
def close(self):
raise NotImplementedError()
@abstractmethod
def connected(self) -> bool:
raise NotImplementedError()
@abstractmethod
def __enter__(self):
raise NotImplementedError()
@abstractmethod
def __exit__(self, exc_type, exc_val, exc_tb):
raise NotImplementedError()
shared/repositories/src/interfaces/image_interface.py
+26
View File
@@ -0,0 +1,26 @@
"""Definition of ImageInterface."""
from abc import abstractmethod
from ..dto import ImageData
from .database_interface import DatabaseInterface
class ImageInterface(DatabaseInterface):
"""Image interface class."""
@abstractmethod
def get_data(self, image_name: str) -> ImageData | None:
raise NotImplementedError()
@abstractmethod
def put_data(self, data: ImageData) -> None:
raise NotImplementedError()
@abstractmethod
def remove_data(self, image_name: str) -> None:
raise NotImplementedError()
@abstractmethod
def list_names(self) -> list[str]:
raise NotImplementedError()
shared/repositories/src/interfaces/model_interface.py
+26
View File
@@ -0,0 +1,26 @@
"""Definition of ModelInterface."""
from abc import abstractmethod
from ..dto import ModelData
from .database_interface import DatabaseInterface
class ModelInterface(DatabaseInterface):
"""Model interface class."""
@abstractmethod
def get_data(self, object_name: str) -> ModelData | None:
raise NotImplementedError()
@abstractmethod
def put_data(self, data: ModelData) -> None:
raise NotImplementedError()
@abstractmethod
def remove_data(self, object_name: str) -> None:
raise NotImplementedError()
@abstractmethod
def list_names(self) -> list[str]:
raise NotImplementedError()
shared/repositories/src/interfaces/visual_communication_interface.py
+26
View File
@@ -0,0 +1,26 @@
"""Definition of VisualCommunicationInterface."""
from abc import abstractmethod
from ..dto import VisualCommunicationData
from .database_interface import DatabaseInterface
class VisualCommunicationInterface(DatabaseInterface):
"""Visual communication interface class."""
@abstractmethod
def get_data(self, name: str) -> VisualCommunicationData | None:
raise NotImplementedError()
@abstractmethod
def put_data(self, data: VisualCommunicationData) -> None:
raise NotImplementedError()
@abstractmethod
def remove_data(self, name: str) -> None:
raise NotImplementedError()
@abstractmethod
def list_names(self) -> list[str]:
raise NotImplementedError()
shared/repositories/src/model_repository.py
+77
View File
@@ -0,0 +1,77 @@
"""Definition of ModelRepository class."""
from __future__ import annotations
from pathlib import Path
from .dto import HexadecimalString, ModelData
from .implementations import MinioImplementation
from .interfaces import ModelInterface
class ModelRepository(ModelInterface, MinioImplementation):
"""Model repository class that handles CRUD functionality for ModelData."""
def __enter__(self) -> ModelRepository:
self.connect()
return self
@staticmethod
def _prefix() -> Path:
"""Object name prefix."""
return Path('models')
@classmethod
def _build_object_name(cls, data: ModelData) -> str:
"""Build object name from data."""
return f'{data.class_name}-{data.buffer_checksum}'
def get_data(self, object_name: str) -> ModelData | None:
"""Get model data."""
assert isinstance(object_name, str)
# build path
path = self._prefix() / object_name
# get object from bucket
buffer = self._get(path)
# handle if no data found
if not buffer:
return None
# extract info
class_name, buffer_checksum_str = object_name.split('-')
# convert data
buffer_checksum = HexadecimalString(buffer_checksum_str)
# instantiate data
data = ModelData(
buffer=buffer,
buffer_checksum=buffer_checksum,
class_name=class_name,
)
return data
def put_data(self, data: ModelData) -> None:
"""Put model data."""
assert isinstance(data, ModelData)
# build object name
object_name = self._build_object_name(data)
# build path
path = self._prefix() / object_name
# put object in bucket
self._put(path, data.buffer)
def remove_data(self, object_name: str) -> None:
"""Remove model data."""
assert isinstance(object_name, str)
# build path
path = self._prefix() / object_name
# remove object
self._delete(path)
def list_names(self) -> list[str]:
"""List names of all models."""
# build path
path = self._prefix()
# list object paths
obj_path_list = self._list_objects(path)
# strip prefix
name_list = [obj_path.split('/')[-1] for obj_path in obj_path_list]
return name_list
shared/repositories/src/visual_communication_repository.py
+56
View File
@@ -0,0 +1,56 @@
"""Definition of VisualCommunicationRepository class."""
from __future__ import annotations
from .dto import VisualCommunicationData
from .implementations import MongoImplementation
from .interfaces import VisualCommunicationInterface
class VisualCommunicationRepository(VisualCommunicationInterface, MongoImplementation):
"""Visual communication repository class that handles CRUD functionality
for VisualCommunicationData."""
def __enter__(self) -> VisualCommunicationRepository:
self.connect()
return self
def get_data(self, name: str) -> VisualCommunicationData | None:
"""Get visual communication data."""
assert isinstance(name, str)
assert len(name) > 0
# build query
query = {'name': name}
# get document from mongo
doc = self._get(query)
# handle if no data found
if doc is None:
return None
# instantiate object
data = VisualCommunicationData(**doc)
return data
def put_data(self, data: VisualCommunicationData) -> None:
"""Put visual communication data."""
assert isinstance(data, VisualCommunicationData)
# convert to dict
data_dict: dict = data.model_dump(mode='json')
# build query
query = {'name': data.name}
# save to mongo
self._save(data_dict, query)
def remove_data(self, name: str) -> None:
"""Remove visual communication data."""
assert isinstance(name, str)
assert len(name) > 0
# build query
query = {'name': name}
# remove document from mongo
self._delete(query)
def list_names(self) -> list[str]:
"""List names of all documents."""
# list documents
name_list = self._list_documents(key='name')
return name_list
shared/repositories/tests/integration/conftest.py
+403
View File
@@ -0,0 +1,403 @@
"""Integration tests configuration."""
import os
import random
from collections.abc import Iterator
from io import BytesIO
from pathlib import Path
import minio
import pytest
from dotenv import load_dotenv
from PIL import Image
from pymongo import MongoClient
from model.src.models import VisualCommunicationModel
from shared.repositories import (
HexadecimalString,
ImageData,
ImageRepository,
ModelData,
ModelRepository,
VisualCommunicationData,
VisualCommunicationRepository,
VisualCommunicationValues,
)
from shared.repositories.src.implementations import (
MinioImplementation,
MongoImplementation,
)
# set random seed for reproducibility
random.seed(13)
# 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',
'MINIO_IMAGE_NAME': 'test-image',
'MONGO_DB': 'test-db',
'MONGO_COLLECTION': 'test-collection',
}
@pytest.fixture(scope='session', autouse=True)
def setup_env(
request: pytest.FixtureRequest,
) -> None:
"""Populate environment with variables used for testing."""
# 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():
# set env var
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 raw_minio_client(
setup_env,
) -> Iterator[minio.Minio]:
"""Raw Minio client fixture."""
# 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.Minio(
endpoint=minio_endpoint,
access_key=minio_access_key,
secret_key=minio_secret_key,
secure=False,
)
# ensure bucket exists
if not client.bucket_exists(bucket_name=minio_bucket_name):
client.make_bucket(bucket_name=minio_bucket_name)
# expose client
yield client
# cleanup
object_list = client.list_objects(minio_bucket_name, recursive=True)
for obj in object_list:
client.remove_object(
bucket_name=obj.bucket_name,
object_name=obj.object_name,
)
client.remove_bucket(minio_bucket_name)
@pytest.fixture
def minio_client(
setup_env,
) -> Iterator[MinioImplementation]:
"""MinioImplementation fixture."""
# instantiate and connect client
minio_client = MinioImplementation()
minio_client.connect()
# expose client
yield minio_client
# cleanup
object_name_list = minio_client._list_objects(Path('*'))
for name in object_name_list:
minio_client._delete(Path(name))
minio_client.close()
@pytest.fixture
def buffer() -> Iterator[BytesIO]:
"""Bytes buffer fixture."""
# generate reproducible random data
data = random.randbytes(n=2**21) # 2 MB
# convert data
buffer = BytesIO(data)
# expose buffer
yield buffer
@pytest.fixture
def buffer_in_minio(
raw_minio_client: minio.Minio,
buffer: BytesIO,
) -> Iterator[tuple[Path, BytesIO]]:
"""Buffer in Minio fixture."""
# prepare arguments
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
minio_object_name = str(os.getenv('MINIO_OBJECT_NAME'))
# prepare for saving
num_bytes = len(buffer.getvalue())
buffer.seek(0)
# put data in bucket
raw_minio_client.put_object(
bucket_name=minio_bucket_name,
object_name=minio_object_name,
length=num_bytes,
data=buffer,
)
# expose data
yield Path(minio_object_name), buffer
# cleanup
raw_minio_client.remove_object(
bucket_name=minio_bucket_name,
object_name=minio_object_name,
)
@pytest.fixture
def image_data() -> Iterator[ImageData]:
"""Image data fixture."""
# prepare arguments
name = str(os.getenv('MINIO_IMAGE_NAME'))
image = Image.new(mode='RGB', size=(480, 480))
# instantiate data
image_data = ImageData(image=image, name=name)
# expose data
yield image_data
@pytest.fixture
def image_data_in_minio(
raw_minio_client: minio.Minio,
image_data: ImageData,
) -> Iterator[ImageData]:
"""Image data in Minio fixture."""
# prepare arguments
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
image_name = image_data.name
# build object path
object_path = ImageRepository._build_path(image_name)
# save image to buffer
buffer = BytesIO()
image_data.image.save(buffer, 'png')
# prepare for saving
num_bytes = len(buffer.getvalue())
buffer.seek(0)
# put data in bucket
raw_minio_client.put_object(
bucket_name=minio_bucket_name,
object_name=object_path.as_posix(),
length=num_bytes,
data=buffer,
)
# expose data
yield image_data
# cleanup
raw_minio_client.remove_object(
bucket_name=minio_bucket_name,
object_name=object_path.as_posix(),
)
@pytest.fixture(scope='session')
def image_repo(
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
repo.close()
@pytest.fixture
def model_data() -> Iterator[ModelData]:
"""Model data fixture."""
# prepare arguments
vis_com_model = VisualCommunicationModel().to('cpu')
class_name = type(vis_com_model).__name__
buffer = ModelData.model_to_buffer(vis_com_model)
buffer_checksum = HexadecimalString('77dcab1769563654a6e24f92d40f29bd')
# instantiate data
model_data = ModelData(
buffer=buffer,
buffer_checksum=buffer_checksum,
class_name=class_name,
)
# expose model
yield model_data
@pytest.fixture
def model_data_in_minio(
raw_minio_client: minio.Minio,
model_data: ModelData,
) -> Iterator[ModelData]:
"""Model data in Minio fixture."""
# prepare arguments
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
object_name = ModelRepository._build_object_name(model_data)
buffer = model_data.buffer
# build object path
object_path = ModelRepository._prefix() / object_name
# prepare for saving
num_bytes = len(buffer.getvalue())
buffer.seek(0)
# put data in bucket
raw_minio_client.put_object(
bucket_name=minio_bucket_name,
object_name=object_path.as_posix(),
length=num_bytes,
data=buffer,
)
# expose data
yield model_data
# cleanup
raw_minio_client.remove_object(
bucket_name=minio_bucket_name,
object_name=object_path.as_posix(),
)
@pytest.fixture(scope='session')
def model_repo(
setup_env,
) -> Iterator[ModelRepository]:
"""Model repository fixture."""
repo = ModelRepository()
repo.connect()
yield repo
repo.close()
@pytest.fixture
def raw_mongo_client(
setup_env,
) -> Iterator[MongoClient]:
"""Raw mongo client fixture."""
# prepare arguments
mongo_endpoint = str(os.getenv('MONGO_ENDPOINT'))
mongo_database = str(os.getenv('MONGO_DB'))
# connect client
client: MongoClient = MongoClient(mongo_endpoint)
_ = client[mongo_database]
# expose client
yield client
# cleanup
client.drop_database(mongo_database)
@pytest.fixture
def mongo_client(
setup_env,
) -> Iterator[MongoImplementation]:
"""MongoImplementation fixture."""
# instantiate and connect client
mongo_client = MongoImplementation()
mongo_client.connect()
# expose client
yield mongo_client
# cleanup
mongo_client._collection.drop()
mongo_client.close()
@pytest.fixture
def dictionary() -> Iterator[dict]:
"""Dictionary fixture."""
# prepare data
data = {
'name': 'test-dictionary',
'str_key': 'value',
'int_key': 100,
'float_key': 3.14,
'list_key': [1, 2, 3],
'dict_key': {
'nested_key': 'nested_value',
},
}
# expose data
yield data
@pytest.fixture
def dictionary_in_mongo(
raw_mongo_client: MongoClient,
dictionary: dict,
) -> Iterator[dict]:
"""Dictionary in Mongo fixture."""
# prepare arguments
database = str(os.getenv('MONGO_DB'))
collection = str(os.getenv('MONGO_COLLECTION'))
# save data
_ = raw_mongo_client[database][collection].insert_one(dictionary.copy())
# expose data
yield dictionary
# cleanup
raw_mongo_client[database][collection].delete_one(dictionary)
@pytest.fixture
def visual_communication_values() -> Iterator[VisualCommunicationValues]:
"""Visual communication values fixture."""
# instantiate with random values
visual_communication_values = VisualCommunicationValues.from_random()
# expose values
yield visual_communication_values
@pytest.fixture
def visual_communication_data(
visual_communication_values: VisualCommunicationValues,
) -> Iterator[VisualCommunicationData]:
"""Visual communication data fixture."""
# prepare arguments
name = 'test-visual-communication'
annotation = visual_communication_values
# instantiate data
visual_communication_data = VisualCommunicationData(
name=name,
annotation=annotation,
)
# expose data
yield visual_communication_data
@pytest.fixture
def visual_communication_data_in_mongo(
raw_mongo_client: MongoClient,
visual_communication_data: VisualCommunicationData,
) -> Iterator[VisualCommunicationData]:
"""Visual communication data in Mongo fixture."""
# prepare arguments
database = str(os.getenv('MONGO_DB'))
collection = str(os.getenv('MONGO_COLLECTION'))
# convert data
dictionary = visual_communication_data.model_dump(mode='dict')
# save data
_ = raw_mongo_client[database][collection].insert_one(dictionary.copy())
# expose data
yield visual_communication_data
# cleanup
raw_mongo_client[database][collection].delete_one(dictionary)
@pytest.fixture(scope='session')
def visual_communication_repo(
setup_env,
) -> Iterator[VisualCommunicationRepository]:
"""Visual communication repository fixture."""
repo = VisualCommunicationRepository()
repo.connect()
yield repo
repo.close()
shared/repositories/tests/integration/image_repository_test.py
+156
View File
@@ -0,0 +1,156 @@
"""Integration tests for ImageRepository class."""
import numpy as np
import pytest
from PIL import Image
from shared.repositories import ImageData, ImageRepository
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 same_image_data(
data_a: ImageData,
data_b: ImageData,
) -> bool:
"""Check if two ImageData-objects contain the same data."""
assert isinstance(data_a, ImageData)
assert isinstance(data_b, ImageData)
# compare names
if data_a.name != data_b.name:
return False
# compare images
if not same_image(data_a.image, data_b.image):
return False
return True
def test_should_have_context_handler():
"""Test that class has a working context handler implemented."""
# ACT
with ImageRepository() as repo:
# ASSERT
assert repo.connected()
def test_should_get_image_data(
image_repo: ImageRepository,
image_data_in_minio: ImageData,
):
"""Test getting image data."""
# ARRANGE
image_name = image_data_in_minio.name
# ACT
received_image_data = image_repo.get_data(image_name)
# ASSERT
assert received_image_data is not None
assert isinstance(received_image_data, ImageData)
assert same_image_data(image_data_in_minio, received_image_data)
def test_should_get_none_when_no_image_data(
image_repo: ImageRepository,
image_data: ImageData,
):
"""Test getting None when no data is available."""
# ARRANGE
image_name = image_data.name
# ACT
received_image_data = image_repo.get_data(image_name)
# ASSERT
assert received_image_data is None
def test_should_delete_image_data(
image_repo: ImageRepository,
image_data_in_minio: ImageData,
):
"""Test deleting image data."""
# ARRANGE
image_name = image_data_in_minio.name
# ACT
image_repo.remove_data(image_name)
received_image_data = image_repo.get_data(image_name)
# ASSERT
assert received_image_data is None
def test_should_put_image_data(
image_repo: ImageRepository,
image_data: ImageData,
):
"""Test putting image data."""
# ARRANGE
image_name = image_data.name
# ACT
image_repo.put_data(image_data)
received_image_data = image_repo.get_data(image_name)
# ASSERT
assert received_image_data is not None
assert same_image_data(image_data, received_image_data)
def test_should_update_image_data(
image_repo: ImageRepository,
image_data_in_minio: ImageData,
):
"""Test updating image data."""
# ARRANGE
updated_image_data = image_data_in_minio.model_copy(
update={
'image': Image.new(mode='RGB', size=(480, 480), color='white'),
},
)
image_name = updated_image_data.name
# ACT
image_repo.put_data(updated_image_data)
received_image_data = image_repo.get_data(image_name)
# ASSERT
assert not same_image_data(image_data_in_minio, updated_image_data)
assert received_image_data is not None
assert same_image_data(updated_image_data, received_image_data)
def test_should_list_names(
image_repo: ImageRepository,
image_data_in_minio: ImageData,
):
"""Test get all image names."""
# ARRANGE
updated_image_data = image_data_in_minio.model_copy(
update={
'name': 'updated-test-image',
},
)
image_repo.put_data(updated_image_data)
expected_name_list = [
image_data_in_minio.name,
updated_image_data.name,
]
# ACT
name_list = image_repo.list_names()
# ASSERT
assert len(name_list) == 2
for name in name_list:
assert name in expected_name_list
if __name__ == '__main__':
pytest.main(['-s', '-v', __file__])
shared/repositories/tests/integration/minio_implementation_test.py
+133
View File
@@ -0,0 +1,133 @@
"""Integration tests related to Minio implementation."""
import logging
import os
from io import BytesIO
from pathlib import Path
import pytest
from shared.repositories.src.implementations import MinioImplementation
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_connect_to_minio():
"""Test connection to Minio."""
# ARRANGE
client = MinioImplementation()
# ACT
client.connect()
# ASSERT
assert client.connected()
client.close()
def test_should_have_context_handler():
"""Test that class has a working context handler implemented."""
# ACT
with MinioImplementation() as client:
# ASSERT
assert client.connected()
def test_should_get_data(
minio_client: MinioImplementation,
buffer_in_minio: tuple[Path, BytesIO],
):
"""Test getting data from Minio."""
# ARRANGE
path, buffer = buffer_in_minio
# ACT
received_buffer = minio_client._get(path)
# ASSERT
assert received_buffer is not None
assert same_data(received_buffer, buffer)
def test_should_get_none_when_no_data(
minio_client: MinioImplementation,
):
"""Test getting None when no data is available in Minio."""
# ARRANGE
nonexistent_path = Path('nonexistent-object-name')
# ACT
received_buffer = minio_client._get(nonexistent_path)
# ASSERT
assert received_buffer is None
def test_should_delete_data(
minio_client: MinioImplementation,
buffer_in_minio: tuple[Path, BytesIO],
):
"""Test deleting data from Minio."""
# ARRANGE
path, _ = buffer_in_minio
# ACT
minio_client._delete(path)
# ASSERT
received_buffer = minio_client._get(path)
assert received_buffer is None
def test_should_put_data(
minio_client: MinioImplementation,
buffer: BytesIO,
):
"""Test putting data in Minio."""
# ARRANGE
path = Path(os.getenv('MINIO_OBJECT_NAME', default=''))
# ACT
minio_client._put(path, buffer)
received_buffer = minio_client._get(path)
# ASSERT
assert received_buffer is not None
assert same_data(received_buffer, buffer)
def test_should_update_data(
minio_client: MinioImplementation,
buffer_in_minio: tuple[Path, BytesIO],
):
"""Test updating data in Minio."""
# ARRANGE
path, buffer = buffer_in_minio
updated_buffer = BytesIO(buffer.getvalue() + b'extra data')
# ACT
minio_client._put(path, updated_buffer)
received_buffer = minio_client._get(path)
# ASSERT
assert not same_data(updated_buffer, buffer)
assert received_buffer is not None
assert same_data(received_buffer, updated_buffer)
if __name__ == '__main__':
pytest.main(['-s', '-v', __file__])
shared/repositories/tests/integration/model_repository_test.py
+136
View File
@@ -0,0 +1,136 @@
"""Integration tests for ModelRepository class."""
from io import BytesIO
import pytest
from shared.repositories import ModelData, ModelRepository
def same_buffer(
buffer_a: BytesIO,
buffer_b: BytesIO,
) -> bool:
"""Check if 2 buffers contain the same data."""
assert isinstance(buffer_a, BytesIO)
assert isinstance(buffer_b, BytesIO)
# read buffers
a_values = buffer_a.getvalue()
b_values = buffer_b.getvalue()
# compare length of buffers
if len(a_values) != len(b_values):
return False
# compare content of buffers
if a_values != b_values:
return False
return True
def same_model_data(
data_a: ModelData,
data_b: ModelData,
) -> bool:
"""Check if to ModelData-objects contain the same data."""
assert isinstance(data_a, ModelData)
assert isinstance(data_b, ModelData)
# compare names
if data_a.buffer_checksum != data_b.buffer_checksum:
return False
# compare buffer
if not same_buffer(data_a.buffer, data_b.buffer):
return False
return True
def test_should_have_context_handler():
"""Test that class has a working context handler implemented."""
# ACT
with ModelRepository() as repo:
# ASSERT
assert repo.connected()
def test_should_get_model_data(
model_repo: ModelRepository,
model_data_in_minio: ModelData,
):
"""Test getting model data."""
# ARRANGE
object_name = ModelRepository._build_object_name(model_data_in_minio)
# ACT
received_model_data = model_repo.get_data(object_name)
# ASSERT
assert received_model_data is not None
assert isinstance(received_model_data, ModelData)
assert same_model_data(model_data_in_minio, received_model_data)
def test_should_get_none_when_no_model_data(
model_repo: ModelRepository,
model_data: ModelData,
):
"""Test getting None whne no data is available."""
# ARRANGE
object_name = ModelRepository._build_object_name(model_data)
# ACT
received_model_data = model_repo.get_data(object_name)
# ASSERT
assert received_model_data is None
def test_should_delete_model_data(
model_repo: ModelRepository,
model_data_in_minio: ModelData,
):
"""Test deleting model data."""
# ARRANGE
object_name = ModelRepository._build_object_name(model_data_in_minio)
# ACT
model_repo.remove_data(object_name)
received_model_data = model_repo.get_data(object_name)
# ASSERT
assert received_model_data is None
def test_should_put_model_data(
model_repo: ModelRepository,
model_data: ModelData,
):
"""Test putting model data."""
# ARRANGE
object_name = ModelRepository._build_object_name(model_data)
# ACT
model_repo.put_data(model_data)
received_model_data = model_repo.get_data(object_name)
# ASSERT
assert received_model_data is not None
assert same_model_data(model_data, received_model_data)
def test_should_list_names(
model_repo: ModelRepository,
model_data_in_minio: ModelData,
):
"""Test get all model names."""
# ARRANGE
new_buffer_checksum = ModelData.calculate_checksum(model_data_in_minio.buffer)
updated_model_data = model_data_in_minio.model_copy(
update={
'buffer_checksum': new_buffer_checksum,
},
)
model_repo.put_data(updated_model_data)
expected_name_list = [
ModelRepository._build_object_name(model_data_in_minio),
ModelRepository._build_object_name(updated_model_data),
]
# ACT
name_list = model_repo.list_names()
# ASSERT
assert len(name_list) == 2
for name in name_list:
assert name in expected_name_list
if __name__ == '__main__':
pytest.main(['-s', '-v', __file__])
shared/repositories/tests/integration/mongo_implementation_test.py
+113
View File
@@ -0,0 +1,113 @@
"""Integration tests related to Mongo implementation."""
import pytest
from shared.repositories.src.implementations import MongoImplementation
def test_should_connect_mongo():
"""Test connecting to Mongo."""
# ARRANGE
client = MongoImplementation()
# ACT
client.connect()
# ASSERT
assert client.connected()
client.close()
def test_should_have_context_handler():
"""Test that class has a working context handler implemented."""
# ACT
with MongoImplementation() as client:
# ASSERT
assert client.connected()
def test_should_get_data(
mongo_client: MongoImplementation,
dictionary_in_mongo: dict,
):
"""Test getting data from mongo."""
# ARRANGE
name = dictionary_in_mongo['name']
query = {'name': name}
# ACT
received_dictionary = mongo_client._get(query)
# ASSERT
assert received_dictionary is not None
assert isinstance(received_dictionary, dict)
assert received_dictionary == dictionary_in_mongo
def test_should_get_none_when_no_data(
mongo_client: MongoImplementation,
dictionary: dict,
):
"""Test getting None when no data is available in mongo."""
# ARRANGE
name = dictionary['name']
query = {'name': name}
# ACT
received_dictionary = mongo_client._get(query)
# ASSERT
assert received_dictionary is None
def test_should_delete_data(
mongo_client: MongoImplementation,
dictionary_in_mongo: dict,
):
"""Test deleting data from mongo."""
# ARRANGE
name = dictionary_in_mongo['name']
query = {'name': name}
# ACT
mongo_client._delete(query)
received_dictionary = mongo_client._get(query)
# ASSERT
assert received_dictionary is None
def test_should_save_data(
mongo_client: MongoImplementation,
dictionary: dict,
):
"""Test saving data to mongo."""
# ARRANGE
name = dictionary['name']
query = {'name': name}
# ACT
mongo_client._save(
data=dictionary,
query=query,
)
# ASSERT
received_dictionary = mongo_client._get(query)
assert received_dictionary is not None
assert isinstance(received_dictionary, dict)
assert received_dictionary == dictionary
def test_should_update_data(
mongo_client: MongoImplementation,
dictionary_in_mongo: dict,
):
"""Test updating data in mongo."""
# ARRANGE
name = dictionary_in_mongo['name']
query = {'name': name}
# ACT
dictionary_in_mongo['str-key'] = 'updated-value'
mongo_client._save(
data=dictionary_in_mongo,
query=query,
)
# ASSERT
received_dictionary = mongo_client._get(query)
assert received_dictionary is not None
assert received_dictionary == dictionary_in_mongo
if __name__ == '__main__':
pytest.main(['-s', '-v', __file__])
shared/repositories/tests/integration/visual_communication_repository_test.py
+125
View File
@@ -0,0 +1,125 @@
"""Integration tests for VisualCommunicationRepository class."""
import pytest
from shared.repositories import (
VisualCommunicationData,
VisualCommunicationRepository,
VisualCommunicationValues,
)
def test_should_have_context_handler():
"""Test that class has a working context handler implemented."""
# ACT
with VisualCommunicationRepository() as repo:
# ASSERT
assert repo.connected()
def test_should_get_visual_communication_data(
visual_communication_repo: VisualCommunicationRepository,
visual_communication_data_in_mongo: VisualCommunicationData,
):
"""Test getting visual communication data."""
# ARRANGE
name = visual_communication_data_in_mongo.name
# ACT
received_visual_communication_data = visual_communication_repo.get_data(name)
# ASSERT
assert received_visual_communication_data is not None
assert isinstance(received_visual_communication_data, VisualCommunicationData)
assert received_visual_communication_data == visual_communication_data_in_mongo
def test_should_get_none_when_no_visual_communication_data(
visual_communication_repo: VisualCommunicationRepository,
visual_communication_data: VisualCommunicationData,
):
"""Test getting None when no data is available."""
# ARRANGE
name = visual_communication_data.name
# ACT
received_visual_communication_data = visual_communication_repo.get_data(name)
# ASSERT
assert received_visual_communication_data is None
def test_should_delete_image_data(
visual_communication_repo: VisualCommunicationRepository,
visual_communication_data_in_mongo: VisualCommunicationData,
):
"""Test deleting visual communication data."""
# ARRANGE
name = visual_communication_data_in_mongo.name
# ACT
visual_communication_repo.remove_data(name)
received_visual_communication_data = visual_communication_repo.get_data(name)
# ASSERT
assert received_visual_communication_data is None
def test_should_put_visual_communication_data(
visual_communication_repo: VisualCommunicationRepository,
visual_communication_data: VisualCommunicationData,
):
"""Test putting visual communication data."""
# ARRANGE
name = visual_communication_data.name
# ACT
visual_communication_repo.put_data(visual_communication_data)
received_visual_communication_data = visual_communication_repo.get_data(name)
# ASSERT
assert received_visual_communication_data is not None
assert isinstance(received_visual_communication_data, VisualCommunicationData)
assert received_visual_communication_data == visual_communication_data
def test_should_update_visual_communication_data(
visual_communication_repo: VisualCommunicationRepository,
visual_communication_data_in_mongo: VisualCommunicationData,
):
"""Test updating visual communication data."""
# ARRANGE
name = visual_communication_data_in_mongo.name
updated_annotation = VisualCommunicationValues.from_random()
updated_visual_communication = VisualCommunicationData(
name=name,
annotation=updated_annotation,
)
# ACT
visual_communication_repo.put_data(updated_visual_communication)
received_visual_communication_data = visual_communication_repo.get_data(name)
# ASSERT
assert received_visual_communication_data is not None
assert isinstance(received_visual_communication_data, VisualCommunicationData)
assert received_visual_communication_data == updated_visual_communication
def test_should_list_names(
visual_communication_repo: VisualCommunicationRepository,
visual_communication_data_in_mongo: VisualCommunicationData,
):
"""Test listing names of all documents."""
# ARRANGE
name = 'test-visual-communication-2'
annotation = VisualCommunicationValues.from_random()
second_visual_communication = VisualCommunicationData(
name=name,
annotation=annotation,
)
visual_communication_repo.put_data(second_visual_communication)
expected_name_list = [
visual_communication_data_in_mongo.name,
second_visual_communication.name,
]
# ACT
name_list = visual_communication_repo.list_names()
# ASSERT
assert len(name_list) == 2
for name in name_list:
assert name in expected_name_list
if __name__ == '__main__':
pytest.main(['-s', '-v', __file__])
shared/utils/check_env.py
+4 -1
View File
@@ -10,6 +10,9 @@ def check_env(
assert isinstance(var_list, set)
assert all(isinstance(elem, str) for elem in var_list)
# check that env vars are set
missing_vars = []
for env_var in var_list:
if env_var not in os.environ:
raise OSError(f"environment variable not set: {env_var}")
missing_vars.append(env_var)
if missing_vars:
raise OSError(f"environment variable(s) not set: {missing_vars}")

Some files were not shown because too many files have changed in this diff Show More