add JobParser and template example usage

Author: Jhsmit

pyhdx/batch_processing.py

@@ -1,20 +1,31 @@
 import warnings
+from functools import reduce
 from pathlib import Path
 import os
-from pyhdx.models import PeptideMasterTable, HDXMeasurement, HDXMeasurementSet
-from pyhdx.fileIO import read_dynamx
+import re
 
+from pyhdx import TorchFitResult
+from pyhdx.models import PeptideMasterTable, HDXMeasurement, HDXMeasurementSet
+from pyhdx.fileIO import read_dynamx, csv_to_dataframe, save_fitresult
+from pyhdx.fitting import fit_rates_half_time_interpolate, fit_rates_weighted_average, \
+    fit_gibbs_global, fit_gibbs_global_batch, RatesFitResult, GenericFitResult
+import param
+import pandas as pd
+from pyhdx.support import gen_subclasses
+import yaml
 
 time_factors = {"s": 1, "m": 60.0, "min": 60.0, "h": 3600, "d": 86400}
 temperature_offsets = {"c": 273.15, "celsius": 273.15, "k": 0, "kelvin": 0}
 
-# todo add data filters in yaml spec
+
+# todo add data filters in state spec?
 # todo add proline, n_term options
-class YamlParser(object):
-    ""'object used to parse yaml data input files into PyHDX HDX Measurement object'
+class StateParser(object):
+    ""'object used to parse yaml state input files into PyHDX HDX Measurement object'
 
-    def __init__(self, yaml_dict, data_src=None, data_filters=None):
-        self.yaml_dict = yaml_dict
+    # todo yaml_dict -> state_spec
+    def __init__(self, state_spec, data_src=None, data_filters=None):
+        self.state_spec = state_spec
         if isinstance(data_src, (os.PathLike, str)):
             self.data_src = Path(data_src)
         elif isinstance(data_src, dict):
@@ -44,7 +55,7 @@ def load_data(self, *filenames, reader='dynamx'):
     def load_hdxmset(self):
         """batch read the full yaml spec into a hdxmeasurementset"""
         hdxm_list = []
-        for state in self.yaml_dict.keys():
+        for state in self.state_spec.keys():
             hdxm = self.load_hdxm(state, name=state)
             hdxm_list.append(hdxm)
 
@@ -55,7 +66,7 @@ def load_hdxm(self, state, **kwargs):
         kwargs: additional kwargs passed to hdxmeasurementset
         """
 
-        state_dict = self.yaml_dict[state]
+        state_dict = self.state_spec[state]
 
         filenames = state_dict["filenames"]
         df = self.load_data(*filenames)
@@ -95,8 +106,8 @@ def load_hdxm(self, state, **kwargs):
             raise ValueError("Must specify either 'c_term' or 'sequence'")
 
         state_data = pmt.get_state(state_dict["state"])
-        for filter in self.data_filters:
-            state_data = filter(state_data)
+        for flt in self.data_filters:
+            state_data = flt(state_data)
 
         hdxm = HDXMeasurement(
             state_data,
@@ -111,16 +122,170 @@ def load_hdxm(self, state, **kwargs):
         return hdxm
 
 
+process_functions = {
+    'csv_to_dataframe': csv_to_dataframe,
+    'fit_rates_half_time_interpolate': fit_rates_half_time_interpolate,
+    'fit_rates_weighted_average': fit_rates_weighted_average,
+    'fit_gibbs_global': fit_gibbs_global
+
+}
+
+# task objects should be param
+class Task(param.Parameterized):
+    ...
+
+    scheduler_address = param.String(doc='Optional scheduler adress for dask task')
+
+    cwd = param.ClassSelector(Path, doc='Path of the current working directory')
+
+
+class LoadHDMeasurementSetTask(Task):
+    _type = 'load_hdxm_set'
+
+    state_file = param.String()  # = string path
+
+    out = param.ClassSelector(HDXMeasurementSet)
+
+    def execute(self, *args, **kwargs):
+        print(self.cwd)
+        state_spec = yaml.safe_load((self.cwd / self.state_file).read_text())
+        parser = StateParser(state_spec, self.cwd, default_filters)
+        hdxm_set = parser.load_hdxmset()
+
+        self.out = hdxm_set
+
+
+class EstimateRates(Task):
+    _type = 'estimate_rates'
+
+    hdxm_set = param.ClassSelector(HDXMeasurementSet)
+
+    select_state = param.String(doc='If set, only use this state for creating initial guesses')
+
+    out = param.ClassSelector((RatesFitResult, GenericFitResult))
+
+    def execute(self, *args, **kwargs):
+        if self.select_state: # refactor to 'state' ?
+            hdxm = self.hdxm_set.get(self.select_state)
+            result = fit_rates_half_time_interpolate(hdxm)
+        else:
+            results = []
+            for hdxm in self.hdxm_set:
+                r = fit_rates_half_time_interpolate(hdxm)
+                results.append(r)
+            result = RatesFitResult(results)
+
+        self.out = result
+
+
+# todo allow guesses from deltaG
+class ProcessGuesses(Task):
+    _type = 'create_guess'
+
+    hdxm_set = param.ClassSelector(HDXMeasurementSet)
+
+    select_state = param.String(doc='If set, only use this state for creating initial guesses')
+
+    rates_df = param.ClassSelector(pd.DataFrame)
+
+    out = param.ClassSelector((pd.Series, pd.DataFrame))
+
+    def execute(self, *args, **kwargs):
+        if self.select_state:
+            hdxm = self.hdxm_set.get(self.select_state)
+            if self.rates_df.columns.nlevels == 2:
+                rates_series = self.rates_df[(self.select_state, 'rate')]
+            else:
+                rates_series = self.rates_df['rate']
+
+            guess = hdxm.guess_deltaG(rates_series)
+
+        else:
+            rates = self.rates_df.xs('rate', level=-1, axis=1)
+            guess = self.hdxm_set.guess_deltaG(rates)
+
+        self.out = guess
+
+
+class FitGlobalBatch(Task):
+    _type = 'fit_global_batch'
+
+    hdxm_set = param.ClassSelector(HDXMeasurementSet)
+
+    initial_guess = param.ClassSelector(
+        (pd.Series, pd.DataFrame), doc='Initial guesses for fits')
+
+    out = param.ClassSelector(TorchFitResult)
+
+    def execute(self, *args, **kwargs):
+        result = fit_gibbs_global_batch(self.hdxm_set, self.initial_guess, **kwargs)
+
+        self.out = result
+
+
+class SaveFitResult(Task):
+    _type = 'save_fit_result'
+
+    fit_result = param.ClassSelector(TorchFitResult)
+
+    output_dir = param.String()
+
+    def execute(self, *args, **kwargs):
+        save_fitresult(self.cwd / self.output_dir, self.fit_result)
+
+
+class JobParser(object):
+
+    cwd = param.ClassSelector(Path, doc='Path of the current working directory')
+
+    def __init__(self, job_spec, cwd=None, ):
+        self.job_spec = job_spec
+        self.cwd = cwd or Path().cwd()
+
+        self.tasks = {}
+        self.task_classes = {cls._type: cls for cls in gen_subclasses(Task) if getattr(cls, "_type", None)}
+
+    def resolve_var(self, var_string):
+        task_name, *attrs = var_string.split('.')
+
+        return reduce(getattr, attrs, self.tasks[task_name])
+
+    def execute(self):
+
+        for task_spec in self.job_spec['steps']:
+            task_klass = self.task_classes[task_spec['task']]
+            skip = {'args', 'kwargs', 'task'}
+
+            resolved_params = {}
+            for par_name in task_spec.keys() - skip:
+                value = task_spec[par_name]
+                if isinstance(value, str):
+                    m = re.findall(r'\$\((.*?)\)', value)
+                    if m:
+                        value = self.resolve_var(m[0])
+                resolved_params[par_name] = value
+            task = task_klass(cwd=self.cwd, **resolved_params)
+            task.execute(*task_spec.get('args', []), **task_spec.get('kwargs', {}))
+
+            self.tasks[task.name] = task
+
+
 def yaml_to_hdxmset(yaml_dict, data_dir=None, **kwargs):
     """reads files according to `yaml_dict` spec from `data_dir into HDXMEasurementSet"""
 
+    warnings.warn("yaml_to_hdxmset is deprecated, use 'StateParser'")
     hdxm_list = []
     for k, v in yaml_dict.items():
         hdxm = yaml_to_hdxm(v, data_dir=data_dir, name=k)
         hdxm_list.append(hdxm)
 
     return HDXMeasurementSet(hdxm_list)
 
+# todo configurable
+default_filters = [
+    lambda df: df.query('exposure > 0')
+]
+
 
 def yaml_to_hdxm(yaml_dict, data_dir=None, data_filters=None, **kwargs):
     # todo perhas classmethod on HDXMeasurement object?
@@ -142,7 +307,7 @@ def yaml_to_hdxm(yaml_dict, data_dir=None, data_filters=None, **kwargs):
         Output data object as specified by `yaml_dict`.
     """
 
-    warnings.warn('This method is deprecated in favor of YamlParser', DeprecationWarning)
+    warnings.warn('This method is deprecated in favor of StateParser', DeprecationWarning)
 
     if data_dir is not None:
         input_files = [Path(data_dir) / fname for fname in yaml_dict["filenames"]]
@@ -270,3 +435,5 @@ def load_from_yaml_v040b2(yaml_dict, data_dir=None, **kwargs):  # pragma: no cov
     )
 
     return hdxm
+
+

pyhdx/fileIO.py

@@ -371,7 +371,7 @@ def save_fitresult(output_dir, fit_result, log_lines=None):
     dataframe_to_file(output_dir / "losses.csv", fit_result.losses)
     dataframe_to_file(output_dir / "losses.txt", fit_result.losses, fmt="pprint")
 
-    if isinstance(fit_result.hdxm_set, pyhdx.HDXMeasurement):
+    if isinstance(fit_result.hdxm_set, pyhdx.HDXMeasurement): # check, but this should always be hdxm_set
         fit_result.hdxm_set.to_file(output_dir / "HDXMeasurement.csv")
     if isinstance(fit_result.hdxm_set, pyhdx.HDXMeasurementSet):
         fit_result.hdxm_set.to_file(output_dir / "HDXMeasurements.csv")

pyhdx/fitting.py

@@ -929,6 +929,7 @@ class GenericFitResult:
 
 @dataclass
 class RatesFitResult:
+    """Accumulates multiple Generic/KineticsFit Results"""
     results: list
 
     @property

pyhdx/models.py

@@ -1219,6 +1219,12 @@ def __iter__(self):
     def __getitem__(self, item):
         return self.hdxm_list.__getitem__(item)
 
+    def get(self, name):
+        """find a HDXMeasurement by name"""
+
+        idx = self.names.index(name)
+        return self[idx]
+
     @property
     def Ns(self):
         return len(self.hdxm_list)

pyhdx/web/controllers.py

@@ -20,7 +20,7 @@
 from proplot import to_hex
 from skimage.filters import threshold_multiotsu
 
-from pyhdx.batch_processing import YamlParser
+from pyhdx.batch_processing import StateParser
 from pyhdx.config import cfg
 from pyhdx.fileIO import read_dynamx, csv_to_dataframe, dataframe_to_stringio
 from pyhdx.fitting import (
@@ -499,7 +499,7 @@ def _add_dataset_batch(self):
         ios = {name: StringIO(byte_content.decode("UTF-8")) for name, byte_content in zip(self.widgets['input_files'].filename, self.input_files)}
         filters = [lambda df: df.query('exposure > 0')]
 
-        parser = YamlParser(yaml_dict, data_src=ios, data_filters=filters)
+        parser = StateParser(yaml_dict, data_src=ios, data_filters=filters)
 
         for state in yaml_dict.keys():
             hdxm = parser.load_hdxm(state, name=state)

setup.cfg

@@ -34,6 +34,8 @@ install_requires =
     dask
     distributed
     packaging
+    param
+
 python_requires =
     >=3.8
 
@@ -46,7 +48,6 @@ console_scripts =
 web =
     panel>=0.12.6
     bokeh
-    param
     holoviews
     colorcet >= 3.0.0
     hvplot

templates/02_guesses_from_yaml.py

@@ -1,5 +1,5 @@
 """Load HDX-MS data from yaml spec and perform initial guess of exchange rates"""
-from pyhdx.batch_processing import YamlParser
+from pyhdx.batch_processing import StateParser
 from pathlib import Path
 from pyhdx.fitting import fit_rates_weighted_average
 import yaml
@@ -16,7 +16,7 @@
 # Requires local_cluster.py to be running (or other Dask client on default address in config)
 client = default_client()
 
-parser = YamlParser(data_dict, data_src=data_dir)
+parser = StateParser(data_dict, data_src=data_dir)
 for name in data_dict.keys():
     print(name)
     dic = data_dict[name]

templates/06_fitting_with_logs.py

@@ -1,5 +1,5 @@
 """Perform fitting with a range of regularizers"""
-from pyhdx.batch_processing import yaml_to_hdxmset, YamlParser
+from pyhdx.batch_processing import yaml_to_hdxmset, StateParser
 from pathlib import Path
 from pyhdx.fitting import fit_gibbs_global_batch
 import yaml
@@ -20,7 +20,7 @@
 output_dir = current_dir / 'fit'
 output_dir.mkdir(exist_ok=True)
 
-parser = YamlParser(data_dict, data_src=input_dir)
+parser = StateParser(data_dict, data_src=input_dir)
 hdx_set = parser.load_hdxmset()
 
 rates_list = [csv_to_protein(current_dir / 'guesses' / f'{name}_rates_guess.csv')['rate'] for name in data_dict.keys()]

templates/12_jobfiles.py

@@ -0,0 +1,27 @@
+"""
+Execute a PyHDX data processing pipeline according to a yaml jobfile specification
+
+
+"""
+
+
+from pathlib import Path
+from pyhdx.batch_processing import JobParser
+import yaml
+
+#%%
+# Pycharm scientific mode
+if '__file__' not in locals():
+    __file__ = Path().cwd() / 'templates' / 'script.py'
+
+current_dir = Path(__file__).parent
+output_dir = current_dir / 'output'
+output_dir.mkdir(exist_ok=True)
+test_data_dir = current_dir.parent / 'tests' / 'test_data'
+input_dir = test_data_dir / 'input'
+
+#%%
+
+job_spec = yaml.safe_load((input_dir / 'jobfile.yaml').read_text())
+job_parser = JobParser(job_spec, cwd=input_dir)
+job_parser.execute()

tests/test_batchprocessing.py

@@ -1,4 +1,4 @@
-from pyhdx.batch_processing import yaml_to_hdxm, yaml_to_hdxmset, YamlParser
+from pyhdx.batch_processing import yaml_to_hdxm, yaml_to_hdxmset, StateParser
 from pyhdx.models import HDXMeasurement, HDXMeasurementSet
 import numpy as np
 from pathlib import Path
@@ -27,7 +27,7 @@ def test_load_from_yaml(self):
         assert isinstance(hdxm_set, HDXMeasurementSet)
         assert hdxm_set.names == list(data_dict.keys())
 
-        parser = YamlParser(data_dict, data_src=input_dir)
+        parser = StateParser(data_dict, data_src=input_dir)
 
         hdxm = parser.load_hdxm('SecB_tetramer')
         assert isinstance(hdxm, HDXMeasurement)