Code Editor for app.py

import streamlit as st
import pyopenms as oms
import pymzml
import plotly.graph_objects as go
import pandas as pd
import numpy as np
import tempfile
import os
from pathlib import Path

st.set_page_config(page_title="Mass Spectrometry Analyzer", layout="wide")
st.title("Mass Spectrometry Data Analysis")

# Initialize session state
if 'current_exp' not in st.session_state:
    st.session_state.current_exp = None
if 'feature_map' not in st.session_state:
    st.session_state.feature_map = None
if 'identified_peptides' not in st.session_state:
    st.session_state.identified_peptides = []

def process_mzml(uploaded_file):
    with tempfile.NamedTemporaryFile(delete=False, suffix='.mzML') as tmp_file:
        tmp_file.write(uploaded_file.getvalue())
        tmp_path = tmp_file.name

    exp = oms.MSExperiment()
    oms.MzMLFile().load(tmp_path, exp)
    os.unlink(tmp_path)  # Clean up temp file
    return exp

def plot_chromatogram(exp):
    rts = []
    intensities = []
    for spectrum in exp:
        if spectrum.getMSLevel() == 1:  # MS1 level
            rts.append(spectrum.getRT())
            intensities.append(sum(spectrum.get_peaks()[1]))
    
    fig = go.Figure()
    fig.add_trace(go.Scatter(x=rts, y=intensities, mode='lines', name='TIC'))
    fig.update_layout(
        title="Total Ion Chromatogram",
        xaxis_title="Retention Time (seconds)",
        yaxis_title="Intensity",
        showlegend=True
    )
    return fig

def plot_spectrum(spectrum):
    mzs, intensities = spectrum.get_peaks()
    
    fig = go.Figure()
    fig.add_trace(go.Scatter(x=mzs, y=intensities, mode='lines', name='Mass Spectrum'))
    fig.update_layout(
        title=f"MS Level {spectrum.getMSLevel()} Spectrum (RT: {spectrum.getRT():.2f}s)",
        xaxis_title="m/z",
        yaxis_title="Intensity",
        showlegend=True
    )
    return fig

def feature_detection(exp):
    # Parameters for feature finding
    params = oms.FeatureFindingMetaboParams()
    params.setDefaults()
    
    # Run feature finding
    feature_finder = oms.FeatureFindingMetabo()
    feature_finder.setParameters(params)
    
    feature_map = oms.FeatureMap()
    feature_finder.run(exp, feature_map)
    
    return feature_map

# Sidebar for file upload and processing options
with st.sidebar:
    st.header("Data Input")
    uploaded_file = st.file_uploader("Upload mzML file", type=['mzML'])
    
    if uploaded_file:
        if st.button("Process File"):
            with st.spinner("Processing mzML file..."):
                st.session_state.current_exp = process_mzml(uploaded_file)
                st.success("File processed successfully!")
    
    if st.session_state.current_exp is not None:
        if st.button("Run Feature Detection"):
            with st.spinner("Detecting features..."):
                st.session_state.feature_map = feature_detection(st.session_state.current_exp)
                st.success(f"Found {st.session_state.feature_map.size()} features!")

# Main content area
if st.session_state.current_exp is None:
    st.info("Please upload an mzML file to begin analysis.")
else:
    tabs = st.tabs(["Chromatogram", "Spectra", "Features"])
    
    with tabs[0]:
        st.plotly_chart(plot_chromatogram(st.session_state.current_exp), use_container_width=True)
    
    with tabs[1]:
        # Spectrum viewer
        spec_idx = st.slider("Select spectrum index", 
                           0, 
                           st.session_state.current_exp.size()-1, 
                           0)
        spectrum = st.session_state.current_exp[spec_idx]
        st.plotly_chart(plot_spectrum(spectrum), use_container_width=True)
        
        # Spectrum metadata
        st.subheader("Spectrum Metadata")
        col1, col2 = st.columns(2)
        with col1:
            st.write(f"MS Level: {spectrum.getMSLevel()}")
            st.write(f"RT: {spectrum.getRT():.2f} seconds")
        with col2:
            st.write(f"Precursor m/z: {spectrum.getPrecursors()[0].getMZ() if spectrum.getPrecursors() else 'N/A'}")
            st.write(f"# of peaks: {spectrum.size()}")
    
    with tabs[2]:
        if st.session_state.feature_map is None:
            st.info("Run feature detection to view features.")
        else:
            # Display feature table
            features_data = []
            for feature in st.session_state.feature_map:
                features_data.append({
                    'RT': feature.getRT(),
                    'MZ': feature.getMZ(),
                    'Intensity': feature.getIntensity(),
                    'Charge': feature.getCharge(),
                    'Quality': feature.getOverallQuality()
                })
            
            features_df = pd.DataFrame(features_data)
            st.dataframe(features_df)