1use std::any::Any;
2use std::fs::File;
3use std::io;
4use std::path::PathBuf;
5use std::sync::{
6 atomic::{AtomicU64, Ordering as AtomicOrdering},
7 mpsc::sync_channel,
8 Arc,
9};
10use std::thread;
11use std::time;
12
13use clap::Parser;
14
15use log::info;
16use mzdata::io::MassSpectrometryFormat;
17use mzdata::io::{checksum_file, MassSpectrometryReadWriteProcess, Sink, Source};
18use mzdata::meta::custom_software_name;
19use mzdata::meta::Software;
20use mzdata::meta::{DataProcessing, ProcessingMethod, SourceFile};
21use mzdata::params::ControlledVocabulary;
22use mzdata::prelude::*;
23
24use mzdata::spectrum::bindata::BinaryCompressionType;
25use mzdata::spectrum::ArrayType;
26use mzdata::spectrum::ArrayType::IntensityArray;
27use mzdata::spectrum::ArrayType::MZArray;
28use mzdata::spectrum::BinaryDataArrayType;
29use mzdata::MzMLWriter;
30use mzpeaks::{CentroidPeak, DeconvolutedPeak};
31
32fn compression_parser(compression: &str) -> Result<BinaryCompressionType, String> {
33 let compression = if !compression.ends_with(" compression") {
34 format!("{compression} compression")
35 } else {
36 compression.to_string()
37 };
38
39 BinaryCompressionType::COMPRESSION_METHODS
40 .iter()
41 .find(|x| x.as_param().unwrap().name() == compression)
42 .copied()
43 .ok_or_else(|| compression.to_string())
44}
45
46#[derive(Debug, Clone, Parser)]
47pub struct MZConvert {
48 #[arg()]
49 pub inpath: String,
50 #[arg()]
51 pub outpath: String,
52
53 #[arg(short = 'b', long, default_value_t = 8192)]
54 pub buffer_size: usize,
55
56 #[arg(long, value_parser=compression_parser, default_value="zlib compression")]
57 pub mz_compression: BinaryCompressionType,
58
59 #[arg(long, value_parser=compression_parser, default_value="zlib compression")]
60 pub intensity_compression: BinaryCompressionType,
61
62 #[arg(long, value_parser=compression_parser, default_value="zlib compression")]
63 pub ion_mobility_compression: BinaryCompressionType,
64}
65
66impl MZConvert {
67 pub fn new(inpath: String, outpath: String) -> Self {
68 Self {
69 inpath,
70 outpath,
71 buffer_size: 8192,
72 mz_compression: BinaryCompressionType::Zlib,
73 intensity_compression: BinaryCompressionType::Zlib,
74 ion_mobility_compression: BinaryCompressionType::Zlib,
75 }
76 }
77
78 pub fn main(&self) -> io::Result<()> {
79 let source = if self.inpath == "-" {
80 Source::Stdin
81 } else {
82 Source::<_, _>::from(self.inpath.as_ref())
83 };
84
85 let sink = if self.outpath == "-" {
86 Sink::Writer(Box::new(io::stdout()), MassSpectrometryFormat::MzML)
87 } else {
88 Sink::<CentroidPeak, DeconvolutedPeak>::from(self.outpath.as_ref())
89 };
90
91 self.open_reader(source, sink)
92 }
93
94 fn task<R: SpectrumSource + Send + 'static, W: SpectrumWriter + Send + 'static>(
95 &self,
96 reader: R,
97 mut writer: W,
98 ) -> io::Result<()> {
99 let (send, recv) = sync_channel(self.buffer_size);
100 let buffered = Arc::new(AtomicU64::default());
101 let buffered_w = Arc::clone(&buffered);
102 let reader_handle = thread::spawn(move || {
103 reader.enumerate().for_each(|(i, s)| {
104 let waiting_cnt = buffered.fetch_add(1, AtomicOrdering::SeqCst);
105 if i % 5000 == 0 && i > 0 {
106 log::info!("Reading {} {} ({waiting_cnt} to write)", i, s.id());
107 }
108 if i % 100 == 0 && i > 0 {
109 log::debug!("Reading {} {} ({waiting_cnt} to write)", i, s.id());
110 }
111
112 send.send(s).unwrap()
113 });
114 });
115
116 let writer_handle = thread::spawn(move || {
117 for s in recv.iter() {
118 let i = s.index();
119 buffered_w.fetch_sub(1, AtomicOrdering::SeqCst);
120 writer
121 .write_owned(s)
122 .inspect_err(|e| log::error!("Failed to write spectrum {i}: {e}"))
123 .unwrap();
124 }
125 writer.close().unwrap();
126 });
127
128 reader_handle.join().unwrap();
129 writer_handle.join().unwrap();
130 Ok(())
131 }
132
133 #[allow(unused)]
134 fn configure_reader(&self, reader: &mut dyn Any) {
135 #[cfg(feature = "thermo")]
136 if let Some(reader) = reader.downcast_mut::<mzdata::io::thermo::ThermoRawReader>() {
137 reader.set_load_extended_spectrum_data(false);
138 }
139 #[cfg(feature = "bruker_tdf")]
140 if let Some(reader) = reader.downcast_mut::<mzdata::io::tdf::TDFSpectrumReader>() {
141 reader.set_consolidate_peaks(false);
142 }
143 }
144
145 fn configure_writer(&self, writer: &mut dyn Any) {
146 info!("Configuring writer...");
147 if let Some(writer) = writer.downcast_mut::<MzMLWriter<io::BufWriter<File>>>() {
148 log::debug!("Configuring compression methods: {:?}", self.mz_compression);
149 writer.set_compression_method(
150 MZArray,
151 BinaryDataArrayType::Float32,
152 self.mz_compression,
153 );
154 writer.set_compression_method(
155 MZArray,
156 BinaryDataArrayType::Float64,
157 self.mz_compression,
158 );
159
160 log::debug!(
161 "Configuring compression methods: {:?}",
162 self.intensity_compression
163 );
164 writer.set_compression_method(
165 IntensityArray,
166 BinaryDataArrayType::Float32,
167 self.intensity_compression,
168 );
169 writer.set_compression_method(
170 IntensityArray,
171 BinaryDataArrayType::Float64,
172 self.intensity_compression,
173 );
174
175 log::debug!(
176 "Configuring compression methods: {:?}",
177 self.ion_mobility_compression
178 );
179
180 let im_arrays = [
181 ArrayType::MeanInverseReducedIonMobilityArray,
182 ArrayType::RawDriftTimeArray,
183 ArrayType::MeanDriftTimeArray,
184 ArrayType::RawIonMobilityArray,
185 ArrayType::MeanIonMobilityArray,
186 ];
187
188 for im_array in im_arrays {
189 writer.set_compression_method(
190 im_array.clone(),
191 BinaryDataArrayType::Float64,
192 self.ion_mobility_compression,
193 );
194 writer.set_compression_method(
195 im_array,
196 BinaryDataArrayType::Float32,
197 self.ion_mobility_compression,
198 );
199 }
200 }
201 }
202}
203
204impl MassSpectrometryReadWriteProcess<CentroidPeak, DeconvolutedPeak> for MZConvert {
205 type ErrorType = io::Error;
206
207 fn task<
208 R: RandomAccessSpectrumIterator<CentroidPeak, DeconvolutedPeak>
209 + SpectrumSource<CentroidPeak, DeconvolutedPeak>
210 + Send
211 + Any
212 + 'static,
213 W: SpectrumWriter<CentroidPeak, DeconvolutedPeak> + Send + Any + 'static,
214 >(
215 &self,
216 reader: R,
217 writer: W,
218 ) -> Result<(), Self::ErrorType> {
219 self.task(reader, writer)
220 }
221
222 fn transform_reader<
223 R: RandomAccessSpectrumIterator<CentroidPeak, DeconvolutedPeak>
224 + MSDataFileMetadata
225 + SpectrumSource<CentroidPeak, DeconvolutedPeak>
226 + Send
227 + Any
228 + 'static,
229 >(
230 &self,
231 mut reader: R,
232 _format: MassSpectrometryFormat,
233 ) -> Result<R, Self::ErrorType> {
234 self.configure_reader(&mut reader);
235 Ok(reader)
236 }
237
238 fn transform_writer<
239 R: RandomAccessSpectrumIterator<CentroidPeak, DeconvolutedPeak>
240 + MSDataFileMetadata
241 + SpectrumSource<CentroidPeak, DeconvolutedPeak>
242 + Send
243 + Any
244 + 'static,
245 W: SpectrumWriter<CentroidPeak, DeconvolutedPeak> + MSDataFileMetadata + Send + Any + 'static,
246 >(
247 &self,
248 reader: R,
249 reader_format: MassSpectrometryFormat,
250 mut writer: W,
251 writer_format: MassSpectrometryFormat,
252 ) -> Result<(R, W), Self::ErrorType> {
253 self.configure_writer(&mut writer);
254 if self.inpath != "-" {
255 let pb: PathBuf = self.inpath.clone().into();
256 let checksum = if pb.is_dir() {
257 Default::default()
258 } else {
259 info!("Computing checksum for {}", pb.display());
260
261 checksum_file(&pb)?
262 };
263 let has_already = reader
264 .file_description()
265 .source_files
266 .iter()
267 .flat_map(|f| {
268 f.get_param_by_name("SHA-1")
269 .map(|c| c.value.as_str() == checksum)
270 })
271 .all(|a| a);
272
273 if !has_already {
274 let mut sf = SourceFile {
275 location: pb
276 .parent()
277 .map(|p| format!("file:///{}", p.to_string_lossy()))
278 .unwrap_or("file:///".to_string()),
279 name: pb
280 .file_name()
281 .map(|p| p.to_string_lossy().to_string())
282 .unwrap_or("".to_string()),
283 ..Default::default()
284 };
285 let par = ControlledVocabulary::MS.param_val(1000569, "SHA-1", checksum);
286 sf.add_param(par);
287 sf.file_format = reader_format.as_param();
288
289 if let Some(ref_sf) = reader.file_description().source_files.last() {
290 sf.id_format = ref_sf.id_format.clone()
291 }
292 writer.file_description_mut().source_files.push(sf);
293 }
294 };
295
296 let mut sw = Software {
297 version: format!(
298 "v{}",
299 option_env!("CARGO_PKG_VERSION").unwrap_or_else(|| "?")
300 ),
301 id: Software::find_unique_id("mzconvert", writer.softwares()),
302 ..Default::default()
303 };
304 sw.add_param(custom_software_name("mzconvert"));
305
306 let mut method = ProcessingMethod {
307 software_reference: sw.id.clone(),
308 ..Default::default()
309 };
310 writer.softwares_mut().push(sw);
311
312 if let Some(conv) = writer_format.as_conversion() {
313 method.add_param(conv.into())
314 }
315
316 if writer.data_processings().is_empty() {
317 let mut dp = DataProcessing::default();
318 method.order = 0;
319 dp.push(method.clone());
320 dp.id = "DP1".to_string();
321 writer.data_processings_mut().push(dp)
322 } else {
323 for dp in writer.data_processings_mut() {
324 let mut next_step = method.clone();
325 next_step.order = dp.highest_order() + 1;
326 dp.push(next_step);
327 }
328 }
329 Ok((reader, writer))
330 }
331}
332
333fn main() -> io::Result<()> {
334 env_logger::init();
335
336 let job = MZConvert::parse();
337 let start = time::Instant::now();
338 job.main()?;
339 let end = time::Instant::now();
340 let elapsed = end - start;
341 eprintln!("Conversion finished: {:0.2} seconds", elapsed.as_secs_f64());
342 Ok(())
343}