From 4ee9bb3e5d3c41339f0935435b0eea78f0674fd8 Mon Sep 17 00:00:00 2001 From: Modularius Date: Tue, 28 Nov 2023 14:05:10 +0000 Subject: [PATCH] Trace Reader (#79) * Trace Reader Branch * Enabled Multiple Readings * Ran treefmt * Updated CI * Updated gitignore * Update Gitignore * gitignore * Applied Requested Changes * Fixed clippy with --all_targets * Fixed cargo.toml * Updated changes * Fixed Readme * Made Requested Changes * Updated CI.yml * Resolve Conflict * Fixed ci.yml * Made suggested changes * Modified clap * Stylistic and main return changes * removed anyhow --- .github/workflows/ci.yml | 4 +- .gitignore | 4 +- Cargo.lock | 16 ++ Cargo.toml | 1 + flake.nix | 1 + trace-reader/Cargo.toml | 18 ++ trace-reader/Readme.md | 28 ++++ trace-reader/default.nix | 42 +++++ trace-reader/src/loader.rs | 298 +++++++++++++++++++++++++++++++++ trace-reader/src/main.rs | 103 ++++++++++++ trace-reader/src/processing.rs | 111 ++++++++++++ 11 files changed, 623 insertions(+), 3 deletions(-) create mode 100644 trace-reader/Cargo.toml create mode 100644 trace-reader/Readme.md create mode 100644 trace-reader/default.nix create mode 100644 trace-reader/src/loader.rs create mode 100644 trace-reader/src/main.rs create mode 100644 trace-reader/src/processing.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 883358b0..0bf3aca0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,4 +1,3 @@ ---- name: CI on: @@ -50,6 +49,7 @@ jobs: - simulator - stream-to-file - trace-archiver + - trace-reader - trace-to-events steps: @@ -86,4 +86,4 @@ jobs: # If the trigger was a tag (i.e. a release) if [[ "${{ github.ref_type }}" == 'tag' ]]; then skopeo copy --dest-creds="$remote_cr_creds" "$local_cr" "$remote_cr:latest" - fi + fi \ No newline at end of file diff --git a/.gitignore b/.gitignore index 80f693be..9bb609b5 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,6 @@ target .ipynb_checkpoints/ .direnv/ .vscode/ -result/ +result +*/Saves +.env \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index d81c949d..33d35dbd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1589,6 +1589,22 @@ dependencies = [ "tokio", ] +[[package]] +name = "trace-reader" +version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "clap", + "common", + "env_logger", + "log", + "rand", + "rdkafka", + "streaming-types", + "tokio", +] + [[package]] name = "trace-to-events" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 9aeb13fd..20a44fef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "stream-to-file", "streaming-types", "trace-archiver", + "trace-reader", "trace-to-events", ] diff --git a/flake.nix b/flake.nix index e25f2d80..a41e988e 100644 --- a/flake.nix +++ b/flake.nix @@ -90,6 +90,7 @@ // import ./simulator {inherit pkgs naersk' version git_revision nativeBuildInputs buildInputs;} // import ./stream-to-file {inherit pkgs naersk' version git_revision nativeBuildInputs buildInputs hdf5-joined;} // import ./trace-archiver {inherit pkgs naersk' version git_revision nativeBuildInputs buildInputs hdf5-joined;} + // import ./trace-reader {inherit pkgs naersk' version git_revision nativeBuildInputs buildInputs;} // import ./trace-to-events {inherit pkgs naersk' version git_revision nativeBuildInputs buildInputs;}; } ); diff --git a/trace-reader/Cargo.toml b/trace-reader/Cargo.toml new file mode 100644 index 00000000..70f5471a --- /dev/null +++ b/trace-reader/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "trace-reader" +version.workspace = true +license.workspace = true +edition.workspace = true + +[dependencies] +common = { path = "../common" } +streaming-types = { path = "../streaming-types" } + +anyhow.workspace = true +clap.workspace = true +rdkafka.workspace = true +tokio.workspace = true +chrono = "0.4.22" +env_logger = "0.10" +log = "0.4" +rand = "0.8.5" diff --git a/trace-reader/Readme.md b/trace-reader/Readme.md new file mode 100644 index 00000000..f08fa161 --- /dev/null +++ b/trace-reader/Readme.md @@ -0,0 +1,28 @@ +# trace-reader + +## Introduction +This program reads picscope .trace files, a binary file developed by E.M Schooneveld. + +## Command Line Interface +For command line instructions run +``` +trace-reader --help +``` + +### Options +The `number-of-trace-events` parameter is the number of traces events that are extracted from the file (either randomly or in sequence). The file defines how many channels the digitizer has, each trace event contains one trace for each channel. The number of messages dispatched is equal to the number of trace events times the number of channels. + +If `random-sample` is set then trace-events are read from the file randomly. Selection is made with replacement so duplication is possible. +If this flag is not set then trace-events are read in order. +If `number-of-trace-events` is greater than the number available then trace-events are the reader wraps around to the beginning of the file as often as necessary. + +### Example +The following reads 18 trace events (sampled randomly with replacement) from `Traces/MuSR_A41_B42_C43_D44_Apr2021_Ag_ZF_IntDeg_Slit60_short.trace` and dispatches them to the topic `Traces` on the Kafka broker located at `localhost:19092`: +``` +trace-reader --broker localhost:19092 --consumer-group trace-producer --trace-topic Traces --file-name Traces/MuSR_A41_B42_C43_D44_Apr2021_Ag_ZF_IntDeg_Slit60_short.traces --number-of-trace-events 18 --random-sample +``` + +## Terminology +- Trace: This is continous block of voltage readings from a digitizer channel. +- Trace Event: This is a collection of traces, one for each channel on the digitizer. +Note that "Event" here is meant in a different sense than the trace-to-event tool. The overlap is a result of terminology used in the .trace file. To avoid confusion, the term "Trace Event" is used here. \ No newline at end of file diff --git a/trace-reader/default.nix b/trace-reader/default.nix new file mode 100644 index 00000000..f666202b --- /dev/null +++ b/trace-reader/default.nix @@ -0,0 +1,42 @@ +{ + pkgs, + naersk', + version, + git_revision, + nativeBuildInputs, + buildInputs, +}: rec { + trace-reader = naersk'.buildPackage { + name = "trace-reader"; + version = version; + + src = ./..; + cargoBuildOptions = x: x ++ ["--package" "trace-reader"]; + + nativeBuildInputs = nativeBuildInputs; + buildInputs = buildInputs; + + overrideMain = p: { + GIT_REVISION = git_revision; + }; + }; + + trace-reader-container-image = pkgs.dockerTools.buildImage { + name = "supermusr-trace-reader"; + tag = "latest"; + created = "now"; + + copyToRoot = pkgs.buildEnv { + name = "image-root"; + paths = with pkgs; [bashInteractive coreutils]; + pathsToLink = ["/bin"]; + }; + + config = { + Entrypoint = ["${pkgs.tini}/bin/tini" "--" "${trace-reader}/bin/trace-reader"]; + Env = [ + "SSL_CERT_FILE=${pkgs.cacert}/etc/ssl/certs/ca-bundle.crt" + ]; + }; + }; +} diff --git a/trace-reader/src/loader.rs b/trace-reader/src/loader.rs new file mode 100644 index 00000000..70c63945 --- /dev/null +++ b/trace-reader/src/loader.rs @@ -0,0 +1,298 @@ +use std::{ + fmt::Debug, + fs::File, + io::{Error, ErrorKind, Read, Seek, SeekFrom}, + mem::size_of, + path::PathBuf, + usize, +}; + +#[derive(Default, Debug)] +pub(crate) struct TraceFileHeader { + pub(crate) prog_version: String, + pub(crate) run_descript: String, + pub(crate) _resolution: i32, + pub(crate) number_of_channels: i32, + pub(crate) _channel_enabled: Vec, + pub(crate) _volts_scale_factor: Vec, + pub(crate) _channel_offset_volts: Vec, + pub(crate) _sample_time: f64, + pub(crate) number_of_samples: i32, + pub(crate) _trigger_enabled: Vec, + pub(crate) _ex_trigger_enabled: bool, + pub(crate) _trigger_level: Vec, + pub(crate) _ex_trigger_level: f64, + pub(crate) _trigger_slope: Vec, + pub(crate) _ex_trigger_slope: i32, + total_bytes: usize, +} + +impl TraceFileHeader { + pub(crate) fn load(file: &mut File) -> Result { + let mut total_bytes = usize::default(); + let prog_version = load_string(file, &mut total_bytes)?; + let run_descript = load_string(file, &mut total_bytes)?; + let _resolution = load_i32(file, &mut total_bytes)?; + let number_of_channels = load_i32(file, &mut total_bytes)?; + Ok(TraceFileHeader { + prog_version, + run_descript, + _resolution, + number_of_channels, + _channel_enabled: load_bool_vec(file, number_of_channels as usize, &mut total_bytes)?, + _volts_scale_factor: load_f64_vec(file, number_of_channels as usize, &mut total_bytes)?, + _channel_offset_volts: load_f64_vec( + file, + number_of_channels as usize, + &mut total_bytes, + )?, + _sample_time: load_f64(file, &mut total_bytes)?, + number_of_samples: load_i32(file, &mut total_bytes)?, + _trigger_enabled: load_bool_vec(file, number_of_channels as usize, &mut total_bytes)?, + _ex_trigger_enabled: load_bool(file, &mut total_bytes)?, + _trigger_level: load_f64_vec(file, number_of_channels as usize, &mut total_bytes)?, + _ex_trigger_level: load_f64(file, &mut total_bytes)?, + _trigger_slope: load_i32_vec(file, number_of_channels as usize, &mut total_bytes)?, + _ex_trigger_slope: load_i32(file, &mut total_bytes)?, + total_bytes, + }) + } +} + +impl TraceFileHeader { + fn get_total_bytes(&self) -> usize { + self.total_bytes + } + + fn get_size(&self) -> usize { + size_of::() + self.prog_version.len() + //pub prog_version : String, + size_of::() + self.run_descript.len() + //pub run_descript : String, + size_of::() + //pub resolution : i32, + size_of::() + //pub number_of_channels : i32, + size_of::()*self.number_of_channels as usize +//pub channel_enabled : Vec, + size_of::()*self.number_of_channels as usize + //pub volts_scale_factor : Vec, + size_of::()*self.number_of_channels as usize + //pub channel_offset_volts : Vec, + size_of::() + //pub sample_time : f64, + size_of::() + //pub number_of_samples : i32, + size_of::()*self.number_of_channels as usize + //pub trigger_enabled : Vec, + size_of::() + //pub ex_trigger_enabled : bool, + size_of::()*self.number_of_channels as usize + //pub trigger_level : Vec, + size_of::() + //pub ex_trigger_level : f64, + size_of::()*self.number_of_channels as usize + //pub trigger_slope : Vec, + size_of::() //pub ex_trigger_slope : i32, + } + + fn get_event_size(&self) -> usize { + TraceFileEvent::get_size( + self.number_of_channels as usize, + self.number_of_samples as usize, + ) + } + + fn get_trace_event(&self, file: &mut File) -> Result { + TraceFileEvent::load_raw( + file, + self.number_of_channels as usize, + self.number_of_samples as usize, + ) + } +} + +#[derive(Default, Debug)] +pub(crate) struct TraceFileEvent { + pub(crate) cur_trace_event: i32, + pub(crate) trace_event_runtime: f64, + pub(crate) number_saved_traces: i32, + pub(crate) saved_channels: Vec, + pub(crate) trigger_time: f64, + pub(crate) _trace: Vec>, + pub(crate) raw_trace: Vec>, + total_bytes: usize, +} + +impl TraceFileEvent { + fn get_size(num_channels: usize, num_samples: usize) -> usize { + size_of::() + //pub cur_trace_event : i32, + size_of::() + //pub trace_event_runtime : f64, + size_of::() + //pub number_saved_traces : i32, + size_of::()*num_channels + //pub saved_channels : Vec, + size_of::() + //pub trigger_time : f64, + size_of::()*num_channels*num_samples //pub raw_trace : Vec>, + } + + pub(crate) fn load(file: &mut File, num_channels: usize) -> Result { + let mut total_bytes = usize::default(); + Ok(TraceFileEvent { + cur_trace_event: load_i32(file, &mut total_bytes)?, + trace_event_runtime: load_f64(file, &mut total_bytes)?, + number_saved_traces: load_i32(file, &mut total_bytes)?, + saved_channels: load_bool_vec(file, num_channels, &mut total_bytes)?, + trigger_time: load_f64(file, &mut total_bytes)?, + total_bytes, + ..Default::default() + }) + } + + pub(crate) fn load_raw( + file: &mut File, + num_channels: usize, + num_samples: usize, + ) -> Result { + let mut total_bytes = usize::default(); + let trace_event = Self::load(file, num_channels)?; + Ok(TraceFileEvent { + cur_trace_event: trace_event.cur_trace_event, + trace_event_runtime: trace_event.trace_event_runtime, + number_saved_traces: trace_event.number_saved_traces, + saved_channels: trace_event.saved_channels, + trigger_time: trace_event.trigger_time, + raw_trace: (0..num_channels) + .map(|_| load_raw_trace(file, num_samples, &mut total_bytes)) + .collect::>()?, + total_bytes: trace_event.total_bytes + total_bytes, + ..Default::default() + }) + } +} + +#[derive(Debug)] +pub(crate) struct TraceFile { + file: File, + header: TraceFileHeader, + num_trace_events: usize, +} + +impl TraceFile { + pub(crate) fn get_trace_event(&mut self, event: usize) -> Result { + if event < self.num_trace_events { + self.file.seek(SeekFrom::Start( + (self.header.get_size() + event * self.header.get_event_size()) as u64, + ))?; + self.header.get_trace_event(&mut self.file) + } else { + Err(Error::new( + ErrorKind::InvalidInput, + "Invalid event index: {event} should be less than {num_events}", + )) + } + } + + pub(crate) fn get_number_of_trace_events(&self) -> usize { + self.num_trace_events + } + + pub(crate) fn get_num_channels(&self) -> usize { + self.header.number_of_channels as usize + } + + pub(crate) fn get_num_samples(&self) -> usize { + self.header.number_of_samples as usize + } +} + +pub(crate) fn load_trace_file(name: PathBuf) -> Result { + let mut file = File::open(name)?; + let header: TraceFileHeader = TraceFileHeader::load(&mut file)?; + let file_size = file + .metadata() + .map_err(|e| Error::new(ErrorKind::InvalidInput, e))? + .len() as usize; + let size_minus_header = file_size - header.get_total_bytes(); + let trace_event_size = header.get_event_size(); + if size_minus_header % trace_event_size != 0 { + Err(Error::new( + ErrorKind::Other, + format!("Problem: {0} != 0", size_minus_header % trace_event_size), + )) + } else { + Ok(TraceFile { + file, + header, + num_trace_events: size_minus_header / trace_event_size, + }) + } +} + +fn load_scalar( + file: &mut File, + bytes: &mut [u8], + total_bytes: &mut usize, +) -> Result<(), Error> { + let num_bytes = file.read(bytes)?; + *total_bytes += num_bytes; + if num_bytes == B { + Ok(()) + } else { + Err(Error::new( + ErrorKind::UnexpectedEof, + format!("Expected {B} bytes, got {num_bytes}."), + )) + } +} + +pub(crate) fn load_i32(file: &mut File, total_bytes: &mut usize) -> Result { + let mut bytes = i32::to_le_bytes(0); + load_scalar::<4>(file, &mut bytes, total_bytes)?; + Ok(i32::from_le_bytes(bytes)) +} + +pub(crate) fn load_f64(file: &mut File, total_bytes: &mut usize) -> Result { + let mut bytes = f64::to_le_bytes(0.); + load_scalar::<8>(file, &mut bytes, total_bytes)?; + Ok(f64::from_le_bytes(bytes)) +} + +pub(crate) fn load_bool(file: &mut File, total_bytes: &mut usize) -> Result { + let mut bytes = u8::to_le_bytes(0); + load_scalar::<1>(file, &mut bytes, total_bytes)?; + Ok(u8::from_le_bytes(bytes) != 0) +} + +pub(crate) fn load_bool_vec( + file: &mut File, + size: usize, + total_bytes: &mut usize, +) -> Result, Error> { + (0..size).map(|_| load_bool(file, total_bytes)).collect() +} + +pub(crate) fn load_f64_vec( + file: &mut File, + size: usize, + total_bytes: &mut usize, +) -> Result, Error> { + (0..size).map(|_| load_f64(file, total_bytes)).collect() +} + +pub(crate) fn load_i32_vec( + file: &mut File, + size: usize, + total_bytes: &mut usize, +) -> Result, Error> { + (0..size).map(|_| load_i32(file, total_bytes)).collect() +} + +pub(crate) fn load_string(file: &mut File, total_bytes: &mut usize) -> Result { + let size = load_i32(file, total_bytes)?; + *total_bytes += size as usize; + let mut string_bytes = Vec::::new(); + string_bytes.resize(size as usize, 0); + file.read_exact(&mut string_bytes)?; + String::from_utf8(string_bytes).map_err(|e| Error::new(ErrorKind::InvalidData, e)) +} + +pub(crate) fn load_raw_trace( + file: &mut File, + size: usize, + total_bytes: &mut usize, +) -> Result, Error> { + let mut trace_bytes = Vec::::new(); + let bytes = (u16::BITS / u8::BITS) as usize * size; + *total_bytes += bytes; + + trace_bytes.resize(bytes, 0); + file.read_exact(&mut trace_bytes)?; + Ok((0..size) + .map(|i| u16::from_be_bytes([trace_bytes[2 * i], trace_bytes[2 * i + 1]])) + .collect()) +} diff --git a/trace-reader/src/main.rs b/trace-reader/src/main.rs new file mode 100644 index 00000000..704be2e3 --- /dev/null +++ b/trace-reader/src/main.rs @@ -0,0 +1,103 @@ +use clap::Parser; +use common::{DigitizerId, FrameNumber}; +use rand::{seq::IteratorRandom, thread_rng}; +use rdkafka::producer::FutureProducer; +use std::path::PathBuf; + +mod loader; +mod processing; +use loader::load_trace_file; +use processing::dispatch_trace_file; + +#[derive(Debug, Parser)] +#[clap(author, version, about)] +struct Cli { + /// Kafka message broker, should have format `host:port`, e.g. `localhost:19092` + #[clap(long)] + broker: String, + + /// Optional Kafka username + #[clap(long)] + username: Option, + + /// Optional Kafka password + #[clap(long)] + password: Option, + + /// Name of the Kafka consumer group + #[clap(long)] + consumer_group: String, + + /// The Kafka topic that trace messages are produced to + #[clap(long)] + trace_topic: String, + + /// Relative path to the .trace file to be read + #[clap(long)] + file_name: PathBuf, + + /// The frame number to assign the message + #[clap(long, default_value = "0")] + frame_number: FrameNumber, + + /// The digitizer id to assign the message + #[clap(long, default_value = "0")] + digitizer_id: DigitizerId, + + /// The number of trace events to read. If zero, then all trace events are read + #[clap(long, default_value = "1")] + number_of_trace_events: usize, + + /// If set, then trace events are sampled randomly with replacement, if not set then trace events are read in order + #[clap(long, default_value = "false")] + random_sample: bool, +} + +#[tokio::main] +async fn main() { + env_logger::init(); + + let args = Cli::parse(); + + let client_config = + common::generate_kafka_client_config(&args.broker, &args.username, &args.password); + + let producer: FutureProducer = client_config + .create() + .expect("Kafka Producer should be created"); + + let trace_file = load_trace_file(args.file_name).expect("Trace File should load"); + let total_trace_events = trace_file.get_number_of_trace_events(); + let num_trace_events = if args.number_of_trace_events == 0 { + total_trace_events + } else { + args.number_of_trace_events + }; + + let trace_event_indices: Vec<_> = if args.random_sample { + (0..num_trace_events) + .map(|_| { + (0..num_trace_events) + .choose(&mut thread_rng()) + .unwrap_or_default() + }) + .collect() + } else { + (0..num_trace_events) + .cycle() + .take(num_trace_events) + .collect() + }; + + dispatch_trace_file( + trace_file, + trace_event_indices, + args.frame_number, + args.digitizer_id, + &producer, + &args.trace_topic, + 6000, + ) + .await + .expect("Trace File should be dispatched to Kafka"); +} diff --git a/trace-reader/src/processing.rs b/trace-reader/src/processing.rs new file mode 100644 index 00000000..230b0d70 --- /dev/null +++ b/trace-reader/src/processing.rs @@ -0,0 +1,111 @@ +//! This module allows one to simulate instances of DigitizerAnalogTraceMessage +//! using the FlatBufferBuilder. + +use super::loader::{TraceFile, TraceFileEvent}; +use anyhow::{Error, Result}; +use chrono::Utc; +use log::{debug, error}; +use rdkafka::{ + producer::{FutureProducer, FutureRecord}, + util::Timeout, +}; +use std::time::Duration; + +use common::{Channel, DigitizerId, FrameNumber, Intensity}; +use streaming_types::{ + dat1_digitizer_analog_trace_v1_generated::{ + finish_digitizer_analog_trace_message_buffer, ChannelTrace, ChannelTraceArgs, + DigitizerAnalogTraceMessage, DigitizerAnalogTraceMessageArgs, + }, + flatbuffers::{FlatBufferBuilder, WIPOffset}, + frame_metadata_v1_generated::{FrameMetadataV1, FrameMetadataV1Args, GpsTime}, +}; + +/// Reads the contents of trace_file and dispatches messages to the given Kafka topic. +pub(crate) async fn dispatch_trace_file( + mut trace_file: TraceFile, + trace_event_indices: Vec, + frame_number: FrameNumber, + digitizer_id: DigitizerId, + producer: &FutureProducer, + topic: &str, + timeout_ms: u64, +) -> Result<()> { + let mut fbb = FlatBufferBuilder::new(); + for index in trace_event_indices { + let event = trace_file.get_trace_event(index)?; + create_message( + &mut fbb, + Utc::now().into(), + frame_number, + digitizer_id, + trace_file.get_num_channels(), + trace_file.get_num_samples(), + &event, + )?; + + let future_record = FutureRecord::to(topic).payload(fbb.finished_data()).key(""); + let timeout = Timeout::After(Duration::from_millis(timeout_ms)); + match producer.send(future_record, timeout).await { + Ok(r) => debug!("Delivery: {:?}", r), + Err(e) => error!("Delivery failed: {:?}", e.0), + }; + } + Ok(()) +} + +pub(crate) fn create_channel<'a>( + fbb: &mut FlatBufferBuilder<'a>, + channel: Channel, + trace: &[Intensity], +) -> WIPOffset> { + let voltage = Some(fbb.create_vector::(trace)); + ChannelTrace::create(fbb, &ChannelTraceArgs { channel, voltage }) +} + +/// Loads a FlatBufferBuilder with a new DigitizerAnalogTraceMessage instance with a custom timestamp. +/// #Arguments +/// * `fbb` - A mutable reference to the FlatBufferBuilder to use. +/// * `time` - A `frame_metadata_v1_generated::GpsTime` instance containing the timestamp. +/// * `frame_number` - The frame number to use. +/// * `digitizer_id` - The id of the digitizer to use. +/// * `measurements_per_frame` - The number of measurements to simulate in each channel. +/// * `num_channels` - The number of channels to simulate. +/// #Returns +/// A string result, or an error. +pub(crate) fn create_message( + fbb: &mut FlatBufferBuilder<'_>, + time: GpsTime, + frame_number: u32, + digitizer_id: u8, + number_of_channels: usize, + number_of_samples: usize, + event: &TraceFileEvent, +) -> Result { + fbb.reset(); + + let metadata: FrameMetadataV1Args = FrameMetadataV1Args { + frame_number, + period_number: 0, + protons_per_pulse: 0, + running: true, + timestamp: Some(&time), + veto_flags: 0, + }; + let metadata: WIPOffset = FrameMetadataV1::create(fbb, &metadata); + + let channels: Vec<_> = (0..number_of_channels) + .map(|c| create_channel(fbb, c as u32, event.raw_trace[c].as_slice())) + .collect(); + + let message = DigitizerAnalogTraceMessageArgs { + digitizer_id, + metadata: Some(metadata), + sample_rate: 1_000_000_000, + channels: Some(fbb.create_vector_from_iter(channels.iter())), + }; + let message = DigitizerAnalogTraceMessage::create(fbb, &message); + finish_digitizer_analog_trace_message_buffer(fbb, message); + + Ok(format!("New message created for digitizer {digitizer_id}, frame number {frame_number}, and has {number_of_channels} channels, and {number_of_samples} measurements.")) +}