[SYSTEMDS-#] Add contrastive learning operator for flexible modality pairing

This patch introduces a new operator to Scuro for building contrastive
learning pipelines with greater flexibility in handling input modalities.

Previously, contrastive pairs had to be structurally aligned in a
preprocessing step before being used in Scuro. This limited the ability
to work with independently transformed or dynamically generated
modalities.

The new operator constructs contrastive pairs via a Cartesian product of
modalities and optionally extends them with additional modalities that
are already aligned. The resulting combinations are evaluated using a
user-defined function to determine whether a pair represents a positive
or negative sample. Based on this evaluation, the operator outputs both
the assigned label and the corresponding modality pair.

This design enables dynamic label generation and supports scenarios where
modalities are windowed, reshuffled, or transformed differently. It also
allows flexible fusion of modalities after contrastive pairing, improving
the expressiveness of contrastive learning workflows.

Limitations: The Cartesian product can introduce significant
computational overhead for large modality sets, which may require further
optimization.

Author: b-enedict

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

@@ -62,19 +62,17 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
         self.file_sanity_check(file)
         if not self.load_data_from_file:
             import numpy as np
-
-            self.metadata[file] = self.modality_type.create_metadata(
-                1000, np.array([0])
-            )
+            audio = np.array([0])
+            sr = 1000
         else:
             audio, sr = librosa.load(file, dtype=self._data_type)
 
             if self.normalize:
                 audio = librosa.util.normalize(audio)
 
-            self.metadata[file] = self.modality_type.create_metadata(sr, audio)
+        self.metadata.append(self.modality_type.create_metadata(sr, audio))
 
-            self.data.append(audio)
+        self.data.append(audio)
 
     def get_stats(self, source_path: str):
         sampling_rate = 0

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

@@ -44,9 +44,7 @@ def __init__(
         (otherwise please provide your own Dataloader that knows about the file name convention)
         """
         self.data = []
-        self.metadata = (
-            {}
-        )  # TODO: check what the index should be for storing the metadata (file_name, counter, ...)
+        self.metadata = []
         self.source_path = source_path
         self.indices = indices
         self.modality_type = modality_type
@@ -87,7 +85,7 @@ def data_type(self, data_type):
     def reset(self):
         self._next_chunk = 0
         self.data = []
-        self.metadata = {}
+        self.metadata = []
 
     def load(self):
         """
@@ -134,6 +132,7 @@ def _load_next_chunk(self):
         Loads the next chunk of data
         """
         self.data = []
+        # TODO: Handle metadata correctly
         next_chunk_indices = self.indices[
             self._next_chunk
             * self._chunk_size : (self._next_chunk + 1)

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

@@ -71,9 +71,9 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
 
         image = image.astype(np.uint8, copy=False)
 
-        self.metadata[file] = self.modality_type.create_metadata(
+        self.metadata.append(self.modality_type.create_metadata(
             width, height, channels
-        )
+        ))
 
         self.data.append(image)
 

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

@@ -69,7 +69,7 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
 
                 text = " ".join(text) if isinstance(text, list) else text
                 self.data.append(text)
-                self.metadata[idx] = self.modality_type.create_metadata(len(text), text)
+                self.metadata.append(self.modality_type.create_metadata(len(text), text) | json_file[idx])
 
     def get_stats(self, source_path: str):
         self.file_sanity_check(source_path)

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

@@ -63,8 +63,8 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
 
             image = image.astype(np.uint8, copy=False)
 
-            self.metadata[f"{file}_{i}"] = self.modality_type.create_metadata(
+            self.metadata.append(self.modality_type.create_metadata(
                 width, height, channels
-            )
+            ))
 
             self.data.append(image)

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

@@ -56,9 +56,9 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
                 if self.prefix:
                     line = re.sub(self.prefix, "", line)
                 line = line.replace("\n", "")
-                self.metadata[file] = self.modality_type.create_metadata(
+                self.metadata.append(self.modality_type.create_metadata(
                     len(line.split()), line
-                )
+                ))
                 self.data.append(line)
 
     def get_stats(self, source_path: str):

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

@@ -81,15 +81,16 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
             data = self._normalize_signals(data)
 
         if file:
-            self.metadata[index] = self.modality_type.create_metadata(
+            self.metadata.append(self.modality_type.create_metadata(
                 self.signal_names, data, self.sampling_rate
-            )
+            ))
+            self.data.append(data)
         else:
             for i, index in enumerate(self.indices):
-                self.metadata[str(index)] = self.modality_type.create_metadata(
+                self.metadata.append(self.modality_type.create_metadata(
                     self.signal_names, data[i], self.sampling_rate
-                )
-        self.data.append(data)
+                ))
+                self.data.append(data[i])
 
     def _normalize_signals(self, data: np.ndarray) -> np.ndarray:
         if data.ndim == 1:

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

@@ -56,6 +56,6 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
             md["timestamp_end"] = seg.end
             md["text"] = seg.text
 
-            self.metadata[f"{file}_{i}"] = md
+            self.metadata.append(md)
 
             self.data.append(seg.text)

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

@@ -87,9 +87,9 @@ def extract(self, file: str, index: Optional[Union[str, List[str]]] = None):
         height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
         num_channels = 3
 
-        self.metadata[file] = self.modality_type.create_metadata(
+        self.metadata.append(self.modality_type.create_metadata(
             self.fps, length, width, height, num_channels
-        )
+        ))
 
         frames = []
         idx = 0

src/main/python/systemds/scuro/modality/joined.py

@@ -77,7 +77,7 @@ def execute(self, starting_idx=0):
             )
 
         for i in range(start, end):
-            idx_1 = list(self.left_modality.metadata.values())[i + starting_idx][
+            idx_1 = self.left_modality.metadata[i + starting_idx][
                 self.condition.leftField
             ]
             if (
@@ -90,7 +90,7 @@ def execute(self, starting_idx=0):
             if self.chunk_left:
                 i = i + starting_idx
 
-            idx_2 = list(self.right_modality.metadata.values())[i][
+            idx_2 = self.right_modality.metadata[i][
                 self.condition.rightField
             ]
             self.joined_right.data.append([])
@@ -246,11 +246,11 @@ def _apply_representation_chunked(
                 self.joined_right, representation
             )
             new_right.data.extend(right_transformed.data)
-            new_right.metadata.update(right_transformed.metadata)
+            new_right.metadata.extend(right_transformed.metadata)
 
             left_transformed = self._apply_representation(left_modality, representation)
             new_left.data.extend(left_transformed.data)
-            new_left.metadata.update(left_transformed.metadata)
+            new_left.metadata.extend(left_transformed.metadata)
 
         new_left.update_metadata()
         new_right.update_metadata()