moved webui and updated poetry packages

shared/database/classes/__init__.py

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

shared/database/classes/dataset.py

@@ -0,0 +1,71 @@
+"""Definition of database Dataset class."""
+from __future__ import annotations
+
+import logging
+import random
+from datetime import datetime
+from datetime import UTC
+
+from pydantic import BaseModel
+from pydantic import Field
+from pymongo.collection import Collection
+
+
+class Dataset(BaseModel):
+    """Database Dataset model."""
+    create_time: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    train_names: list[str]
+    test_names: list[str]
+    validation_names: list[str]
+
+    @classmethod
+    def fraction_map(cls) -> dict[str, float]:
+        """Dict with train, test and validation fractions."""
+        # define map
+        split_map = {
+            'train': 0.7,
+            'test': 0.2,
+            'validation': 0.1,
+        }
+        # sanity check
+        assert sum(split_map.values()) == 1.0
+        return split_map
+
+    @classmethod
+    def new_from_name_list(
+        cls,
+        name_list: list[str],
+    ) -> Dataset:
+        """Generate new dataset from list of filenames."""
+        # calculate split fractions
+        fraction_map = cls.fraction_map()
+        num_total = len(name_list)
+        num_validation = round(num_total * fraction_map['validation'])
+        num_test = round(num_total * fraction_map['test'])
+        # split data
+        validation_name_list = random.choices(name_list, k=num_validation)
+        name_list = [
+            name for name in name_list if name not in validation_name_list
+        ]
+        test_name_list = random.choices(name_list, k=num_test)
+        train_name_list = [
+            name for name in name_list if name not in test_name_list
+        ]
+        # instantiate object
+        dataset = Dataset(
+            train_names=train_name_list,
+            test_names=test_name_list,
+            validation_names=validation_name_list,
+        )
+        logging.debug('finished')
+        return dataset
+
+    def save(
+        self,
+        collection: Collection,
+    ) -> None:
+        """Save dataset to database."""
+        res = collection.insert_one(
+            document=self.model_dump(),
+        )
+        logging.debug('inserted document: %s', res)

shared/database/classes/exceptions.py

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

shared/database/classes/visual_communication.py

@@ -0,0 +1,85 @@
+"""Definition of VisualCommunication model."""
+from __future__ import annotations
+
+import logging
+from base64 import b64decode
+from base64 import b64encode
+from io import BytesIO
+from pathlib import Path
+
+from PIL import Image
+from pydantic import BaseModel
+from pydantic import field_serializer
+from pydantic import field_validator
+
+from shared.dto import ModelData
+
+
+class VisualCommunication(BaseModel):
+    """Visual communication model."""
+
+    name: str
+    image: Image.Image
+    annotation: ModelData | None = None
+    prediction: ModelData | None = None
+
+    class Config:
+        """BaseModel configuration."""
+        arbitrary_types_allowed = True
+
+    @classmethod
+    def classname(cls) -> str:
+        """Return classname."""
+        return cls.__name__
+
+    @classmethod
+    def from_file(cls, path: Path) -> VisualCommunication:
+        """Instantiate from file."""
+        name = path.stem
+        image = Image.open(path)
+        image.load()
+        return VisualCommunication(name=name, image=image)
+
+    @classmethod
+    def decode_image(cls, content: str) -> Image.Image:
+        """Extract image from webencoded content."""
+        _, content_data = content.split(',')
+        return Image.open(BytesIO(b64decode(content_data)))
+
+    @field_serializer('image')
+    @classmethod
+    def serialize_image(cls, image: Image.Image) -> bytes:   # type: ignore
+        """Convert image to bytes for storage in database."""
+        buffer = BytesIO()
+        image.save(buffer, format='JPEG')
+        return buffer.getvalue()
+
+    @field_validator('image', mode='before')
+    @classmethod
+    def convert_to_image(
+        cls,
+        image: Image.Image | BytesIO | bytes,
+    ) -> Image.Image:
+        """Convert bytes input from database into image."""
+        if isinstance(image, bytes):
+            image = BytesIO(image)
+        if isinstance(image, BytesIO):
+            image = Image.open(image)
+        return image
+
+    def __repr__(self) -> str:
+        return f"{self.classname()}(name='{self.name}')"
+
+    def webencoded_image(self) -> str:
+        """Convert image to be displayed on webpage."""
+        # convert images to bytes string
+        buffer = BytesIO()
+        self.image.save(buffer, format='png')
+        img_enc = b64encode(buffer.getvalue()).decode('utf-8')
+        return f"data:image/png;base64, {img_enc}"
+
+    def generate_random_prediction(self, force: bool = False) -> None:
+        """Generate random prediction values."""
+        if not force and self.prediction is not None:
+            logging.warning('set force=True to overwrite existing values.')
+        self.prediction = ModelData.from_random()