fix many; but not finished

integrationtests/evaluator/integrationtest_evaluator.py

@@ -160,15 +160,14 @@ def test_evaluator(dataset_helper: ImageDatasetHelper) -> None:
     evaluator = EvaluatorStub(evaluator_channel)
     split_ts1, split_ts2, split1_size, split2_size, split3_size = prepare_dataset(dataset_helper)
     model_id = prepare_model()
-    eval_model_resp = evaluate_model(model_id, evaluator, [(split_ts2, split_ts1)])
-    print(eval_model_resp)
-    assert not eval_model_resp.evaluation_started, "Evaluation should not start if start_timestamp > end_timestamp"
-    assert eval_model_resp.eval_aborted_reasons == EvaluationAbortedReason.EMPTY_DATASET
+    # assert not eval_model_resp.evaluation_started, "Evaluation should not start if start_timestamp > end_timestamp"
+    # assert eval_model_resp.eval_aborted_reasons == EvaluationAbortedReason.EMPTY_DATASET
 
     # (start_timestamp, end_timestamp)
     intervals = [
         (None, split_ts1),
         (None, split_ts2),
+        (split_ts2, split_ts1),
         (split_ts1, split_ts2),
         (split_ts1, None),
         (split_ts2, None),
@@ -178,6 +177,7 @@ def test_evaluator(dataset_helper: ImageDatasetHelper) -> None:
     expected_data_sizes = [
         split1_size,
         split1_size + split2_size,
+        None,
         split2_size,
         split2_size + split3_size,
         split3_size,
@@ -188,11 +188,17 @@ def test_evaluator(dataset_helper: ImageDatasetHelper) -> None:
     eval_model_resp = evaluate_model(model_id, evaluator, intervals)
     print(eval_model_resp)
     assert eval_model_resp.evaluation_started
-    assert eval_model_resp.dataset_sizes == expected_data_sizes
+    assert len(eval_model_resp.interval_responses) == len(intervals)
+    for interval_resp, expected_size in zip(eval_model_resp.interval_responses, expected_data_sizes):
+        if expected_size is None:
+            assert interval_resp.eval_aborted_reason == EvaluationAbortedReason.EMPTY_DATASET
+        else:
+            assert interval_resp.dataset_size == expected_size
+            assert interval_resp.eval_aborted_reason == EvaluationAbortedReason.NOT_ABORTED
 
     eval_result_resp = wait_for_evaluation(eval_model_resp.evaluation_id, evaluator)
     assert eval_result_resp.valid
-    assert len(eval_result_resp.evaluation_results) == len(intervals)
+    assert len(eval_result_resp.evaluation_results) == len(intervals) - 1
     for single_eval_data in eval_result_resp.evaluation_results:
         assert len(single_eval_data.evaluation_data) == 1
         assert single_eval_data.evaluation_data[0].metric == "Accuracy"

modyn/evaluator/internal/grpc/evaluator_grpc_servicer.py

@@ -22,8 +22,9 @@
     EvaluationResultResponse,
     EvaluationStatusRequest,
     EvaluationStatusResponse,
-    SingleEvaluationData,
+    EvaluateModelIntervalResponse,
     SingleMetricResult,
+    EvaluationIntervalData,
 )
 from modyn.evaluator.internal.grpc.generated.evaluator_pb2_grpc import EvaluatorServicer
 from modyn.evaluator.internal.pytorch_evaluator import evaluate
@@ -103,15 +104,19 @@ def evaluate_model(self, request: EvaluateModelRequest, context: grpc.ServicerCo
                 logger.error(f"Trained model {request.model_id} does not exist!")
                 return EvaluateModelResponse(
                     evaluation_started=False,
-                    eval_aborted_reasons=[EvaluationAbortedReason.MODEL_NOT_EXIST_IN_METADATA] * num_intervals,
+                    interval_responses=[EvaluateModelIntervalResponse(
+                        eval_aborted_reason=EvaluationAbortedReason.MODEL_NOT_EXIST_IN_METADATA
+                    )] * num_intervals,
                 )
             model_class_name, model_config, amp = database.get_model_configuration(trained_model.pipeline_id)
 
         if not hasattr(dynamic_module_import("modyn.models"), model_class_name):
             logger.error(f"Model {model_class_name} not available!")
             return EvaluateModelResponse(
                 evaluation_started=False,
-                eval_aborted_reasons=[EvaluationAbortedReason.MODEL_IMPORT_FAILURE] * num_intervals,
+                interval_responses=[EvaluateModelIntervalResponse(
+                    eval_aborted_reason=EvaluationAbortedReason.MODEL_IMPORT_FAILURE
+                )] * num_intervals,
             )
 
         fetch_request = FetchModelRequest(model_id=request.model_id, load_metadata=False)
@@ -124,11 +129,12 @@ def evaluate_model(self, request: EvaluateModelRequest, context: grpc.ServicerCo
             )
             return EvaluateModelResponse(
                 evaluation_started=False,
-                eval_aborted_reasons=[EvaluationAbortedReason.MODEL_NOT_EXIST_IN_STORAGE] * num_intervals,
+                interval_responses=[EvaluateModelIntervalResponse(
+                    eval_aborted_reason=EvaluationAbortedReason.MODEL_NOT_EXIST_IN_STORAGE
+                )] * num_intervals,
             )
 
-        data_sizes: list[int] = []
-        eval_aborted_reasons = []
+        interval_responses = []
         not_failed_interval_ids: list[int] = []
         for idx, interval in enumerate(request.dataset_info.evaluation_intervals):
             dataset_size_req = GetDatasetSizeRequest(
@@ -140,19 +146,22 @@ def evaluate_model(self, request: EvaluateModelRequest, context: grpc.ServicerCo
 
             dataset_size = dataset_size_response.num_keys
             if not dataset_size_response.success:
-                eval_aborted_reasons.append(EvaluationAbortedReason.DATASET_NOT_FOUND)
-                data_sizes.append(0)
+                interval_responses.append(EvaluateModelIntervalResponse(
+                    eval_aborted_reason=EvaluationAbortedReason.DATASET_NOT_FOUND
+                ))
             elif dataset_size == 0:
-                eval_aborted_reasons.append(EvaluationAbortedReason.EMPTY_DATASET)
-                data_sizes.append(0)
+                interval_responses.append(EvaluateModelIntervalResponse(
+                    eval_aborted_reason=EvaluationAbortedReason.EMPTY_DATASET
+                ))
             else:
-                eval_aborted_reasons.append(EvaluationAbortedReason.UNKNOWN)
-                data_sizes.append(dataset_size)
+                interval_responses.append(EvaluateModelIntervalResponse(
+                    eval_aborted_reason=EvaluationAbortedReason.NOT_ABORTED,dataset_size=dataset_size
+                ))
                 not_failed_interval_ids.append(idx)
 
         if len(not_failed_interval_ids) == 0:
             logger.error("All evaluations failed. Evaluation cannot be started.")
-            return EvaluateModelResponse(evaluation_started=False, eval_aborted_reasons=eval_aborted_reasons)
+            return EvaluateModelResponse(evaluation_started=False, interval_responses=interval_responses)
 
         with self._lock:
             evaluation_id = self._next_evaluation_id
@@ -171,9 +180,10 @@ def evaluate_model(self, request: EvaluateModelRequest, context: grpc.ServicerCo
             logger.error("Trained model could not be downloaded. Evaluation cannot be started.")
             return EvaluateModelResponse(
                 evaluation_started=False,
-                eval_aborted_reasons=[EvaluationAbortedReason.DOWNLOAD_MODEL_FAILURE] * num_intervals,
+                interval_responses=[
+                   EvaluateModelIntervalResponse(eval_aborted_reason=EvaluationAbortedReason.DOWNLOAD_MODEL_FAILURE)
+               ] * num_intervals,
             )
-
         evaluation_info = EvaluationInfo(
             request,
             evaluation_id,
@@ -192,8 +202,7 @@ def evaluate_model(self, request: EvaluateModelRequest, context: grpc.ServicerCo
         return EvaluateModelResponse(
             evaluation_started=True,
             evaluation_id=evaluation_id,
-            dataset_sizes=data_sizes,
-            eval_aborted_reasons=eval_aborted_reasons,
+            interval_responses=interval_responses,
         )
 
     def _run_evaluation(self, evaluation_id: int) -> None:
@@ -263,7 +272,7 @@ def get_evaluation_result(
             return EvaluationResultResponse(valid=False)
 
         logger.info("Returning results of all metrics.")
-        evaluation_data: list[SingleEvaluationData] = []
+        evaluation_data: list[EvaluationIntervalData] = []
 
         metric_result_queue = self._evaluation_process_dict[evaluation_id].metric_result_queue
 
@@ -273,7 +282,7 @@ def get_evaluation_result(
             except queue.Empty:
                 break
             metric_result = [SingleMetricResult(metric=name, result=result) for name, result in metric_result]
-            single_eval_data = SingleEvaluationData(interval_index=interval_idx, evaluation_data=metric_result)
+            single_eval_data = EvaluationIntervalData(interval_index=interval_idx, evaluation_data=metric_result)
 
             evaluation_data.append(single_eval_data)
         if len(evaluation_data) < len(self._evaluation_dict[evaluation_id].not_failed_interval_ids):