From b4cb598ab6386c016a56d627f9ed93bd70dc5340 Mon Sep 17 00:00:00 2001 From: Christina Dionysio Date: Tue, 1 Sep 2026 10:35:32 +0200 Subject: [PATCH] add evaluation procedure for test set --- .../scuro/dataloader/timeseries_loader.py | 2 + .../python/systemds/scuro/drsearch/ranking.py | 136 +++++++++++++++++- .../python/systemds/scuro/drsearch/task.py | 110 ++++++++++++++ .../scuro/drsearch/test_set_evaluation.py | 121 ++++++++++++++++ .../scuro/drsearch/unimodal_optimizer.py | 23 ++- .../timeseries_representations.py | 8 +- .../tests/scuro/test_unimodal_optimizer.py | 31 +++- 7 files changed, 422 insertions(+), 9 deletions(-) create mode 100644 src/main/python/systemds/scuro/drsearch/test_set_evaluation.py diff --git a/src/main/python/systemds/scuro/dataloader/timeseries_loader.py b/src/main/python/systemds/scuro/dataloader/timeseries_loader.py index 8e6c11316b0..a2f515f7d26 100644 --- a/src/main/python/systemds/scuro/dataloader/timeseries_loader.py +++ b/src/main/python/systemds/scuro/dataloader/timeseries_loader.py @@ -52,11 +52,13 @@ def __init__( normalize: bool = True, file_format: str = "npy", modality_type: Optional[ModalityType] = ModalityType.TIMESERIES, + channel_index: Optional[int] = None, ): super().__init__(source_path, indices, data_type, chunk_size, modality_type) self.signal_names = signal_names self.sampling_rate = sampling_rate self.normalize = normalize + self.channel_index = channel_index self.file_format = file_format.lower() self.stats = self.get_stats(source_path, sampling_rate) if self.file_format not in ["npy", "mat", "hdf5", "txt"]: diff --git a/src/main/python/systemds/scuro/drsearch/ranking.py b/src/main/python/systemds/scuro/drsearch/ranking.py index 381a6d1fb9d..935bc565356 100644 --- a/src/main/python/systemds/scuro/drsearch/ranking.py +++ b/src/main/python/systemds/scuro/drsearch/ranking.py @@ -21,11 +21,127 @@ from typing import Callable, Iterable, Optional +import numpy as np + + +def _operator_signature(entry) -> frozenset: + """Set of operator class names in an entry's DAG, ignoring hyperparameters.""" + try: + return frozenset( + node.operation.__name__ + for node in entry.dag.nodes + if getattr(node, "operation", None) is not None + ) + except Exception: + return frozenset() + + +def _dag_size(entry) -> int: + try: + return sum( + 1 + for node in entry.dag.nodes + if getattr(node, "operation", None) is not None + ) + except Exception: + return 0 + + +def rank_by_robustness( + entries: Iterable, + *, + performance_metric_name: str = "accuracy", + neighbourhood_weight: float = 0.5, + sharpness: int = 4, + one_se_parsimony: bool = True, + cache_scores: bool = True, + score_attr: str = "robustness_score", +): + entries = list(entries) + if not entries: + return [], [] + + def perf_of(entry): + if entry is None: + return None + try: + score = float(entry.val_score[performance_metric_name]) + except (KeyError, TypeError, ValueError): + return None + return score if np.isfinite(score) else None + + indexed_entries = [ + (index, entry, score) + for index, entry in enumerate(entries) + if (score := perf_of(entry)) is not None + ] + if not indexed_entries: + return [], [] + + original_indices, entries, performance = zip(*indexed_entries) + entries = list(entries) + perf = np.array(performance, dtype=float) + sizes = np.array([_dag_size(e) if e is not None else 0 for e in entries], float) + + smoothed = perf + if neighbourhood_weight > 0.0 and len(entries) > 1: + signatures = [ + _operator_signature(e) if e is not None else frozenset() for e in entries + ] + vocabulary = sorted({op for sig in signatures for op in sig}) + if vocabulary: + position = {op: i for i, op in enumerate(vocabulary)} + membership = np.zeros((len(entries), len(vocabulary)), dtype=np.float32) + for row, sig in enumerate(signatures): + for op in sig: + membership[row, position[op]] = 1.0 + intersection = membership @ membership.T + counts = membership.sum(1) + union = counts[:, None] + counts[None, :] - intersection + jaccard = np.divide( + intersection, + union, + out=np.zeros_like(intersection), + where=union > 0, + ) + weights = jaccard**sharpness + denominator = weights.sum(1) + neighbourhood = np.divide( + (weights * perf[None, :].astype(np.float32)).sum(1), + denominator, + out=perf.astype(np.float32).copy(), + where=denominator > 0, + ) + smoothed = ( + 1.0 - neighbourhood_weight + ) * perf + neighbourhood_weight * neighbourhood + + if cache_scores: + for entry, score in zip(entries, smoothed): + if entry is not None: + setattr(entry, score_attr, float(score)) + + if one_se_parsimony and len(entries) > 1: + standard_error = float(smoothed.std()) / np.sqrt(len(smoothed)) + threshold = float(smoothed.max()) - standard_error + keys = [ + (True, -sz, float(s)) if s >= threshold else (False, 0.0, float(s)) + for s, sz in zip(smoothed, sizes) + ] + else: + keys = [(True, 0.0, float(s)) for s in smoothed] + + local_indices = sorted(range(len(entries)), key=lambda i: keys[i], reverse=True) + sorted_entries = [entries[i] for i in local_indices] + sorted_indices = [original_indices[i] for i in local_indices] + + return sorted_entries, sorted_indices + def rank_by_tradeoff( entries: Iterable, *, - weights=(0.7, 0.3), + weights=(1.0, 0.0), performance_metric_name: str = "accuracy", runtime_accessor: Optional[Callable[[object], float]] = None, cache_scores: bool = True, @@ -48,8 +164,10 @@ def runtime_accessor(entry): task = getattr(entry, "task_time", 0.0) return rep + task - performance = [float(performance_score_accessor(e)) for e in entries] - runtimes = [float(runtime_accessor(e)) for e in entries] + performance = [ + float(performance_score_accessor(e)) if e is not None else 0.0 for e in entries + ] + runtimes = [float(runtime_accessor(e)) if e is not None else 0.0 for e in entries] perf_min, perf_max = min(performance), max(performance) run_min, run_max = min(runtimes), max(runtimes) @@ -76,17 +194,25 @@ def safe_normalize(values, vmin, vmax): if cache_scores: for entry, score in zip(entries, scores): + if entry is None: + continue if hasattr(entry, score_attr): setattr(entry, score_attr, score) else: setattr(entry, score_attr, score) - sorted_entries = sorted(entries, key=lambda e: e.tradeoff_score, reverse=True) + sorted_entries = sorted( + entries, + key=lambda e: e.tradeoff_score if hasattr(e, "tradeoff_score") else 0.0, + reverse=True, + ) sorted_indices = [ i for i, _ in sorted( - enumerate(entries), key=lambda pair: pair[1].tradeoff_score, reverse=True + enumerate(entries), + key=lambda pair: pair[1].tradeoff_score if pair is not None else None, + reverse=True, ) ] diff --git a/src/main/python/systemds/scuro/drsearch/task.py b/src/main/python/systemds/scuro/drsearch/task.py index 7977c628c12..7ddeaede76c 100644 --- a/src/main/python/systemds/scuro/drsearch/task.py +++ b/src/main/python/systemds/scuro/drsearch/task.py @@ -62,6 +62,18 @@ def compute_averages(self): self.average_scores[self.metrics] = np.mean(self.scores[self.metrics]) return self + def fold_scores(self): + return { + metric: [float(v) for v in values] for metric, values in self.scores.items() + } + + def score_stds(self): + """Per-metric standard deviation across folds (0.0 for a single fold).""" + return { + metric: (float(np.std(values, ddof=1)) if len(values) > 1 else 0.0) + for metric, values in self.scores.items() + } + class Task: def __init__( @@ -95,6 +107,7 @@ def __init__( self.measure_performance = measure_performance self.inference_time = [] self.training_time = [] + self.last_run_timing = {} self.expected_dim = 1 self.performance_measures = performance_measures self.train_scores = PerformanceMeasure("train", performance_measures) @@ -286,12 +299,109 @@ def run(self, data): if hasattr(model, "clean_up"): model.clean_up() del model + + self.last_run_timing = { + "train_time_per_fold_s": list(self.training_time), + "test_inference_time_per_fold_s": list(self.inference_time), + "train_time_mean_s": ( + float(np.mean(self.training_time)) if self.training_time else 0.0 + ), + "test_inference_time_mean_s": ( + float(np.mean(self.inference_time)) if self.inference_time else 0.0 + ), + "n_test_instances": len(self.test_indices) if self.test_indices else 0, + } return [ self.train_scores.compute_averages(), self.val_scores.compute_averages(), self.test_scores.compute_averages(), ] + def fit_once_and_time_inference( + self, data, latency_repeats: int = 200, latency_warmup: int = 20 + ): + model = self.create_model() + + train_X = self._gather_by_indices(data, self.train_indices) + train_y = self._gather_by_indices(self.labels, self.train_indices) + test_X = self._gather_by_indices(data, self.test_indices) + test_y = self._gather_by_indices(self.labels, self.test_indices) + + t0 = time.perf_counter() + model.fit(train_X, train_y, test_X, test_y) + fit_time = time.perf_counter() - t0 + + t0 = time.perf_counter() + test_score = model.test(np.asarray(test_X), test_y) + batch_inference_time = time.perf_counter() - t0 + + latency = self._measure_single_sample_latency( + model, test_X, test_y, latency_repeats, latency_warmup + ) + + scores = PerformanceMeasure("test_single_fit", self.performance_measures) + scores.add_scores(test_score[0]) + scores.compute_averages() + + if hasattr(model, "clean_up"): + model.clean_up() + + n_test = len(test_X) + return { + "single_fit_train_time_s": fit_time, + "test_batch_inference_time_s": batch_inference_time, + "test_inference_time_per_instance_ms": ( + batch_inference_time / n_test * 1000.0 if n_test else 0.0 + ), + "n_test_instances": n_test, + "single_fit_test_scores": scores.average_scores, + **latency, + } + + @staticmethod + def _prediction_callable(model): + if hasattr(model, "predict"): + return model.predict, "model.predict" + clf = getattr(model, "clf", None) + if clf is not None and hasattr(clf, "predict"): + return clf.predict, "clf.predict" + return None, None + + def _measure_single_sample_latency( + self, model, test_X, test_y, repeats: int, warmup: int + ): + empty = { + "inference_latency_ms_median": None, + "inference_latency_ms_p95": None, + "inference_latency_samples": 0, + "inference_latency_source": None, + } + if repeats <= 0 or not len(test_X): + return empty + + predict, source = self._prediction_callable(model) + if predict is None: + return empty + + timings = [] + for i in range(warmup + repeats): + sample_X = np.asarray([test_X[i % len(test_X)]]) + t0 = time.perf_counter() + try: + predict(sample_X) + except Exception: + return empty + elapsed = (time.perf_counter() - t0) * 1000.0 + if i >= warmup: + timings.append(elapsed) + + return { + "inference_latency_ms_median": float(np.median(timings)), + "inference_latency_ms_p95": float(np.percentile(timings, 95)), + "inference_latency_samples": len(timings), + "inference_latency_source": source, + } + def _reset_params(self): self.inference_time = [] self.training_time = [] diff --git a/src/main/python/systemds/scuro/drsearch/test_set_evaluation.py b/src/main/python/systemds/scuro/drsearch/test_set_evaluation.py new file mode 100644 index 00000000000..f160cd4ccd0 --- /dev/null +++ b/src/main/python/systemds/scuro/drsearch/test_set_evaluation.py @@ -0,0 +1,121 @@ +# ------------------------------------------------------------- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +# ------------------------------------------------------------- +from __future__ import annotations + +import time +from typing import Any, Dict, List + +from systemds.scuro.modality.modality import Modality +from systemds.scuro.drsearch.representation_dag import RepresentationDag +from systemds.scuro.representations.aggregated_representation import ( + AggregatedRepresentation, +) +from systemds.scuro.utils.schema_helpers import get_shape + + +def _unwrap(result): + if isinstance(result, dict): + if not result: + return None + return result[list(result.keys())[-1]] + return result + + +def _match_expected_dim(modality, task): + if modality is None or task is None: + return modality + if getattr(task, "expected_dim", 1) == 1 and get_shape(modality.metadata) > 1: + return AggregatedRepresentation().transform(modality) + return modality + + +def measure_representation_time_on_test_set( + dag: RepresentationDag, + modalities: List[Modality], + test_indices: List[int], + task=None, + repeats: int = 1, +) -> Dict[str, Any]: + subsets = [modality.subset(test_indices) for modality in modalities] + + timings = [] + output = None + for _ in range(max(1, repeats)): + t0 = time.perf_counter() + output = _match_expected_dim( + _unwrap(dag.execute(subsets, task, enable_cache=False)), task + ) + timings.append(time.perf_counter() - t0) + + timings.sort() + median = timings[len(timings) // 2] + n_test = len(test_indices) + + output_shape = None + if output is not None and getattr(output, "data", None) is not None: + try: + output_shape = tuple(getattr(output.data[0], "shape", ())) + except (IndexError, TypeError): + output_shape = None + + return { + "test_only_representation_time_s": median, + "test_only_representation_time_all_runs_s": timings, + "test_only_representation_time_per_instance_ms": ( + median / n_test * 1000.0 if n_test else 0.0 + ), + "test_only_n_instances": n_test, + "test_only_output_shape": output_shape, + "test_only_features_are_valid_for_scoring": False, + } + + +def measure_test_set_application( + dag: RepresentationDag, + modalities: List[Modality], + task, + full_data=None, + repeats: int = 1, + latency_repeats: int = 200, + latency_warmup: int = 20, +) -> Dict[str, Any]: + record = measure_representation_time_on_test_set( + dag, modalities, task.test_indices, task=task, repeats=repeats + ) + + if full_data is None: + t0 = time.perf_counter() + output = _match_expected_dim( + _unwrap(dag.execute(modalities, task, enable_cache=False)), task + ) + record["full_representation_time_s"] = time.perf_counter() - t0 + full_data = None if output is None else output.data + + if full_data is not None: + record.update( + task.fit_once_and_time_inference( + full_data, + latency_repeats=latency_repeats, + latency_warmup=latency_warmup, + ) + ) + + return record diff --git a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py index b185bd45dfb..5f130c369a3 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_tradeoff +from systemds.scuro.drsearch.ranking import rank_by_tradeoff, rank_by_robustness from systemds.scuro.drsearch.task import PerformanceMeasure from systemds.scuro.representations.concatenation import Concatenation from systemds.scuro.representations.hadamard import Hadamard @@ -994,6 +994,27 @@ def get_k_best_results( return results, cache + def get_k_most_robust_results( + self, + modality, + task, + performance_metric_name, + k=None, + neighbourhood_weight=0.5, + one_se_parsimony=True, + ): + task_results = self.results[modality.modality_id][task.model.name] + + results, sorted_indices = rank_by_robustness( + task_results, + performance_metric_name=performance_metric_name, + neighbourhood_weight=neighbourhood_weight, + one_se_parsimony=one_se_parsimony, + ) + + limit = self.k if k is None else k + return results[:limit], sorted_indices[:limit] + def add_worker_stat(self, worker_stats, modality_id): self.worker_stats[modality_id] = worker_stats diff --git a/src/main/python/systemds/scuro/representations/timeseries_representations.py b/src/main/python/systemds/scuro/representations/timeseries_representations.py index e6aa999e8f2..c67fde8692e 100644 --- a/src/main/python/systemds/scuro/representations/timeseries_representations.py +++ b/src/main/python/systemds/scuro/representations/timeseries_representations.py @@ -210,7 +210,9 @@ def __init__(self, params=None): super().__init__("Skew", min_input_length=3) def compute_feature(self, signal, axis=-1): - return np.array(stats.skew(signal, axis=axis)) + result = np.asarray(stats.skew(signal, axis=axis)) + zero_variance = np.std(signal, axis=axis) <= np.finfo(float).eps + return np.where(zero_variance, 0.0, result) @register_representation([ModalityType.TIMESERIES, ModalityType.PHYSIOLOGICAL]) @@ -253,7 +255,9 @@ def __init__(self, params=None): super().__init__("Kurtosis", min_input_length=4) def compute_feature(self, signal, axis=-1): - return np.array(stats.kurtosis(signal, fisher=True, bias=True, axis=axis)) + result = np.asarray(stats.kurtosis(signal, fisher=True, bias=True, axis=axis)) + zero_variance = np.std(signal, axis=axis) <= np.finfo(float).eps + return np.where(zero_variance, 0.0, result) @register_representation([ModalityType.TIMESERIES, ModalityType.PHYSIOLOGICAL]) diff --git a/src/main/python/tests/scuro/test_unimodal_optimizer.py b/src/main/python/tests/scuro/test_unimodal_optimizer.py index 41717b4e3a8..f27c721aa25 100644 --- a/src/main/python/tests/scuro/test_unimodal_optimizer.py +++ b/src/main/python/tests/scuro/test_unimodal_optimizer.py @@ -21,11 +21,16 @@ import unittest +from types import SimpleNamespace import numpy as np from systemds.scuro.representations.color_histogram import ColorHistogram from systemds.scuro.drsearch.operator_registry import Registry -from systemds.scuro.drsearch.unimodal_optimizer import UnimodalOptimizer +from systemds.scuro.drsearch.node_executor import ResultEntry +from systemds.scuro.drsearch.unimodal_optimizer import ( + UnimodalOptimizer, + UnimodalResults, +) from systemds.scuro.representations.covarep_audio_features import ZeroCrossing from systemds.scuro.representations.covarep_audio_features import ( @@ -133,6 +138,30 @@ def test_unimodal_optimizer_for_text_modality(self): ) self.optimize_unimodal_representation_for_modality([text]) + def test_robust_results_ignore_non_finite_scores(self): + modality = SimpleNamespace(modality_id="modality") + task = SimpleNamespace(model=SimpleNamespace(name="task")) + results = UnimodalResults([modality], [task], k=2) + scores = [0.8, np.nan, np.inf, -np.inf, None, 0.6] + entries = [ + ResultEntry( + val_score=None if score is None else {"accuracy": score}, + dag=SimpleNamespace( + nodes=[SimpleNamespace(operation=UnimodalOptimizer)] + ), + ) + for score in scores + ] + results.results[modality.modality_id][task.model.name] = entries + + robust, indices = results.get_k_most_robust_results( + modality, task, "accuracy", one_se_parsimony=False + ) + + self.assertEqual(robust, [entries[0], entries[5]]) + self.assertEqual(indices, [0, 5]) + self.assertTrue(all(np.isfinite(entry.robustness_score) for entry in robust)) + def test_bow_and_tfidf_require_dimensionality_reduction_before_task(self): text_data, text_md = ModalityRandomDataGenerator().create_text_data( self.num_instances, 10