averaging_writer/
averaging_writer.rs

1/*!
2 * Demo how to use the deferred spectrum averaging iterator with `rayon`
3 * to quickly average over an LC-MS run and write the averaged spectra
4 * out to an mzML file on disk.
5 */
6use std::env;
7use std::io;
8use std::path;
9use std::thread;
10use std::time::Instant;
11
12use std::sync::mpsc::sync_channel;
13
14use mzdata::spectrum::SignalContinuity;
15use rayon::prelude::*;
16
17use mzdata::prelude::*;
18use mzdata::spectrum::utils::Collator;
19use mzdata::{MZReader, MzMLWriter};
20
21fn main() -> io::Result<()> {
22    let path = path::PathBuf::from(
23        env::args()
24            .nth(1)
25            .expect("Please pass an MS data file path"),
26    );
27
28    let mut reader = MZReader::open_path(path)?;
29    let mut writer = MzMLWriter::new(io::BufWriter::new(io::stdout()));
30    writer.copy_metadata_from(&reader);
31
32    let (input_sender, input_receiver) = sync_channel(5000);
33    let (output_sender, output_receiver) = sync_channel(5000);
34
35    let start = Instant::now();
36    let reader_task = thread::spawn(move || {
37        let (grouper, averager, _reprofiler) =
38            reader.groups().averaging_deferred(1, 120.0, 2000.1, 0.002);
39        grouper
40            .enumerate()
41            .par_bridge()
42            .map_init(
43                || averager.clone(),
44                |averager, (i, g)| {
45                    let (mut g, arrays) = g.average_with(averager);
46                    if let Some(p) = g.precursor_mut() {
47                        p.arrays = Some(arrays.into());
48                        p.description_mut().signal_continuity = SignalContinuity::Profile;
49                    }
50                    (i, g)
51                },
52            )
53            .for_each(|(i, g)| {
54                input_sender.send((i, g)).unwrap();
55            });
56        let end_read = Instant::now();
57        eprintln!(
58            "Finished reading all spectra and averaging in {:0.3?}",
59            end_read - start
60        );
61    });
62
63    let collator_task =
64        thread::spawn(move || Collator::collate_sync(input_receiver, output_sender));
65
66    let writer_task = thread::spawn(move || -> io::Result<()> {
67        for (_, group) in output_receiver {
68            writer.write_group(&group)?;
69        }
70        writer.close().unwrap();
71        let end_write = Instant::now();
72        eprintln!("Finished writing all spectra in {:0.3?}", end_write - start);
73        Ok(())
74    });
75
76    if let Err(e) = reader_task.join() {
77        eprintln!("An error occurred while joining processing spectra task: {e:?}")
78    }
79
80    if let Err(e) = collator_task.join() {
81        eprintln!("An error occurred while joining collating spectra task: {e:?}")
82    }
83
84    match writer_task.join() {
85        Ok(r) => {
86            r?;
87        }
88        Err(e) => {
89            eprintln!("An error occurred while joining writing spectra task: {e:?}")
90        }
91    }
92
93    Ok(())
94}
averaging_writer/averaging_writer.rs

averaging_writer/
averaging_writer.rs
