[SYSTEMDS-3835] Improve efficiency of Representations in Scuro

.github/workflows/python.yml

@@ -167,7 +167,7 @@ jobs:
         pip install --upgrade pip wheel setuptools
         # Use CUDA 12.1 wheels to avoid slow/source builds
         pip install --extra-index-url https://download.pytorch.org/whl/cu121 \
-          torch==2.4.1 torchvision==0.19.1
+          torch==2.4.1 torchvision==0.19.1 torchaudio==2.4.1
         pip install \
           transformers \
           opencv-python \
@@ -179,7 +179,10 @@ jobs:
           fvcore \
           scikit-optimize \
           flair \
-          optuna 
+          optuna \
+          openface-test \
+          imagebind \
+         "pytorchvideo @ git+https://github.com/facebookresearch/pytorchvideo.git@eb04d1b"
         kill $KA 
         cd src/main/python
         python -m unittest discover -s tests/scuro -p 'test_*.py' -v

src/main/python/systemds/scuro/__init__.py

@@ -24,6 +24,7 @@
 from systemds.scuro.dataloader.video_loader import VideoLoader
 from systemds.scuro.dataloader.text_loader import TextLoader
 from systemds.scuro.dataloader.json_loader import JSONLoader
+from systemds.scuro.dataloader.tabular_loader import TabularLoader
 from systemds.scuro.representations.representation import Representation
 from systemds.scuro.representations.aggregate import Aggregation
 from systemds.scuro.representations.aggregated_representation import (
@@ -71,9 +72,23 @@
 from systemds.scuro.representations.representation_dataloader import JSON
 from systemds.scuro.representations.representation_dataloader import Pickle
 from systemds.scuro.representations.resnet import ResNet
+from systemds.scuro.representations.openface import OpenFace
+
+try:
+    from systemds.scuro.representations.image_bind import ImageBind
+except ImportError as _imagebind_import_error:  # pragma: no cover
+    import warnings as _warnings
+
+    _warnings.warn(
+        f"ImageBind representation unavailable, it is excluded from the search "
+        f"({_imagebind_import_error})."
+    )
+    ImageBind = None
+
 from systemds.scuro.representations.spectrogram import Spectrogram
 from systemds.scuro.representations.sum import Sum
 from systemds.scuro.representations.swin_video_transformer import SwinVideoTransformer
+from systemds.scuro.representations.tabular_features import TabularFeatures
 from systemds.scuro.representations.tfidf import TfIdf
 from systemds.scuro.representations.unimodal import UnimodalRepresentation
 from systemds.scuro.representations.wav2vec import Wav2Vec
@@ -82,6 +97,10 @@
     DynamicWindow,
     StaticWindow,
 )
+from systemds.scuro.representations.physiological_window import (
+    AdaptiveWindow,
+    PhysiologicalEventWindow,
+)
 from systemds.scuro.representations.word2vec import W2V
 from systemds.scuro.representations.x3d import X3D
 from systemds.scuro.representations.color_histogram import ColorHistogram
@@ -124,6 +143,28 @@
     MLPLearnedDimReduction,
 )
 
+
+from systemds.scuro.representations.physiological_representations import (
+    SDNN,
+    RMSSD,
+    pNN,
+    RRPerMinute,
+    HRVBandPower,
+    HRVVLF,
+    HRVLF,
+    HRVHF,
+    HRVLFHF,
+    PoincareSD1,
+    PoincareSD2,
+    # SampleEntropy,
+    # DFAAlpha,
+    SCLSlope,
+    SCLDynamicRange,
+    SCRPeaksPerMinute,
+    SCRAverageAmplitude,
+    SCRAverageDuration,
+)
+
 __all__ = [
     "BaseLoader",
     "ImageLoader",
@@ -156,6 +197,8 @@
     "JSON",
     "Pickle",
     "ResNet",
+    "OpenFace",
+    "ImageBind",
     "Spectrogram",
     "Sum",
     "BoW",
@@ -187,6 +230,8 @@
     "AttentionFusion",
     "DynamicWindow",
     "StaticWindow",
+    "AdaptiveWindow",
+    "PhysiologicalEventWindow",
     "Min",
     "Max",
     "Mean",
@@ -211,4 +256,22 @@
     "MLPAveraging",
     "MLPLearnedDimReduction",
     "DimensionalityReduction",
+    "MultimodalGAPymooOptimizer",
+    "SDNN",
+    "RMSSD",
+    "pNN",
+    "RRPerMinute",
+    "HRVBandPower",
+    "HRVVLF",
+    "HRVLF",
+    "HRVHF",
+    "HRVLFHF",
+    "PoincareSD1",
+    "PoincareSD2",
+    # "DFAAlpha",
+    "SCLSlope",
+    "SCLDynamicRange",
+    "SCRPeaksPerMinute",
+    "SCRAverageAmplitude",
+    "SCRAverageDuration",
 ]

src/main/python/systemds/scuro/dataloader/base_loader.py

@@ -20,12 +20,33 @@
 # -------------------------------------------------------------
 import os
 from abc import ABC, abstractmethod
-from typing import Iterator, List, Optional, Tuple, Union
+from collections.abc import Sequence
+from typing import Callable, Iterator, List, Optional, Tuple, Union
 import math
+from numbers import Integral
 
 import numpy as np
 
 
+class LazyFileSequence(Sequence):
+    """List-like file references decoded only when a sample is requested."""
+
+    def __init__(self, file_names: List[str], decoder: Callable[[str], object]):
+        self.file_names = tuple(file_names)
+        self.decoder = decoder
+
+    def __getitem__(self, index):
+        if isinstance(index, slice):
+            return [self[i] for i in range(*index.indices(len(self)))]
+        return self.decoder(self.file_names[index])
+
+    def __len__(self):
+        return len(self.file_names)
+
+    def subset(self, indices):
+        return type(self)([self.file_names[i] for i in indices], self.decoder)
+
+
 class BaseLoader(ABC):
     def __init__(
         self,
@@ -54,17 +75,36 @@ def __init__(
         self._data_type = data_type
         self._ext = ext
         self.stats = None
-        if chunk_size:
-            self.chunk_size = chunk_size
+        self.chunk_size = chunk_size
 
     @property
     def chunk_size(self):
         return self._chunk_size
 
     @chunk_size.setter
     def chunk_size(self, value):
-        self._chunk_size = value
-        self._num_chunks = int(math.ceil(len(self.indices) / self._chunk_size))
+        if value is None:
+            self._chunk_size = None
+            self._num_chunks = 1
+        else:
+            if isinstance(value, bool) or not isinstance(value, Integral) or value <= 0:
+                raise ValueError("chunk_size must be None or a positive integer")
+            self._chunk_size = int(value)
+            self._num_chunks = int(math.ceil(len(self.indices) / self._chunk_size))
+
+        stats = getattr(self, "stats", None)
+        if stats is not None and hasattr(stats, "num_instances"):
+            stats.num_instances = (
+                len(self.indices)
+                if self._chunk_size is None
+                else min(len(self.indices), self._chunk_size)
+            )
+        if stats is not None and hasattr(stats, "num_total_instances"):
+            stats.num_total_instances = len(self.indices)
+
+    @property
+    def is_chunked(self):
+        return self._chunk_size is not None
 
     @property
     def num_chunks(self):
@@ -91,7 +131,7 @@ def load(self):
         """
         Takes care of loading the raw data either chunk wise (if chunk size is defined) or all at once
         """
-        if self._chunk_size:
+        if self.is_chunked:
             return self._load_next_chunk()
 
         return self._load(self.indices)
@@ -102,8 +142,8 @@ def iter_loaded_chunks(
         if reset:
             self.reset()
 
-        if not self._chunk_size:
-            data, metadata = self._load(self.indices)
+        if not self.is_chunked:
+            data, metadata = self.load()
             yield data, metadata, self.indices
             return
 
@@ -115,24 +155,21 @@ def iter_loaded_chunks(
             yield data, metadata, chunk_indices
 
     def update_chunk_sizes(self, other):
-        if not self._chunk_size and not other.chunk_size:
+        sizes = [
+            size for size in (self.chunk_size, other.chunk_size) if size is not None
+        ]
+        if not sizes:
             return
-
-        if (
-            self._chunk_size
-            and not other.chunk_size
-            or self._chunk_size < other.chunk_size
-        ):
-            other.chunk_size = self.chunk_size
-        else:
-            self.chunk_size = other.chunk_size
+        shared_size = min(sizes)
+        self.chunk_size = shared_size
+        other.chunk_size = shared_size
 
     def _load_next_chunk(self):
         """
         Loads the next chunk of data
         """
         self.data = []
-        # TODO: Handle metadata correctly
+        self.metadata = []
         next_chunk_indices = self.indices[
             self._next_chunk
             * self._chunk_size : (self._next_chunk + 1)
@@ -161,7 +198,7 @@ def get_file_names(self, indices=None):
             if self._ext is None:
                 _, self._ext = os.path.splitext(os.listdir(self.source_path)[0])
             for index in self.indices if indices is None else indices:
-                file_names.append(self.source_path + index + self._ext)
+                file_names.append(os.path.join(self.source_path, index + self._ext))
             return file_names
         else:
             return self.source_path

src/main/python/systemds/scuro/dataloader/image_loader.py

@@ -24,7 +24,7 @@
 
 import numpy as np
 
-from systemds.scuro.dataloader.base_loader import BaseLoader
+from systemds.scuro.dataloader.base_loader import BaseLoader, LazyFileSequence
 import cv2
 from systemds.scuro.modality.type import ModalityType
 
@@ -55,29 +55,43 @@ def __init__(
             source_path, indices, data_type, chunk_size, ModalityType.IMAGE, ext
         )
         self.load_data_from_file = load
+        self._all_metadata = []
         self.stats = self.get_stats(source_path)
 
-    def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
-        self.file_sanity_check(file)
+    def load(self):
+        if self.chunk_size:
+            return super().load()
 
+        self.data = LazyFileSequence(
+            self.get_file_names(self.indices), self._decode_file
+        )
+        self.metadata = self._all_metadata.copy()
+        return self.data, self.metadata
+
+    def _decode_file(self, file: str):
+        self.file_sanity_check(file)
         image = cv2.imread(file, cv2.IMREAD_COLOR)
-        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+        if image is None:
+            raise ValueError(f"Could not read image at path: {file}")
+        return cv2.cvtColor(image, cv2.COLOR_BGR2RGB).astype(np.uint8, copy=False)
+
+    def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
+        image = self._decode_file(file)
 
         if image.ndim == 2:
             height, width = image.shape
             channels = 1
         else:
             height, width, channels = image.shape
 
-        image = image.astype(np.uint8, copy=False)
-
         self.metadata.append(
             self.modality_type.create_metadata(width, height, channels)
         )
 
         self.data.append(image)
 
     def get_stats(self, source_path: str):
+        self._all_metadata = []
         max_width = 0
         max_height = 0
         max_channels = 0
@@ -87,18 +101,15 @@ def get_stats(self, source_path: str):
         average_channels = 0
         for file in self.indices:
             path = os.path.join(source_path, f"{file}{self._ext}")
-            # if self.chunk_size is None:
-            #     self.extract(path)
-            #     md = self.metadata[path]
-            #     max_width = max(max_width, md["width"])
-            #     max_height = max(max_height, md["height"])
-            #     max_channels = max(max_channels, md["num_channels"])
-            #     num_instances += 1
-            # else:
             self.file_sanity_check(path)
             image = cv2.imread(path, cv2.IMREAD_COLOR)
+            if image is None:
+                raise ValueError(f"Could not read image at path: {path}")
             image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
             height, width, channels = image.shape
+            self._all_metadata.append(
+                self.modality_type.create_metadata(width, height, channels)
+            )
             max_width = max(max_width, width)
             max_height = max(max_height, height)
             max_channels = max(max_channels, channels)
@@ -119,3 +130,15 @@ def get_stats(self, source_path: str):
             average_height,
             average_channels,
         )
+
+    def estimate_peak_memory_bytes(self) -> dict:
+        n = self.chunk_size if self.chunk_size is not None else 1
+        per_instance = (
+            self.stats.average_width
+            * self.stats.average_height
+            * self.stats.average_channels
+        )
+        return {
+            "cpu_peak_bytes": int(n * per_instance * np.dtype(np.uint8).itemsize),
+            "gpu_peak_bytes": 0,
+        }