Skills › Scientific Agent Skills › pyopenms › consensus_to_matrix.py
consensus_to_matrix.py
70 lines · skills/pyopenms/scripts/consensus_to_matrix.py at 5f2bc03
1#!/usr/bin/env python3
2"""
3Consensus Map to Quantification Matrix
5Convert a consensusXML into analysis-ready tables: a wide intensity matrix
6(consensus features x samples) joined with feature metadata (RT, m/z, charge,
7quality), plus an optional long/tidy format. Optionally normalize intensities
8across samples (median or quantile).
10Usage:
11 python consensus_to_matrix.py study.consensusXML --out quant.csv
12 python consensus_to_matrix.py study.consensusXML --out quant.csv --normalize median
13 python consensus_to_matrix.py study.consensusXML --out quant.csv --long long.csv
14"""
16import argparse
17import os
18import sys
20try:
21 import pyopenms as ms
22except ImportError:
23 print("Error: pyopenms not installed. Install with: uv pip install pyopenms")
24 sys.exit(1)
27def main():
28 parser = argparse.ArgumentParser(description="Export a consensusXML to a quant matrix CSV.")
29 parser.add_argument("input", help="Input consensusXML")
30 parser.add_argument("--out", required=True, help="Output wide-matrix CSV")
31 parser.add_argument("--long", help="Also write a tidy/long-format CSV")
32 parser.add_argument("--normalize", choices=["median", "quantile", "none"], default="none",
33 help="Cross-sample intensity normalization (default none)")
34 args = parser.parse_args()
36 if not os.path.exists(args.input):
37 print(f"Error: file not found: {args.input}")
38 sys.exit(1)
40 cm = ms.ConsensusMap()
41 ms.ConsensusXMLFile().load(args.input, cm)
42 print(f"Loaded {cm.size()} consensus features, {len(cm.getColumnHeaders())} samples")
44 if args.normalize == "median":
45 ms.ConsensusMapNormalizerAlgorithmMedian().normalizeMaps(
46 cm, ms.NormalizationMethod.NM_SCALE, "", "")
47 print("Applied median normalization")
48 elif args.normalize == "quantile":
49 ms.ConsensusMapNormalizerAlgorithmQuantile().normalizeMaps(cm)
50 print("Applied quantile normalization")
52 import pandas as pd
53 # Ensure unique row identity; consensus ids may be 0/duplicated on some maps.
54 cm.setUniqueIds()
55 intensity_df = cm.get_intensity_df().reset_index(drop=True)
56 meta_df = cm.get_metadata_df().reset_index(drop=True)
57 matrix = pd.concat([meta_df, intensity_df], axis=1)
58 matrix.to_csv(args.out, index_label="consensus_id")
59 print(f"Wrote {args.out} ({matrix.shape[0]} features x {intensity_df.shape[1]} samples)")
61 if args.long:
62 long_df = intensity_df.copy()
63 long_df.insert(0, "consensus_id", range(len(long_df)))
64 long_df = long_df.melt(id_vars="consensus_id", var_name="sample", value_name="intensity")
65 long_df.to_csv(args.long, index=False)
66 print(f"Wrote {args.long} ({len(long_df)} rows)")
69if __name__ == "__main__":
70 main()