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    /// [`Self::create_streaming_dataset_chunked`] with a frame axis that holds
284    /// at most `max_frames` frames (`None`: unlimited) and an explicit stored
285    /// datatype, filtered through `pipeline` when one is given: NDFileHDF5's
286    /// maxdims\[0\] = NumCapture. A bounded frame axis is indexed by a fixed
287    /// array, the index libhdf5 picks for it, and an append past
288    /// `max_frames` is refused.
289    pub fn create_streaming_dataset_with_max_frames(
290        &mut self,
291        name: &str,
292        datatype: crate::DatatypeMessage,
293        frame_dims: &[u64],
294        chunk: &[u64],
295        max_frames: Option<u64>,
296        pipeline: Option<crate::format::messages::filter::FilterPipeline>,
297    ) -> Result<usize> {
298        let idx = self.inner.create_streaming_dataset_with_max_frames(
299            name, datatype, frame_dims, chunk, max_frames, pipeline,
300        )?;
301        Ok(idx)
302    }
303
304    /// Create a fixed-shape multi-dimensional grid dataset that fills at
305    /// explicit positions as frames arrive.
306    ///
307    /// Unlike [`create_streaming_dataset`](Self::create_streaming_dataset),
308    /// which appends frames along a single unlimited leading axis, this
309    /// creates a dataset of the full bounded shape `dims` (no unlimited axis)
310    /// and lets you place each frame at an arbitrary chunk position with
311    /// [`write_chunk_at`](Self::write_chunk_at). This mirrors AreaDetector's
312    /// "extra dimensions" layout, where a scan of known size (e.g.
313    /// `[Na, Nb, H, W]`) is filled in odometer order. `chunk` is the per-chunk
314    /// shape of the same rank (typically `[1, …, 1, H, W]`). `dims` is also
315    /// the maximum shape; [`set_extent`](Self::set_extent) moves the extent
316    /// within it. Returns the dataset index.
317    pub fn create_grid_dataset<T: H5Type>(
318        &mut self,
319        name: &str,
320        dims: &[u64],
321        chunk: &[u64],
322    ) -> Result<usize> {
323        self.create_grid_dataset_with_datatype(name, T::hdf5_type(), dims, chunk)
324    }
325
326    /// [`Self::create_grid_dataset`] with an explicit stored datatype instead of a
327    /// carrier `T`, for a type no `T` describes (e.g. a reduced-precision
328    /// type for [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an
329    /// element is `datatype.element_size()` bytes.
330    pub fn create_grid_dataset_with_datatype(
331        &mut self,
332        name: &str,
333        datatype: crate::DatatypeMessage,
334        dims: &[u64],
335        chunk: &[u64],
336    ) -> Result<usize> {
337        let idx = self
338            .inner
339            .create_grid_dataset(name, datatype, dims, chunk)?;
340        Ok(idx)
341    }
342
343    /// Compressed variant of [`create_grid_dataset`](Self::create_grid_dataset);
344    /// each chunk is filtered independently through `pipeline`, and
345    /// [`write_chunk_raw_at`](Self::write_chunk_raw_at) can place chunks a
346    /// codec already compressed.
347    pub fn create_grid_dataset_compressed<T: H5Type>(
348        &mut self,
349        name: &str,
350        dims: &[u64],
351        chunk: &[u64],
352        pipeline: crate::format::messages::filter::FilterPipeline,
353    ) -> Result<usize> {
354        self.create_grid_dataset_compressed_with_datatype(
355            name,
356            T::hdf5_type(),
357            dims,
358            chunk,
359            pipeline,
360        )
361    }
362
363    /// [`Self::create_grid_dataset_compressed`] with an explicit stored
364    /// datatype instead of a carrier `T`, for a type no `T` describes (e.g. a
365    /// reduced-precision type for
366    /// [`FilterPipeline::nbit`](crate::FilterPipeline::nbit)); an element is
367    /// `datatype.element_size()` bytes.
368    pub fn create_grid_dataset_compressed_with_datatype(
369        &mut self,
370        name: &str,
371        datatype: crate::DatatypeMessage,
372        dims: &[u64],
373        chunk: &[u64],
374        pipeline: crate::format::messages::filter::FilterPipeline,
375    ) -> Result<usize> {
376        let idx = self
377            .inner
378            .create_grid_dataset_compressed(name, datatype, dims, chunk, pipeline)?;
379        Ok(idx)
380    }
381
382    /// Create a hard link: an additional name for a dataset or group that
383    /// already exists in the file.
384    ///
385    /// No data is copied — the link and its target share one object header,
386    /// exactly as `h5py` / libhdf5 hard links do. This is the NeXus-style way
387    /// to expose a streaming dataset at an aliased path.
388    ///
389    /// * `parent_group_path` — full path of the group that will hold the
390    ///   link (`"/"` for the root group).
391    /// * `link_name` — leaf name of the new link within that group.
392    /// * `target_path` — full path of an existing dataset or group.
393    ///
394    /// # Visibility relative to SWMR mode
395    ///
396    /// A link created **before** [`start_swmr`](Self::start_swmr) is committed
397    /// by `start_swmr` and is visible to SWMR readers for the whole streaming
398    /// window. A link created **after** `start_swmr` is committed only by
399    /// [`close`](Self::close); it does not appear to readers that attach
400    /// during the live SWMR window. Create layout links before `start_swmr`
401    /// when readers must resolve them while streaming.
402    pub fn create_hard_link(
403        &mut self,
404        parent_group_path: &str,
405        link_name: &str,
406        target_path: &str,
407    ) -> Result<()> {
408        self.inner
409            .writer_mut()
410            .create_hard_link(parent_group_path, link_name, target_path)?;
411        Ok(())
412    }
413
414    /// Create a group in the file hierarchy.
415    ///
416    /// * `parent_group_path` — full path of the parent group (`"/"` for the
417    ///   root group).
418    /// * `name` — leaf name of the new group.
419    ///
420    /// A nested NeXus layout is built one level at a time, parent first:
421    ///
422    /// ```no_run
423    /// # use rust_hdf5::swmr::SwmrFileWriter;
424    /// # let mut writer = SwmrFileWriter::create("stream.h5").unwrap();
425    /// writer.create_group("/", "entry").unwrap();
426    /// writer.create_group("/entry", "data").unwrap();
427    /// ```
428    ///
429    /// Like [`create_hard_link`](Self::create_hard_link), a group created
430    /// before [`start_swmr`](Self::start_swmr) is visible to SWMR readers for
431    /// the whole streaming window; one created after is committed only by
432    /// [`close`](Self::close).
433    pub fn create_group(&mut self, parent_group_path: &str, name: &str) -> Result<()> {
434        self.inner
435            .writer_mut()
436            .create_group(parent_group_path, name)?;
437        Ok(())
438    }
439
440    /// Set a string attribute on a group, or on the root group when
441    /// `group_path` is `"/"`.
442    ///
443    /// This is the NeXus way to tag a group with its class — for example
444    /// `set_group_attr_string("/entry", "NX_class", "NXentry")`. An existing
445    /// attribute of the same name is replaced.
446    ///
447    /// Attributes must be set before [`start_swmr`](Self::start_swmr):
448    /// object headers are frozen while readers stream, so every attribute
449    /// setter is refused once SWMR is active — libhdf5's rule for SWMR
450    /// writes too.
451    pub fn set_group_attr_string(
452        &mut self,
453        group_path: &str,
454        name: &str,
455        value: &str,
456    ) -> Result<()> {
457        self.inner.writer_mut().set_vlen_string_attribute(
458            group_attr_target(group_path),
459            name,
460            value,
461        )?;
462        Ok(())
463    }
464
465    /// Set a numeric scalar attribute on a group, or on the root group when
466    /// `group_path` is `"/"`. An existing attribute of the same name is
467    /// replaced. Refused after [`start_swmr`](Self::start_swmr) — see
468    /// [`set_group_attr_string`](Self::set_group_attr_string).
469    pub fn set_group_attr_numeric<T: H5Type>(
470        &mut self,
471        group_path: &str,
472        name: &str,
473        value: &T,
474    ) -> Result<()> {
475        let attr = AttributeMessage::scalar_numeric(name, T::hdf5_type(), scalar_to_bytes(value));
476        self.inner
477            .writer_mut()
478            .set_attribute(group_attr_target(group_path), attr)?;
479        Ok(())
480    }
481
482    /// Set a numeric array attribute on a group, or on the root group when
483    /// `group_path` is `"/"` — the group form of
484    /// [`set_dataset_attr_array`](Self::set_dataset_attr_array), with the
485    /// same shape rule. Refused after [`start_swmr`](Self::start_swmr) — see
486    /// [`set_group_attr_string`](Self::set_group_attr_string).
487    pub fn set_group_attr_array<T: H5Type>(
488        &mut self,
489        group_path: &str,
490        name: &str,
491        dims: &[u64],
492        values: &[T],
493    ) -> Result<()> {
494        let attr = array_attr(name, dims, values)?;
495        self.inner
496            .writer_mut()
497            .set_attribute(group_attr_target(group_path), attr)?;
498        Ok(())
499    }
500
501    /// Set a scalar attribute of a caller-supplied datatype on a group, or on
502    /// the root group when `group_path` is `"/"` — the SWMR form of
503    /// [`H5Group::set_attr_typed`](crate::H5Group::set_attr_typed). Refused
504    /// after [`start_swmr`](Self::start_swmr) — see
505    /// [`set_group_attr_string`](Self::set_group_attr_string).
506    pub fn set_group_attr_typed(
507        &mut self,
508        group_path: &str,
509        name: &str,
510        datatype: DatatypeMessage,
511        value: Vec<u8>,
512    ) -> Result<()> {
513        let attr = crate::attribute::typed_scalar_attr(name, datatype, value)?;
514        self.inner
515            .writer_mut()
516            .set_attribute(group_attr_target(group_path), attr)?;
517        Ok(())
518    }
519
520    /// Create a fixed-shape (non-streaming) dataset and write all its data in
521    /// one call. Returns the dataset index.
522    ///
523    /// This is for the NeXus metadata that surrounds the image stream —
524    /// coordinate axes, detector geometry, and (with `dims = &[]`) scalar
525    /// values such as `/entry/instrument/detector/distance`. Unlike a
526    /// streaming dataset, it is written once and not appended to.
527    pub fn write_dataset<T: H5Type>(
528        &mut self,
529        name: &str,
530        dims: &[u64],
531        data: &[T],
532    ) -> Result<usize> {
533        let expected: u64 = if dims.is_empty() {
534            1
535        } else {
536            dims.iter().product()
537        };
538        if data.len() as u64 != expected {
539            return Err(crate::error::Hdf5Error::InvalidState(format!(
540                "write_dataset: data has {} elements but shape {dims:?} needs {expected}",
541                data.len()
542            )));
543        }
544        let idx = self
545            .inner
546            .writer_mut()
547            .create_dataset(name, T::hdf5_type(), dims)?;
548        self.inner
549            .writer_mut()
550            .write_dataset_raw(idx, slice_to_bytes(data))?;
551        Ok(idx)
552    }
553
554    /// Create a variable-length string dataset (one element per string).
555    /// Returns the dataset index. Useful for NeXus metadata such as
556    /// `/entry/start_time` or per-frame timestamp arrays.
557    ///
558    /// The datatype declares UTF-8;
559    /// [`write_string_dataset_ascii`](Self::write_string_dataset_ascii)
560    /// declares ASCII instead.
561    pub fn write_string_dataset(&mut self, name: &str, strings: &[&str]) -> Result<usize> {
562        let idx = self
563            .inner
564            .writer_mut()
565            .create_vlen_string_dataset(name, strings, 1)?;
566        Ok(idx)
567    }
568
569    /// [`write_string_dataset`](Self::write_string_dataset) under an **ASCII**
570    /// datatype, the one h5py's `string_dtype("ascii")` produces. A string
571    /// that is not 7-bit is rejected rather than mislabelled.
572    pub fn write_string_dataset_ascii(&mut self, name: &str, strings: &[&str]) -> Result<usize> {
573        let idx = self
574            .inner
575            .writer_mut()
576            .create_vlen_string_dataset(name, strings, 0)?;
577        Ok(idx)
578    }
579
580    /// Set a string attribute on a dataset, addressed by its index. The
581    /// NeXus way to record `units`, `long_name`, `signal`, etc. An existing
582    /// attribute of the same name is replaced. Refused after
583    /// [`start_swmr`](Self::start_swmr) — see
584    /// [`set_group_attr_string`](Self::set_group_attr_string).
585    pub fn set_dataset_attr_string(
586        &mut self,
587        ds_index: usize,
588        name: &str,
589        value: &str,
590    ) -> Result<()> {
591        self.inner.writer_mut().set_vlen_string_attribute(
592            crate::io::writer::AttrTarget::Dataset(ds_index),
593            name,
594            value,
595        )?;
596        Ok(())
597    }
598
599    /// Set a numeric scalar attribute on a dataset, addressed by its index.
600    /// An existing attribute of the same name is replaced. Refused after
601    /// [`start_swmr`](Self::start_swmr) — see
602    /// [`set_group_attr_string`](Self::set_group_attr_string).
603    pub fn set_dataset_attr_numeric<T: H5Type>(
604        &mut self,
605        ds_index: usize,
606        name: &str,
607        value: &T,
608    ) -> Result<()> {
609        let attr = AttributeMessage::scalar_numeric(name, T::hdf5_type(), scalar_to_bytes(value));
610        self.inner
611            .writer_mut()
612            .add_dataset_attribute(ds_index, attr)?;
613        Ok(())
614    }
615
616    /// Set a scalar attribute of a caller-supplied datatype on a dataset,
617    /// addressed by its index — the SWMR form of
618    /// [`H5Dataset::set_attr_typed`](crate::H5Dataset::set_attr_typed).
619    /// Refused after [`start_swmr`](Self::start_swmr) — see
620    /// [`set_group_attr_string`](Self::set_group_attr_string).
621    pub fn set_dataset_attr_typed(
622        &mut self,
623        ds_index: usize,
624        name: &str,
625        datatype: DatatypeMessage,
626        value: Vec<u8>,
627    ) -> Result<()> {
628        let attr = crate::attribute::typed_scalar_attr(name, datatype, value)?;
629        self.inner
630            .writer_mut()
631            .add_dataset_attribute(ds_index, attr)?;
632        Ok(())
633    }
634
635    /// Set a numeric array attribute on a dataset, addressed by its index.
636    ///
637    /// `dims` are the dimension sizes (e.g. `&[3]` for a 1-D array) and the
638    /// number of `values` must equal their product. This is the SWMR
639    /// counterpart of [`H5Attribute::write_array`](crate::H5Attribute::write_array),
640    /// for the length-ndims `int32` array attributes AreaDetector writes
641    /// (`NDArrayDimOffset`, `NDArrayDimBinning`, `NDArrayDimReverse`). An
642    /// existing attribute of the same name is replaced.
643    ///
644    /// In SWMR mode, attributes can only be added before
645    /// [`start_swmr`](Self::start_swmr); HDF5 forbids adding dataset
646    /// attributes once the file is in SWMR write mode. Resolve a dataset path
647    /// to its index with [`dataset_index`](Self::dataset_index).
648    pub fn set_dataset_attr_array<T: H5Type>(
649        &mut self,
650        ds_index: usize,
651        name: &str,
652        dims: &[u64],
653        values: &[T],
654    ) -> Result<()> {
655        let attr = array_attr(name, dims, values)?;
656        self.inner
657            .writer_mut()
658            .add_dataset_attribute(ds_index, attr)?;
659        Ok(())
660    }
661
662    /// Set the fill value of a streaming dataset, addressed by its index.
663    ///
664    /// Call this before the first [`append_frame`](Self::append_frame): it
665    /// determines the value of chunk regions that are never written (a
666    /// partial final band, or unwritten tiles).
667    pub fn set_dataset_fill_value<T: H5Type>(&mut self, ds_index: usize, value: &T) -> Result<()> {
668        self.set_dataset_fill_value_bytes(ds_index, &scalar_to_bytes(value))
669    }
670
671    /// [`set_dataset_fill_value`](Self::set_dataset_fill_value) from one
672    /// element encoded in the dataset's stored datatype — for a type no `T`
673    /// carries, such as one made with a `_with_datatype` constructor.
674    /// `bytes` is the element's host-order image, `datatype.element_size()`
675    /// bytes long; it is put in the order the datatype declares.
676    pub fn set_dataset_fill_value_bytes(&mut self, ds_index: usize, bytes: &[u8]) -> Result<()> {
677        let writer = self.inner.writer_mut();
678        if ds_index >= writer.dataset_count() {
679            return Err(Hdf5Error::InvalidState(format!(
680                "dataset index {ds_index} out of range"
681            )));
682        }
683        let datatype = writer.ds(ds_index).lock().datatype.clone();
684        let width = datatype.element_size() as usize;
685        if bytes.len() != width {
686            return Err(Hdf5Error::TypeMismatch(format!(
687                "fill value is {} bytes but datatype {datatype} stores {width}",
688                bytes.len()
689            )));
690        }
691        let stored = crate::dataset::to_stored_byte_order(bytes, &datatype, width)?;
692        writer.set_dataset_fill_value(ds_index, stored.into_owned())?;
693        Ok(())
694    }
695
696    /// Place an existing dataset inside a group.
697    ///
698    /// By default a dataset created through this writer lives at the root
699    /// level; this moves its link record into `group_path` (which must
700    /// already exist). The group must be created before `start_swmr` for the
701    /// placement to be visible to readers during streaming.
702    pub fn assign_dataset_to_group(&mut self, group_path: &str, ds_index: usize) -> Result<()> {
703        self.inner
704            .writer_mut()
705            .assign_dataset_to_group(group_path, ds_index)?;
706        Ok(())
707    }
708
709    /// Signal the start of SWMR mode.
710    pub fn start_swmr(&mut self) -> Result<()> {
711        self.inner.start_swmr()?;
712        Ok(())
713    }
714
715    /// Append a frame of raw data to a streaming dataset.
716    ///
717    /// The data size must match one frame (product of frame_dims * element_size).
718    pub fn append_frame(&mut self, ds_index: usize, data: &[u8]) -> Result<()> {
719        self.inner.append_frame(ds_index, data)?;
720        Ok(())
721    }
722
723    /// Append a frame that is already filtered (e.g. compressed by a codec
724    /// plugin upstream), storing its bytes verbatim — HDF5's direct chunk
725    /// write (`H5Dwrite_chunk`) for a streaming dataset.
726    ///
727    /// `data` is the stored (compressed) image of one frame; the dataset's
728    /// filter pipeline, recorded when it was created with a `_compressed`
729    /// constructor, is what readers run to decode it. `filter_mask` bit *i*
730    /// set means filter *i* was not applied to this frame; pass 0 when the
731    /// whole pipeline was. The dataset must store one whole frame per chunk
732    /// ([`create_streaming_dataset_compressed`](Self::create_streaming_dataset_compressed),
733    /// or a `_chunked_compressed` or filtered `_with_max_frames` chunk of
734    /// `[1, frame_dims...]`).
735    pub fn append_frame_raw(
736        &mut self,
737        ds_index: usize,
738        data: &[u8],
739        filter_mask: u32,
740    ) -> Result<()> {
741        self.inner.append_frame_raw(ds_index, data, filter_mask)?;
742        Ok(())
743    }
744
745    /// Write one already-filtered chunk verbatim at an explicit chunk
746    /// position of a grid dataset made by
747    /// [`create_grid_dataset_compressed`](Self::create_grid_dataset_compressed)
748    /// — HDF5's direct chunk write (`H5Dwrite_chunk`) at a chunk offset.
749    /// `filter_mask` bit *i* set means filter *i* was not applied to this
750    /// chunk; pass 0 when the whole pipeline was.
751    pub fn write_chunk_raw_at(
752        &mut self,
753        ds_index: usize,
754        chunk_coords: &[u64],
755        data: &[u8],
756        filter_mask: u32,
757    ) -> Result<()> {
758        self.inner
759            .write_chunk_raw_at(ds_index, chunk_coords, data, filter_mask)?;
760        Ok(())
761    }
762
763    /// Write one frame at an explicit chunk position of a grid dataset
764    /// created with [`create_grid_dataset`](Self::create_grid_dataset).
765    ///
766    /// `chunk_coords` are in units of chunks (row-major over the chunk grid)
767    /// and `data` must be exactly one full chunk (`product(chunk) *
768    /// element_size` bytes; edge chunks are zero-padded by the caller).
769    /// Positions may be written in any order, beyond the extent too while
770    /// inside the maximum shape, and unwritten positions read back as fill.
771    /// As with the streaming path, call [`flush`](Self::flush) to make writes
772    /// visible to SWMR readers and set dataset attributes before
773    /// [`start_swmr`](Self::start_swmr).
774    pub fn write_chunk_at(
775        &mut self,
776        ds_index: usize,
777        chunk_coords: &[u64],
778        data: &[u8],
779    ) -> Result<()> {
780        self.inner.write_chunk_at(ds_index, chunk_coords, data)?;
781        Ok(())
782    }
783
784    /// Set a dataset's logical extent within its maximum shape, growing or
785    /// shrinking it — libhdf5's `H5Dset_extent`, e.g. a grid's extent
786    /// following the positions written so far. A multi-frame-chunk stream,
787    /// whose extent counts the frames it buffers, is refused.
788    pub fn set_extent(&mut self, ds_index: usize, dims: &[u64]) -> Result<()> {
789        self.inner.set_extent(ds_index, dims)?;
790        Ok(())
791    }
792
793    /// Flush all dataset index structures to disk with SWMR ordering.
794    pub fn flush(&mut self) -> Result<()> {
795        self.inner.flush()?;
796        Ok(())
797    }
798
799    /// Close and finalize the file.
800    pub fn close(self) -> Result<()> {
801        self.inner.close()?;
802        Ok(())
803    }
804}
805
806/// SWMR reader for monitoring a streaming HDF5 file.
807///
808/// Opens a file being written by a concurrent [`SwmrFileWriter`] and
809/// periodically calls [`refresh`](Self::refresh) to pick up new data.
810///
811/// ```no_run
812/// use rust_hdf5::swmr::SwmrFileReader;
813///
814/// let mut reader = SwmrFileReader::open("stream.h5").unwrap();
815///
816/// loop {
817///     reader.refresh().unwrap();
818///     let names = reader.dataset_names();
819///     if let Some(shape) = reader.dataset_shape("frames").ok() {
820///         println!("frames shape: {:?}", shape);
821///         if shape[0] > 0 {
822///             let data = reader.read_dataset_raw("frames").unwrap();
823///             println!("got {} bytes", data.len());
824///             break;
825///         }
826///     }
827///     std::thread::sleep(std::time::Duration::from_millis(100));
828/// }
829/// ```
830pub struct SwmrFileReader {
831    reader: Hdf5Reader,
832}
833
834impl SwmrFileReader {
835    /// Open an HDF5 file for SWMR reading using the env-var-derived
836    /// locking policy. Takes a shared lock so it coexists with the
837    /// downgraded shared lock held by [`SwmrFileWriter`] after
838    /// `start_swmr`, and with other concurrent SWMR readers.
839    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
840        let reader = Hdf5Reader::open_swmr(path.as_ref())?;
841        Ok(Self { reader })
842    }
843
844    /// Open an HDF5 file for SWMR reading with an explicit locking policy.
845    pub fn open_with_locking<P: AsRef<Path>>(path: P, locking: FileLocking) -> Result<Self> {
846        let reader = Hdf5Reader::open_swmr_with_locking(path.as_ref(), locking)?;
847        Ok(Self { reader })
848    }
849
850    /// Re-read the superblock and dataset metadata from disk.
851    ///
852    /// Call this periodically to pick up new data written by the concurrent
853    /// SWMR writer.
854    pub fn refresh(&mut self) -> Result<()> {
855        self.reader.refresh()?;
856        Ok(())
857    }
858
859    /// Return the names of all datasets.
860    pub fn dataset_names(&self) -> Vec<String> {
861        self.reader
862            .dataset_names()
863            .iter()
864            .map(|s| s.to_string())
865            .collect()
866    }
867
868    /// Return the current shape of a dataset.
869    pub fn dataset_shape(&mut self, name: &str) -> Result<Vec<u64>> {
870        Ok(self.reader.dataset_shape(name)?)
871    }
872
873    /// Read the raw bytes of a dataset.
874    pub fn read_dataset_raw(&mut self, name: &str) -> Result<Vec<u8>> {
875        Ok(self.reader.read_dataset_raw(name)?)
876    }
877
878    /// Read a dataset as a typed vector.
879    ///
880    /// `T` must have the dataset's exact element width — the same
881    /// width-checked byte reinterpretation as
882    /// [`H5Dataset::read_raw`](crate::dataset::H5Dataset::read_raw), with no
883    /// datatype conversion. Use
884    /// [`dataset_element_size`](Self::dataset_element_size) to size-check a
885    /// type at runtime.
886    pub fn read_dataset<T: H5Type>(&mut self, name: &str) -> Result<Vec<T>> {
887        self.check_element_width::<T>(name)?;
888        let datatype = self.dataset_datatype(name)?;
889        let es = T::element_size();
890        let total = self.reader.dataset_raw_size(name)? as usize;
891        if es == 0 || !total.is_multiple_of(es) {
892            return Err(crate::error::Hdf5Error::TypeMismatch(format!(
893                "raw data size {total} is not a multiple of element size {es}"
894            )));
895        }
896        // As the non-SWMR full read: the image lands in the vector this
897        // returns rather than in a byte buffer copied into it afterwards.
898        crate::io::reader::read_image_into_new(total / es, |image| {
899            self.reader.read_dataset_raw_into_dst(
900                name,
901                image,
902                crate::io::file_handle::ReadDst::Fresh,
903            )?;
904            crate::dataset::to_host_byte_order(image, &datatype, es)
905        })
906    }
907
908    /// Read a slice (hyperslab) of a dataset as raw bytes.
909    ///
910    /// `starts[d]` is the first index along dimension `d`, `counts[d]` is how
911    /// many. For a streaming dataset this reads only the chunks the slice
912    /// overlaps — the efficient way for a live viewer to fetch the latest
913    /// frame without re-reading the whole stream.
914    pub fn read_slice_raw(
915        &mut self,
916        name: &str,
917        starts: &[u64],
918        counts: &[u64],
919    ) -> Result<Vec<u8>> {
920        Ok(self.reader.read_slice(name, starts, counts)?)
921    }
922
923    /// Read a slice (hyperslab) of a dataset as a typed vector.
924    /// See [`read_slice_raw`](Self::read_slice_raw); `T`'s width is checked
925    /// like [`read_dataset`](Self::read_dataset).
926    pub fn read_slice<T: H5Type>(
927        &mut self,
928        name: &str,
929        starts: &[u64],
930        counts: &[u64],
931    ) -> Result<Vec<T>> {
932        self.check_element_width::<T>(name)?;
933        let datatype = self.dataset_datatype(name)?;
934        bytes_to_typed(self.reader.read_slice(name, starts, counts)?, &datatype)
935    }
936
937    /// The width gate for the typed reads: without it, reinterpreting the
938    /// raw bytes splits or merges elements — an f64 dataset read as `i32`
939    /// passed the divisibility check and silently returned twice as many
940    /// garbage values.
941    fn check_element_width<T: H5Type>(&mut self, name: &str) -> Result<()> {
942        let stored = self.dataset_element_size(name)?;
943        if T::element_size() != stored {
944            return Err(crate::error::Hdf5Error::TypeMismatch(format!(
945                "read type has element size {} but dataset has element size {stored}",
946                T::element_size(),
947            )));
948        }
949        Ok(())
950    }
951
952    /// Read a variable-length string dataset.
953    pub fn read_vlen_strings(&mut self, name: &str) -> Result<Vec<String>> {
954        Ok(self.reader.read_vlen_strings(name)?)
955    }
956
957    /// Element size of a dataset's datatype, in bytes — enough to size a
958    /// read buffer without knowing the concrete element type at compile time.
959    pub fn dataset_element_size(&mut self, name: &str) -> Result<usize> {
960        self.reader
961            .dataset_info(name)
962            .map(|i| i.datatype.element_size() as usize)
963            .ok_or_else(|| crate::error::Hdf5Error::NotFound(name.to_string()))
964    }
965
966    /// The stored datatype of a dataset, which the typed reads need to put
967    /// the element image into host byte order.
968    fn dataset_datatype(&mut self, name: &str) -> Result<DatatypeMessage> {
969        self.reader
970            .dataset_info(name)
971            .map(|i| i.datatype.clone())
972            .ok_or_else(|| crate::error::Hdf5Error::NotFound(name.to_string()))
973    }
974
975    /// All group paths in the file.
976    pub fn group_paths(&self) -> Vec<String> {
977        self.reader.group_paths().iter().cloned().collect()
978    }
979
980    /// Whether a group exists. A leading `/` is tolerated.
981    pub fn has_group(&self, group_path: &str) -> bool {
982        self.reader.has_group(group_path.trim_start_matches('/'))
983    }
984
985    /// Names of the attributes on a dataset.
986    pub fn dataset_attr_names(&mut self, name: &str) -> Result<Vec<String>> {
987        Ok(self.reader.dataset_attr_names(name)?)
988    }
989
990    /// Read a dataset's attribute as a string (e.g. `units`, `NX_class`).
991    pub fn dataset_attr_string(&mut self, dataset: &str, attr: &str) -> Result<String> {
992        let a = self.reader.dataset_attr(dataset, attr)?.clone();
993        Ok(self.reader.attr_string_value(&a)?)
994    }
995
996    /// Names of the attributes on a group, or on the root group when
997    /// `group_path` is `"/"`. A leading `/` is tolerated.
998    pub fn group_attr_names(&mut self, group_path: &str) -> Result<Vec<String>> {
999        Ok(if group_path == "/" {
1000            self.reader.root_attr_names()?
1001        } else {
1002            self.reader
1003                .group_attr_names(group_path.trim_start_matches('/'))?
1004        })
1005    }
1006
1007    /// Read a group's attribute as a string (the NeXus `NX_class` etc.), or
1008    /// a root attribute when `group_path` is `"/"`. A leading `/` is tolerated.
1009    pub fn group_attr_string(&mut self, group_path: &str, attr: &str) -> Result<String> {
1010        let a = if group_path == "/" {
1011            self.reader.root_attr(attr)
1012        } else {
1013            self.reader
1014                .group_attr(group_path.trim_start_matches('/'), attr)
1015        }?
1016        .clone();
1017        Ok(self.reader.attr_string_value(&a)?)
1018    }
1019}
1020
1021/// The writer-side attribute list a group path names: `"/"` is the
1022/// file-level (root) list, anything else the group's own.
1023fn group_attr_target(group_path: &str) -> crate::io::writer::AttrTarget<'_> {
1024    if group_path == "/" {
1025        crate::io::writer::AttrTarget::Root
1026    } else {
1027        crate::io::writer::AttrTarget::Group(group_path)
1028    }
1029}
1030
1031/// Reinterpret a raw byte buffer as a typed vector. The buffer length must
1032/// be a whole multiple of `T`'s element size, and the stored byte order is
1033/// converted to the host's first — reinterpretation is only the stored value
1034/// when the two agree.
1035fn bytes_to_typed<T: H5Type>(mut raw: Vec<u8>, datatype: &DatatypeMessage) -> Result<Vec<T>> {
1036    let es = T::element_size();
1037    if es == 0 || !raw.len().is_multiple_of(es) {
1038        return Err(crate::error::Hdf5Error::TypeMismatch(format!(
1039            "raw data size {} is not a multiple of element size {es}",
1040            raw.len()
1041        )));
1042    }
1043    crate::dataset::to_host_byte_order(&mut raw, datatype, es)?;
1044    let count = raw.len() / es;
1045    let mut result = Vec::<T>::with_capacity(count);
1046    // Safety: `T: H5Type` is a `Copy` POD primitive exactly `element_size()`
1047    // bytes wide, so the byte buffer is a valid array of `count` `T`s.
1048    unsafe {
1049        std::ptr::copy_nonoverlapping(raw.as_ptr(), result.as_mut_ptr() as *mut u8, raw.len());
1050        result.set_len(count);
1051    }
1052    Ok(result)
1053}
1054
1055/// The array attribute message for `values` of shape `dims`, whose product
1056/// must be the value count (an empty shape is a scalar, one value).
1057fn array_attr<T: H5Type>(name: &str, dims: &[u64], values: &[T]) -> Result<AttributeMessage> {
1058    let expected: u64 = dims.iter().product();
1059    if values.len() as u64 != expected {
1060        return Err(crate::error::Hdf5Error::InvalidState(format!(
1061            "attribute '{name}': {} values but shape {dims:?} needs {expected}",
1062            values.len()
1063        )));
1064    }
1065    Ok(AttributeMessage::array_numeric(
1066        name,
1067        T::hdf5_type(),
1068        dims,
1069        slice_to_bytes(values).to_vec(),
1070    ))
1071}
1072
1073/// Raw bytes of one `H5Type` scalar.
1074fn scalar_to_bytes<T: H5Type>(value: &T) -> Vec<u8> {
1075    let es = T::element_size();
1076    // Safety: `T: H5Type` is a `Copy` POD primitive exactly `element_size()`
1077    // bytes wide.
1078    unsafe { std::slice::from_raw_parts(value as *const T as *const u8, es) }.to_vec()
1079}
1080
1081/// Raw bytes of an `H5Type` slice, in element order.
1082fn slice_to_bytes<T: H5Type>(data: &[T]) -> &[u8] {
1083    // Safety: as `scalar_to_bytes`; the slice is a contiguous POD array.
1084    unsafe { std::slice::from_raw_parts(data.as_ptr() as *const u8, std::mem::size_of_val(data)) }
1085}