Propagate incomplete frame data (#295)

## Summary of changes

Writes the `complete` flag of each frame into the nexus file, as well as
recording the digitisers present as a run log, in the case of `complete
== false`.

Also modified the type of the time field in the runlogs: `i32` changed
to `u64` as `i32` does not have the range for recording `ns` precision.

## Instruction for review/testing

General code review.
Has been tested with simulated data.

Please excuse the use of `anyhow`, will be changed later.

Author: Modularius

nexus-writer/src/main.rs

@@ -293,6 +293,7 @@ where
         metadata_veto_flags = tracing::field::Empty,
         metadata_protons_per_pulse = tracing::field::Empty,
         metadata_running = tracing::field::Empty,
+        frame_is_complete = tracing::field::Empty,
     )
 )]
 fn process_frame_assembled_event_list_message(nexus_engine: &mut NexusEngine, payload: &[u8]) {
@@ -307,6 +308,7 @@ fn process_frame_assembled_event_list_message(nexus_engine: &mut NexusEngine, pa
                 .try_into()
                 .map(|metadata: FrameMetadata| {
                     record_metadata_fields_to_span!(metadata, tracing::Span::current());
+                    tracing::Span::current().record("frame_is_complete", data.complete());
                 })
                 .ok();
             match nexus_engine.process_event_list(&data) {
@@ -321,6 +323,7 @@ fn process_frame_assembled_event_list_message(nexus_engine: &mut NexusEngine, pa
                                 "metadata_veto_flags" = tracing::field::Empty,
                                 "metadata_protons_per_pulse" = tracing::field::Empty,
                                 "metadata_running" = tracing::field::Empty,
+                                "frame_is_complete" = data.complete(),
                             );
                             data.metadata()
                                 .try_into()

nexus-writer/src/nexus/engine.rs

@@ -210,7 +210,7 @@ impl NexusEngine {
             .iter_mut()
             .find(|run| run.is_message_timestamp_valid(&timestamp))
         {
-            run.push_message(self.local_path.as_deref(), message)?;
+            run.push_message(self.local_path.as_deref(), message, &self.nexus_settings)?;
             Some(run)
         } else {
             warn!("No run found for message with timestamp: {timestamp}");

nexus-writer/src/nexus/hdf5_file/run_file.rs

@@ -316,8 +316,24 @@ impl RunFile {
     pub(crate) fn push_message_to_runfile(
         &mut self,
         message: &FrameAssembledEventListMessage,
+        nexus_settings: &NexusSettings,
     ) -> anyhow::Result<()> {
-        self.contents.lists.push_message_to_event_runfile(message)
+        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> {

nexus-writer/src/nexus/hdf5_file/run_file_components/event_run_file.rs

@@ -20,6 +20,7 @@ pub(crate) struct EventRun {
     event_time_zero: Dataset,
     period_number: Dataset,
     frame_number: Dataset,
+    frame_complete: Dataset,
     //  Events
     event_id: Dataset,
     pulse_height: Dataset,
@@ -81,6 +82,13 @@ impl EventRun {
             nexus_settings.framelist_chunk_size,
         )?;
 
+        let frame_complete = create_resizable_dataset::<u64>(
+            &detector,
+            "is_frame_complete",
+            0,
+            nexus_settings.framelist_chunk_size,
+        )?;
+
         Ok(Self {
             offset: None,
             num_events: 0,
@@ -92,6 +100,7 @@ impl EventRun {
             event_time_zero,
             period_number,
             frame_number,
+            frame_complete,
         })
     }
 
@@ -107,6 +116,7 @@ impl EventRun {
         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 offset: Option<DateTime<Utc>> = {
             if let Ok(offset) = event_time_zero.attr("offset") {
@@ -128,6 +138,7 @@ impl EventRun {
             event_time_zero,
             period_number,
             frame_number,
+            frame_complete,
         })
     }
 
@@ -156,19 +167,9 @@ impl EventRun {
         self.event_index
             .write_slice(&[self.num_events], next_message_slice)?;
 
-        let timestamp: DateTime<Utc> = (*message
-            .metadata()
-            .timestamp()
-            .ok_or(anyhow::anyhow!("Message timestamp missing."))?)
-        .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;
+        let time_zero = self.get_time_zero(message)?;
 
         self.event_time_zero.resize(self.num_messages + 1)?;
         self.event_time_zero
@@ -182,6 +183,10 @@ impl EventRun {
         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)?;
+
         // Fields Indexed By Event
         let num_new_events = message.channel().unwrap_or_default().len();
         let total_events = self.num_events + num_new_events;
@@ -223,4 +228,25 @@ impl EventRun {
         tracing::Span::current().record("num_events", num_new_events);
         Ok(())
     }
+
+    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()?;
+
+        // 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(time_zero)
+    }
 }

nexus-writer/src/nexus/hdf5_file/run_file_components/runlog_file.rs

@@ -10,6 +10,8 @@ use hdf5::{
     types::{IntSize, TypeDescriptor},
     Group, SimpleExtents,
 };
+use ndarray::s;
+use supermusr_common::DigitizerId;
 use supermusr_streaming_types::ecs_f144_logdata_generated::f144_LogData;
 use tracing::debug;
 
@@ -48,7 +50,7 @@ impl RunLog {
             let group = add_new_group_to(&self.parent, logdata.source_name(), NX::LOG)
                 .map_err(|e| e.context(err))?;
 
-            let time = create_resizable_dataset::<i32>(
+            let time = create_resizable_dataset::<u64>(
                 &group,
                 "time",
                 0,
@@ -80,7 +82,7 @@ impl RunLog {
             let group =
                 add_new_group_to(&self.parent, LOG_NAME, NX::LOG).map_err(|e| e.context(err))?;
 
-            let _time = create_resizable_dataset::<i32>(
+            let _time = create_resizable_dataset::<u64>(
                 &group,
                 "time",
                 0,
@@ -100,4 +102,60 @@ impl RunLog {
 
         Ok(())
     }
+
+    pub(crate) fn push_incomplete_frame_log(
+        &mut self,
+        event_time_zero: u64,
+        digitisers_present: Vec<DigitizerId>,
+        nexus_settings: &NexusSettings,
+    ) -> anyhow::Result<()> {
+        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)];
+
+        timestamps.resize(current_size + 1)?;
+        timestamps.write_slice(&[event_time_zero], next_slice)?;
+
+        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)?;
+
+        Ok(())
+    }
 }

nexus-writer/src/nexus/run.rs

@@ -130,10 +130,11 @@ impl Run {
         &mut self,
         local_path: Option<&Path>,
         message: &FrameAssembledEventListMessage,
+        nexus_settings: &NexusSettings,
     ) -> anyhow::Result<()> {
         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)?;
+            hdf5.push_message_to_runfile(message, nexus_settings)?;
             hdf5.close()?;
         }