gondola_core/io/caraspace/
writer.rs

1// This file is part of gaps-online-software and published 
2// under the GPLv3 license
3
4use crate::prelude::*;
5
6/// Write CRFrames ("caraspace frames") to disk. 
7/// 
8/// A frame can hold any TOF/TelemetryPacket.
9/// Data are written to disk in sequential matter,
10/// where frames can be added one at a time and 
11/// buffered for efficient writing
12///
13/// -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- 
14///
15/// How to create a new CRWriter (python)
16///
17/// # Arguments
18///
19/// * file_path        (str) : Path to store the file under
20/// * run_id           (u32) : Run ID for this file (will be written in filename)
21/// * subrun_id        (u32) : Sub-Run ID for this file (will be written in filename. 
22///                            If None, a generic "0" will be used
23/// * timestamp        (str) : The writer will add an automatic timestamp to the current file
24///                            based on the current time. This option allows to overwrite 
25///                            that behaviour
26/// * file_len_gcu_sec (u32) : If not None, this will use the gcu time from added packets 
27///                            as the timestamp in the filename, and will write a new file 
28///                            every given number of seconds
29#[cfg_attr(feature="pybindings", pyclass)]
30pub struct CRWriter {
31
32  pub file            : File,
33  /// location to store the file
34  pub file_path       : String,
35  /// The maximum number of packets 
36  /// for a single file. Ater this 
37  /// number is reached, a new 
38  /// file is started.
39  pub pkts_per_file   : usize,
40  /// The maximum number of (Mega)bytes
41  /// per file. After this a new file 
42  /// is started
43  pub mbytes_per_file     : usize,
44  /// the name of the current file we are writing to
45  pub file_name           : String,
46  pub run_id              : u32,
47  file_id                 : usize,
48  /// internal packet counter, number of 
49  /// packets which went through the writer
50  n_packets               : usize,
51  /// internal counter for bytes written in 
52  /// this file
53  file_nbytes_wr          : usize,
54  /// it can also take a timestamp, in case we 
55  /// don't want to use the current time when the 
56  /// file is written
57  pub file_timestamp      : Option<String>,
58  /// if writing a new file every gcu seconds, keep 
59  /// the first gcutimestamp 
60  pub first_gcu_timestamp : Option<u64>,
61  /// Write a new file every gcu seconds (from telemetry packet header) 
62  pub file_len_gcu_sec    : Option<u32>,
63}
64
65impl CRWriter {
66
67  /// Instantiate a new PacketWriter 
68  ///
69  /// # Arguments
70  ///
71  /// * file_path        : Path to store the file under
72  /// * run_id           : Run ID for this file (will be written in filename)
73  /// * subrun_id        : Sub-Run ID for this file (will be written in filename. 
74  ///                      If None, a generic "0" will be used
75  /// * timestamp        : The writer will add an automatic timestamp to the current file
76  ///                      based on the current time. This option allows to overwrite 
77  ///                      that behaviour
78  /// * file_len_gcu_sec : If not None, this will use the gcu time from added packets 
79  ///                      as the timestamp in the filename, and will write a new file 
80  ///                      every given number of seconds
81  pub fn new(mut file_path : String, run_id : u32,
82             subrun_id : Option<u64>, timestamp : Option<String>, file_len_gcu_sec : Option<u32>) -> Self {
83    if file_len_gcu_sec.is_some() {
84      if timestamp.is_none() {
85        panic!("If the writer should operate in the mode where it writes a new file every x seconds depending on the delta of gcu times in the packets, it needs the first timestamp given. This needs to be the first timestamp of the packets to be written");
86      }
87    }
88    let file : File;
89    let file_name : String;
90    if !file_path.ends_with("/") {
91      file_path += "/";
92    }
93    // check if this path exists 
94    let tmppath = Path::new(&file_path);
95    if !tmppath.exists() {
96      fs::create_dir_all(tmppath)
97      .expect(&format!("Failed to create directory at: {}", file_path));
98    }
99
100    let filename : String;
101    let first_timestamp = timestamp.clone();
102    if let Some(subrun) = subrun_id {
103      filename = format!("{}{}", file_path, get_runfilename(run_id, subrun, None, timestamp, false));
104    } else {
105      filename = format!("{}{}", file_path, get_runfilename(run_id, 0, None, timestamp, false));
106    }
107    let path     = Path::new(&filename); 
108    //println!("Writing to file {filename}");
109    //FIXME - do we really want to have this append here?
110    file = OpenOptions::new().create(true).append(true).open(path).expect(&format!("Unable to open file {}", filename));
111    file_name = filename;
112    let mut first_gcu_timestamp = None;
113    if first_timestamp.is_some() {
114      first_gcu_timestamp = get_unix_timestamp(&first_timestamp.unwrap(), None);
115    }
116    Self {
117      file,
118      file_path           : file_path,
119      pkts_per_file       : 0,
120      mbytes_per_file     : 420,
121      run_id              : run_id,
122      file_nbytes_wr      : 0,    
123      file_id             : 1,
124      n_packets           : 0,
125      file_name           : file_name,
126      file_timestamp      : None,
127      first_gcu_timestamp : first_gcu_timestamp,
128      file_len_gcu_sec    : file_len_gcu_sec,
129    }
130  }
131
132  pub fn get_file(&self, timestamp : Option<String>) -> File { 
133    let file : File;
134    let filename = format!("{}{}", self.file_path, get_runfilename(self.run_id, self.file_id as u64, None, timestamp, false));
135    //let filename = self.file_path.clone() + &get_runfilename(runid,self.file_id as u64, None);
136    let path     = Path::new(&filename); 
137    info!("Writing to file {filename}");
138    file = OpenOptions::new().create(true).append(true).open(path).expect("Unable to open file {filename}");
139    file
140  }
141
142  /// Induce serialization to disk for a CRFrame
143  ///
144  ///
145  pub fn add_frame(&mut self, frame : &CRFrame) {
146    let buffer = frame.to_bytestream();
147    self.file_nbytes_wr += buffer.len();
148    match self.file.write_all(buffer.as_slice()) {
149      Err(err) => error!("Writing to file to path {} failed! {}", self.file_path, err),
150      Ok(_)    => ()
151    }
152    self.n_packets += 1;
153    let mut newfile = false;
154    if self.pkts_per_file != 0 {
155      if self.n_packets == self.pkts_per_file {
156        newfile = true;
157        self.n_packets = 0;
158      }
159    } else if self.mbytes_per_file != 0 {
160      // multiply by mebibyte
161      if self.file_nbytes_wr >= self.mbytes_per_file * 1_048_576 {
162        newfile = true;
163        self.file_nbytes_wr = 0;
164      }
165    } else if self.file_len_gcu_sec.is_some() {
166      if frame.timestamp.clone().unwrap() - self.first_gcu_timestamp.unwrap() as f64 > self.file_len_gcu_sec.unwrap() as f64 {
167        newfile = true; 
168        self.file_timestamp = get_utc_timestamp_from_unix(frame.timestamp.clone().unwrap()); 
169        //println!("starting new file with {}", self.file_timestamp.clone().unwrap());
170        self.first_gcu_timestamp = Some(frame.timestamp.unwrap() as u64);
171      }
172    }
173    if newfile {
174        //let filename = self.file_prefix.clone() + "_" + &self.file_id.to_string() + ".tof.gaps";
175        match self.file.sync_all() {
176          Err(err) => {
177            error!("Unable to sync file to disc! {err}");
178          },
179          Ok(_) => ()
180        }
181        self.file = self.get_file(self.file_timestamp.clone());
182        self.file_id += 1;
183        //let path  = Path::new(&filename);
184        //println!("==> [TOFPACKETWRITER] Will start a new file {}", path.display());
185        //self.file = OpenOptions::new().create(true).append(true).open(path).expect("Unable to open file {filename}");
186        //self.n_packets = 0;
187        //self.file_id += 1;
188      }
189  debug!("CRFrame written!");
190  }
191}
192
193impl Default for CRWriter {
194  fn default() -> Self {
195    CRWriter::new(String::from(""),0,None, None, None)
196  }
197}
198
199impl fmt::Display for CRWriter {
200  fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
201    let mut repr = String::from("<CRWriter:");
202    repr += &(format!("\n  path: {}", self.file_path)); 
203    repr += ">";
204    write!(f, "{}", repr)
205  }
206}
207
208#[cfg(feature="pybindings")]
209#[pymethods]
210impl CRWriter {
211  
212  #[new]
213  #[pyo3(signature = (filename, run_id, subrun_id = None, timestamp = None, file_len_gcu_sec = None))]
214  fn new_py(filename : String, run_id : u32, subrun_id : Option<u64>, timestamp : Option<String>, file_len_gcu_sec : Option<u32>) -> Self {
215    Self::new(filename, run_id, subrun_id, timestamp, file_len_gcu_sec)
216  }
217
218  #[getter] 
219  #[pyo3(name="current_filename")] 
220  fn get_current_filename_py(&self) -> String { 
221    self.file_name.clone()
222  }
223
224  fn set_mbytes_per_file(&mut self, fsize : usize) {
225    self.mbytes_per_file = fsize;
226  }
227
228  fn set_file_timestamp(&mut self, timestamp : String) {
229    self.file_timestamp = Some(timestamp);
230  }
231  
232  #[pyo3(name="add_frame")]
233  fn add_frame_py(&mut self, frame : CRFrame) {
234    self.add_frame(&frame);  
235  }
236}
237
238#[cfg(feature="pybindings")]
239pythonize_display!(CRWriter);
240
gondola_core/io/caraspace/writer.rs

gondola_core/io/caraspace/
writer.rs
