[SYSTEMDS-3887] Improve Scuro DAGs

src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py

@@ -75,7 +75,20 @@ def _param_values_to_spec(
     return None
 
 
-def _expand_aggregation_param_specs(op_id: str, agg_cls: Any) -> List[Dict[str, Any]]:
+def _window_input_stats(node_parameters: Optional[Dict[str, Any]]):
+    from systemds.scuro.representations.representation import RepresentationStats
+
+    if not node_parameters:
+        return None
+    window_length = node_parameters.get("window_size")
+    if window_length is None:
+        return None
+    return RepresentationStats(1, (int(window_length),))
+
+
+def _expand_aggregation_param_specs(
+    op_id: str, agg_cls: Any, input_stats=None
+) -> List[Dict[str, Any]]:
     if not inspect.isclass(agg_cls):
         return []
 
@@ -98,6 +111,10 @@ def _expand_aggregation_param_specs(op_id: str, agg_cls: Any) -> List[Dict[str,
         nested_values = search_template.get(nested_name)
         if nested_values is None:
             continue
+        if input_stats is not None and isinstance(nested_values, list):
+            narrow = getattr(instance, "filter_parameter_domain", None)
+            if narrow is not None:
+                nested_values = narrow(nested_name, nested_values, input_stats)
         full_name = f"{op_id}-aggregation_function_{nested_name}"
         spec = _param_values_to_spec(full_name, nested_values)
         if spec is not None:
@@ -251,7 +268,7 @@ def setup_mm(self, optimize_unimodal):
             for task in self.tasks:
                 self.results[task.model.name] = {"mm_results": []}
 
-    def get_k_best_results(self, modality, task, performance_metric_name):
+    def get_k_best_dags(self, modality, task):
         results = self.results[task.model.name][modality.modality_id]
         dags = []
         for result in results:
@@ -269,6 +286,10 @@ def get_k_best_results(self, modality, task, performance_metric_name):
                     )
 
             dags.append(dag_with_best_params.build(prev_node_id))
+        return results, dags
+
+    def get_k_best_results(self, modality, task, performance_metric_name):
+        results, dags = self.get_k_best_dags(modality, task)
         representations = [list(dag.execute([modality]).values())[-1] for dag in dags]
         return results, representations
 
@@ -488,7 +509,11 @@ def visit_node(node_id):
         if not hyperparams:
             all_results = [baseline]
         else:
-            param_specs = self._build_param_specs(hyperparams)
+            node_parameters = {
+                node_id: (dag.get_node_by_id(node_id).parameters or {})
+                for node_id in hyperparams
+            }
+            param_specs = self._build_param_specs(hyperparams, node_parameters)
             discrete_size = self._estimate_discrete_search_size(param_specs)
             n_calls = min(discrete_size, max_evals) if max_evals else discrete_size
             all_results = self._search_best_configs(
@@ -575,7 +600,12 @@ def __get_params_for_node(self, node: RepresentationNode) -> Dict[str, Any]:
         if node.parameters:
             if inspect.isclass(node.parameters.get("aggregation_function")):
                 params["aggregation_function"] = node.parameters["aggregation_function"]
-            for fixed_key in ("target_dimensions", "self_contained"):
+            for fixed_key in (
+                "target_dimensions",
+                "self_contained",
+                "aggregate_leading",
+                "preserve_leading_axis",
+            ):
                 if fixed_key in node.parameters:
                     params[fixed_key] = node.parameters[fixed_key]
 
@@ -587,13 +617,19 @@ def __get_params_for_node(self, node: RepresentationNode) -> Dict[str, Any]:
         return params
 
     def _build_param_specs(
-        self, hyperparams: Dict[str, Dict[str, Any]]
+        self,
+        hyperparams: Dict[str, Dict[str, Any]],
+        node_parameters: Optional[Dict[str, Dict[str, Any]]] = None,
     ) -> List[Dict[str, Any]]:
         param_specs = []
+        node_parameters = node_parameters or {}
         for op_id, op_params in hyperparams.items():
+            input_stats = _window_input_stats(node_parameters.get(op_id))
             for param_name, param_values in op_params.items():
                 if param_name == "aggregation_function":
-                    expanded = _expand_aggregation_param_specs(op_id, param_values)
+                    expanded = _expand_aggregation_param_specs(
+                        op_id, param_values, input_stats
+                    )
                     if expanded:
                         param_specs.extend(expanded)
                         continue

src/main/python/systemds/scuro/drsearch/node_executor.py

@@ -381,6 +381,7 @@ def __init__(
         result_path: Optional[str] = None,
         enable_checkpointing: bool = False,
         worker_pool: Optional[PersistentWorkerPool] = None,
+        search_start: Optional[float] = None,
     ):
         self.enable_checkpointing = enable_checkpointing
         available_total_cpu = cpu_memory_budget_bytes()
@@ -411,6 +412,11 @@ def __init__(
         )
         self._memory_usage_data: Dict[str, Any] = {}
         self.statistics = {"worker_stats": {}, "node_stats": {}}
+        self._eval_counter = 0
+        self._nodes_executed = 0
+        self._search_start = (
+            search_start if search_start is not None else time.perf_counter()
+        )
 
         self._node_attempts: Dict[str, int] = {}
 

src/main/python/systemds/scuro/drsearch/operator_registry.py

@@ -47,7 +47,7 @@ def __new__(cls):
 
     def set_fusion_operators(self, fusion_operators):
         if isinstance(fusion_operators, list):
-            self._context_operators = fusion_operators
+            self._fusion_operators = fusion_operators
         else:
             self._fusion_operators = [fusion_operators]
 
@@ -155,7 +155,7 @@ def get_representation_by_name(self, representation_name, modality_type):
         return None, False
 
     def get_context_representations(self, modality_type):
-        return self._context_representation_operators[modality_type]
+        return self._context_representation_operators.get(modality_type, [])
 
     def get_context_lenghts_for_modality(self, modality_type, statistics):
         if modality_type == ModalityType.AUDIO:
@@ -176,7 +176,7 @@ def get_context_lenghts_for_modality(self, modality_type, statistics):
             modality_type == ModalityType.TIMESERIES
             or modality_type == ModalityType.PHYSIOLOGICAL
         ):
-            window_lengths = [0.05, 0.1, 0.5, 0.75, 1, 2, 5, 10, 30, 60]  # seconds
+            window_lengths = [0.5, 0.75, 1, 2, 5, 10, 30, 60]  # seconds
 
         if modality_type == ModalityType.VIDEO:
             window_lengths = [0.5, 1, 2, 5, 10]  # seconds
@@ -198,7 +198,9 @@ def get_context_lenghts_for_modality(self, modality_type, statistics):
                 math.ceil(statistics.avg_length / length)
                 for length in effective_window_lenghts
             ]
-            return effective_window_lenghts, num_windows
+            return self._drop_windows_below_min_count(
+                effective_window_lenghts, num_windows
+            )
 
         if modality_type == ModalityType.VIDEO:
             max_length_in_seconds = statistics.max_length / statistics.fps
@@ -213,7 +215,22 @@ def get_context_lenghts_for_modality(self, modality_type, statistics):
                 math.ceil(statistics.avg_length / length)
                 for length in effective_window_lenghts
             ]
+            return self._drop_windows_below_min_count(
+                effective_window_lenghts, num_windows
+            )
+
+    MIN_TUNABLE_NUM_WINDOWS = 5
+
+    def _drop_windows_below_min_count(self, effective_window_lenghts, num_windows):
+        filtered = [
+            (length, count)
+            for length, count in zip(effective_window_lenghts, num_windows)
+            if count >= self.MIN_TUNABLE_NUM_WINDOWS and length >= 1
+        ]
+        if not filtered:
             return effective_window_lenghts, num_windows
+        lengths, counts = zip(*filtered)
+        return list(lengths), list(counts)
 
 
 def register_representation(modalities: Union[ModalityType, List[ModalityType]]):

src/main/python/systemds/scuro/drsearch/representation_dag.py

@@ -19,6 +19,8 @@
 #
 # -------------------------------------------------------------
 import copy
+import hashlib
+import json
 from dataclasses import dataclass, field
 from typing import List, Dict, Set, Tuple, Union, Any, Hashable, Optional
 from systemds.scuro.modality.modality import Modality
@@ -112,6 +114,65 @@ def filter_connected_nodes(self, nodes):
 
         return [node for node in nodes if node.node_id in visited]
 
+    def to_spec(self) -> Dict[str, Any]:
+        order = self._topological_order()
+        relabel = {node_id: f"n{i}" for i, node_id in enumerate(order)}
+
+        nodes = []
+        for node_id in order:
+            node = self.get_node_by_id(node_id)
+            operation = getattr(node, "operation", None)
+            nodes.append(
+                {
+                    "id": relabel[node_id],
+                    "op": (
+                        getattr(operation, "__name__", None)
+                        if operation is not None
+                        else None
+                    ),
+                    "params": {
+                        k: _spec_safe(v)
+                        for k, v in sorted((node.parameters or {}).items())
+                    },
+                    "inputs": [relabel[i] for i in node.inputs if i in relabel],
+                    "modality_id": node.modality_id,
+                    "representation_index": node.representation_index,
+                    "aggregation": (
+                        type(node.aggregation).__name__
+                        if node.aggregation is not None
+                        else None
+                    ),
+                }
+            )
+
+        return {
+            "root": relabel.get(self.root_node_id),
+            "nodes": nodes,
+            "representation_names": self.get_represntation_names(),
+        }
+
+    def pipeline_id(self) -> str:
+        blob = json.dumps(self.to_spec(), sort_keys=True, separators=(",", ":"))
+        return hashlib.sha1(blob.encode()).hexdigest()[:16]
+
+    def _topological_order(self) -> List[str]:
+        node_map = {node.node_id: node for node in self.nodes}
+        visited = []
+        seen = set()
+
+        def visit(node_id):
+            if node_id in seen or node_id not in node_map:
+                return
+            seen.add(node_id)
+            for input_id in sorted(node_map[node_id].inputs or []):
+                visit(input_id)
+            visited.append(node_id)
+
+        visit(self.root_node_id)
+        for node_id in sorted(node_map):
+            visit(node_id)
+        return visited
+
     def get_leaf_nodes(self) -> List[str]:
         leaf_nodes = []
         for node in self.nodes:
@@ -414,6 +475,26 @@ def get_leaf_node_id(self) -> str:
         return None
 
 
+def _spec_safe(value):
+    if isinstance(value, (str, int, float, bool)) or value is None:
+        return value
+    if isinstance(value, (list, tuple)):
+        return [_spec_safe(v) for v in value]
+    if isinstance(value, dict):
+        return {
+            str(k): _spec_safe(v)
+            for k, v in sorted(value.items(), key=lambda kv: str(kv[0]))
+        }
+    if hasattr(value, "item") and hasattr(value, "dtype"):
+        try:
+            return value.item()
+        except Exception:
+            pass
+    if isinstance(value, type):
+        return value.__name__
+    return type(value).__name__
+
+
 def get_modality_by_id_and_instance_id(
     modalities: List[Modality], modality_id: int, instance_id: int
 ):
@@ -565,12 +646,13 @@ def __init__(self):
         self.node_to_signature: Dict[str, Hashable] = {}
         self.node_counter = 0
         self.dag_counter = 0
+        self._dag_by_root: Dict[str, RepresentationDag] = {}
 
     def _compute_node_signature(
         self, operation: Any, inputs: List[str], parameters: Dict[str, Any] = None
     ) -> Hashable:
         ip = [self.node_to_signature[inp] for inp in inputs]
-        input_sigs = tuple(sorted(ip)) if inputs else ()
+        input_sigs = tuple(sorted(ip, key=repr)) if inputs else ()
         op_cls = operation().name
         params_items = tuple(sorted((parameters or {}).items()))
         return ("op", op_cls, params_items, input_sigs)
@@ -644,6 +726,11 @@ def create_operation_node(
         )
 
     def build(self, root_node_id: str, dag_id: int = None) -> RepresentationDag:
+        if dag_id is None:
+            memoized = self._dag_by_root.get(root_node_id)
+            if memoized is not None:
+                return memoized
+
         dag = RepresentationDag(
             nodes=self.global_nodes,
             root_node_id=root_node_id,
@@ -652,6 +739,9 @@ def build(self, root_node_id: str, dag_id: int = None) -> RepresentationDag:
         self.dag_counter += 1
         if not dag.validate():
             raise ValueError("Invalid DAG construction")
+
+        if dag_id is None:
+            self._dag_by_root[root_node_id] = dag
         return dag
 
     def get_node(self, node_id: str) -> Optional[RepresentationNode]: