Skip to main content

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::format::messages::attribute::AttributeMessage;
9use crate::format::messages::datatype::DatatypeMessage;
10use crate::io::locking::FileLocking;
11use crate::io::Hdf5Reader;
12use crate::io::SwmrWriter as IoSwmrWriter;
13
14use crate::error::{Hdf5Error, Result};
15use crate::types::H5Type;
16
17/// SWMR writer for streaming frame-based data to an HDF5 file.
18///
19/// Usage:
20/// ```no_run
21/// use rust_hdf5::swmr::SwmrFileWriter;
22///
23/// let mut writer = SwmrFileWriter::create("stream.h5").unwrap();
24/// let ds = writer.create_streaming_dataset::<f32>("frames", &[256, 256]).unwrap();
25/// writer.start_swmr().unwrap();
26///
27/// // Write frames
28/// let frame_data = vec![0.0f32; 256 * 256];
29/// let raw: Vec<u8> = frame_data.iter()
30///     .flat_map(|v| v.to_le_bytes())
31///     .collect();
32/// writer.append_frame(ds, &raw).unwrap();
33/// writer.flush().unwrap();
34///
35/// writer.close().unwrap();
36/// ```
37pub struct SwmrFileWriter {
38    inner: IoSwmrWriter,
39}
40
41impl SwmrFileWriter {
42    /// Create a new HDF5 file for SWMR streaming using the env-var-derived
43    /// locking policy.
44    pub fn create<P: AsRef<Path>>(path: P) -> Result<Self> {
45        let inner = IoSwmrWriter::create(path.as_ref())?;
46        Ok(Self { inner })
47    }
48
49    /// Create a new HDF5 file for SWMR streaming with an explicit locking
50    /// policy. The writer holds an exclusive lock until [`Self::start_swmr`]
51    /// is called, at which point the lock is downgraded to shared so
52    /// concurrent SWMR readers can attach.
53    pub fn create_with_locking<P: AsRef<Path>>(path: P, locking: FileLocking) -> Result<Self> {
54        let inner = IoSwmrWriter::create_with_locking(path.as_ref(), locking)?;
55        Ok(Self { inner })
56    }
57
58    /// Reopen a cleanly-closed HDF5 file to resume SWMR streaming.
59    ///
60    /// Existing datasets are reconstructed; locate them with
61    /// [`dataset_index`](Self::dataset_index), call [`start_swmr`](Self::start_swmr)
62    /// to re-enter SWMR mode, then continue with [`append_frame`](Self::append_frame).
63    /// Appending to a multi-frame-chunk dataset (`chunk[0] > 1`) after reopen
64    /// is rejected — its final partial band was zero-padded at the original
65    /// close. Recovering a crashed (never cleanly closed) file is not supported.
66    pub fn open_append<P: AsRef<Path>>(path: P) -> Result<Self> {
67        let inner = IoSwmrWriter::open_append(path.as_ref())?;
68        Ok(Self { inner })
69    }
70
71    /// Reopen a cleanly-closed HDF5 file to resume SWMR streaming with an
72    /// explicit locking policy. See [`Self::open_append`].
73    pub fn open_append_with_locking<P: AsRef<Path>>(path: P, locking: FileLocking) -> Result<Self> {
74        let inner = IoSwmrWriter::open_append_with_locking(path.as_ref(), locking)?;
75        Ok(Self { inner })
76    }
77
78    /// Return the index of a dataset by name, or `None` if absent.
79    ///
80    /// Mainly used after [`open_append`](Self::open_append) to recover the
81    /// index of a reconstructed dataset for [`append_frame`](Self::append_frame).
82    pub fn dataset_index(&self, name: &str) -> Option<usize> {
83        self.inner.dataset_index(name)
84    }
85
86    /// Create a streaming dataset.
87    ///
88    /// The dataset will have shape `[0, frame_dims...]` initially, with
89    /// chunk dimensions `[1, frame_dims...]` and unlimited first dimension.
90    ///
91    /// Returns the dataset index for use with `append_frame`.
92    pub fn create_streaming_dataset<T: H5Type>(
93        &mut self,
94        name: &str,
95        frame_dims: &[u64],
96    ) -> Result<usize> {
97        self.create_streaming_dataset_with_datatype(name, T::hdf5_type(), frame_dims)
98    }
99
100    /// [`Self::create_streaming_dataset`] with an explicit stored datatype instead of a
101    /// carrier `T`, for a type no `T` describes (e.g. a reduced-precision
102    /// type for [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an
103    /// element is `datatype.element_size()` bytes.
104    pub fn create_streaming_dataset_with_datatype(
105        &mut self,
106        name: &str,
107        datatype: crate::DatatypeMessage,
108        frame_dims: &[u64],
109    ) -> Result<usize> {
110        let idx = self
111            .inner
112            .create_streaming_dataset(name, datatype, frame_dims)?;
113        Ok(idx)
114    }
115
116    /// Create a streaming dataset whose frames are compressed.
117    ///
118    /// Like [`create_streaming_dataset`](Self::create_streaming_dataset) but
119    /// each appended frame is run through `pipeline` (e.g.
120    /// `FilterPipeline::deflate(4)`). SWMR appends and in-place header
121    /// updates work the same as for uncompressed streaming datasets.
122    pub fn create_streaming_dataset_compressed<T: H5Type>(
123        &mut self,
124        name: &str,
125        frame_dims: &[u64],
126        pipeline: crate::format::messages::filter::FilterPipeline,
127    ) -> Result<usize> {
128        self.create_streaming_dataset_compressed_with_datatype(
129            name,
130            T::hdf5_type(),
131            frame_dims,
132            pipeline,
133        )
134    }
135
136    /// [`Self::create_streaming_dataset_compressed`] with an explicit stored datatype instead of a
137    /// carrier `T`, for a type no `T` describes (e.g. a reduced-precision
138    /// type for [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an
139    /// element is `datatype.element_size()` bytes.
140    pub fn create_streaming_dataset_compressed_with_datatype(
141        &mut self,
142        name: &str,
143        datatype: crate::DatatypeMessage,
144        frame_dims: &[u64],
145        pipeline: crate::format::messages::filter::FilterPipeline,
146    ) -> Result<usize> {
147        let idx = self
148            .inner
149            .create_streaming_dataset_compressed(name, datatype, frame_dims, pipeline)?;
150        Ok(idx)
151    }
152
153    /// Create a streaming dataset whose frames are split into fixed-size
154    /// chunk tiles.
155    ///
156    /// `frame_dims` is the per-frame shape (e.g. `[1024, 1024]`);
157    /// `frame_chunk` is the tile shape within a frame (e.g. `[256, 256]`),
158    /// of the same rank. The on-disk chunk shape becomes
159    /// `[1, frame_chunk...]`, so each frame is stored as
160    /// `product(frame_dims / frame_chunk)` chunks instead of one. This
161    /// mirrors area-detector tiling controls such as NDFileHDF5's
162    /// `nRowChunks` / `nColChunks`: it changes only the partial-read
163    /// granularity and compression unit, not the stored data.
164    /// [`append_frame`](Self::append_frame) accepts a whole frame and
165    /// splits it into tiles automatically.
166    pub fn create_streaming_dataset_tiled<T: H5Type>(
167        &mut self,
168        name: &str,
169        frame_dims: &[u64],
170        frame_chunk: &[u64],
171    ) -> Result<usize> {
172        let idx = self.inner.create_streaming_dataset_tiled(
173            name,
174            T::hdf5_type(),
175            frame_dims,
176            frame_chunk,
177        )?;
178        Ok(idx)
179    }
180
181    /// Create a compressed streaming dataset whose frames are split into
182    /// fixed-size chunk tiles. See
183    /// [`create_streaming_dataset_tiled`](Self::create_streaming_dataset_tiled)
184    /// for the meaning of `frame_chunk`; each tile is the compression unit.
185    pub fn create_streaming_dataset_tiled_compressed<T: H5Type>(
186        &mut self,
187        name: &str,
188        frame_dims: &[u64],
189        frame_chunk: &[u64],
190        pipeline: crate::format::messages::filter::FilterPipeline,
191    ) -> Result<usize> {
192        self.create_streaming_dataset_tiled_compressed_with_datatype(
193            name,
194            T::hdf5_type(),
195            frame_dims,
196            frame_chunk,
197            pipeline,
198        )
199    }
200
201    /// [`Self::create_streaming_dataset_tiled_compressed`] with an explicit stored datatype instead of a
202    /// carrier `T`, for a type no `T` describes (e.g. a reduced-precision
203    /// type for [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an
204    /// element is `datatype.element_size()` bytes.
205    pub fn create_streaming_dataset_tiled_compressed_with_datatype(
206        &mut self,
207        name: &str,
208        datatype: crate::DatatypeMessage,
209        frame_dims: &[u64],
210        frame_chunk: &[u64],
211        pipeline: crate::format::messages::filter::FilterPipeline,
212    ) -> Result<usize> {
213        let idx = self.inner.create_streaming_dataset_tiled_compressed(
214            name,
215            datatype,
216            frame_dims,
217            frame_chunk,
218            pipeline,
219        )?;
220        Ok(idx)
221    }
222
223    /// Create a streaming dataset with full control over the chunk shape,
224    /// including the frame axis.
225    ///
226    /// `chunk` is the complete per-chunk shape, of rank
227    /// `frame_dims.len() + 1`: `chunk[0]` frames per chunk (the NDFileHDF5
228    /// `nFramesChunks` control) and `chunk[1..]` the per-frame tile shape
229    /// (`nRowChunks` / `nColChunks`). When `chunk[0] > 1`,
230    /// [`append_frame`](Self::append_frame) buffers whole frames until a
231    /// chunk band fills; the final partial band is written (zero-padded) at
232    /// [`close`](Self::close), and the dataset's logical frame count always
233    /// equals the exact number of frames appended.
234    pub fn create_streaming_dataset_chunked<T: H5Type>(
235        &mut self,
236        name: &str,
237        frame_dims: &[u64],
238        chunk: &[u64],
239    ) -> Result<usize> {
240        let idx =
241            self.inner
242                .create_streaming_dataset_chunked(name, T::hdf5_type(), frame_dims, chunk)?;
243        Ok(idx)
244    }
245
246    /// Compressed variant of
247    /// [`create_streaming_dataset_chunked`](Self::create_streaming_dataset_chunked);
248    /// each chunk is filtered independently through `pipeline`.
249    pub fn create_streaming_dataset_chunked_compressed<T: H5Type>(
250        &mut self,
251        name: &str,
252        frame_dims: &[u64],
253        chunk: &[u64],
254        pipeline: crate::format::messages::filter::FilterPipeline,
255    ) -> Result<usize> {
256        self.create_streaming_dataset_chunked_compressed_with_datatype(
257            name,
258            T::hdf5_type(),
259            frame_dims,
260            chunk,
261            pipeline,
262        )
263    }
264
265    /// [`Self::create_streaming_dataset_chunked_compressed`] with an explicit stored datatype instead of a
266    /// carrier `T`, for a type no `T` describes (e.g. a reduced-precision
267    /// type for [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an
268    /// element is `datatype.element_size()` bytes.
269    pub fn create_streaming_dataset_chunked_compressed_with_datatype(
270        &mut self,
271        name: &str,
272        datatype: crate::DatatypeMessage,
273        frame_dims: &[u64],
274        chunk: &[u64],
275        pipeline: crate::format::messages::filter::FilterPipeline,
276    ) -> Result<usize> {
277        let idx = self.inner.create_streaming_dataset_chunked_compressed(
278            name, datatype, frame_dims, chunk, pipeline,
279        )?;
280        Ok(idx)
281    }
282
283    /// Create a fixed-shape multi-dimensional grid dataset that fills at
284    /// explicit positions as frames arrive.
285    ///
286    /// Unlike [`create_streaming_dataset`](Self::create_streaming_dataset),
287    /// which appends frames along a single unlimited leading axis, this
288    /// creates a dataset of the full bounded shape `dims` (no unlimited axis)
289    /// and lets you place each frame at an arbitrary chunk position with
290    /// [`write_chunk_at`](Self::write_chunk_at). This mirrors AreaDetector's
291    /// "extra dimensions" layout, where a scan of known size (e.g.
292    /// `[Na, Nb, H, W]`) is filled in odometer order. `chunk` is the per-chunk
293    /// shape of the same rank (typically `[1, …, 1, H, W]`). Returns the
294    /// dataset index.
295    pub fn create_grid_dataset<T: H5Type>(
296        &mut self,
297        name: &str,
298        dims: &[u64],
299        chunk: &[u64],
300    ) -> Result<usize> {
301        self.create_grid_dataset_with_datatype(name, T::hdf5_type(), dims, chunk)
302    }
303
304    /// [`Self::create_grid_dataset`] with an explicit stored datatype instead of a
305    /// carrier `T`, for a type no `T` describes (e.g. a reduced-precision
306    /// type for [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an
307    /// element is `datatype.element_size()` bytes.
308    pub fn create_grid_dataset_with_datatype(
309        &mut self,
310        name: &str,
311        datatype: crate::DatatypeMessage,
312        dims: &[u64],
313        chunk: &[u64],
314    ) -> Result<usize> {
315        let idx = self
316            .inner
317            .create_grid_dataset(name, datatype, dims, chunk)?;
318        Ok(idx)
319    }
320
321    /// Compressed variant of [`create_grid_dataset`](Self::create_grid_dataset);
322    /// each chunk is filtered independently through `pipeline`, and
323    /// [`write_chunk_raw_at`](Self::write_chunk_raw_at) can place chunks a
324    /// codec already compressed.
325    pub fn create_grid_dataset_compressed<T: H5Type>(
326        &mut self,
327        name: &str,
328        dims: &[u64],
329        chunk: &[u64],
330        pipeline: crate::format::messages::filter::FilterPipeline,
331    ) -> Result<usize> {
332        self.create_grid_dataset_compressed_with_datatype(
333            name,
334            T::hdf5_type(),
335            dims,
336            chunk,
337            pipeline,
338        )
339    }
340
341    /// [`Self::create_grid_dataset_compressed`] with an explicit stored
342    /// datatype instead of a carrier `T`, for a type no `T` describes (e.g. a
343    /// reduced-precision type for
344    /// [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an element is
345    /// `datatype.element_size()` bytes.
346    pub fn create_grid_dataset_compressed_with_datatype(
347        &mut self,
348        name: &str,
349        datatype: crate::DatatypeMessage,
350        dims: &[u64],
351        chunk: &[u64],
352        pipeline: crate::format::messages::filter::FilterPipeline,
353    ) -> Result<usize> {
354        let idx = self
355            .inner
356            .create_grid_dataset_compressed(name, datatype, dims, chunk, pipeline)?;
357        Ok(idx)
358    }
359
360    /// Create a hard link: an additional name for a dataset or group that
361    /// already exists in the file.
362    ///
363    /// No data is copied — the link and its target share one object header,
364    /// exactly as `h5py` / libhdf5 hard links do. This is the NeXus-style way
365    /// to expose a streaming dataset at an aliased path.
366    ///
367    /// * `parent_group_path` — full path of the group that will hold the
368    ///   link (`"/"` for the root group).
369    /// * `link_name` — leaf name of the new link within that group.
370    /// * `target_path` — full path of an existing dataset or group.
371    ///
372    /// # Visibility relative to SWMR mode
373    ///
374    /// A link created **before** [`start_swmr`](Self::start_swmr) is committed
375    /// by `start_swmr` and is visible to SWMR readers for the whole streaming
376    /// window. A link created **after** `start_swmr` is committed only by
377    /// [`close`](Self::close); it does not appear to readers that attach
378    /// during the live SWMR window. Create layout links before `start_swmr`
379    /// when readers must resolve them while streaming.
380    pub fn create_hard_link(
381        &mut self,
382        parent_group_path: &str,
383        link_name: &str,
384        target_path: &str,
385    ) -> Result<()> {
386        self.inner
387            .writer_mut()
388            .create_hard_link(parent_group_path, link_name, target_path)?;
389        Ok(())
390    }
391
392    /// Create a group in the file hierarchy.
393    ///
394    /// * `parent_group_path` — full path of the parent group (`"/"` for the
395    ///   root group).
396    /// * `name` — leaf name of the new group.
397    ///
398    /// A nested NeXus layout is built one level at a time, parent first:
399    ///
400    /// ```no_run
401    /// # use rust_hdf5::swmr::SwmrFileWriter;
402    /// # let mut writer = SwmrFileWriter::create("stream.h5").unwrap();
403    /// writer.create_group("/", "entry").unwrap();
404    /// writer.create_group("/entry", "data").unwrap();
405    /// ```
406    ///
407    /// Like [`create_hard_link`](Self::create_hard_link), a group created
408    /// before [`start_swmr`](Self::start_swmr) is visible to SWMR readers for
409    /// the whole streaming window; one created after is committed only by
410    /// [`close`](Self::close).
411    pub fn create_group(&mut self, parent_group_path: &str, name: &str) -> Result<()> {
412        self.inner
413            .writer_mut()
414            .create_group(parent_group_path, name)?;
415        Ok(())
416    }
417
418    /// Set a string attribute on a group, or on the root group when
419    /// `group_path` is `"/"`.
420    ///
421    /// This is the NeXus way to tag a group with its class — for example
422    /// `set_group_attr_string("/entry", "NX_class", "NXentry")`. An existing
423    /// attribute of the same name is replaced.
424    ///
425    /// Attributes must be set before [`start_swmr`](Self::start_swmr):
426    /// object headers are frozen while readers stream, so every attribute
427    /// setter is refused once SWMR is active — libhdf5's rule for SWMR
428    /// writes too.
429    pub fn set_group_attr_string(
430        &mut self,
431        group_path: &str,
432        name: &str,
433        value: &str,
434    ) -> Result<()> {
435        self.inner.writer_mut().set_vlen_string_attribute(
436            group_attr_target(group_path),
437            name,
438            value,
439        )?;
440        Ok(())
441    }
442
443    /// Set a numeric scalar attribute on a group, or on the root group when
444    /// `group_path` is `"/"`. An existing attribute of the same name is
445    /// replaced. Refused after [`start_swmr`](Self::start_swmr) — see
446    /// [`set_group_attr_string`](Self::set_group_attr_string).
447    pub fn set_group_attr_numeric<T: H5Type>(
448        &mut self,
449        group_path: &str,
450        name: &str,
451        value: &T,
452    ) -> Result<()> {
453        let attr = AttributeMessage::scalar_numeric(name, T::hdf5_type(), scalar_to_bytes(value));
454        self.inner
455            .writer_mut()
456            .set_attribute(group_attr_target(group_path), attr)?;
457        Ok(())
458    }
459
460    /// Create a fixed-shape (non-streaming) dataset and write all its data in
461    /// one call. Returns the dataset index.
462    ///
463    /// This is for the NeXus metadata that surrounds the image stream —
464    /// coordinate axes, detector geometry, and (with `dims = &[]`) scalar
465    /// values such as `/entry/instrument/detector/distance`. Unlike a
466    /// streaming dataset, it is written once and not appended to.
467    pub fn write_dataset<T: H5Type>(
468        &mut self,
469        name: &str,
470        dims: &[u64],
471        data: &[T],
472    ) -> Result<usize> {
473        let expected: u64 = if dims.is_empty() {
474            1
475        } else {
476            dims.iter().product()
477        };
478        if data.len() as u64 != expected {
479            return Err(crate::error::Hdf5Error::InvalidState(format!(
480                "write_dataset: data has {} elements but shape {dims:?} needs {expected}",
481                data.len()
482            )));
483        }
484        let idx = self
485            .inner
486            .writer_mut()
487            .create_dataset(name, T::hdf5_type(), dims)?;
488        self.inner
489            .writer_mut()
490            .write_dataset_raw(idx, slice_to_bytes(data))?;
491        Ok(idx)
492    }
493
494    /// Create a variable-length string dataset (one element per string).
495    /// Returns the dataset index. Useful for NeXus metadata such as
496    /// `/entry/start_time` or per-frame timestamp arrays.
497    ///
498    /// The datatype declares UTF-8;
499    /// [`write_string_dataset_ascii`](Self::write_string_dataset_ascii)
500    /// declares ASCII instead.
501    pub fn write_string_dataset(&mut self, name: &str, strings: &[&str]) -> Result<usize> {
502        let idx = self
503            .inner
504            .writer_mut()
505            .create_vlen_string_dataset(name, strings, 1)?;
506        Ok(idx)
507    }
508
509    /// [`write_string_dataset`](Self::write_string_dataset) under an **ASCII**
510    /// datatype, the one h5py's `string_dtype("ascii")` produces. A string
511    /// that is not 7-bit is rejected rather than mislabelled.
512    pub fn write_string_dataset_ascii(&mut self, name: &str, strings: &[&str]) -> Result<usize> {
513        let idx = self
514            .inner
515            .writer_mut()
516            .create_vlen_string_dataset(name, strings, 0)?;
517        Ok(idx)
518    }
519
520    /// Set a string attribute on a dataset, addressed by its index. The
521    /// NeXus way to record `units`, `long_name`, `signal`, etc. An existing
522    /// attribute of the same name is replaced. Refused after
523    /// [`start_swmr`](Self::start_swmr) — see
524    /// [`set_group_attr_string`](Self::set_group_attr_string).
525    pub fn set_dataset_attr_string(
526        &mut self,
527        ds_index: usize,
528        name: &str,
529        value: &str,
530    ) -> Result<()> {
531        self.inner.writer_mut().set_vlen_string_attribute(
532            crate::io::writer::AttrTarget::Dataset(ds_index),
533            name,
534            value,
535        )?;
536        Ok(())
537    }
538
539    /// Set a numeric scalar attribute on a dataset, addressed by its index.
540    /// An existing attribute of the same name is replaced. Refused after
541    /// [`start_swmr`](Self::start_swmr) — see
542    /// [`set_group_attr_string`](Self::set_group_attr_string).
543    pub fn set_dataset_attr_numeric<T: H5Type>(
544        &mut self,
545        ds_index: usize,
546        name: &str,
547        value: &T,
548    ) -> Result<()> {
549        let attr = AttributeMessage::scalar_numeric(name, T::hdf5_type(), scalar_to_bytes(value));
550        self.inner
551            .writer_mut()
552            .add_dataset_attribute(ds_index, attr)?;
553        Ok(())
554    }
555
556    /// Set a numeric array attribute on a dataset, addressed by its index.
557    ///
558    /// `dims` are the dimension sizes (e.g. `&[3]` for a 1-D array) and the
559    /// number of `values` must equal their product. This is the SWMR
560    /// counterpart of [`H5Attribute::write_array`](crate::H5Attribute::write_array),
561    /// for the length-ndims `int32` array attributes AreaDetector writes
562    /// (`NDArrayDimOffset`, `NDArrayDimBinning`, `NDArrayDimReverse`). An
563    /// existing attribute of the same name is replaced.
564    ///
565    /// In SWMR mode, attributes can only be added before
566    /// [`start_swmr`](Self::start_swmr); HDF5 forbids adding dataset
567    /// attributes once the file is in SWMR write mode. Resolve a dataset path
568    /// to its index with [`dataset_index`](Self::dataset_index).
569    pub fn set_dataset_attr_array<T: H5Type>(
570        &mut self,
571        ds_index: usize,
572        name: &str,
573        dims: &[u64],
574        values: &[T],
575    ) -> Result<()> {
576        // Product of an empty shape is 1 (a scalar holds one element).
577        let expected: u64 = dims.iter().product();
578        if values.len() as u64 != expected {
579            return Err(crate::error::Hdf5Error::InvalidState(format!(
580                "set_dataset_attr_array: {} values but shape {dims:?} needs {expected}",
581                values.len()
582            )));
583        }
584        let attr = AttributeMessage::array_numeric(
585            name,
586            T::hdf5_type(),
587            dims,
588            slice_to_bytes(values).to_vec(),
589        );
590        self.inner
591            .writer_mut()
592            .add_dataset_attribute(ds_index, attr)?;
593        Ok(())
594    }
595
596    /// Set the fill value of a streaming dataset, addressed by its index.
597    ///
598    /// Call this before the first [`append_frame`](Self::append_frame): it
599    /// determines the value of chunk regions that are never written (a
600    /// partial final band, or unwritten tiles).
601    pub fn set_dataset_fill_value<T: H5Type>(&mut self, ds_index: usize, value: &T) -> Result<()> {
602        self.set_dataset_fill_value_bytes(ds_index, &scalar_to_bytes(value))
603    }
604
605    /// [`set_dataset_fill_value`](Self::set_dataset_fill_value) from one
606    /// element encoded in the dataset's stored datatype — for a type no `T`
607    /// carries, such as one made with a `_with_datatype` constructor.
608    /// `bytes` is the element's host-order image, `datatype.element_size()`
609    /// bytes long; it is put in the order the datatype declares.
610    pub fn set_dataset_fill_value_bytes(&mut self, ds_index: usize, bytes: &[u8]) -> Result<()> {
611        let writer = self.inner.writer_mut();
612        if ds_index >= writer.dataset_count() {
613            return Err(Hdf5Error::InvalidState(format!(
614                "dataset index {ds_index} out of range"
615            )));
616        }
617        let datatype = writer.ds(ds_index).lock().datatype.clone();
618        let width = datatype.element_size() as usize;
619        if bytes.len() != width {
620            return Err(Hdf5Error::TypeMismatch(format!(
621                "fill value is {} bytes but datatype {datatype} stores {width}",
622                bytes.len()
623            )));
624        }
625        let stored = crate::dataset::to_stored_byte_order(bytes, &datatype, width)?;
626        writer.set_dataset_fill_value(ds_index, stored.into_owned())?;
627        Ok(())
628    }
629
630    /// Place an existing dataset inside a group.
631    ///
632    /// By default a dataset created through this writer lives at the root
633    /// level; this moves its link record into `group_path` (which must
634    /// already exist). The group must be created before `start_swmr` for the
635    /// placement to be visible to readers during streaming.
636    pub fn assign_dataset_to_group(&mut self, group_path: &str, ds_index: usize) -> Result<()> {
637        self.inner
638            .writer_mut()
639            .assign_dataset_to_group(group_path, ds_index)?;
640        Ok(())
641    }
642
643    /// Signal the start of SWMR mode.
644    pub fn start_swmr(&mut self) -> Result<()> {
645        self.inner.start_swmr()?;
646        Ok(())
647    }
648
649    /// Append a frame of raw data to a streaming dataset.
650    ///
651    /// The data size must match one frame (product of frame_dims * element_size).
652    pub fn append_frame(&mut self, ds_index: usize, data: &[u8]) -> Result<()> {
653        self.inner.append_frame(ds_index, data)?;
654        Ok(())
655    }
656
657    /// Append a frame that is already filtered (e.g. compressed by a codec
658    /// plugin upstream), storing its bytes verbatim — HDF5's direct chunk
659    /// write (`H5Dwrite_chunk`) for a streaming dataset.
660    ///
661    /// `data` is the stored (compressed) image of one frame; the dataset's
662    /// filter pipeline, recorded when it was created with a `_compressed`
663    /// constructor, is what readers run to decode it. `filter_mask` bit *i*
664    /// set means filter *i* was not applied to this frame; pass 0 when the
665    /// whole pipeline was. The dataset must store one whole frame per chunk
666    /// ([`create_streaming_dataset_compressed`](Self::create_streaming_dataset_compressed)
667    /// or a `_chunked_compressed` chunk of `[1, frame_dims...]`).
668    pub fn append_frame_raw(
669        &mut self,
670        ds_index: usize,
671        data: &[u8],
672        filter_mask: u32,
673    ) -> Result<()> {
674        self.inner.append_frame_raw(ds_index, data, filter_mask)?;
675        Ok(())
676    }
677
678    /// Write one already-filtered chunk verbatim at an explicit chunk
679    /// position of a grid dataset made by
680    /// [`create_grid_dataset_compressed`](Self::create_grid_dataset_compressed)
681    /// — HDF5's direct chunk write (`H5Dwrite_chunk`) at a chunk offset.
682    /// `filter_mask` bit *i* set means filter *i* was not applied to this
683    /// chunk; pass 0 when the whole pipeline was.
684    pub fn write_chunk_raw_at(
685        &mut self,
686        ds_index: usize,
687        chunk_coords: &[u64],
688        data: &[u8],
689        filter_mask: u32,
690    ) -> Result<()> {
691        self.inner
692            .write_chunk_raw_at(ds_index, chunk_coords, data, filter_mask)?;
693        Ok(())
694    }
695
696    /// Write one frame at an explicit chunk position of a grid dataset
697    /// created with [`create_grid_dataset`](Self::create_grid_dataset).
698    ///
699    /// `chunk_coords` are in units of chunks (row-major over the chunk grid)
700    /// and `data` must be exactly one full chunk (`product(chunk) *
701    /// element_size` bytes; edge chunks are zero-padded by the caller). The
702    /// logical extent is fixed, so positions may be written in any order and
703    /// unwritten positions read back as fill. As with the streaming path,
704    /// call [`flush`](Self::flush) to make writes visible to SWMR readers and
705    /// set dataset attributes before [`start_swmr`](Self::start_swmr).
706    pub fn write_chunk_at(
707        &mut self,
708        ds_index: usize,
709        chunk_coords: &[u64],
710        data: &[u8],
711    ) -> Result<()> {
712        self.inner.write_chunk_at(ds_index, chunk_coords, data)?;
713        Ok(())
714    }
715
716    /// Flush all dataset index structures to disk with SWMR ordering.
717    pub fn flush(&mut self) -> Result<()> {
718        self.inner.flush()?;
719        Ok(())
720    }
721
722    /// Close and finalize the file.
723    pub fn close(self) -> Result<()> {
724        self.inner.close()?;
725        Ok(())
726    }
727}
728
729/// SWMR reader for monitoring a streaming HDF5 file.
730///
731/// Opens a file being written by a concurrent [`SwmrFileWriter`] and
732/// periodically calls [`refresh`](Self::refresh) to pick up new data.
733///
734/// ```no_run
735/// use rust_hdf5::swmr::SwmrFileReader;
736///
737/// let mut reader = SwmrFileReader::open("stream.h5").unwrap();
738///
739/// loop {
740///     reader.refresh().unwrap();
741///     let names = reader.dataset_names();
742///     if let Some(shape) = reader.dataset_shape("frames").ok() {
743///         println!("frames shape: {:?}", shape);
744///         if shape[0] > 0 {
745///             let data = reader.read_dataset_raw("frames").unwrap();
746///             println!("got {} bytes", data.len());
747///             break;
748///         }
749///     }
750///     std::thread::sleep(std::time::Duration::from_millis(100));
751/// }
752/// ```
753pub struct SwmrFileReader {
754    reader: Hdf5Reader,
755}
756
757impl SwmrFileReader {
758    /// Open an HDF5 file for SWMR reading using the env-var-derived
759    /// locking policy. Takes a shared lock so it coexists with the
760    /// downgraded shared lock held by [`SwmrFileWriter`] after
761    /// `start_swmr`, and with other concurrent SWMR readers.
762    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
763        let reader = Hdf5Reader::open_swmr(path.as_ref())?;
764        Ok(Self { reader })
765    }
766
767    /// Open an HDF5 file for SWMR reading with an explicit locking policy.
768    pub fn open_with_locking<P: AsRef<Path>>(path: P, locking: FileLocking) -> Result<Self> {
769        let reader = Hdf5Reader::open_swmr_with_locking(path.as_ref(), locking)?;
770        Ok(Self { reader })
771    }
772
773    /// Re-read the superblock and dataset metadata from disk.
774    ///
775    /// Call this periodically to pick up new data written by the concurrent
776    /// SWMR writer.
777    pub fn refresh(&mut self) -> Result<()> {
778        self.reader.refresh()?;
779        Ok(())
780    }
781
782    /// Return the names of all datasets.
783    pub fn dataset_names(&self) -> Vec<String> {
784        self.reader
785            .dataset_names()
786            .iter()
787            .map(|s| s.to_string())
788            .collect()
789    }
790
791    /// Return the current shape of a dataset.
792    pub fn dataset_shape(&mut self, name: &str) -> Result<Vec<u64>> {
793        Ok(self.reader.dataset_shape(name)?)
794    }
795
796    /// Read the raw bytes of a dataset.
797    pub fn read_dataset_raw(&mut self, name: &str) -> Result<Vec<u8>> {
798        Ok(self.reader.read_dataset_raw(name)?)
799    }
800
801    /// Read a dataset as a typed vector.
802    ///
803    /// `T` must have the dataset's exact element width — the same
804    /// width-checked byte reinterpretation as
805    /// [`H5Dataset::read_raw`](crate::dataset::H5Dataset::read_raw), with no
806    /// datatype conversion. Use
807    /// [`dataset_element_size`](Self::dataset_element_size) to size-check a
808    /// type at runtime.
809    pub fn read_dataset<T: H5Type>(&mut self, name: &str) -> Result<Vec<T>> {
810        self.check_element_width::<T>(name)?;
811        let datatype = self.dataset_datatype(name)?;
812        let es = T::element_size();
813        let total = self.reader.dataset_raw_size(name)? as usize;
814        if es == 0 || !total.is_multiple_of(es) {
815            return Err(crate::error::Hdf5Error::TypeMismatch(format!(
816                "raw data size {total} is not a multiple of element size {es}"
817            )));
818        }
819        // As the non-SWMR full read: the image lands in the vector this
820        // returns rather than in a byte buffer copied into it afterwards.
821        crate::io::reader::read_image_into_new(total / es, |image| {
822            self.reader.read_dataset_raw_into_dst(
823                name,
824                image,
825                crate::io::file_handle::ReadDst::Fresh,
826            )?;
827            crate::dataset::to_host_byte_order(image, &datatype, es)
828        })
829    }
830
831    /// Read a slice (hyperslab) of a dataset as raw bytes.
832    ///
833    /// `starts[d]` is the first index along dimension `d`, `counts[d]` is how
834    /// many. For a streaming dataset this reads only the chunks the slice
835    /// overlaps — the efficient way for a live viewer to fetch the latest
836    /// frame without re-reading the whole stream.
837    pub fn read_slice_raw(
838        &mut self,
839        name: &str,
840        starts: &[u64],
841        counts: &[u64],
842    ) -> Result<Vec<u8>> {
843        Ok(self.reader.read_slice(name, starts, counts)?)
844    }
845
846    /// Read a slice (hyperslab) of a dataset as a typed vector.
847    /// See [`read_slice_raw`](Self::read_slice_raw); `T`'s width is checked
848    /// like [`read_dataset`](Self::read_dataset).
849    pub fn read_slice<T: H5Type>(
850        &mut self,
851        name: &str,
852        starts: &[u64],
853        counts: &[u64],
854    ) -> Result<Vec<T>> {
855        self.check_element_width::<T>(name)?;
856        let datatype = self.dataset_datatype(name)?;
857        bytes_to_typed(self.reader.read_slice(name, starts, counts)?, &datatype)
858    }
859
860    /// The width gate for the typed reads: without it, reinterpreting the
861    /// raw bytes splits or merges elements — an f64 dataset read as `i32`
862    /// passed the divisibility check and silently returned twice as many
863    /// garbage values.
864    fn check_element_width<T: H5Type>(&mut self, name: &str) -> Result<()> {
865        let stored = self.dataset_element_size(name)?;
866        if T::element_size() != stored {
867            return Err(crate::error::Hdf5Error::TypeMismatch(format!(
868                "read type has element size {} but dataset has element size {stored}",
869                T::element_size(),
870            )));
871        }
872        Ok(())
873    }
874
875    /// Read a variable-length string dataset.
876    pub fn read_vlen_strings(&mut self, name: &str) -> Result<Vec<String>> {
877        Ok(self.reader.read_vlen_strings(name)?)
878    }
879
880    /// Element size of a dataset's datatype, in bytes — enough to size a
881    /// read buffer without knowing the concrete element type at compile time.
882    pub fn dataset_element_size(&mut self, name: &str) -> Result<usize> {
883        self.reader
884            .dataset_info(name)
885            .map(|i| i.datatype.element_size() as usize)
886            .ok_or_else(|| crate::error::Hdf5Error::NotFound(name.to_string()))
887    }
888
889    /// The stored datatype of a dataset, which the typed reads need to put
890    /// the element image into host byte order.
891    fn dataset_datatype(&mut self, name: &str) -> Result<DatatypeMessage> {
892        self.reader
893            .dataset_info(name)
894            .map(|i| i.datatype.clone())
895            .ok_or_else(|| crate::error::Hdf5Error::NotFound(name.to_string()))
896    }
897
898    /// All group paths in the file.
899    pub fn group_paths(&self) -> Vec<String> {
900        self.reader.group_paths().iter().cloned().collect()
901    }
902
903    /// Whether a group exists. A leading `/` is tolerated.
904    pub fn has_group(&self, group_path: &str) -> bool {
905        self.reader.has_group(group_path.trim_start_matches('/'))
906    }
907
908    /// Names of the attributes on a dataset.
909    pub fn dataset_attr_names(&mut self, name: &str) -> Result<Vec<String>> {
910        Ok(self.reader.dataset_attr_names(name)?)
911    }
912
913    /// Read a dataset's attribute as a string (e.g. `units`, `NX_class`).
914    pub fn dataset_attr_string(&mut self, dataset: &str, attr: &str) -> Result<String> {
915        let a = self.reader.dataset_attr(dataset, attr)?.clone();
916        Ok(self.reader.attr_string_value(&a)?)
917    }
918
919    /// Names of the attributes on a group, or on the root group when
920    /// `group_path` is `"/"`. A leading `/` is tolerated.
921    pub fn group_attr_names(&mut self, group_path: &str) -> Result<Vec<String>> {
922        Ok(if group_path == "/" {
923            self.reader.root_attr_names()?
924        } else {
925            self.reader
926                .group_attr_names(group_path.trim_start_matches('/'))?
927        })
928    }
929
930    /// Read a group's attribute as a string (the NeXus `NX_class` etc.), or
931    /// a root attribute when `group_path` is `"/"`. A leading `/` is tolerated.
932    pub fn group_attr_string(&mut self, group_path: &str, attr: &str) -> Result<String> {
933        let a = if group_path == "/" {
934            self.reader.root_attr(attr)
935        } else {
936            self.reader
937                .group_attr(group_path.trim_start_matches('/'), attr)
938        }?
939        .clone();
940        Ok(self.reader.attr_string_value(&a)?)
941    }
942}
943
944/// The writer-side attribute list a group path names: `"/"` is the
945/// file-level (root) list, anything else the group's own.
946fn group_attr_target(group_path: &str) -> crate::io::writer::AttrTarget<'_> {
947    if group_path == "/" {
948        crate::io::writer::AttrTarget::Root
949    } else {
950        crate::io::writer::AttrTarget::Group(group_path)
951    }
952}
953
954/// Reinterpret a raw byte buffer as a typed vector. The buffer length must
955/// be a whole multiple of `T`'s element size, and the stored byte order is
956/// converted to the host's first — reinterpretation is only the stored value
957/// when the two agree.
958fn bytes_to_typed<T: H5Type>(mut raw: Vec<u8>, datatype: &DatatypeMessage) -> Result<Vec<T>> {
959    let es = T::element_size();
960    if es == 0 || !raw.len().is_multiple_of(es) {
961        return Err(crate::error::Hdf5Error::TypeMismatch(format!(
962            "raw data size {} is not a multiple of element size {es}",
963            raw.len()
964        )));
965    }
966    crate::dataset::to_host_byte_order(&mut raw, datatype, es)?;
967    let count = raw.len() / es;
968    let mut result = Vec::<T>::with_capacity(count);
969    // Safety: `T: H5Type` is a `Copy` POD primitive exactly `element_size()`
970    // bytes wide, so the byte buffer is a valid array of `count` `T`s.
971    unsafe {
972        std::ptr::copy_nonoverlapping(raw.as_ptr(), result.as_mut_ptr() as *mut u8, raw.len());
973        result.set_len(count);
974    }
975    Ok(result)
976}
977
978/// Raw bytes of one `H5Type` scalar.
979fn scalar_to_bytes<T: H5Type>(value: &T) -> Vec<u8> {
980    let es = T::element_size();
981    // Safety: `T: H5Type` is a `Copy` POD primitive exactly `element_size()`
982    // bytes wide.
983    unsafe { std::slice::from_raw_parts(value as *const T as *const u8, es) }.to_vec()
984}
985
986/// Raw bytes of an `H5Type` slice, in element order.
987fn slice_to_bytes<T: H5Type>(data: &[T]) -> &[u8] {
988    // Safety: as `scalar_to_bytes`; the slice is a contiguous POD array.
989    unsafe { std::slice::from_raw_parts(data.as_ptr() as *const u8, std::mem::size_of_val(data)) }
990}