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(
 53        self,
 54        time_range: Optional[
 55            Union[Tuple[float, float], List[Tuple[float, float]]]
 56        ] = None,
 57    ):
 58        """
 59        Return scan data as a pandas DataFrame.
 60
 61        Parameters
 62        ----------
 63        time_range : tuple or list of tuples, optional
 64            Retention time range(s) to filter scans. Can be:
 65            - Single range: (start_time, end_time) in minutes
 66            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
 67            If None, returns all scans.
 68
 69        Returns
 70        -------
 71        pd.DataFrame
 72            DataFrame containing scan information, optionally filtered by time range.
 73        """
 74        pass
 75
 76    @abstractmethod
 77    def get_ms_raw(
 78        self,
 79        spectra,
 80        scan_df,
 81        time_range: Optional[
 82            Union[Tuple[float, float], List[Tuple[float, float]]]
 83        ] = None,
 84    ):
 85        """
 86        Return a dictionary of mass spectra data as pandas DataFrames.
 87
 88        Parameters
 89        ----------
 90        spectra : str or dict
 91            Specifies which spectra to load (e.g., 'ms1', 'ms2', or custom dict)
 92        scan_df : pd.DataFrame
 93            Scan information DataFrame
 94        time_range : tuple or list of tuples, optional
 95            Retention time range(s) to filter scans. Can be:
 96            - Single range: (start_time, end_time) in minutes
 97            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
 98            If None, returns all scans.
 99
100        Returns
101        -------
102        dict
103            Dictionary of raw mass spectra data, optionally filtered by time range.
104        """
105        pass
106
107    @abstractmethod
108    def get_mass_spectra_obj(
109        self,
110        time_range: Optional[
111            Union[Tuple[float, float], List[Tuple[float, float]]]
112        ] = None,
113    ):
114        """
115        Return mass spectra data object.
116
117        Parameters
118        ----------
119        time_range : tuple or list of tuples, optional
120            Retention time range(s) to load. Can be:
121            - Single range: (start_time, end_time) in minutes
122            - Multiple ranges: [(start1, end1), (start2, end2), ...] in minutes
123            If None, loads all scans. Useful for targeted workflows to improve performance.
124
125        Returns
126        -------
127        MassSpectraBase
128            Mass spectra object, optionally filtered to specified time range(s).
129        """
130        pass
131
132    @abstractmethod
133    def get_mass_spectrum_from_scan(
134        self, scan_number, spectrum_mode, auto_process=True
135    ):
136        """
137        Return mass spectrum data object from scan number.
138        """
139        pass
140
141    @abstractmethod
142    def get_mass_spectra_from_scan_list(
143        self, scan_list, spectrum_mode, auto_process=True
144    ):
145        """
146        Return a list of mass spectrum data objects from a list of scan numbers.
147        """
148        pass
149
150    @abstractmethod
151    def get_scans_in_time_range(
152        self,
153        time_range: Union[Tuple[float, float], List[Tuple[float, float]]],
154        ms_level: Optional[int] = None,
155    ) -> List[int]:
156        """
157        Return scan numbers within specified retention time range(s).
158
159        This method provides efficient filtering of scans by retention time,
160        which is particularly useful for targeted workflows where only specific
161        time windows are of interest.
162
163        Parameters
164        ----------
165        time_range : tuple or list of tuples
166            Retention time range(s) in minutes. Can be:
167            - Single range: (start_time, end_time)
168            - Multiple ranges: [(start1, end1), (start2, end2), ...]
169        ms_level : int, optional
170            If specified, only return scans of this MS level (e.g., 1 for MS1, 2 for MS2).
171            If None, returns scans of all MS levels.
172
173        Returns
174        -------
175        list of int
176            List of scan numbers within the specified time range(s) and MS level.
177
178        Examples
179        --------
180        Get MS1 scans between 1.0 and 2.0 minutes:
181
182        >>> scans = parser.get_scans_in_time_range((1.0, 2.0), ms_level=1)
183
184        Get scans in multiple time windows:
185
186        >>> scans = parser.get_scans_in_time_range([(0.5, 1.5), (3.0, 4.0)])
187        """
188        pass
189
190    @abstractmethod
191    def get_instrument_info(self):
192        """
193        Return instrument information.
194
195        Returns
196        -------
197        dict
198            A dictionary with the keys 'model', and 'serial_number'.
199        """
200        pass
201
202    @abstractmethod
203    def get_creation_time(self) -> datetime.datetime:
204        """
205        Return the creation time of the mass spectra data.
206
207        Returns
208        -------
209        datetime.datetime
210            The creation time of the mass spectra data.
211        """
212        pass
213
214    @staticmethod
215    def _normalize_time_range(
216        time_range: Optional[Union[Tuple[float, float], List[Tuple[float, float]]]],
217    ) -> Optional[List[Tuple[float, float]]]:
218        """
219        Normalize time range input to a list of tuples.
220
221        Helper method for implementations to standardize time_range parameter.
222        Converts single tuple to list of tuples for consistent handling.
223
224        Parameters
225        ----------
226        time_range : tuple, list of tuples, or None
227            Input time range(s)
228
229        Returns
230        -------
231        list of tuples or None
232            Normalized time ranges as list of (start, end) tuples, or None if input is None.
233
234        Examples
235        --------
236        >>> SpectraParserInterface._normalize_time_range((1.0, 2.0))
237        [(1.0, 2.0)]
238
239        >>> SpectraParserInterface._normalize_time_range([(1.0, 2.0), (3.0, 4.0)])
240        [(1.0, 2.0), (3.0, 4.0)]
241
242        >>> SpectraParserInterface._normalize_time_range(None)
243        None
244        """
245        if time_range is None:
246            return None
247
248        # Check if it's a single tuple (two numbers)
249        if isinstance(time_range, tuple) and len(time_range) == 2:
250            # Use numbers.Number to catch int, float, and numpy scalar types
251            if isinstance(time_range[0], numbers.Number) and isinstance(
252                time_range[1], numbers.Number
253            ):
254                # Convert to float to ensure consistency (handles numpy scalars)
255                return [(float(time_range[0]), float(time_range[1]))]
256
257        # Otherwise assume it's already a list of tuples
258        return list(time_range)