Release 1.7.4, Merge pull request #321 from sentinel-hub/develop

Release 1.7.4

.pre-commit-config.yaml

@@ -13,20 +13,20 @@ repos:
       - id: debug-statements
 
   - repo: https://github.com/pre-commit/mirrors-prettier
-    rev: "v4.0.0-alpha.4"
+    rev: "v4.0.0-alpha.8"
     hooks:
       - id: prettier
         exclude: "tests/(test_stats|test_project)/"
         types_or: [json]
 
   - repo: https://github.com/psf/black
-    rev: 23.11.0
+    rev: 23.12.1
     hooks:
       - id: black
         language_version: python3
 
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: "v0.1.7"
+    rev: "v0.1.11"
     hooks:
       - id: ruff
 

CHANGELOG.md

@@ -1,3 +1,8 @@
+## [Version 1.7.4] - 2024-01-03
+
+- Pipelines now have an additional parameter `raise_if_failed` to raise an error if the pipeline failed.
+
+
 ## [Version 1.7.3] - 2023-12-11
 
 - Fix bug with versions of `sentinelhub-py >= 3.10.0` due to bad version string comparison.

eogrow/__init__.py

@@ -1,3 +1,3 @@
 """The main module of the eo-grow package."""
 
-__version__ = "1.7.3"
+__version__ = "1.7.4"

eogrow/cli.py

@@ -220,9 +220,9 @@ def _prepare_chain(
     for i, run_config in enumerate(config):
         if "pipeline_config" not in run_config:
             raise ValueError(f"Pipeline-chain element {i} is missing the field `pipeline_config`.")
-        raw_chain.append({
-            **run_config, "pipeline_config": _prepare_config(run_config["pipeline_config"], variables, test_patches)
-        })
+        raw_chain.append(
+            {**run_config, "pipeline_config": _prepare_config(run_config["pipeline_config"], variables, test_patches)}
+        )
     return raw_chain  # type: ignore[return-value]
 
 

eogrow/core/pipeline.py

@@ -27,6 +27,10 @@
 LOGGER = logging.getLogger(__name__)
 
 
+class PipelineExecutionError(RuntimeError):
+    """Raised when the pipeline failed some executions."""
+
+
 class Pipeline(EOGrowObject):
     """A base class for execution of processing procedures which may or may not include running EOWorkflows, running
     EOExecutions, creating maps, etc.
@@ -212,6 +216,7 @@ def run(self) -> None:
         """The main method for pipeline execution. It sets up logging and runs the pipeline procedure."""
         timestamp = current_timestamp()
         self.current_execution_name = self.get_pipeline_execution_name(timestamp)
+        log_folder = self.logging_manager.get_pipeline_logs_folder(self.current_execution_name, full_path=True)
 
         root_logger = logging.getLogger()
         handlers = self.logging_manager.start_logging(root_logger, self.current_execution_name, "pipeline.log")
@@ -231,10 +236,7 @@ def run(self) -> None:
             elapsed_time = time.time() - pipeline_start
 
             if failed:
-                LOGGER.info(
-                    "Pipeline finished with some errors! Check %s",
-                    self.logging_manager.get_pipeline_logs_folder(self.current_execution_name, full_path=True),
-                )
+                LOGGER.info("Pipeline finished with some errors! Check %s", log_folder)
             else:
                 LOGGER.info("Pipeline finished successfully!")
 
@@ -250,6 +252,9 @@ def run(self) -> None:
             self.logging_manager.save_eopatch_execution_status(
                 pipeline_execution_name=self.current_execution_name, finished=finished, failed=failed
             )
+
+            if failed and self.config.raise_if_failed:
+                raise PipelineExecutionError(f"Pipeline failed some executions. Check {log_folder}.")
         finally:
             self.logging_manager.stop_logging(root_logger, handlers)
 

eogrow/core/schemas.py

@@ -69,6 +69,9 @@ class PipelineSchema(BaseSchema):
     raise_on_temporal_mismatch: bool = Field(
         False, description="Treat `TemporalDimensionWarning` as an exception during EOExecution."
     )
+    raise_if_failed: bool = Field(
+        False, description="Raise an exception if `run_procedure` returns some executions in `failed`."
+    )
     debug: bool = Field(False, description="Run pipeline without the `ray` wrapper to enable debugging.")
 
 

eogrow/tasks/common.py

@@ -46,7 +46,7 @@ def execute(self, eopatch: EOPatch) -> EOPatch:
 
         for label in self.labels:
             label_mask = np.squeeze((mask == label).astype(np.uint8), axis=-1)
-            mask_mod = morp_func(label_mask) * label
+            mask_mod = morp_func(label_mask) * label  # type: ignore[operator]
             mask_mod = mask_mod[..., np.newaxis]
             mask[mask == label] = mask_mod[mask == label]
 

tests/core/test_pipeline.py

@@ -8,7 +8,7 @@
 from sentinelhub import CRS, BBox
 
 from eogrow.core.config import interpret_config_from_path
-from eogrow.core.pipeline import Pipeline
+from eogrow.core.pipeline import Pipeline, PipelineExecutionError
 
 
 @pytest.fixture(scope="session", name="simple_config_filename")
@@ -33,6 +33,14 @@ def run_procedure(self) -> Tuple[List[str], List[str]]:
         return finished[:-1], finished[-1:] + failed
 
 
+class FailingPipeline(Pipeline):
+    class Schema(Pipeline.Schema):
+        fail: bool
+
+    def run_procedure(self) -> Tuple[List[str], List[str]]:
+        return [], ([0] if self.config.fail else [])
+
+
 def test_pipeline_execution(simple_config_filename: str) -> None:
     """Tests that appropriate folders and log files are created."""
     config = interpret_config_from_path(simple_config_filename)
@@ -88,3 +96,17 @@ def test_get_patch_list_filtration_error(test_subset: List[Union[int, str]], sim
     pipeline = SimplePipeline.from_raw_config(config)
     with pytest.raises(ValueError):
         pipeline.get_patch_list()
+
+
+@pytest.mark.parametrize("fail", [True, False])
+@pytest.mark.parametrize("raise_if_failed", [True, False])
+def test_pipeline_raises_on_failure(fail: bool, raise_if_failed: bool, simple_config_filename: str):
+    config = interpret_config_from_path(simple_config_filename)
+    config.pop("test_param")
+    pipeline = FailingPipeline.from_raw_config({"fail": fail, "raise_if_failed": raise_if_failed, **config})
+
+    if fail and raise_if_failed:
+        with pytest.raises(PipelineExecutionError):
+            pipeline.run()
+    else:
+        pipeline.run()