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}