Repository: matchms/matchms
Pull request #483 (2023-08-15 09:58:46)

Add workflow setters
Solves suggestion of @florian-huber 

Adds setters for workflow attributes back in to prevent changing anything in workflow that will not be passed on the SpectrumProcessor. 


Patch:
diff --git a/matchms/Pipeline.py b/matchms/Pipeline.py
index 96602dc7..baf579bb 100644
--- a/matchms/Pipeline.py
+++ b/matchms/Pipeline.py
@@ -1,7 +1,7 @@
 import logging
 from collections import OrderedDict
 from datetime import datetime
-from typing import Iterable, List, Optional, Tuple, Union
+from typing import Iterable, List, Optional, Union
 import yaml
 import matchms.similarity as mssimilarity
 from matchms import calculate_scores
@@ -10,6 +10,7 @@ from matchms.logging_functions import (add_logging_to_file,
                                        reset_matchms_logger,
                                        set_matchms_logger_level)
 from matchms.SpectrumProcessor import SpectrumProcessor
+from matchms.typing import SpectrumType
 
 
 _masking_functions = ["filter_by_range"]
@@ -48,9 +49,11 @@ def create_workflow(yaml_file_name: Optional[str] = None,
     """
     # pylint: disable=too-many-arguments
     workflow = OrderedDict()
-    queries_processor = initialize_spectrum_processor(predefined_processing_queries, additional_filters_queries)
+    queries_processor = initialize_spectrum_processor(predefined_processing_queries,
+                                                      additional_filters_queries)
     workflow["query_filters"] = queries_processor.processing_steps
-    reference_processor = initialize_spectrum_processor(predefined_processing_reference, additional_filters_references)
+    reference_processor = initialize_spectrum_processor(predefined_processing_reference,
+                                                        additional_filters_references)
     workflow["reference_filters"] = reference_processor.processing_steps
     workflow["score_computations"] = score_computations
     if yaml_file_name is not None:
@@ -63,7 +66,7 @@ def create_workflow(yaml_file_name: Optional[str] = None,
 
 
 def initialize_spectrum_processor(predefined_pipeline: Optional[str],
-                                  additional_filters: Tuple[str]) -> SpectrumProcessor:
+                                  additional_filters: Iterable[Union[str, List[dict]]]) -> SpectrumProcessor:
     """Initialize spectrum processing workflow."""
     processor = SpectrumProcessor(predefined_pipeline)
     for step in additional_filters:
@@ -108,7 +111,7 @@ def ordered_dump(data: OrderedDict, stream=None, dumper=yaml.SafeDumper, **kwds)
     return yaml.dump(data, stream, OrderedDumper, **kwds)
 
 
-def check_score_computation(score_computations: Tuple[str]):
+def check_score_computation(score_computations: Iterable[Union[str, List[dict]]]):
     """Check if the score computations seem OK before running.
     Aim is to avoid pipeline crashing after long computation.
     """
@@ -208,7 +211,7 @@ class Pipeline:
         self.logging_level = logging_level
         self.logging_file = logging_file
         self.progress_bar = progress_bar
-        self.workflow = workflow
+        self.__workflow = workflow
         self.check_workflow()
 
         self.write_to_logfile("--- Processing pipeline: ---")
@@ -220,31 +223,30 @@ class Pipeline:
         """Initialize spectrum processing workflow for the query spectra."""
         self.processing_queries = initialize_spectrum_processor(
             None,
-            self.workflow["query_filters"]
+            self.__workflow["query_filters"]
             )
 
         self.write_to_logfile(str(self.processing_queries))
-        if self.processing_queries.processing_steps != self.workflow["query_filters"]:
+        if self.processing_queries.processing_steps != self.__workflow["query_filters"]:
             logger.warning("The order of the filters has been changed compared to the Yaml file.")
 
     def _initialize_spectrum_processor_references(self):
         """Initialize spectrum processing workflow for the reference spectra."""
         self.processing_references = initialize_spectrum_processor(
             None,
-            self.workflow["reference_filters"]
+            self.__workflow["reference_filters"]
             )
         self.write_to_logfile(str(self.processing_references))
-        self.write_to_logfile(str(self.processing_queries))
-        if self.processing_queries.processing_steps != self.workflow["query_filters"]:
+        if self.processing_queries.processing_steps != self.__workflow["query_filters"]:
             logger.warning("The order of the filters has been changed compared to the Yaml file.")
 
     def check_workflow(self):
         """Define Pipeline workflow based on a yaml file (config_file).
         """
-        assert isinstance(self.workflow, OrderedDict), \
-            f"Workflow is expectd to be a OrderedDict, instead it was of type {type(self.workflow)}"
+        assert isinstance(self.__workflow, OrderedDict), \
+            f"Workflow is expectd to be a OrderedDict, instead it was of type {type(self.__workflow)}"
         expected_keys = {"query_filters", "reference_filters", "score_computations"}
-        assert set(self.workflow.keys()) == expected_keys
+        assert set(self.__workflow.keys()) == expected_keys
         check_score_computation(score_computations=self.score_computations)
 
     def run(self, query_files, reference_files = None):
@@ -390,21 +392,36 @@ class Pipeline:
 
     # Getter & Setters
     @property
-    def score_computations(self):
-        return self.workflow.get("score_computations")
+    def score_computations(self) -> Iterable[Union[str, List[dict]]]:
+        return self.__workflow.get("score_computations")
+
+    @score_computations.setter
+    def score_computations(self, computations):
+        self.__workflow["score_computations"] = computations
+        check_score_computation(score_computations=self.score_computations)
 
     @property
-    def query_filters(self):
-        return self.workflow.get("query_filters")
+    def query_filters(self) -> Iterable[Union[str, List[dict]]]:
+        return self.__workflow.get("query_filters")
+
+    @query_filters.setter
+    def query_filters(self, filters: Iterable[Union[str, List[dict]]]):
+        self.__workflow["query_filters"] = filters
+        self._initialize_spectrum_processor_queries()
 
     @property
-    def reference_filters(self):
-        return self.workflow.get("reference_filters")
+    def reference_filters(self) -> Iterable[Union[str, List[dict]]]:
+        return self.__workflow.get("reference_filters")
+
+    @reference_filters.setter
+    def reference_filters(self, filters: Iterable[Union[str, List[dict]]]):
+        self.__workflow["reference_filters"] = filters
+        self._initialize_spectrum_processor_references()
 
     @property
-    def spectrums_queries(self):
+    def spectrums_queries(self) -> List[SpectrumType]:
         return self._spectrums_queries
 
     @property
-    def spectrums_references(self):
+    def spectrums_references(self) -> List[SpectrumType]:
         return self._spectrums_references


Test patch:
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 49d483f6..54eaf2e5 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -191,3 +191,27 @@ def test_FingerprintSimilarity_pipeline():
     assert len(pipeline.spectrums_queries[0].get("fingerprint")) == 2048
     assert pipeline.scores.scores.shape == (5, 5, 2)
     assert pipeline.scores.score_names == ('MetadataMatch', 'FingerprintSimilarity')
+
+
+def test_pipeline_changing_workflow():
+    """Test if changing workflow after creating Pipeline results in the expected change of the pipeline"""
+    workflow = create_workflow(predefined_processing_queries=None,
+                               additional_filters_queries=["make_charge_int"],
+                               predefined_processing_reference=None,
+                               additional_filters_references=["make_charge_int"],
+                               score_computations=['precursormzmatch'],
+                               )
+    pipeline = Pipeline(workflow)
+    pipeline.query_filters = ["add_fingerprint"]
+    pipeline.reference_filters = ["add_fingerprint"]
+    pipeline.score_computations = [["modifiedcosine", {"tolerance": 10.0}]]
+    pipeline.run(spectrums_file_msp, spectrums_file_msp)
+    assert len(pipeline.spectrums_queries[0].get("fingerprint")) == 2048, \
+        "The query filters were not modified correctly"
+    assert len(pipeline.spectrums_references[0].get("fingerprint")) == 2048, \
+        "The reference filters were not modified correctly"
+    assert pipeline.scores.score_names == ('ModifiedCosine_score', 'ModifiedCosine_matches')
+    all_scores = pipeline.scores.to_array()
+    expected = np.array([[1., 0.30384404],
+                         [0.30384404, 1.]])
+    assert np.allclose(all_scores["ModifiedCosine_score"][3:, 3:], expected)