flow_fcs/
write.rs

1//! FCS file writing utilities
2//!
3//! This module provides functionality to write FCS files to disk, including:
4//! - Duplicating existing files
5//! - Editing metadata and persisting changes
6//! - Creating new FCS files with data modifications (filtering, concatenation, column addition)
7//!
8//! ## Memory-Mapping Implications
9//!
10//! **Important**: When writing FCS files, the original memory-mapped file is not modified.
11//! All write operations create new files. The original `Fcs` struct remains valid and
12//! can continue to access the original file via memory-mapping until it's dropped.
13//!
14//! When you call `write_fcs_file()` or any of the modification functions:
15//! 1. The data is read from the DataFrame (which is already in memory)
16//! 2. A new file is created on disk
17//! 3. The original memory-mapped file remains unchanged
18//!
19//! This means:
20//! - You can safely write modified versions without affecting the original
21//! - The original `Fcs` struct can still be used after writing
22//! - No special handling is needed to "close" or "unmap" before writing
23//! - Multiple writes can happen concurrently from the same source file
24
25use crate::{
26    Fcs,
27    byteorder::ByteOrder,
28    keyword::{IntegerKeyword, Keyword},
29    metadata::Metadata,
30    version::Version,
31};
32use anyhow::{Result, anyhow};
33use byteorder::{LittleEndian, WriteBytesExt};
34use polars::prelude::*;
35use std::fs::File;
36use std::io::Write;
37use std::path::Path;
38use std::sync::Arc;
39
40/// Write an FCS file to disk
41///
42/// **Important**: This function closes the memory-mapped file before writing.
43/// The original Fcs struct will no longer be able to access the original file
44/// after this operation, but the data is preserved in the DataFrame.
45///
46/// # Arguments
47/// * `fcs` - The FCS struct to write (will consume the struct)
48/// * `path` - Output file path
49///
50/// # Errors
51/// Returns an error if:
52/// - The path is invalid
53/// - The file cannot be written
54/// - Metadata cannot be serialized
55pub fn write_fcs_file(fcs: Fcs, path: impl AsRef<Path>) -> Result<()> {
56    let path = path.as_ref();
57
58    // Validate file extension
59    if path.extension().and_then(|s| s.to_str()) != Some("fcs") {
60        return Err(anyhow!("Output file must have .fcs extension"));
61    }
62
63    // Get data from DataFrame
64    let df = &*fcs.data_frame;
65    let n_events = df.height();
66    let n_params = df.width();
67
68    if n_events == 0 {
69        return Err(anyhow!("Cannot write FCS file with 0 events"));
70    }
71    if n_params == 0 {
72        return Err(anyhow!("Cannot write FCS file with 0 parameters"));
73    }
74
75    // Serialize data segment first (we need its size for metadata)
76    let data_segment = serialize_data(df, &fcs.metadata)?;
77
78    // Calculate offsets
79    let header_size = 58;
80    let text_start = header_size;
81    // Estimate text segment size (will recalculate after)
82    let estimated_text_size = estimate_text_segment_size(&fcs.metadata, n_events, n_params);
83    let estimated_text_end = text_start + estimated_text_size - 1;
84    let data_start = estimated_text_end + 1;
85    let data_end = data_start + data_segment.len() - 1;
86
87    // Serialize metadata to text segment (now we know data offsets)
88    let text_segment = serialize_metadata(&fcs.metadata, n_events, n_params, data_start, data_end)?;
89
90    // Recalculate offsets with actual text segment size
91    let text_end = text_start + text_segment.len() - 1;
92    let data_start = text_end + 1;
93    let data_end = data_start + data_segment.len() - 1;
94
95    // Build header
96    let header = build_header(
97        &fcs.header.version,
98        text_start,
99        text_end,
100        data_start,
101        data_end,
102    )?;
103
104    // Write file
105    let mut file = File::create(path)?;
106    file.write_all(&header)?;
107    file.write_all(&text_segment)?;
108    file.write_all(&data_segment)?;
109    file.sync_all()?;
110
111    Ok(())
112}
113
114/// Duplicate an existing FCS file to a new path
115///
116/// This creates an exact copy of the file on disk. The original Fcs struct
117/// remains valid and can continue to be used.
118///
119/// # Arguments
120/// * `fcs` - Reference to the FCS struct to duplicate
121/// * `path` - Output file path
122///
123/// # Errors
124/// Returns an error if the file cannot be written
125pub fn duplicate_fcs_file(fcs: &Fcs, path: impl AsRef<Path>) -> Result<()> {
126    use std::fs;
127
128    let path = path.as_ref();
129
130    // Simply copy the file on disk
131    fs::copy(&fcs.file_access.path, path)?;
132
133    Ok(())
134}
135
136/// Edit metadata and persist changes to disk
137///
138/// This function:
139/// 1. Updates the metadata in the Fcs struct
140/// 2. Writes the modified file to disk
141/// 3. Returns a new Fcs struct pointing to the new file
142///
143/// **Note**: The original file is not modified. A new file is created.
144///
145/// # Arguments
146/// * `fcs` - The FCS struct to modify
147/// * `path` - Output file path for the modified file
148/// * `updates` - Function that modifies the metadata
149///
150/// # Errors
151/// Returns an error if the file cannot be written
152pub fn edit_metadata_and_save<F>(mut fcs: Fcs, path: impl AsRef<Path>, updates: F) -> Result<Fcs>
153where
154    F: FnOnce(&mut Metadata),
155{
156    // Apply updates to metadata
157    updates(&mut fcs.metadata);
158
159    // Update $TOT if event count changed
160    let n_events = fcs.get_event_count_from_dataframe();
161    use crate::keyword::match_and_parse_keyword;
162    let tot_keyword = match_and_parse_keyword("$TOT", &n_events.to_string());
163    if let crate::keyword::KeywordCreationResult::Int(int_kw) = tot_keyword {
164        fcs.metadata
165            .keywords
166            .insert("$TOT".to_string(), Keyword::Int(int_kw));
167    }
168
169    // Write to new file
170    write_fcs_file(fcs.clone(), &path)?;
171
172    // Open the new file
173    Fcs::open(
174        path.as_ref()
175            .to_str()
176            .ok_or_else(|| anyhow!("Invalid path"))?,
177    )
178}
179
180/// Create a new FCS file by filtering events
181///
182/// Removes events where `mask[i] == false`. The mask must have the same length
183/// as the number of events in the original file.
184///
185/// # Arguments
186/// * `fcs` - The FCS struct to filter
187/// * `path` - Output file path
188/// * `mask` - Boolean mask (true = keep, false = remove)
189///
190/// # Errors
191/// Returns an error if:
192/// - The mask length doesn't match the number of events
193/// - The file cannot be written
194pub fn filter_events(fcs: Fcs, path: impl AsRef<Path>, mask: &[bool]) -> Result<Fcs> {
195    let df = &*fcs.data_frame;
196    let n_events = df.height();
197
198    if mask.len() != n_events {
199        return Err(anyhow!(
200            "Mask length {} doesn't match number of events {}",
201            mask.len(),
202            n_events
203        ));
204    }
205
206    // Filter DataFrame using Polars
207    let mask_vec: Vec<bool> = mask.to_vec();
208    let mask_series = Series::new("mask".into(), mask_vec);
209    let mask_ca = mask_series.bool()?;
210    let filtered_df = df.filter(&mask_ca)?;
211
212    // Create new Fcs with filtered data
213    let mut new_fcs = fcs.clone();
214    new_fcs.data_frame = Arc::new(filtered_df);
215
216    // Update metadata
217    let n_events_after = new_fcs.get_event_count_from_dataframe();
218    use crate::keyword::match_and_parse_keyword;
219    let tot_keyword = match_and_parse_keyword("$TOT", &n_events_after.to_string());
220    if let crate::keyword::KeywordCreationResult::Int(int_kw) = tot_keyword {
221        new_fcs
222            .metadata
223            .keywords
224            .insert("$TOT".to_string(), Keyword::Int(int_kw));
225    }
226
227    // Write to file
228    write_fcs_file(new_fcs.clone(), &path)?;
229
230    // Open the new file
231    Fcs::open(
232        path.as_ref()
233            .to_str()
234            .ok_or_else(|| anyhow!("Invalid path"))?,
235    )
236}
237
238/// Create a new FCS file by concatenating events from multiple files
239///
240/// All files must have the same parameters (same names and order).
241///
242/// # Arguments
243/// * `files` - Vector of FCS structs to concatenate
244/// * `path` - Output file path
245///
246/// # Errors
247/// Returns an error if:
248/// - Files have different parameters
249/// - The file cannot be written
250pub fn concatenate_events(files: Vec<Fcs>, path: impl AsRef<Path>) -> Result<Fcs> {
251    if files.is_empty() {
252        return Err(anyhow!("Cannot concatenate empty list of files"));
253    }
254
255    if files.len() == 1 {
256        // Just duplicate the single file
257        return duplicate_fcs_file(&files[0], &path).and_then(|_| {
258            Fcs::open(
259                path.as_ref()
260                    .to_str()
261                    .ok_or_else(|| anyhow!("Invalid path"))?,
262            )
263        });
264    }
265
266    // Verify all files have the same parameters
267    let first_params: Vec<String> = files[0].get_parameter_names_from_dataframe();
268
269    for (idx, fcs) in files.iter().enumerate().skip(1) {
270        let params: Vec<String> = fcs.get_parameter_names_from_dataframe();
271        if params != first_params {
272            return Err(anyhow!("File {} has different parameters than file 0", idx));
273        }
274    }
275
276    // Concatenate DataFrames using vstack
277    let dfs: Vec<DataFrame> = files.iter().map(|f| (*f.data_frame).clone()).collect();
278    let concatenated_df = dfs
279        .into_iter()
280        .reduce(|acc, df| acc.vstack(&df).unwrap_or(acc))
281        .unwrap();
282
283    // Create new Fcs using first file as template
284    let mut new_fcs = files[0].clone();
285    new_fcs.data_frame = Arc::new(concatenated_df);
286
287    // Update metadata
288    let n_events_after = new_fcs.get_event_count_from_dataframe();
289    use crate::keyword::match_and_parse_keyword;
290    let tot_keyword = match_and_parse_keyword("$TOT", &n_events_after.to_string());
291    if let crate::keyword::KeywordCreationResult::Int(int_kw) = tot_keyword {
292        new_fcs
293            .metadata
294            .keywords
295            .insert("$TOT".to_string(), Keyword::Int(int_kw));
296    }
297
298    // Generate new GUID
299    new_fcs.metadata.validate_guid();
300
301    // Write to file
302    write_fcs_file(new_fcs.clone(), &path)?;
303
304    // Open the new file
305    Fcs::open(
306        path.as_ref()
307            .to_str()
308            .ok_or_else(|| anyhow!("Invalid path"))?,
309    )
310}
311
312/// Create a new FCS file by adding a column (parameter) to existing data
313///
314/// This is useful for adding QC results (e.g., a boolean column indicating
315/// good/bad events) or other event-level annotations.
316///
317/// # Arguments
318/// * `fcs` - The FCS struct to modify
319/// * `path` - Output file path
320/// * `column_name` - Name of the new parameter
321/// * `values` - Values for the new parameter (must match number of events)
322///
323/// # Errors
324/// Returns an error if:
325/// - The values length doesn't match the number of events
326/// - The column name already exists
327/// - The file cannot be written
328pub fn add_column(
329    mut fcs: Fcs,
330    path: impl AsRef<Path>,
331    column_name: &str,
332    values: Vec<f32>,
333) -> Result<Fcs> {
334    let df = &*fcs.data_frame;
335    let n_events = df.height();
336
337    if values.len() != n_events {
338        return Err(anyhow!(
339            "Values length {} doesn't match number of events {}",
340            values.len(),
341            n_events
342        ));
343    }
344
345    // Check if column already exists
346    if df
347        .get_column_names()
348        .iter()
349        .any(|&name| name == column_name)
350    {
351        return Err(anyhow!("Column {} already exists", column_name));
352    }
353
354    // Add column to DataFrame
355    let mut new_df = df.clone();
356    let new_series = Series::new(column_name.into(), values);
357    new_df
358        .with_column(new_series)
359        .map_err(|e| anyhow!("Failed to add column: {}", e))?;
360
361    // Update Fcs struct
362    fcs.data_frame = Arc::new(new_df);
363
364    // Add parameter metadata
365    let n_params = fcs.get_parameter_count_from_dataframe();
366    let param_num = n_params; // 1-based indexing in FCS
367
368    // Update $PAR keyword
369    use crate::keyword::match_and_parse_keyword;
370    let par_keyword = match_and_parse_keyword("$PAR", &n_params.to_string());
371    if let crate::keyword::KeywordCreationResult::Int(int_kw) = par_keyword {
372        fcs.metadata
373            .keywords
374            .insert("$PAR".to_string(), Keyword::Int(int_kw));
375    }
376
377    // Add parameter keywords ($PnN, $PnB, etc.)
378    fcs.metadata
379        .insert_string_keyword(format!("$P{}N", param_num), column_name.to_string());
380
381    // Default: 32 bits (4 bytes) for float32
382    let pnb_keyword = match_and_parse_keyword(&format!("$P{}B", param_num), "32");
383    if let crate::keyword::KeywordCreationResult::Int(int_kw) = pnb_keyword {
384        fcs.metadata
385            .keywords
386            .insert(format!("$P{}B", param_num), Keyword::Int(int_kw));
387    }
388
389    // Default range
390    let pnr_keyword = match_and_parse_keyword(&format!("$P{}R", param_num), "262144");
391    if let crate::keyword::KeywordCreationResult::Int(int_kw) = pnr_keyword {
392        fcs.metadata
393            .keywords
394            .insert(format!("$P{}R", param_num), Keyword::Int(int_kw));
395    }
396
397    // Default amplification
398    fcs.metadata
399        .insert_string_keyword(format!("$P{}E", param_num), "0,0".to_string());
400
401    // Add to parameter map
402    use crate::TransformType;
403    use crate::parameter::Parameter;
404    fcs.parameters.insert(
405        column_name.to_string().into(),
406        Parameter::new(&param_num, column_name, column_name, &TransformType::Linear),
407    );
408
409    // Write to file
410    write_fcs_file(fcs.clone(), &path)?;
411
412    // Open the new file
413    Fcs::open(
414        path.as_ref()
415            .to_str()
416            .ok_or_else(|| anyhow!("Invalid path"))?,
417    )
418}
419
420// ==================== Internal Helper Functions ====================
421
422fn estimate_text_segment_size(metadata: &Metadata, _n_events: usize, n_params: usize) -> usize {
423    // Rough estimate: base size + keywords
424    let base_size = 200; // Base keywords
425    let keyword_size = metadata.keywords.len() * 50; // Average keyword size
426    let param_keywords = n_params * 100; // Parameter keywords
427    base_size + keyword_size + param_keywords
428}
429
430fn serialize_metadata(
431    metadata: &Metadata,
432    n_events: usize,
433    n_params: usize,
434    data_start: usize,
435    data_end: usize,
436) -> Result<Vec<u8>> {
437    let delimiter = metadata.delimiter as u8;
438    let mut text_segment = Vec::new();
439
440    // Helper to add keyword-value pair
441    let mut add_keyword = |key: &str, value: &str| {
442        text_segment.push(delimiter);
443        text_segment.extend_from_slice(format!("${}", key).as_bytes());
444        text_segment.push(delimiter);
445        text_segment.extend_from_slice(value.as_bytes());
446    };
447
448    // Required keywords (order matters for FCS compatibility)
449    add_keyword("BEGINANALYSIS", "0");
450    add_keyword("ENDANALYSIS", "0");
451    add_keyword("BEGINSTEXT", "0");
452    add_keyword("ENDSTEXT", "0");
453    add_keyword("BEGINDATA", &data_start.to_string());
454    add_keyword("ENDDATA", &data_end.to_string());
455
456    // Serialize all keywords from metadata
457    let mut sorted_keys: Vec<_> = metadata.keywords.keys().collect();
458    sorted_keys.sort();
459
460    for key in sorted_keys {
461        // Skip keywords we've already added
462        if matches!(
463            key.as_str(),
464            "$BEGINANALYSIS"
465                | "$ENDANALYSIS"
466                | "$BEGINSTEXT"
467                | "$ENDSTEXT"
468                | "$BEGINDATA"
469                | "$ENDDATA"
470        ) {
471            continue;
472        }
473
474        let keyword = metadata.keywords.get(key).unwrap();
475        let value_str = match keyword {
476            Keyword::Int(int_kw) => match int_kw {
477                IntegerKeyword::TOT(_) => {
478                    // Use actual event count
479                    n_events.to_string()
480                }
481                IntegerKeyword::PAR(_) => {
482                    // Use actual parameter count
483                    n_params.to_string()
484                }
485                _ => int_kw.to_string(),
486            },
487            Keyword::String(str_kw) => str_kw.to_string(),
488            Keyword::Float(float_kw) => float_kw.to_string(),
489            Keyword::Byte(byte_kw) => byte_kw.to_string(),
490            Keyword::Mixed(mixed_kw) => mixed_kw.to_string(),
491        };
492
493        // Remove $ prefix for serialization (it will be added back)
494        let key_without_prefix = key.strip_prefix('$').unwrap_or(key);
495        add_keyword(key_without_prefix, &value_str);
496    }
497
498    Ok(text_segment)
499}
500
501fn serialize_data(df: &DataFrame, metadata: &Metadata) -> Result<Vec<u8>> {
502    let n_events = df.height();
503    let n_params = df.width();
504
505    // Get bytes per parameter from metadata
506    let bytes_per_param = metadata
507        .calculate_bytes_per_event()
508        .map(|bytes_per_event| bytes_per_event / n_params)
509        .unwrap_or(4); // Default to 4 bytes (float32)
510
511    let mut data = Vec::with_capacity(n_events * n_params * bytes_per_param);
512
513    // Get byte order
514    let byte_order = metadata
515        .get_byte_order()
516        .unwrap_or(&ByteOrder::LittleEndian);
517    let is_little_endian = matches!(byte_order, ByteOrder::LittleEndian);
518
519    // Serialize row by row (FCS format: event1_param1, event1_param2, ..., event2_param1, ...)
520    // Get all columns as f32 slices for efficient access
521    let column_names = df.get_column_names();
522    let mut column_data: Vec<&[f32]> = Vec::with_capacity(n_params);
523
524    for col_name in &column_names {
525        let series = df.column(col_name)?;
526        let f32_series = series
527            .f32()
528            .map_err(|e| anyhow!("Column {} is not f32: {}", col_name, e))?;
529        let slice = f32_series
530            .cont_slice()
531            .map_err(|e| anyhow!("Column {} data is not contiguous: {}", col_name, e))?;
532        column_data.push(slice);
533    }
534
535    // Write row by row
536    for row_idx in 0..n_events {
537        for col_data in &column_data {
538            let value = col_data[row_idx];
539
540            // Write as float32 (4 bytes)
541            if is_little_endian {
542                data.write_f32::<LittleEndian>(value)?;
543            } else {
544                use byteorder::BigEndian;
545                data.write_f32::<BigEndian>(value)?;
546            }
547        }
548    }
549
550    Ok(data)
551}
552
553fn build_header(
554    version: &Version,
555    text_start: usize,
556    text_end: usize,
557    data_start: usize,
558    data_end: usize,
559) -> Result<Vec<u8>> {
560    let mut header = vec![0u8; 58];
561
562    // Version string (bytes 0-5)
563    let version_str = format!("{}", version);
564    if version_str.len() > 6 {
565        return Err(anyhow!("Version string too long: {}", version_str));
566    }
567    header[0..version_str.len()].copy_from_slice(version_str.as_bytes());
568
569    // 4 spaces (bytes 6-9)
570    header[6..10].fill(b' ');
571
572    // Text segment offsets (bytes 10-17 and 18-25) - right-aligned, space-padded
573    let text_start_str = format!("{:>8}", text_start);
574    header[10..18].copy_from_slice(text_start_str.as_bytes());
575    let text_end_str = format!("{:>8}", text_end);
576    header[18..26].copy_from_slice(text_end_str.as_bytes());
577
578    // Data segment offsets (bytes 26-33 and 34-41)
579    let data_start_str = format!("{:>8}", data_start);
580    header[26..34].copy_from_slice(data_start_str.as_bytes());
581    let data_end_str = format!("{:>8}", data_end);
582    header[34..42].copy_from_slice(data_end_str.as_bytes());
583
584    // Analysis segment offsets (bytes 42-49 and 50-57) - set to 0
585    header[42..50].copy_from_slice(b"       0");
586    header[50..58].copy_from_slice(b"       0");
587
588    Ok(header)
589}
flow_fcs/write.rs

flow_fcs/
write.rs
