diff --git a/README.md b/README.md index 5716afc..176edeb 100644 --- a/README.md +++ b/README.md @@ -299,6 +299,75 @@ peaks_df = spec.to_dataframe() print(peaks_df.head()) ``` +### Working with Chromatograms + +Chromatograms (e.g., extracted ion chromatograms, total ion chromatograms) are now fully supported with a Pythonic interface similar to spectra. + +```python +from openms_python import Py_MSExperiment, Py_MSChromatogram +import pandas as pd + +# Load experiment with chromatograms +exp = Py_MSExperiment.from_file('data.mzML') + +# Check chromatogram count +print(f"Chromatograms: {exp.chromatogram_count}") + +# Access individual chromatograms +if exp.chromatogram_count > 0: + chrom = exp.get_chromatogram(0) + print(f"MZ: {chrom.mz:.4f}") + print(f"Name: {chrom.name}") + print(f"Data points: {len(chrom)}") + print(f"RT range: {chrom.rt_range}") + print(f"TIC: {chrom.total_ion_current:.2e}") + +# Iterate over all chromatograms +for chrom in exp.chromatograms(): + print(f"Chromatogram: MZ={chrom.mz:.4f}, Points={len(chrom)}") + +# Create chromatogram from DataFrame +df = pd.DataFrame({ + 'rt': [10.0, 20.0, 30.0, 40.0], + 'intensity': [1000.0, 5000.0, 3000.0, 500.0] +}) +chrom = Py_MSChromatogram.from_dataframe( + df, + mz=445.12, + name="XIC m/z 445.12", + native_id="chromatogram=1" +) + +# Add to experiment +exp.add_chromatogram(chrom) + +# Chromatogram properties +print(f"MZ: {chrom.mz:.4f}") +print(f"Max intensity: {chrom.max_intensity}") +print(f"RT at max: {chrom.rt[chrom.intensity.argmax()]:.2f}") + +# Get data as arrays +rt, intensity = chrom.data +# Or individually +rt_values = chrom.rt +intensity_values = chrom.intensity + +# Convert to DataFrame +chrom_df = chrom.to_dataframe() + +# Filter chromatogram by RT +filtered = chrom.filter_by_rt(min_rt=15.0, max_rt=35.0) + +# Normalize intensities +normalized = chrom.normalize_intensity(max_value=100.0) +normalized_tic = chrom.normalize_to_tic() + +# Metadata access +chrom["sample_id"] = "Sample_A" +chrom["replicate"] = 1 +print(chrom.get("sample_id")) + + ### Ion Mobility Support `openms-python` provides comprehensive support for ion mobility data through float data arrays and mobilograms. @@ -738,6 +807,8 @@ plt.show() **Properties:** - `n_spectra`: Number of spectra +- `nr_chromatograms`: Number of chromatograms +- `chromatogram_count`: Alias for nr_chromatograms - `rt_range`: Tuple of (min_rt, max_rt) - `ms_levels`: Set of MS levels present @@ -750,12 +821,42 @@ plt.show() - `ms2_spectra()`: Iterator over MS2 spectra - `spectra_by_level(level)`: Iterator over specific MS level - `spectra_in_rt_range(min_rt, max_rt)`: Iterator over RT range +- `get_chromatogram(index)`: Get chromatogram by index +- `chromatograms()`: Iterator over all chromatograms +- `add_chromatogram(chromatogram)`: Add a chromatogram to the experiment - `filter_by_ms_level(level)`: Filter by MS level - `filter_by_rt(min_rt, max_rt)`: Filter by RT range - `filter_top_n_peaks(n)`: Keep top N peaks per spectrum - `summary()`: Get summary statistics - `print_summary()`: Print formatted summary +### Py_MSChromatogram + +**Properties:** +- `mz`: M/z value for the chromatogram +- `name`: Chromatogram name +- `native_id`: Native chromatogram ID +- `chromatogram_type`: Type of chromatogram +- `rt_range`: Tuple of (min_rt, max_rt) +- `min_rt`: Minimum retention time +- `max_rt`: Maximum retention time +- `min_intensity`: Minimum intensity +- `max_intensity`: Maximum intensity +- `total_ion_current`: Sum of all intensities +- `data`: Tuple of (rt_array, intensity_array) +- `rt`: RT values as NumPy array +- `intensity`: Intensity values as NumPy array + +**Methods:** +- `from_numpy(rt, intensity, **metadata)`: Create from NumPy arrays (class method) +- `from_dataframe(df, **metadata)`: Create from DataFrame (class method) +- `to_numpy()`: Convert to NumPy arrays +- `to_dataframe()`: Convert to DataFrame +- `filter_by_rt(min_rt, max_rt)`: Filter data points by RT range +- `filter_by_intensity(min_intensity)`: Filter by minimum intensity +- `normalize_intensity(max_value)`: Normalize intensities to max value +- `normalize_to_tic()`: Normalize so intensities sum to 1.0 + ### Py_MSSpectrum **Properties:** @@ -835,10 +936,13 @@ pip install -e ".[dev]" | Feature | pyOpenMS | openms-python | |---------|----------|---------------| | Get spectrum count | `exp.getNrSpectra()` | `len(exp)` | +| Get chromatogram count | `exp.getNrChromatograms()` | `exp.chromatogram_count` | | Get retention time | `spec.getRT()` | `spec.retention_time` | +| Get chromatogram m/z | `chrom.getMZ()` | `chrom.mz` | | Check MS1 | `spec.getMSLevel() == 1` | `spec.is_ms1` | | Load file | `MzMLFile().load(path, exp)` | `exp = MSExperiment.from_file(path)` | | Iterate MS1 | Manual loop + level check | `for spec in exp.ms1_spectra():` | +| Iterate chromatograms | Manual loop + range check | `for chrom in exp.chromatograms():` | | Peak data | `peaks = spec.get_peaks(); mz = peaks[0]` | `mz, intensity = spec.peaks` | | DataFrame | Not available | `df = exp.to_dataframe()` | diff --git a/openms_python/__init__.py b/openms_python/__init__.py index fe9bfaa..9c49e04 100644 --- a/openms_python/__init__.py +++ b/openms_python/__init__.py @@ -23,6 +23,7 @@ from .py_msexperiment import Py_MSExperiment from .py_msspectrum import Py_MSSpectrum +from .py_chromatogram import Py_MSChromatogram from .py_mobilogram import Py_Mobilogram from .py_feature import Py_Feature from .py_featuremap import Py_FeatureMap @@ -100,6 +101,7 @@ def get_example(name: str, *, load: bool = False, target_dir: Union[str, Path, N __all__ = [ "Py_MSExperiment", "Py_MSSpectrum", + "Py_MSChromatogram", "Py_Mobilogram", "Py_Feature", "Py_FeatureMap", diff --git a/openms_python/py_chromatogram.py b/openms_python/py_chromatogram.py new file mode 100644 index 0000000..98f49bb --- /dev/null +++ b/openms_python/py_chromatogram.py @@ -0,0 +1,391 @@ +""" +Pythonic wrapper for pyOpenMS MSChromatogram class. +""" + +from typing import Tuple, Optional +import numpy as np +import pandas as pd +import pyopenms as oms + +from ._meta_mapping import MetaInfoMappingMixin + + +class Py_MSChromatogram(MetaInfoMappingMixin): + """ + A Pythonic wrapper around pyOpenMS MSChromatogram. + + This class provides intuitive properties and methods for working with + chromatograms, hiding the verbose C++ API underneath. + + Example: + >>> chrom = Py_MSChromatogram(native_chromatogram) + >>> print(f"MZ: {chrom.mz:.4f}") + >>> print(f"Name: {chrom.name}") + >>> print(f"Number of data points: {len(chrom)}") + >>> peaks_df = chrom.to_dataframe() + """ + + def __init__(self, native_chromatogram: oms.MSChromatogram): + """ + Initialize Chromatogram wrapper. + + Args: + native_chromatogram: pyOpenMS MSChromatogram object + """ + self._chromatogram = native_chromatogram + + # ==================== Meta-info support ==================== + + def _meta_object(self) -> oms.MetaInfoInterface: + return self._chromatogram + + # ==================== Pythonic Properties ==================== + + @property + def mz(self) -> float: + """Get m/z value for this chromatogram.""" + return self._chromatogram.getMZ() + + @mz.setter + def mz(self, value: float): + """Set m/z value for this chromatogram.""" + product = self._chromatogram.getProduct() + product.setMZ(value) + self._chromatogram.setProduct(product) + + @property + def name(self) -> str: + """Get name of the chromatogram.""" + return self._chromatogram.getName() + + @name.setter + def name(self, value: str): + """Set name of the chromatogram.""" + self._chromatogram.setName(value) + + @property + def native_id(self) -> str: + """Get native ID of the chromatogram.""" + return self._chromatogram.getNativeID() + + @native_id.setter + def native_id(self, value: str): + """Set native ID of the chromatogram.""" + self._chromatogram.setNativeID(value) + + @property + def chromatogram_type(self) -> int: + """Get chromatogram type.""" + return self._chromatogram.getChromatogramType() + + @chromatogram_type.setter + def chromatogram_type(self, value: int): + """Set chromatogram type.""" + self._chromatogram.setChromatogramType(value) + + @property + def rt_range(self) -> Tuple[float, float]: + """Get retention time range (min, max).""" + if len(self) == 0: + return (0.0, 0.0) + rt, _ = self.data + return (float(np.min(rt)), float(np.max(rt))) + + @property + def min_rt(self) -> float: + """Get minimum retention time.""" + if len(self) == 0: + return 0.0 + return float(self._chromatogram.getMinRT()) + + @property + def max_rt(self) -> float: + """Get maximum retention time.""" + if len(self) == 0: + return 0.0 + return float(self._chromatogram.getMaxRT()) + + @property + def max_intensity(self) -> float: + """Get maximum intensity.""" + if len(self) == 0: + return 0.0 + return float(self._chromatogram.getMaxIntensity()) + + @property + def min_intensity(self) -> float: + """Get minimum intensity.""" + if len(self) == 0: + return 0.0 + return float(self._chromatogram.getMinIntensity()) + + @property + def total_ion_current(self) -> float: + """Get total ion current (sum of all intensities).""" + _, intensities = self.data + return float(np.sum(intensities)) + + @property + def data(self) -> Tuple[np.ndarray, np.ndarray]: + """ + Get chromatogram data as NumPy arrays. + + Returns: + Tuple of (rt_array, intensity_array) + """ + rt, intensity = self._chromatogram.get_peaks() + return np.array(rt), np.array(intensity) + + @data.setter + def data(self, values: Tuple[np.ndarray, np.ndarray]): + """ + Set chromatogram data from NumPy arrays. + + Args: + values: Tuple of (rt_array, intensity_array) + """ + rt, intensity = values + self._chromatogram.set_peaks((rt.tolist(), intensity.tolist())) + + @property + def rt(self) -> np.ndarray: + """Get retention time values as a NumPy array.""" + rt, _ = self.data + return rt + + @property + def intensity(self) -> np.ndarray: + """Get intensity values as a NumPy array.""" + _, intensity = self.data + return intensity + + # ==================== Magic Methods ==================== + + def __len__(self) -> int: + """Return number of data points in the chromatogram.""" + return self._chromatogram.size() + + def __repr__(self) -> str: + """Return string representation.""" + mz_str = f"m/z={self.mz:.4f}" if self.mz > 0 else "m/z=N/A" + name_str = f", name='{self.name}'" if self.name else "" + return ( + f"Chromatogram({mz_str}{name_str}, " + f"points={len(self)}, TIC={self.total_ion_current:.2e})" + ) + + def __str__(self) -> str: + """Return human-readable string.""" + return self.__repr__() + + def __iter__(self): + """Allow dict(self) and list(self) conversions.""" + yield "rt", self.rt.tolist() + yield "intensity", self.intensity.tolist() + + # ==================== Conversion Methods ==================== + + def to_numpy(self) -> np.ndarray: + """ + Convert chromatogram data to NumPy arrays. + + Returns: + Tuple of (rt_array, intensity_array) + + Example: + >>> rt, intensity = chrom.to_numpy() + """ + return np.array(self.data) + + def to_dataframe(self) -> pd.DataFrame: + """ + Convert chromatogram data to pandas DataFrame. + + Returns: + DataFrame with columns: rt, intensity + + Example: + >>> df = chrom.to_dataframe() + >>> df.head() + rt intensity + 0 10.050 1250.5 + 1 10.100 5678.2 + ... + """ + rt, intensity = self.data + return pd.DataFrame({ + 'rt': rt, + 'intensity': intensity + }) + + @classmethod + def from_numpy(cls, rt: np.ndarray, intensity: np.ndarray, **metadata) -> 'Py_MSChromatogram': + """ + Create chromatogram from NumPy arrays. + + Args: + rt: Array of retention time values + intensity: Array of intensity values + **metadata: Optional metadata (mz, name, native_id, etc.) + + Returns: + Py_MSChromatogram object + """ + chrom = oms.MSChromatogram() + chrom.set_peaks((rt.tolist(), intensity.tolist())) + + # Set metadata + if 'mz' in metadata: + product = chrom.getProduct() + product.setMZ(metadata['mz']) + chrom.setProduct(product) + if 'name' in metadata: + chrom.setName(metadata['name']) + if 'native_id' in metadata: + chrom.setNativeID(metadata['native_id']) + if 'chromatogram_type' in metadata: + chrom.setChromatogramType(metadata['chromatogram_type']) + + return cls(chrom) + + @classmethod + def from_dataframe(cls, df: pd.DataFrame, **metadata) -> 'Py_MSChromatogram': + """ + Create chromatogram from pandas DataFrame. + + Args: + df: DataFrame with 'rt' and 'intensity' columns + **metadata: Optional metadata (mz, name, native_id, etc.) + + Returns: + Py_MSChromatogram object + + Example: + >>> df = pd.DataFrame({'rt': [10.0, 20.0], 'intensity': [50, 100]}) + >>> chrom = Py_MSChromatogram.from_dataframe(df, mz=445.12, name='XIC') + """ + chrom = oms.MSChromatogram() + chrom.set_peaks((df['rt'].values.tolist(), df['intensity'].values.tolist())) + + # Set metadata + if 'mz' in metadata: + product = chrom.getProduct() + product.setMZ(metadata['mz']) + chrom.setProduct(product) + if 'name' in metadata: + chrom.setName(metadata['name']) + if 'native_id' in metadata: + chrom.setNativeID(metadata['native_id']) + if 'chromatogram_type' in metadata: + chrom.setChromatogramType(metadata['chromatogram_type']) + + return cls(chrom) + + # ==================== Data Manipulation ==================== + + def filter_by_rt(self, min_rt: float, max_rt: float) -> 'Py_MSChromatogram': + """ + Filter data points by retention time range. + + Args: + min_rt: Minimum retention time + max_rt: Maximum retention time + + Returns: + New Py_MSChromatogram with filtered data points + """ + rt, intensity = self.data + mask = (rt >= min_rt) & (rt <= max_rt) + + new_chrom = oms.MSChromatogram() + new_chrom.set_peaks((rt[mask].tolist(), intensity[mask].tolist())) + + # Copy metadata + product = self._chromatogram.getProduct() + new_chrom.setProduct(product) + new_chrom.setName(self.name) + new_chrom.setNativeID(self.native_id) + new_chrom.setChromatogramType(self.chromatogram_type) + + return Py_MSChromatogram(new_chrom) + + def filter_by_intensity(self, min_intensity: float) -> 'Py_MSChromatogram': + """ + Filter data points by minimum intensity. + + Args: + min_intensity: Minimum intensity threshold + + Returns: + New Py_MSChromatogram with filtered data points + """ + rt, intensity = self.data + mask = intensity >= min_intensity + + new_chrom = oms.MSChromatogram() + new_chrom.set_peaks((rt[mask].tolist(), intensity[mask].tolist())) + + # Copy metadata + product = self._chromatogram.getProduct() + new_chrom.setProduct(product) + new_chrom.setName(self.name) + new_chrom.setNativeID(self.native_id) + new_chrom.setChromatogramType(self.chromatogram_type) + + return Py_MSChromatogram(new_chrom) + + def normalize_to_tic(self) -> 'Py_MSChromatogram': + """ + Scale intensities so their sum equals one. + + Returns: + New Py_MSChromatogram with normalized intensities + """ + rt, intensity = self.data + total = float(np.sum(intensity)) + if total <= 0: + return self + + normalized_chrom = oms.MSChromatogram(self._chromatogram) + normalized_chrom.set_peaks((rt.tolist(), (intensity / total).tolist())) + return Py_MSChromatogram(normalized_chrom) + + def normalize_intensity(self, max_value: float = 100.0) -> 'Py_MSChromatogram': + """ + Normalize intensities to a maximum value. + + Args: + max_value: Target maximum intensity (default: 100.0) + + Returns: + New Py_MSChromatogram with normalized intensities + """ + rt, intensity = self.data + if len(intensity) == 0 or np.max(intensity) == 0: + return self + + normalized = intensity * (max_value / np.max(intensity)) + + new_chrom = oms.MSChromatogram() + new_chrom.set_peaks((rt.tolist(), normalized.tolist())) + + # Copy metadata + product = self._chromatogram.getProduct() + new_chrom.setProduct(product) + new_chrom.setName(self.name) + new_chrom.setNativeID(self.native_id) + new_chrom.setChromatogramType(self.chromatogram_type) + + return Py_MSChromatogram(new_chrom) + + # ==================== Access to Native Object ==================== + + @property + def native(self) -> oms.MSChromatogram: + """ + Get the underlying pyOpenMS MSChromatogram object. + + Use this when you need to access pyOpenMS-specific methods + not wrapped by this class. + """ + return self._chromatogram diff --git a/openms_python/py_msexperiment.py b/openms_python/py_msexperiment.py index 030d2a0..a501cf0 100644 --- a/openms_python/py_msexperiment.py +++ b/openms_python/py_msexperiment.py @@ -8,6 +8,7 @@ import numpy as np import pyopenms as oms from .py_msspectrum import Py_MSSpectrum +from .py_chromatogram import Py_MSChromatogram from .py_featuremap import Py_FeatureMap from ._io_utils import ensure_allowed_suffix, MS_EXPERIMENT_EXTENSIONS @@ -266,6 +267,16 @@ def rt_filter(self) -> _Py_MSExperimentRTSlicing: """ return _Py_MSExperimentRTSlicing(self) + @property + def nr_chromatograms(self) -> int: + """Get number of chromatograms in the experiment.""" + return self._experiment.getNrChromatograms() + + @property + def chromatogram_count(self) -> int: + """Alias for nr_chromatograms.""" + return self.nr_chromatograms + # ==================== Magic Methods ==================== def __len__(self) -> int: @@ -281,8 +292,9 @@ def __repr__(self) -> str: """Return string representation.""" ms_levels_str = ', '.join(f"MS{level}" for level in sorted(self.ms_levels)) rt_min, rt_max = self.rt_range + chrom_str = f", chromatograms={self.nr_chromatograms}" if self.nr_chromatograms > 0 else "" return ( - f"MSExperiment(spectra={len(self)}, " + f"MSExperiment(spectra={len(self)}{chrom_str}, " f"levels=[{ms_levels_str}], " f"rt_range=[{rt_min:.2f}, {rt_max:.2f}]s)" ) @@ -344,6 +356,52 @@ def spectra_in_rt_range(self, min_rt: float, max_rt: float) -> Iterator[Py_MSSpe if min_rt <= spec.retention_time <= max_rt: yield spec + # ==================== Chromatogram Access ==================== + + def get_chromatogram(self, index: int) -> Py_MSChromatogram: + """ + Get a chromatogram by index. + + Args: + index: Chromatogram index (0-based) + + Returns: + Py_MSChromatogram wrapper + + Example: + >>> chrom = exp.get_chromatogram(0) + >>> print(f"MZ: {chrom.mz:.4f}, Points: {len(chrom)}") + """ + native_chrom = self._experiment.getChromatogram(index) + return Py_MSChromatogram(native_chrom) + + def chromatograms(self) -> Iterator[Py_MSChromatogram]: + """ + Iterate over all chromatograms. + + Example: + >>> for chrom in exp.chromatograms(): + ... print(f"MZ: {chrom.mz:.4f}, TIC: {chrom.total_ion_current:.2e}") + """ + for i in range(self.nr_chromatograms): + yield self.get_chromatogram(i) + + def add_chromatogram(self, chromatogram: Union[Py_MSChromatogram, oms.MSChromatogram]): + """ + Add a chromatogram to the experiment. + + Args: + chromatogram: Chromatogram to add (Py_MSChromatogram or native) + + Example: + >>> chrom = Py_MSChromatogram.from_dataframe(df, mz=445.12) + >>> exp.add_chromatogram(chrom) + """ + if isinstance(chromatogram, Py_MSChromatogram): + self._experiment.addChromatogram(chromatogram.native) + else: + self._experiment.addChromatogram(chromatogram) + # ==================== DataFrame Conversion ==================== def to_dataframe(self, include_peaks: bool = True, ms_level: Optional[int] = None) -> pd.DataFrame: diff --git a/tests/test_msexperiment_chromatograms.py b/tests/test_msexperiment_chromatograms.py new file mode 100644 index 0000000..4a85c7f --- /dev/null +++ b/tests/test_msexperiment_chromatograms.py @@ -0,0 +1,154 @@ +""" +Tests for MSExperiment chromatogram integration. +""" + +import pytest +import numpy as np +import pandas as pd +import pyopenms as oms + +from openms_python import Py_MSExperiment, Py_MSChromatogram + + +def test_msexperiment_chromatogram_count(): + """Test chromatogram count properties.""" + exp = Py_MSExperiment() + + # Initially empty + assert exp.nr_chromatograms == 0 + assert exp.chromatogram_count == 0 + + # Add a chromatogram + chrom = Py_MSChromatogram(oms.MSChromatogram()) + exp.add_chromatogram(chrom) + + assert exp.nr_chromatograms == 1 + assert exp.chromatogram_count == 1 + + +def test_msexperiment_add_chromatogram(): + """Test adding chromatograms to experiment.""" + exp = Py_MSExperiment() + + # Create chromatogram from DataFrame + df = pd.DataFrame({ + 'rt': [10.0, 20.0, 30.0], + 'intensity': [100.0, 200.0, 150.0] + }) + chrom = Py_MSChromatogram.from_dataframe(df, mz=445.12, name="XIC 445.12") + + # Add it + exp.add_chromatogram(chrom) + + assert exp.nr_chromatograms == 1 + + # Add another + chrom2 = Py_MSChromatogram.from_dataframe(df, mz=500.0, name="XIC 500.0") + exp.add_chromatogram(chrom2) + + assert exp.nr_chromatograms == 2 + + +def test_msexperiment_get_chromatogram(): + """Test getting chromatograms by index.""" + exp = Py_MSExperiment() + + # Add chromatograms + df1 = pd.DataFrame({'rt': [10.0, 20.0], 'intensity': [100.0, 200.0]}) + chrom1 = Py_MSChromatogram.from_dataframe(df1, mz=445.12, name="First") + exp.add_chromatogram(chrom1) + + df2 = pd.DataFrame({'rt': [30.0, 40.0], 'intensity': [150.0, 250.0]}) + chrom2 = Py_MSChromatogram.from_dataframe(df2, mz=500.0, name="Second") + exp.add_chromatogram(chrom2) + + # Get them back + retrieved1 = exp.get_chromatogram(0) + assert retrieved1.mz == pytest.approx(445.12) + assert retrieved1.name == "First" + assert len(retrieved1) == 2 + + retrieved2 = exp.get_chromatogram(1) + assert retrieved2.mz == pytest.approx(500.0) + assert retrieved2.name == "Second" + assert len(retrieved2) == 2 + + +def test_msexperiment_chromatograms_iteration(): + """Test iterating over chromatograms.""" + exp = Py_MSExperiment() + + # Add multiple chromatograms + mz_values = [445.12, 500.0, 550.25] + for mz in mz_values: + df = pd.DataFrame({'rt': [10.0, 20.0, 30.0], 'intensity': [100.0, 200.0, 150.0]}) + chrom = Py_MSChromatogram.from_dataframe(df, mz=mz) + exp.add_chromatogram(chrom) + + # Iterate and check + retrieved_mzs = [] + for chrom in exp.chromatograms(): + assert isinstance(chrom, Py_MSChromatogram) + assert len(chrom) == 3 + retrieved_mzs.append(chrom.mz) + + assert len(retrieved_mzs) == 3 + np.testing.assert_array_almost_equal(retrieved_mzs, mz_values) + + +def test_msexperiment_repr_with_chromatograms(): + """Test string representation includes chromatogram count.""" + exp = Py_MSExperiment() + + # Add a spectrum + spec = oms.MSSpectrum() + spec.setRT(10.0) + spec.setMSLevel(1) + spec.set_peaks(([100.0, 200.0], [50.0, 100.0])) + exp._experiment.addSpectrum(spec) + + # Initially no chromatograms + repr_str = repr(exp) + assert "chromatograms" not in repr_str + + # Add a chromatogram + df = pd.DataFrame({'rt': [10.0, 20.0], 'intensity': [100.0, 200.0]}) + chrom = Py_MSChromatogram.from_dataframe(df, mz=445.12) + exp.add_chromatogram(chrom) + + repr_str = repr(exp) + assert "chromatograms=1" in repr_str + + +def test_msexperiment_add_native_chromatogram(): + """Test adding native pyOpenMS chromatogram.""" + exp = Py_MSExperiment() + + # Create native chromatogram + native_chrom = oms.MSChromatogram() + native_chrom.set_peaks(([10.0, 20.0, 30.0], [100.0, 200.0, 150.0])) + product = native_chrom.getProduct() + product.setMZ(445.12) + native_chrom.setProduct(product) + + # Add it directly + exp.add_chromatogram(native_chrom) + + assert exp.nr_chromatograms == 1 + + # Retrieve and verify + retrieved = exp.get_chromatogram(0) + assert retrieved.mz == pytest.approx(445.12) + assert len(retrieved) == 3 + + +def test_msexperiment_empty_chromatograms(): + """Test experiment with no chromatograms.""" + exp = Py_MSExperiment() + + # Should not raise error + assert exp.nr_chromatograms == 0 + + # Should return empty iterator + chroms = list(exp.chromatograms()) + assert len(chroms) == 0 diff --git a/tests/test_py_chromatogram.py b/tests/test_py_chromatogram.py new file mode 100644 index 0000000..edbf684 --- /dev/null +++ b/tests/test_py_chromatogram.py @@ -0,0 +1,285 @@ +""" +Tests for Py_MSChromatogram wrapper. +""" + +import pytest +import numpy as np +import pandas as pd +import pyopenms as oms + +from openms_python import Py_MSChromatogram + + +def test_py_chromatogram_properties(): + """Test basic properties of Py_MSChromatogram.""" + # Create a native chromatogram + native_chrom = oms.MSChromatogram() + product = native_chrom.getProduct() + product.setMZ(445.12) + native_chrom.setProduct(product) + native_chrom.setName("XIC 445.12") + native_chrom.setNativeID("chromatogram=1") + native_chrom.set_peaks(([10.0, 20.0, 30.0], [100.0, 200.0, 150.0])) + + # Wrap it + chrom = Py_MSChromatogram(native_chrom) + + # Test properties + assert chrom.mz == pytest.approx(445.12) + assert chrom.name == "XIC 445.12" + assert chrom.native_id == "chromatogram=1" + assert len(chrom) == 3 + + # Test RT range + assert chrom.min_rt == pytest.approx(10.0) + assert chrom.max_rt == pytest.approx(30.0) + assert chrom.rt_range == (pytest.approx(10.0), pytest.approx(30.0)) + + # Test intensity range + assert chrom.max_intensity == pytest.approx(200.0) + assert chrom.min_intensity == pytest.approx(100.0) + assert chrom.total_ion_current == pytest.approx(450.0) + + +def test_py_chromatogram_data_access(): + """Test data access methods.""" + native_chrom = oms.MSChromatogram() + native_chrom.set_peaks(([10.0, 20.0, 30.0], [100.0, 200.0, 150.0])) + + chrom = Py_MSChromatogram(native_chrom) + + # Test data property + rt, intensity = chrom.data + assert isinstance(rt, np.ndarray) + assert isinstance(intensity, np.ndarray) + assert len(rt) == 3 + assert len(intensity) == 3 + np.testing.assert_array_almost_equal(rt, [10.0, 20.0, 30.0]) + np.testing.assert_array_almost_equal(intensity, [100.0, 200.0, 150.0]) + + # Test individual accessors + np.testing.assert_array_almost_equal(chrom.rt, [10.0, 20.0, 30.0]) + np.testing.assert_array_almost_equal(chrom.intensity, [100.0, 200.0, 150.0]) + + +def test_py_chromatogram_setters(): + """Test property setters.""" + chrom = Py_MSChromatogram(oms.MSChromatogram()) + + # Test setters + chrom.mz = 500.25 + assert chrom.mz == pytest.approx(500.25) + + chrom.name = "Test Chromatogram" + assert chrom.name == "Test Chromatogram" + + chrom.native_id = "chrom=test" + assert chrom.native_id == "chrom=test" + + # Test data setter + chrom.data = (np.array([1.0, 2.0, 3.0]), np.array([10.0, 20.0, 30.0])) + rt, intensity = chrom.data + np.testing.assert_array_almost_equal(rt, [1.0, 2.0, 3.0]) + np.testing.assert_array_almost_equal(intensity, [10.0, 20.0, 30.0]) + + +def test_py_chromatogram_from_dataframe(): + """Test creating chromatogram from DataFrame.""" + df = pd.DataFrame({ + 'rt': [10.0, 20.0, 30.0], + 'intensity': [100.0, 200.0, 150.0] + }) + + chrom = Py_MSChromatogram.from_dataframe( + df, + mz=445.12, + name="Test XIC", + native_id="chrom=1" + ) + + assert len(chrom) == 3 + assert chrom.mz == pytest.approx(445.12) + assert chrom.name == "Test XIC" + assert chrom.native_id == "chrom=1" + + rt, intensity = chrom.data + np.testing.assert_array_almost_equal(rt, [10.0, 20.0, 30.0]) + np.testing.assert_array_almost_equal(intensity, [100.0, 200.0, 150.0]) + + +def test_py_chromatogram_to_dataframe(): + """Test converting chromatogram to DataFrame.""" + native_chrom = oms.MSChromatogram() + native_chrom.set_peaks(([10.0, 20.0, 30.0], [100.0, 200.0, 150.0])) + + chrom = Py_MSChromatogram(native_chrom) + df = chrom.to_dataframe() + + assert isinstance(df, pd.DataFrame) + assert 'rt' in df.columns + assert 'intensity' in df.columns + assert len(df) == 3 + + np.testing.assert_array_almost_equal(df['rt'].values, [10.0, 20.0, 30.0]) + np.testing.assert_array_almost_equal(df['intensity'].values, [100.0, 200.0, 150.0]) + + +def test_py_chromatogram_dataframe_roundtrip(): + """Test DataFrame round-trip conversion.""" + original_df = pd.DataFrame({ + 'rt': [10.5, 20.3, 30.1], + 'intensity': [123.4, 234.5, 178.9] + }) + + chrom = Py_MSChromatogram.from_dataframe(original_df, mz=500.0) + result_df = chrom.to_dataframe() + + # Check values with tolerance for float32/float64 conversion + np.testing.assert_array_almost_equal(result_df['rt'].values, original_df['rt'].values, decimal=5) + np.testing.assert_array_almost_equal(result_df['intensity'].values, original_df['intensity'].values, decimal=5) + + +def test_py_chromatogram_from_numpy(): + """Test creating chromatogram from NumPy arrays.""" + rt = np.array([10.0, 20.0, 30.0]) + intensity = np.array([100.0, 200.0, 150.0]) + + chrom = Py_MSChromatogram.from_numpy( + rt, intensity, + mz=445.12, + name="NumPy Chromatogram" + ) + + assert len(chrom) == 3 + assert chrom.mz == pytest.approx(445.12) + assert chrom.name == "NumPy Chromatogram" + + rt_out, intensity_out = chrom.data + np.testing.assert_array_almost_equal(rt_out, rt) + np.testing.assert_array_almost_equal(intensity_out, intensity) + + +def test_py_chromatogram_filter_by_rt(): + """Test filtering by retention time.""" + native_chrom = oms.MSChromatogram() + native_chrom.set_peaks(([10.0, 20.0, 30.0, 40.0], [100.0, 200.0, 150.0, 120.0])) + product = native_chrom.getProduct() + product.setMZ(445.12) + native_chrom.setProduct(product) + + chrom = Py_MSChromatogram(native_chrom) + filtered = chrom.filter_by_rt(15.0, 35.0) + + assert len(filtered) == 2 + rt, intensity = filtered.data + np.testing.assert_array_almost_equal(rt, [20.0, 30.0]) + np.testing.assert_array_almost_equal(intensity, [200.0, 150.0]) + + # Ensure metadata is preserved + assert filtered.mz == pytest.approx(445.12) + + +def test_py_chromatogram_filter_by_intensity(): + """Test filtering by intensity.""" + native_chrom = oms.MSChromatogram() + native_chrom.set_peaks(([10.0, 20.0, 30.0, 40.0], [100.0, 200.0, 150.0, 80.0])) + native_chrom.setName("Test") + + chrom = Py_MSChromatogram(native_chrom) + filtered = chrom.filter_by_intensity(100.0) + + assert len(filtered) == 3 + rt, intensity = filtered.data + np.testing.assert_array_almost_equal(rt, [10.0, 20.0, 30.0]) + np.testing.assert_array_almost_equal(intensity, [100.0, 200.0, 150.0]) + + # Ensure metadata is preserved + assert filtered.name == "Test" + + +def test_py_chromatogram_normalize_intensity(): + """Test intensity normalization.""" + native_chrom = oms.MSChromatogram() + native_chrom.set_peaks(([10.0, 20.0, 30.0], [100.0, 200.0, 150.0])) + + chrom = Py_MSChromatogram(native_chrom) + normalized = chrom.normalize_intensity(max_value=1000.0) + + assert len(normalized) == 3 + rt, intensity = normalized.data + np.testing.assert_array_almost_equal(rt, [10.0, 20.0, 30.0]) + assert intensity.max() == pytest.approx(1000.0) + assert intensity[0] == pytest.approx(500.0) # 100/200 * 1000 + assert intensity[2] == pytest.approx(750.0) # 150/200 * 1000 + + +def test_py_chromatogram_normalize_to_tic(): + """Test normalization to TIC.""" + native_chrom = oms.MSChromatogram() + native_chrom.set_peaks(([10.0, 20.0, 30.0], [100.0, 200.0, 150.0])) + + chrom = Py_MSChromatogram(native_chrom) + normalized = chrom.normalize_to_tic() + + assert len(normalized) == 3 + assert normalized.total_ion_current == pytest.approx(1.0) + + +def test_py_chromatogram_empty(): + """Test behavior with empty chromatogram.""" + chrom = Py_MSChromatogram(oms.MSChromatogram()) + + assert len(chrom) == 0 + assert chrom.total_ion_current == 0.0 + assert chrom.rt_range == (0.0, 0.0) + assert chrom.min_rt == 0.0 + assert chrom.max_rt == 0.0 + + df = chrom.to_dataframe() + assert len(df) == 0 + + +def test_py_chromatogram_repr(): + """Test string representation.""" + native_chrom = oms.MSChromatogram() + product = native_chrom.getProduct() + product.setMZ(445.12) + native_chrom.setProduct(product) + native_chrom.setName("XIC") + native_chrom.set_peaks(([10.0, 20.0], [100.0, 200.0])) + + chrom = Py_MSChromatogram(native_chrom) + repr_str = repr(chrom) + + assert "445.12" in repr_str + assert "XIC" in repr_str + assert "points=2" in repr_str + + +def test_py_chromatogram_meta_mapping(): + """Test dictionary-style meta data access.""" + chrom = Py_MSChromatogram(oms.MSChromatogram()) + + # Set and get meta values + chrom["custom_label"] = "test_sample" + chrom["scan_rate"] = 5.0 + + assert chrom["custom_label"] == "test_sample" + assert chrom["scan_rate"] == 5.0 + + # Test 'in' operator + assert "custom_label" in chrom + assert "nonexistent" not in chrom + + # Test get with default + assert chrom.get("custom_label") == "test_sample" + assert chrom.get("missing", "default") == "default" + + +def test_py_chromatogram_native_access(): + """Test access to native pyOpenMS object.""" + native_chrom = oms.MSChromatogram() + chrom = Py_MSChromatogram(native_chrom) + + assert chrom.native is native_chrom + assert isinstance(chrom.native, oms.MSChromatogram)