tessl/pypi-pyopenms

Python wrapper for C++ LC-MS library OpenMS for comprehensive mass spectrometry data analysis

—

Pending

Overview

Eval results

Files

Map Alignment and Consensus Features

Name: tessl/pypi-pyopenms
Author: tessl

Retention time alignment algorithms and consensus feature generation for comparative analysis across multiple LC-MS experiments. Essential for label-free quantification and multi-sample proteomics studies.

Capabilities

Map Alignment

Pose Clustering Alignment

Advanced alignment algorithm using pose clustering for robust retention time correction.

class MapAlignmentAlgorithmPoseClustering:
    def __init__(self) -> None: ...
    
    def align(self, maps: list[FeatureMap], transformations: list[TransformationDescription]) -> None:
        """
        Align multiple feature maps using pose clustering.
        
        Args:
            maps (list[FeatureMap]): Feature maps to align
            transformations (list[TransformationDescription]): Output transformations
        """
    
    def getParameters(self) -> Param:
        """
        Get alignment parameters.
        
        Returns:
            Param: Parameter configuration
        """
    
    def setParameters(self, param: Param) -> None:
        """
        Set alignment parameters.
        
        Args:
            param (Param): Parameter configuration
        """
    
    def setReference(self, reference: FeatureMap) -> None:
        """
        Set reference map for alignment.
        
        Args:
            reference (FeatureMap): Reference feature map
        """

class MapAlignmentAlgorithmIdentification:
    def __init__(self) -> None: ...
    
    def align(self, maps: list[FeatureMap], transformations: list[TransformationDescription]) -> None:
        """
        Align maps using peptide identifications.
        
        Args:
            maps (list[FeatureMap]): Feature maps with identifications
            transformations (list[TransformationDescription]): Output transformations
        """
    
    def getParameters(self) -> Param:
        """Get alignment parameters."""
    
    def setParameters(self, param: Param) -> None:
        """Set alignment parameters."""

Map Alignment Transformer

Apply retention time transformations to align data.

class MapAlignmentTransformer:
    def __init__(self) -> None: ...
    
    @staticmethod
    def transformRetentionTimes(map: FeatureMap, trafo: TransformationDescription, 
                                store_original_rt: bool = False) -> None:
        """
        Transform retention times in feature map.
        
        Args:
            map (FeatureMap): Feature map to transform
            trafo (TransformationDescription): Transformation to apply
            store_original_rt (bool): Store original RT as meta value
        """
    
    @staticmethod
    def transformRetentionTimes(exp: MSExperiment, trafo: TransformationDescription, 
                                store_original_rt: bool = False) -> None:
        """
        Transform retention times in MS experiment.
        
        Args:
            exp (MSExperiment): Experiment to transform
            trafo (TransformationDescription): Transformation to apply
            store_original_rt (bool): Store original RT as meta value
        """
    
    @staticmethod
    def transformRetentionTimes(peptide_ids: list[PeptideIdentification], 
                                trafo: TransformationDescription, 
                                store_original_rt: bool = False) -> None:
        """
        Transform retention times in peptide identifications.
        
        Args:
            peptide_ids (list[PeptideIdentification]): IDs to transform
            trafo (TransformationDescription): Transformation to apply
            store_original_rt (bool): Store original RT as meta value
        """

Transformation Models

TransformationDescription

Container for retention time transformation metadata and models.

class TransformationDescription:
    def __init__(self) -> None: ...
    
    def getDataPoints(self) -> list[tuple[float, float]]:
        """
        Get transformation data points.
        
        Returns:
            list[tuple[float, float]]: List of (input_rt, output_rt) pairs
        """
    
    def setDataPoints(self, data: list[tuple[float, float]]) -> None:
        """
        Set transformation data points.
        
        Args:
            data (list[tuple[float, float]]): (input_rt, output_rt) pairs
        """
    
    def apply(self, value: float) -> float:
        """
        Apply transformation to retention time value.
        
        Args:
            value (float): Input retention time
            
        Returns:
            float: Transformed retention time
        """
    
    def getModelType(self) -> str:
        """
        Get transformation model type.
        
        Returns:
            str: Model type ("linear", "b_spline", etc.)
        """
    
    def setModelType(self, type: str) -> None:
        """
        Set transformation model type.
        
        Args:
            type (str): Model type
        """
    
    def fitModel(self, type: str, params: Param = None) -> None:
        """
        Fit transformation model to data points.
        
        Args:
            type (str): Model type to fit
            params (Param): Model parameters (optional)
        """
    
    def invert(self) -> TransformationDescription:
        """
        Get inverted transformation.
        
        Returns:
            TransformationDescription: Inverted transformation
        """

class TransformationModel:
    def __init__(self, data: list[tuple[float, float]], params: Param) -> None:
        """
        Base transformation model.
        
        Args:
            data (list[tuple[float, float]]): Training data points
            params (Param): Model parameters
        """
    
    def evaluate(self, value: float) -> float:
        """
        Evaluate transformation at given point.
        
        Args:
            value (float): Input value
            
        Returns:
            float: Transformed value
        """

class TransformationModelLinear(TransformationModel):
    def __init__(self, data: list[tuple[float, float]], params: Param) -> None:
        """
        Linear transformation model.
        
        Args:
            data (list[tuple[float, float]]): Training data points
            params (Param): Model parameters
        """

class TransformationModelBSpline(TransformationModel):
    def __init__(self, data: list[tuple[float, float]], params: Param) -> None:
        """
        B-spline transformation model.
        
        Args:
            data (list[tuple[float, float]]): Training data points
            params (Param): Model parameters including wavelength and boundary
        """

Consensus Features

ConsensusMap

Container for consensus features across multiple experiments.

class ConsensusMap:
    def __init__(self) -> None: ...
    
    def size(self) -> int:
        """
        Get number of consensus features.
        
        Returns:
            int: Number of consensus features
        """
    
    def empty(self) -> bool:
        """
        Check if consensus map is empty.
        
        Returns:
            bool: True if empty
        """
    
    def push_back(self, feature: ConsensusFeature) -> None:
        """
        Add consensus feature.
        
        Args:
            feature (ConsensusFeature): Consensus feature to add
        """
    
    def __getitem__(self, index: int) -> ConsensusFeature:
        """
        Get consensus feature by index.
        
        Args:
            index (int): Feature index
            
        Returns:
            ConsensusFeature: Consensus feature
        """
    
    def clear(self) -> None:
        """Remove all consensus features."""
    
    def getColumnHeaders(self) -> dict[int, ColumnHeader]:
        """
        Get column headers describing input maps.
        
        Returns:
            dict[int, ColumnHeader]: Map index to column header mapping
        """
    
    def setColumnHeaders(self, headers: dict[int, ColumnHeader]) -> None:
        """
        Set column headers.
        
        Args:
            headers (dict[int, ColumnHeader]): Column headers
        """
    
    def getExperimentType(self) -> str:
        """
        Get experiment type.
        
        Returns:
            str: Experiment type ("label-free", "itraq", etc.)
        """
    
    def setExperimentType(self, type: str) -> None:
        """
        Set experiment type.
        
        Args:
            type (str): Experiment type
        """
    
    def sortByRT(self) -> None:
        """Sort consensus features by retention time."""
    
    def sortByMZ(self) -> None:
        """Sort consensus features by m/z."""
    
    def sortByIntensity(self, reverse: bool = True) -> None:
        """
        Sort by intensity.
        
        Args:
            reverse (bool): Sort in descending order
        """
    
    def updateRanges(self) -> None:
        """Update RT/m/z ranges from features."""
    
    def get_intensity_df(self) -> DataFrame:
        """
        Export intensity data to pandas DataFrame.
        
        Returns:
            DataFrame: Intensity data in long or wide format
        """
    
    def get_metadata_df(self) -> DataFrame:
        """
        Export feature metadata to pandas DataFrame.
        
        Returns:
            DataFrame: Feature metadata (RT, m/z, quality, etc.)
        """
    
    def get_df(self) -> DataFrame:
        """
        Export complete consensus data to pandas DataFrame.
        
        Returns:
            DataFrame: Combined metadata and intensity data
        """

class ColumnHeader:
    def __init__(self) -> None: ...
    
    def filename(self) -> str:
        """
        Get filename.
        
        Returns:
            str: Input file name
        """
    
    def setFilename(self, filename: str) -> None:
        """
        Set filename.
        
        Args:
            filename (str): Input file name
        """
    
    def label(self) -> str:
        """
        Get label.
        
        Returns:
            str: Sample label
        """
    
    def setLabel(self, label: str) -> None:
        """
        Set label.
        
        Args:
            label (str): Sample label
        """
    
    def size(self) -> int:
        """
        Get number of features in original map.
        
        Returns:
            int: Feature count
        """
    
    def setSize(self, size: int) -> None:
        """
        Set feature count.
        
        Args:
            size (int): Feature count
        """

ConsensusFeature

Individual consensus feature linking features across experiments.

class ConsensusFeature:
    def __init__(self) -> None: ...
    
    def getRT(self) -> float:
        """
        Get consensus retention time.
        
        Returns:
            float: Consensus retention time
        """
    
    def setRT(self, rt: float) -> None:
        """
        Set retention time.
        
        Args:
            rt (float): Retention time
        """
    
    def getMZ(self) -> float:
        """
        Get consensus m/z.
        
        Returns:
            float: Consensus m/z
        """
    
    def setMZ(self, mz: float) -> None:
        """
        Set m/z.
        
        Args:
            mz (float): m/z value
        """
    
    def getIntensity(self) -> float:
        """
        Get consensus intensity.
        
        Returns:
            float: Consensus intensity
        """
    
    def setIntensity(self, intensity: float) -> None:
        """
        Set intensity.
        
        Args:
            intensity (float): Intensity value
        """
    
    def getCharge(self) -> int:
        """
        Get consensus charge.
        
        Returns:
            int: Charge state
        """
    
    def setCharge(self, charge: int) -> None:
        """
        Set charge.
        
        Args:
            charge (int): Charge state
        """
    
    def getQuality(self) -> float:
        """
        Get consensus quality.
        
        Returns:
            float: Quality score
        """
    
    def setQuality(self, quality: float) -> None:
        """
        Set quality.
        
        Args:
            quality (float): Quality score
        """
    
    def getFeatureList(self) -> list[FeatureHandle]:
        """
        Get constituent features.
        
        Returns:
            list[FeatureHandle]: Features from individual maps
        """
    
    def setFeatureList(self, features: list[FeatureHandle]) -> None:
        """
        Set constituent features.
        
        Args:
            features (list[FeatureHandle]): Features from individual maps
        """
    
    def insert(self, handle: FeatureHandle) -> None:
        """
        Add feature handle.
        
        Args:
            handle (FeatureHandle): Feature handle to add
        """
    
    def getUniqueId(self) -> int:
        """
        Get unique consensus feature ID.
        
        Returns:
            int: Unique identifier
        """
    
    def getPeptideIdentifications(self) -> list[PeptideIdentification]:
        """
        Get assigned peptide identifications.
        
        Returns:
            list[PeptideIdentification]: Peptide IDs
        """

class FeatureHandle:
    def __init__(self) -> None: ...
    
    def getRT(self) -> float:
        """Get retention time."""
    
    def getMZ(self) -> float:
        """Get m/z."""
    
    def getIntensity(self) -> float:
        """Get intensity."""
    
    def getCharge(self) -> int:
        """Get charge."""
    
    def getMapIndex(self) -> int:
        """
        Get source map index.
        
        Returns:
            int: Index of source map
        """
    
    def setMapIndex(self, index: int) -> None:
        """
        Set source map index.
        
        Args:
            index (int): Source map index
        """
    
    def getUniqueId(self) -> int:
        """
        Get unique feature ID.
        
        Returns:
            int: Unique ID in source map
        """
    
    def setUniqueId(self, id: int) -> None:
        """
        Set unique feature ID.
        
        Args:
            id (int): Unique ID
        """

Feature Linking

FeatureGroupingAlgorithm

Base class for feature grouping across maps.

class FeatureGroupingAlgorithm:
    def __init__(self) -> None: ...
    
    def group(self, maps: list[FeatureMap], consensus_map: ConsensusMap) -> None:
        """
        Group features across maps.
        
        Args:
            maps (list[FeatureMap]): Input feature maps
            consensus_map (ConsensusMap): Output consensus map
        """
    
    def getParameters(self) -> Param:
        """Get grouping parameters."""
    
    def setParameters(self, param: Param) -> None:
        """Set grouping parameters."""

class FeatureGroupingAlgorithmQT:
    def __init__(self) -> None: ...
    
    def group(self, maps: list[FeatureMap], consensus_map: ConsensusMap) -> None:
        """
        Group features using QT clustering.
        
        Args:
            maps (list[FeatureMap]): Input feature maps
            consensus_map (ConsensusMap): Output consensus map
        """
    
    def getParameters(self) -> Param:
        """
        Get QT clustering parameters.
        
        Returns:
            Param: Parameters including distance thresholds
        """

class FeatureGroupingAlgorithmUnlabeled:
    def __init__(self) -> None: ...
    
    def group(self, maps: list[FeatureMap], consensus_map: ConsensusMap) -> None:
        """
        Group features for unlabeled data.
        
        Args:
            maps (list[FeatureMap]): Input feature maps
            consensus_map (ConsensusMap): Output consensus map
        """

Usage Examples

Map Alignment Workflow

import pyopenms

# Load feature maps from multiple runs
feature_maps = []
filenames = ["run1_features.featureXML", "run2_features.featureXML", "run3_features.featureXML"]

for filename in filenames:
    features = pyopenms.FeatureMap()
    pyopenms.FeatureXMLFile().load(filename, features)
    feature_maps.append(features)

print(f"Loaded {len(feature_maps)} feature maps")

# Set up alignment algorithm
alignment = pyopenms.MapAlignmentAlgorithmPoseClustering()

# Configure alignment parameters
params = alignment.getParameters()
params.setValue("max_num_peaks_considered", 1000)
params.setValue("superimposer:mz_pair_max_distance", 0.05)
params.setValue("superimposer:rt_pair_distance_fraction", 0.1)
alignment.setParameters(params)

# Use first map as reference
alignment.setReference(feature_maps[0])

# Perform alignment
transformations = []
alignment.align(feature_maps, transformations)

print(f"Generated {len(transformations)} transformations")

# Apply transformations to feature maps
transformer = pyopenms.MapAlignmentTransformer()
for i, (fmap, trafo) in enumerate(zip(feature_maps[1:], transformations[1:]), 1):
    print(f"Applying transformation to map {i+1}")
    transformer.transformRetentionTimes(fmap, trafo, store_original_rt=True)

# Save aligned feature maps
for i, fmap in enumerate(feature_maps):
    output_file = f"aligned_run{i+1}_features.featureXML"
    pyopenms.FeatureXMLFile().store(output_file, fmap)

Consensus Feature Generation

import pyopenms

# Load aligned feature maps
feature_maps = []
for i in range(3):
    features = pyopenms.FeatureMap()
    pyopenms.FeatureXMLFile().load(f"aligned_run{i+1}_features.featureXML", features)
    feature_maps.append(features)

# Set up feature grouping
grouper = pyopenms.FeatureGroupingAlgorithmQT()

# Configure grouping parameters
params = grouper.getParameters()
params.setValue("distance_RT:max_difference", 30.0)  # 30 seconds RT tolerance
params.setValue("distance_MZ:max_difference", 0.01)   # 0.01 Da m/z tolerance
params.setValue("distance_MZ:unit", "Da")
grouper.setParameters(params)

# Create consensus map
consensus_map = pyopenms.ConsensusMap()

# Set column headers for input maps
headers = {}
for i, fmap in enumerate(feature_maps):
    header = pyopenms.ColumnHeader()
    header.setFilename(f"run{i+1}.mzML")
    header.setLabel(f"run{i+1}")
    header.setSize(fmap.size())
    headers[i] = header

consensus_map.setColumnHeaders(headers)
consensus_map.setExperimentType("label-free")

# Group features
grouper.group(feature_maps, consensus_map)

print(f"Generated {consensus_map.size()} consensus features")

# Save consensus map
pyopenms.ConsensusXMLFile().store("consensus_features.consensusXML", consensus_map)

# Export to DataFrame for analysis
df = consensus_map.get_df()
print(f"Exported consensus data: {df.shape}")
print(df.head())

Analyzing Consensus Results

import pyopenms
import pandas as pd

# Load consensus map
consensus_map = pyopenms.ConsensusMap()
pyopenms.ConsensusXMLFile().load("consensus_features.consensusXML", consensus_map)

# Export intensity data
intensity_df = consensus_map.get_intensity_df()
metadata_df = consensus_map.get_metadata_df()

print("Intensity data shape:", intensity_df.shape)
print("Metadata shape:", metadata_df.shape)

# Analyze feature presence across runs
feature_presence = (intensity_df > 0).sum(axis=1)
print("Features detected in all runs:", (feature_presence == intensity_df.shape[1]).sum())
print("Features detected in at least 2 runs:", (feature_presence >= 2).sum())

# Quality assessment
print("Quality score distribution:")
print(metadata_df['quality'].describe())

# Find high-quality features present in all runs
high_quality_complete = metadata_df[
    (metadata_df['quality'] > 0.7) & 
    (feature_presence == intensity_df.shape[1])
]

print(f"High-quality features in all runs: {len(high_quality_complete)}")

Custom Transformation Models

import pyopenms

# Create transformation description with data points
trafo = pyopenms.TransformationDescription()

# Add alignment points (input_rt, reference_rt)
data_points = [
    (100.0, 105.0),
    (200.0, 210.0),
    (300.0, 315.0),
    (400.0, 420.0),
    (500.0, 525.0)
]

trafo.setDataPoints(data_points)

# Fit different models
print("Linear model:")
trafo.fitModel("linear")
print(f"Transform 250.0 -> {trafo.apply(250.0):.2f}")

print("B-spline model:")
params = pyopenms.Param()
params.setValue("num_breakpoints", 5)
params.setValue("break_positions", "uniform")
trafo.fitModel("b_spline", params)
print(f"Transform 250.0 -> {trafo.apply(250.0):.2f}")

# Apply transformation to feature map
features = pyopenms.FeatureMap()
pyopenms.FeatureXMLFile().load("features.featureXML", features)

transformer = pyopenms.MapAlignmentTransformer()
transformer.transformRetentionTimes(features, trafo, store_original_rt=True)

# Check transformation results
for i in range(min(5, features.size())):
    feature = features[i]
    original_rt = feature.getMetaValue("original_RT") if feature.metaValueExists("original_RT") else "N/A"
    print(f"Feature {i}: RT {feature.getRT():.2f} (original: {original_rt})")

Install with Tessl CLI