Refactor nexus writer

Cargo.toml

@@ -44,6 +44,7 @@ rayon = "1.10.0"
 rdkafka = { version = "0.37.0", features = ["cmake-build", "ssl", "gssapi", "sasl"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1.0.140" }
+strum = { version = "0.26.3", features = ["derive"] }
 supermusr-common = { path = "./common" }
 supermusr-streaming-types = { path = "./streaming-types" }
 taos = { version = "0.10.27", default-features = false, features = ["ws"] }

nexus-writer/Cargo.toml

@@ -14,6 +14,7 @@ metrics.workspace = true
 metrics-exporter-prometheus.workspace = true
 ndarray.workspace = true
 rdkafka.workspace = true
+strum.workspace = true
 supermusr-common.workspace = true
 supermusr-streaming-types.workspace = true
 thiserror.workspace = true

nexus-writer/docs/README.md

@@ -0,0 +1,95 @@
+# nexus-writer source code description
+
+This file explains the nexus-writer source code file structure.
+
+## Top-Level Modules
+
+### main.rs
+
+This module operates in a similar way to other components, containing the `clap` Cli structure.
+
+### message_handlers.rs
+
+Consists of functions which bridge the main `main` module and `run_engine`.
+
+### flush_to_archive.rs
+
+Implements
+
+### error.rs
+
+### run_engine.rs
+
+Manages the run cache, accepts flatbuffer messages via the `message_handler` module and passes them to the appropriate `run` object.
+
+### hdf5_handlers.rs
+
+Extension Traits for `hdf5` objects.
+
+### nexus
+
+Defines general structs and traits used in the `nexus_structure` module.
+
+### nexus_structure
+
+This is where the nexus file structure is found, files and directories are set out in a way that roughly follows the group structure of the nexus format.
+
+## Lower-Level Modules
+
+### nexus.rs
+
+Defines the `NexusSchematic` trait which is implemented by every struct in the `nexus_structure` module.
+This trait standardises the creating and opening of the appropriate fields in nexus files.
+
+Defines the `NexusMessageHandler<M>` trait which takes a generic `M`. This trait is implemented by structs in `nexus_structure` module which process messages `M`.
+
+Defines the `NexusGroup<S>` struct which takes a generic `S : NexusSchematic`. This struct is a wrapper around the structs defined in the `nexus_structure` module.
+
+Note that for every `S : NexusSchematic` which implements `NexusMessageHandler<M>`, `NexusGroup<S>` automatically implements `NexusMessageHandler<M>` and passes `M` to `S`.
+
+#### file_interface.rs
+
+This defines a trait which abstracts the hdf5 file functionality. This interface is injected into the Run object. This allows the hdf5 file to be mocked in unit tests.
+
+#### logs.rs
+
+This defines the `LogMessage` and `AlarmMessage` traits which are implemented on appropriate types of log message (e.g. flatbuffer logs).
+This trait allows performs tasks such as getting the approprate `type_description` for the message, or writing its time/value to a given `Dataset`.
+
+#### classes.rs
+
+Implements the enum `NexusClass` which standardises the adding of attributes specifying the nexus classes of groups.
+
+#### units.rs
+
+Implements the enum `NexusUnits` which standardises the adding of attributes specifying physical units.
+
+### nexus_structure.rs
+
+Defines the `Root` struct which is the top level of the nexus format.
+
+#### entry.rs
+
+Defines the `Entry` struct which contains most of the fields, and implements `NexusSchematic` for it.
+
+##### event_data.rs
+
+Defines the `EventData` struct, and implements `NexusSchematic` for it.
+
+##### runlog.rs
+
+Defines the `RunLog` struct, and implements `NexusSchematic` for it.
+
+`RunLog` maintains a `HashMap` of `NexusGroup<Log>` keyed by `String`.
+
+##### selog.rs
+
+Defines the `SELog` struct, and implements `NexusSchematic` for it.
+
+`RunLog` maintains a `HashMap` of `NexusGroup<ValueLog>` keyed by `String`.
+
+#### logs
+
+Defines the `Log`, `ValueLog` and `AlarmLog` structs which are used in `runlog.rs` and `selog.rs`, and implements `NexusSchematic` for them.
+
+The struct `ValueLog` consists of `Option<Log>` and `Option<AlarmLog>` fields.

nexus-writer/src/nexus/error.rs → nexus-writer/src/error.rs

@@ -1,4 +1,4 @@
-use super::{hdf5_file::NexusHDF5Error, NexusDateTime};
+use crate::{hdf5_handlers::NexusHDF5Error, run_engine::NexusDateTime};
 use glob::{GlobError, PatternError};
 use std::{num::TryFromIntError, path::PathBuf};
 use supermusr_streaming_types::time_conversions::GpsTimeConversionError;
@@ -88,4 +88,10 @@ pub(crate) enum FlatBufferMissingError {
     RunName,
     #[error("Flatbuffer Instrument Name Missing")]
     InstrumentName,
+    #[error("Flatbuffer Alarm Source Name")]
+    AlarmName,
+    #[error("Flatbuffer Alarm Severity")]
+    AlarmSeverity,
+    #[error("Flatbuffer Alarm Message")]
+    AlarmMessage,
 }

nexus-writer/src/flush_to_archive.rs

@@ -1,6 +1,6 @@
 use crate::{
-    nexus::{ErrorCodeLocation, NexusWriterError},
-    NexusSettings, NexusWriterResult,
+    error::{ErrorCodeLocation, NexusWriterError, NexusWriterResult},
+    NexusSettings,
 };
 use std::path::{Path, PathBuf};
 use tokio::{

nexus-writer/src/hdf5_handlers/attribute.rs

@@ -0,0 +1,28 @@
+use super::{
+    error::{ConvertResult, NexusHDF5Result},
+    AttributeExt,
+};
+use crate::run_engine::NexusDateTime;
+use hdf5::{types::VarLenUnicode, Attribute};
+
+impl AttributeExt for Attribute {
+    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
+    fn get_datetime(&self) -> NexusHDF5Result<NexusDateTime> {
+        let string: VarLenUnicode = self.read_scalar().err_attribute(self)?;
+        string.parse().err_attribute(self)
+    }
+
+    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
+    fn set_string(&self, value: &str) -> NexusHDF5Result<()> {
+        self.write_scalar(&value.parse::<VarLenUnicode>().err_attribute(self)?)
+            .err_attribute(self)
+    }
+
+    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
+    fn get_string(&self) -> NexusHDF5Result<String> {
+        Ok(self
+            .read_scalar::<VarLenUnicode>()
+            .err_attribute(self)?
+            .to_string())
+    }
+}

nexus-writer/src/hdf5_handlers/dataset.rs

@@ -0,0 +1,78 @@
+use hdf5::{types::VarLenUnicode, Attribute, Dataset, H5Type};
+use ndarray::s;
+
+use crate::run_engine::NexusDateTime;
+
+use super::{
+    error::{ConvertResult, NexusHDF5Result},
+    DatasetExt, HasAttributesExt,
+};
+
+impl HasAttributesExt for Dataset {
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn add_attribute<T: H5Type>(&self, attr: &str) -> NexusHDF5Result<Attribute> {
+        let attr = self.new_attr::<T>().create(attr).err_dataset(self)?;
+        Ok(attr)
+    }
+
+    fn add_string_attribute(&self, attr: &str) -> NexusHDF5Result<Attribute> {
+        self.add_attribute::<VarLenUnicode>(attr)
+    }
+
+    fn add_constant_string_attribute(&self, attr: &str, value: &str) -> NexusHDF5Result<Attribute> {
+        let attr = self.add_string_attribute(attr)?;
+        attr.write_scalar(&value.parse::<VarLenUnicode>().err_dataset(self)?)
+            .err_dataset(self)?;
+        Ok(attr)
+    }
+
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn get_attribute(&self, attr: &str) -> NexusHDF5Result<Attribute> {
+        self.attr(attr).err_dataset(self)
+    }
+}
+
+impl DatasetExt for Dataset {
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn set_scalar<T: H5Type>(&self, value: &T) -> NexusHDF5Result<()> {
+        self.write_scalar(value).err_dataset(self)
+    }
+
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn set_string(&self, value: &str) -> NexusHDF5Result<()> {
+        self.write_scalar(&value.parse::<VarLenUnicode>().err_dataset(self)?)
+            .err_dataset(self)
+    }
+
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn get_string(&self) -> NexusHDF5Result<String> {
+        let string: VarLenUnicode = self.read_scalar().err_dataset(self)?;
+        Ok(string.into())
+    }
+
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn get_datetime(&self) -> NexusHDF5Result<NexusDateTime> {
+        let string: VarLenUnicode = self.read_scalar().err_dataset(self)?;
+        string.parse().err_dataset(self)
+    }
+
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn set_slice<T: H5Type>(&self, value: &[T]) -> NexusHDF5Result<()> {
+        self.resize(value.len()).err_dataset(self)?;
+        self.write_raw(value).err_dataset(self)
+    }
+
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn append_value<T: H5Type>(&self, value: T) -> NexusHDF5Result<()> {
+        self.append_slice(&[value]).err_dataset(self)
+    }
+
+    #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
+    fn append_slice<T: H5Type>(&self, value: &[T]) -> NexusHDF5Result<()> {
+        let cur_size = self.size();
+        let new_size = cur_size + value.len();
+        self.resize(new_size).err_dataset(self)?;
+        self.write_slice(value, s![cur_size..new_size])
+            .err_dataset(self)
+    }
+}