rust_hdf5/
swmr.rs

1//! Single Writer / Multiple Reader (SWMR) API.
2//!
3//! Provides a high-level wrapper around the SWMR protocol for streaming
4//! frame-based data (e.g., area detector images).
5
6use std::path::Path;
7
8use crate::io::Hdf5Reader;
9use crate::io::SwmrWriter as IoSwmrWriter;
10
11use crate::error::Result;
12use crate::types::H5Type;
13
14/// SWMR writer for streaming frame-based data to an HDF5 file.
15///
16/// Usage:
17/// ```no_run
18/// use rust_hdf5::swmr::SwmrFileWriter;
19///
20/// let mut writer = SwmrFileWriter::create("stream.h5").unwrap();
21/// let ds = writer.create_streaming_dataset::<f32>("frames", &[256, 256]).unwrap();
22/// writer.start_swmr().unwrap();
23///
24/// // Write frames
25/// let frame_data = vec![0.0f32; 256 * 256];
26/// let raw: Vec<u8> = frame_data.iter()
27///     .flat_map(|v| v.to_le_bytes())
28///     .collect();
29/// writer.append_frame(ds, &raw).unwrap();
30/// writer.flush().unwrap();
31///
32/// writer.close().unwrap();
33/// ```
34pub struct SwmrFileWriter {
35    inner: IoSwmrWriter,
36}
37
38impl SwmrFileWriter {
39    /// Create a new HDF5 file for SWMR streaming.
40    pub fn create<P: AsRef<Path>>(path: P) -> Result<Self> {
41        let inner = IoSwmrWriter::create(path.as_ref())?;
42        Ok(Self { inner })
43    }
44
45    /// Create a streaming dataset.
46    ///
47    /// The dataset will have shape `[0, frame_dims...]` initially, with
48    /// chunk dimensions `[1, frame_dims...]` and unlimited first dimension.
49    ///
50    /// Returns the dataset index for use with `append_frame`.
51    pub fn create_streaming_dataset<T: H5Type>(
52        &mut self,
53        name: &str,
54        frame_dims: &[u64],
55    ) -> Result<usize> {
56        let datatype = T::hdf5_type();
57        let idx = self
58            .inner
59            .create_streaming_dataset(name, datatype, frame_dims)?;
60        Ok(idx)
61    }
62
63    /// Signal the start of SWMR mode.
64    pub fn start_swmr(&mut self) -> Result<()> {
65        self.inner.start_swmr()?;
66        Ok(())
67    }
68
69    /// Append a frame of raw data to a streaming dataset.
70    ///
71    /// The data size must match one frame (product of frame_dims * element_size).
72    pub fn append_frame(&mut self, ds_index: usize, data: &[u8]) -> Result<()> {
73        self.inner.append_frame(ds_index, data)?;
74        Ok(())
75    }
76
77    /// Flush all dataset index structures to disk with SWMR ordering.
78    pub fn flush(&mut self) -> Result<()> {
79        self.inner.flush()?;
80        Ok(())
81    }
82
83    /// Close and finalize the file.
84    pub fn close(self) -> Result<()> {
85        self.inner.close()?;
86        Ok(())
87    }
88}
89
90/// SWMR reader for monitoring a streaming HDF5 file.
91///
92/// Opens a file being written by a concurrent [`SwmrFileWriter`] and
93/// periodically calls [`refresh`](Self::refresh) to pick up new data.
94///
95/// ```no_run
96/// use rust_hdf5::swmr::SwmrFileReader;
97///
98/// let mut reader = SwmrFileReader::open("stream.h5").unwrap();
99///
100/// loop {
101///     reader.refresh().unwrap();
102///     let names = reader.dataset_names();
103///     if let Some(shape) = reader.dataset_shape("frames").ok() {
104///         println!("frames shape: {:?}", shape);
105///         if shape[0] > 0 {
106///             let data = reader.read_dataset_raw("frames").unwrap();
107///             println!("got {} bytes", data.len());
108///             break;
109///         }
110///     }
111///     std::thread::sleep(std::time::Duration::from_millis(100));
112/// }
113/// ```
114pub struct SwmrFileReader {
115    reader: Hdf5Reader,
116}
117
118impl SwmrFileReader {
119    /// Open an HDF5 file for SWMR reading.
120    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
121        let reader = Hdf5Reader::open_swmr(path.as_ref())?;
122        Ok(Self { reader })
123    }
124
125    /// Re-read the superblock and dataset metadata from disk.
126    ///
127    /// Call this periodically to pick up new data written by the concurrent
128    /// SWMR writer.
129    pub fn refresh(&mut self) -> Result<()> {
130        self.reader.refresh()?;
131        Ok(())
132    }
133
134    /// Return the names of all datasets.
135    pub fn dataset_names(&self) -> Vec<String> {
136        self.reader
137            .dataset_names()
138            .iter()
139            .map(|s| s.to_string())
140            .collect()
141    }
142
143    /// Return the current shape of a dataset.
144    pub fn dataset_shape(&self, name: &str) -> Result<Vec<u64>> {
145        Ok(self.reader.dataset_shape(name)?)
146    }
147
148    /// Read the raw bytes of a dataset.
149    pub fn read_dataset_raw(&mut self, name: &str) -> Result<Vec<u8>> {
150        Ok(self.reader.read_dataset_raw(name)?)
151    }
152
153    /// Read a dataset as a typed vector.
154    pub fn read_dataset<T: H5Type>(&mut self, name: &str) -> Result<Vec<T>> {
155        let raw = self.reader.read_dataset_raw(name)?;
156        if raw.len() % T::element_size() != 0 {
157            return Err(crate::error::Hdf5Error::TypeMismatch(format!(
158                "raw data size {} is not a multiple of element size {}",
159                raw.len(),
160                T::element_size(),
161            )));
162        }
163        let count = raw.len() / T::element_size();
164        let mut result = Vec::<T>::with_capacity(count);
165        unsafe {
166            std::ptr::copy_nonoverlapping(raw.as_ptr(), result.as_mut_ptr() as *mut u8, raw.len());
167            result.set_len(count);
168        }
169        Ok(result)
170    }
171}
rust_hdf5/swmr.rs

rust_hdf5/
swmr.rs
