diff --git a/src/main/python/systemds/scuro/modality/joined.py b/src/main/python/systemds/scuro/modality/joined.py index 9b4bdc4791b..1551c5fe3ed 100644 --- a/src/main/python/systemds/scuro/modality/joined.py +++ b/src/main/python/systemds/scuro/modality/joined.py @@ -102,7 +102,7 @@ def execute(self, starting_idx=0): self.joined_right.data[i - starting_idx].append([]) right = np.array([]) if self.condition.join_type == "<": - while c < len(idx_2) - 1 and idx_2[c] < nextIdx[j]: + while c < len(idx_2) and idx_2[c] < nextIdx[j]: if right.size == 0: right = self.right_modality.data[i][c] if right.ndim == 1: @@ -123,15 +123,21 @@ def execute(self, starting_idx=0): ) c = c + 1 else: - while c < len(idx_2) - 1 and idx_2[c] <= idx_1[j]: + matches = [] + while c < len(idx_2) and idx_2[c] <= idx_1[j]: if idx_2[c] == idx_1[j]: - right.append(self.right_modality.data[i][c]) + match = self.right_modality.data[i][c] + if match.ndim == 1: + match = match[np.newaxis, :] + matches.append(match) c = c + 1 + if matches: + right = np.concatenate(matches, axis=0) if ( len(right) == 0 ): # Audio and video length sometimes do not match so we add the average all audio samples for this specific frame - right = np.mean(self.right_modality.data[i][c - 1 : c], axis=0) + right = np.mean(self.right_modality.data[i], axis=0) if right.ndim == 1: right = right[ np.newaxis, : diff --git a/src/main/python/tests/scuro/test_data_loaders.py b/src/main/python/tests/scuro/test_data_loaders.py index 0d5ace01a47..f0ee32ef23c 100644 --- a/src/main/python/tests/scuro/test_data_loaders.py +++ b/src/main/python/tests/scuro/test_data_loaders.py @@ -49,19 +49,32 @@ def setUpClass(cls): def tearDownClass(cls): shutil.rmtree(cls.test_file_path) - def test_audio_loader_loads_all_instances(self): - loader = AudioLoader( - self.data_generator.get_modality_path(ModalityType.AUDIO), - self.data_generator.indices, - ) - data, metadata = loader.load() - - self.assertEqual(len(data), self.num_instances) - self.assertEqual(len(metadata), self.num_instances) - - for arr in data: - self.assertIsInstance(arr, np.ndarray) - self.assertEqual(arr.ndim, 1) + # Loading is the same contract for every loader -- one array plus one + # metadata entry per instance, at the dimensionality that modality has -- + # so the three cases only differed in the loader class and the expected + # ndim. The stats tests below stay separate: each stats class exposes + # different fields, so there is no shared assertion to parameterise. + LOADERS_AND_DIMENSIONS = [ + (AudioLoader, ModalityType.AUDIO, 1), + (VideoLoader, ModalityType.VIDEO, 4), + (ImageLoader, ModalityType.IMAGE, 3), + ] + + def test_loaders_load_all_instances(self): + for loader_class, modality_type, expected_ndim in self.LOADERS_AND_DIMENSIONS: + with self.subTest(loader=loader_class.__name__): + loader = loader_class( + self.data_generator.get_modality_path(modality_type), + self.data_generator.indices, + ) + data, metadata = loader.load() + + self.assertEqual(len(data), self.num_instances) + self.assertEqual(len(metadata), self.num_instances) + + for arr in data: + self.assertIsInstance(arr, np.ndarray) + self.assertEqual(arr.ndim, expected_ndim) def test_audio_loader_stats(self): loader = AudioLoader( @@ -76,20 +89,6 @@ def test_audio_loader_stats(self): self.assertEqual(stats.max_length, 44100) self.assertAlmostEqual(stats.avg_length, (44100 * 2) / 2.0) - def test_video_loader_loads_all_instances(self): - loader = VideoLoader( - self.data_generator.get_modality_path(ModalityType.VIDEO), - self.data_generator.indices, - ) - data, metadata = loader.load() - - self.assertEqual(len(data), self.num_instances) - self.assertEqual(len(metadata), self.num_instances) - - for arr in data: - self.assertIsInstance(arr, np.ndarray) - self.assertEqual(arr.ndim, 4) - def test_video_loader_stats(self): loader = VideoLoader( self.data_generator.get_modality_path(ModalityType.VIDEO), @@ -129,20 +128,6 @@ def test_text_loader_stats(self): self.assertEqual(stats.max_length, 7) self.assertAlmostEqual(stats.avg_length, (7 + 7) / 2.0) - def test_image_loader_loads_all_instances(self): - loader = ImageLoader( - self.data_generator.get_modality_path(ModalityType.IMAGE), - self.data_generator.indices, - ) - data, metadata = loader.load() - - self.assertEqual(len(data), self.num_instances) - self.assertEqual(len(metadata), self.num_instances) - - for arr in data: - self.assertIsInstance(arr, np.ndarray) - self.assertEqual(arr.ndim, 3) - def test_image_loader_stats(self): loader = ImageLoader( self.data_generator.get_modality_path(ModalityType.IMAGE), diff --git a/src/main/python/tests/scuro/test_fusion_orders.py b/src/main/python/tests/scuro/test_fusion_orders.py index 22d64bcc0bf..789d7d8c6fa 100644 --- a/src/main/python/tests/scuro/test_fusion_orders.py +++ b/src/main/python/tests/scuro/test_fusion_orders.py @@ -19,77 +19,79 @@ # # ------------------------------------------------------------- -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): + """ + The interesting content is the table below rather than the call sequence: + which operator is commutative, whose result depends on the order of a + pairwise chain, and where a pairwise chain equals the n-ary form. Written as + a table those differences are visible at a glance and a new operator is one + line. + """ + + # (operator, chain_order_independent, chain_equals_nary) + # Commutativity is not listed: every Fusion operator declares a + # "commutative" attribute, so the test compares the measured behaviour + # against that declaration instead of against a second copy of it. A new + # operator whose declaration contradicts its implementation fails here + # without anyone having to remember to extend this table. + # Combining a pair is never the same as combining all three, so that case + # is asserted for every operator instead of being listed here. + FUSION_PROPERTIES = [ + (Average, True, False), + (Concatenation, False, True), + (RowMax, True, True), + (Hadamard, True, True), + ] + @classmethod def setUpClass(cls): - cls.num_instances = 40 + # The properties under test hold for any input shape. + 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)) diff --git a/src/main/python/tests/scuro/test_hp_tuner.py b/src/main/python/tests/scuro/test_hp_tuner.py index 8f7aa0b1284..ab5f5ce905d 100644 --- a/src/main/python/tests/scuro/test_hp_tuner.py +++ b/src/main/python/tests/scuro/test_hp_tuner.py @@ -74,16 +74,31 @@ def setUpClass(cls): TestTask("UnimodalRepresentationTask2", "TestSVM2", cls.num_instances), ] - def test_hp_tuner_for_text_modality(self): - text_data, text_md = ModalityRandomDataGenerator().create_text_data( - self.num_instances - ) - text = UnimodalModality( - TestDataLoader( - self.indices, None, ModalityType.TEXT, text_data, str, text_md + def _create_modality(self, modality_type): + if modality_type is ModalityType.TEXT: + data, metadata = ModalityRandomDataGenerator().create_text_data( + self.num_instances + ) + data_type = str + else: + data, metadata = ModalityRandomDataGenerator().create_visual_modality( + self.num_instances, 1 ) + data_type = np.float32 + + return UnimodalModality( + TestDataLoader(self.indices, None, modality_type, data, data_type, metadata) ) - self.run_hp_for_modality([text]) + + def test_hp_tuner_per_modality(self): + # The text and image cases ran the same tuner over the same registry and + # asserted the same thing -- every assertion lives in + # run_hp_for_modality, so the two tests differed only in how the + # modality is built. Building it is a factory and the modality type is a + # subTest dimension; both cases still run and are reported separately. + for modality_type in [ModalityType.TEXT, ModalityType.IMAGE]: + with self.subTest(modality=modality_type.name): + self.run_hp_for_modality([self._create_modality(modality_type)]) # TODO: Add once the final multimodal optimizer is implemented # def test_multimodal_hp_tuning(self): @@ -109,17 +124,6 @@ def test_hp_tuner_for_text_modality(self): # [audio, text], multimodal=True, tune_unimodal_representations=False # ) - def test_hp_tuner_for_image_modality(self): - image_data, image_md = ModalityRandomDataGenerator().create_visual_modality( - self.num_instances, 1 - ) - image = UnimodalModality( - TestDataLoader( - self.indices, None, ModalityType.IMAGE, image_data, np.float32, image_md - ) - ) - self.run_hp_for_modality([image]) - def run_hp_for_modality( self, modalities, multimodal=False, tune_unimodal_representations=False ): diff --git a/src/main/python/tests/scuro/test_multimodal_join.py b/src/main/python/tests/scuro/test_multimodal_join.py index 4a53129db33..3b55b5b5200 100644 --- a/src/main/python/tests/scuro/test_multimodal_join.py +++ b/src/main/python/tests/scuro/test_multimodal_join.py @@ -20,72 +20,246 @@ # TODO: Test edge cases: unequal number of audio-video timestamps (should still work and add the average over all audio/video samples) +import copy import unittest import numpy as np -import copy -from systemds.scuro.modality.joined import JoinCondition + +from systemds.scuro.modality.joined import JoinCondition, JoinedModality +from systemds.scuro.modality.modality import Modality +from systemds.scuro.modality.transformed import TransformedModality +from systemds.scuro.modality.type import ModalityType from systemds.scuro.modality.unimodal_modality import UnimodalModality from systemds.scuro.representations.mel_spectrogram import MelSpectrogram from systemds.scuro.representations.resnet import ResNet -from tests.scuro.data_generator import TestDataLoader, ModalityRandomDataGenerator -from systemds.scuro.modality.type import ModalityType +from systemds.scuro.representations.unimodal import UnimodalRepresentation +from tests.scuro.data_generator import ModalityRandomDataGenerator, TestDataLoader -class TestMultimodalJoin(unittest.TestCase): - test_file_path = None - mods = None - text = None - audio = None - video = None - data_generator = None - num_instances = 0 - indizes = [] +class SpyRepresentation(UnimodalRepresentation): + """ + Cheap stand-in for a real representation. - @classmethod - def setUpClass(cls): - cls.num_instances = 4 - cls.indices = np.array(range(cls.num_instances)) - cls.audio_data, cls.audio_md = ModalityRandomDataGenerator().create_audio_data( - cls.num_instances, 500 + Whether a representation is a neural network or a summary statistic is + irrelevant to the join: it only has to turn every element into one row. A + deterministic stand-in keeps the join behaviour comparable across + configurations and keeps these tests off the torch path. + """ + + def __init__(self): + super().__init__("Spy", ModalityType.EMBEDDING) + self.call_count = 0 + self.seen_element_counts = [] + + def transform(self, modality, aggregation=None): + self.call_count += 1 + self.seen_element_counts.append([len(instance) for instance in modality.data]) + transformed = TransformedModality(modality, self, self.output_modality_type) + transformed.data = [ + np.stack([self._embed(element) for element in instance]) + for instance in modality.data + ] + return transformed + + @staticmethod + def _embed(element): + values = np.asarray(element, dtype=np.float32).reshape(-1) + return np.array( + [values.mean(), values.min(), values.max(), float(values.size)], + dtype=np.float32, ) - cls.video_data, cls.video_md = ( - ModalityRandomDataGenerator().create_visual_modality(cls.num_instances, 60) + +def _modality_with_timestamps(modality_type, instances, timestamps, metadata_args): + """ + Build a modality whose timestamps are written by hand, one entry per list. + + Modality.data is a property: assigning it recomputes the timestamps from the + frequency stored in the metadata, so the timestamps have to be overwritten + afterwards. The assertion at the end guards that order - swap the two and + the generated timestamps would silently win. + """ + modality = Modality( + modality_type, + 0, + [modality_type.create_metadata(*args) for args in metadata_args], + np.float32, + ) + modality.data = list(instances) + for metadata, stamps in zip(modality.metadata, timestamps): + metadata["timestamp"] = np.asarray(stamps) + + for metadata, stamps in zip(modality.metadata, timestamps): + assert np.array_equal(metadata["timestamp"], np.asarray(stamps)), ( + "timestamps were recomputed - Modality.data has to be assigned " + "before the timestamps are written, not after" ) + return modality - def test_video_audio_join(self): - self._execute_va_join() - def test_chunked_video_audio_join(self): - self._execute_va_join(2) +def _video_modality(frame_timestamps): + """frame_timestamps: one list of frame timestamps per instance.""" + return _modality_with_timestamps( + ModalityType.VIDEO, + [np.zeros((len(stamps), 2), dtype=np.float32) for stamps in frame_timestamps], + frame_timestamps, + [(30, len(stamps), 2, 2, 3) for stamps in frame_timestamps], + ) - def test_video_chunked_audio_join(self): - self._execute_va_join(None, 2) - def test_chunked_video_chunked_audio_join(self): - self._execute_va_join(2, 2) +def _audio_modality(rows, row_timestamps): + """rows / row_timestamps: one entry per instance.""" + return _modality_with_timestamps( + ModalityType.AUDIO, + rows, + row_timestamps, + [(1, np.zeros(len(stamps), dtype=np.float32)) for stamps in row_timestamps], + ) - def test_audio_video_join(self): - # Audio has a much higher frequency than video, hence we would need to - # duplicate or interpolate frames to match them to the audio frequency - self._execute_av_join() - # TODO - # def test_chunked_audio_video_join(self): - # self._execute_av_join(2) +class TestJoinMapping(unittest.TestCase): + """ + Unit tests for the join mapping itself: fixed timestamps, no representation. + These are the tests that pin down which right hand rows end up under which + left hand frame. + """ + + # Two instances so that a mix-up between them shows: instance 0 carries the + # row values 0..5, instance 1 carries 100..102. + LEFT_TIMESTAMPS = [[0, 10, 20], [0, 5]] + RIGHT_TIMESTAMPS = [[0, 1, 5, 11, 12, 25], [0, 3, 7]] + RIGHT_VALUES = [[0, 1, 2, 3, 4, 5], [100, 101, 102]] + + def setUp(self): + self.right_rows = [ + np.array([[value, value] for value in values], dtype=np.float32) + for values in self.RIGHT_VALUES + ] + joined = JoinedModality( + ModalityType.VIDEO, + _video_modality(self.LEFT_TIMESTAMPS), + _audio_modality(self.right_rows, self.RIGHT_TIMESTAMPS), + JoinCondition("timestamp", "timestamp", "<"), + ) + joined.execute() + self.blocks = [ + [np.asarray(block) for block in instance] + for instance in joined.joined_right.data + ] + + def _rows(self, instance, *indices): + return self.right_rows[instance][list(indices)] + + def test_join_assigns_every_left_frame_a_block(self): + self.assertEqual(len(self.blocks), len(self.LEFT_TIMESTAMPS)) + for instance, frame_timestamps in enumerate(self.LEFT_TIMESTAMPS): + self.assertEqual(len(self.blocks[instance]), len(frame_timestamps)) + + def test_join_maps_right_samples_before_the_next_left_frame(self): + # instance 0: frames at t = 0, 10, 20 over rows at t = 0, 1, 5, 11, 12, 25 + np.testing.assert_array_equal(self.blocks[0][0], self._rows(0, 0, 1, 2)) + np.testing.assert_array_equal(self.blocks[0][1], self._rows(0, 3, 4)) + # the last frame covers everything from 20 on -> the row at t = 25. + # Regression: the final right row has to be reachable, the last frame + # must not fall back to a copy of the previous one. + np.testing.assert_array_equal(self.blocks[0][2], self._rows(0, 5)) + + def test_join_keeps_instances_apart(self): + # instance 1: frames at t = 0, 5 over rows at t = 0, 3, 7. The values are + # from the 100 range, so borrowing a row from instance 0 would show. + np.testing.assert_array_equal(self.blocks[1][0], self._rows(1, 0, 1)) + np.testing.assert_array_equal(self.blocks[1][1], self._rows(1, 2)) + + def test_join_without_matching_samples_falls_back_to_the_instance_average(self): + # A left frame that no right sample falls into gets the average over all + # right rows of that instance, as described by the TODO at the top of + # this file. Regression: an empty match must not produce NaN. + # every right row lies before the first left frame, so frame 1 matches + # nothing at all + right_rows = np.array([[2.0, 2.0], [4.0, 4.0]], dtype=np.float32) + joined = JoinedModality( + ModalityType.VIDEO, + _video_modality([[0, 100]]), + _audio_modality([right_rows], [[0, 1]]), + JoinCondition("timestamp", "timestamp", "<"), + ) + joined.execute() + blocks = [np.asarray(block) for block in joined.joined_right.data[0]] + + np.testing.assert_array_equal(blocks[0], right_rows) + np.testing.assert_array_equal(blocks[1], np.array([[3.0, 3.0]])) + self.assertFalse(np.isnan(blocks[1]).any()) + + def test_chunked_execution_offsets_into_the_right_modality(self): + # execute(starting_idx) is the index arithmetic chunked runs depend on: + # the left modality holds one chunk while the right one holds every + # instance, so the chunk has to be paired with the right instances that + # start at starting_idx. Off by one here pairs instance A's video with + # instance B's audio, silently and with the expected shapes. + right_rows = [ + np.array([[10 * instance, 10 * instance]], dtype=np.float32) + for instance in range(4) + ] + joined = JoinedModality( + ModalityType.VIDEO, + _video_modality([[0, 10], [0, 10]]), + _audio_modality(right_rows, [[0]] * 4), + JoinCondition("timestamp", "timestamp", "<"), + ) + joined.chunked_execution = True + joined.chunk_left = True + + joined.execute(starting_idx=2) + + # the chunk is instances 0 and 1 of the left modality, so it must pick up + # right instances 2 and 3, i.e. the rows carrying 20 and 30 + for chunk_position, right_instance in enumerate([2, 3]): + with self.subTest(chunk_position=chunk_position): + for block in joined.joined_right.data[chunk_position]: + np.testing.assert_array_equal( + np.asarray(block), right_rows[right_instance] + ) + + def test_equality_join_maps_rows_with_matching_timestamps(self): + # Covers the branch taken for join types other than "<", which had no + # test at all and could not run: it called .append() on a numpy array. + right_timestamps = [0, 10, 10, 20, 30, 40] + right_rows = np.array( + [[value, value] for value in range(len(right_timestamps))], + dtype=np.float32, + ) + joined = JoinedModality( + ModalityType.VIDEO, + _video_modality([[0, 10, 20]]), + _audio_modality([right_rows], [right_timestamps]), + JoinCondition("timestamp", "timestamp", "=="), + ) + joined.execute() + blocks = [np.asarray(block) for block in joined.joined_right.data[0]] + + # every left frame collects the right rows carrying the same timestamp + np.testing.assert_array_equal(blocks[0], right_rows[[0]]) + np.testing.assert_array_equal(blocks[1], right_rows[[1, 2]]) + np.testing.assert_array_equal(blocks[2], right_rows[[3]]) - # TODO - # def test_chunked_audio_chunked_video_join(self): - # self._execute_av_join(2, 2) - def _execute_va_join(self, l_chunk_size=None, r_chunk_size=None): - video, audio = self._prepare_data(l_chunk_size, r_chunk_size) - self._join(video, audio, 2) +class TestMultimodalJoin(unittest.TestCase): + """ + End to end joins over generated data. These check that the pipeline holds + together and that chunking does not change the result; the mapping itself + is covered by TestJoinMapping above. + """ - def _execute_av_join(self, l_chunk_size=None, r_chunk_size=None): - video, audio = self._prepare_data(l_chunk_size, r_chunk_size) - self._join(audio, video, 2) + @classmethod + def setUpClass(cls): + cls.num_instances = 4 + cls.indices = np.array(range(cls.num_instances)) + cls.audio_data, cls.audio_md = ModalityRandomDataGenerator().create_audio_data( + cls.num_instances, 500 + ) + cls.video_data, cls.video_md = ( + ModalityRandomDataGenerator().create_visual_modality(cls.num_instances, 60) + ) def _prepare_data(self, l_chunk_size=None, r_chunk_size=None): audio = UnimodalModality( @@ -104,32 +278,112 @@ def _prepare_data(self, l_chunk_size=None, r_chunk_size=None): l_chunk_size, ModalityType.VIDEO, copy.deepcopy(self.video_data), - np.uint8, + np.float32, copy.deepcopy(self.video_md), ) ) + return video, audio.apply_representation(MelSpectrogram()) - mel_audio = audio.apply_representation(MelSpectrogram()) - - return video, mel_audio - - def _join(self, left_modality, right_modality, window_size): - resnet_modality = ( + def _join(self, left_modality, right_modality, representation, window_size=2): + return ( left_modality.join( right_modality, JoinCondition("timestamp", "timestamp", "<") ) - .apply_representation(ResNet()) + .apply_representation(representation) .window_aggregation(window_size, "mean") .combine("concat") ) - assert resnet_modality.left_modality is not None - assert resnet_modality.right_modality is not None - assert len(resnet_modality.left_modality.data) == self.num_instances - assert len(resnet_modality.right_modality.data) == self.num_instances - assert resnet_modality.data is not None + def test_video_audio_join(self): + video, mel_audio = self._prepare_data() + joined = self._join(video, mel_audio, SpyRepresentation()) + + self.assertEqual(len(joined.left_modality.data), self.num_instances) + self.assertEqual(len(joined.right_modality.data), self.num_instances) + self.assertEqual(len(joined.data), self.num_instances) + + # TODO + # def test_chunked_audio_video_join(self): + # self._execute_av_join(2) + + # TODO + # def test_chunked_audio_chunked_video_join(self): + # self._execute_av_join(2, 2) + + def test_audio_video_join(self): + # Audio has a much higher frequency than video, hence we would need to + # duplicate or interpolate frames to match them to the audio frequency + video, mel_audio = self._prepare_data() + joined = self._join(mel_audio, video, SpyRepresentation()) + + self.assertEqual(len(joined.left_modality.data), self.num_instances) + self.assertEqual(len(joined.data), self.num_instances) + + def test_chunked_and_unchunked_joins_agree(self): + # Chunking is a memory setting, so it must not change the result. Every + # combination is compared against the unchunked run. + chunk_configurations = [(None, None), (2, None), (None, 2), (2, 2)] + results = {} + + for l_chunk_size, r_chunk_size in chunk_configurations: + video, mel_audio = self._prepare_data(l_chunk_size, r_chunk_size) + joined = self._join(video, mel_audio, SpyRepresentation()) + results[(l_chunk_size, r_chunk_size)] = [ + np.asarray(instance) for instance in joined.data + ] + + expected = results[(None, None)] + for configuration, actual in results.items(): + with self.subTest(chunk_sizes=configuration): + self.assertEqual(len(actual), len(expected)) + for instance, expected_instance in zip(actual, expected): + self.assertEqual(instance.shape, expected_instance.shape) + # Regression: an empty match must never produce NaN, so no + # equal_nan here. + self.assertFalse(np.isnan(instance).any()) + self.assertTrue(np.allclose(instance, expected_instance)) + + def test_join_applies_the_representation_to_both_sides(self): + video, mel_audio = self._prepare_data() + spy = SpyRepresentation() + + video.join( + mel_audio, JoinCondition("timestamp", "timestamp", "<") + ).apply_representation(spy) + + self.assertEqual(spy.call_count, 2) + # Both sides have to arrive with one entry per left hand frame: the + # video frames themselves, and the block of right hand rows the join + # assigned to each of those frames. + left_counts, right_counts = spy.seen_element_counts + self.assertEqual(left_counts, right_counts) + self.assertEqual(len(left_counts), self.num_instances) + + def test_video_audio_join_with_resnet(self): + # The one test that runs a real representation end to end, so that a + # change breaking the torch path is still caught. + video_data, video_md = ModalityRandomDataGenerator().create_visual_modality( + 2, 12 + ) + audio_data, audio_md = ModalityRandomDataGenerator().create_audio_data(2, 500) + indices = np.array(range(2)) + + audio = UnimodalModality( + TestDataLoader( + indices, None, ModalityType.AUDIO, audio_data, np.float32, audio_md + ) + ) + video = UnimodalModality( + TestDataLoader( + indices, None, ModalityType.VIDEO, video_data, np.float32, video_md + ) + ) + + joined = self._join( + video, audio.apply_representation(MelSpectrogram()), ResNet() + ) - return resnet_modality + self.assertEqual(len(joined.data), 2) if __name__ == "__main__": diff --git a/src/main/python/tests/scuro/test_scheduler.py b/src/main/python/tests/scuro/test_scheduler.py index e46d59c15f3..dad07029e04 100644 --- a/src/main/python/tests/scuro/test_scheduler.py +++ b/src/main/python/tests/scuro/test_scheduler.py @@ -177,6 +177,9 @@ def test_get_ready_nodes_second_level(self): scheduler.move_to_running(node) scheduler.complete_node(node) ready_nodes_2 = scheduler.get_runnable() + # Without this the loop below is skipped when the scheduler returns + # nothing and the test passes without checking anything. + self.assertEqual(len(ready_nodes_2), 1) for node in ready_nodes_2: self.assertGreaterEqual( get_node_from_dags(self.dags, node).parameters["dimensionality"], 1000 @@ -218,6 +221,12 @@ def test_finished_when_no_nodes_are_runnable(self): self.assertTrue(scheduler.success) def test_deadlock_when_no_nodes_are_runnable(self): + # The memory budget is too small for any node, so the scheduler has to + # stop without having run anything. is_finished() alone cannot show + # that: the loop below only exits once it returns True, so asserting it + # afterwards is a tautology. success is what separates this case from + # test_finished_when_no_nodes_are_runnable, which is otherwise the same + # test with a larger budget. scheduler = MemoryAwareNodeScheduler( self.dags, self.modalities, self.tasks, 1024 * 1024 * 3, 0 ) @@ -226,5 +235,5 @@ def test_deadlock_when_no_nodes_are_runnable(self): for node in ready_nodes.copy(): scheduler.move_to_running(node) scheduler.complete_node(node) - self.assertTrue(scheduler.is_finished()) self.assertFalse(scheduler.success) + self.assertEqual(len(scheduler.completed_nodes), 0) diff --git a/src/main/python/tests/scuro/test_text_context_operators.py b/src/main/python/tests/scuro/test_text_context_operators.py index e9bc7032a22..b9603809b8c 100644 --- a/src/main/python/tests/scuro/test_text_context_operators.py +++ b/src/main/python/tests/scuro/test_text_context_operators.py @@ -29,78 +29,112 @@ SentenceBoundarySplitIndices, OverlappingSplitIndices, ) -from tests.scuro.data_generator import ( - ModalityRandomDataGenerator, - TestDataLoader, - TestTask, -) +from tests.scuro.data_generator import TestDataLoader from systemds.scuro.modality.unimodal_modality import UnimodalModality from systemds.scuro.modality.type import ModalityType -from systemds.scuro.representations.bert import Bert class TestTextContextOperator(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.data_generator = ModalityRandomDataGenerator() - cls.data, cls.md = cls.data_generator.create_text_data(10, 50) - cls.text_modality = UnimodalModality( + """ + The input is fixed so that the exact chunk boundaries and character spans + can be written down. With randomly generated sentences the only assertable + properties are invariants ("a chunk has at most max_words", "consecutive + chunks share their first/last words"), and those pass for a large family of + wrong implementations. + """ + + # 3 sentences, 5 words each, 77 characters + THREE_SENTENCES = ( + "The cat reads the document. A dog writes the code. The bird studies the data." + ) + # 1 sentence, 5 words, 27 characters - stays below max_words for every case + ONE_SENTENCE = "The cat reads the document." + + SENTENCE_MAX_WORDS = 10 + SENTENCE_MIN_WORDS = 4 + # sentence 1 + 2 fill the 10 word budget, sentence 3 starts a new chunk + EXPECTED_SENTENCE_CHUNKS = [ + [ + "The cat reads the document. A dog writes the code.", + "The bird studies the data.", + ], + [ONE_SENTENCE], + ] + + OVERLAP_MAX_WORDS = 6 + OVERLAP = 0.5 # -> stride of 3 words, i.e. 3 words shared per chunk pair + EXPECTED_OVERLAPPING_CHUNKS = [ + [ + "The cat reads the document. A", + "the document. A dog writes the", + "dog writes the code. The bird", + "code. The bird studies the data.", + ], + [ONE_SENTENCE], + ] + + def setUp(self): + # Rebuilt for every test: the *Indices operators write "text_spans" + # into the modality metadata, so a class level modality would leak the + # spans of one test into the next one (test order is alphabetical). + self.texts = [self.THREE_SENTENCES, self.ONE_SENTENCE] + metadata = [ + ModalityType.TEXT.create_metadata(len(text), text) for text in self.texts + ] + self.text_modality = UnimodalModality( TestDataLoader( - [i for i in range(0, 10)], + list(range(len(self.texts))), None, ModalityType.TEXT, - cls.data, + list(self.texts), str, - cls.md, + metadata, ) ) - cls.text_modality.extract_raw_data() - cls.task = TestTask("TextContextTask", "Test1", 10) + self.text_modality.extract_raw_data() + + def _spans(self): + return [metadata["text_spans"] for metadata in self.text_modality.metadata] + + def _sliced_by_spans(self): + return [ + [text[start:end] for start, end in spans] + for text, spans in zip(self.text_modality.data, self._spans()) + ] def test_sentence_boundary_split(self): - sentence_boundary_split = SentenceBoundarySplit(10, min_words=4) - chunks = sentence_boundary_split.execute(self.text_modality) - for i in range(0, len(chunks)): - for chunk in chunks[i]: - assert len(chunk.split(" ")) <= 10 and ( - chunk[-1] == "." or chunk[-1] == "!" or chunk[-1] == "?" - ) + chunks = SentenceBoundarySplit( + self.SENTENCE_MAX_WORDS, min_words=self.SENTENCE_MIN_WORDS + ).execute(self.text_modality) + + self.assertEqual(chunks, self.EXPECTED_SENTENCE_CHUNKS) def test_overlapping_split(self): - overlapping_split = OverlappingSplit(40, 0.05) - chunks = overlapping_split.execute(self.text_modality) - for i in range(len(chunks)): - prev_chunk = "" - for j, chunk in enumerate(chunks[i]): - if j > 0: - prev_words = prev_chunk.split(" ") - curr_words = chunk.split(" ") - assert prev_words[-2:] == curr_words[:2] - prev_chunk = chunk - assert len(chunk.split(" ")) <= 40 + chunks = OverlappingSplit(self.OVERLAP_MAX_WORDS, self.OVERLAP).execute( + self.text_modality + ) + + self.assertEqual(chunks, self.EXPECTED_OVERLAPPING_CHUNKS) def test_sentence_boundary_split_indices(self): - sentence_boundary_split = SentenceBoundarySplitIndices(10, min_words=4) - sentence_boundary_split.execute(self.text_modality) - for instance, md in zip(self.text_modality.data, self.text_modality.metadata): - for chunk in md["text_spans"]: - text = instance[chunk[0] : chunk[1]].split(" ") - assert len(text) <= 10 and ( - text[-1][-1] == "." or text[-1][-1] == "!" or text[-1][-1] == "?" - ) + SentenceBoundarySplitIndices( + self.SENTENCE_MAX_WORDS, min_words=self.SENTENCE_MIN_WORDS + ).execute(self.text_modality) + + self.assertEqual(self._spans(), [[(0, 50), (51, 77)], [(0, 27)]]) + # the spans have to cut the original text into the same chunks the + # string returning variant produces + self.assertEqual(self._sliced_by_spans(), self.EXPECTED_SENTENCE_CHUNKS) def test_overlapping_split_indices(self): - overlapping_split = OverlappingSplitIndices(40, 0.1) - overlapping_split.execute(self.text_modality) - for instance, md in zip(self.text_modality.data, self.text_modality.metadata): - prev_chunk = (0, 0) - for j, chunk in enumerate(md["text_spans"]): - if j > 0: - prev_words = instance[prev_chunk[0] : prev_chunk[1]].split(" ") - curr_words = instance[chunk[0] : chunk[1]].split(" ") - assert prev_words[-4:] == curr_words[:4] - prev_chunk = chunk - assert len(instance[chunk[0] : chunk[1]].split(" ")) <= 40 + OverlappingSplitIndices(self.OVERLAP_MAX_WORDS, self.OVERLAP).execute( + self.text_modality + ) + + self.assertEqual( + self._spans(), [[(0, 29), (14, 44), (30, 59), (45, 77)], [(0, 27)]] + ) + self.assertEqual(self._sliced_by_spans(), self.EXPECTED_OVERLAPPING_CHUNKS) if __name__ == "__main__": diff --git a/src/main/python/tests/scuro/test_unimodal_optimizer.py b/src/main/python/tests/scuro/test_unimodal_optimizer.py index f27c721aa25..8a53fdab3b5 100644 --- a/src/main/python/tests/scuro/test_unimodal_optimizer.py +++ b/src/main/python/tests/scuro/test_unimodal_optimizer.py @@ -127,16 +127,57 @@ 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 + # Every case below runs the same optimizer over the same registry and + # asserts the same thing -- optimize_unimodal_representation_for_modality + # holds all the assertions, and the helper already loops over the modality + # list, so the multi-modality case is just a set with two entries. The five + # tests differed only in which modalities they built, so that is a subTest + # dimension and the building is a factory. + # + # The keyword arguments keep the inputs exactly as the individual tests had + # them: video used ten frames where image used one, and the multi-modality + # case built its text with the generator default of one sentence rather than + # the ten the standalone text case used. + 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: + 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") @@ -194,59 +235,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 # ------------------------------------------------------------------ diff --git a/src/main/python/tests/scuro/test_unimodal_representations.py b/src/main/python/tests/scuro/test_unimodal_representations.py index 27e09d48711..3816b189f3f 100644 --- a/src/main/python/tests/scuro/test_unimodal_representations.py +++ b/src/main/python/tests/scuro/test_unimodal_representations.py @@ -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: diff --git a/src/main/python/tests/scuro/test_window_operations.py b/src/main/python/tests/scuro/test_window_operations.py index c6a258fb465..c253ae30020 100644 --- a/src/main/python/tests/scuro/test_window_operations.py +++ b/src/main/python/tests/scuro/test_window_operations.py @@ -75,103 +75,91 @@ def setUpClass(cls): cls.data_generator = ModalityRandomDataGenerator() cls.aggregations = ["mean", "sum", "max", "min"] - def test_static_window(self): + def test_fixed_window_count_operators(self): + # StaticWindow and DynamicWindow place their window boundaries + # differently, but the only thing asserted here is the contract they + # share: exactly num_windows segments per instance, whatever the input + # length. The two operators were spelled out as separate tests whose + # bodies differed in one word. num_windows = 5 data, md = self.data_generator.create_visual_modality(self.num_instances, 10) - modality = UnimodalModality( - TestDataLoader( - [i for i in range(0, self.num_instances)], - None, - ModalityType.VIDEO, - data, - np.float32, - md, - ) - ) - aggregated_window = modality.context(StaticWindow(num_windows=num_windows)) - - for i in range(0, self.num_instances): - assert len(aggregated_window.data[i]) == num_windows - - def test_dynamic_window(self): - num_windows = 5 - data, md = self.data_generator.create_visual_modality(self.num_instances, 10) - modality = UnimodalModality( - TestDataLoader( - [i for i in range(0, self.num_instances)], - None, - ModalityType.VIDEO, - data, - np.float32, - md, - ) - ) - aggregated_window = modality.context(DynamicWindow(num_windows=num_windows)) - for i in range(0, self.num_instances): - assert len(aggregated_window.data[i]) == num_windows - - def test_window_aggregation_on_audio_representations(self): - window_size = 10 - self.run_window_aggregation_for_modality(ModalityType.AUDIO, window_size) + for window_operator in [StaticWindow, DynamicWindow]: + with self.subTest(operator=window_operator.__name__): + modality = UnimodalModality( + TestDataLoader( + [i for i in range(0, self.num_instances)], + None, + ModalityType.VIDEO, + data, + np.float32, + md, + ) + ) + aggregated_window = modality.context( + window_operator(num_windows=num_windows) + ) - def test_window_operations_on_video_representations(self): - window_size = 10 - self.run_window_aggregation_for_modality(ModalityType.VIDEO, window_size) + for i in range(0, self.num_instances): + self.assertEqual(len(aggregated_window.data[i]), num_windows) - def test_window_operations_on_text_representations(self): + def test_window_aggregation_on_1d_modalities(self): + # create1DModality produces the same random matrix for all three + # modality types -- only the metadata label differs -- and + # window_aggregation dispatches on the data layout, not on the modality + # type. The three per-modality tests therefore ran identical code over + # identical numbers; the modality is a subTest dimension instead. 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 compresses the first (time) axis and leaves every + # feature axis untouched, so the expected shape is + # (num_windows,) + dims[1:] for any number of dimensions. One + # expression covers the 3d and 2d cases that were written out as two + # otherwise identical tests. 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(