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}