From c0cf6187dbd68ec46d433d8b67d6b59a3be486d8 Mon Sep 17 00:00:00 2001
From: Christina Dionysio <dionysio@tu-berlin.de>
Date: Mon, 17 Aug 2026 22:18:32 +0200
Subject: [PATCH 1/7] remove duplicate DAGs

---
 .../scuro/drsearch/representation_dag.py      |  11 +-
 .../scuro/drsearch/unimodal_optimizer.py      | 139 +++++++++++++-----
 2 files changed, 116 insertions(+), 34 deletions(-)

diff --git a/src/main/python/systemds/scuro/drsearch/representation_dag.py b/src/main/python/systemds/scuro/drsearch/representation_dag.py
index a19c44396fd..c575645fa70 100644
--- a/src/main/python/systemds/scuro/drsearch/representation_dag.py
+++ b/src/main/python/systemds/scuro/drsearch/representation_dag.py
@@ -565,12 +565,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 +645,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 +658,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]:
diff --git a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
index a632c97e973..0076e9e81a6 100644
--- a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
+++ b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
@@ -65,10 +65,12 @@ def __init__(
         checkpoint_every: Optional[int] = 1,
         resume: bool = False,
         max_num_workers: int = -1,
-        enable_checkpointing: bool = True,
+        enable_checkpointing: bool = False,
         enable_execution_profile: bool = False,
         execution_profile_path: Optional[str] = None,
+        window_combination_chains: int = 1,
     ):
+        self.window_combination_chains = window_combination_chains
         self.enable_checkpointing = enable_checkpointing
         self.modalities = modalities
         self.tasks = tasks
@@ -348,10 +350,9 @@ def _process_modality(self, modality, skip_remaining: int = 0, scheduler=None):
         for task_result in task_results:
             local_results.add_task_result(task_result, dags)
         statistics = exec_out["statistics"]
-        for worker_stat in statistics["worker_stats"]:
-            local_results.add_worker_stat(worker_stat, modality.modality_id)
-        for node_stat in statistics["node_stats"]:
-            local_results.add_node_stat(node_stat, modality.modality_id)
+
+        local_results.add_worker_stat(statistics["worker_stats"], modality.modality_id)
+        local_results.add_node_stat(statistics["node_stats"], modality.modality_id)
 
         if self.save_all_results:
             timestr = time.strftime("%Y%m%d-%H%M%S")
@@ -401,6 +402,12 @@ def _merge_results(self, local_results):
                 self.operator_performance.results[modality_id][task_name].extend(
                     local_results.results[modality_id][task_name]
                 )
+                self.operator_performance.add_worker_stat(
+                    local_results.worker_stats[modality_id], modality_id
+                )
+                self.operator_performance.add_node_stat(
+                    local_results.node_stats[modality_id], modality_id
+                )
 
     def add_dimensionality_reduction_operators(self, builder, current_node_id):
         dags = []
@@ -483,7 +490,10 @@ def _build_modality_dag(
             not_self_contained_reps = [
                 rep for rep in not_self_contained_reps if rep != operator.__class__
             ]
-            rep_id = current_node_id
+            chain_tips = {
+                combination.__class__: current_node_id
+                for combination in self._combination_operators
+            }
 
             for rep in not_self_contained_reps:
                 other_rep_id = builder.create_operation_node(
@@ -492,9 +502,10 @@ def _build_modality_dag(
                 for combination in self._combination_operators:
                     combine_id = builder.create_operation_node(
                         combination.__class__,
-                        [rep_id, other_rep_id],
+                        [chain_tips[combination.__class__], other_rep_id],
                         combination.get_current_parameters(),
                     )
+                    chain_tips[combination.__class__] = combine_id
                     rep_dag = builder.build(combine_id)
                     dags.append(rep_dag)
                     if modality.modality_type in [
@@ -507,15 +518,6 @@ def _build_modality_dag(
                                 modality, builder, leaf_id, rep_dag, False
                             )
                         )
-                    elif modality.modality_type == ModalityType.TIMESERIES:
-                        dags.extend(
-                            self.temporal_context_operators(
-                                modality,
-                                builder,
-                                leaf_id,
-                            )
-                        )
-                rep_id = combine_id
 
         if rep_dag.nodes[-1].operation().output_modality_type in [
             ModalityType.EMBEDDING
@@ -586,19 +588,27 @@ def default_context_operators(
                     )
                     dags.append(builder.build(context_node_id))
 
-        context_operators = self._get_context_operators(
-            rep_dag.nodes[-1].operation().output_modality_type
-        )
-        for context_op in context_operators:
-            context_node_id = builder.create_operation_node(
-                context_op,
-                [rep_dag.nodes[-1].node_id],
-                context_op().get_current_parameters(),
+        if self._representations_keep_time_axis(modality.modality_type):
+            context_operators = self._get_context_operators(
+                rep_dag.nodes[-1].operation().output_modality_type
             )
-            dags.append(builder.build(context_node_id))
+            for context_op in context_operators:
+                context_node_id = builder.create_operation_node(
+                    context_op,
+                    [rep_dag.nodes[-1].node_id],
+                    context_op().get_current_parameters(),
+                )
+                dags.append(builder.build(context_node_id))
 
         return dags
 
+    @staticmethod
+    def _representations_keep_time_axis(modality_type) -> bool:
+        return modality_type not in (
+            ModalityType.TIMESERIES,
+            ModalityType.PHYSIOLOGICAL,
+        )
+
     def temporal_context_operators(self, modality, builder, leaf_id):
         aggregators = self.operator_registry.get_context_representations(
             modality.modality_type
@@ -610,23 +620,81 @@ def temporal_context_operators(self, modality, builder, leaf_id):
             )
         )
         dags = []
-        for agg in aggregators:
-            for context_operator in context_operators:
-                for window_size, num_window in zip(window_lengths, num_windows):
+        for context_operator in context_operators:
+            for window_size, num_window in zip(window_lengths, num_windows):
+                window_node_ids = []
+                for agg in aggregators:
                     context_operator_instance = context_operator(agg())
-                    if hasattr(context_operator_instance, "num_windows"):
-                        context_operator_instance.num_windows = num_window
-                    elif hasattr(context_operator_instance, "window_size"):
-                        context_operator_instance.window_size = window_size
+                    self._apply_granularity(
+                        context_operator_instance, window_size, num_window
+                    )
                     context_node_id = builder.create_operation_node(
                         context_operator,
                         [leaf_id],
                         context_operator_instance.get_current_parameters(),
                     )
+                    window_node_ids.append(context_node_id)
                     dags.append(builder.build(context_node_id))
 
+                dags.extend(
+                    self.combine_windowed_representations(builder, window_node_ids)
+                )
+
         return dags
 
+    @staticmethod
+    def _apply_granularity(context_operator_instance, window_size, num_window):
+        parameter = getattr(context_operator_instance, "granularity_parameter", None)
+        kind = getattr(context_operator_instance, "granularity_kind", None)
+        if parameter is None or kind not in ("length", "count"):
+            raise ValueError(
+                f"{type(context_operator_instance).__name__} is registered as a "
+                "context operator but does not declare granularity_parameter / "
+                "granularity_kind, so the window-length search cannot vary it."
+            )
+        value = window_size if kind == "length" else num_window
+        setattr(context_operator_instance, parameter, int(value))
+
+    def combine_windowed_representations(self, builder, window_node_ids):
+        dags = []
+        num_chains = min(self.window_combination_chains, len(window_node_ids))
+        if len(window_node_ids) < 2 or num_chains < 1:
+            return dags
+
+        for start in range(num_chains):
+            ordered = window_node_ids[start:] + window_node_ids[:start]
+            for combination in self._combination_operators:
+                parameters = combination.get_current_parameters()
+                if "preserve_leading_axis" not in parameters:
+                    continue
+                parameters["preserve_leading_axis"] = True
+
+                chain_tip = ordered[0]
+                for next_node_id in ordered[1:]:
+                    chain_tip = builder.create_operation_node(
+                        combination.__class__,
+                        [chain_tip, next_node_id],
+                        parameters,
+                    )
+                    summary_id = self._summarize_windows(builder, chain_tip)
+                    dags.append(
+                        builder.build(chain_tip if summary_id is None else summary_id)
+                    )
+        return dags
+
+    def _summarize_windows(self, builder, node_id):
+        if not (self._tasks_require_same_dims and self.expected_dimensions == 1):
+            return None
+
+        agg_operator = AggregatedRepresentation(
+            target_dimensions=self.expected_dimensions, aggregate_leading=True
+        )
+        return builder.create_operation_node(
+            agg_operator.__class__,
+            [node_id],
+            agg_operator.get_current_parameters(),
+        )
+
 
 class UnimodalResults:
     def __init__(
@@ -651,13 +719,18 @@ def __init__(
             self.cache[modality] = {task_name: [] for task_name in self.task_names}
         self.worker_stats = {}
         self.node_stats = {}
+        self._dag_index = None
+        self._dag_index_source = None
 
     def add_task_result(self, task_result: ResultEntry, dags: List[RepresentationDag]):
         dag_id = task_result.dag.dag_id
         task_name = self.task_names[
             task_result.dag.nodes[-1].parameters.get("_task_idx", 0)
         ]
-        task_result.dag = get_dag_by_id(dags, dag_id)
+        if self._dag_index_source is not dags:
+            self._dag_index = {dag.dag_id: dag for dag in dags}
+            self._dag_index_source = dags
+        task_result.dag = self._dag_index.get(dag_id)
         self.results[task_result.dag.nodes[0].modality_id][task_name].append(
             task_result
         )

From 971e22e403228789379ca5398f741e4beae999a0 Mon Sep 17 00:00:00 2001
From: Christina Dionysio <dionysio@tu-berlin.de>
Date: Mon, 17 Aug 2026 22:20:33 +0200
Subject: [PATCH 2/7] adapt window operators to more finegrained window sizes

---
 .../scuro/drsearch/operator_registry.py       |  21 +-
 .../scuro/representations/aggregate.py        |  32 +-
 .../aggregated_representation.py              |  21 +-
 .../representations/window_aggregation.py     | 332 +++++++++++-------
 4 files changed, 281 insertions(+), 125 deletions(-)

diff --git a/src/main/python/systemds/scuro/drsearch/operator_registry.py b/src/main/python/systemds/scuro/drsearch/operator_registry.py
index 4c5641fdd91..ab8fde78a0c 100644
--- a/src/main/python/systemds/scuro/drsearch/operator_registry.py
+++ b/src/main/python/systemds/scuro/drsearch/operator_registry.py
@@ -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:
@@ -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]]):
diff --git a/src/main/python/systemds/scuro/representations/aggregate.py b/src/main/python/systemds/scuro/representations/aggregate.py
index e8a44faa34c..68f42d14e98 100644
--- a/src/main/python/systemds/scuro/representations/aggregate.py
+++ b/src/main/python/systemds/scuro/representations/aggregate.py
@@ -41,11 +41,38 @@ def _min_agg(data, aggregate_dim=0):
     def _sum_agg(data, aggregate_dim=0):
         return np.sum(data, axis=aggregate_dim)
 
+    @staticmethod
+    def _median_agg(data, aggregate_dim=0):
+        return np.median(data, axis=Aggregation._normalize_axis(aggregate_dim))
+
+    @staticmethod
+    def _mode_agg(data, aggregate_dim=0):
+        axis = Aggregation._normalize_axis(aggregate_dim)
+        arr = np.asarray(data)
+        if arr.ndim == 1:
+            values, counts = np.unique(arr, return_counts=True)
+            return values[np.argmax(counts)]
+        moved = np.moveaxis(arr, axis, -1)
+        flat = moved.reshape(-1, moved.shape[-1])
+        modes = np.empty(flat.shape[0], dtype=arr.dtype)
+        for i in range(flat.shape[0]):
+            values, counts = np.unique(flat[i], return_counts=True)
+            modes[i] = values[np.argmax(counts)]
+        return modes.reshape(moved.shape[:-1])
+
+    @staticmethod
+    def _normalize_axis(aggregate_dim):
+        if isinstance(aggregate_dim, tuple):
+            return aggregate_dim[0] if aggregate_dim else 0
+        return aggregate_dim
+
     _aggregation_function = {
         "mean": _mean_agg.__func__,
         "max": _max_agg.__func__,
         "min": _min_agg.__func__,
         "sum": _sum_agg.__func__,
+        "median": _median_agg.__func__,
+        "mode": _mode_agg.__func__,
     }
 
     def __init__(self, aggregation_function="mean", pad_modality=True, params=None):
@@ -71,7 +98,7 @@ def get_current_parameters(self):
             "pad_modality": self.pad_modality,
         }
 
-    def execute(self, modality, aggregate_dim=(0,)):
+    def execute(self, modality, aggregate_dim=(0,), squeeze_singleton=True):
         data = []
         max_len = 0
         for i, instance in enumerate(modality.data):
@@ -83,7 +110,8 @@ def execute(self, modality, aggregate_dim=(0,)):
                 ) and instance.ndim > 2:
                     aggregated_data = instance.flatten()
                 elif (
-                    isinstance(instance, np.ndarray)
+                    squeeze_singleton
+                    and isinstance(instance, np.ndarray)
                     and instance.ndim == 2
                     and instance.shape[1] == 1
                 ):
diff --git a/src/main/python/systemds/scuro/representations/aggregated_representation.py b/src/main/python/systemds/scuro/representations/aggregated_representation.py
index 85744f9209a..b37b1eceda9 100644
--- a/src/main/python/systemds/scuro/representations/aggregated_representation.py
+++ b/src/main/python/systemds/scuro/representations/aggregated_representation.py
@@ -30,7 +30,13 @@
 
 
 class AggregatedRepresentation(Representation):
-    def __init__(self, aggregation="mean", target_dimensions=None, params=None):
+    def __init__(
+        self,
+        aggregation="mean",
+        target_dimensions=None,
+        params=None,
+        aggregate_leading=False,
+    ):
         if params is not None:
             if "aggregation_function_aggregation_function" in params:
                 aggregation = params["aggregation_function_aggregation_function"]
@@ -40,6 +46,7 @@ def __init__(self, aggregation="mean", target_dimensions=None, params=None):
                 aggregation = params["aggregation"]
             if "target_dimensions" in params:
                 target_dimensions = params["target_dimensions"]
+            aggregate_leading = params.get("aggregate_leading", aggregate_leading)
         parameters = {
             "aggregation": list(Aggregation().get_aggregation_functions()),
         }
@@ -48,13 +55,14 @@ def __init__(self, aggregation="mean", target_dimensions=None, params=None):
         self.aggregation = Aggregation(aggregation)
         self.self_contained = True
         self.target_dimensions = target_dimensions
+        self.aggregate_leading = bool(aggregate_leading)
         self.data_type = np.float32
 
     def get_output_stats(self, input_stats: RepresentationStats) -> RepresentationStats:
         input_shape = list(copy.deepcopy(input_stats.output_shape))
         if self.target_dimensions is not None:
             while len(input_shape) > self.target_dimensions:
-                input_shape.pop()
+                input_shape.pop(0 if self.aggregate_leading else -1)
         out_shape = tuple(input_shape)
         self.stats = RepresentationStats(
             input_stats.num_instances,
@@ -99,6 +107,10 @@ def transform(self, modality):
 
             if len(input_dimensions) == self.target_dimensions:
                 return modality
+            elif self.aggregate_leading:
+                aggregate_dim = tuple(
+                    range(len(input_dimensions) - self.target_dimensions)
+                )
             else:
                 i = len(input_dimensions) - 1
                 aggregate_dim = ()
@@ -107,7 +119,9 @@ def transform(self, modality):
                     i -= 1
                     input_dimensions = input_dimensions[:-1]
 
-        aggregated_data = self.aggregation.execute(modality, aggregate_dim)
+        aggregated_data = self.aggregation.execute(
+            modality, aggregate_dim, squeeze_singleton=not self.aggregate_leading
+        )
 
         aggregated_modality.data = aggregated_data
         end = time.perf_counter()
@@ -122,6 +136,7 @@ def get_current_parameters(self):
             current_params[f"aggregation_function_{key}"] = value
         current_params["self_contained"] = self.self_contained
         current_params["target_dimensions"] = self.target_dimensions
+        current_params["aggregate_leading"] = self.aggregate_leading
         return current_params
 
     def assert_output_stats(self, aggregated_data):
diff --git a/src/main/python/systemds/scuro/representations/window_aggregation.py b/src/main/python/systemds/scuro/representations/window_aggregation.py
index 52a17401959..a9a1f1eb41b 100644
--- a/src/main/python/systemds/scuro/representations/window_aggregation.py
+++ b/src/main/python/systemds/scuro/representations/window_aggregation.py
@@ -29,8 +29,10 @@
 from systemds.scuro.representations.aggregate import Aggregation
 from systemds.scuro.representations.context import Context
 from systemds.scuro.representations.representation import (
+    NDARRAY_OBJECT_OVERHEAD_BYTES,
     Representation,
     RepresentationStats,
+    stats_itemsize,
 )
 
 
@@ -65,7 +67,59 @@ def instantiate_nested_aggregation(agg_cls, nested):
     return agg_cls(**filtered)
 
 
+def _pad_stack(arrays):
+    arrays = [np.asarray(a) for a in arrays]
+    if len({a.shape for a in arrays}) == 1:
+        return np.stack(arrays)
+
+    ndim = max(a.ndim for a in arrays)
+    arrays = [a.reshape((1,) * (ndim - a.ndim) + a.shape) for a in arrays]
+    target_shape = tuple(max(a.shape[d] for a in arrays) for d in range(ndim))
+    stacked = np.zeros((len(arrays), *target_shape), dtype=arrays[0].dtype)
+    for i, a in enumerate(arrays):
+        slices = tuple(slice(0, s) for s in a.shape)
+        stacked[(i, *slices)] = a
+    return stacked
+
+
+def _append_tail_row(full_result, tail_result):
+    full_result = np.asarray(full_result)
+    tail_result = np.asarray(tail_result)
+    target_shape = full_result.shape[1:]
+    if tail_result.shape == target_shape:
+        tail_row = tail_result
+    else:
+        tail_row = np.zeros(target_shape, dtype=full_result.dtype)
+        slices = tuple(
+            slice(0, min(d, s)) for d, s in zip(target_shape, tail_result.shape)
+        )
+        tail_row[slices] = tail_result[slices]
+    return np.concatenate([full_result, tail_row[None, ...]])
+
+
+def resolve_aggregation_function(aggregation_function, params):
+    if params is None:
+        return aggregation_function
+    if isinstance(params.get("aggregation_function"), (Aggregation, Representation)):
+        return params["aggregation_function"]
+
+    nested_agg = {
+        key[len("aggregation_function_") :]: value
+        for key, value in params.items()
+        if key.startswith("aggregation_function_")
+    }
+    agg_value = params.get("aggregation_function")
+    if nested_agg and inspect.isclass(agg_value):
+        return instantiate_nested_aggregation(agg_value, nested_agg)
+    if inspect.isclass(agg_value):
+        return agg_value()
+    return params.get("aggregation_function", aggregation_function)
+
+
 class Window(Context):
+    granularity_parameter = None
+    granularity_kind = None  # "length" | "count"
+
     def __init__(self, name, aggregation_function):
         self.aggregation_function = aggregation_function
         parameters = {}
@@ -143,11 +197,26 @@ def _shape_numel(shape):
     def _rest_numel(shape):
         return int(np.prod(shape[1:])) if len(shape) > 1 else 1
 
+    def _per_window_feature_shape(self, approx_window_size):
+        windowed_input_stats = RepresentationStats(1, (approx_window_size,))
+        feat_shape = self.aggregation_function.get_output_stats(
+            windowed_input_stats
+        ).output_shape
+        return () if self._shape_numel(feat_shape) <= 1 else tuple(feat_shape)
+
 
 @register_context_operator(
-    [ModalityType.TIMESERIES, ModalityType.AUDIO, ModalityType.EMBEDDING]
+    [
+        ModalityType.TIMESERIES,
+        ModalityType.PHYSIOLOGICAL,
+        ModalityType.AUDIO,
+        ModalityType.EMBEDDING,
+    ]
 )
 class WindowAggregation(Window):
+    granularity_parameter = "window_size"
+    granularity_kind = "length"
+
     def __init__(
         self,
         aggregation_function="mean",
@@ -163,30 +232,12 @@ def __init__(
             window_size_set = True
 
         if params is not None:
-            if isinstance(
-                params.get("aggregation_function"), (Aggregation, Representation)
-            ):
-                aggregation_function = params["aggregation_function"]
-                if hasattr(aggregation_function, "window_size"):
-                    window_size = aggregation_function.window_size
-                    window_size_set = True
-            else:
-                nested_agg = {
-                    key[len("aggregation_function_") :]: value
-                    for key, value in params.items()
-                    if key.startswith("aggregation_function_")
-                }
-                agg_value = params.get("aggregation_function")
-                if nested_agg and inspect.isclass(agg_value):
-                    aggregation_function = instantiate_nested_aggregation(
-                        agg_value, nested_agg
-                    )
-                elif inspect.isclass(agg_value):
-                    aggregation_function = agg_value()
-                else:
-                    aggregation_function = params.get(
-                        "aggregation_function", aggregation_function
-                    )
+            aggregation_function = resolve_aggregation_function(
+                aggregation_function, params
+            )
+            if hasattr(aggregation_function, "window_size"):
+                window_size = aggregation_function.window_size
+                window_size_set = True
 
             window_size = params["window_size"] if not window_size_set else window_size
             pad = params.get("pad", True)
@@ -197,13 +248,8 @@ def __init__(
 
     def get_output_stats(self, input_stats: RepresentationStats) -> tuple:
         if not isinstance(self.aggregation_function, Aggregation):
-            windowed_input_stats = RepresentationStats(
-                input_stats.num_instances, (self.window_size,)
-            )
-            in_shape = self.aggregation_function.get_output_stats(
-                windowed_input_stats
-            ).output_shape
-            in_shape = (input_stats.output_shape[0], *in_shape)
+            feat_shape = self._per_window_feature_shape(self.window_size)
+            in_shape = (input_stats.output_shape[0], *feat_shape)
         else:
             in_shape = tuple(int(s) for s in input_stats.output_shape)
         if len(in_shape) == 1:
@@ -226,40 +272,35 @@ def estimate_output_memory_bytes(self, input_stats: RepresentationStats) -> int:
 
         out_seq_len = math.ceil(in_shape[0] / self.window_size)
         output_bytes = out_seq_len * self._rest_numel(in_shape)
-        return (
-            input_stats.num_instances * output_bytes * np.dtype(self.data_type).itemsize
-        )
+        return input_stats.num_instances * output_bytes * stats_itemsize(input_stats)
 
     def estimate_peak_memory_bytes(self, input_stats: RepresentationStats) -> dict:
         in_shape = tuple(int(s) for s in input_stats.output_shape)
         if len(in_shape) == 0:
             return {"cpu_peak_bytes": 0, "gpu_peak_bytes": 0}
 
-        out_stats = self.get_output_stats(input_stats)
-        out_shape = out_stats.output_shape
-        output_bytes = (
-            input_stats.num_instances
-            * np.prod(out_shape)
-            * np.dtype(self.data_type).itemsize
-        )
-
         effective_seq_len = in_shape[0]
         in_numel = effective_seq_len * self._rest_numel(in_shape)
         output_bytes = self.estimate_output_memory_bytes(input_stats)
-        one_instance_bytes = in_numel * np.dtype(self.data_type).itemsize
+        one_instance_bytes = in_numel * stats_itemsize(input_stats)
         input_bytes = one_instance_bytes * input_stats.num_instances
 
-        output_transient = output_bytes
+        list_bytes = output_bytes + input_stats.num_instances * (
+            NDARRAY_OBJECT_OVERHEAD_BYTES
+        )
 
-        pad_overhead = 0
+        pad_bytes = 0
         if getattr(self, "pad", False):
             out_seq_len = math.ceil(in_shape[0] / self.window_size)
-            pad_overhead = int(input_stats.num_instances * out_seq_len * 8)
+            padded_elems = (
+                input_stats.num_instances * out_seq_len * self._rest_numel(in_shape)
+            )
+            pad_bytes = int(
+                padded_elems * np.dtype(np.float64).itemsize
+                + padded_elems * stats_itemsize(input_stats)
+            )
 
-        cpu_peak = int(
-            (input_bytes + output_bytes + output_transient + pad_overhead) * 1.15
-            + 16 * 1024 * 1024
-        )
+        cpu_peak = int((input_bytes + list_bytes + pad_bytes) * 1.15 + 16 * 1024 * 1024)
         return {"cpu_peak_bytes": cpu_peak, "gpu_peak_bytes": 0}
 
     def execute(self, modality):
@@ -317,34 +358,51 @@ def window_aggregate_single_level(self, instance, new_length):
 
         arr = np.asarray(instance)
         cut_length = (new_length - 1) * self.window_size
-
+        tail = arr[cut_length:]
+        sig = inspect.signature(self.aggregation_function.compute_feature)
+        if new_length <= 1:
+            if not tail.size:
+                raise ValueError(
+                    "Cannot window-aggregate an empty instance "
+                    f"(window_size={self.window_size})."
+                )
+            if tail.shape[0] < self.window_size:
+                pad_len = self.window_size - tail.shape[0]
+                if tail.ndim == 1:
+                    tail = np.pad(tail, (0, pad_len), mode="constant")
+                else:
+                    pad_width = [(0, 0)] * tail.ndim
+                    pad_width[0] = (0, pad_len)
+                    tail = np.pad(tail, pad_width=pad_width, mode="constant")
+            if "axis" in sig.parameters:
+                return np.array([self.aggregation_function.compute_feature(tail)])
+            tail_result = self.aggregation_function.compute_feature(tail)
+            return (
+                tail_result[None, :]
+                if tail_result.ndim > 0
+                else np.array([tail_result])
+            )
         full_batches = arr[:cut_length].reshape(
             new_length - 1, self.window_size, *arr.shape[1:]
         )
-        tail = arr[cut_length:]
 
-        sig = inspect.signature(self.aggregation_function.compute_feature)
         if "axis" in sig.parameters:
             full_result = self.aggregation_function.compute_feature(
                 full_batches, axis=1
             )
             if tail.size:
                 tail_result = self.aggregation_function.compute_feature(tail)
-                full_result = np.concatenate([full_result, np.array([tail_result])])
+                full_result = _append_tail_row(full_result, tail_result)
         else:
-            full_result = self.aggregation_function.compute_feature(full_batches)
+            full_result = np.stack(
+                [
+                    self.aggregation_function.compute_feature(full_batches[i])
+                    for i in range(full_batches.shape[0])
+                ]
+            )
             if tail.size:
                 tail_result = self.aggregation_function.compute_feature(tail)
-                if tail_result.shape == full_result.shape[1:]:
-                    tail_row = tail_result
-                else:
-                    tail_row = np.zeros_like(full_result[0])
-                    slices = tuple(
-                        slice(0, min(d, s))
-                        for d, s in zip(tail_row.shape, tail_result.shape)
-                    )
-                    tail_row[slices] = tail_result[slices]
-                full_result = np.concatenate([full_result, tail_row[None, :]])
+                full_result = _append_tail_row(full_result, tail_result)
 
         return full_result
 
@@ -359,27 +417,42 @@ def window_aggregate_nested_level(self, instance, new_length):
 
 
 @register_context_operator(
-    [ModalityType.TIMESERIES, ModalityType.AUDIO, ModalityType.EMBEDDING]
+    [
+        ModalityType.TIMESERIES,
+        ModalityType.PHYSIOLOGICAL,
+        ModalityType.AUDIO,
+        ModalityType.EMBEDDING,
+    ]
 )
 class StaticWindow(Window):
+    granularity_parameter = "num_windows"
+    granularity_kind = "count"
+
     def __init__(self, aggregation_function="mean", num_windows=100, params=None):
-        super().__init__("StaticWindow", aggregation_function)
         if params is not None:
-            num_windows = params.get("num_windows", 100)
+            aggregation_function = resolve_aggregation_function(
+                aggregation_function, params
+            )
+            num_windows = params.get("num_windows", num_windows)
 
-        self.parameters["num_windows"] = (5, num_windows)
+        super().__init__("StaticWindow", aggregation_function)
+        self.parameters["num_windows"] = (min(5, num_windows), max(5, num_windows))
         self.num_windows = int(num_windows)
 
+    def _feature_shape(self, in_shape):
+        if isinstance(self.aggregation_function, Aggregation):
+            return in_shape[1:]
+        approx_window_size = (
+            max(1, int(in_shape[0] / self.num_windows)) if in_shape else 1
+        )
+        return self._per_window_feature_shape(approx_window_size)
+
     def get_output_stats(self, input_stats: RepresentationStats) -> tuple:
         in_shape = tuple(int(s) for s in input_stats.output_shape)
-        if len(in_shape) <= 1:
-            self.stats = RepresentationStats(
-                input_stats.num_instances, (self.num_windows,)
-            )
-        else:
-            self.stats = RepresentationStats(
-                input_stats.num_instances, (self.num_windows, *in_shape[1:])
-            )
+        feat_shape = self._feature_shape(in_shape)
+        self.stats = RepresentationStats(
+            input_stats.num_instances, (self.num_windows, *feat_shape)
+        )
         self.stats.output_shape_is_known = input_stats.output_shape_is_known
 
         return self.stats
@@ -389,11 +462,9 @@ def estimate_output_memory_bytes(self, input_stats: RepresentationStats) -> int:
         if len(in_shape) == 0:
             return 0
 
-        out_seq_len = self.num_windows
-        output_bytes = out_seq_len * self._rest_numel(in_shape)
-        return (
-            input_stats.num_instances * output_bytes * np.dtype(self.data_type).itemsize
-        )
+        out_shape = self.get_output_stats(input_stats).output_shape
+        out_numel = int(np.prod(out_shape)) if len(out_shape) > 0 else 1
+        return input_stats.num_instances * out_numel * stats_itemsize(input_stats)
 
     def estimate_peak_memory_bytes(self, input_stats: RepresentationStats) -> dict:
         in_shape = tuple(int(s) for s in input_stats.output_shape)
@@ -401,7 +472,7 @@ def estimate_peak_memory_bytes(self, input_stats: RepresentationStats) -> dict:
             return {"cpu_peak_bytes": 0, "gpu_peak_bytes": 0}
         effective_seq_len = in_shape[0]
         in_numel = effective_seq_len * self._rest_numel(in_shape)
-        one_instance_bytes = in_numel * np.dtype(self.data_type).itemsize
+        one_instance_bytes = in_numel * stats_itemsize(input_stats)
         input_bytes = one_instance_bytes * input_stats.num_instances
         output_bytes = self.estimate_output_memory_bytes(input_stats)
         output_transient = output_bytes
@@ -428,34 +499,71 @@ def execute(self, modality):
             if "axis" in sig.parameters:
                 f = self.aggregation_function.compute_feature(full_batches, axis=1)
             else:
-                f = self.aggregation_function.compute_feature(full_batches)
+                f = np.stack(
+                    [
+                        self.aggregation_function.compute_feature(full_batches[i])
+                        for i in range(full_batches.shape[0])
+                    ]
+                )
 
             windowed_data.append(f)
-        windowed_data = np.array(windowed_data)
+        windowed_data = _pad_stack(windowed_data)
         return windowed_data
 
 
 @register_context_operator(
-    [ModalityType.TIMESERIES, ModalityType.AUDIO, ModalityType.EMBEDDING]
+    [
+        ModalityType.TIMESERIES,
+        ModalityType.PHYSIOLOGICAL,
+        ModalityType.AUDIO,
+        ModalityType.EMBEDDING,
+    ]
 )
 class DynamicWindow(Window):
+    granularity_parameter = "num_windows"
+    granularity_kind = "count"
+
     def __init__(self, aggregation_function="mean", num_windows=100, params=None):
-        super().__init__("DynamicWindow", aggregation_function)
         if params is not None:
-            num_windows = params.get("num_windows", 100)
-        self.parameters["num_windows"] = (5, num_windows)
+            aggregation_function = resolve_aggregation_function(
+                aggregation_function, params
+            )
+            num_windows = params.get("num_windows", num_windows)
+        super().__init__("DynamicWindow", aggregation_function)
+        self.parameters["num_windows"] = (min(5, num_windows), max(5, num_windows))
         self.num_windows = int(num_windows)
 
+    def _effective_num_windows(self, signal_length: int) -> int:
+        if signal_length <= 0:
+            return max(1, self.num_windows)
+        return max(1, min(self.num_windows, int(signal_length)))
+
+    def _window_sizes(self, signal_length: int) -> np.ndarray:
+        num_windows = self._effective_num_windows(signal_length)
+        length = max(int(signal_length), num_windows)
+        weights = np.geomspace(4, 256, num=num_windows)
+        weights = weights / np.sum(weights)
+
+        sizes = 1 + (weights * (length - num_windows)).astype(int)
+        sizes[-1] += length - int(sizes.sum())
+        return sizes
+
+    def _feature_shape(self, in_shape):
+        if isinstance(self.aggregation_function, Aggregation):
+            return in_shape[1:]
+        length = in_shape[0] if in_shape else 0
+        approx_window_size = (
+            max(1, int(length / self._effective_num_windows(length))) if in_shape else 1
+        )
+        return self._per_window_feature_shape(approx_window_size)
+
     def get_output_stats(self, input_stats: RepresentationStats) -> tuple:
         in_shape = tuple(int(s) for s in input_stats.output_shape)
-        if len(in_shape) <= 1:
-            self.stats = RepresentationStats(
-                input_stats.num_instances, (self.num_windows,)
-            )
-        else:
-            self.stats = RepresentationStats(
-                input_stats.num_instances, (self.num_windows, *in_shape[1:])
-            )
+        feat_shape = self._feature_shape(in_shape)
+        num_windows = self._effective_num_windows(in_shape[0] if in_shape else 0)
+        self.stats = RepresentationStats(
+            input_stats.num_instances, (num_windows, *feat_shape)
+        )
         self.stats.output_shape_is_known = input_stats.output_shape_is_known
         return self.stats
 
@@ -464,11 +572,9 @@ def estimate_output_memory_bytes(self, input_stats: RepresentationStats) -> int:
         if len(in_shape) == 0:
             return 0
 
-        out_seq_len = self.num_windows
-        output_bytes = out_seq_len * self._rest_numel(in_shape)
-        return (
-            input_stats.num_instances * output_bytes * np.dtype(self.data_type).itemsize
-        )
+        out_shape = self.get_output_stats(input_stats).output_shape
+        out_numel = int(np.prod(out_shape)) if len(out_shape) > 0 else 1
+        return input_stats.num_instances * out_numel * stats_itemsize(input_stats)
 
     def estimate_peak_memory_bytes(self, input_stats: RepresentationStats) -> dict:
         in_shape = tuple(int(s) for s in input_stats.output_shape)
@@ -477,7 +583,7 @@ def estimate_peak_memory_bytes(self, input_stats: RepresentationStats) -> dict:
         effective_seq_len = in_shape[0]
         in_numel = effective_seq_len * self._rest_numel(in_shape)
         output_bytes = self.estimate_output_memory_bytes(input_stats)
-        one_instance_bytes = in_numel * np.dtype(self.data_type).itemsize
+        one_instance_bytes = in_numel * stats_itemsize(input_stats)
         cpu_peak = (
             output_bytes * 2
             + one_instance_bytes * input_stats.num_instances
@@ -489,26 +595,16 @@ def execute(self, modality):
         windowed_data = []
 
         for instance in modality.data:
-            N = len(instance)
-            weights = np.geomspace(4, 256, num=self.num_windows)
-            weights = weights / np.sum(weights)
-            window_sizes = (weights * N).astype(int)
-            window_sizes[-1] += N - np.sum(window_sizes)
-            indices = np.cumsum(window_sizes)
+            indices = np.cumsum(self._window_sizes(len(instance)))
             output = []
             start = 0
             for end in indices:
                 window = instance[start:end]
                 window.setflags(write=False)
-                val = (
-                    self.aggregation_function.compute_feature(window)
-                    if len(window) > 0
-                    else np.zeros_like(instance[0])
-                )
-                output.append(val)
+                output.append(self.aggregation_function.compute_feature(window))
                 start = end
 
-            windowed_data.append(output)
-        windowed_data = np.array(windowed_data)
+            windowed_data.append(_pad_stack(output))
+        windowed_data = _pad_stack(windowed_data)
         self.assert_output_stats(windowed_data)
         return windowed_data

From 2ab6e07f618129612bb7f31d06fc49d918435998 Mon Sep 17 00:00:00 2001
From: Christina Dionysio <dionysio@tu-berlin.de>
Date: Wed, 26 Aug 2026 08:09:27 +0200
Subject: [PATCH 3/7] add more finegrained statistics

---
 .../scuro/drsearch/hyperparameter_tuner.py    |  48 +++-
 .../scuro/drsearch/operator_registry.py       |   4 +-
 .../scuro/drsearch/representation_dag.py      |  81 +++++++
 .../scuro/drsearch/unimodal_optimizer.py      | 126 +++++++++--
 .../systemds/scuro/modality/modality.py       | 208 ++++++++++--------
 .../scuro/modality/unimodal_modality.py       |  38 +++-
 6 files changed, 384 insertions(+), 121 deletions(-)

diff --git a/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py b/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py
index a62b990fa99..094a4b9585e 100644
--- a/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py
+++ b/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
diff --git a/src/main/python/systemds/scuro/drsearch/operator_registry.py b/src/main/python/systemds/scuro/drsearch/operator_registry.py
index ab8fde78a0c..7a80aafa913 100644
--- a/src/main/python/systemds/scuro/drsearch/operator_registry.py
+++ b/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]
 
@@ -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
diff --git a/src/main/python/systemds/scuro/drsearch/representation_dag.py b/src/main/python/systemds/scuro/drsearch/representation_dag.py
index c575645fa70..ad7f174acfa 100644
--- a/src/main/python/systemds/scuro/drsearch/representation_dag.py
+++ b/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
 ):
diff --git a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
index 0076e9e81a6..1b24e6ef192 100644
--- a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
+++ b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
@@ -19,23 +19,23 @@
 #
 # -------------------------------------------------------------
 import copy
+import math
 import pickle
-import csv
-from pathlib import Path
 import time
 from concurrent.futures import ProcessPoolExecutor, as_completed
-from dataclasses import dataclass
 import multiprocessing as mp
 from typing import List, Any, Optional, Dict
 from functools import lru_cache
 
 from systemds.scuro import ModalityType
 from systemds.scuro.drsearch.node_executor import NodeExecutor, ResultEntry
-from systemds.scuro.drsearch.ranking import rank_by_tradeoff
+from systemds.scuro.representations.representation import RepresentationStats
+from systemds.scuro.drsearch.ranking import rank_by_robustness, rank_by_tradeoff
 from systemds.scuro.drsearch.task import PerformanceMeasure
 from systemds.scuro.representations.concatenation import Concatenation
 from systemds.scuro.representations.hadamard import Hadamard
 from systemds.scuro.representations.sum import Sum
+from systemds.scuro.representations.average import Average
 from systemds.scuro.representations.aggregated_representation import (
     AggregatedRepresentation,
 )
@@ -66,10 +66,9 @@ def __init__(
         resume: bool = False,
         max_num_workers: int = -1,
         enable_checkpointing: bool = False,
-        enable_execution_profile: bool = False,
-        execution_profile_path: Optional[str] = None,
         window_combination_chains: int = 1,
     ):
+        self._node_stats: Dict[str, Any] = {}
         self.window_combination_chains = window_combination_chains
         self.enable_checkpointing = enable_checkpointing
         self.modalities = modalities
@@ -94,13 +93,13 @@ def __init__(
         }
 
         self.debug = debug
+        self._search_start = time.perf_counter()
+        self._search_start_unix = time.time()
 
         self.operator_registry = Registry()
         self.operator_performance = UnimodalResults(
             modalities, tasks, debug, True, k, self.metric_name
         )
-        self.enable_execution_profile = enable_execution_profile
-        self.execution_profile_path = execution_profile_path
         self._tasks_require_same_dims = True
         self.expected_dimensions = tasks[0].expected_dim
 
@@ -142,6 +141,22 @@ def store_results(self, file_name=None):
         with open(file_name, "wb") as f:
             pickle.dump(self.operator_performance.results, f)
 
+        stats_file_name = file_name.replace(".pkl", "_exec_stats.pkl")
+        if stats_file_name == file_name:
+            stats_file_name = file_name + "_exec_stats.pkl"
+        with open(stats_file_name, "wb") as f:
+            pickle.dump(
+                {
+                    "worker_stats": self.operator_performance.worker_stats,
+                    "node_stats": self.operator_performance.node_stats,
+                    "reuse_stats": self.operator_performance.reuse_stats,
+                    "wall_clock_s": self.operator_performance.wall_clock_s,
+                    "search_start_unix": self._search_start_unix,
+                    "max_num_workers": self.max_num_workers,
+                },
+                f,
+            )
+
     def store_cache(self, file_name=None):
         if file_name is None:
             import time
@@ -341,6 +356,7 @@ def _process_modality(self, modality, skip_remaining: int = 0, scheduler=None):
             max_num_workers=self.max_num_workers,
             result_path=self.result_path,
             enable_checkpointing=self.enable_checkpointing,
+            search_start=self._search_start,
         )
         start_time = time.perf_counter()
         exec_out = node_executor.run()
@@ -353,6 +369,8 @@ def _process_modality(self, modality, skip_remaining: int = 0, scheduler=None):
 
         local_results.add_worker_stat(statistics["worker_stats"], modality.modality_id)
         local_results.add_node_stat(statistics["node_stats"], modality.modality_id)
+        local_results.add_reuse_stat(statistics.get("reuse", {}), modality.modality_id)
+        local_results.wall_clock_s[modality.modality_id] = end_time - start_time
 
         if self.save_all_results:
             timestr = time.strftime("%Y%m%d-%H%M%S")
@@ -362,12 +380,29 @@ def _process_modality(self, modality, skip_remaining: int = 0, scheduler=None):
 
         return local_results, end_time - start_time
 
+    def _window_input_stats(self, modality: Modality, window_length: int):
+        modality_stats = modality.get_output_stats()
+        return RepresentationStats(
+            modality_stats.num_instances,
+            (int(window_length),),
+            output_shape_is_known=modality_stats.output_shape_is_known,
+            dtype=getattr(modality_stats, "dtype", None),
+            sampling_rate=getattr(modality_stats, "sampling_rate", None),
+        )
+
+    @staticmethod
+    def _effective_window_length(context_operator, window_size, num_window, signal_len):
+        if context_operator.granularity_kind == "count":
+            return max(1, int(math.ceil(signal_len / max(1, int(num_window)))))
+        return max(1, int(window_size))
+
     def _build_execution_dags_for_modality(
         self, modality: Modality, skip_remaining: int = 0
     ) -> tuple:
         modality_specific_operators = self._get_modality_operators(
             modality.modality_type
         )
+        self._node_stats = {}
         dags = []
         for operator in modality_specific_operators:
             dags.extend(self._build_modality_dag(modality, operator()))
@@ -408,6 +443,12 @@ def _merge_results(self, local_results):
                 self.operator_performance.add_node_stat(
                     local_results.node_stats[modality_id], modality_id
                 )
+                self.operator_performance.add_reuse_stat(
+                    local_results.reuse_stats.get(modality_id, {}), modality_id
+                )
+                self.operator_performance.wall_clock_s[modality_id] = (
+                    local_results.wall_clock_s.get(modality_id, 0.0)
+                )
 
     def add_dimensionality_reduction_operators(self, builder, current_node_id):
         dags = []
@@ -530,22 +571,35 @@ def _build_modality_dag(
 
         return dags
 
-    def _aggregation_needed(self, dag: RepresentationDag) -> bool:
-        input_stats = {}
+    def _node_output_stats(self, dag: RepresentationDag) -> Dict[str, Any]:
+        stats = self._node_stats
         for modality in self.modalities:
             if modality.modality_id == dag.nodes[0].modality_id:
-                input_stats[dag.nodes[0].node_id] = modality.stats
+                stats.setdefault(dag.nodes[0].node_id, modality.stats)
                 break
         for node in dag.nodes[1:]:
+            if node.node_id in stats or node.operation is None:
+                continue
             previous_stats = [
-                input_stats.get(input_node_id, None) for input_node_id in node.inputs
+                stats.get(input_node_id, None) for input_node_id in node.inputs
             ]
-            current_stats = node.operation(params=node.parameters).get_output_stats(
+            stats[node.node_id] = node.operation(
+                params=node.parameters
+            ).get_output_stats(
                 previous_stats if len(previous_stats) > 1 else previous_stats[0]
             )
-            input_stats[node.node_id] = current_stats
 
-        return len(input_stats.get(dag.root_node_id, None).output_shape) > 1
+        return stats
+
+    def _dag_output_length(self, dag: RepresentationDag) -> Optional[int]:
+        stats = self._node_output_stats(dag).get(dag.root_node_id, None)
+        output_shape = getattr(stats, "output_shape", None)
+        if not output_shape:
+            return None
+        return int(output_shape[0])
+
+    def _aggregation_needed(self, dag: RepresentationDag) -> bool:
+        return len(self._node_output_stats(dag)[dag.root_node_id].output_shape) > 1
 
     def add_aggregation_operator(self, builder, dags):
         new_dags = []
@@ -589,19 +643,41 @@ def default_context_operators(
                     dags.append(builder.build(context_node_id))
 
         if self._representations_keep_time_axis(modality.modality_type):
+            rep_root = rep_dag.get_node_by_id(rep_dag.root_node_id)
             context_operators = self._get_context_operators(
-                rep_dag.nodes[-1].operation().output_modality_type
+                rep_root.operation().output_modality_type
             )
+            output_length = self._dag_output_length(rep_dag)
             for context_op in context_operators:
+                context_operator_instance = context_op()
+                if not self._size_context_operator(
+                    context_operator_instance, output_length
+                ):
+                    continue
                 context_node_id = builder.create_operation_node(
                     context_op,
-                    [rep_dag.nodes[-1].node_id],
-                    context_op().get_current_parameters(),
+                    [rep_root.node_id],
+                    context_operator_instance.get_current_parameters(),
                 )
                 dags.append(builder.build(context_node_id))
 
         return dags
 
+    def _size_context_operator(self, context_operator_instance, output_length) -> bool:
+        parameter = getattr(context_operator_instance, "granularity_parameter", None)
+        kind = getattr(context_operator_instance, "granularity_kind", None)
+        if parameter is None or kind not in ("length", "count"):
+            return True
+        if output_length is None or output_length < 4:
+            return False
+        current = int(getattr(context_operator_instance, parameter))
+        setattr(
+            context_operator_instance,
+            parameter,
+            max(2, min(current, output_length // 2)),
+        )
+        return True
+
     @staticmethod
     def _representations_keep_time_axis(modality_type) -> bool:
         return modality_type not in (
@@ -719,6 +795,10 @@ def __init__(
             self.cache[modality] = {task_name: [] for task_name in self.task_names}
         self.worker_stats = {}
         self.node_stats = {}
+        self.reuse_stats = {}
+        self.wall_clock_s = {}
+        self._eval_counter = 0
+        self._search_start = time.perf_counter()
         self._dag_index = None
         self._dag_index_source = None
 
@@ -765,10 +845,17 @@ def add_result(
             train_score=scores[0].average_scores,
             val_score=scores[1].average_scores,
             test_score=scores[2].average_scores,
+            train_fold_scores=scores[0].fold_scores(),
+            val_fold_scores=scores[1].fold_scores(),
+            test_fold_scores=scores[2].fold_scores(),
             representation_time=transform_time,
             task_time=task_time,
             dag=dag,
+            eval_index=self._eval_counter,
+            t_since_search_start_s=time.perf_counter() - self._search_start,
+            t_eval_end_unix=time.time(),
         )
+        self._eval_counter += 1
 
         scores = [
             -item.val_score[self.metric_name]
@@ -858,6 +945,9 @@ def get_k_best_results(
     def add_worker_stat(self, worker_stats, modality_id):
         self.worker_stats[modality_id] = worker_stats
 
+    def add_reuse_stat(self, reuse_stats, modality_id):
+        self.reuse_stats[modality_id] = reuse_stats
+
     def add_node_stat(self, node_stats, modality_id):
         self.node_stats[modality_id] = node_stats
 
diff --git a/src/main/python/systemds/scuro/modality/modality.py b/src/main/python/systemds/scuro/modality/modality.py
index 477e4e45f35..a0d1e36377d 100644
--- a/src/main/python/systemds/scuro/modality/modality.py
+++ b/src/main/python/systemds/scuro/modality/modality.py
@@ -84,13 +84,19 @@ def update_metadata(self):
         """
         Updates the metadata of the modality (i.e.: updates timestamps)
         """
-        if (
-            not self.has_metadata()
-            or not self.has_data()
-            or len(self.data) < len(self.metadata)
-        ):
+        if not self.has_metadata() or not self.has_data():
+            return
+
+        num_instances = len(self.data)
+        if num_instances < len(self.metadata):
             return
 
+        while len(self.metadata) < num_instances:
+            template = (
+                selective_copy_metadata(self.metadata[0]) if self.metadata else {}
+            )
+            self.metadata.append(template)
+
         for i, md_v in enumerate(self.metadata):
             md_v = selective_copy_metadata(md_v)
             updated_md = self.modality_type.update_metadata(md_v, self.data[i])
@@ -132,88 +138,118 @@ def flatten(self, padding=False):
         self.data = np.array(data)
         return self
 
-    def pad(self, value=0, max_len=None):
-        try:
-            if max_len is None:
-                result = np.array(self.data)
-            elif isinstance(self.data, np.ndarray) and self.data.shape[1] == max_len:
-                result = self.data
-            else:
-                raise "Needs padding to max_len"
-        except:
-            first = self.data[0]
-            if isinstance(first, np.ndarray) and first.ndim == 3:
-                maxlen = (
-                    max([seq.shape[0] for seq in self.data])
-                    if max_len is None
-                    else max_len
-                )
-                tail_shape = first.shape[1:]
-                result = np.full(
-                    (len(self.data), maxlen, *tail_shape),
-                    value,
-                    dtype=self.data_type or first.dtype,
-                )
-                for i, seq in enumerate(self.data):
-                    data = seq[:maxlen]
-                    result[i, : len(data), ...] = data
-                    if self.has_metadata():
-                        attention_mask = np.zeros(maxlen, dtype=np.int8)
-                        attention_mask[: len(data)] = 1
-                        if "attention_mask" in self.metadata[i]:
-                            self.metadata[i]["attention_mask"] = attention_mask
-                        else:
-                            self.metadata[i].update({"attention_mask": attention_mask})
-            elif (
-                isinstance(first, list)
-                and len(first) > 0
-                and isinstance(first[0], np.ndarray)
-                and first[0].ndim == 2
-            ):
-                maxlen = (
-                    max([len(seq) for seq in self.data]) if max_len is None else max_len
-                )
-                row_dim, col_dim = first[0].shape
-                result = np.full(
-                    (len(self.data), maxlen, row_dim, col_dim),
-                    value,
-                    dtype=self.data_type or first[0].dtype,
-                )
-                for i, seq in enumerate(self.data):
-                    data = seq[:maxlen]
-                    # stack list of 2D arrays into 3D then assign
-                    if len(data) > 0:
-                        result[i, : len(data), :, :] = np.stack(data, axis=0)
-                    if self.has_metadata():
-                        attention_mask = np.zeros(maxlen, dtype=np.int8)
-                        attention_mask[: len(data)] = 1
-                        if "attention_mask" in self.metadata[i]:
-                            self.metadata[i]["attention_mask"] = attention_mask
-                        else:
-                            self.metadata[i].update({"attention_mask": attention_mask})
-            else:
-                maxlen = (
-                    max([len(seq) for seq in self.data]) if max_len is None else max_len
-                )
-                result = np.full((len(self.data), maxlen), value, dtype=self.data_type)
-                for i, seq in enumerate(self.data):
-                    data = seq[:maxlen]
-                    try:
-                        result[i, : len(data)] = data
-                    except:
-                        print(f"Error padding data for modality {self.modality_id}")
-                        print(f"Data shape: {data.shape}")
-                        print(f"Result shape: {result.shape}")
-                        raise Exception("Error padding data")
-                    if self.has_metadata():
-                        attention_mask = np.zeros(result.shape[1], dtype=np.int8)
-                        attention_mask[: len(data)] = 1
-                        if "attention_mask" in self.metadata[i]:
-                            self.metadata[i]["attention_mask"] = attention_mask
-                        else:
-                            self.metadata[i].update({"attention_mask": attention_mask})
-        # TODO: this might need to be a new modality (otherwise we loose the original data)
+    def _set_attention_mask(self, index, attention_mask):
+        if not self.has_metadata() or index >= len(self.metadata):
+            return
+        if "attention_mask" in self.metadata[index]:
+            self.metadata[index]["attention_mask"] = attention_mask
+        else:
+            self.metadata[index].update({"attention_mask": attention_mask})
+
+    def _reshape_single_instance_embedding(self, arr):
+        if arr.ndim != 1:
+            return arr
+        if self.has_metadata() and len(self.metadata) == 1:
+            return arr.reshape(1, -1)
+        if not self.has_metadata() or len(self.metadata) <= 1:
+            return arr.reshape(1, -1)
+        return arr
+
+    def _pad_embedding_matrix(self, value, max_len):
+        arr = np.asarray(self.data)
+        if arr.dtype == object:
+            return False
+
+        arr = self._reshape_single_instance_embedding(arr)
+        if arr.ndim != 2:
+            return False
+
+        if arr.shape[1] == max_len:
+            self.data = arr.copy()
+            return True
+
+        result = np.full(
+            (arr.shape[0], max_len),
+            value,
+            dtype=self.data_type or arr.dtype,
+        )
+        copy_width = min(arr.shape[1], max_len)
+        result[:, :copy_width] = arr[:, :copy_width]
         self.data = result
+        return True
+
+    def _pad_variable_length_sequences(self, value, max_len):
+        first = self.data[0]
+        if isinstance(first, np.ndarray) and first.ndim == 3:
+            maxlen = (
+                max([seq.shape[0] for seq in self.data]) if max_len is None else max_len
+            )
+            tail_shape = first.shape[1:]
+            result = np.full(
+                (len(self.data), maxlen, *tail_shape),
+                value,
+                dtype=self.data_type or first.dtype,
+            )
+            for i, seq in enumerate(self.data):
+                data = seq[:maxlen]
+                result[i, : len(data), ...] = data
+                attention_mask = np.zeros(maxlen, dtype=np.int8)
+                attention_mask[: len(data)] = 1
+                self._set_attention_mask(i, attention_mask)
+        elif (
+            isinstance(first, list)
+            and len(first) > 0
+            and isinstance(first[0], np.ndarray)
+            and first[0].ndim == 2
+        ):
+            maxlen = (
+                max([len(seq) for seq in self.data]) if max_len is None else max_len
+            )
+            row_dim, col_dim = first[0].shape
+            result = np.full(
+                (len(self.data), maxlen, row_dim, col_dim),
+                value,
+                dtype=self.data_type or first[0].dtype,
+            )
+            for i, seq in enumerate(self.data):
+                data = seq[:maxlen]
+                if len(data) > 0:
+                    result[i, : len(data), :, :] = np.stack(data, axis=0)
+                attention_mask = np.zeros(maxlen, dtype=np.int8)
+                attention_mask[: len(data)] = 1
+                self._set_attention_mask(i, attention_mask)
+        else:
+            maxlen = (
+                max([len(seq) for seq in self.data]) if max_len is None else max_len
+            )
+            result = np.full((len(self.data), maxlen), value, dtype=self.data_type)
+            for i, seq in enumerate(self.data):
+                data = seq[:maxlen]
+                try:
+                    result[i, : len(data)] = data
+                except Exception as exc:
+                    raise ValueError(
+                        f"Error padding data for modality {self.modality_id}: "
+                        f"data shape {getattr(data, 'shape', None)}, "
+                        f"result shape {result.shape}"
+                    ) from exc
+                attention_mask = np.zeros(result.shape[1], dtype=np.int8)
+                attention_mask[: len(data)] = 1
+                self._set_attention_mask(i, attention_mask)
+        self.data = result
+
+    def pad(self, value=0, max_len=None):
+        if not self.has_data():
+            return
+
+        if max_len is None:
+            self.data = np.array(self.data)
+            return
+
+        if self._pad_embedding_matrix(value, max_len):
+            return
+
+        self._pad_variable_length_sequences(value, max_len)
 
     def get_data_layout(self):
         if self.has_metadata():
diff --git a/src/main/python/systemds/scuro/modality/unimodal_modality.py b/src/main/python/systemds/scuro/modality/unimodal_modality.py
index 0535c64bcee..e1bba7df9da 100644
--- a/src/main/python/systemds/scuro/modality/unimodal_modality.py
+++ b/src/main/python/systemds/scuro/modality/unimodal_modality.py
@@ -27,7 +27,11 @@
 from systemds.scuro.modality.modality import Modality
 from systemds.scuro.modality.joined import JoinedModality
 from systemds.scuro.modality.transformed import TransformedModality
-from systemds.scuro.representations.representation import RepresentationStats
+from systemds.scuro.representations.representation import (
+    CONTAINER_ARRAY,
+    RepresentationStats,
+    stats_bytes,
+)
 from systemds.scuro.utils.identifier import Identifier
 
 
@@ -64,21 +68,36 @@ def get_metadata_at_position(self, position: int):
         return self.metadata[position]
 
     def get_stats(self):
+        if self.stats is not None and getattr(self.stats, "dtype", None) is None:
+            try:
+                self.stats.dtype = np.dtype(self.data_loader.data_type)
+            except (AttributeError, TypeError):
+                pass
         return self.stats
 
     def get_output_stats(self):
-        return RepresentationStats(self.stats.num_instances, self.stats.output_shape)
+        stats = self.get_stats()
+        return RepresentationStats(
+            stats.num_instances,
+            stats.output_shape,
+            output_shape_is_known=getattr(stats, "output_shape_is_known", True),
+            dtype=getattr(stats, "dtype", None),
+            container=getattr(stats, "container", CONTAINER_ARRAY),
+            shape_variance=getattr(stats, "shape_variance", 0.0),
+            sampling_rate=getattr(stats, "sampling_rate", None),
+        )
 
     def estimate_memory_bytes(self):
-        memory_bytes = 1
-        for i in self.stats.output_shape:
-            memory_bytes *= i
-
-        return (
-            self.stats.num_instances * memory_bytes * 4
-        )  # TODO: check how to meausure str size
+        return stats_bytes(self.get_stats())
 
     def estimate_peak_memory_bytes(self):
+        loader_estimate = getattr(self.data_loader, "estimate_peak_memory_bytes", None)
+        if callable(loader_estimate):
+            estimate = loader_estimate()
+            return {
+                "cpu_peak_bytes": float(estimate["cpu_peak_bytes"]),
+                "gpu_peak_bytes": float(estimate.get("gpu_peak_bytes", 0.0)),
+            }
         return {"cpu_peak_bytes": self.estimate_memory_bytes(), "gpu_peak_bytes": 0.0}
 
     def extract_raw_data(self):
@@ -149,6 +168,7 @@ def apply_representations(self, representations, aggregation=None, parallel=Fals
         for representation in representations:
             transformed_modality = TransformedModality(self, representation.name)
             transformed_modality.data = []
+            transformed_modality.metadata = []
             transformed_modalities_per_representation[representation.name] = (
                 transformed_modality
             )

From d42aae33ed5157e2dd570fd3d34030be28d41690 Mon Sep 17 00:00:00 2001
From: Christina Dionysio <dionysio@tu-berlin.de>
Date: Wed, 26 Aug 2026 08:21:02 +0200
Subject: [PATCH 4/7] refine import

---
 src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
index 1b24e6ef192..cbf7cfb730a 100644
--- a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
+++ b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
@@ -27,7 +27,7 @@
 from typing import List, Any, Optional, Dict
 from functools import lru_cache
 
-from systemds.scuro import ModalityType
+from systemds.scuro.modality.type import ModalityType
 from systemds.scuro.drsearch.node_executor import NodeExecutor, ResultEntry
 from systemds.scuro.representations.representation import RepresentationStats
 from systemds.scuro.drsearch.ranking import rank_by_robustness, rank_by_tradeoff

From 1e0065bd2292ed03854288287f6364e3f8b6308e Mon Sep 17 00:00:00 2001
From: Christina Dionysio <dionysio@tu-berlin.de>
Date: Wed, 26 Aug 2026 08:28:11 +0200
Subject: [PATCH 5/7] remove import

---
 src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
index cbf7cfb730a..1b2227b773c 100644
--- a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
+++ b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py
@@ -30,7 +30,7 @@
 from systemds.scuro.modality.type import ModalityType
 from systemds.scuro.drsearch.node_executor import NodeExecutor, ResultEntry
 from systemds.scuro.representations.representation import RepresentationStats
-from systemds.scuro.drsearch.ranking import rank_by_robustness, rank_by_tradeoff
+from systemds.scuro.drsearch.ranking import rank_by_tradeoff
 from systemds.scuro.drsearch.task import PerformanceMeasure
 from systemds.scuro.representations.concatenation import Concatenation
 from systemds.scuro.representations.hadamard import Hadamard

From 5e3130c640b9835c4f17dae9d07e1047c8433552 Mon Sep 17 00:00:00 2001
From: Christina Dionysio <dionysio@tu-berlin.de>
Date: Wed, 26 Aug 2026 08:43:55 +0200
Subject: [PATCH 6/7] add node executor changes

---
 .../python/systemds/scuro/drsearch/node_executor.py    |  6 ++++++
 .../systemds/scuro/representations/representation.py   | 10 ++++++++++
 2 files changed, 16 insertions(+)

diff --git a/src/main/python/systemds/scuro/drsearch/node_executor.py b/src/main/python/systemds/scuro/drsearch/node_executor.py
index ec5d9f40c36..3400e6c6c50 100644
--- a/src/main/python/systemds/scuro/drsearch/node_executor.py
+++ b/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] = {}
 
diff --git a/src/main/python/systemds/scuro/representations/representation.py b/src/main/python/systemds/scuro/representations/representation.py
index c7b6d69d730..ae7df5e4477 100644
--- a/src/main/python/systemds/scuro/representations/representation.py
+++ b/src/main/python/systemds/scuro/representations/representation.py
@@ -42,6 +42,7 @@ class RepresentationStats:
     dtype: Optional[Any] = None
     container: str = CONTAINER_ARRAY
     shape_variance: float = 0.0
+    sampling_rate: Optional[float] = None
 
 
 def stats_dtype(stats) -> np.dtype:
@@ -180,6 +181,15 @@ def set_parameters(self, parameters):
         for parameter in parameters:
             setattr(self, parameter, parameters[parameter])
 
+    def check_preconditions(self, input_stats) -> Optional[str]:
+        return None
+
+    def configure_for_input(self, input_stats) -> None:
+        return None
+
+    def filter_parameter_domain(self, name, values, input_stats):
+        return values
+
     def estimate_memory_bytes(self, input_stats):
         output_memory_bytes = self.estimate_output_memory_bytes(input_stats)
         return output_memory_bytes

From 5804055ad5576ef9f863e340877dff728e67bfdd Mon Sep 17 00:00:00 2001
From: Christina Dionysio <dionysio@tu-berlin.de>
Date: Wed, 26 Aug 2026 09:15:33 +0200
Subject: [PATCH 7/7] add unique folder names to avoid race condition in
 parallel tests

---
 src/main/python/tests/iotests/test_io_csv.py                    | 2 +-
 src/main/python/tests/iotests/test_io_pandas_systemds.py        | 2 +-
 .../tests/python_java_data_transfer/test_dense_numpy_matrix.py  | 2 +-
 .../python/tests/python_java_data_transfer/test_pandas_frame.py | 2 +-
 4 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/main/python/tests/iotests/test_io_csv.py b/src/main/python/tests/iotests/test_io_csv.py
index 042e7e308a4..9d6860454dc 100644
--- a/src/main/python/tests/iotests/test_io_csv.py
+++ b/src/main/python/tests/iotests/test_io_csv.py
@@ -31,7 +31,7 @@
 class TestReadCSV(unittest.TestCase):
 
     sds: SystemDSContext = None
-    temp_dir: str = "tests/iotests/temp_write_csv/"
+    temp_dir: str = "tests/iotests/temp_write_csv_read/"
     n_cols = 3
     n_rows = 100
 
diff --git a/src/main/python/tests/iotests/test_io_pandas_systemds.py b/src/main/python/tests/iotests/test_io_pandas_systemds.py
index 214bc7475ad..91482da7619 100644
--- a/src/main/python/tests/iotests/test_io_pandas_systemds.py
+++ b/src/main/python/tests/iotests/test_io_pandas_systemds.py
@@ -42,7 +42,7 @@ def create_dataframe(n_rows, n_cols, mixed=True):
 class TestPandasFromToSystemds(unittest.TestCase):
 
     sds: SystemDSContext = None
-    temp_dir: str = "tests/iotests/temp_write_csv/"
+    temp_dir: str = "tests/iotests/temp_write_csv_pandas/"
 
     @classmethod
     def setUpClass(cls):
diff --git a/src/main/python/tests/python_java_data_transfer/test_dense_numpy_matrix.py b/src/main/python/tests/python_java_data_transfer/test_dense_numpy_matrix.py
index fcfe683dc7f..5d84d2a0666 100644
--- a/src/main/python/tests/python_java_data_transfer/test_dense_numpy_matrix.py
+++ b/src/main/python/tests/python_java_data_transfer/test_dense_numpy_matrix.py
@@ -32,7 +32,7 @@
 class TestMatrixBlockConverterUnixPipe(unittest.TestCase):
 
     sds: SystemDSContext = None
-    temp_dir: str = "tests/iotests/temp_write_csv/"
+    temp_dir: str = "tests/iotests/temp_write_csv_matrix/"
 
     @classmethod
     def setUpClass(cls):
diff --git a/src/main/python/tests/python_java_data_transfer/test_pandas_frame.py b/src/main/python/tests/python_java_data_transfer/test_pandas_frame.py
index a841795363a..28982bc4e0a 100644
--- a/src/main/python/tests/python_java_data_transfer/test_pandas_frame.py
+++ b/src/main/python/tests/python_java_data_transfer/test_pandas_frame.py
@@ -32,7 +32,7 @@
 class TestFrameConverterUnixPipe(unittest.TestCase):
 
     sds: SystemDSContext = None
-    temp_dir: str = "tests/iotests/temp_write_csv/"
+    temp_dir: str = "tests/iotests/temp_write_csv_frame/"
 
     @classmethod
     def setUpClass(cls):