Skip to content

[SYSTEMDS-3971] Improve Scuro test suite #2614

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shieru1214 wants to merge 3 commits into
base: main
Choose a base branch
from
+155 −187
Open

[SYSTEMDS-3971] Improve Scuro test suite #2614

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7d62874
[MINOR] Merge the per-modality duplicates in the Scuro optimizer tests
shieru1214 Sep 12, 2026
4c29f1d
[MINOR] Merge redundant subtests in the Scuro window operation tests
shieru1214 Sep 12, 2026
929d9ff
[MINOR] Turn the Scuro fusion order tests into a property table
shieru1214 Sep 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 63 additions & 60 deletions src/main/python/tests/scuro/test_fusion_orders.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,77 +19,80 @@
#
# -------------------------------------------------------------

import os
import shutil
import unittest
import numpy as np

from systemds.scuro import Concatenation, RowMax, Hadamard
from systemds.scuro.modality.unimodal_modality import UnimodalModality
from systemds.scuro.representations.bert import Bert
from systemds.scuro.representations.mel_spectrogram import MelSpectrogram
from systemds.scuro.representations.average import Average
from tests.scuro.data_generator import ModalityRandomDataGenerator
from systemds.scuro.modality.type import ModalityType


class TestFusionOrders(unittest.TestCase):
"""
Tests the order properties of the fusion operators: commutativity, whether
the order of a pairwise chain matters, and whether a pairwise chain gives
the same result as the n-ary call.
"""

# (operator, chain_order_independent, chain_equals_nary)
#
# Commutativity is not in the table. Every Fusion operator has a
# "commutative" attribute. It is False on the base class and a subclass
# can override it. The test compares the measured result with that
# attribute, so an operator whose attribute does not match its
# implementation fails here.
#
# Combining a pair is never the same as combining all three. That is
# checked for every operator, so it is not in the table either.
FUSION_PROPERTIES = [
(Average, True, False),
(Concatenation, False, True),
(RowMax, True, True),
(Hadamard, True, True),
]

@classmethod
def setUpClass(cls):
cls.num_instances = 40
# These properties do not depend on the input shape, so the data can
# be small.
cls.num_instances = 4
cls.num_features = 8
cls.data_generator = ModalityRandomDataGenerator()
cls.r_1 = cls.data_generator.create1DModality(40, 100, ModalityType.AUDIO)
cls.r_2 = cls.data_generator.create1DModality(40, 100, ModalityType.TEXT)
cls.r_3 = cls.data_generator.create1DModality(40, 100, ModalityType.TEXT)

def test_fusion_order_avg(self):
r_1_r_2 = self.r_1.combine(self.r_2, Average())
r_2_r_1 = self.r_2.combine(self.r_1, Average())
r_1_r_2_r_3 = r_1_r_2.combine(self.r_3, Average())
r_2_r_1_r_3 = r_2_r_1.combine(self.r_3, Average())

r1_r2_r3 = self.r_1.combine([self.r_2, self.r_3], Average())

self.assertTrue(np.array_equal(r_1_r_2.data, r_2_r_1.data))
self.assertTrue(np.array_equal(r_1_r_2_r_3.data, r_2_r_1_r_3.data))
self.assertFalse(np.array_equal(r_1_r_2_r_3.data, r1_r2_r3.data))
self.assertFalse(np.array_equal(r_1_r_2.data, r1_r2_r3.data))

def test_fusion_order_concat(self):
r_1_r_2 = self.r_1.combine(self.r_2, Concatenation())
r_2_r_1 = self.r_2.combine(self.r_1, Concatenation())
r_1_r_2_r_3 = r_1_r_2.combine(self.r_3, Concatenation())
r_2_r_1_r_3 = r_2_r_1.combine(self.r_3, Concatenation())

r1_r2_r3 = self.r_1.combine([self.r_2, self.r_3], Concatenation())

self.assertFalse(np.array_equal(r_1_r_2.data, r_2_r_1.data))
self.assertFalse(np.array_equal(r_1_r_2_r_3.data, r_2_r_1_r_3.data))
self.assertFalse(np.array_equal(r_2_r_1.data, r1_r2_r3.data))
self.assertFalse(np.array_equal(r_1_r_2.data, r1_r2_r3.data))

def test_fusion_order_max(self):
r_1_r_2 = self.r_1.combine(self.r_2, RowMax())
r_2_r_1 = self.r_2.combine(self.r_1, RowMax())
r_1_r_2_r_3 = r_1_r_2.combine(self.r_3, RowMax())
r_2_r_1_r_3 = r_2_r_1.combine(self.r_3, RowMax())

r1_r2_r3 = self.r_1.combine([self.r_2, self.r_3], RowMax())

self.assertTrue(np.array_equal(r_1_r_2.data, r_2_r_1.data))
self.assertTrue(np.array_equal(r_1_r_2_r_3.data, r_2_r_1_r_3.data))
self.assertTrue(np.array_equal(r_1_r_2_r_3.data, r1_r2_r3.data))
self.assertFalse(np.array_equal(r_1_r_2.data, r1_r2_r3.data))

def test_fusion_order_hadamard(self):
r_1_r_2 = self.r_1.combine(self.r_2, Hadamard())
r_2_r_1 = self.r_2.combine(self.r_1, Hadamard())
r_1_r_2_r_3 = r_1_r_2.combine(self.r_3, Hadamard())
r_2_r_1_r_3 = r_2_r_1.combine(self.r_3, Hadamard())

r1_r2_r3 = self.r_1.combine([self.r_2, self.r_3], Hadamard())

self.assertTrue(np.array_equal(r_1_r_2.data, r_2_r_1.data))
self.assertTrue(np.array_equal(r_1_r_2_r_3.data, r_2_r_1_r_3.data))
self.assertTrue(np.array_equal(r_1_r_2_r_3.data, r1_r2_r3.data))
self.assertFalse(np.array_equal(r_1_r_2.data, r1_r2_r3.data))
def setUp(self):
self.r_1 = self.data_generator.create1DModality(
self.num_instances, self.num_features, ModalityType.AUDIO
)
self.r_2 = self.data_generator.create1DModality(
self.num_instances, self.num_features, ModalityType.TEXT
)
self.r_3 = self.data_generator.create1DModality(
self.num_instances, self.num_features, ModalityType.TEXT
)

@staticmethod
def _equal(left, right):
return np.array_equal(np.asarray(left.data), np.asarray(right.data))

def test_fusion_order_properties(self):
for (
fusion_operator,
chain_order_independent,
chain_equals_nary,
) in self.FUSION_PROPERTIES:
with self.subTest(fusion=fusion_operator.__name__):
r_1_r_2 = self.r_1.combine(self.r_2, fusion_operator())
r_2_r_1 = self.r_2.combine(self.r_1, fusion_operator())
r_1_r_2_r_3 = r_1_r_2.combine(self.r_3, fusion_operator())
r_2_r_1_r_3 = r_2_r_1.combine(self.r_3, fusion_operator())
r1_r2_r3 = self.r_1.combine([self.r_2, self.r_3], fusion_operator())

self.assertEqual(
self._equal(r_1_r_2, r_2_r_1), fusion_operator().commutative
)
self.assertEqual(
self._equal(r_1_r_2_r_3, r_2_r_1_r_3), chain_order_independent
)
self.assertEqual(self._equal(r_1_r_2_r_3, r1_r2_r3), chain_equals_nary)
self.assertFalse(self._equal(r_1_r_2, r1_r2_r3))
104 changes: 43 additions & 61 deletions src/main/python/tests/scuro/test_unimodal_optimizer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,16 +127,51 @@ def setUpClass(cls):
TestTask("UnimodalRepresentationTask1", "Test1", cls.num_instances),
]

def test_unimodal_optimizer_for_text_modality(self):
text_data, text_md = ModalityRandomDataGenerator().create_text_data(
self.num_instances, 10
)
text = UnimodalModality(
TestDataLoader(
self.indices, None, ModalityType.TEXT, text_data, str, text_md
# (label, [(modality type, generator keyword arguments)]). A set with two
# entries is passed to the optimizer as one multi-modality search, because
# optimize_unimodal_representation_for_modality loops over the list.
MODALITY_SETS = [
("text", [(ModalityType.TEXT, {})]),
("image", [(ModalityType.IMAGE, {})]),
("audio", [(ModalityType.AUDIO, {})]),
("video", [(ModalityType.VIDEO, {"num_frames": 10})]),
(
"text+image",
[(ModalityType.TEXT, {"num_sentences": 1}), (ModalityType.IMAGE, {})],
),
]

def _create_modality(self, modality_type, num_sentences=10, num_frames=1):
generator = ModalityRandomDataGenerator()
if modality_type is ModalityType.TEXT:
data, metadata = generator.create_text_data(
self.num_instances, num_sentences
)
data_type = str
elif modality_type is ModalityType.AUDIO:
data, metadata = generator.create_audio_data(self.num_instances, 3000)
data_type = np.float32
else:
# IMAGE and VIDEO use the same generator. The number of frames is
# the difference between them.
data, metadata = generator.create_visual_modality(
self.num_instances, num_frames, 10, 10
)
data_type = np.float32

return UnimodalModality(
TestDataLoader(self.indices, None, modality_type, data, data_type, metadata)
)
self.optimize_unimodal_representation_for_modality([text])

def test_unimodal_optimizer_per_modality_set(self):
for label, modality_specs in self.MODALITY_SETS:
with self.subTest(modalities=label):
self.optimize_unimodal_representation_for_modality(
[
self._create_modality(modality_type, **kwargs)
for modality_type, kwargs in modality_specs
]
)

def test_robust_results_ignore_non_finite_scores(self):
modality = SimpleNamespace(modality_id="modality")
Expand Down Expand Up @@ -194,59 +229,6 @@ def test_bow_and_tfidf_require_dimensionality_reduction_before_task(self):
task_input = dag.get_node_by_id(task_node.inputs[0])
self.assertIs(task_input.operation, MLPAveraging)

def test_unimodal_optimizer_for_image_modality(self):
image_data, image_md = ModalityRandomDataGenerator().create_visual_modality(
self.num_instances, 1, 10, 10
)
image = UnimodalModality(
TestDataLoader(
self.indices, None, ModalityType.IMAGE, image_data, np.float32, image_md
)
)
self.optimize_unimodal_representation_for_modality([image])

def test_unimodal_optimizer_for_multiple_modalities(self):
image_data, image_md = ModalityRandomDataGenerator().create_visual_modality(
self.num_instances, 1, 10, 10
)
image = UnimodalModality(
TestDataLoader(
self.indices, None, ModalityType.IMAGE, image_data, np.float32, image_md
)
)
text_data, text_md = ModalityRandomDataGenerator().create_text_data(
self.num_instances
)
text = UnimodalModality(
TestDataLoader(
self.indices, None, ModalityType.TEXT, text_data, str, text_md
)
)
self.optimize_unimodal_representation_for_modality([text, image])

def test_unimodal_optimizer_for_audio_modality(self):
audio_data, audio_md = ModalityRandomDataGenerator().create_audio_data(
self.num_instances, 3000
)
audio = UnimodalModality(
TestDataLoader(
self.indices, None, ModalityType.AUDIO, audio_data, np.float32, audio_md
)
)

self.optimize_unimodal_representation_for_modality([audio])

def test_unimodal_optimizer_for_video_modality(self):
video_data, video_md = ModalityRandomDataGenerator().create_visual_modality(
self.num_instances, 10, 10, 10
)
video = UnimodalModality(
TestDataLoader(
self.indices, None, ModalityType.VIDEO, video_data, np.float32, video_md
)
)
self.optimize_unimodal_representation_for_modality([video])

# ------------------------------------------------------------------
# Every registered representation, run through the optimizer
# ------------------------------------------------------------------
Expand Down
12 changes: 1 addition & 11 deletions src/main/python/tests/scuro/test_unimodal_representations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,17 +205,7 @@ def test_audio_representations(self):
RMSE(),
Pitch(),
]
audio_data, audio_md = ModalityRandomDataGenerator().create_audio_data(
self.num_instances, 200
)

audio = UnimodalModality(
TestDataLoader(
self.indices, None, ModalityType.AUDIO, audio_data, np.float32, audio_md
)
)

audio.extract_raw_data()
audio = self._create_audio_modality(signal_length=200)
original_data = copy.deepcopy(audio.data)

for representation in audio_representations:
Expand Down
103 changes: 48 additions & 55 deletions src/main/python/tests/scuro/test_window_operations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,67 +111,60 @@ def test_dynamic_window(self):
for i in range(0, self.num_instances):
assert len(aggregated_window.data[i]) == num_windows

def test_window_aggregation_on_audio_representations(self):
def test_window_aggregation_on_1d_modalities(self):
# create1DModality returns the same shape and dtype for all three
# modality types. window_aggregation looks at the data layout and not
# at the modality type, so the result should be the same for all of
# them.
window_size = 10
self.run_window_aggregation_for_modality(ModalityType.AUDIO, window_size)

def test_window_operations_on_video_representations(self):
window_size = 10
self.run_window_aggregation_for_modality(ModalityType.VIDEO, window_size)

def test_window_operations_on_text_representations(self):
window_size = 10

self.run_window_aggregation_for_modality(ModalityType.TEXT, window_size)

def run_window_aggregation_for_modality(self, modality_type, window_size):
r = self.data_generator.create1DModality(self.num_instances, 200, modality_type)
for aggregation in self.aggregations:
windowed_modality = r.window_aggregation(window_size, aggregation)

self.verify_window_operation(aggregation, r, windowed_modality, window_size)

def test_window_aggregation_on_3d_modality(self):
data, _ = self.data_generator.create_3d_modality(
self.num_instances, (100, 8, 8)
)
embedding_modality = TransformedModality(
self.data_generator, "test_transformation"
)
embedding_modality.data = data
embedding_modality.stats = RepresentationStats(self.num_instances, (100, 8, 8))
num_windows = 10

for window_operator in [
StaticWindow(num_windows=num_windows),
DynamicWindow(num_windows=num_windows),
WindowAggregation(window_size=10),
for modality_type in [
ModalityType.AUDIO,
ModalityType.VIDEO,
ModalityType.TEXT,
]:
stats = window_operator.get_output_stats(embedding_modality.stats)
assert stats.num_instances == self.num_instances
assert stats.output_shape == (num_windows, 8, 8)

windowed_modality = embedding_modality.context(window_operator)
r = self.data_generator.create1DModality(
self.num_instances, 200, modality_type
)
for aggregation in self.aggregations:
with self.subTest(modality=modality_type.name, aggregation=aggregation):
windowed_modality = r.window_aggregation(window_size, aggregation)
self.verify_window_operation(
aggregation, r, windowed_modality, window_size
)

def test_window_aggregation_on_2d_modality(self):
data, _ = self.data_generator.create_2d_modality(self.num_instances, (100, 8))
embedding_modality = TransformedModality(
self.data_generator, "test_transformation"
)
embedding_modality.data = data
embedding_modality.stats = RepresentationStats(self.num_instances, (100, 8))
def test_window_aggregation_on_nd_modality(self):
# Window aggregation only changes the first (time) axis and keeps the
# feature axes as they are. The expected shape is therefore
# (num_windows,) + dims[1:] for any number of dimensions.
num_windows = 10

for window_operator in [
StaticWindow(num_windows=num_windows),
DynamicWindow(num_windows=num_windows),
WindowAggregation(window_size=10),
]:
stats = window_operator.get_output_stats(embedding_modality.stats)
assert stats.num_instances == self.num_instances
assert stats.output_shape == (num_windows, 8)

windowed_modality = embedding_modality.context(window_operator)
for dims in [(100, 8, 8), (100, 8)]:
if len(dims) == 3:
data, _ = self.data_generator.create_3d_modality(
self.num_instances, dims
)
else:
data, _ = self.data_generator.create_2d_modality(
self.num_instances, dims
)
embedding_modality = TransformedModality(
self.data_generator, "test_transformation"
)
embedding_modality.data = data
embedding_modality.stats = RepresentationStats(self.num_instances, dims)

for window_operator in [
StaticWindow(num_windows=num_windows),
DynamicWindow(num_windows=num_windows),
WindowAggregation(window_size=10),
]:
with self.subTest(dims=dims, operator=type(window_operator).__name__):
stats = window_operator.get_output_stats(embedding_modality.stats)
self.assertEqual(stats.num_instances, self.num_instances)
self.assertEqual(stats.output_shape, (num_windows,) + dims[1:])

embedding_modality.context(window_operator)

def _timeseries_modality(self, signal_length=100):
return self.data_generator.create1DModality(
Expand Down