diff --git a/src/main/python/tests/scuro/test_fusion_orders.py b/src/main/python/tests/scuro/test_fusion_orders.py index 22d64bcc0bf..5b9e8e60879 100644 --- a/src/main/python/tests/scuro/test_fusion_orders.py +++ b/src/main/python/tests/scuro/test_fusion_orders.py @@ -19,77 +19,81 @@ # # ------------------------------------------------------------- -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 systemds.scuro.drsearch.operator_registry import Registry from tests.scuro.data_generator import ModalityRandomDataGenerator from systemds.scuro.modality.type import ModalityType +# built-in fusions only: other test modules register their own, and a fusion +# that needs training cannot be combined without labels +FUSION_OPERATORS = [ + op + for op in Registry().get_fusion_operators() + if op.__module__.startswith("systemds.") and not op().needs_training +] + class TestFusionOrders(unittest.TestCase): + """ + Tests the order properties of the fusion operators against the flags they + declare. Swapping the inputs of a pair, or the first pair of a chain, only + keeps the result for a commutative operator. A pairwise chain only matches + the n-ary call for an associative one. + """ + @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 in FUSION_OPERATORS: + name = fusion_operator.__name__ + commutative = fusion_operator().commutative + associative = fusion_operator().associative + with self.subTest(fusion=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), + commutative, + f"{name}: swapping the two inputs", + ) + self.assertEqual( + self._equal(r_1_r_2_r_3, r_2_r_1_r_3), + commutative, + f"{name}: swapping the first pair of a chain", + ) + self.assertEqual( + self._equal(r_1_r_2_r_3, r1_r2_r3), + associative, + f"{name}: pairwise chain against the n-ary call", + ) + self.assertFalse( + self._equal(r_1_r_2, r1_r2_r3), + f"{name}: a pair against all three inputs", + ) diff --git a/src/main/python/tests/scuro/test_hp_tuner.py b/src/main/python/tests/scuro/test_hp_tuner.py index 1e0bad3991e..3586f7cd7a2 100644 --- a/src/main/python/tests/scuro/test_hp_tuner.py +++ b/src/main/python/tests/scuro/test_hp_tuner.py @@ -19,6 +19,7 @@ # # ------------------------------------------------------------- +import sys import unittest from types import SimpleNamespace @@ -41,6 +42,7 @@ ) from systemds.scuro.representations.word2vec import W2V from systemds.scuro.representations.bow import BoW +from systemds.scuro.modality.transformed import TransformedModality from systemds.scuro.modality.unimodal_modality import UnimodalModality from systemds.scuro.representations.resnet import ResNet from tests.scuro.data_generator import ( @@ -50,8 +52,28 @@ ) from systemds.scuro.modality.type import ModalityType -from systemds.scuro.drsearch.hyperparameter_tuner import HyperparameterTuner -from systemds.scuro.drsearch.representation_dag import CSEAwareDAGBuilder +from systemds.scuro.drsearch.hyperparameter_tuner import ( + HyperparamResult, + HyperparamResults, + HyperparameterTuner, + _apply_pushdown_trial_params, + _apply_trial_params_to_node, + _expand_aggregation_param_specs, + _is_aggregated_representation_operation, + _is_window_operation, + _materialize_node_params, + _param_values_to_spec, + _window_input_stats, +) +from systemds.scuro.drsearch.representation_dag import ( + CSEAwareDAGBuilder, + RepresentationDAGBuilder, + RepresentationNode, +) +from systemds.scuro.representations.aggregate import Aggregation +from systemds.scuro.representations.aggregated_representation import ( + AggregatedRepresentation, +) from systemds.scuro.representations.bert import Bert from systemds.scuro.representations.clip import CLIPText from systemds.scuro.representations.image_bind import ImageBind @@ -59,6 +81,7 @@ OverlappingSplitIndices, SentenceBoundarySplitIndices, ) +from systemds.scuro.representations.window_aggregation import WindowAggregation from unittest.mock import patch @@ -286,5 +309,362 @@ def fake_evaluate( self.assertEqual([r[0]["x"] for r in results], [1, 2]) +class _NarrowingAggregation: + """An aggregation that drops window sizes larger than the input. It stands + in for operators that adapt their domain to the data they receive.""" + + def __init__(self): + self.parameters = {"window_size": [2, 4, 8]} + + def filter_parameter_domain(self, name, values, input_stats): + return [value for value in values if value <= input_stats.output_shape[0]] + + +class TestSearchSpaceConstruction(unittest.TestCase): + """Every operator declares its parameters in its own shape. These helpers + translate them into the specifications optuna samples from.""" + + def test_param_values_to_spec_maps_a_list_to_a_categorical_domain(self): + spec = _param_values_to_spec("node0-window_size", [2, 4, 8]) + self.assertEqual( + spec, + {"name": "node0-window_size", "type": "categorical", "domain": [2, 4, 8]}, + ) + + def test_param_values_to_spec_maps_an_integer_pair_to_an_integer_domain(self): + # a pair describes a range and is sorted, so the declared order does + # not matter + spec = _param_values_to_spec("node0-n_mfcc", (20, 13)) + self.assertEqual(spec["type"], "integer") + self.assertEqual(spec["domain"], (13, 20)) + + def test_param_values_to_spec_maps_a_float_pair_to_a_real_domain(self): + spec = _param_values_to_spec("node0-ratio", (0.9, 0.1)) + self.assertEqual(spec["type"], "real") + self.assertEqual(spec["domain"], (0.1, 0.9)) + + def test_param_values_to_spec_wraps_a_single_value_in_a_categorical_domain(self): + # a fixed value becomes a domain too, so every parameter reaches optuna + # in the same shape + spec = _param_values_to_spec("node0-aggregation", "mean") + self.assertEqual(spec["type"], "categorical") + self.assertEqual(spec["domain"], ["mean"]) + + def test_param_values_to_spec_accepts_any_other_iterable(self): + spec = _param_values_to_spec("node0-window_size", range(3)) + self.assertEqual(spec["type"], "categorical") + self.assertEqual(spec["domain"], [0, 1, 2]) + + def test_param_values_to_spec_returns_none_for_values_without_a_domain(self): + for param_values in (None, {"window_size": 4}, range(0)): + with self.subTest(param_values=param_values): + self.assertIsNone(_param_values_to_spec("node0-p", param_values)) + + def test_param_values_to_spec_returns_none_when_the_value_cannot_be_listed(self): + # a zero dimensional array offers __iter__ but raises on list() + self.assertIsNone(_param_values_to_spec("node0-p", np.array(5))) + + def test_window_input_stats_reports_the_window_size_as_the_input_shape(self): + stats = _window_input_stats({"window_size": 8}) + self.assertEqual(stats.num_instances, 1) + self.assertEqual(stats.output_shape, (8,)) + + def test_window_input_stats_returns_none_without_a_window_size(self): + for node_parameters in (None, {}, {"aggregation_function": Aggregation}): + with self.subTest(node_parameters=node_parameters): + self.assertIsNone(_window_input_stats(node_parameters)) + + def test_expand_aggregation_param_specs_expands_the_nested_parameters(self): + specs = _expand_aggregation_param_specs("node0", Aggregation) + # Aggregation offers two nested names, but pad_modality carries no + # domain and is skipped + self.assertEqual(len(specs), 1) + self.assertEqual( + specs[0]["name"], "node0-aggregation_function_aggregation_function" + ) + self.assertIn("mean", specs[0]["domain"]) + + def test_expand_aggregation_param_specs_narrows_the_domain_to_the_input(self): + specs = _expand_aggregation_param_specs( + "node0", _NarrowingAggregation, _window_input_stats({"window_size": 4}) + ) + self.assertEqual([spec["domain"] for spec in specs], [[2, 4]]) + + def test_expand_aggregation_param_specs_returns_nothing_for_an_instance(self): + self.assertEqual(_expand_aggregation_param_specs("node0", Aggregation()), []) + + def test_expand_aggregation_param_specs_returns_nothing_without_nested_names(self): + class _PlainAggregation: + pass + + self.assertEqual( + _expand_aggregation_param_specs("node0", _PlainAggregation), [] + ) + + def test_expand_aggregation_param_specs_returns_nothing_for_a_failing_class(self): + # nested names are looked up by class name, so a class called + # Aggregation reports them before any instance exists -- only then can + # building the instance still fail + class Aggregation: + def __init__(self): + raise ValueError("cannot be built") + + self.assertEqual(_expand_aggregation_param_specs("node0", Aggregation), []) + + +class TestApplyTrialParams(unittest.TestCase): + """A trial hands back one flat value per parameter. Putting a value back is + not always a top level assignment: a pushed down aggregation keeps its + parameters in a nested key.""" + + def _pushdown_node_parameters(self): + # pushdown_aggregation moves the parameters of an aggregation node into + # this nested key and leaves the representation parameters on top + return {"layer": "avgpool", "_pushdown_aggregation": {"aggregation": "mean"}} + + def test_apply_pushdown_trial_params_moves_nested_values_into_the_pushdown(self): + result = _apply_pushdown_trial_params( + self._pushdown_node_parameters(), + {"aggregation_function_pad_modality": False}, + ) + self.assertEqual( + result["_pushdown_aggregation"], + {"aggregation": "mean", "aggregation_function_pad_modality": False}, + ) + + def test_apply_pushdown_trial_params_renames_the_aggregation_value(self): + # AggregatedRepresentation reads aggregation_function_aggregation_function + # before the plain aggregation key, so the sampled value wins + result = _apply_pushdown_trial_params( + self._pushdown_node_parameters(), {"aggregation": "max"} + ) + self.assertEqual( + result["_pushdown_aggregation"][ + "aggregation_function_aggregation_function" + ], + "max", + ) + + def test_apply_pushdown_trial_params_keeps_other_values_at_the_top_level(self): + result = _apply_pushdown_trial_params( + self._pushdown_node_parameters(), + {"layer": "fc", "_pushdown_aggregation": "ignored"}, + ) + self.assertEqual(result["layer"], "fc") + self.assertEqual(result["_pushdown_aggregation"], {"aggregation": "mean"}) + + def test_apply_pushdown_trial_params_leaves_the_base_parameters_unchanged(self): + base_params = self._pushdown_node_parameters() + _apply_pushdown_trial_params(base_params, {"aggregation": "max", "layer": "fc"}) + self.assertEqual(base_params, self._pushdown_node_parameters()) + + def test_apply_pushdown_trial_params_removes_nested_keys_from_the_top_level(self): + # a nested key left on top would reach the representation instead of + # the aggregation + base_params = { + "aggregation_function_pad_modality": True, + "_pushdown_aggregation": {}, + } + result = _apply_pushdown_trial_params(base_params, {"n_components": 4}) + self.assertNotIn("aggregation_function_pad_modality", result) + self.assertEqual(result["n_components"], 4) + + def test_apply_trial_params_to_node_routes_a_pushdown_node_to_the_merge(self): + node = RepresentationNode( + node_id="n0", + operation=WindowAggregation, + inputs=[], + parameters=self._pushdown_node_parameters(), + ) + result = _apply_trial_params_to_node(node, {"n0-aggregation": "max"}) + self.assertEqual(result["layer"], "avgpool") + self.assertEqual( + result["_pushdown_aggregation"][ + "aggregation_function_aggregation_function" + ], + "max", + ) + + def test_apply_trial_params_to_node_mirrors_the_aggregation(self): + node = RepresentationNode( + node_id="n1", + operation=AggregatedRepresentation, + inputs=[], + parameters={ + "aggregation_function_aggregation_function": "mean", + "aggregation_function_pad_modality": True, + "target_dimensions": 8, + }, + ) + result = _apply_trial_params_to_node(node, {"n1-aggregation": "max"}) + self.assertEqual(result["aggregation_function_aggregation_function"], "max") + self.assertNotIn("aggregation_function_pad_modality", result) + self.assertEqual(result["target_dimensions"], 8) + + def test_is_window_operation_recognises_only_window_classes(self): + for operation, expected in ( + (WindowAggregation, True), + (AggregatedRepresentation, False), + ("WindowAggregation", False), + ): + with self.subTest(operation=operation): + self.assertEqual(_is_window_operation(operation), expected) + + def test_is_aggregated_representation_operation_recognises_only_its_classes(self): + for operation, expected in ( + (AggregatedRepresentation, True), + (WindowAggregation, False), + (None, False), + ): + with self.subTest(operation=operation): + self.assertEqual( + _is_aggregated_representation_operation(operation), expected + ) + + def test_operation_checks_fall_back_when_the_module_is_unavailable(self): + # both checks import their base class inside the function + checks = ( + ( + _is_window_operation, + "systemds.scuro.representations.window_aggregation", + WindowAggregation, + ), + ( + _is_aggregated_representation_operation, + "systemds.scuro.representations.aggregated_representation", + AggregatedRepresentation, + ), + ) + for check, module_name, operation in checks: + with self.subTest(check=check.__name__): + with patch.dict(sys.modules, {module_name: None}): + self.assertFalse(check(operation)) + + def test_materialize_node_params_returns_the_input_outside_a_window(self): + window_node = RepresentationNode( + node_id="n0", operation=WindowAggregation, inputs=[] + ) + leaf_node = RepresentationNode(node_id="n1", operation=None, inputs=[]) + for node, flat_params in ((leaf_node, {"window_size": 4}), (window_node, {})): + with self.subTest(operation=node.operation): + self.assertEqual( + _materialize_node_params(node, flat_params), flat_params + ) + + +class TestHyperparamResults(unittest.TestCase): + """HyperparamResults stores what the tuner found. Reading it back rebuilds + the DAG of a result with the parameters that scored best.""" + + def setUp(self): + self.task = SimpleNamespace(model=SimpleNamespace(name="TuningTask")) + self.modality = SimpleNamespace(modality_id="audio_0") + self.results = HyperparamResults([self.task], [self.modality]) + + def _dag_with_one_operation(self): + # the unimodal optimizer builds its DAGs through the same builder + builder = RepresentationDAGBuilder() + leaf_id = builder.create_leaf_node(self.modality.modality_id) + operation_id = builder.create_operation_node( + WindowAggregation, [leaf_id], {"window_size": 4} + ) + return builder.build(operation_id), operation_id + + def _result(self, dag, best_params, mm_opt=False): + return HyperparamResult( + representation_name="WindowAggregation", + best_params=best_params, + best_score=0.8, + all_results=[], + tuning_time=0.1, + modality_id=self.modality.modality_id, + task_name=self.task.model.name, + dag=dag, + mm_opt=mm_opt, + ) + + def _store(self, results): + self.results.results[self.task.model.name][self.modality.modality_id] = results + + def test_add_result_skips_a_missing_result(self): + # a representation that could not be tuned arrives as None + self.results.add_result([None]) + self.assertEqual( + self.results.results[self.task.model.name][self.modality.modality_id], [] + ) + + def test_add_result_stores_a_multimodal_result_under_its_own_key(self): + dag, _ = self._dag_with_one_operation() + self.results.setup_mm(optimize_unimodal=False) + self.results.add_result([self._result(dag, {}, mm_opt=True)]) + self.assertEqual( + len(self.results.results[self.task.model.name]["mm_results"]), 1 + ) + + def test_setup_mm_replaces_the_results_with_a_multimodal_slot(self): + self.results.setup_mm(optimize_unimodal=False) + self.assertEqual( + self.results.results, {self.task.model.name: {"mm_results": []}} + ) + + def test_setup_mm_keeps_the_results_when_the_unimodal_step_runs(self): + self.results.setup_mm(optimize_unimodal=True) + self.assertEqual( + self.results.results, + {self.task.model.name: {self.modality.modality_id: []}}, + ) + + def test_get_k_best_dags_rebuilds_the_dag_with_the_best_parameters(self): + dag, operation_id = self._dag_with_one_operation() + self._store([self._result(dag, {f"{operation_id}-window_size": 8})]) + + _, dags = self.results.get_k_best_dags(self.modality, self.task) + + operation_nodes = [node for node in dags[0].nodes if node.operation is not None] + leaf_nodes = [node for node in dags[0].nodes if node.operation is None] + self.assertEqual(operation_nodes[0].parameters["window_size"], 8) + self.assertEqual(leaf_nodes[0].modality_id, self.modality.modality_id) + self.assertEqual(dags[0].root_node_id, operation_nodes[0].node_id) + # the stored result keeps the parameters it was evaluated with + self.assertEqual(dag.get_node_by_id(operation_id).parameters["window_size"], 4) + + def test_get_k_best_dags_returns_one_dag_per_stored_result(self): + first_dag, _ = self._dag_with_one_operation() + second_dag, _ = self._dag_with_one_operation() + stored = [self._result(first_dag, {}), self._result(second_dag, {})] + self._store(stored) + + results, dags = self.results.get_k_best_dags(self.modality, self.task) + + self.assertIs(results, stored) + self.assertEqual(len(dags), 2) + + def test_get_k_best_dags_returns_nothing_without_stored_results(self): + self.assertEqual( + self.results.get_k_best_dags(self.modality, self.task), ([], []) + ) + + def test_get_k_best_results_executes_every_rebuilt_dag(self): + modality = ModalityRandomDataGenerator().create1DModality( + 6, 4, ModalityType.TIMESERIES + ) + modality.modality_id = 0 + results = HyperparamResults([self.task], [modality]) + builder = RepresentationDAGBuilder() + leaf_id = builder.create_leaf_node(modality.modality_id) + operation_id = builder.create_operation_node( + WindowAggregation, [leaf_id], {"window_size": 2} + ) + results.results[self.task.model.name][modality.modality_id] = [ + self._result(builder.build(operation_id), {}) + ] + + _, representations = results.get_k_best_results(modality, self.task, "accuracy") + + self.assertEqual(len(representations), 1) + self.assertIsInstance(representations[0], TransformedModality) + self.assertEqual(len(representations[0].data), 6) + + if __name__ == "__main__": unittest.main() diff --git a/src/main/python/tests/scuro/test_multimodal_fusion.py b/src/main/python/tests/scuro/test_multimodal_fusion.py index 8086bcc3f07..b50c75d11cb 100644 --- a/src/main/python/tests/scuro/test_multimodal_fusion.py +++ b/src/main/python/tests/scuro/test_multimodal_fusion.py @@ -20,10 +20,14 @@ # ------------------------------------------------------------- import unittest +from types import SimpleNamespace import numpy as np -from systemds.scuro.drsearch.multimodal_optimizer import MultimodalOptimizer +from systemds.scuro.drsearch.multimodal_optimizer import ( + MultimodalOptimizer, + OptimizationResult, +) from systemds.scuro.drsearch.unimodal_optimizer import UnimodalOptimizer from systemds.scuro.representations.concatenation import Concatenation from systemds.scuro.representations.lstm import LSTM @@ -187,5 +191,192 @@ class TestMultimodalRepresentationOptimizer(unittest.TestCase): # ) +class _FakeModality: + def __init__(self, modality_id): + self.modality_id = modality_id + + +class _FakeUnimodalResults: + """Stand-in for UnimodalOptimizer.operator_performance: hands back a + fixed list of representations per modality without running any real + unimodal search.""" + + def __init__(self, reps_per_modality): + self.reps_per_modality = reps_per_modality + + def get_k_best_results(self, modality, task, performance_metric_name): + reps = self.reps_per_modality.get(modality.modality_id, []) + return list(range(len(reps))), reps + + +def _make_task(name="task0"): + return SimpleNamespace(model=SimpleNamespace(name=name)) + + +def _make_optimizer(reps_per_modality, **kwargs): + modalities = [_FakeModality(modality_id) for modality_id in reps_per_modality] + task = _make_task() + kwargs.setdefault("debug", False) + # a resumed run loads any checkpoint left in the working directory + kwargs.setdefault("resume", False) + optimizer = MultimodalOptimizer( + modalities, _FakeUnimodalResults(reps_per_modality), [task], **kwargs + ) + return optimizer, task + + +class TestFusionDagGeneration(unittest.TestCase): + def test_modality_combinations_respect_min_max_modalities(self): + reps = {"m0": [object()], "m1": [object()], "m2": [object()]} + pairs = [["m0", "m1"], ["m0", "m2"], ["m1", "m2"]] + for max_modalities, expected in ( + (2, pairs), + (3, pairs + [["m0", "m1", "m2"]]), + (5, pairs + [["m0", "m1", "m2"]]), + ): + with self.subTest(max_modalities=max_modalities): + optimizer, _ = _make_optimizer( + reps, min_modalities=2, max_modalities=max_modalities + ) + self.assertEqual( + list(optimizer._generate_modality_combinations()), expected + ) + + def test_representation_combinations_pick_one_per_modality(self): + reps = {"m0": [object(), object()], "m1": [object(), object(), object()]} + optimizer, task = _make_optimizer(reps) + combinations = list( + optimizer._generate_representation_combinations( + ["m0", "m1"], task.model.name + ) + ) + self.assertEqual( + combinations, + [ + {"m0": 0, "m1": 0}, + {"m0": 0, "m1": 1}, + {"m0": 0, "m1": 2}, + {"m0": 1, "m1": 0}, + {"m0": 1, "m1": 1}, + {"m0": 1, "m1": 2}, + ], + ) + + def test_fusion_dags_contain_every_selected_representation(self): + reps = { + "m0": [object(), object()], + "m1": [object()], + "m2": [object(), object()], + } + optimizer, _ = _make_optimizer(reps) + optimizer.fusion_operators = [Concatenation, Average] + selection = {"m0": 1, "m1": 0, "m2": 1} + dags = list(optimizer._generate_fusion_dags(["m0", "m1", "m2"], selection)) + # the same tree can be yielded more than once + self.assertTrue(dags) + for dag in dags: + leaves = [node for node in dag.nodes if not node.inputs] + fusions = [node for node in dag.nodes if node.inputs] + self.assertEqual( + {(leaf.modality_id, leaf.representation_index) for leaf in leaves}, + set(selection.items()), + ) + self.assertEqual(len(fusions), len(leaves) - 1) + self.assertTrue(all(len(node.inputs) == 2 for node in fusions)) + self.assertIn(dag.root_node_id, {node.node_id for node in fusions}) + + def test_fusion_dags_use_every_fusion_operator(self): + reps = {"m0": [object()], "m1": [object()]} + optimizer, _ = _make_optimizer(reps) + optimizer.fusion_operators = [Concatenation, Average] + used = { + node.operation + for dag in optimizer._generate_fusion_dags(["m0", "m1"], {"m0": 0, "m1": 0}) + for node in dag.nodes + if node.inputs + } + self.assertEqual(used, {Concatenation, Average}) + + +class TestConstructorValidation(unittest.TestCase): + def test_min_modalities_clamped_to_two(self): + optimizer, _ = _make_optimizer( + {"m0": [object()], "m1": [object()]}, min_modalities=1 + ) + self.assertEqual(optimizer.min_modalities, 2) + + def test_max_modalities_defaults_to_number_of_modalities(self): + optimizer, _ = _make_optimizer( + {"m0": [object()], "m1": [object()], "m2": [object()]} + ) + self.assertEqual(optimizer.max_modalities, 3) + + def test_k_best_representations_hold_the_cached_data(self): + reps = {"m0": [object(), object()], "m1": [object()]} + optimizer, task = _make_optimizer(reps) + k_best = optimizer.k_best_representations[task.model.name] + self.assertEqual(set(k_best), {"m0", "m1"}) + for modality_id in reps: + self.assertIs(k_best[modality_id], reps[modality_id]) + + +def _stored_result(dag, task): + return OptimizationResult(dag=dag, task_name=task.model.name) + + +class TestOptimizeLoop(unittest.TestCase): + def test_stops_at_max_combinations(self): + optimizer, task = _make_optimizer( + {"m0": [object()], "m1": [object()], "m2": [object()]} + ) + optimizer.fusion_operators = [Concatenation] + with patch.object( + optimizer, "_evaluate_dag", side_effect=_stored_result + ) as evaluate_dag: + results = optimizer.optimize(max_combinations=5) + + self.assertEqual(evaluate_dag.call_count, 5) + self.assertEqual(len(results[task.model.name]), 5) + + def test_drops_failed_evaluations(self): + optimizer, task = _make_optimizer( + {"m0": [object()], "m1": [object()], "m2": [object()]} + ) + optimizer.fusion_operators = [Concatenation] + call_counter = {"n": 0} + + def flaky_evaluate(dag, _task): + call_counter["n"] += 1 + if call_counter["n"] % 2 == 0: + return None + return _stored_result(dag, _task) + + with patch.object(optimizer, "_evaluate_dag", side_effect=flaky_evaluate): + results = optimizer.optimize(max_combinations=6) + + self.assertEqual(call_counter["n"], 6) + self.assertEqual(len(results[task.model.name]), 3) + + def test_keeps_results_and_budget_per_task(self): + tasks = [_make_task("task_a"), _make_task("task_b")] + optimizer = MultimodalOptimizer( + [_FakeModality("m0"), _FakeModality("m1")], + _FakeUnimodalResults({"m0": [object()], "m1": [object()]}), + tasks, + debug=False, + resume=False, + ) + optimizer.fusion_operators = [Concatenation] + with patch.object(optimizer, "_evaluate_dag", side_effect=_stored_result): + results = optimizer.optimize(max_combinations=2) + + self.assertEqual( + {name: len(stored) for name, stored in results.items()}, + {"task_a": 2, "task_b": 2}, + ) + for name, stored in results.items(): + self.assertEqual({result.task_name for result in stored}, {name}) + + if __name__ == "__main__": unittest.main() diff --git a/src/main/python/tests/scuro/test_representation_dag.py b/src/main/python/tests/scuro/test_representation_dag.py new file mode 100644 index 00000000000..dbc59886476 --- /dev/null +++ b/src/main/python/tests/scuro/test_representation_dag.py @@ -0,0 +1,222 @@ +# ------------------------------------------------------------- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +# ------------------------------------------------------------- + +import unittest + +from systemds.scuro.drsearch.representation_dag import ( + CSEAwareDAGBuilder, + RepresentationDAGBuilder, + RepresentationDag, + RepresentationNode, + get_modality_by_id_and_instance_id, + pushdown_aggregation, + pushdown_aggregation_for_node, +) +from systemds.scuro.modality.transformed import TransformedModality +from systemds.scuro.modality.type import ModalityType +from systemds.scuro.representations.aggregated_representation import ( + AggregatedRepresentation, +) +from systemds.scuro.representations.concatenation import Concatenation +from systemds.scuro.representations.window_aggregation import WindowAggregation +from tests.scuro.data_generator import ModalityRandomDataGenerator + + +def _leaf(node_id, modality_id=0): + return RepresentationNode(node_id, None, [], modality_id, 0) + + +def _op(node_id, inputs, operation=Concatenation): + return RepresentationNode(node_id, operation, inputs) + + +def _modality(modality_id): + modality = ModalityRandomDataGenerator().create1DModality( + 6, 4, ModalityType.TIMESERIES + ) + modality.modality_id = modality_id + return modality + + +class _TrainedFusion(Concatenation): + def __init__(self, params=None): + super().__init__() + self.needs_training = True + + +class _RecordingNodeCache: + def __init__(self): + self.entries = {} + self.stored = [] + + def get(self, node_id): + return self.entries.get(node_id) + + def put(self, node_id, value, compute_seconds): + self.stored.append(node_id) + self.entries[node_id] = value + + +class TestDagValidation(unittest.TestCase): + def test_validate_accepts_a_connected_dag(self): + dag = RepresentationDag( + [_op("root", ["a", "b"]), _leaf("a"), _leaf("b", 1)], "root" + ) + self.assertTrue(dag.validate()) + + def test_validate_rejects_malformed_dags(self): + malformed = { + "empty": ([], "root"), + "missing root": ([_leaf("a")], "root"), + "dangling input": ([_op("root", ["a", "ghost"]), _leaf("a")], "root"), + "cycle": ([_op("a", ["b"]), _op("b", ["a"])], "a"), + "self loop": ([_op("a", ["a"])], "a"), + } + for case, (nodes, root) in malformed.items(): + with self.subTest(case=case): + self.assertFalse(RepresentationDag(nodes, root).validate()) + + def test_unreachable_nodes_are_dropped(self): + dag = RepresentationDag( + [_op("root", ["a", "b"]), _leaf("a"), _leaf("b", 1), _leaf("orphan", 2)], + "root", + ) + self.assertEqual([n.node_id for n in dag.nodes], ["root", "a", "b"]) + self.assertTrue(dag.validate()) + + +class TestDagBuilders(unittest.TestCase): + def test_build_rejects_an_unknown_root(self): + for builder_cls in (RepresentationDAGBuilder, CSEAwareDAGBuilder): + with self.subTest(builder=builder_cls.__name__): + builder = builder_cls() + builder.create_leaf_node(0, 0) + with self.assertRaises(ValueError): + builder.build("missing") + + def test_cse_builder_reuses_identical_nodes(self): + builder = CSEAwareDAGBuilder() + leaf_id = builder.create_leaf_node(0, 0) + self.assertEqual(builder.create_leaf_node(0, 0), leaf_id) + + def window(size): + return builder.create_operation_node( + WindowAggregation, [leaf_id], {"window_size": size} + ) + + window_id = window(2) + self.assertEqual(window(2), window_id) + self.assertNotEqual(window(4), window_id) + + def test_cse_builder_returns_the_same_dag_for_a_root(self): + builder = CSEAwareDAGBuilder() + root_id = builder.create_operation_node( + WindowAggregation, [builder.create_leaf_node(0, 0)], {"window_size": 2} + ) + dag = builder.build(root_id) + + self.assertIs(builder.build(root_id), dag) + with_id = builder.build(root_id, dag_id=99) + self.assertIsNot(with_id, dag) + self.assertEqual(with_id.dag_id, 99) + + +class TestDagExecution(unittest.TestCase): + def setUp(self): + self.modalities = [_modality(0), _modality(1)] + + def _fusion_dag(self, fusion=Concatenation): + builder = RepresentationDAGBuilder() + window_id = builder.create_operation_node( + WindowAggregation, [builder.create_leaf_node(0, 0)], {"window_size": 2} + ) + root_id = builder.create_operation_node( + fusion, [window_id, builder.create_leaf_node(1, 0)] + ) + return builder.build(root_id), window_id + + def test_execute_returns_the_root_output(self): + dag, _ = self._fusion_dag() + result = dag.execute(self.modalities) + self.assertIsInstance(result, TransformedModality) + self.assertEqual(result.data.shape, (6, 6)) # 2 windowed + 4 raw + + def test_fusion_that_needs_training_is_rejected(self): + dag, _ = self._fusion_dag(_TrainedFusion) + with self.assertRaisesRegex(RuntimeError, "SplitAwareDAGEvaluator"): + dag.execute(self.modalities) + + def test_node_cache_holds_only_intermediate_nodes(self): + dag, window_id = self._fusion_dag() + node_cache = _RecordingNodeCache() + + dag.execute(self.modalities, node_cache=node_cache) + self.assertEqual(node_cache.stored, [window_id]) + + dag.execute(self.modalities, node_cache=node_cache) + self.assertEqual(node_cache.stored, [window_id]) + + +class TestDagHelpers(unittest.TestCase): + def test_modality_lookup_picks_the_requested_instance(self): + first, second, other = _modality(1), _modality(1), _modality(0) + modalities = [first, second, other] + for (modality_id, instance_id), expected in { + (1, -1): first, + (1, 1): second, + (0, 1): other, + }.items(): + with self.subTest(modality_id=modality_id, instance_id=instance_id): + self.assertIs( + get_modality_by_id_and_instance_id( + modalities, modality_id, instance_id + ), + expected, + ) + self.assertIsNone(get_modality_by_id_and_instance_id(modalities, 9, 0)) + + def test_pushdown_parameters_become_an_aggregation(self): + self.assertIsNone(pushdown_aggregation_for_node(None)) + self.assertIsNone(pushdown_aggregation_for_node({"window_size": 2})) + parameters = AggregatedRepresentation().get_current_parameters() + aggregation = pushdown_aggregation_for_node( + {"_pushdown_aggregation": parameters} + ) + self.assertIsInstance(aggregation, AggregatedRepresentation) + + def test_pushdown_aggregation_keeps_nodes_it_cannot_merge(self): + dag = RepresentationDag( + [ + _op("agg", ["window"], AggregatedRepresentation), + _op("window", ["a"], WindowAggregation), + _leaf("a"), + ], + "agg", + ) + before = [(n.node_id, n.operation, list(n.inputs)) for n in dag.nodes] + pushdown_aggregation([dag]) + self.assertEqual( + [(n.node_id, n.operation, n.inputs) for n in dag.nodes], before + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/main/python/tests/scuro/test_unimodal_optimizer.py b/src/main/python/tests/scuro/test_unimodal_optimizer.py index 215b4712466..3fb018ed2ec 100644 --- a/src/main/python/tests/scuro/test_unimodal_optimizer.py +++ b/src/main/python/tests/scuro/test_unimodal_optimizer.py @@ -20,6 +20,10 @@ # ------------------------------------------------------------- +import os +import pickle +import shutil +import tempfile import unittest from types import SimpleNamespace @@ -30,6 +34,7 @@ from systemds.scuro.drsearch.unimodal_optimizer import ( UnimodalOptimizer, UnimodalResults, + get_dag_by_id, ) from systemds.scuro.representations.covarep_audio_features import ZeroCrossing @@ -140,16 +145,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_fitted_text_leaves_are_refit_per_fold_in_fusion(self): text_data, text_md = ModalityRandomDataGenerator().create_text_data( @@ -476,59 +516,6 @@ def test_windows_are_kept_for_representations_with_a_time_axis(self): self.assertIs(operation, ResNet) self.assertGreaterEqual(rank, 2) - 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 # ------------------------------------------------------------------ @@ -737,6 +724,231 @@ def optimize_unimodal_representation_for_modality(self, modalities): assert len(result) == 1 +class TestUnimodalOptimizerPersistence(unittest.TestCase): + """store_results writes the search results to disk so that a long search + can be resumed or inspected afterwards. The files are written from the + result container, so no search has to run.""" + + # TestTask stratifies its train and validation split, so it needs at least + # one instance per class in each part. + num_instances = 10 + + def setUp(self): + self.result_path = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.result_path) + indices = np.array(range(self.num_instances)) + data, metadata = ModalityRandomDataGenerator().create_audio_data( + self.num_instances, 200 + ) + self.modality = UnimodalModality( + TestDataLoader( + indices, None, ModalityType.AUDIO, data, np.float32, metadata + ) + ) + self.task = TestTask("PersistenceTask", "Test1", self.num_instances) + self.optimizer = UnimodalOptimizer( + [self.modality], + [self.task], + result_path=self.result_path, + enable_checkpointing=False, + ) + + def _read(self, file_name): + with open(os.path.join(self.result_path, file_name), "rb") as f: + return pickle.load(f) + + def test_store_results_writes_the_results_and_the_execution_statistics(self): + self.optimizer.store_results("results.pkl") + + self.assertEqual( + sorted(os.listdir(self.result_path)), + ["results.pkl", "results_exec_stats.pkl"], + ) + self.assertEqual( + self._read("results.pkl"), self.optimizer.operator_performance.results + ) + self.assertEqual( + set(self._read("results_exec_stats.pkl")), + { + "worker_stats", + "node_stats", + "reuse_stats", + "wall_clock_s", + "search_start_unix", + "max_num_workers", + }, + ) + + def test_store_results_names_the_file_after_the_optimizer_and_the_time(self): + self.optimizer.store_results() + + written = sorted(os.listdir(self.result_path)) + self.assertEqual(len(written), 2) + for name in written: + self.assertTrue(name.startswith("unimodal_optimizer")) + self.assertTrue(name.endswith(".pkl")) + + def test_store_results_appends_the_statistics_suffix_without_a_pkl_ending(self): + # The statistics file name is derived by replacing ".pkl". Without that + # ending the name stays unchanged, so the suffix is appended. + self.optimizer.store_results("results") + + self.assertEqual( + sorted(os.listdir(self.result_path)), + ["results", "results_exec_stats.pkl"], + ) + + def test_results_survive_a_store_and_load_round_trip(self): + modality_id = self.modality.modality_id + task_name = self.task.model.name + entry = ResultEntry( + val_score={"accuracy": 0.75}, representation_time=1.0, task_time=2.0 + ) + self.optimizer.operator_performance.results[modality_id][task_name] = [entry] + + self.optimizer.store_results("results.pkl") + self.optimizer.operator_performance.results[modality_id][task_name] = [] + self.optimizer.load_results(os.path.join(self.result_path, "results.pkl")) + + restored = self.optimizer.operator_performance.results[modality_id][task_name] + self.assertEqual(len(restored), 1) + self.assertEqual(restored[0].val_score, {"accuracy": 0.75}) + + def test_count_results_by_modality_counts_the_first_task_of_every_modality(self): + # The count drives the checkpoint progress, where every task evaluates + # the same representations, so one task stands for all of them. + counts = self.optimizer._count_results_by_modality( + {"audio": {"t1": [1, 2], "t2": [3]}, "text": {"t1": []}} + ) + + self.assertEqual(counts, {"audio": 2, "text": 0}) + + def test_resume_from_checkpoint_restores_a_stored_result_set(self): + restored = {"audio": {"t1": ["entry"]}} + with patch.object( + self.optimizer._checkpoint_manager, + "resume_from_checkpoint", + return_value=(restored, None, None), + ): + self.optimizer.resume_from_checkpoint() + + self.assertEqual(self.optimizer.operator_performance.results, restored) + + def test_resume_from_checkpoint_keeps_the_results_without_a_checkpoint(self): + current = self.optimizer.operator_performance.results + with patch.object( + self.optimizer._checkpoint_manager, + "resume_from_checkpoint", + return_value=None, + ): + self.optimizer.resume_from_checkpoint() + + self.assertIs(self.optimizer.operator_performance.results, current) + + +class TestUnimodalResultsReadout(unittest.TestCase): + """UnimodalResults holds the scores of every evaluated representation and a + cache of the data the best ones produced. get_k_best_results returns both. + The entries are written straight into the container, so no search has to + run.""" + + num_instances = 10 + + def setUp(self): + indices = np.array(range(self.num_instances)) + data, metadata = ModalityRandomDataGenerator().create_audio_data( + self.num_instances, 200 + ) + self.modality = UnimodalModality( + TestDataLoader( + indices, None, ModalityType.AUDIO, data, np.float32, metadata + ) + ) + self.task = TestTask("ReadoutTask", "Test1", self.num_instances) + self.results = UnimodalResults( + [self.modality], [self.task], k=2, metric_name="accuracy" + ) + self.modality_id = self.modality.modality_id + self.task_name = self.task.model.name + + def _entry(self, accuracy, dag=None): + return ResultEntry( + val_score={"accuracy": accuracy}, + representation_time=1.0, + task_time=1.0, + dag=dag, + ) + + def _fill(self, accuracies): + entries = [self._entry(a) for a in accuracies] + self.results.results[self.modality_id][self.task_name] = entries + return entries + + def test_get_k_best_results_returns_the_k_highest_scores(self): + self._fill([0.5, 0.9, 0.7, 0.3]) + + best, _ = self.results.get_k_best_results( + self.modality, self.task, "accuracy", cache_needed=False + ) + + self.assertEqual([entry.val_score["accuracy"] for entry in best], [0.9, 0.7]) + + def test_get_k_best_results_uses_the_cache_when_it_holds_entries(self): + self._fill([0.9, 0.7]) + self.results.cache[self.modality_id][self.task_name] = ["first", "second"] + + _, cache = self.results.get_k_best_results(self.modality, self.task, "accuracy") + + self.assertEqual(cache, ["first", "second"]) + + def test_get_k_best_results_executes_the_dag_when_the_cache_is_empty(self): + # load_results restores the scores but not the cached data, so the + # cache is empty after reading a file. The dag of every returned entry + # is executed to rebuild it. + executed = [] + + class RecordingDag: + # no node needs fitting, so the dag is executed directly + nodes = [] + + def __init__(self, label): + self.label = label + + def execute(self, modalities): + executed.append(self.label) + return self.label + + entries = [ + self._entry(0.9, dag=RecordingDag("best")), + self._entry(0.7, dag=RecordingDag("second")), + ] + self.results.results[self.modality_id][self.task_name] = entries + self.results.cache[self.modality_id][self.task_name] = [] + + _, cache = self.results.get_k_best_results(self.modality, self.task, "accuracy") + + self.assertEqual(executed, ["best", "second"]) + self.assertEqual(cache, ["best", "second"]) + + def test_get_k_best_results_returns_nothing_without_entries(self): + best, cache = self.results.get_k_best_results( + self.modality, self.task, "accuracy", cache_needed=False + ) + + self.assertEqual(best, []) + self.assertEqual(cache, []) + + def test_get_dag_by_id_finds_the_matching_dag(self): + dags = [SimpleNamespace(dag_id=1), SimpleNamespace(dag_id=7)] + + self.assertIs(get_dag_by_id(dags, 7), dags[1]) + + def test_get_dag_by_id_returns_none_for_an_unknown_id(self): + dags = [SimpleNamespace(dag_id=1)] + + self.assertIsNone(get_dag_by_id(dags, 99)) + + class TestSelectionStrategies(unittest.TestCase): """The three selection strategies must genuinely disagree. 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 ddfecc16862..f526851098e 100644 --- a/src/main/python/tests/scuro/test_window_operations.py +++ b/src/main/python/tests/scuro/test_window_operations.py @@ -114,67 +114,74 @@ 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) + 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), + ]: + name = type(window_operator).__name__ + with self.subTest(dims=dims, operator=name): + stats = window_operator.get_output_stats(embedding_modality.stats) + self.assertEqual( + stats.num_instances, + self.num_instances, + f"{name}: number of instances in the stats", + ) + self.assertEqual( + stats.output_shape, + (num_windows,) + dims[1:], + f"{name}: output shape in the stats", + ) - windowed_modality = embedding_modality.context(window_operator) + windowed = embedding_modality.context(window_operator) + self.assertEqual( + np.asarray(windowed.data).shape, + (stats.num_instances,) + stats.output_shape, + f"{name}: shape of the windowed data against the stats", + ) def _timeseries_modality(self, signal_length=100): return self.data_generator.create1DModality(