corems.mass_spectra.input.parserbase

  1from abc import ABC, abstractmethod
  2import datetime
  3import numbers
  4from typing import Optional, Union, List, Tuple
  5
  6
  7class SpectraParserInterface(ABC):
  8    """
  9    Interface for parsing mass spectra data into MassSpectraBase objects.
 10
 11    Methods
 12    -------
 13    * load().
 14        Load mass spectra data.
 15    * run().
 16        Parse mass spectra data.
 17    * get_mass_spectra_obj().
 18        Return MassSpectraBase object with several attributes populated
 19    * get_mass_spectrum_from_scan(scan_number).
 20        Return MassSpecBase data object from scan number.
 21    * get_scans_in_time_range(time_range).
 22        Return scan numbers within specified retention time range(s).
 23
 24    Notes
 25    -----
 26    This is an abstract class and should not be instantiated directly.
 27    
 28    Time Range Filtering
 29    --------------------
 30    Many methods support optional time_range parameter to load only scans within
 31    specific retention time windows. This significantly improves performance for
 32    targeted workflows. Time ranges can be specified as:
 33    - Single range: (start_time, end_time) in minutes
 34    - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
 35    """
 36
 37    @abstractmethod
 38    def load(self):
 39        """
 40        Load mass spectra data.
 41        """
 42        pass
 43
 44    @abstractmethod
 45    def run(self):
 46        """
 47        Parse mass spectra data.
 48        """
 49        pass
 50
 51    @abstractmethod
 52    def get_scan_df(self, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
 53        """
 54        Return scan data as a pandas DataFrame.
 55        
 56        Parameters
 57        ----------
 58        time_range : tuple or list of tuples, optional
 59            Retention time range(s) to filter scans. Can be:
 60            - Single range: (start_time, end_time) in minutes
 61            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
 62            If None, returns all scans.
 63        
 64        Returns
 65        -------
 66        pd.DataFrame
 67            DataFrame containing scan information, optionally filtered by time range.
 68        """
 69        pass
 70
 71    @abstractmethod
 72    def get_ms_raw(self, spectra, scan_df, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
 73        """
 74        Return a dictionary of mass spectra data as pandas DataFrames.
 75        
 76        Parameters
 77        ----------
 78        spectra : str or dict
 79            Specifies which spectra to load (e.g., 'ms1', 'ms2', or custom dict)
 80        scan_df : pd.DataFrame
 81            Scan information DataFrame
 82        time_range : tuple or list of tuples, optional
 83            Retention time range(s) to filter scans. Can be:
 84            - Single range: (start_time, end_time) in minutes
 85            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
 86            If None, returns all scans.
 87        
 88        Returns
 89        -------
 90        dict
 91            Dictionary of raw mass spectra data, optionally filtered by time range.
 92        """
 93        pass
 94
 95    @abstractmethod
 96    def get_mass_spectra_obj(self, time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]] = None):
 97        """
 98        Return mass spectra data object.
 99        
100        Parameters
101        ----------
102        time_range : tuple or list of tuples, optional
103            Retention time range(s) to load. Can be:
104            - Single range: (start_time, end_time) in minutes
105            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
106            If None, loads all scans. Useful for targeted workflows to improve performance.
107        
108        Returns
109        -------
110        MassSpectraBase
111            Mass spectra object, optionally filtered to specified time range(s).
112        """
113        pass
114
115    @abstractmethod
116    def get_mass_spectrum_from_scan(
117        self, scan_number, spectrum_mode, auto_process=True
118    ):
119        """
120        Return mass spectrum data object from scan number.
121        """
122        pass
123
124    @abstractmethod
125    def get_mass_spectra_from_scan_list(
126        self, scan_list, spectrum_mode, auto_process=True
127    ):
128        """
129        Return a list of mass spectrum data objects from a list of scan numbers.
130        """
131        pass
132
133    @abstractmethod
134    def get_scans_in_time_range(
135        self, 
136        time_range: Union[Tuple[float, float], List[Tuple[float, float]]],
137        ms_level: Optional[int] = None
138    ) -> List[int]:
139        """
140        Return scan numbers within specified retention time range(s).
141        
142        This method provides efficient filtering of scans by retention time,
143        which is particularly useful for targeted workflows where only specific
144        time windows are of interest.
145        
146        Parameters
147        ----------
148        time_range : tuple or list of tuples
149            Retention time range(s) in minutes. Can be:
150            - Single range: (start_time, end_time)
151            - Multiple ranges: [(start1, end1), (start2, end2), ...]
152        ms_level : int, optional
153            If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2).
154            If None, returns scans of all MS levels.
155        
156        Returns
157        -------
158        list of int
159            List of scan numbers within the specified time range(s) and MS level.
160        
161        Examples
162        --------
163        Get MS1 scans between 1.0 and 2.0 minutes:
164        
165        >>> scans = parser.get_scans_in_time_range((1.0, 2.0), ms_level=1)
166        
167        Get scans in multiple time windows:
168        
169        >>> scans = parser.get_scans_in_time_range([(0.5, 1.5), (3.0, 4.0)])
170        """
171        pass
172
173    @abstractmethod
174    def get_instrument_info(self):
175        """
176        Return instrument information.
177
178        Returns
179        -------
180        dict
181            A dictionary with the keys 'model', and 'serial_number'.
182        """
183        pass
184
185    @abstractmethod
186    def get_creation_time(self) -> datetime.datetime:
187        """
188        Return the creation time of the mass spectra data.
189
190        Returns
191        -------
192        datetime.datetime
193            The creation time of the mass spectra data.
194        """
195        pass
196    
197    @staticmethod
198    def _normalize_time_range(
199        time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]]
200    ) -> Optional[List[Tuple[float, float]]]:
201        """
202        Normalize time range input to a list of tuples.
203        
204        Helper method for implementations to standardize time_range parameter.
205        Converts single tuple to list of tuples for consistent handling.
206        
207        Parameters
208        ----------
209        time_range : tuple, list of tuples, or None
210            Input time range(s)
211        
212        Returns
213        -------
214        list of tuples or None
215            Normalized time ranges as list of (start, end) tuples, or None if input is None.
216        
217        Examples
218        --------
219        >>> SpectraParserInterface._normalize_time_range((1.0, 2.0))
220        [(1.0, 2.0)]
221        
222        >>> SpectraParserInterface._normalize_time_range([(1.0, 2.0), (3.0, 4.0)])
223        [(1.0, 2.0), (3.0, 4.0)]
224        
225        >>> SpectraParserInterface._normalize_time_range(None)
226        None
227        """
228        if time_range is None:
229            return None
230        
231        # Check if it's a single tuple (two numbers)
232        if isinstance(time_range, tuple) and len(time_range) == 2:
233            # Use numbers.Number to catch int, float, and numpy scalar types
234            if isinstance(time_range[0], numbers.Number) and isinstance(time_range[1], numbers.Number):
235                # Convert to float to ensure consistency (handles numpy scalars)
236                return [(float(time_range[0]), float(time_range[1]))]
237        
238        # Otherwise assume it's already a list of tuples
239        return list(time_range)