fls2

Cargo.toml

@@ -14,24 +14,24 @@ repository = "https://github.com/quartiq/stabilizer-streaming"
 clap = { version = "4.3", features = ["derive"] }
 num_enum = "0.7"
 log = "0.4"
-eframe = { version = "0.33", default-features = false, features = [
+eframe = { version = "0.34", default-features = false, features = [
     "wayland",
     "x11",
     "glow",
     "default_fonts",
     "persistence",
 ] }
-egui_plot = "0.34"
+egui_plot = "0.35"
 env_logger = "0.11"
 bytemuck = { version = "1.13.1", features = [
     "latest_stable_rust",
     "extern_crate_std",
 ] }
 thiserror = "2"
 anyhow = "1.0.75"
-socket2 = "0.6.0"
-idsp = "0.20"
-dsp-process = "0.1.0"
+socket2 = { version = "0.6.0", features = [ "all" ] }
+idsp = "0.21"
+dsp-process = "0.2.0"
 rustfft = "6.1.0"
 rand = "0.10.0"
 derive_builder = "0.20.0"

src/bin/psd.rs

@@ -123,36 +123,40 @@ struct Trace {
 }
 
 impl Trace {
-    fn plot(&self, acq: &AcqOpts) -> (f32, Vec<[f64; 2]>) {
+    fn plot(&self, acq: &AcqOpts) -> (f32, f32, Vec<[f64; 2]>) {
         let logfs = acq.fs.log10();
         let mut p0 = Trapezoidal::default();
         let mut pi = 0.0;
-        let plot = self
+        let points = self
             .psd
             .iter()
             .zip(self.frequencies.iter())
-            .filter_map(|(&p, &f)| {
+            .skip(1) // DC
+            .map(|(&p, &f)| {
                 // TODO: check at stage breaks
                 let dp = p0.push(f, p);
                 if (acq.integral_start..=acq.integral_end).contains(&(acq.fs * f)) {
                     // TODO: correctly interpolate at range limits
                     pi += dp;
                 }
-                if f.is_normal() {
-                    Some([
-                        (f.log10() + logfs) as f64,
-                        (if acq.integrate {
-                            p0.get().sqrt()
-                        } else {
-                            10.0 * (p.log10() - logfs)
-                        }) as f64,
-                    ])
-                } else {
-                    None
-                }
+                [
+                    (f.log10() + logfs) as f64,
+                    (if acq.integrate {
+                        p0.get().sqrt()
+                    } else {
+                        10.0 * (p.log10() - logfs)
+                    }) as f64,
+                ]
             })
             .collect();
-        (pi.sqrt(), plot)
+        debug_assert_eq!(self.frequencies.get(0).copied().unwrap_or_default(), 0.0);
+        let (p0, f1) = self
+            .psd
+            .get(0)
+            .copied()
+            .zip(self.frequencies.get(1).copied())
+            .unwrap_or_default();
+        ((p0 * f1).sqrt(), pi.sqrt(), points)
     }
 }
 
@@ -221,15 +225,14 @@ fn main() -> Result<()> {
         Ok(())
     });
 
-    let options = eframe::NativeOptions {
-        viewport: egui::ViewportBuilder::default()
-            .with_inner_size([1000.0, 700.0])
-            .with_drag_and_drop(true),
-        ..Default::default()
-    };
     eframe::run_native(
         "PSD",
-        options,
+        eframe::NativeOptions {
+            viewport: egui::ViewportBuilder::default()
+                .with_inner_size([1000.0, 700.0])
+                .with_drag_and_drop(true),
+            ..Default::default()
+        },
         Box::new(move |cc| {
             cc.egui_ctx.set_visuals(egui::Visuals::light());
             Ok(Box::new(App::new(trace_recv, cmd_send, acq)))
@@ -271,29 +274,29 @@ impl eframe::App for App {
         self.cmd_send.send(Cmd::Exit).unwrap();
     }
 
-    fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) {
+    fn ui(&mut self, ui: &mut egui::Ui, _frame: &mut eframe::Frame) {
         match self.trace_recv.try_recv() {
             Err(mpsc::TryRecvError::Empty) => {}
             Ok(trace) => {
                 if !self.hold {
                     self.current = trace
                         .into_iter()
                         .map(|t| {
-                            let (pi, xy) = t.plot(&self.acq);
-                            (format!("{}: {pi:.2e}", t.name), xy, t.breaks)
+                            let (pdf, pi, xy) = t.plot(&self.acq);
+                            (format!("{}: {pdf:.2e}±{pi:.2e}", t.name,), xy, t.breaks)
                         })
                         .collect();
-                    ctx.request_repaint_after(Duration::from_secs_f32(self.repaint));
+                    ui.request_repaint_after(Duration::from_secs_f32(self.repaint));
                 }
             }
             Err(mpsc::TryRecvError::Disconnected) => panic!("lost data processing thread"),
         };
 
-        egui::SidePanel::left("left panel")
+        egui::Panel::left("left panel")
             .resizable(true)
-            .default_width(250.0)
-            .show(ctx, |ui| self.settings(ui));
-        egui::CentralPanel::default().show(ctx, |ui| self.plot(ui));
+            .default_size(250.0)
+            .show_inside(ui, |ui| self.settings(ui));
+        egui::CentralPanel::default().show_inside(ui, |ui| self.plot(ui));
         self.cmd_send.send(Cmd::Send(self.acq)).unwrap();
     }
 }
@@ -390,7 +393,7 @@ impl App {
                         .style(LineStyle::dashed_loose()),
                 );
                 for (name, trace, _) in self.current.iter() {
-                    plot_ui.line(Line::new(name, PlotPoints::from(trace.clone())).name(name));
+                    plot_ui.line(Line::new(name, PlotPoints::new(trace.clone())).name(name));
                 }
             });
     }

src/de/data.rs

@@ -210,3 +210,29 @@ impl<'a> Payload<'a> for Mpll<'a> {
         ])
     }
 }
+
+#[derive(Clone, Debug, PartialEq, PartialOrd)]
+pub struct Fls2<'a> {
+    data: &'a [[[u8; 4]; 1 + 2 * (2 + 2 + 1 + 1)]],
+}
+
+impl<'a> Payload<'a> for Fls2<'a> {
+    fn new(batches: usize, data: &'a [u8]) -> Result<Self, Error> {
+        let data = bytemuck::try_cast_slice(data)?;
+        // (pll, [(demod_re, demod_im, phase[2], ftw, pow_amp); 2])
+        assert_eq!(batches, data.len());
+        Ok(Self { data })
+    }
+
+    fn traces(&self) -> Result<Vec<(&'static str, Vec<f32>)>, Error> {
+        Ok(vec![(
+            "AP",
+            self.data
+                .iter()
+                .map(|b| {
+                    i32::from_le_bytes(b[3]) as f32 * (core::f32::consts::TAU / (5i64 << 32) as f32)
+                })
+                .collect(),
+        )])
+    }
+}

src/de/frame.rs

@@ -55,6 +55,7 @@ impl<'a> Frame<'a> {
             Format::Fls => Box::new(data::Fls::new(batches, data)?),
             Format::ThermostatEem => Box::new(data::ThermostatEem::new(batches, data)?),
             Format::Mpll => Box::new(data::Mpll::new(header.batches as _, data)?),
+            Format::Fls2 => Box::new(data::Fls2::new(batches, data)?),
         };
         Ok(Self { header, payload })
     }

src/de/mod.rs

@@ -14,6 +14,7 @@ pub enum Format {
     Fls = 2,
     ThermostatEem = 3,
     Mpll = 4,
+    Fls2 = 5,
 }
 
 #[derive(Debug, Clone, Error)]

src/psd.rs

@@ -454,8 +454,8 @@ impl<const N: usize> PsdCascade<N> {
 
     /// Process input items
     pub fn process(&mut self, x: &[f32]) {
-        let mut a = ([0f32; N], [0f32; N]);
-        let (mut y, mut z) = (&mut a.0, &mut a.1);
+        let mut a = [[0f32; N]; 2];
+        let [mut y, mut z] = a.each_mut();
         for mut x in x.chunks(N << DEPTH) {
             let mut i = 0;
             while !x.is_empty() {
@@ -519,6 +519,7 @@ impl<const N: usize> PsdCascade<N> {
                 end = start;
             }
         }
+        debug_assert_eq!(decimation, 1);
         // Do not "correct" DC and Nyquist bins.
         // Common psd algorithms argue that as both only contribute once to the one-sided
         // spectrum, they should be scaled by 0.5.

src/source.rs

@@ -82,7 +82,9 @@ impl Source {
             socket.set_read_timeout(Some(Duration::from_millis(1000)))?;
             socket.set_recv_buffer_size(1 << 20)?;
             socket.set_reuse_address(true)?;
+            socket.set_reuse_port(true)?;
             if opts.ip.is_multicast() {
+                socket.set_multicast_all_v4(false)?;
                 socket.join_multicast_v4(&opts.ip, &Ipv4Addr::UNSPECIFIED)?;
             }
             #[cfg(windows)]