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

removed deprecated modules

Browse Source
This commit is contained in:
brian
2025-03-18 19:21:17 +00:00
parent 0b54ccaadb
commit a379270add
41 changed files with 10 additions and 2113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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 internal variables
self._client: Minio | None = None
self._bucket_name: str | None = None
def connect(self) -> None:
"""Connect to Minio server."""
# prepare arguments
minio_endpoint = str(os.getenv('MINIO_ENDPOINT'))
minio_access_key = str(os.getenv('MINIO_ACCESS_KEY'))
minio_secret_key = str(os.getenv('MINIO_SECRET_KEY'))
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
# connect client
client = Minio(
endpoint=minio_endpoint,
access_key=minio_access_key,
secret_key=minio_secret_key,
secure=False,
)
# ensure bucket exists
if not client.bucket_exists(bucket_name=minio_bucket_name):
logging.debug('creating bucket: %s', minio_bucket_name)
client.make_bucket(bucket_name=minio_bucket_name)
logging.debug('finished')
# persist state
self._client = client
self._bucket_name = minio_bucket_name
def close(self) -> None:
"""Close connection to Minio server.
N.B. Minio connection cannot be closed manually.
"""
self._client = None
self._bucket_name = None
def __enter__(self) -> DatastoreMinio:
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
if any(
(
exc_type is not None,
exc_val is not None,
exc_tb is not None,
),
):
logging.error('error while exiting context')
self.close()
def _put(
self,
object_name: str,
buffer: BytesIO,
) -> None:
"""Save in-memory buffer as object in Minio."""
assert isinstance(object_name, str)
assert len(object_name) > 0
assert isinstance(buffer, BytesIO)
assert isinstance(self._client, Minio)
assert isinstance(self._bucket_name, str)
# prepare for saving
num_bytes = len(buffer.getvalue())
buffer.seek(0)
# send data to bucket
try:
self._client.put_object(
bucket_name=self._bucket_name,
object_name=object_name,
length=num_bytes,
data=buffer,
)
logging.debug('saved data to %s', object_name)
except Exception as exc:
logging.error('failed saving data to MinIO')
raise exc
def _get(
self,
object_name: str,
) -> BytesIO:
"""Get object from Minio as in-memory buffer."""
assert isinstance(object_name, str)
assert len(object_name) > 0
assert isinstance(self._client, Minio)
assert isinstance(self._bucket_name, str)
try:
# make request
response = self._client.get_object(
bucket_name=self._bucket_name,
object_name=object_name,
)
assert response.status == 200
# get buffer
buffer = BytesIO()
chunk_size = 2**14
while chunk := response.read(chunk_size):
buffer.write(chunk)
buffer.seek(0)
logging.debug('got %s', object_name)
return buffer
except Exception as exc:
logging.error('failed getting data from MinIO')
logging.debug(format_exc())
raise exc
finally:
# close connection if established
if 'response' in locals():
response.close()
response.release_conn()
def _delete(
self,
object_name: str,
) -> None:
"""Delete object from Minio."""
assert isinstance(object_name, str)
assert len(object_name) > 0
assert isinstance(self._client, Minio)
assert isinstance(self._bucket_name, str)
# remove object
try:
self._client.remove_object(
bucket_name=self._bucket_name,
object_name=object_name,
)
logging.debug('deleted %s', object_name)
except Exception as exc:
logging.error('failed deleting %s', object_name)
logging.debug(format_exc())
raise exc
def put_image(
self,
image: Image.Image,
) -> str:
"""Put image in Minio."""
assert isinstance(image, Image.Image)
# save data to buffer
buffer = BytesIO()
image.save(buffer, 'png')
# get md5 of buffer
checksum = md5(buffer.getbuffer()).hexdigest()
# build object path
object_path = f'images/{checksum}'
# send data to bucket
self._put(
object_name=object_path,
buffer=buffer,
)
logging.debug('saved data to %s', object_path)
return checksum
def get_image(
self,
object_name: str,
) -> Image.Image:
"""Get image from Minio."""
assert isinstance(object_name, str)
assert len(object_name) > 0
# build object path
object_path = f'images/{object_name}'
# get object from bucket
buffer = self._get(
object_name=object_path,
)
# convert data to image
image = Image.open(buffer)
logging.debug('got data from %s', object_path)
return image
def put_model(
self,
model: torch.nn.Module,
) -> str:
"""Put model in Minio."""
assert isinstance(model, torch.nn.Module)
# save data to buffer
buffer = BytesIO()
torch.save(model.state_dict(), buffer)
# get md5 of image
checksum = md5(buffer.getbuffer()).hexdigest()
# build object path
object_path = f'models/{checksum}'
# send data to bucket
self._put(
object_name=object_path,
buffer=buffer,
)
logging.debug('saved data to %s', object_path)
return checksum
def get_model(
self,
object_name: str,
) -> OrderedDict:
"""Get model data from Minio."""
assert isinstance(object_name, str)
assert len(object_name) > 0
# build object path
object_path = f'models/{object_name}'
# get object from bucket
buffer = self._get(
object_name=object_path,
)
# convert data to model checkpoint
model_content = torch.load(buffer)
logging.debug('got data from %s', object_path)
return model_content
shared/datastore/tests/integration/base_crud_test.py
-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 Minio implementation."""
import os
from hashlib import md5
from io import BytesIO
from unittest import TestCase
from unittest.mock import ANY, MagicMock, patch
from minio import Minio
from PIL import Image
from urllib3 import BaseHTTPResponse
from shared.datastore.src.datastore_minio import DatastoreMinio
class TestDatastoreMinioImplementation(TestCase):
def setUp(self):
# define relevant env vars
self.env_var_map = {
'MINIO_ENDPOINT': '192.168.1.2',
'MINIO_ACCESS_KEY': 'randomAccess_key',
'MINIO_SECRET_KEY': 'randomSecret_key',
'MINIO_BUCKET_NAME': 'test-bucket-name',
}
# set env vars
for key, val in self.env_var_map.items():
os.environ[key] = val
# set other variables
self.object_name = '46KXJMFIAPVLM0TKRFZR5YPPTVJ6PJNX'
self.image = Image.new(mode='RGB', size=(480, 480))
self.buffer = BytesIO()
self.image.save(self.buffer, 'png')
self.num_bytes = len(self.buffer.getvalue())
self.checksum = md5(self.buffer.getbuffer()).hexdigest()
def tearDown(self):
# clear env vars
for key in self.env_var_map:
_ = os.environ.pop(key, default=None)
def test_instantiation_should_fail_when_env_not_set(self):
# ensure env not set
self.tearDown()
# run test
with self.assertRaises(OSError):
_ = DatastoreMinio()
@patch('shared.datastore.src.datastore_minio.Minio')
def test_connect_should_call_Minio_with_env_vars(self, minio_mock):
# ARRANGE
datastore = DatastoreMinio()
minio_mock().bucket_exists.return_value = False
# ACT
datastore.connect()
# ASSERT
minio_mock.assert_called_with(
endpoint=self.env_var_map['MINIO_ENDPOINT'],
access_key=self.env_var_map['MINIO_ACCESS_KEY'],
secret_key=self.env_var_map['MINIO_SECRET_KEY'],
secure=False,
)
minio_mock().bucket_exists.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
)
minio_mock().make_bucket.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
)
def test_close_should_overwrite_private_variables(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock()
datastore._bucket_name = MagicMock()
assert isinstance(datastore._client, MagicMock)
assert isinstance(datastore._bucket_name, MagicMock)
# ACT
datastore.close()
# ASSERT
assert datastore._client is None
assert datastore._bucket_name is None
@patch('shared.datastore.src.datastore_minio.DatastoreMinio.connect')
@patch('shared.datastore.src.datastore_minio.DatastoreMinio.close')
def test_context_management_implemented(
self,
mocked_close_method,
mocked_connect_method,
):
# ARRANGE, ACT and ASSERT
with DatastoreMinio() as ds:
mocked_connect_method.assert_called_once()
assert isinstance(ds, DatastoreMinio)
mocked_close_method.assert_called_once()
def test_should_call_put_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
self.buffer.seek(0)
# ACT
datastore._put(
object_name=self.object_name,
buffer=self.buffer,
)
# ASSERT
datastore._client.put_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=self.object_name,
length=self.num_bytes,
data=self.buffer,
)
def test_should_call_get_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._client.get_object.return_value = MagicMock(
BaseHTTPResponse,
status=200,
)
# datastore._client.get_object.read.side_effect = [b'random ', b'test', b'text']
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
# ACT
with self.assertRaises(TypeError): # dont care to mock even more...
datastore._get(
object_name=self.object_name,
)
# ASSERT
datastore._client.get_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=self.object_name,
)
def test_should_call_remove_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
# ACT
datastore._delete(
object_name=self.object_name,
)
# ASSERT
datastore._client.remove_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=self.object_name,
)
def test_put_image_should_call_put_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
object_path = f'images/{self.checksum}'
# ACT
datastore.put_image(
image=self.image,
)
# ASSERT
datastore._client.put_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=object_path,
length=self.num_bytes,
data=ANY, # saved to different buffer when converting image
)
def test_get_image_should_call_get_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._client.get_object.return_value = MagicMock(
BaseHTTPResponse,
status=200,
)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
object_path = f'images/{self.checksum}'
# ACT
with self.assertRaises(TypeError): # dont care to mock even more...
datastore.get_image(
object_name=self.checksum,
)
# ASSERT
datastore._client.get_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=object_path,
)
# @patch('shared.datastore.src.datastore_minio.torch.serialization.save')
# def test_put_model_should_call_put_object_with_arguments(self, mocked_torch_fn):
# # ARRANGE
# datastore = DatastoreMinio()
# datastore._client = MagicMock(Minio)
# datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
# object_path = f'images/{self.checksum}'
# mocked_torch_fn.return_value = None
# model = MagicMock(torch.nn.Module)
# # ACT
# datastore.put_model(
# model=model,
# )
# # ASSERT
# datastore._client.put_object.assert_called_with(
# bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
# object_name=object_path,
# length=self.num_bytes,
# data=ANY, # saved to different buffer when converting data
# )
def test_get_model_should_call_get_object_with_arguments(self):
# ARRANGE
datastore = DatastoreMinio()
datastore._client = MagicMock(Minio)
datastore._client.get_object.return_value = MagicMock(
BaseHTTPResponse,
status=200,
)
datastore._bucket_name = self.env_var_map['MINIO_BUCKET_NAME']
object_path = f'models/{self.checksum}'
# ACT
with self.assertRaises(TypeError): # dont care to mock even more...
datastore.get_model(
object_name=self.checksum,
)
# ASSERT
datastore._client.get_object.assert_called_with(
bucket_name=self.env_var_map['MINIO_BUCKET_NAME'],
object_name=object_path,
)
shared/docstore/__init__.py
-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/docstore/src/__init__.py
-3
View File
@@ -1,3 +0,0 @@
"""Database utils module content."""
from .connect_mongodb import connect_mongodb
shared/docstore/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/docstore/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/docstore/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/docstore/src/classes/data_model.py
-67
View File
@@ -1,67 +0,0 @@
"""Definition of DataModel base class."""
import random
from pydantic import BaseModel, ValidationError
from torch import Tensor
class DataModel(BaseModel):
"""DataModel base class."""
@classmethod
def classname(cls) -> str:
"""Return classname."""
return cls.__name__
@classmethod
def list_fields(cls) -> list[str]:
"""List options that are stored as attributes."""
return list(cls.model_fields.keys())
@classmethod
def from_random(cls):
"""Instantiate with random numbers."""
kwargs = {field: random.random() for field in cls.list_fields()}
return cls(**kwargs)
@classmethod
def from_choice(cls, option: str):
"""Instantiate from choice."""
if option is None:
raise ValidationError()
assert isinstance(option, str), 'option is not a string'
allowed_options_list = cls.list_fields()
assert (
option in allowed_options_list
), f"{option} is not among allowed fields {allowed_options_list}"
kwargs = {field: 0 for field in cls.list_fields()}
kwargs[option] = 1
return cls(**kwargs)
@classmethod
def from_tensor(cls, tensor: Tensor):
"""Instantiate from list of values."""
assert tensor.size(dim=0) == 1, f'tensor batch larger than 1: {tensor}'
data_list = [float(t.item()) for t in tensor[0]]
kwargs = dict(zip(cls.list_fields(), data_list))
return cls(**kwargs)
def __repr__(self) -> str:
model_dict = self.model_dump()
model_repr_str = f'{self.classname()}('
model_repr_str += ', '.join(
[f'{field}={value:.3f}' for field, value in model_dict.items()],
)
model_repr_str += ')'
return model_repr_str
def highest_score_field(self) -> str:
"""Return name of field with highest score."""
model_dict = self.model_dump()
return max(model_dict, key=lambda k: model_dict[k])
def highest_score_value(self) -> float:
"""Return value of field with highest score."""
model_dict = self.model_dump()
return max(model_dict.values())
shared/docstore/src/classes/distance.py
-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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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/docstore/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.docstore.src.classes import ModelData
class VisualCommunication(BaseModel):
"""Visual communication model."""
name: str
object_name: str
annotation: ModelData | None = None
prediction: ModelData | None = None
model_config = ConfigDict(arbitrary_types_allowed=True)
@classmethod
def classname(cls) -> str:
"""Return classname."""
return cls.__name__
@classmethod
def upload_image_to_minio(
cls,
image: Image.Image,
minio_client: Minio,
) -> str:
"""Upload image to MinIO and return MD5 checksum of hashed image."""
assert isinstance(image, Image.Image)
assert isinstance(minio_client, Minio)
with Datastore() as ds:
object_name = ds.put_image(
image=image,
)
return object_name
@classmethod
def from_name_and_image(
cls,
name: str,
image: Image.Image,
minio_client: Minio,
) -> VisualCommunication:
"""Instantiate from filename and image that is automatically uploaded
to MinIO."""
assert isinstance(name, str)
assert isinstance(image, Image.Image)
assert isinstance(minio_client, Minio)
# upload file to minio
object_name = VisualCommunication.upload_image_to_minio(
image=image,
minio_client=minio_client,
)
return VisualCommunication(name=name, object_name=object_name)
@classmethod
def from_file(cls, path: Path, minio_client: Minio) -> VisualCommunication:
"""Instantiate from file."""
assert isinstance(path, Path)
assert isinstance(minio_client, Minio)
# determine name
name = path.stem
# open image
image = Image.open(path)
image.load()
# instantiate object
return VisualCommunication.from_name_and_image(
name=name,
image=image,
minio_client=minio_client,
)
@classmethod
def decode_image(cls, content: str) -> Image.Image:
"""Extract image from webencoded content."""
_, content_data = content.split(',')
return Image.open(BytesIO(b64decode(content_data)))
def get_image(self, minio_client: Minio) -> Image.Image:
"""Load image data from minio."""
assert isinstance(minio_client, Minio)
# get image from minio
with Datastore() as ds:
image = ds.get_image(
object_name=self.object_name,
)
return image
def save_to_mongo(self, collection: Collection) -> None:
"""Save self as document in MongoDB."""
res = collection.insert_one(
document=self.model_dump(),
)
assert res.acknowledged
def webencoded_image(self, minio_client: Minio) -> str:
"""Convert image to be displayed on webpage."""
assert isinstance(minio_client, Minio)
# get image from minio
image = self.get_image(minio_client)
# convert images to bytes string
buffer = BytesIO()
image.save(buffer, format='png')
img_enc = b64encode(buffer.getvalue()).decode('utf-8')
return f"data:image/png;base64, {img_enc}"
def generate_random_prediction(self, force: bool = False) -> None:
"""Generate random prediction values."""
if not force and self.prediction is not None:
logging.warning('set force=True to overwrite existing values.')
self.prediction = ModelData.from_random()
shared/docstore/src/classes/visual_syntax.py
-28
View File
@@ -1,28 +0,0 @@
"""Definition of VisualSyntaxData data model."""
from __future__ import annotations
from .data_model import DataModel
class VisualSyntaxData(DataModel):
"""VisualSyntaxData data model."""
non_transactional_action: float
non_transactional_reaction: float
unidirectional_transactional_action: float
unidirectional_transactional_reaction: float
bidirectional_transactional_action: float
bidirectional_transactional_reaction: float
conversion: float
speech_process: float
classification_overt_taxonomy: float
analytical_exhaustive: float
analytical_disarranged: float
analytical_temporal: float
analytical_distributed: float
analytical_topological: float
analytical_exploded: float
analytical_inclusive: float
symbolic_suggestive: float
symbolic_attributive: float
shared/docstore/src/connect_mongodb.py
-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/docstore/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/docstore/src/docstore_interface.py
-83
View File
@@ -1,83 +0,0 @@
"""Definition of docstore interface."""
from __future__ import annotations
from abc import ABC, abstractmethod
from shared.docstore.src.classes import ModelData, VisualCommunication
class DocstoreInterface(ABC):
"""Docstore interface class."""
@abstractmethod
def connect(
self,
) -> None:
pass
@abstractmethod
def close(
self,
) -> None:
pass
@abstractmethod
def __enter__(
self,
) -> DocstoreInterface:
pass
@abstractmethod
def __exit__(
self,
exc_type,
exc_val,
exc_tb,
) -> None:
pass
@abstractmethod
def count_documents(
self,
only_with_annotation: bool,
) -> int:
pass
@abstractmethod
def list_names(
self,
only_with_annotation: bool,
) -> list[str]:
pass
@abstractmethod
def upsert_visual_comminucations(
self,
visual_communication_list: list[VisualCommunication],
) -> None:
pass
@abstractmethod
def upsert_annotations(
self,
visual_communication_name: str,
annotations: ModelData,
) -> None:
pass
@abstractmethod
def upsert_prediction(
self,
visual_communication_name: str,
predictions: ModelData,
) -> None:
pass
@abstractmethod
def get_visual_communication(
self,
with_annotation: bool,
name: str | None = None,
) -> VisualCommunication:
pass
shared/docstore/src/docstore_mongo.py
-241
View File
@@ -1,241 +0,0 @@
"""Definition of docstore mongodb implementation."""
from __future__ import annotations
import logging
import os
from typing import Any
from pymongo import MongoClient
from pymongo.collection import Collection
from pymongo.database import Database
from shared.utils import check_env
from .classes import ModelData, VisualCommunication
from .docstore_interface import DocstoreInterface
from .exceptions import NoDocumentFoundException
class DocstoreMongo(DocstoreInterface):
"""Docstore interface."""
def __init__(self):
# ensure necessary env vars available
var_list = {
'MONGO_ENDPOINT',
'MONGO_DB',
'MONGO_COLLECTION',
}
check_env(var_list)
# prepare internal variables
self.client: MongoClient | None = None
self.db: Database | None = None
self.collection: Collection | None = None
def connect(self) -> None:
"""Connect to Mongo server."""
# prepare arguments
mongo_endpoint = str(os.getenv('MONGO_ENDPOINT'))
mongo_database = str(os.getenv('MONGO_DB'))
mongo_collection = str(os.getenv('MONGO_COLLECTION'))
# connect client
client: MongoClient = MongoClient(mongo_endpoint)
database = client[mongo_database]
collection = database[mongo_collection]
# set unique index on 'name'
collection.create_index(keys='name', unique=True)
# persist state
self._client = client
self._database = database
self._collection = collection
def close(self) -> None:
"""Close connection to Mongo server."""
self._client.close()
self._client = None # type: ignore
self._database = None # type: ignore
self._collection = None # type: ignore
def __enter__(self) -> DocstoreMongo:
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
if any(
(
exc_type is not None,
exc_val is not None,
exc_tb is not None,
),
):
logging.error('error while exiting context')
self.close()
def count_documents(
self,
only_with_annotation: bool = False,
):
"""Get the total number of Visual Communication documents matching the
filter."""
assert isinstance(only_with_annotation, bool)
# build query
query = {}
if only_with_annotation:
query['annotation'] = {'$ne': None}
# execute query
num_docs = self._collection.count_documents(filter=query)
return num_docs
def list_names(
self,
only_with_annotation: bool,
) -> list[str]:
"""List names of Visual Communication documents that match the
filters."""
assert isinstance(only_with_annotation, bool)
assert isinstance(self._collection, Collection)
# build query
query = {}
if only_with_annotation:
query['annotation'] = {'$ne': None}
# execute query
res_list = self._collection.find(
filter=query,
projection={
'_id': False, # don't get id
'name': True, # include document name
},
)
# extract info
name_list = [elem['name'] for elem in res_list]
return name_list
def upsert_visual_comminucations(
self,
visual_communication_list: list[VisualCommunication],
) -> None:
"""Upsert Visual Communication document in the database."""
assert isinstance(visual_communication_list, list)
assert all(
isinstance(vis_com, VisualCommunication)
for vis_com in visual_communication_list
)
assert isinstance(self._collection, Collection)
# convert to dict
doc_list = [vis_com.model_dump() for vis_com in visual_communication_list]
# insert documents
response = self._collection.insert_many(documents=doc_list)
# check response
if not response.acknowledged:
raise OSError('failed inserting documents')
def upsert_annotations(
self,
visual_communication_name: str,
annotations: ModelData,
) -> None:
"""Upsert annotation for the document with matching name."""
assert isinstance(visual_communication_name, str)
assert len(visual_communication_name) > 0
assert isinstance(annotations, ModelData)
assert isinstance(self._collection, Collection)
# convert to dict
doc = annotations.model_dump()
# build query
query = {
'name': visual_communication_name,
}
update = {
'$set': {
'annotation': doc,
},
}
# execute query
res = self._collection.update_one(
filter=query,
update=update,
upsert=True,
)
# check response
if not res.acknowledged:
raise OSError(
f'failed upserting annotations for {visual_communication_name}',
)
def upsert_prediction(
self,
visual_communication_name: str,
predictions: ModelData,
) -> None:
"""Upsert prediction for the document with matching name."""
assert isinstance(visual_communication_name, str)
assert len(visual_communication_name) > 0
assert isinstance(annotations, ModelData)
assert isinstance(self._collection, Collection)
# convert to dict
doc = predictions.model_dump()
# build query
query = {
'name': visual_communication_name,
}
update = {
'$set': {
'prediction': doc,
},
}
# execute query
res = self._collection.update_one(
filter=query,
update=update,
upsert=True,
)
# check response
if not res.acknowledged:
raise OSError(
f'failed upserting predictions for {visual_communication_name}',
)
def get_visual_communication(
self,
with_annotation: bool = False,
name: str | None = None,
) -> VisualCommunication:
"""Get a random Visual Communication document that matches annotation
filter.
If name is not specified, a random document matching filter is
returned.
"""
assert isinstance(with_annotation, bool)
if name is not None:
assert isinstance(name, str)
assert len(name) > 0
# build query
query: dict[str, Any] = {}
if with_annotation:
query['annotation'] = {'$ne': None}
else:
query['annotation'] = {'$eq': None}
if name is not None:
query['name'] = {'$eq': name}
# execute query
res_list = self._collection.aggregate(
pipeline=[
{
'$match': query, # find using filters
},
{
'$sample': {
'size': 1, # get one random
},
},
],
)
doc_list = list(res_list)
# check result
if len(doc_list) == 0:
raise NoDocumentFoundException()
# convert
vis_com = VisualCommunication.model_validate(doc_list[0])
return vis_com
shared/docstore/src/exceptions/__init__.py
-1
View File
@@ -1 +0,0 @@
from .no_document_found import NoDocumentFoundException
shared/docstore/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/docstore/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/docstore/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.docstore.src.classes import VisualCommunication
from shared.docstore.src.exceptions import NoDocumentFoundException
def get_visual_communication(
collection: Collection,
with_annotation: bool = False,
) -> VisualCommunication:
"""Get a random visual communication from the database."""
query = {}
if with_annotation:
query['annotation'] = {'$ne': None}
else:
query['annotation'] = {'$eq': None}
data = collection.aggregate(
pipeline=[
{
'$match': query, # find using filters
},
{
'$sample': {
'size': 1, # get one random
},
},
],
)
data_list = list(data) # read data from cursor object
if len(data_list) == 0:
raise NoDocumentFoundException()
vis_com = VisualCommunication.model_validate(data_list[0])
logging.debug('finished')
return vis_com
shared/docstore/src/list_names.py
-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/docstore/src/upsert_annotation.py
-30
View File
@@ -1,30 +0,0 @@
from __future__ import annotations
import logging
from pymongo.collection import Collection
from shared.docstore.src.classes import ModelData
def upsert_annotation(
collection: Collection,
vis_com_name: str,
annotations: ModelData,
) -> None:
"""Upserts annotation data in the database."""
query = {
'name': vis_com_name,
}
update = {
'$set': {
'annotation': annotations.model_dump(),
},
}
res = collection.update_one(
filter=query,
update=update,
upsert=True,
)
logging.info('upserted document: %s', res)
logging.info('finished')
shared/docstore/src/upsert_prediction.py
-30
View File
@@ -1,30 +0,0 @@
from __future__ import annotations
import logging
from pymongo.collection import Collection
from shared.docstore.src.classes import ModelData
def upsert_prediction(
collection: Collection,
vis_com_name: str,
predictions: ModelData,
) -> None:
"""Upsert prediction data in the database."""
query = {
'name': vis_com_name,
}
update = {
'$set': {
'prediction': predictions.model_dump(),
},
}
res = collection.update_one(
filter=query,
update=update,
upsert=True,
)
logging.debug('upserted document: %s', res)
logging.info('finished')
shared/docstore/src/upsert_visual_communication.py
-19
View File
@@ -1,19 +0,0 @@
from __future__ import annotations
from pymongo.collection import Collection
from shared.docstore.src.classes import VisualCommunication
def upsert_visual_communication(
collection: Collection,
visual_communication_list: list[VisualCommunication],
) -> bool:
"""Upsert VisualCommunication object in the database.
Returns bool stating success.
"""
response = collection.insert_many(
[vis_com.model_dump() for vis_com in visual_communication_list],
)
return response.acknowledged
shared/docstore/tests/unit/docstore_mongo_test.py
-45
View File
@@ -1,45 +0,0 @@
"""Definition of unittests for Docstore MongoDB implementation."""
import os
from unittest import TestCase
from unittest.mock import MagicMock
from pymongo import MongoClient
from pymongo.collection import Collection
from pymongo.database import Database
from shared.docstore.src.docstore_mongo import DocstoreMongo
class TestDocstoreMongoImplementation(TestCase):
def setUp(self):
# define relevant env vars
self.env_var_map = {
'MONGO_ENDPOINT': '192.168.1.2:27017',
'MONGO_DB': 'visual_critical_discourse_analysis',
'MONGO_COLLECTION': 'test-collection',
}
# set env vars
for key, val in self.env_var_map.items():
os.environ[key] = val
# set other variables
self.mongo_client_mock = MagicMock(MongoClient)
self.mongo_database_mock = MagicMock(Database)
self.mongo_collection_mock = MagicMock(Collection)
def tearDown(self):
# clear env vars
for key in self.env_var_map:
_ = os.environ.pop(key, default=None)
# reset reuseable mocks
self.mongo_client_mock.reset_mock()
self.mongo_database_mock.reset_mock()
self.mongo_collection_mock.reset_mock()
def test_instantiation_should_fail_when_env_not_set(self):
# ensure env not set
self.tearDown()
# run test
with self.assertRaises(OSError):
_ = DocstoreMongo()
shared/repositories/tests/integration/conftest.py
+10 -4
View File
@@ -32,14 +32,17 @@ from shared.repositories.src.implementations import (
random.seed(13) random.seed(13)
# define test environment variables # define test environment variables
necessary_env_vars = {
'MINIO_ENDPOINT',
'MINIO_ACCESS_KEY',
'MINIO_SECRET_KEY',
'MONGO_ENDPOINT',
}
env_var_map = { env_var_map = {
'MINIO_ENDPOINT': '192.168.1.2:9000', 'MINIO_ENDPOINT': 'localhost:9000',
'MINIO_ACCESS_KEY': '1rMQTezbALeJyOgAbYYE',
'MINIO_SECRET_KEY': 'O76GM756ybKYLFokZFJneaCoYwpaIbAlOdB7mcc0',
'MINIO_BUCKET_NAME': 'test-bucket', 'MINIO_BUCKET_NAME': 'test-bucket',
'MINIO_OBJECT_NAME': 'test-object', 'MINIO_OBJECT_NAME': 'test-object',
'MINIO_IMAGE_NAME': 'test-image', 'MINIO_IMAGE_NAME': 'test-image',
'MONGO_ENDPOINT': 'mongodb://192.168.1.2:27017/',
'MONGO_DB': 'test-db', 'MONGO_DB': 'test-db',
'MONGO_COLLECTION': 'test-collection', 'MONGO_COLLECTION': 'test-collection',
} }
@@ -53,6 +56,9 @@ def setup_env(
# load in optional local test environment variables # load in optional local test environment variables
test_env_path = Path(__file__).parent.parent.parent.parent.parent / 'test.env' test_env_path = Path(__file__).parent.parent.parent.parent.parent / 'test.env'
load_dotenv(test_env_path) 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 testing env vars # set testing env vars
for key, val in env_var_map.items(): for key, val in env_var_map.items():
# set env var # set env var