dspk_view/
signal.rs

1//! A signal as a viewer sees it ([`SignalBackend`]), wherever it lives: [`LocalSignal`] is a
2//! recording in this process with its pyramid and background builder; a remote session
3//! (dspk-stream) answers the same [`View`]s over the network. A viewer (an app) only asks for views
4//! and draws the [`Envelope`]s it gets.
5
6use std::ops::Range;
7use std::path::Path;
8use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
9use std::sync::{Arc, Mutex};
10
11use dspk_core::{DspResult, RecordingInfo, RecordingSource};
12
13use crate::pyramid::{pyramid_path, OnProgress, Progress, Pyramid, PyramidBuilder, PyramidIdentity, FILE_BASE, MEMORY_BASE};
14use crate::view::{Envelope, View, RAW_BLOCK_VALUES};
15
16/// Recordings with at least this many stored bytes keep their pyramid in a file next to them
17/// (built once, nearest the view first, and reused by every later open); smaller ones keep it in
18/// memory for the session.
19pub const FILE_PYRAMID_MIN_BYTES: u64 = 64 << 20;
20
21/// What a viewer needs from a signal. Calls block; viewers call them from worker threads.
22pub trait SignalBackend: Send + Sync {
23    /// The recording's description (channels, samples, exact rate and start, units).
24    fn info(&self) -> &RecordingInfo;
25
26    /// The contents of `view` into `out` (its buffer is reused). Zoomed-out columns not built yet
27    /// are NaN and `complete` is false: ask again after [`Self::watch`] reports progress.
28    fn view(&self, view: &View, out: &mut Envelope) -> DspResult<()>;
29
30    /// Samples `samples` of `channels`, channel-major (e.g. the value under the cursor).
31    fn read(&self, channels: &[usize], samples: Range<u64>, out: &mut [f32]) -> DspResult<()>;
32
33    /// The viewer looks at sample `sample`: zoomed-out data there is prepared first.
34    fn focus(&self, sample: u64);
35
36    /// Prepares the zoomed-out data of samples `start..end` now, blocking until it is ready (for
37    /// a single exact image, e.g. a snapshot; viewers use [`Self::focus`]).
38    fn prepare(&self, start: u64, end: u64) -> DspResult<()>;
39
40    /// Samples whose zoomed-out data is ready, of all (`done == total` once complete).
41    fn progress(&self) -> Progress;
42
43    /// Calls `on_progress` whenever more zoomed-out data is ready.
44    fn watch(&self, on_progress: OnProgress);
45}
46
47/// A recording in this process, its min/max [`Pyramid`] (a file for recordings of at least
48/// [`FILE_PYRAMID_MIN_BYTES`], else memory) and the builder that fills it in the background,
49/// nearest the focus first. The recording is read once to build the pyramid, while views already
50/// draw from it.
51pub struct LocalSignal {
52    source: Arc<dyn RecordingSource>,
53    pyramid: Arc<Pyramid>,
54    builder: PyramidBuilder,
55    cancel: Arc<AtomicBool>,
56    /// Samples built so far.
57    done: Arc<AtomicU64>,
58    listeners: Arc<Mutex<Vec<OnProgress>>>,
59}
60
61impl LocalSignal {
62    /// `source`, with the pyramid of source `id` of the recording file `path` when given: the
63    /// complete file next to it, else (large recordings) a new file there, else memory.
64    pub fn open(source: Arc<dyn RecordingSource>, recording: Option<(&Path, &str)>) -> DspResult<Self> {
65        let pyramid = Self::choose_pyramid(source.as_ref(), recording)?;
66        let total = source.info().samples;
67        let done = if pyramid.is_complete() { total } else { 0 };
68        Ok(Self {
69            source,
70            pyramid: Arc::new(pyramid),
71            builder: PyramidBuilder::default(),
72            cancel: Arc::new(AtomicBool::new(false)),
73            done: Arc::new(AtomicU64::new(done)),
74            listeners: Arc::new(Mutex::new(Vec::new())),
75        })
76    }
77
78    fn choose_pyramid(source: &dyn RecordingSource, recording: Option<(&Path, &str)>) -> DspResult<Pyramid> {
79        let Some((path, id)) = recording else { return Pyramid::in_memory(source, MEMORY_BASE) };
80        let identity = PyramidIdentity::of(path, id, source)?;
81        let file = pyramid_path(path, id);
82        if let Some(complete) = Pyramid::open_file(&file, &identity, FILE_BASE)? {
83            return Ok(complete);
84        }
85        if source.info().data_bytes() < FILE_PYRAMID_MIN_BYTES {
86            return Pyramid::in_memory(source, MEMORY_BASE);
87        }
88        match Pyramid::create_file(&file, &identity, FILE_BASE) {
89            Ok(pyramid) => Ok(pyramid),
90            Err(e) => {
91                tracing::warn!("cannot write {}: {e}; keeping the min/max pyramid in memory", file.display());
92                Pyramid::in_memory(source, MEMORY_BASE)
93            }
94        }
95    }
96
97    /// The recording.
98    pub fn source(&self) -> &Arc<dyn RecordingSource> {
99        &self.source
100    }
101
102    /// The pyramid (complete or filling).
103    pub fn pyramid(&self) -> &Arc<Pyramid> {
104        &self.pyramid
105    }
106}
107
108impl SignalBackend for LocalSignal {
109    fn info(&self) -> &RecordingInfo {
110        self.source.info()
111    }
112
113    fn view(&self, view: &View, out: &mut Envelope) -> DspResult<()> {
114        view.read_into(self.source.as_ref(), Some(&self.pyramid), out)
115    }
116
117    fn read(&self, channels: &[usize], samples: Range<u64>, out: &mut [f32]) -> DspResult<()> {
118        self.source.read(channels, samples, out)
119    }
120
121    fn focus(&self, sample: u64) {
122        if self.pyramid.is_complete() {
123            return;
124        }
125        let (done, listeners) = (self.done.clone(), self.listeners.clone());
126        let on_progress: OnProgress = Arc::new(move |p: Progress| {
127            done.store(p.done, Ordering::Relaxed);
128            for listener in listeners.lock().expect("listeners lock").iter() {
129                listener(p.clone());
130            }
131        });
132        // Starts once; later calls only move the focus
133        self.builder.run(self.source.clone(), self.pyramid.clone(), sample, self.cancel.clone(), on_progress);
134    }
135
136    fn prepare(&self, start: u64, end: u64) -> DspResult<()> {
137        let end = end.min(self.source.info().samples);
138        if start < end {
139            self.pyramid.fill(self.source.as_ref(), start, end, RAW_BLOCK_VALUES, || false)?;
140        }
141        Ok(())
142    }
143
144    fn progress(&self) -> Progress {
145        Progress { filled: 0..0, done: self.done.load(Ordering::Relaxed), total: self.source.info().samples }
146    }
147
148    fn watch(&self, on_progress: OnProgress) {
149        self.listeners.lock().expect("listeners lock").push(on_progress);
150    }
151}
152
153impl Drop for LocalSignal {
154    fn drop(&mut self) {
155        self.cancel.store(true, Ordering::Relaxed);
156    }
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162    use dspk_core::MemoryRecording;
163
164    #[test]
165    fn small_recordings_keep_the_pyramid_in_memory_and_fill_it() {
166        let data: Vec<f32> = (0..2 * 50_000).map(|i| ((i * 7919) % 101) as f32).collect();
167        let source: Arc<dyn RecordingSource> = Arc::new(MemoryRecording::new("s", data, 2, 1000.0).unwrap());
168        let signal = LocalSignal::open(source, None).unwrap();
169        assert!(signal.pyramid().path().is_none());
170        assert_eq!(signal.progress().done, 0);
171        let (tx, rx) = std::sync::mpsc::channel();
172        signal.watch(Arc::new(move |p: Progress| {
173            if p.done == p.total {
174                let _ = tx.send(());
175            }
176        }));
177        signal.focus(25_000);
178        rx.recv_timeout(std::time::Duration::from_secs(30)).expect("pyramid filled");
179        assert!(signal.pyramid().is_complete());
180        let mut out = Envelope::Samples(Vec::new());
181        signal.view(&View { channels: vec![1], start: 0, end: 50_000, width: 10 }, &mut out).unwrap();
182        assert!(matches!(out, Envelope::Columns { complete: true, .. }));
183    }
184}
dspk_view/signal.rs

dspk_view/
signal.rs
