diff --git a/Cargo.lock b/Cargo.lock
index 1017aada7..5f318db8f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1984,6 +1984,7 @@ dependencies = [
  "rdkafka",
  "supermusr-common",
  "supermusr-streaming-types",
+ "thiserror 2.0.11",
  "tokio",
  "tracing",
 ]
diff --git a/nexus-writer/Cargo.toml b/nexus-writer/Cargo.toml
index 775e99618..6431a57df 100644
--- a/nexus-writer/Cargo.toml
+++ b/nexus-writer/Cargo.toml
@@ -16,6 +16,7 @@ ndarray.workspace = true
 rdkafka.workspace = true
 supermusr-common.workspace = true
 supermusr-streaming-types.workspace = true
+thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 
diff --git a/nexus-writer/src/nexus/engine.rs b/nexus-writer/src/nexus/engine.rs
index 2bdde3c81..640931793 100644
--- a/nexus-writer/src/nexus/engine.rs
+++ b/nexus-writer/src/nexus/engine.rs
@@ -1,5 +1,8 @@
-use super::{Run, RunParameters};
-use chrono::{DateTime, Duration, Utc};
+use super::{
+    error::{ErrorCodeLocation, FlatBufferMissingError, NexusWriterError, NexusWriterResult},
+    NexusDateTime, Run, RunParameters,
+};
+use chrono::Duration;
 use glob::glob;
 #[cfg(test)]
 use std::collections::vec_deque;
@@ -43,10 +46,13 @@ impl NexusEngine {
         }
     }
 
-    pub(crate) fn resume_partial_runs(&mut self) -> anyhow::Result<()> {
+    pub(crate) fn resume_partial_runs(&mut self) -> NexusWriterResult<()> {
         if let Some(local_path) = &self.local_path {
             let local_path_str = local_path.as_os_str().to_str().ok_or_else(|| {
-                anyhow::anyhow!("Cannot convert local path to string: {0:?}", local_path)
+                NexusWriterError::CannotConvertPath {
+                    path: local_path.clone(),
+                    location: ErrorCodeLocation::ResumePartialRunsLocalDirectoryPath,
+                }
             })?;
 
             for filename in glob(&format!("{local_path_str}/*.nxs"))? {
@@ -55,15 +61,18 @@ impl NexusEngine {
                     filename
                         .file_stem()
                         .and_then(OsStr::to_str)
-                        .ok_or_else(|| {
-                            anyhow::anyhow!("Cannot convert filename to string: {0:?}", filename)
+                        .ok_or_else(|| NexusWriterError::CannotConvertPath {
+                            path: filename.clone(),
+                            location: ErrorCodeLocation::ResumePartialRunsFilePath,
                         })?;
                 let mut run = info_span!(
                     "Partial Run Found",
                     path = local_path_str,
                     file_name = filename_str
                 )
-                .in_scope(|| Run::resume_partial_run(local_path, filename_str))?;
+                .in_scope(|| {
+                    Run::resume_partial_run(local_path, filename_str, &self.nexus_settings)
+                })?;
                 if let Err(e) = run.span_init() {
                     warn!("Run span initiation failed {e}")
                 }
@@ -86,8 +95,8 @@ impl NexusEngine {
     pub(crate) fn sample_envionment(
         &mut self,
         data: se00_SampleEnvironmentData<'_>,
-    ) -> anyhow::Result<Option<&Run>> {
-        let timestamp = DateTime::<Utc>::from_timestamp_nanos(data.packet_timestamp());
+    ) -> NexusWriterResult<Option<&Run>> {
+        let timestamp = NexusDateTime::from_timestamp_nanos(data.packet_timestamp());
         if let Some(run) = self
             .run_cache
             .iter_mut()
@@ -102,8 +111,8 @@ impl NexusEngine {
     }
 
     #[tracing::instrument(skip_all)]
-    pub(crate) fn logdata(&mut self, data: &f144_LogData<'_>) -> anyhow::Result<Option<&Run>> {
-        let timestamp = DateTime::<Utc>::from_timestamp_nanos(data.timestamp());
+    pub(crate) fn logdata(&mut self, data: &f144_LogData<'_>) -> NexusWriterResult<Option<&Run>> {
+        let timestamp = NexusDateTime::from_timestamp_nanos(data.timestamp());
         if let Some(run) = self
             .run_cache
             .iter_mut()
@@ -118,14 +127,14 @@ impl NexusEngine {
     }
 
     #[tracing::instrument(skip_all)]
-    pub(crate) fn alarm(&mut self, data: Alarm<'_>) -> anyhow::Result<Option<&Run>> {
-        let timestamp = DateTime::<Utc>::from_timestamp_nanos(data.timestamp());
+    pub(crate) fn alarm(&mut self, data: Alarm<'_>) -> NexusWriterResult<Option<&Run>> {
+        let timestamp = NexusDateTime::from_timestamp_nanos(data.timestamp());
         if let Some(run) = self
             .run_cache
             .iter_mut()
             .find(|run| run.is_message_timestamp_valid(&timestamp))
         {
-            run.push_alarm_to_run(self.local_path.as_deref(), data)?;
+            run.push_alarm_to_run(self.local_path.as_deref(), data, &self.nexus_settings)?;
             Ok(Some(run))
         } else {
             warn!("No run found for alarm message with timestamp: {timestamp}");
@@ -134,7 +143,7 @@ impl NexusEngine {
     }
 
     #[tracing::instrument(skip_all)]
-    pub(crate) fn start_command(&mut self, data: RunStart<'_>) -> anyhow::Result<&mut Run> {
+    pub(crate) fn start_command(&mut self, data: RunStart<'_>) -> NexusWriterResult<&mut Run> {
         //  If a run is already in progress, and is missing a run-stop
         //  then call an abort run on the current run.
         if self.run_cache.back().is_some_and(|run| !run.has_run_stop()) {
@@ -161,7 +170,7 @@ impl NexusEngine {
             start_time = data.start_time(),
         )
     )]
-    fn abort_back_run(&mut self, data: &RunStart<'_>) -> anyhow::Result<()> {
+    fn abort_back_run(&mut self, data: &RunStart<'_>) -> NexusWriterResult<()> {
         self.run_cache
             .back_mut()
             .expect("run_cache::back_mut should exist")
@@ -174,13 +183,15 @@ impl NexusEngine {
     }
 
     #[tracing::instrument(skip_all)]
-    pub(crate) fn stop_command(&mut self, data: RunStop<'_>) -> anyhow::Result<&Run> {
+    pub(crate) fn stop_command(&mut self, data: RunStop<'_>) -> NexusWriterResult<&Run> {
         if let Some(last_run) = self.run_cache.back_mut() {
             last_run.set_stop_if_valid(self.local_path.as_deref(), data)?;
 
             Ok(last_run)
         } else {
-            Err(anyhow::anyhow!("Unexpected RunStop Command"))
+            Err(NexusWriterError::UnexpectedRunStop(
+                ErrorCodeLocation::StopCommand,
+            ))
         }
     }
 
@@ -197,19 +208,27 @@ impl NexusEngine {
     pub(crate) fn process_event_list(
         &mut self,
         message: &FrameAssembledEventListMessage<'_>,
-    ) -> anyhow::Result<Option<&Run>> {
-        let timestamp: DateTime<Utc> = (*message
-            .metadata()
-            .timestamp()
-            .ok_or(anyhow::anyhow!("Message timestamp missing."))?)
-        .try_into()?;
+    ) -> NexusWriterResult<Option<&Run>> {
+        let timestamp: NexusDateTime =
+            (*message
+                .metadata()
+                .timestamp()
+                .ok_or(NexusWriterError::FlatBufferMissing(
+                    FlatBufferMissingError::Timestamp,
+                    ErrorCodeLocation::ProcessEventList,
+                ))?)
+            .try_into()?;
 
         let run: Option<&Run> = if let Some(run) = self
             .run_cache
             .iter_mut()
             .find(|run| run.is_message_timestamp_valid(&timestamp))
         {
-            run.push_message(self.local_path.as_deref(), message, &self.nexus_settings)?;
+            run.push_frame_eventlist_message(
+                self.local_path.as_deref(),
+                message,
+                &self.nexus_settings,
+            )?;
             Some(run)
         } else {
             warn!("No run found for message with timestamp: {timestamp}");
@@ -288,7 +307,7 @@ mod test {
         fbb: &'b mut FlatBufferBuilder,
         name: &str,
         start_time: u64,
-    ) -> anyhow::Result<RunStart<'a>, InvalidFlatbuffer> {
+    ) -> Result<RunStart<'a>, InvalidFlatbuffer> {
         let args = RunStartArgs {
             start_time,
             run_name: Some(fbb.create_string(name)),
@@ -304,7 +323,7 @@ mod test {
         fbb: &'b mut FlatBufferBuilder,
         name: &str,
         stop_time: u64,
-    ) -> anyhow::Result<RunStop<'a>, InvalidFlatbuffer> {
+    ) -> Result<RunStop<'a>, InvalidFlatbuffer> {
         let args = RunStopArgs {
             stop_time,
             run_name: Some(fbb.create_string(name)),
@@ -329,7 +348,7 @@ mod test {
     fn create_frame_assembled_message<'a, 'b: 'a>(
         fbb: &'b mut FlatBufferBuilder,
         timestamp: &GpsTime,
-    ) -> anyhow::Result<FrameAssembledEventListMessage<'a>, InvalidFlatbuffer> {
+    ) -> Result<FrameAssembledEventListMessage<'a>, InvalidFlatbuffer> {
         let metadata = FrameMetadataV2::create(fbb, &create_metadata(timestamp));
         let args = FrameAssembledEventListMessageArgs {
             metadata: Some(metadata),
diff --git a/nexus-writer/src/nexus/error.rs b/nexus-writer/src/nexus/error.rs
new file mode 100644
index 000000000..c0e427d09
--- /dev/null
+++ b/nexus-writer/src/nexus/error.rs
@@ -0,0 +1,87 @@
+use super::{hdf5_file::NexusHDF5Error, NexusDateTime};
+use glob::{GlobError, PatternError};
+use std::{num::TryFromIntError, path::PathBuf};
+use supermusr_streaming_types::time_conversions::GpsTimeConversionError;
+use thiserror::Error;
+
+pub(crate) type NexusWriterResult<T> = Result<T, NexusWriterError>;
+
+#[derive(Debug, Error)]
+pub(crate) enum ErrorCodeLocation {
+    #[error("set_stop_if_valid")]
+    SetStopIfValid,
+    #[error("set_aborted_run")]
+    SetAbortedRun,
+    #[error("RunParameters::new")]
+    NewRunParamemters,
+    #[error("stop_command")]
+    StopCommand,
+    #[error("process_event_list")]
+    ProcessEventList,
+    #[error("resume_partial_runs local directory path")]
+    ResumePartialRunsLocalDirectoryPath,
+    #[error("resume_partial_runs file path")]
+    ResumePartialRunsFilePath,
+}
+
+#[derive(Debug, Error)]
+pub(crate) enum NexusWriterError {
+    #[error("{0}")]
+    HDF5(#[from] NexusHDF5Error),
+    #[error("Flatbuffer Timestamp Conversion Error {0}")]
+    FlatBufferTimestampConversion(#[from] GpsTimeConversionError),
+    #[error("{0}")]
+    FlatBufferMissing(FlatBufferMissingError, ErrorCodeLocation),
+    #[error("Unexpected RunStop Command")]
+    UnexpectedRunStop(ErrorCodeLocation),
+    #[error("Cannot convert local path to string: {path}")]
+    CannotConvertPath {
+        path: PathBuf,
+        location: ErrorCodeLocation,
+    },
+    #[error("Glob Pattern Error: {0}")]
+    GlobPattern(#[from] PatternError),
+    #[error("Glob Error: {0}")]
+    Glob(#[from] GlobError),
+    #[error("Integer Conversion Error")]
+    TryFromInt(#[from] TryFromIntError),
+    #[error("Start Time {int} Out of Range for DateTime at {location}")]
+    IntOutOfRangeForDateTime {
+        int: u64,
+        location: ErrorCodeLocation,
+    },
+    #[error("Stop Command before Start Command at {0}")]
+    StopCommandBeforeStartCommand(ErrorCodeLocation),
+    #[error("Stop Time {stop} earlier than current Start Time {start} at {location}")]
+    StopTimeEarlierThanStartTime {
+        start: NexusDateTime,
+        stop: NexusDateTime,
+        location: ErrorCodeLocation,
+    },
+    #[error("RunStop already set at {0}")]
+    RunStopAlreadySet(ErrorCodeLocation),
+}
+
+#[derive(Debug, Error)]
+pub(crate) enum FlatBufferInvalidDataTypeContext {
+    #[error("Run Log")]
+    RunLog,
+    #[error("Sample Environment Log")]
+    SELog,
+}
+
+#[derive(Debug, Error)]
+pub(crate) enum FlatBufferMissingError {
+    #[error("Flatbuffer Timestamp Missing")]
+    Timestamp,
+    #[error("Flatbuffer Channels Missing")]
+    Channels,
+    #[error("Flatbuffer Intensities Missing")]
+    Intensities,
+    #[error("Flatbuffer Times Missing")]
+    Times,
+    #[error("Flatbuffer Run Name Missing")]
+    RunName,
+    #[error("Flatbuffer Instrument Name Missing")]
+    InstrumentName,
+}
diff --git a/nexus-writer/src/nexus/hdf5_file/error.rs b/nexus-writer/src/nexus/hdf5_file/error.rs
new file mode 100644
index 000000000..d02716a4a
--- /dev/null
+++ b/nexus-writer/src/nexus/hdf5_file/error.rs
@@ -0,0 +1,266 @@
+use crate::nexus::error::{FlatBufferInvalidDataTypeContext, FlatBufferMissingError};
+use hdf5::{types::TypeDescriptor, Attribute, Dataset, Group};
+use std::error::Error;
+use supermusr_streaming_types::time_conversions::GpsTimeConversionError;
+use thiserror::Error;
+
+pub(crate) type NexusHDF5Result<T> = Result<T, NexusHDF5Error>;
+
+const NO_HDF5_PATH_SET: &str = "[No HDF5 Path Set]";
+
+#[derive(Debug, Error)]
+pub(crate) enum NexusHDF5Error {
+    #[error("{error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    HDF5 {
+        error: hdf5::Error,
+        hdf5_path: Option<String>,
+    },
+    #[error("{error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    DateTimeConversion {
+        error: chrono::ParseError,
+        hdf5_path: Option<String>,
+    },
+    #[error("{error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    HDF5String {
+        error: hdf5::types::StringError,
+        hdf5_path: Option<String>,
+    },
+    #[error("Flatbuffer Timestamp Conversion Error {error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    FlatBufferTimestampConversion {
+        error: GpsTimeConversionError,
+        hdf5_path: Option<String>,
+    },
+    #[error("Flatbuffer Timestamp Error Converting to Nanoseconds at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    FlatBufferTimestampConvertToNanoseconds { hdf5_path: Option<String> },
+    #[error("{error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    FlatBufferMissing {
+        error: FlatBufferMissingError,
+        hdf5_path: Option<String>,
+    },
+    #[error("Invalid FlatBuffer {context} Data Type {error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    FlatBufferInvalidDataType {
+        context: FlatBufferInvalidDataTypeContext,
+        error: String,
+        hdf5_path: Option<String>,
+    },
+    #[error("Inconsistent Numbers of Sample Environment Log Times and Values {0} != {1} at {2}", sizes.0, sizes.1, hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    FlatBufferInconsistentSELogTimeValueSizes {
+        sizes: (usize, usize),
+        hdf5_path: Option<String>,
+    },
+    #[error("Invalid HDF5 Type {error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    InvalidHDF5Type {
+        error: TypeDescriptor,
+        hdf5_path: Option<String>,
+    },
+    #[error("Invalid HDF5 Conversion {error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    InvalidHDF5TypeConversion {
+        error: TypeDescriptor,
+        hdf5_path: Option<String>,
+    },
+    #[error("IO Error {error} at {0}", hdf5_path.as_deref().unwrap_or(NO_HDF5_PATH_SET))]
+    IO {
+        error: std::io::Error,
+        hdf5_path: Option<String>,
+    },
+}
+
+impl NexusHDF5Error {
+    fn with_hdf5_path(self, path: String) -> Self {
+        match self {
+            Self::HDF5 {
+                error,
+                hdf5_path: _,
+            } => Self::HDF5 {
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::DateTimeConversion {
+                error,
+                hdf5_path: _,
+            } => Self::DateTimeConversion {
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::HDF5String {
+                error,
+                hdf5_path: _,
+            } => Self::HDF5String {
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::FlatBufferTimestampConversion {
+                error,
+                hdf5_path: _,
+            } => Self::FlatBufferTimestampConversion {
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::FlatBufferTimestampConvertToNanoseconds { hdf5_path: _ } => {
+                Self::FlatBufferTimestampConvertToNanoseconds {
+                    hdf5_path: Some(path),
+                }
+            }
+            Self::FlatBufferMissing {
+                error,
+                hdf5_path: _,
+            } => Self::FlatBufferMissing {
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::FlatBufferInvalidDataType {
+                context,
+                error,
+                hdf5_path: _,
+            } => Self::FlatBufferInvalidDataType {
+                context,
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::FlatBufferInconsistentSELogTimeValueSizes {
+                sizes,
+                hdf5_path: _,
+            } => Self::FlatBufferInconsistentSELogTimeValueSizes {
+                sizes,
+                hdf5_path: Some(path),
+            },
+            Self::InvalidHDF5Type {
+                error,
+                hdf5_path: _,
+            } => Self::InvalidHDF5Type {
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::InvalidHDF5TypeConversion {
+                error,
+                hdf5_path: _,
+            } => Self::InvalidHDF5TypeConversion {
+                error,
+                hdf5_path: Some(path),
+            },
+            Self::IO {
+                error,
+                hdf5_path: _,
+            } => Self::IO {
+                error,
+                hdf5_path: Some(path),
+            },
+        }
+    }
+
+    pub(crate) fn new_flatbuffer_missing(error: FlatBufferMissingError) -> Self {
+        Self::FlatBufferMissing {
+            error,
+            hdf5_path: None,
+        }
+    }
+
+    pub(crate) fn new_flatbuffer_timestamp_convert_to_nanoseconds() -> Self {
+        Self::FlatBufferTimestampConvertToNanoseconds { hdf5_path: None }
+    }
+
+    pub(crate) fn new_flatbuffer_invalid_data_type(
+        context: FlatBufferInvalidDataTypeContext,
+        error: String,
+    ) -> Self {
+        Self::FlatBufferInvalidDataType {
+            context,
+            error,
+            hdf5_path: None,
+        }
+    }
+
+    pub(crate) fn new_invalid_hdf5_type_conversion(error: TypeDescriptor) -> Self {
+        Self::InvalidHDF5TypeConversion {
+            error,
+            hdf5_path: None,
+        }
+    }
+}
+
+impl From<hdf5::Error> for NexusHDF5Error {
+    fn from(error: hdf5::Error) -> Self {
+        NexusHDF5Error::HDF5 {
+            error,
+            hdf5_path: None,
+        }
+    }
+}
+
+impl From<hdf5::types::StringError> for NexusHDF5Error {
+    fn from(error: hdf5::types::StringError) -> Self {
+        NexusHDF5Error::HDF5String {
+            error,
+            hdf5_path: None,
+        }
+    }
+}
+
+impl From<chrono::ParseError> for NexusHDF5Error {
+    fn from(error: chrono::ParseError) -> Self {
+        NexusHDF5Error::DateTimeConversion {
+            error,
+            hdf5_path: None,
+        }
+    }
+}
+
+impl From<GpsTimeConversionError> for NexusHDF5Error {
+    fn from(error: GpsTimeConversionError) -> Self {
+        NexusHDF5Error::FlatBufferTimestampConversion {
+            error,
+            hdf5_path: None,
+        }
+    }
+}
+
+impl From<FlatBufferMissingError> for NexusHDF5Error {
+    fn from(error: FlatBufferMissingError) -> Self {
+        NexusHDF5Error::FlatBufferMissing {
+            error,
+            hdf5_path: None,
+        }
+    }
+}
+
+impl From<std::io::Error> for NexusHDF5Error {
+    fn from(error: std::io::Error) -> Self {
+        NexusHDF5Error::IO {
+            error,
+            hdf5_path: None,
+        }
+    }
+}
+
+/// Used to allow errors which can be convertex to NexusHDF5Errors to be
+/// appended with hdf5 paths
+pub(crate) trait ConvertResult<T, E>
+where
+    E: Error + Into<NexusHDF5Error>,
+{
+    fn err_group(self, group: &Group) -> NexusHDF5Result<T>;
+    fn err_dataset(self, dataset: &Dataset) -> NexusHDF5Result<T>;
+    fn err_attribute(self, attribute: &Attribute) -> NexusHDF5Result<T>;
+    fn err_file(self) -> NexusHDF5Result<T>;
+}
+
+impl<T, E> ConvertResult<T, E> for Result<T, E>
+where
+    E: Error + Into<NexusHDF5Error>,
+{
+    fn err_group(self, group: &Group) -> NexusHDF5Result<T> {
+        self.map_err(|e| e.into().with_hdf5_path(group.name()))
+    }
+
+    fn err_dataset(self, dataset: &Dataset) -> NexusHDF5Result<T> {
+        self.map_err(|e| e.into().with_hdf5_path(dataset.name()))
+    }
+
+    fn err_attribute(self, attribute: &Attribute) -> NexusHDF5Result<T> {
+        self.map_err(|e| e.into().with_hdf5_path(attribute.name()))
+    }
+
+    fn err_file(self) -> NexusHDF5Result<T> {
+        self.map_err(|e| e.into().with_hdf5_path("File Level".to_owned()))
+    }
+}
diff --git a/nexus-writer/src/nexus/hdf5_file/hdf5_writer.rs b/nexus-writer/src/nexus/hdf5_file/hdf5_writer.rs
index 9df09a8f1..d69af01b1 100644
--- a/nexus-writer/src/nexus/hdf5_file/hdf5_writer.rs
+++ b/nexus-writer/src/nexus/hdf5_file/hdf5_writer.rs
@@ -1,80 +1,79 @@
+use super::{
+    error::{ConvertResult, NexusHDF5Result},
+    NexusHDF5Error,
+};
+use crate::nexus::NexusDateTime;
 use hdf5::{
-    types::{IntSize, TypeDescriptor, VarLenUnicode},
-    Dataset, Group, H5Type, Location, SimpleExtents,
+    types::{FloatSize, IntSize, TypeDescriptor, VarLenUnicode},
+    Attribute, Dataset, DatasetBuilderEmpty, Group, H5Type, SimpleExtents,
 };
+use ndarray::s;
 
-pub(super) fn add_new_group_to(parent: &Group, name: &str, class: &str) -> anyhow::Result<Group> {
-    let group = parent.create_group(name)?;
-    set_group_nx_class(&group, class)?;
-    Ok(group)
-}
-
-pub(super) fn add_attribute_to(parent: &Location, attr: &str, value: &str) -> anyhow::Result<()> {
-    parent
-        .new_attr::<VarLenUnicode>()
-        .create(attr)?
-        .write_scalar(&value.parse::<VarLenUnicode>()?)?;
-    Ok(())
+pub(crate) trait HasAttributesExt {
+    fn add_attribute_to(&self, attr: &str, value: &str) -> NexusHDF5Result<()>;
+    fn get_attribute(&self, attr: &str) -> NexusHDF5Result<Attribute>;
 }
 
-pub(super) fn set_group_nx_class(parent: &Group, class: &str) -> anyhow::Result<()> {
-    add_attribute_to(parent, "NX_class", class)
+pub(crate) trait GroupExt {
+    fn add_new_group_to(&self, name: &str, class: &str) -> NexusHDF5Result<Group>;
+    fn set_nx_class(&self, class: &str) -> NexusHDF5Result<()>;
+    fn create_resizable_empty_dataset<T: H5Type>(
+        &self,
+        name: &str,
+        chunk_size: usize,
+    ) -> NexusHDF5Result<Dataset>;
+    fn create_dynamic_resizable_empty_dataset(
+        &self,
+        name: &str,
+        type_descriptor: &TypeDescriptor,
+        chunk_size: usize,
+    ) -> NexusHDF5Result<Dataset>;
+    fn create_scalar_dataset<T: H5Type>(&self, name: &str) -> NexusHDF5Result<Dataset>;
+    fn get_dataset(&self, name: &str) -> NexusHDF5Result<Dataset>;
+    fn get_dataset_or_create_dynamic_resizable_empty_dataset(
+        &self,
+        name: &str,
+        type_descriptor: &TypeDescriptor,
+        chunk_size: usize,
+    ) -> NexusHDF5Result<Dataset>;
+    fn get_dataset_or_else<F>(&self, name: &str, f: F) -> NexusHDF5Result<Dataset>
+    where
+        F: Fn(&Group) -> NexusHDF5Result<Dataset>;
+    fn get_group(&self, name: &str) -> NexusHDF5Result<Group>;
+    fn get_group_or_create_new(&self, name: &str, class: &str) -> NexusHDF5Result<Group>;
 }
 
-pub(super) fn set_string_to(target: &Dataset, value: &str) -> anyhow::Result<()> {
-    Ok(target.write_scalar(&value.parse::<VarLenUnicode>()?)?)
+pub(crate) trait AttributeExt {
+    fn get_datetime_from(&self) -> NexusHDF5Result<NexusDateTime>;
 }
 
-pub(super) fn set_slice_to<T: H5Type>(target: &Dataset, value: &[T]) -> anyhow::Result<()> {
-    target.resize(value.len())?;
-    Ok(target.write_raw(value)?)
+impl AttributeExt for Attribute {
+    fn get_datetime_from(&self) -> NexusHDF5Result<NexusDateTime> {
+        let string: VarLenUnicode = self.read_scalar().err_attribute(self)?;
+        string.parse().err_attribute(self)
+    }
 }
 
-pub(super) fn create_resizable_dataset<T: H5Type>(
-    parent: &Group,
-    name: &str,
-    initial_size: usize,
-    chunk_size: usize,
-) -> anyhow::Result<Dataset> {
-    Ok(parent
-        .new_dataset::<T>()
-        .shape(SimpleExtents::resizable(vec![initial_size]))
-        .chunk(vec![chunk_size])
-        .create(name)?)
-}
+impl HasAttributesExt for Group {
+    fn add_attribute_to(&self, attr: &str, value: &str) -> NexusHDF5Result<()> {
+        self.new_attr::<VarLenUnicode>()
+            .create(attr)
+            .err_group(self)?
+            .write_scalar(&value.parse::<VarLenUnicode>().err_group(self)?)
+            .err_group(self)?;
+        Ok(())
+    }
 
-pub(super) fn _create_resizable_2d_dataset<T: H5Type>(
-    parent: &Group,
-    name: &str,
-    initial_size: (usize, usize),
-    chunk_size: (usize, usize),
-) -> anyhow::Result<Dataset> {
-    Ok(parent
-        .new_dataset::<T>()
-        .shape(SimpleExtents::resizable(vec![
-            initial_size.0,
-            initial_size.1,
-        ]))
-        .chunk(vec![chunk_size.0, chunk_size.1])
-        .create(name)?)
+    fn get_attribute(&self, attr: &str) -> NexusHDF5Result<Attribute> {
+        self.attr(attr).err_group(self)
+    }
 }
 
-pub(super) fn _create_resizable_2d_dataset_dyn_type(
+fn get_dataset_builder(
+    type_descriptor: &TypeDescriptor,
     parent: &Group,
-    name: &str,
-    hdf5_type: &TypeDescriptor,
-    initial_size: (usize, usize),
-    chunk_size: (usize, usize),
-) -> anyhow::Result<Dataset> {
-    let hdf5_type = {
-        if let TypeDescriptor::VarLenArray(t) = hdf5_type {
-            t
-        } else {
-            hdf5_type
-        }
-    };
-
-    let dataset = match hdf5_type {
+) -> Result<DatasetBuilderEmpty, NexusHDF5Error> {
+    Ok(match type_descriptor {
         TypeDescriptor::Integer(sz) => match sz {
             IntSize::U1 => parent.new_dataset::<i8>(),
             IntSize::U2 => parent.new_dataset::<i16>(),
@@ -88,16 +87,272 @@ pub(super) fn _create_resizable_2d_dataset_dyn_type(
             IntSize::U8 => parent.new_dataset::<u64>(),
         },
         TypeDescriptor::Float(sz) => match sz {
-            hdf5::types::FloatSize::U4 => parent.new_dataset::<f32>(),
-            hdf5::types::FloatSize::U8 => parent.new_dataset::<f64>(),
+            FloatSize::U4 => parent.new_dataset::<f32>(),
+            FloatSize::U8 => parent.new_dataset::<f64>(),
         },
-        _ => unreachable!(),
-    };
-    Ok(dataset
-        .shape(SimpleExtents::resizable(vec![
-            initial_size.0,
-            initial_size.1,
-        ]))
-        .chunk(vec![chunk_size.0, chunk_size.1])
-        .create(name)?)
+        TypeDescriptor::VarLenUnicode => parent.new_dataset::<VarLenUnicode>(),
+        _ => {
+            return Err(NexusHDF5Error::InvalidHDF5Type {
+                error: type_descriptor.clone(),
+                hdf5_path: None,
+            })
+        }
+    })
+}
+
+impl GroupExt for Group {
+    fn add_new_group_to(&self, name: &str, class: &str) -> NexusHDF5Result<Group> {
+        let group = self.create_group(name).err_group(self)?;
+        group.set_nx_class(class)?;
+        Ok(group)
+    }
+
+    fn set_nx_class(&self, class: &str) -> NexusHDF5Result<()> {
+        self.add_attribute_to("NX_class", class)
+    }
+
+    fn create_scalar_dataset<T: H5Type>(&self, name: &str) -> NexusHDF5Result<Dataset> {
+        self.new_dataset::<T>().create(name).err_group(self)
+    }
+
+    fn create_resizable_empty_dataset<T: H5Type>(
+        &self,
+        name: &str,
+        chunk_size: usize,
+    ) -> NexusHDF5Result<Dataset> {
+        self.new_dataset::<T>()
+            .shape(SimpleExtents::resizable(vec![0]))
+            .chunk(vec![chunk_size])
+            .create(name)
+            .err_group(self)
+    }
+
+    fn create_dynamic_resizable_empty_dataset(
+        &self,
+        name: &str,
+        type_descriptor: &TypeDescriptor,
+        chunk_size: usize,
+    ) -> NexusHDF5Result<Dataset> {
+        get_dataset_builder(type_descriptor, self)
+            .err_group(self)?
+            .shape(SimpleExtents::resizable(vec![0]))
+            .chunk(chunk_size)
+            .create(name)
+            .err_group(self)
+    }
+
+    fn get_dataset(&self, name: &str) -> NexusHDF5Result<Dataset> {
+        self.dataset(name).err_group(self)
+    }
+
+    fn get_dataset_or_else<F>(&self, name: &str, f: F) -> NexusHDF5Result<Dataset>
+    where
+        F: Fn(&Group) -> NexusHDF5Result<Dataset>,
+    {
+        self.dataset(name).or_else(|_| f(self))
+    }
+
+    fn get_dataset_or_create_dynamic_resizable_empty_dataset(
+        &self,
+        name: &str,
+        type_descriptor: &TypeDescriptor,
+        chunk_size: usize,
+    ) -> NexusHDF5Result<Dataset> {
+        self.dataset(name).or_else(|_| {
+            self.create_dynamic_resizable_empty_dataset(name, type_descriptor, chunk_size)
+        })
+    }
+
+    fn get_group(&self, name: &str) -> NexusHDF5Result<Group> {
+        self.group(name).err_group(self)
+    }
+
+    fn get_group_or_create_new(&self, name: &str, class: &str) -> NexusHDF5Result<Group> {
+        self.group(name)
+            .or_else(|_| self.add_new_group_to(name, class))
+    }
+}
+
+impl HasAttributesExt for Dataset {
+    fn add_attribute_to(&self, attr: &str, value: &str) -> NexusHDF5Result<()> {
+        self.new_attr::<VarLenUnicode>()
+            .create(attr)
+            .err_dataset(self)?
+            .write_scalar(&value.parse::<VarLenUnicode>().err_dataset(self)?)
+            .err_dataset(self)?;
+        Ok(())
+    }
+
+    fn get_attribute(&self, attr: &str) -> NexusHDF5Result<Attribute> {
+        self.attr(attr).err_dataset(self)
+    }
+}
+
+pub(crate) trait DatasetExt {
+    fn set_scalar_to<T: H5Type>(&self, value: &T) -> NexusHDF5Result<()>;
+    fn get_scalar_from<T: H5Type>(&self) -> NexusHDF5Result<T>;
+    fn set_string_to(&self, value: &str) -> NexusHDF5Result<()>;
+    fn get_string_from(&self) -> NexusHDF5Result<String>;
+    fn get_datetime_from(&self) -> NexusHDF5Result<NexusDateTime>;
+    fn set_slice_to<T: H5Type>(&self, value: &[T]) -> NexusHDF5Result<()>;
+    fn append_slice<T: H5Type>(&self, value: &[T]) -> NexusHDF5Result<()>;
+}
+
+impl DatasetExt for Dataset {
+    fn set_scalar_to<T: H5Type>(&self, value: &T) -> NexusHDF5Result<()> {
+        self.write_scalar(value).err_dataset(self)
+    }
+
+    fn get_scalar_from<T: H5Type>(&self) -> NexusHDF5Result<T> {
+        self.read_scalar().err_dataset(self)
+    }
+
+    fn set_string_to(&self, value: &str) -> NexusHDF5Result<()> {
+        self.write_scalar(&value.parse::<VarLenUnicode>().err_dataset(self)?)
+            .err_dataset(self)
+    }
+
+    fn get_string_from(&self) -> NexusHDF5Result<String> {
+        let string: VarLenUnicode = self.read_scalar().err_dataset(self)?;
+        Ok(string.into())
+    }
+
+    fn get_datetime_from(&self) -> NexusHDF5Result<NexusDateTime> {
+        let string: VarLenUnicode = self.read_scalar().err_dataset(self)?;
+        string.parse().err_dataset(self)
+    }
+
+    fn set_slice_to<T: H5Type>(&self, value: &[T]) -> NexusHDF5Result<()> {
+        self.resize(value.len()).err_dataset(self)?;
+        self.write_raw(value).err_dataset(self)
+    }
+
+    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)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{env::temp_dir, ops::Deref, path::PathBuf};
+
+    use super::*;
+
+    // Helper struct to create and tidy-up a temp hdf5 file
+    struct OneTempFile(Option<hdf5::File>, PathBuf);
+    // Suitably long temp file name, unlikely to clash with anything else
+    const TEMP_FILE_PREFIX: &str = "temp_supermusr_pipeline_nexus_writer_file";
+
+    impl OneTempFile {
+        //  We need a different file for each test, so they can run in parallel
+        fn new(test_name: &str) -> Self {
+            let mut path = temp_dir();
+            path.push(format!("{TEMP_FILE_PREFIX}_{test_name}.nxs"));
+            Self(Some(hdf5::File::create(&path).unwrap()), path)
+        }
+    }
+
+    //  Cleans up the temp directory after our test
+    impl Drop for OneTempFile {
+        fn drop(&mut self) {
+            let file = self.0.take().unwrap();
+            file.close().unwrap();
+            std::fs::remove_file(&self.1).unwrap();
+        }
+    }
+
+    //  So we can use our OneTempFile as an hdf5 file
+    impl Deref for OneTempFile {
+        type Target = hdf5::File;
+
+        fn deref(&self) -> &Self::Target {
+            self.0.as_ref().unwrap()
+        }
+    }
+
+    #[test]
+    fn create_group() {
+        let file = OneTempFile::new("create_group");
+        let maybe_group = file.get_group_or_create_new("my_group", "my_class");
+
+        assert!(maybe_group.is_ok());
+        assert_eq!(maybe_group.unwrap().name().as_str(), "/my_group");
+    }
+
+    #[test]
+    fn create_nested_group() {
+        let file = OneTempFile::new("create_nested_group");
+        let group = file
+            .get_group_or_create_new("my_group", "my_class")
+            .unwrap();
+        let maybe_subgroup = group.get_group_or_create_new("my_subgroup", "my_subclass");
+
+        assert!(maybe_subgroup.is_ok());
+        assert_eq!(
+            maybe_subgroup.unwrap().name().as_str(),
+            "/my_group/my_subgroup"
+        );
+    }
+
+    #[test]
+    fn create_dataset() {
+        let file = OneTempFile::new("create_dataset");
+        let maybe_dataset = file.get_dataset_or_else("my_dataset", |group| {
+            group.create_scalar_dataset::<u8>("my_dataset")
+        });
+
+        assert!(maybe_dataset.is_ok());
+        assert_eq!(maybe_dataset.unwrap().name().as_str(), "/my_dataset");
+    }
+
+    #[test]
+    fn open_nonexistant_group() {
+        let file = OneTempFile::new("open_nonexistant_group");
+        let maybe_group = file.get_group("non_existant_group");
+
+        assert!(maybe_group.is_err());
+
+        const EXPECTED_ERR_MSG : &str = "H5Gopen2(): unable to synchronously open group: object 'non_existant_group' doesn't exist at /";
+        assert_eq!(maybe_group.err().unwrap().to_string(), EXPECTED_ERR_MSG);
+    }
+
+    #[test]
+    fn open_nonexistant_dataset() {
+        let file = OneTempFile::new("open_nonexistant_dataset");
+        let maybe_dataset = file.get_dataset("non_existant_dataset");
+
+        assert!(maybe_dataset.is_err());
+
+        const EXPECTED_ERR_MSG : &str = "H5Dopen2(): unable to synchronously open dataset: object 'non_existant_dataset' doesn't exist at /";
+        assert_eq!(maybe_dataset.err().unwrap().to_string(), EXPECTED_ERR_MSG);
+    }
+
+    #[test]
+    fn open_nonexistant_nested_dataset() {
+        let file = OneTempFile::new("open_nonexistant_nested_dataset");
+        let group = file
+            .get_group_or_create_new("my_group", "my_class")
+            .unwrap();
+        let maybe_subgroup = group.get_dataset("my_subgroup");
+
+        assert!(maybe_subgroup.is_err());
+
+        const EXPECTED_ERR_MSG : &str = "H5Dopen2(): unable to synchronously open dataset: object 'my_subgroup' doesn't exist at /my_group";
+        assert_eq!(maybe_subgroup.err().unwrap().to_string(), EXPECTED_ERR_MSG);
+    }
+
+    #[test]
+    fn open_nonexistant_attribute() {
+        let file = OneTempFile::new("open_nonexistant_attribute");
+        let maybe_dataset = file.get_attribute("non_existant_attribute");
+
+        assert!(maybe_dataset.is_err());
+
+        const EXPECTED_ERR_MSG : &str = "H5Aopen(): unable to synchronously open attribute: can't locate attribute: 'non_existant_attribute' at /";
+        assert_eq!(maybe_dataset.err().unwrap().to_string(), EXPECTED_ERR_MSG);
+    }
 }
diff --git a/nexus-writer/src/nexus/hdf5_file/mod.rs b/nexus-writer/src/nexus/hdf5_file/mod.rs
index a00165ca3..35321af61 100644
--- a/nexus-writer/src/nexus/hdf5_file/mod.rs
+++ b/nexus-writer/src/nexus/hdf5_file/mod.rs
@@ -1,10 +1,8 @@
+mod error;
 mod hdf5_writer;
 mod run_file;
 mod run_file_components;
 
-use hdf5_writer::{
-    add_attribute_to, add_new_group_to, create_resizable_dataset, set_group_nx_class, set_slice_to,
-    set_string_to,
-};
+pub(crate) use error::NexusHDF5Error;
 pub(crate) use run_file::RunFile;
 use run_file_components::EventRun;
diff --git a/nexus-writer/src/nexus/hdf5_file/run_file.rs b/nexus-writer/src/nexus/hdf5_file/run_file.rs
index 2991ff2d6..9bffe9fb5 100644
--- a/nexus-writer/src/nexus/hdf5_file/run_file.rs
+++ b/nexus-writer/src/nexus/hdf5_file/run_file.rs
@@ -1,15 +1,16 @@
 use super::{
-    add_attribute_to, add_new_group_to, create_resizable_dataset, set_group_nx_class, set_slice_to,
-    set_string_to, EventRun,
+    error::{ConvertResult, NexusHDF5Result},
+    hdf5_writer::{DatasetExt, GroupExt, HasAttributesExt},
+    EventRun,
 };
 use crate::nexus::{
     hdf5_file::run_file_components::{RunLog, SeLog},
     nexus_class as NX,
     run_parameters::RunStopParameters,
-    NexusConfiguration, NexusSettings, RunParameters, DATETIME_FORMAT,
+    NexusConfiguration, NexusDateTime, NexusSettings, RunParameters, DATETIME_FORMAT,
 };
-use chrono::{DateTime, Utc};
-use hdf5::{types::VarLenUnicode, Dataset, File, H5Type};
+use chrono::Utc;
+use hdf5::{types::VarLenUnicode, Dataset, File};
 use std::{fs::create_dir_all, path::Path};
 use supermusr_streaming_types::{
     aev2_frame_assembled_event_v2_generated::FrameAssembledEventListMessage,
@@ -58,55 +59,48 @@ impl RunFileContents {
     pub(crate) fn populate_new_runfile(
         file: &File,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<Self> {
-        set_group_nx_class(file, NX::ROOT)?;
+    ) -> NexusHDF5Result<Self> {
+        file.set_nx_class(NX::ROOT)?;
 
-        add_attribute_to(file, "HDF5_version", "1.14.3")?; // Can this be taken directly from the nix package;
-        add_attribute_to(file, "NeXus_version", "")?; // Where does this come from?
-        add_attribute_to(file, "file_name", &file.filename())?; //  This should be absolutized at some point
-        add_attribute_to(file, "file_time", Utc::now().to_string().as_str())?; //  This should be formatted, the nanoseconds are overkill.
+        file.add_attribute_to("HDF5_version", "1.14.3")?; // Can this be taken directly from the nix package;
+        file.add_attribute_to("NeXus_version", "")?; // Where does this come from?
+        file.add_attribute_to("file_name", &file.filename())?; //  This should be absolutized at some point
+        file.add_attribute_to("file_time", Utc::now().to_string().as_str())?; //  This should be formatted, the nanoseconds are overkill.
 
-        let entry = add_new_group_to(file, "raw_data_1", NX::ENTRY)?;
+        let entry = file.add_new_group_to("raw_data_1", NX::ENTRY)?;
 
-        let idf_version = entry.new_dataset::<u32>().create("IDF_version")?;
-        let definition = entry.new_dataset::<VarLenUnicode>().create("definition")?;
-        let program_name = entry
-            .new_dataset::<VarLenUnicode>()
-            .create("program_name")?;
+        let idf_version = entry.create_scalar_dataset::<u32>("IDF_version")?;
+        let definition = entry.create_scalar_dataset::<VarLenUnicode>("definition")?;
+        let program_name = entry.create_scalar_dataset::<VarLenUnicode>("program_name")?;
 
-        let run_number = entry.new_dataset::<u32>().create("run_number")?;
-        let experiment_identifier = entry
-            .new_dataset::<VarLenUnicode>()
-            .create("experiment_identifier")?;
+        let run_number = entry.create_scalar_dataset::<u32>("run_number")?;
+        let experiment_identifier =
+            entry.create_scalar_dataset::<VarLenUnicode>("experiment_identifier")?;
 
-        let start_time = entry.new_dataset::<VarLenUnicode>().create("start_time")?;
-        let end_time = entry.new_dataset::<VarLenUnicode>().create("end_time")?;
+        let start_time = entry.create_scalar_dataset::<VarLenUnicode>("start_time")?;
+        let end_time = entry.create_scalar_dataset::<VarLenUnicode>("end_time")?;
 
-        let name = entry.new_dataset::<VarLenUnicode>().create("name")?;
-        let title = entry.new_dataset::<VarLenUnicode>().create("title")?;
+        let name = entry.create_scalar_dataset::<VarLenUnicode>("name")?;
+        let title = entry.create_scalar_dataset::<VarLenUnicode>("title")?;
 
-        let instrument = add_new_group_to(&entry, "instrument", NX::INSTRUMENT)?;
-        let instrument_name = instrument.new_dataset::<VarLenUnicode>().create("name")?;
+        let instrument = entry.add_new_group_to("instrument", NX::INSTRUMENT)?;
+        let instrument_name = instrument.create_scalar_dataset::<VarLenUnicode>("name")?;
 
         let logs = RunLog::new_runlog(&entry)?;
 
-        let periods = add_new_group_to(&entry, "periods", NX::PERIOD)?;
-        let period_number = periods.new_dataset::<u32>().create("number")?;
-        let period_type = create_resizable_dataset::<u32>(
-            &periods,
-            "type",
-            0,
-            nexus_settings.periodlist_chunk_size,
-        )?;
+        let periods = entry.add_new_group_to("periods", NX::PERIOD)?;
+        let period_number = periods.create_scalar_dataset::<u32>("number")?;
+        let period_type = periods
+            .create_resizable_empty_dataset::<u32>("type", nexus_settings.periodlist_chunk_size)?;
 
         let selogs = SeLog::new_selog(&entry)?;
 
-        let source = add_new_group_to(&instrument, "source", NX::SOURCE)?;
-        let source_name = source.new_dataset::<VarLenUnicode>().create("name")?;
-        let source_type = source.new_dataset::<VarLenUnicode>().create("type")?;
-        let source_probe = source.new_dataset::<VarLenUnicode>().create("probe")?;
+        let source = instrument.add_new_group_to("source", NX::SOURCE)?;
+        let source_name = source.create_scalar_dataset::<VarLenUnicode>("name")?;
+        let source_type = source.create_scalar_dataset::<VarLenUnicode>("type")?;
+        let source_probe = source.create_scalar_dataset::<VarLenUnicode>("probe")?;
 
-        let _detector = add_new_group_to(&instrument, "detector", NX::DETECTOR)?;
+        let _detector = instrument.add_new_group_to("detector", NX::DETECTOR)?;
 
         let lists = EventRun::new_event_runfile(&entry, nexus_settings)?;
 
@@ -132,38 +126,38 @@ impl RunFileContents {
         })
     }
 
-    fn populate_open_runfile(file: &File) -> anyhow::Result<Self> {
-        let entry = file.group("raw_data_1")?;
+    fn populate_open_runfile(file: &File) -> NexusHDF5Result<Self> {
+        let entry = file.get_group("raw_data_1")?;
 
-        let idf_version = entry.dataset("IDF_version")?;
-        let definition = entry.dataset("definition")?;
-        let run_number = entry.dataset("run_number")?;
-        let program_name = entry.dataset("program_name")?;
-        let experiment_identifier = entry.dataset("experiment_identifier")?;
+        let idf_version = entry.get_dataset("IDF_version")?;
+        let definition = entry.get_dataset("definition")?;
+        let run_number = entry.get_dataset("run_number")?;
+        let program_name = entry.get_dataset("program_name")?;
+        let experiment_identifier = entry.get_dataset("experiment_identifier")?;
 
-        let start_time = entry.dataset("start_time")?;
-        let end_time = entry.dataset("end_time")?;
+        let start_time = entry.get_dataset("start_time")?;
+        let end_time = entry.get_dataset("end_time")?;
 
-        let name = entry.dataset("name")?;
-        let title = entry.dataset("title")?;
+        let name = entry.get_dataset("name")?;
+        let title = entry.get_dataset("title")?;
 
-        let periods = entry.group("periods")?;
-        let period_number = periods.dataset("number")?;
-        let period_type = periods.dataset("type")?;
+        let periods = entry.get_group("periods")?;
+        let period_number = periods.get_dataset("number")?;
+        let period_type = periods.get_dataset("type")?;
 
         let selogs = SeLog::open_selog(&entry)?;
 
-        let instrument = entry.group("instrument")?;
-        let instrument_name = instrument.dataset("name")?;
+        let instrument = entry.get_group("instrument")?;
+        let instrument_name = instrument.get_dataset("name")?;
 
         let logs = RunLog::open_runlog(&entry)?;
 
-        let source = instrument.group("source")?;
-        let source_name = source.dataset("name")?;
-        let source_type = source.dataset("type")?;
-        let source_probe = source.dataset("probe")?;
+        let source = instrument.get_group("source")?;
+        let source_name = source.get_dataset("name")?;
+        let source_type = source.get_dataset("type")?;
+        let source_probe = source.get_dataset("probe")?;
 
-        let _detector = instrument.group("detector")?;
+        let _detector = instrument.get_group("detector")?;
 
         let lists = EventRun::open_event_runfile(&entry)?;
 
@@ -196,31 +190,31 @@ impl RunFile {
         path: &Path,
         run_name: &str,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<Self> {
-        create_dir_all(path)?;
+    ) -> NexusHDF5Result<Self> {
+        create_dir_all(path).err_file()?;
         let filename = RunParameters::get_hdf5_filename(path, run_name);
         debug!("File save begin. File: {0}.", filename.display());
 
-        let file = File::create(filename)?;
+        let file = File::create(filename).err_file()?;
         match RunFileContents::populate_new_runfile(&file, nexus_settings) {
             Ok(contents) => Ok(Self { file, contents }),
             Err(e) => {
-                file.close()?;
+                file.close().err_file()?;
                 Err(e)
             }
         }
     }
 
     #[tracing::instrument(skip_all, err(level = "warn"))]
-    pub(crate) fn open_runfile(local_path: &Path, run_name: &str) -> anyhow::Result<Self> {
+    pub(crate) fn open_runfile(local_path: &Path, run_name: &str) -> NexusHDF5Result<Self> {
         let filename = RunParameters::get_hdf5_filename(local_path, run_name);
         debug!("File open begin. File: {0}.", filename.display());
 
-        let file = File::open_rw(filename)?;
+        let file = File::open_rw(filename).err_file()?;
         match RunFileContents::populate_open_runfile(&file) {
             Ok(contents) => Ok(Self { file, contents }),
             Err(e) => {
-                file.close()?;
+                file.close().err_file()?;
                 Err(e)
             }
         }
@@ -231,57 +225,57 @@ impl RunFile {
         &mut self,
         parameters: &RunParameters,
         nexus_configuration: &NexusConfiguration,
-    ) -> anyhow::Result<()> {
-        self.contents.idf_version.write_scalar(&2)?;
+    ) -> NexusHDF5Result<()> {
+        self.contents.idf_version.set_scalar_to(&2)?;
         self.contents
             .run_number
-            .write_scalar(&parameters.run_number)?;
-
-        set_string_to(&self.contents.definition, "muonTD")?;
-        set_string_to(&self.contents.experiment_identifier, "")?;
-
-        set_string_to(
-            &self.contents.program_name,
-            "SuperMuSR Data Pipeline Nexus Writer",
-        )?;
-        add_attribute_to(&self.contents.program_name, "version", "1.0")?;
-        add_attribute_to(
-            &self.contents.program_name,
-            "configuration",
-            &nexus_configuration.configuration,
-        )?;
+            .set_scalar_to(&parameters.run_number)?;
+
+        self.contents.definition.set_string_to("muonTD")?;
+        self.contents.experiment_identifier.set_string_to("")?;
+
+        self.contents
+            .program_name
+            .set_string_to("SuperMuSR Data Pipeline Nexus Writer")?;
+        self.contents
+            .program_name
+            .add_attribute_to("version", "1.0")?;
+        self.contents
+            .program_name
+            .add_attribute_to("configuration", &nexus_configuration.configuration)?;
 
         let start_time = parameters.collect_from.format(DATETIME_FORMAT).to_string();
 
-        set_string_to(&self.contents.start_time, &start_time)?;
-        set_string_to(&self.contents.end_time, "")?;
+        self.contents.start_time.set_string_to(&start_time)?;
+        self.contents.end_time.set_string_to("")?;
 
-        set_string_to(&self.contents.name, &parameters.run_name)?;
-        set_string_to(&self.contents.title, "")?;
+        self.contents.name.set_string_to(&parameters.run_name)?;
+        self.contents.title.set_string_to("")?;
 
-        set_string_to(&self.contents.instrument_name, &parameters.instrument_name)?;
+        self.contents
+            .instrument_name
+            .set_string_to(&parameters.instrument_name)?;
 
         self.contents
             .period_number
-            .write_scalar(&parameters.num_periods)?;
-        set_slice_to(
-            &self.contents.period_type,
-            &vec![1; parameters.num_periods as usize],
-        )?;
+            .set_scalar_to(&parameters.num_periods)?;
+        self.contents
+            .period_type
+            .set_slice_to(&vec![1; parameters.num_periods as usize])?;
 
-        set_string_to(&self.contents.source_name, "MuSR")?;
-        set_string_to(&self.contents.source_type, "")?;
-        set_string_to(&self.contents.source_probe, "")?;
+        self.contents.source_name.set_string_to("MuSR")?;
+        self.contents.source_type.set_string_to("")?;
+        self.contents.source_probe.set_string_to("")?;
 
         self.contents.lists.init(&parameters.collect_from)?;
         Ok(())
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn set_end_time(&mut self, end_time: &DateTime<Utc>) -> anyhow::Result<()> {
+    pub(crate) fn set_end_time(&mut self, end_time: &NexusDateTime) -> NexusHDF5Result<()> {
         let end_time = end_time.format(DATETIME_FORMAT).to_string();
 
-        set_string_to(&self.contents.end_time, &end_time)?;
+        self.contents.end_time.set_string_to(&end_time)?;
         Ok(())
     }
 
@@ -289,81 +283,65 @@ impl RunFile {
     pub(crate) fn push_logdata_to_runfile(
         &mut self,
         logdata: &f144_LogData,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         self.contents
             .logs
-            .push_logdata_to_runlog(logdata, nexus_settings)
+            .push_logdata_to_runlog(logdata, origin_time, nexus_settings)
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn push_alarm_to_runfile(&mut self, alarm: Alarm) -> anyhow::Result<()> {
-        self.contents.selogs.push_alarm_to_selog(alarm)
+    pub(crate) fn push_alarm_to_runfile(
+        &mut self,
+        alarm: Alarm,
+        origin_time: &NexusDateTime,
+        nexus_settings: &NexusSettings,
+    ) -> NexusHDF5Result<()> {
+        self.contents
+            .selogs
+            .push_alarm_to_selog(alarm, origin_time, nexus_settings)
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
     pub(crate) fn push_selogdata(
         &mut self,
         selogdata: se00_SampleEnvironmentData,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         self.contents
             .selogs
-            .push_selogdata_to_selog(&selogdata, nexus_settings)
+            .push_selogdata_to_selog(&selogdata, origin_time, nexus_settings)
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn push_message_to_runfile(
+    pub(crate) fn push_frame_eventlist_message_to_runfile(
         &mut self,
         message: &FrameAssembledEventListMessage,
-        nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
-        self.contents.lists.push_message_to_event_runfile(message)?;
-
-        if !message.complete() {
-            let time_zero = self.contents.lists.get_time_zero(message)?;
-
-            self.contents.logs.push_incomplete_frame_log(
-                time_zero,
-                message
-                    .digitizers_present()
-                    .unwrap_or_default()
-                    .iter()
-                    .collect(),
-                nexus_settings,
-            )?;
-        }
-        Ok(())
-    }
-
-    fn try_read_scalar<T: H5Type>(dataset: &Dataset) -> anyhow::Result<T> {
-        if dataset.storage_size() != 0 {
-            if dataset.is_scalar() {
-                Ok(dataset.read_scalar::<T>()?)
-            } else {
-                anyhow::bail!("{} is not a scalar", dataset.name())
-            }
-        } else {
-            anyhow::bail!("{} is not allocated", dataset.name())
-        }
+    ) -> NexusHDF5Result<()> {
+        self.contents
+            .lists
+            .push_frame_eventlist_message_to_runfile(message)
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn extract_run_parameters(&self) -> anyhow::Result<RunParameters> {
-        let collect_from: DateTime<Utc> =
-            Self::try_read_scalar::<VarLenUnicode>(&self.contents.start_time)?.parse()?;
-        let run_name = Self::try_read_scalar::<VarLenUnicode>(&self.contents.name)?.into();
-        let run_number = Self::try_read_scalar::<u32>(&self.contents.run_number)?;
-        let num_periods = Self::try_read_scalar::<u32>(&self.contents.period_number)?;
-        let instrument_name =
-            Self::try_read_scalar::<VarLenUnicode>(&self.contents.instrument_name)?.into();
-        let run_stop_parameters = Self::try_read_scalar::<VarLenUnicode>(&self.contents.end_time)?
-            .parse()
+    pub(crate) fn extract_run_parameters(&self) -> NexusHDF5Result<RunParameters> {
+        let collect_from = self.contents.start_time.get_datetime_from()?;
+        let run_name = self.contents.name.get_string_from()?;
+        let run_number = self.contents.run_number.get_scalar_from()?;
+        let num_periods = self.contents.period_number.get_scalar_from()?;
+        let instrument_name = self.contents.instrument_name.get_string_from()?;
+        let run_stop_parameters = self
+            .contents
+            .end_time
+            .get_datetime_from()
             .map(|collect_until| RunStopParameters {
                 collect_until,
                 last_modified: Utc::now(),
             })
             .ok();
+
         Ok(RunParameters {
             collect_from,
             run_stop_parameters,
@@ -375,20 +353,59 @@ impl RunFile {
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn set_aborted_run_warning(
+    pub(crate) fn push_incomplete_frame_warning(
+        &mut self,
+        message: &FrameAssembledEventListMessage,
+        nexus_settings: &NexusSettings,
+    ) -> NexusHDF5Result<()> {
+        let time_zero = self.contents.lists.get_time_zero(message).err_file()?;
+        let origin = self
+            .contents
+            .lists
+            .get_offset()
+            .expect("This should never fail.");
+
+        self.contents.logs.push_incomplete_frame_log(
+            time_zero,
+            message
+                .digitizers_present()
+                .unwrap_or_default()
+                .iter()
+                .collect(),
+            origin,
+            nexus_settings,
+        )
+    }
+
+    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
+    pub(crate) fn push_run_resumed_warning(
+        &mut self,
+        current_time: &NexusDateTime,
+        origin_time: &NexusDateTime,
+        nexus_settings: &NexusSettings,
+    ) -> NexusHDF5Result<()> {
+        self.contents
+            .logs
+            .push_run_resumed_warning(current_time, origin_time, nexus_settings)?;
+        Ok(())
+    }
+
+    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
+    pub(crate) fn push_aborted_run_warning(
         &mut self,
-        stop_time: i32,
+        stop_time_ms: i64,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         self.contents
             .logs
-            .set_aborted_run_warning(stop_time, nexus_settings)?;
+            .push_aborted_run_warning(stop_time_ms, origin_time, nexus_settings)?;
         Ok(())
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn close(self) -> anyhow::Result<()> {
-        self.file.close()?;
+    pub(crate) fn close(self) -> NexusHDF5Result<()> {
+        self.file.close().err_file()?;
         Ok(())
     }
 }
diff --git a/nexus-writer/src/nexus/hdf5_file/run_file_components/event_run_file.rs b/nexus-writer/src/nexus/hdf5_file/run_file_components/event_run_file.rs
index dedf4e8e3..3633bf77e 100644
--- a/nexus-writer/src/nexus/hdf5_file/run_file_components/event_run_file.rs
+++ b/nexus-writer/src/nexus/hdf5_file/run_file_components/event_run_file.rs
@@ -1,16 +1,19 @@
 use crate::nexus::{
-    hdf5_file::{add_attribute_to, add_new_group_to, create_resizable_dataset},
-    nexus_class as NX, NexusSettings,
+    error::FlatBufferMissingError,
+    hdf5_file::{
+        error::{ConvertResult, NexusHDF5Error, NexusHDF5Result},
+        hdf5_writer::{AttributeExt, DatasetExt, GroupExt, HasAttributesExt},
+    },
+    nexus_class as NX, NexusDateTime, NexusSettings,
 };
-use chrono::{DateTime, Utc};
-use hdf5::{types::VarLenUnicode, Dataset, Group};
-use ndarray::s;
+use hdf5::{Dataset, Group};
 use supermusr_common::{Channel, Time};
 use supermusr_streaming_types::aev2_frame_assembled_event_v2_generated::FrameAssembledEventListMessage;
 
 #[derive(Debug)]
 pub(crate) struct EventRun {
-    offset: Option<DateTime<Utc>>,
+    parent: Group,
+    offset: Option<NexusDateTime>,
 
     num_messages: usize,
     num_events: usize,
@@ -34,78 +37,59 @@ impl EventRun {
     pub(crate) fn new_event_runfile(
         parent: &Group,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<Self> {
-        let detector = add_new_group_to(parent, "detector_1", NX::EVENT_DATA)?;
+    ) -> NexusHDF5Result<Self> {
+        let detector = parent.add_new_group_to("detector_1", NX::EVENT_DATA)?;
 
-        let pulse_height = create_resizable_dataset::<f64>(
-            &detector,
+        let pulse_height = detector.create_resizable_empty_dataset::<f64>(
             "pulse_height",
-            0,
             nexus_settings.eventlist_chunk_size,
         )?;
-        let event_id = create_resizable_dataset::<Channel>(
-            &detector,
+        let event_id = detector.create_resizable_empty_dataset::<Channel>(
             "event_id",
-            0,
             nexus_settings.eventlist_chunk_size,
         )?;
-        let event_time_offset = create_resizable_dataset::<Time>(
-            &detector,
+        let event_time_offset = detector.create_resizable_empty_dataset::<Time>(
             "event_time_offset",
-            0,
             nexus_settings.eventlist_chunk_size,
         )?;
-        add_attribute_to(&event_time_offset, "units", "ns")?;
+        event_time_offset.add_attribute_to("units", "ns")?;
 
-        let event_index = create_resizable_dataset::<u32>(
-            &detector,
+        let event_index = detector.create_resizable_empty_dataset::<u32>(
             "event_index",
-            0,
             nexus_settings.framelist_chunk_size,
         )?;
-        let event_time_zero = create_resizable_dataset::<u64>(
-            &detector,
+        let event_time_zero = detector.create_resizable_empty_dataset::<u64>(
             "event_time_zero",
-            0,
             nexus_settings.framelist_chunk_size,
         )?;
-        let period_number = create_resizable_dataset::<u64>(
-            &detector,
+        let period_number = detector.create_resizable_empty_dataset::<u64>(
             "period_number",
-            0,
             nexus_settings.framelist_chunk_size,
         )?;
-        add_attribute_to(&event_time_zero, "units", "ns")?;
+        event_time_zero.add_attribute_to("units", "ns")?;
 
-        let frame_number = create_resizable_dataset::<u64>(
-            &detector,
+        let frame_number = detector.create_resizable_empty_dataset::<u64>(
             "frame_number",
-            0,
             nexus_settings.framelist_chunk_size,
         )?;
 
-        let frame_complete = create_resizable_dataset::<u64>(
-            &detector,
+        let frame_complete = detector.create_resizable_empty_dataset::<u64>(
             "is_frame_complete",
-            0,
             nexus_settings.framelist_chunk_size,
         )?;
 
-        let running = create_resizable_dataset::<bool>(
-            &detector,
+        let running = detector.create_resizable_empty_dataset::<bool>(
             "running",
-            0,
             nexus_settings.framelist_chunk_size,
         )?;
 
-        let veto_flags = create_resizable_dataset::<u16>(
-            &detector,
+        let veto_flags = detector.create_resizable_empty_dataset::<u16>(
             "veto_flag",
-            0,
             nexus_settings.framelist_chunk_size,
         )?;
 
         Ok(Self {
+            parent: detector,
             offset: None,
             num_events: 0,
             num_messages: 0,
@@ -123,31 +107,31 @@ impl EventRun {
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn open_event_runfile(parent: &Group) -> anyhow::Result<Self> {
-        let detector = parent.group("detector_1")?;
-
-        let pulse_height = detector.dataset("pulse_height")?;
-        let event_id = detector.dataset("event_id")?;
-        let event_time_offset = detector.dataset("event_time_offset")?;
-
-        let event_index = detector.dataset("event_index")?;
-        let event_time_zero = detector.dataset("event_time_zero")?;
-        let period_number = detector.dataset("period_number")?;
-        let frame_number = detector.dataset("frame_number")?;
-        let frame_complete = detector.dataset("is_frame_complete")?;
-        let running = detector.dataset("running")?;
-        let veto_flags = detector.dataset("veto_flag")?;
-
-        let offset: Option<DateTime<Utc>> = {
-            if let Ok(offset) = event_time_zero.attr("offset") {
-                let offset: VarLenUnicode = offset.read_scalar()?;
-                Some(offset.parse()?)
+    pub(crate) fn open_event_runfile(parent: &Group) -> NexusHDF5Result<Self> {
+        let detector = parent.get_group("detector_1")?;
+
+        let pulse_height = detector.get_dataset("pulse_height")?;
+        let event_id = detector.get_dataset("event_id")?;
+        let event_time_offset = detector.get_dataset("event_time_offset")?;
+
+        let event_index = detector.get_dataset("event_index")?;
+        let event_time_zero = detector.get_dataset("event_time_zero")?;
+        let period_number = detector.get_dataset("period_number")?;
+        let frame_number = detector.get_dataset("frame_number")?;
+        let frame_complete = detector.get_dataset("is_frame_complete")?;
+        let running = detector.get_dataset("running")?;
+        let veto_flags = detector.get_dataset("veto_flag")?;
+
+        let offset: Option<NexusDateTime> = {
+            if let Ok(offset) = event_time_zero.get_attribute("offset") {
+                Some(offset.get_datetime_from()?)
             } else {
                 None
             }
         };
 
         Ok(Self {
+            parent: detector,
             offset,
             num_messages: event_time_zero.size(),
             num_events: event_time_offset.size(),
@@ -165,9 +149,10 @@ impl EventRun {
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn init(&mut self, offset: &DateTime<Utc>) -> anyhow::Result<()> {
+    pub(crate) fn init(&mut self, offset: &NexusDateTime) -> NexusHDF5Result<()> {
         self.offset = Some(*offset);
-        add_attribute_to(&self.event_time_zero, "offset", &offset.to_rfc3339())?;
+        self.event_time_zero
+            .add_attribute_to("offset", &offset.to_rfc3339())?;
         Ok(())
     }
 
@@ -177,80 +162,67 @@ impl EventRun {
         fields(message_number, num_events),
         err(level = "warn")
     )]
-    pub(crate) fn push_message_to_event_runfile(
+    pub(crate) fn push_frame_eventlist_message_to_runfile(
         &mut self,
         message: &FrameAssembledEventListMessage,
-    ) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         tracing::Span::current().record("message_number", self.num_messages);
 
         // Fields Indexed By Frame
-        let next_message_slice = s![self.num_messages..(self.num_messages + 1)];
-        self.event_index.resize(self.num_messages + 1)?;
-        self.event_index
-            .write_slice(&[self.num_events], next_message_slice)?;
+        self.event_index.append_slice(&[self.num_events])?;
 
         // Recalculate time_zero of the frame to be relative to the offset value
         // (set at the start of the run).
-        let time_zero = self.get_time_zero(message)?;
-
-        self.event_time_zero.resize(self.num_messages + 1)?;
-        self.event_time_zero
-            .write_slice(&[time_zero], next_message_slice)?;
+        let time_zero = self
+            .get_time_zero(message)
+            .err_dataset(&self.event_time_zero)?;
 
-        self.period_number.resize(self.num_messages + 1)?;
+        self.event_time_zero.append_slice(&[time_zero])?;
         self.period_number
-            .write_slice(&[message.metadata().period_number()], next_message_slice)?;
-
-        self.frame_number.resize(self.num_messages + 1)?;
+            .append_slice(&[message.metadata().period_number()])?;
         self.frame_number
-            .write_slice(&[message.metadata().frame_number()], next_message_slice)?;
-
-        self.frame_complete.resize(self.num_messages + 1)?;
-        self.frame_complete
-            .write_slice(&[message.complete()], next_message_slice)?;
+            .append_slice(&[message.metadata().frame_number()])?;
+        self.frame_complete.append_slice(&[message.complete()])?;
 
-        self.running.resize(self.num_messages + 1)?;
-        self.running
-            .write_slice(&[message.metadata().running()], next_message_slice)?;
+        self.running.append_slice(&[message.metadata().running()])?;
 
-        self.veto_flags.resize(self.num_messages + 1)?;
         self.veto_flags
-            .write_slice(&[message.metadata().veto_flags()], next_message_slice)?;
+            .append_slice(&[message.metadata().veto_flags()])?;
 
         // Fields Indexed By Event
         let num_new_events = message.channel().unwrap_or_default().len();
         let total_events = self.num_events + num_new_events;
-        let next_event_block_slice = s![self.num_events..total_events];
-
-        self.pulse_height.resize(total_events)?;
-        self.pulse_height.write_slice(
-            &message
-                .voltage()
-                .unwrap_or_default()
-                .iter()
-                .collect::<Vec<_>>(),
-            next_event_block_slice,
-        )?;
-
-        self.event_time_offset.resize(total_events)?;
-        self.event_time_offset.write_slice(
-            &message
-                .time()
-                .unwrap_or_default()
-                .iter()
-                .collect::<Vec<_>>(),
-            next_event_block_slice,
-        )?;
 
-        self.event_id.resize(total_events)?;
-        self.event_id.write_slice(
-            &message
-                .channel()
-                .unwrap_or_default()
-                .iter()
-                .collect::<Vec<_>>(),
-            next_event_block_slice,
-        )?;
+        let intensities = &message
+            .voltage()
+            .ok_or(NexusHDF5Error::new_flatbuffer_missing(
+                FlatBufferMissingError::Intensities,
+            ))
+            .err_group(&self.parent)?
+            .iter()
+            .collect::<Vec<_>>();
+
+        let times = &message
+            .time()
+            .ok_or(NexusHDF5Error::new_flatbuffer_missing(
+                FlatBufferMissingError::Times,
+            ))
+            .err_group(&self.parent)?
+            .iter()
+            .collect::<Vec<_>>();
+
+        let channels = &message
+            .channel()
+            .ok_or(NexusHDF5Error::new_flatbuffer_missing(
+                FlatBufferMissingError::Channels,
+            ))
+            .err_group(&self.parent)?
+            .iter()
+            .collect::<Vec<_>>();
+
+        self.pulse_height.append_slice(intensities)?;
+        self.event_time_offset.append_slice(times)?;
+        self.event_id.append_slice(channels)?;
 
         self.num_events = total_events;
         self.num_messages += 1;
@@ -262,21 +234,27 @@ impl EventRun {
     pub(crate) fn get_time_zero(
         &self,
         message: &FrameAssembledEventListMessage,
-    ) -> anyhow::Result<u64> {
-        let timestamp: DateTime<Utc> = (*message
-            .metadata()
-            .timestamp()
-            .ok_or(anyhow::anyhow!("Message timestamp missing."))?)
-        .try_into()?;
+    ) -> Result<i64, NexusHDF5Error> {
+        let timestamp: NexusDateTime =
+            (*message
+                .metadata()
+                .timestamp()
+                .ok_or(NexusHDF5Error::new_flatbuffer_missing(
+                    FlatBufferMissingError::Timestamp,
+                ))?)
+            .try_into()?;
 
         // Recalculate time_zero of the frame to be relative to the offset value
         // (set at the start of the run).
         let time_zero = self
             .offset
             .and_then(|offset| (timestamp - offset).num_nanoseconds())
-            .ok_or(anyhow::anyhow!("event_time_zero cannot be calculated."))?
-            as u64;
+            .ok_or(NexusHDF5Error::new_flatbuffer_timestamp_convert_to_nanoseconds())?;
 
         Ok(time_zero)
     }
+
+    pub(crate) fn get_offset(&self) -> Option<&NexusDateTime> {
+        self.offset.as_ref()
+    }
 }
diff --git a/nexus-writer/src/nexus/hdf5_file/run_file_components/mod.rs b/nexus-writer/src/nexus/hdf5_file/run_file_components/mod.rs
index da268803b..772d96000 100644
--- a/nexus-writer/src/nexus/hdf5_file/run_file_components/mod.rs
+++ b/nexus-writer/src/nexus/hdf5_file/run_file_components/mod.rs
@@ -3,8 +3,6 @@ mod runlog_file;
 mod selog_file;
 mod timeseries_file;
 
-use super::add_new_group_to;
-
 pub(crate) use event_run_file::EventRun;
 pub(crate) use runlog_file::RunLog;
 pub(crate) use selog_file::SeLog;
diff --git a/nexus-writer/src/nexus/hdf5_file/run_file_components/runlog_file.rs b/nexus-writer/src/nexus/hdf5_file/run_file_components/runlog_file.rs
index 64b460c76..45b1b2b9d 100644
--- a/nexus-writer/src/nexus/hdf5_file/run_file_components/runlog_file.rs
+++ b/nexus-writer/src/nexus/hdf5_file/run_file_components/runlog_file.rs
@@ -1,19 +1,18 @@
-use super::{add_new_group_to, timeseries_file::TimeSeriesDataSource};
+use super::timeseries_file::TimeSeriesDataSource;
 use crate::nexus::{
     hdf5_file::{
-        hdf5_writer::{create_resizable_dataset, set_slice_to},
-        run_file_components::timeseries_file::get_dataset_builder,
+        error::{ConvertResult, NexusHDF5Result},
+        hdf5_writer::{DatasetExt, GroupExt, HasAttributesExt},
+        run_file_components::timeseries_file::adjust_nanoseconds_by_origin_to_sec,
     },
-    nexus_class as NX, NexusSettings,
+    nexus_class as NX, NexusDateTime, NexusSettings,
 };
 use hdf5::{
     types::{IntSize, TypeDescriptor},
-    Group, SimpleExtents,
+    Dataset, Group, H5Type,
 };
-use ndarray::s;
 use supermusr_common::DigitizerId;
 use supermusr_streaming_types::ecs_f144_logdata_generated::f144_LogData;
-use tracing::debug;
 
 #[derive(Debug)]
 pub(crate) struct RunLog {
@@ -22,139 +21,142 @@ pub(crate) struct RunLog {
 
 impl RunLog {
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn new_runlog(parent: &Group) -> anyhow::Result<Self> {
-        let logs = add_new_group_to(parent, "runlog", NX::RUNLOG)?;
+    pub(crate) fn new_runlog(parent: &Group) -> NexusHDF5Result<Self> {
+        let logs = parent.add_new_group_to("runlog", NX::RUNLOG)?;
         Ok(Self { parent: logs })
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn open_runlog(parent: &Group) -> anyhow::Result<Self> {
-        let parent = parent.group("runlog")?;
+    pub(crate) fn open_runlog(parent: &Group) -> NexusHDF5Result<Self> {
+        let parent = parent.get_group("runlog")?;
         Ok(Self { parent })
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn push_logdata_to_runlog(
+    pub(crate) fn create_runlog_group<T: H5Type>(
         &mut self,
-        logdata: &f144_LogData,
+        name: &str,
+        type_descriptor: &TypeDescriptor,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
-        debug!("Type: {0:?}", logdata.value_type());
-
-        let timeseries = self.parent.group(logdata.source_name()).or_else(|err| {
-            debug!(
-                "Cannot find {0}. Creating new group.",
-                logdata.source_name()
-            );
-
-            let group = add_new_group_to(&self.parent, logdata.source_name(), NX::LOG)
-                .map_err(|e| e.context(err))?;
-
-            let time = create_resizable_dataset::<u64>(
-                &group,
+        time_units: &'static str,
+    ) -> NexusHDF5Result<(Dataset, Dataset)> {
+        let runlog = self.parent.get_group_or_create_new(name, NX::RUNLOG)?;
+        let timestamps = runlog.get_dataset_or_else("time", |_| {
+            let times = runlog.create_resizable_empty_dataset::<T>(
                 "time",
-                0,
                 nexus_settings.runloglist_chunk_size,
             )?;
-            logdata.write_initial_timestamp(&time)?;
-            get_dataset_builder(&logdata.get_hdf5_type()?, &group)?
-                .shape(SimpleExtents::resizable(vec![0]))
-                .chunk(nexus_settings.runloglist_chunk_size)
-                .create("value")?;
-            Ok::<_, anyhow::Error>(group)
+            let start = origin_time.to_rfc3339();
+            times.add_attribute_to("Start", &start)?;
+            times.add_attribute_to("Units", time_units)?;
+            Ok(times)
         })?;
-        let timestamps = timeseries.dataset("time")?;
-        let values = timeseries.dataset("value")?;
+
+        let values = runlog.get_dataset_or_create_dynamic_resizable_empty_dataset(
+            "value",
+            type_descriptor,
+            nexus_settings.runloglist_chunk_size,
+        )?;
+
+        Ok((timestamps, values))
+    }
+
+    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
+    pub(crate) fn push_logdata_to_runlog(
+        &mut self,
+        logdata: &f144_LogData,
+        origin_time: &NexusDateTime,
+        nexus_settings: &NexusSettings,
+    ) -> NexusHDF5Result<()> {
+        let (timestamps, values) = self.create_runlog_group::<f32>(
+            logdata.source_name(),
+            &logdata.get_hdf5_type().err_group(&self.parent)?,
+            origin_time,
+            nexus_settings,
+            "second",
+        )?;
 
         logdata.write_values_to_dataset(&values)?;
-        logdata.write_timestamps_to_dataset(&timestamps, 1)?;
+        logdata.write_timestamps_to_dataset(&timestamps, 1, origin_time)?;
         Ok(())
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn set_aborted_run_warning(
+    pub(crate) fn push_aborted_run_warning(
         &mut self,
-        stop_time: i32,
+        stop_time_ms: i64,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         const LOG_NAME: &str = "SuperMuSRDataPipeline_RunAborted";
-        let timeseries = self.parent.group(LOG_NAME).or_else(|err| {
-            let group =
-                add_new_group_to(&self.parent, LOG_NAME, NX::LOG).map_err(|e| e.context(err))?;
+        let (timestamps, values) = self.create_runlog_group::<f32>(
+            LOG_NAME,
+            &TypeDescriptor::Unsigned(IntSize::U1),
+            origin_time,
+            nexus_settings,
+            "second",
+        )?;
+
+        timestamps.append_slice(&[adjust_nanoseconds_by_origin_to_sec(
+            1_000 * stop_time_ms,
+            origin_time,
+        )])?;
+        values.append_slice(&[0])?; // This is a default value, I'm not sure if this field is needed
 
-            let _time = create_resizable_dataset::<u64>(
-                &group,
-                "time",
-                0,
-                nexus_settings.runloglist_chunk_size,
-            )?;
-            get_dataset_builder(&TypeDescriptor::Unsigned(IntSize::U1), &group)?
-                .shape(SimpleExtents::resizable(vec![0]))
-                .chunk(nexus_settings.runloglist_chunk_size)
-                .create("value")?;
-            Ok::<_, anyhow::Error>(group)
-        })?;
-        let timestamps = timeseries.dataset("time")?;
-        let values = timeseries.dataset("value")?;
+        Ok(())
+    }
 
-        set_slice_to(&timestamps, &[stop_time])?;
-        set_slice_to(&values, &[0])?; // This is a default value, I'm not sure if this field is needed
+    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
+    pub(crate) fn push_run_resumed_warning(
+        &mut self,
+        current_time: &NexusDateTime,
+        origin_time: &NexusDateTime,
+        nexus_settings: &NexusSettings,
+    ) -> NexusHDF5Result<()> {
+        const LOG_NAME: &str = "SuperMuSRDataPipeline_RunResumed";
+        let (timestamps, values) = self.create_runlog_group::<f32>(
+            LOG_NAME,
+            &TypeDescriptor::Unsigned(IntSize::U1),
+            origin_time,
+            nexus_settings,
+            "second",
+        )?;
+
+        timestamps.append_slice(&[(*current_time - origin_time)
+            .num_nanoseconds()
+            .unwrap_or_default()])?;
+        values.append_slice(&[0])?; // This is a default value, I'm not sure if this field is needed
 
         Ok(())
     }
 
     pub(crate) fn push_incomplete_frame_log(
         &mut self,
-        event_time_zero: u64,
+        event_time_zero: i64,
         digitisers_present: Vec<DigitizerId>,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         const LOG_NAME: &str = "SuperMuSRDataPipeline_DigitisersPresentInIncompleteFrame";
-        let timeseries = self.parent.group(LOG_NAME).or_else(|err| {
-            debug!("Cannot find {LOG_NAME}. Creating new group.");
-
-            let group =
-                add_new_group_to(&self.parent, LOG_NAME, NX::LOG).map_err(|e| e.context(err))?;
-
-            create_resizable_dataset::<u64>(
-                &group,
-                "time",
-                0,
-                nexus_settings.runloglist_chunk_size,
-            )?;
-            create_resizable_dataset::<hdf5::types::VarLenUnicode>(
-                &group,
-                "value",
-                0,
-                nexus_settings.runloglist_chunk_size,
-            )?;
-            Ok::<_, anyhow::Error>(group)
-        })?;
-        let timestamps = timeseries.dataset("time")?;
-        let values = timeseries.dataset("value")?;
-
-        if timestamps.size() != values.size() {
-            anyhow::bail!(
-                "time length ({}) and value length ({}) differ",
-                timestamps.size(),
-                values.size()
-            )
-        }
-
-        let current_size = timestamps.size();
-        let next_slice = s![current_size..(current_size + 1)];
+        let (timestamps, values) = self.create_runlog_group::<i64>(
+            LOG_NAME,
+            &TypeDescriptor::VarLenUnicode,
+            origin_time,
+            nexus_settings,
+            "ns",
+        )?;
 
-        timestamps.resize(current_size + 1)?;
-        timestamps.write_slice(&[event_time_zero], next_slice)?;
+        timestamps.append_slice(&[event_time_zero])?;
 
-        values.resize(current_size + 1)?;
         let value = digitisers_present
             .iter()
             .map(DigitizerId::to_string)
             .collect::<Vec<_>>()
             .join(",")
-            .parse::<hdf5::types::VarLenUnicode>()?;
-        values.write_slice(&[value], next_slice)?;
+            .parse::<hdf5::types::VarLenUnicode>()
+            .err_group(&self.parent)?;
+        values.append_slice(&[value])?;
 
         Ok(())
     }
diff --git a/nexus-writer/src/nexus/hdf5_file/run_file_components/selog_file.rs b/nexus-writer/src/nexus/hdf5_file/run_file_components/selog_file.rs
index 08fd9de3c..19b3bda3f 100644
--- a/nexus-writer/src/nexus/hdf5_file/run_file_components/selog_file.rs
+++ b/nexus-writer/src/nexus/hdf5_file/run_file_components/selog_file.rs
@@ -1,13 +1,12 @@
-use super::{add_new_group_to, timeseries_file::TimeSeriesDataSource};
+use super::timeseries_file::{adjust_nanoseconds_by_origin_to_sec, TimeSeriesDataSource};
 use crate::nexus::{
     hdf5_file::{
-        hdf5_writer::create_resizable_dataset,
-        run_file_components::timeseries_file::get_dataset_builder,
+        error::{ConvertResult, NexusHDF5Result},
+        hdf5_writer::{DatasetExt, GroupExt, HasAttributesExt},
     },
-    nexus_class as NX, NexusSettings,
+    nexus_class as NX, NexusDateTime, NexusSettings,
 };
-use hdf5::{types::VarLenUnicode, Group, SimpleExtents};
-use ndarray::s;
+use hdf5::{types::VarLenUnicode, Group};
 use supermusr_streaming_types::{
     ecs_al00_alarm_generated::Alarm, ecs_se00_data_generated::se00_SampleEnvironmentData,
 };
@@ -20,92 +19,73 @@ pub(crate) struct SeLog {
 
 impl SeLog {
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn new_selog(parent: &Group) -> anyhow::Result<Self> {
-        let logs = add_new_group_to(parent, "selog", NX::SELOG)?;
+    pub(crate) fn new_selog(parent: &Group) -> NexusHDF5Result<Self> {
+        let logs = parent.add_new_group_to("selog", NX::SELOG)?;
         Ok(Self { parent: logs })
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn open_selog(parent: &Group) -> anyhow::Result<Self> {
-        let parent = parent.group("selog")?;
+    pub(crate) fn open_selog(parent: &Group) -> NexusHDF5Result<Self> {
+        let parent = parent.get_group("selog")?;
         Ok(Self { parent })
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    fn create_new_selogdata_in_selog(
+    pub(crate) fn push_alarm_to_selog(
         &mut self,
-        selog: &se00_SampleEnvironmentData,
+        alarm: Alarm,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<Group> {
-        add_new_group_to(&self.parent, selog.name(), NX::SELOG_BLOCK).and_then(|parent_group| {
-            let group = add_new_group_to(&parent_group, "value_log", NX::LOG)?;
-            let time = create_resizable_dataset::<i32>(
-                &group,
-                "time",
-                0,
-                nexus_settings.seloglist_chunk_size,
-            )?;
-            selog.write_initial_timestamp(&time)?;
-            get_dataset_builder(&selog.get_hdf5_type()?, &group)?
-                .shape(SimpleExtents::resizable(vec![0]))
-                .chunk(nexus_settings.seloglist_chunk_size)
-                .create("value")?;
-
-            create_resizable_dataset::<VarLenUnicode>(
-                &group,
-                "alarm_severity",
-                0,
-                nexus_settings.alarmlist_chunk_size,
-            )?;
-            create_resizable_dataset::<VarLenUnicode>(
-                &group,
-                "alarm_status",
-                0,
-                nexus_settings.alarmlist_chunk_size,
-            )?;
-            create_resizable_dataset::<i64>(
-                &group,
-                "alarm_time",
-                0,
-                nexus_settings.alarmlist_chunk_size,
-            )?;
-
-            Ok::<_, anyhow::Error>(parent_group)
-        })
-    }
-
-    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn push_alarm_to_selog(&mut self, alarm: Alarm) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         if let Some(source_name) = alarm.source_name() {
-            if let Ok(timeseries) = self
+            let seblock = self
                 .parent
-                .group(source_name)
-                .and_then(|group| group.group("value_log"))
-            {
-                let alarm_time = timeseries.dataset("alarm_time")?;
-                let alarm_status = timeseries.dataset("alarm_status")?;
-                let alarm_severity = timeseries.dataset("alarm_severity")?;
-                alarm_time.resize(alarm_time.size() + 1)?;
-                alarm_time.write_slice(
-                    &[alarm.timestamp()],
-                    s![(alarm_time.size() - 1)..alarm_time.size()],
+                .get_group_or_create_new(source_name, NX::SELOG_BLOCK)?;
+            let value_log = seblock.get_group_or_create_new("value_log", NX::LOG)?;
+
+            let alarm_time = value_log.get_dataset_or_else("alarm_time", |group| {
+                let alarm_time = group.create_resizable_empty_dataset::<f32>(
+                    "alarm_time",
+                    nexus_settings.alarmlist_chunk_size,
                 )?;
+                let start = origin_time.to_rfc3339();
+                alarm_time.add_attribute_to("Start", &start)?;
+                alarm_time.add_attribute_to("Units", "second")?;
+                Ok(alarm_time)
+            })?;
+
+            let alarm_status = value_log.get_dataset_or_else("alarm_status", |group| {
+                group.create_resizable_empty_dataset::<VarLenUnicode>(
+                    "alarm_status",
+                    nexus_settings.alarmlist_chunk_size,
+                )
+            })?;
+
+            let alarm_severity = value_log.get_dataset_or_else("alarm_severity", |group| {
+                group.create_resizable_empty_dataset::<VarLenUnicode>(
+                    "alarm_severity",
+                    nexus_settings.alarmlist_chunk_size,
+                )
+            })?;
 
-                alarm_severity.resize(alarm_severity.size() + 1)?;
-                if let Some(severity) = alarm.severity().variant_name() {
-                    alarm_severity.write_slice(
-                        &[severity.parse::<VarLenUnicode>()?],
-                        s![(alarm_severity.size() - 1)..alarm_severity.size()],
-                    )?;
-                }
+            alarm_time.append_slice(&[adjust_nanoseconds_by_origin_to_sec(
+                alarm.timestamp(),
+                origin_time,
+            )])?;
+
+            if let Some(severity) = alarm.severity().variant_name() {
+                alarm_severity.append_slice(&[severity
+                    .parse::<VarLenUnicode>()
+                    .err_group(&self.parent)?])?;
+            } else {
+                alarm_severity.append_slice(&[VarLenUnicode::default()])?;
+            }
 
-                alarm_status.resize(alarm_status.size() + 1)?;
-                if let Some(message) = alarm.message() {
-                    alarm_status.write_slice(
-                        &[message.parse::<VarLenUnicode>()?],
-                        s![(alarm_status.size() - 1)..alarm_status.size()],
-                    )?;
-                }
+            if let Some(message) = alarm.message() {
+                alarm_status
+                    .append_slice(&[message.parse::<VarLenUnicode>().err_group(&self.parent)?])?;
+            } else {
+                alarm_severity.append_slice(&[VarLenUnicode::default()])?;
             }
         }
         Ok(())
@@ -115,22 +95,35 @@ impl SeLog {
     pub(crate) fn push_selogdata_to_selog(
         &mut self,
         selog: &se00_SampleEnvironmentData,
+        origin_time: &NexusDateTime,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusHDF5Result<()> {
         debug!("Type: {0:?}", selog.values_type());
 
-        let timeseries = self
+        let seblock = self
             .parent
-            .group(selog.name())
-            .or_else(|err| {
-                self.create_new_selogdata_in_selog(selog, nexus_settings)
-                    .map_err(|e| e.context(err))
-            })?
-            .group("value_log")?;
-        let timestamps = timeseries.dataset("time")?;
-        let values = timeseries.dataset("value")?;
+            .get_group_or_create_new(selog.name(), NX::SELOG_BLOCK)?;
+        let value_log = seblock.get_group_or_create_new("value_log", NX::LOG)?;
+
+        let timestamps = value_log.get_dataset_or_else("time", |_| {
+            let times = value_log.create_resizable_empty_dataset::<f32>(
+                "time",
+                nexus_settings.runloglist_chunk_size,
+            )?;
+            let start = origin_time.to_rfc3339();
+            times.add_attribute_to("Start", &start)?;
+            times.add_attribute_to("Units", "second")?;
+            Ok(times)
+        })?;
+
+        let values = value_log.get_dataset_or_create_dynamic_resizable_empty_dataset(
+            "value",
+            &selog.get_hdf5_type().err_group(&self.parent)?,
+            nexus_settings.seloglist_chunk_size,
+        )?;
+
         let num_values = selog.write_values_to_dataset(&values)?;
-        selog.write_timestamps_to_dataset(&timestamps, num_values)?;
+        selog.write_timestamps_to_dataset(&timestamps, num_values, origin_time)?;
         Ok(())
     }
 }
diff --git a/nexus-writer/src/nexus/hdf5_file/run_file_components/timeseries_file.rs b/nexus-writer/src/nexus/hdf5_file/run_file_components/timeseries_file.rs
index 0a085d074..df156802b 100644
--- a/nexus-writer/src/nexus/hdf5_file/run_file_components/timeseries_file.rs
+++ b/nexus-writer/src/nexus/hdf5_file/run_file_components/timeseries_file.rs
@@ -1,7 +1,14 @@
-use chrono::{DateTime, Utc};
+use crate::nexus::{
+    error::FlatBufferInvalidDataTypeContext,
+    hdf5_file::{
+        error::{ConvertResult, NexusHDF5Error, NexusHDF5Result},
+        hdf5_writer::DatasetExt,
+    },
+    NexusDateTime,
+};
 use hdf5::{
     types::{FloatSize, IntSize, TypeDescriptor},
-    Dataset, DatasetBuilderEmpty, Group, H5Type,
+    Dataset, H5Type,
 };
 use ndarray::{s, Dim, SliceInfo, SliceInfoElem};
 use std::fmt::Debug;
@@ -10,135 +17,138 @@ use supermusr_streaming_types::{
     ecs_se00_data_generated::{se00_SampleEnvironmentData, ValueUnion},
     flatbuffers::{Follow, Vector},
 };
-use tracing::{debug, trace};
-
-use crate::nexus::hdf5_file::hdf5_writer::add_attribute_to;
+use tracing::{debug, trace, warn};
 
 pub(super) type Slice1D = SliceInfo<[SliceInfoElem; 1], Dim<[usize; 1]>, Dim<[usize; 1]>>;
 
+pub(super) fn adjust_nanoseconds_by_origin_to_sec(
+    nanoseconds: i64,
+    origin_time: &NexusDateTime,
+) -> f64 {
+    (origin_time
+        .timestamp_nanos_opt()
+        .map(|origin_time_ns| nanoseconds - origin_time_ns)
+        .unwrap_or_default() as f64)
+        / 1_000_000_000.0
+}
+
 pub(super) trait TimeSeriesDataSource<'a>: Debug {
-    fn write_initial_timestamp(&self, target: &Dataset) -> anyhow::Result<()>;
     fn write_timestamps_to_dataset(
         &self,
         target: &Dataset,
         num_values: usize,
-    ) -> anyhow::Result<()>;
-    fn write_values_to_dataset(&self, target: &Dataset) -> anyhow::Result<usize>;
-    fn get_hdf5_type(&self) -> anyhow::Result<TypeDescriptor>;
-}
-
-pub(super) fn get_dataset_builder(
-    type_descriptor: &TypeDescriptor,
-    parent: &Group,
-) -> anyhow::Result<DatasetBuilderEmpty> {
-    Ok(match type_descriptor {
-        TypeDescriptor::Integer(sz) => match sz {
-            IntSize::U1 => parent.new_dataset::<i8>(),
-            IntSize::U2 => parent.new_dataset::<i16>(),
-            IntSize::U4 => parent.new_dataset::<i32>(),
-            IntSize::U8 => parent.new_dataset::<i64>(),
-        },
-        TypeDescriptor::Unsigned(sz) => match sz {
-            IntSize::U1 => parent.new_dataset::<u8>(),
-            IntSize::U2 => parent.new_dataset::<u16>(),
-            IntSize::U4 => parent.new_dataset::<u32>(),
-            IntSize::U8 => parent.new_dataset::<u64>(),
-        },
-        TypeDescriptor::Float(sz) => match sz {
-            FloatSize::U4 => parent.new_dataset::<f32>(),
-            FloatSize::U8 => parent.new_dataset::<f64>(),
-        },
-        _ => {
-            return Err(anyhow::anyhow!(
-                "Invalid HDF5 array type: {}",
-                type_descriptor.to_string()
-            ))
-        }
-    })
+        origin_time: &NexusDateTime,
+    ) -> NexusHDF5Result<()>;
+    fn write_values_to_dataset(&self, target: &Dataset) -> NexusHDF5Result<usize>;
+    fn get_hdf5_type(&self) -> Result<TypeDescriptor, NexusHDF5Error>;
 }
 
 fn write_generic_logdata_slice_to_dataset<T: H5Type>(
     val: T,
     target: &Dataset,
-) -> anyhow::Result<Slice1D> {
+) -> NexusHDF5Result<Slice1D> {
     let position = target.size();
     let slice = s![position..(position + 1)];
-    target.resize(position + 1)?;
-    target.write_slice(&[val], slice)?;
+    target.append_slice(&[val])?;
     Ok(slice)
 }
 
 impl<'a> TimeSeriesDataSource<'a> for f144_LogData<'a> {
-    #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    fn write_initial_timestamp(&self, target: &Dataset) -> anyhow::Result<()> {
-        let time = DateTime::<Utc>::from_timestamp_nanos(self.timestamp()).to_rfc3339();
-        add_attribute_to(target, "Start", &time)?;
-        add_attribute_to(target, "Units", "second")?;
-        Ok(())
-    }
-
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
     fn write_timestamps_to_dataset(
         &self,
         target: &Dataset,
         _num_values: usize,
-    ) -> anyhow::Result<()> {
+        origin_time: &NexusDateTime,
+    ) -> NexusHDF5Result<()> {
         let position = target.size();
-        target.resize(position + 1)?;
         let slice = s![position..(position + 1)];
         debug!("Timestamp Slice: {slice:?}, Value: {0:?}", self.timestamp());
-        target.write_slice(&[self.timestamp()], slice)?;
+        target.append_slice(&[adjust_nanoseconds_by_origin_to_sec(
+            self.timestamp(),
+            origin_time,
+        )])?;
         Ok(())
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    fn write_values_to_dataset(&self, target: &Dataset) -> anyhow::Result<usize> {
-        let type_descriptor = self.get_hdf5_type()?;
-        let error = anyhow::anyhow!("Could not convert value to type {type_descriptor:?}");
+    fn write_values_to_dataset(&self, target: &Dataset) -> NexusHDF5Result<usize> {
+        let type_descriptor = self.get_hdf5_type().err_dataset(target)?;
+        let error = NexusHDF5Error::new_invalid_hdf5_type_conversion(type_descriptor.clone());
         match type_descriptor {
             TypeDescriptor::Integer(sz) => match sz {
                 IntSize::U1 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_byte().ok_or(error)?.value(),
+                    self.value_as_byte()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U2 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_short().ok_or(error)?.value(),
+                    self.value_as_short()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U4 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_int().ok_or(error)?.value(),
+                    self.value_as_int()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U8 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_long().ok_or(error)?.value(),
+                    self.value_as_long()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
             },
             TypeDescriptor::Unsigned(sz) => match sz {
                 IntSize::U1 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_ubyte().ok_or(error)?.value(),
+                    self.value_as_ubyte()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U2 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_ushort().ok_or(error)?.value(),
+                    self.value_as_ushort()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U4 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_uint().ok_or(error)?.value(),
+                    self.value_as_uint()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U8 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_ulong().ok_or(error)?.value(),
+                    self.value_as_ulong()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
             },
             TypeDescriptor::Float(sz) => match sz {
                 FloatSize::U4 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_float().ok_or(error)?.value(),
+                    self.value_as_float()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 FloatSize::U8 => write_generic_logdata_slice_to_dataset(
-                    self.value_as_double().ok_or(error)?.value(),
+                    self.value_as_double()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
             },
@@ -147,7 +157,7 @@ impl<'a> TimeSeriesDataSource<'a> for f144_LogData<'a> {
         Ok(1)
     }
 
-    fn get_hdf5_type(&self) -> anyhow::Result<TypeDescriptor> {
+    fn get_hdf5_type(&self) -> Result<TypeDescriptor, NexusHDF5Error> {
         let datatype = match self.value_type() {
             Value::Byte => TypeDescriptor::Integer(IntSize::U1),
             Value::UByte => TypeDescriptor::Unsigned(IntSize::U1),
@@ -159,7 +169,12 @@ impl<'a> TimeSeriesDataSource<'a> for f144_LogData<'a> {
             Value::ULong => TypeDescriptor::Unsigned(IntSize::U8),
             Value::Float => TypeDescriptor::Float(FloatSize::U4),
             Value::Double => TypeDescriptor::Float(FloatSize::U8),
-            t => anyhow::bail!("Invalid flatbuffers logdata type {:?}", t.variant_name()),
+            t => {
+                return Err(NexusHDF5Error::new_flatbuffer_invalid_data_type(
+                    FlatBufferInvalidDataTypeContext::RunLog,
+                    t.variant_name().map(ToOwned::to_owned).unwrap_or_default(),
+                ))
+            }
         };
         Ok(datatype)
     }
@@ -168,110 +183,140 @@ impl<'a> TimeSeriesDataSource<'a> for f144_LogData<'a> {
 fn write_generic_se_slice_to_dataset<'a, T: Follow<'a>>(
     vec: Vector<'a, T>,
     target: &Dataset,
-) -> anyhow::Result<usize>
+) -> NexusHDF5Result<usize>
 where
     T::Inner: H5Type,
 {
     let size = vec.len();
-    let position = target.size();
-    let slice = s![position..(position + size)];
-    target.resize(position + size)?;
-    target.write_slice(&vec.iter().collect::<Vec<_>>(), slice)?;
+    target.append_slice(&vec.iter().collect::<Vec<_>>())?;
     Ok(size)
 }
 
 impl<'a> TimeSeriesDataSource<'a> for se00_SampleEnvironmentData<'a> {
-    #[tracing::instrument(skip_all, err(level = "warn"))]
-    fn write_initial_timestamp(&self, target: &Dataset) -> anyhow::Result<()> {
-        let time = DateTime::<Utc>::from_timestamp_nanos(self.packet_timestamp()).to_rfc3339();
-        add_attribute_to(target, "Start", &time)?;
-        add_attribute_to(target, "Units", "second")?;
-        Ok(())
-    }
-
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
     fn write_timestamps_to_dataset(
         &self,
         target: &Dataset,
         num_values: usize,
-    ) -> anyhow::Result<()> {
+        origin_time: &NexusDateTime,
+    ) -> NexusHDF5Result<()> {
         let position = target.size();
         if let Some(timestamps) = self.timestamps() {
             trace!("Times given explicitly.");
 
-            let timestamps = timestamps.iter().collect::<Vec<_>>();
+            let timestamps = timestamps
+                .iter()
+                .map(|t| adjust_nanoseconds_by_origin_to_sec(t, origin_time))
+                .collect::<Vec<_>>();
+
             if timestamps.len() != num_values {
-                anyhow::bail!("Different number of values and times");
+                return Err(NexusHDF5Error::FlatBufferInconsistentSELogTimeValueSizes {
+                    sizes: (timestamps.len(), num_values),
+                    hdf5_path: None,
+                })
+                .err_dataset(target);
             }
             let slice = s![position..(position + num_values)];
 
             debug!("Timestamp Slice: {slice:?}, Times: {0:?}", timestamps);
-            target.resize(position + num_values)?;
-            target.write_slice(timestamps.as_slice(), slice)?;
+            target.append_slice(timestamps.as_slice())?;
         } else if self.time_delta() > 0.0 {
             trace!("Calculate times automatically.");
 
             let timestamps = (0..num_values)
-                .map(|v| v as f64 * self.time_delta())
+                .map(|v| (v as f64 * self.time_delta()) as i64)
+                .map(|t| {
+                    adjust_nanoseconds_by_origin_to_sec(t + self.packet_timestamp(), origin_time)
+                })
                 .collect::<Vec<_>>();
             let slice = s![position..(position + num_values)];
 
             debug!("Timestamp Slice: {slice:?}, Times: {0:?}", timestamps,);
-            target.write_slice(timestamps.as_slice(), slice)?;
+            target.append_slice(timestamps.as_slice())?;
         } else {
-            trace!("No time data.");
+            warn!("No time data.");
         }
         Ok(())
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    fn write_values_to_dataset(&self, target: &Dataset) -> anyhow::Result<usize> {
-        let type_descriptor = self.get_hdf5_type()?;
-        let error = anyhow::anyhow!("Could not convert value to type {type_descriptor:?}");
+    fn write_values_to_dataset(&self, target: &Dataset) -> NexusHDF5Result<usize> {
+        let type_descriptor = self.get_hdf5_type().err_dataset(target)?;
+        let error = NexusHDF5Error::new_invalid_hdf5_type_conversion(type_descriptor.clone());
         match type_descriptor {
             TypeDescriptor::Integer(sz) => match sz {
                 IntSize::U1 => write_generic_se_slice_to_dataset(
-                    self.values_as_int_8_array().ok_or(error)?.value(),
+                    self.values_as_int_8_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U2 => write_generic_se_slice_to_dataset(
-                    self.values_as_int_16_array().ok_or(error)?.value(),
+                    self.values_as_int_16_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U4 => write_generic_se_slice_to_dataset(
-                    self.values_as_int_32_array().ok_or(error)?.value(),
+                    self.values_as_int_32_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U8 => write_generic_se_slice_to_dataset(
-                    self.values_as_int_64_array().ok_or(error)?.value(),
+                    self.values_as_int_64_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
             },
             TypeDescriptor::Unsigned(sz) => match sz {
                 IntSize::U1 => write_generic_se_slice_to_dataset(
-                    self.values_as_uint_8_array().ok_or(error)?.value(),
+                    self.values_as_uint_8_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U2 => write_generic_se_slice_to_dataset(
-                    self.values_as_uint_16_array().ok_or(error)?.value(),
+                    self.values_as_uint_16_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U4 => write_generic_se_slice_to_dataset(
-                    self.values_as_uint_32_array().ok_or(error)?.value(),
+                    self.values_as_uint_32_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 IntSize::U8 => write_generic_se_slice_to_dataset(
-                    self.values_as_uint_64_array().ok_or(error)?.value(),
+                    self.values_as_uint_64_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
             },
             TypeDescriptor::Float(sz) => match sz {
                 FloatSize::U4 => write_generic_se_slice_to_dataset(
-                    self.values_as_float_array().ok_or(error)?.value(),
+                    self.values_as_float_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
                 FloatSize::U8 => write_generic_se_slice_to_dataset(
-                    self.values_as_double_array().ok_or(error)?.value(),
+                    self.values_as_double_array()
+                        .ok_or(error)
+                        .err_dataset(target)?
+                        .value(),
                     target,
                 ),
             },
@@ -279,7 +324,7 @@ impl<'a> TimeSeriesDataSource<'a> for se00_SampleEnvironmentData<'a> {
         }
     }
 
-    fn get_hdf5_type(&self) -> anyhow::Result<TypeDescriptor> {
+    fn get_hdf5_type(&self) -> Result<TypeDescriptor, NexusHDF5Error> {
         let datatype = match self.values_type() {
             ValueUnion::Int8Array => TypeDescriptor::Integer(IntSize::U1),
             ValueUnion::UInt8Array => TypeDescriptor::Unsigned(IntSize::U1),
@@ -291,7 +336,12 @@ impl<'a> TimeSeriesDataSource<'a> for se00_SampleEnvironmentData<'a> {
             ValueUnion::UInt64Array => TypeDescriptor::Unsigned(IntSize::U8),
             ValueUnion::FloatArray => TypeDescriptor::Float(FloatSize::U4),
             ValueUnion::DoubleArray => TypeDescriptor::Float(FloatSize::U8),
-            t => anyhow::bail!("Invalid flatbuffers logdata type {:?}", t.variant_name()),
+            t => {
+                return Err(NexusHDF5Error::new_flatbuffer_invalid_data_type(
+                    FlatBufferInvalidDataTypeContext::SELog,
+                    t.variant_name().map(ToOwned::to_owned).unwrap_or_default(),
+                ))
+            }
         };
         Ok(datatype)
     }
diff --git a/nexus-writer/src/nexus/mod.rs b/nexus-writer/src/nexus/mod.rs
index 2307eff20..aeeb62346 100644
--- a/nexus-writer/src/nexus/mod.rs
+++ b/nexus-writer/src/nexus/mod.rs
@@ -1,13 +1,16 @@
 mod engine;
+mod error;
 mod hdf5_file;
 mod run;
 mod run_parameters;
 
+use chrono::{DateTime, Utc};
 pub(crate) use engine::{NexusConfiguration, NexusEngine, NexusSettings};
 pub(crate) use run::Run;
 pub(crate) use run_parameters::RunParameters;
 
 pub(crate) const DATETIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%z";
+pub(crate) type NexusDateTime = DateTime<Utc>;
 
 pub(crate) mod nexus_class {
     pub(crate) const DETECTOR: &str = "NXdetector";
diff --git a/nexus-writer/src/nexus/run.rs b/nexus-writer/src/nexus/run.rs
index fe342276f..a9a3294a6 100644
--- a/nexus-writer/src/nexus/run.rs
+++ b/nexus-writer/src/nexus/run.rs
@@ -1,5 +1,8 @@
-use super::{hdf5_file::RunFile, NexusConfiguration, NexusSettings, RunParameters};
-use chrono::{DateTime, Duration, Utc};
+use super::{
+    error::NexusWriterResult, hdf5_file::RunFile, NexusConfiguration, NexusDateTime, NexusSettings,
+    RunParameters,
+};
+use chrono::{Duration, Utc};
 use std::{fs::create_dir_all, future::Future, io, path::Path};
 use supermusr_common::spanned::{SpanOnce, SpanOnceError, Spanned, SpannedAggregator, SpannedMut};
 use supermusr_streaming_types::{
@@ -21,7 +24,7 @@ impl Run {
         parameters: RunParameters,
         nexus_settings: &NexusSettings,
         nexus_configuration: &NexusConfiguration,
-    ) -> anyhow::Result<Self> {
+    ) -> NexusWriterResult<Self> {
         if let Some(local_path) = local_path {
             let mut hdf5 = RunFile::new_runfile(local_path, &parameters.run_name, nexus_settings)?;
             hdf5.init(&parameters, nexus_configuration)?;
@@ -34,9 +37,14 @@ impl Run {
         })
     }
 
-    pub(crate) fn resume_partial_run(local_path: &Path, filename: &str) -> anyhow::Result<Self> {
-        let run = RunFile::open_runfile(local_path, filename)?;
+    pub(crate) fn resume_partial_run(
+        local_path: &Path,
+        filename: &str,
+        nexus_settings: &NexusSettings,
+    ) -> NexusWriterResult<Self> {
+        let mut run = RunFile::open_runfile(local_path, filename)?;
         let parameters = run.extract_run_parameters()?;
+        run.push_run_resumed_warning(&Utc::now(), &parameters.collect_from, nexus_settings)?;
         run.close()?;
 
         Ok(Self {
@@ -81,10 +89,10 @@ impl Run {
         local_path: Option<&Path>,
         logdata: &f144_LogData,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusWriterResult<()> {
         if let Some(local_path) = local_path {
             let mut hdf5 = RunFile::open_runfile(local_path, &self.parameters.run_name)?;
-            hdf5.push_logdata_to_runfile(logdata, nexus_settings)?;
+            hdf5.push_logdata_to_runfile(logdata, &self.parameters.collect_from, nexus_settings)?;
             hdf5.close()?;
         }
 
@@ -97,10 +105,11 @@ impl Run {
         &mut self,
         local_path: Option<&Path>,
         alarm: Alarm,
-    ) -> anyhow::Result<()> {
+        nexus_settings: &NexusSettings,
+    ) -> NexusWriterResult<()> {
         if let Some(local_path) = local_path {
             let mut hdf5 = RunFile::open_runfile(local_path, &self.parameters.run_name)?;
-            hdf5.push_alarm_to_runfile(alarm)?;
+            hdf5.push_alarm_to_runfile(alarm, &self.parameters.collect_from, nexus_settings)?;
             hdf5.close()?;
         }
 
@@ -114,10 +123,10 @@ impl Run {
         local_path: Option<&Path>,
         logdata: se00_SampleEnvironmentData,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusWriterResult<()> {
         if let Some(local_path) = local_path {
             let mut hdf5 = RunFile::open_runfile(local_path, &self.parameters.run_name)?;
-            hdf5.push_selogdata(logdata, nexus_settings)?;
+            hdf5.push_selogdata(logdata, &self.parameters.collect_from, nexus_settings)?;
             hdf5.close()?;
         }
 
@@ -126,15 +135,20 @@ impl Run {
     }
 
     #[tracing::instrument(skip_all, level = "debug", err(level = "warn"))]
-    pub(crate) fn push_message(
+    pub(crate) fn push_frame_eventlist_message(
         &mut self,
         local_path: Option<&Path>,
         message: &FrameAssembledEventListMessage,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusWriterResult<()> {
         if let Some(local_path) = local_path {
             let mut hdf5 = RunFile::open_runfile(local_path, &self.parameters.run_name)?;
-            hdf5.push_message_to_runfile(message, nexus_settings)?;
+            hdf5.push_frame_eventlist_message_to_runfile(message)?;
+
+            if !message.complete() {
+                hdf5.push_incomplete_frame_warning(message, nexus_settings)?;
+            }
+
             hdf5.close()?;
         }
 
@@ -155,7 +169,7 @@ impl Run {
         &mut self,
         local_path: Option<&Path>,
         data: RunStop<'_>,
-    ) -> anyhow::Result<()> {
+    ) -> NexusWriterResult<()> {
         self.parameters.set_stop_if_valid(data)?;
 
         if let Some(local_path) = local_path {
@@ -179,7 +193,7 @@ impl Run {
         local_path: Option<&Path>,
         absolute_stop_time_ms: u64,
         nexus_settings: &NexusSettings,
-    ) -> anyhow::Result<()> {
+    ) -> NexusWriterResult<()> {
         self.parameters.set_aborted_run(absolute_stop_time_ms)?;
 
         if let Some(local_path) = local_path {
@@ -195,17 +209,17 @@ impl Run {
             hdf5.set_end_time(&collect_until)?;
             let relative_stop_time_ms =
                 (collect_until - self.parameters.collect_from).num_milliseconds();
-            if let Ok(relative_stop_time_ms) = relative_stop_time_ms.try_into() {
-                hdf5.set_aborted_run_warning(relative_stop_time_ms, nexus_settings)?;
-            } else {
-                warn!("Cannot convert {relative_stop_time_ms} to i32");
-            }
+            hdf5.push_aborted_run_warning(
+                relative_stop_time_ms,
+                &self.parameters.collect_from,
+                nexus_settings,
+            )?;
             hdf5.close()?;
         }
         Ok(())
     }
 
-    pub(crate) fn is_message_timestamp_valid(&self, timestamp: &DateTime<Utc>) -> bool {
+    pub(crate) fn is_message_timestamp_valid(&self, timestamp: &NexusDateTime) -> bool {
         self.parameters.is_message_timestamp_valid(timestamp)
     }
 
diff --git a/nexus-writer/src/nexus/run_parameters.rs b/nexus-writer/src/nexus/run_parameters.rs
index 1a5e4526a..e284989aa 100644
--- a/nexus-writer/src/nexus/run_parameters.rs
+++ b/nexus-writer/src/nexus/run_parameters.rs
@@ -1,19 +1,22 @@
+use super::{
+    error::{ErrorCodeLocation, FlatBufferMissingError, NexusWriterError, NexusWriterResult},
+    NexusDateTime,
+};
+use chrono::Utc;
 use std::path::{Path, PathBuf};
-
-use chrono::{DateTime, Utc};
 use supermusr_streaming_types::{
     ecs_6s4t_run_stop_generated::RunStop, ecs_pl72_run_start_generated::RunStart,
 };
 
 #[derive(Default, Debug, Clone)]
 pub(crate) struct RunStopParameters {
-    pub(crate) collect_until: DateTime<Utc>,
-    pub(crate) last_modified: DateTime<Utc>,
+    pub(crate) collect_until: NexusDateTime,
+    pub(crate) last_modified: NexusDateTime,
 }
 
 #[derive(Debug, Clone)]
 pub(crate) struct RunParameters {
-    pub(crate) collect_from: DateTime<Utc>,
+    pub(crate) collect_from: NexusDateTime,
     pub(crate) run_stop_parameters: Option<RunStopParameters>,
     pub(crate) num_periods: u32,
     pub(crate) run_name: String,
@@ -23,36 +26,45 @@ pub(crate) struct RunParameters {
 
 impl RunParameters {
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn new(data: RunStart<'_>, run_number: u32) -> anyhow::Result<Self> {
+    pub(crate) fn new(data: RunStart<'_>, run_number: u32) -> NexusWriterResult<Self> {
         Ok(Self {
-            collect_from: DateTime::<Utc>::from_timestamp_millis(data.start_time().try_into()?)
-                .ok_or(anyhow::anyhow!(
-                    "Cannot create start_time from {0}",
-                    &data.start_time()
-                ))?,
+            collect_from: NexusDateTime::from_timestamp_millis(data.start_time().try_into()?)
+                .ok_or(NexusWriterError::IntOutOfRangeForDateTime {
+                    int: data.start_time(),
+                    location: ErrorCodeLocation::NewRunParamemters,
+                })?,
             run_stop_parameters: None,
             num_periods: data.n_periods(),
             run_name: data
                 .run_name()
-                .ok_or(anyhow::anyhow!("Run Name not found"))?
+                .ok_or(NexusWriterError::FlatBufferMissing(
+                    FlatBufferMissingError::RunName,
+                    ErrorCodeLocation::NewRunParamemters,
+                ))?
                 .to_owned(),
             run_number,
             instrument_name: data
                 .instrument_name()
-                .ok_or(anyhow::anyhow!("Instrument Name not found"))?
+                .ok_or(NexusWriterError::FlatBufferMissing(
+                    FlatBufferMissingError::InstrumentName,
+                    ErrorCodeLocation::NewRunParamemters,
+                ))?
                 .to_owned(),
         })
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn set_stop_if_valid(&mut self, data: RunStop<'_>) -> anyhow::Result<()> {
+    pub(crate) fn set_stop_if_valid(&mut self, data: RunStop<'_>) -> NexusWriterResult<()> {
         if self.run_stop_parameters.is_some() {
-            Err(anyhow::anyhow!("Stop Command before Start Command"))
+            Err(NexusWriterError::StopCommandBeforeStartCommand(
+                ErrorCodeLocation::SetStopIfValid,
+            ))
         } else {
-            let stop_time =
-                DateTime::<Utc>::from_timestamp_millis(data.stop_time().try_into()?).ok_or(
-                    anyhow::anyhow!("Cannot create end_time from {0}", data.stop_time()),
-                )?;
+            let stop_time = NexusDateTime::from_timestamp_millis(data.stop_time().try_into()?)
+                .ok_or(NexusWriterError::IntOutOfRangeForDateTime {
+                    int: data.stop_time(),
+                    location: ErrorCodeLocation::SetStopIfValid,
+                })?;
             if self.collect_from < stop_time {
                 self.run_stop_parameters = Some(RunStopParameters {
                     collect_until: stop_time,
@@ -60,20 +72,27 @@ impl RunParameters {
                 });
                 Ok(())
             } else {
-                Err(anyhow::anyhow!(
-                    "Stop Time earlier than current Start Time."
-                ))
+                Err(NexusWriterError::StopTimeEarlierThanStartTime {
+                    start: self.collect_from,
+                    stop: stop_time,
+                    location: ErrorCodeLocation::SetStopIfValid,
+                })
             }
         }
     }
 
     #[tracing::instrument(skip_all, level = "trace", err(level = "warn"))]
-    pub(crate) fn set_aborted_run(&mut self, stop_time: u64) -> anyhow::Result<()> {
-        let collect_until = DateTime::<Utc>::from_timestamp_millis(stop_time.try_into()?).ok_or(
-            anyhow::anyhow!("Cannot create start_time from {0}", &stop_time),
+    pub(crate) fn set_aborted_run(&mut self, stop_time: u64) -> NexusWriterResult<()> {
+        let collect_until = NexusDateTime::from_timestamp_millis(stop_time.try_into()?).ok_or(
+            NexusWriterError::IntOutOfRangeForDateTime {
+                int: stop_time,
+                location: ErrorCodeLocation::SetAbortedRun,
+            },
         )?;
         if self.run_stop_parameters.is_some() {
-            anyhow::bail!("RunStop already set");
+            return Err(NexusWriterError::RunStopAlreadySet(
+                ErrorCodeLocation::SetAbortedRun,
+            ));
         }
         {
             self.run_stop_parameters = Some(RunStopParameters {
@@ -88,7 +107,7 @@ impl RunParameters {
     /// if run_stop_parameters exist then, if timestamp is strictly
     /// before params.collect_until.
     #[tracing::instrument(skip_all, level = "trace")]
-    pub(crate) fn is_message_timestamp_valid(&self, timestamp: &DateTime<Utc>) -> bool {
+    pub(crate) fn is_message_timestamp_valid(&self, timestamp: &NexusDateTime) -> bool {
         if self.collect_from < *timestamp {
             self.run_stop_parameters
                 .as_ref()