visual_cortex_vision/
debug.rs

1//! Debug taps for preprocessors: visual insight into what a masking stage
2//! is doing, without ever stalling the watcher. Implementations push frames
3//! through a bounded channel to a writer thread and DROP on lag (the
4//! Recorder discipline); dropped counts are logged and queryable.
5
6use std::path::PathBuf;
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::mpsc::{sync_channel, SyncSender, TrySendError};
9use std::sync::Arc;
10use std::thread::JoinHandle;
11
12use visual_cortex_capture::Frame;
13
14/// A tap receiving per-tick stage frames from a preprocessor (see
15/// [`StabilityMask::debug_sink`](crate::StabilityMask::debug_sink)).
16///
17/// `write` is called on the watcher's tick path — implementations must
18/// return immediately (hand off to a thread, drop on lag), never block.
19pub trait DebugSink: Send + 'static {
20    fn write(&mut self, tick: u64, stage: DebugStage, frame: &Frame);
21
22    /// Receives the watcher name at subscribe time (used e.g. for output
23    /// subdirectories). Default: ignore.
24    fn set_label(&mut self, _label: &str) {}
25}
26
27/// Which pipeline artifact a debug frame shows.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29#[non_exhaustive]
30pub enum DebugStage {
31    /// The raw cropped input.
32    Input,
33    /// The per-block baseline estimate rendered as grayscale.
34    Baseline,
35    /// The human view: input with suppressed blocks dimmed.
36    Overlay,
37    /// Exactly what the rest of the pipeline sees.
38    Output,
39}
40
41impl DebugStage {
42    fn suffix(self) -> &'static str {
43        match self {
44            DebugStage::Input => "input",
45            DebugStage::Baseline => "baseline",
46            DebugStage::Overlay => "overlay",
47            DebugStage::Output => "output",
48        }
49    }
50}
51
52struct Job {
53    tick: u64,
54    stage: DebugStage,
55    label: String,
56    width: u32,
57    height: u32,
58    /// BGRA, cloned off the tick path.
59    data: Vec<u8>,
60}
61
62/// Push a job without ever blocking: a full channel means the writer is
63/// lagging, and debug output loses to the watcher — drop and count.
64fn try_dispatch(tx: &SyncSender<Job>, dropped: &AtomicU64, job: Job) {
65    match tx.try_send(job) {
66        Ok(()) => {}
67        Err(TrySendError::Full(_)) => {
68            let n = dropped.fetch_add(1, Ordering::Relaxed) + 1;
69            tracing::debug!(dropped = n, "PngDump lagging; dropping debug frame");
70        }
71        Err(TrySendError::Disconnected(_)) => {}
72    }
73}
74
75/// PNG stills per sampled tick:
76/// `<dir>/<watcher>/NNNNNN-{input,baseline,overlay,output}.png`.
77///
78/// Encoding and disk I/O happen on a dedicated writer thread behind a
79/// bounded channel; frames are dropped (and counted) if the writer lags.
80pub struct PngDump {
81    tx: Option<SyncSender<Job>>,
82    writer: Option<JoinHandle<()>>,
83    dropped: Arc<AtomicU64>,
84    sample_every: u64,
85    label: String,
86}
87
88impl PngDump {
89    /// Write stills under `dir` (created on demand).
90    pub fn dir(dir: impl Into<PathBuf>) -> Self {
91        let dir = dir.into();
92        let (tx, rx) = sync_channel::<Job>(8);
93        let writer = std::thread::spawn(move || {
94            for job in rx {
95                let sub = dir.join(&job.label);
96                if let Err(e) = std::fs::create_dir_all(&sub) {
97                    tracing::warn!(dir = %sub.display(), error = %e, "PngDump mkdir failed");
98                    continue;
99                }
100                let path = sub.join(format!("{:06}-{}.png", job.tick, job.stage.suffix()));
101                // BGRA -> RGBA for the PNG encoder.
102                let mut rgba = job.data;
103                for px in rgba.chunks_exact_mut(4) {
104                    px.swap(0, 2);
105                }
106                match image::RgbaImage::from_raw(job.width, job.height, rgba) {
107                    Some(img) => {
108                        if let Err(e) = img.save(&path) {
109                            tracing::warn!(path = %path.display(), error = %e, "PngDump write failed");
110                        }
111                    }
112                    None => tracing::warn!("PngDump: frame buffer size mismatch"),
113                }
114            }
115        });
116        Self {
117            tx: Some(tx),
118            writer: Some(writer),
119            dropped: Arc::default(),
120            sample_every: 1,
121            label: "preprocessor".into(),
122        }
123    }
124
125    /// Only write every Nth tick (default 1 = every tick).
126    pub fn sample_every(mut self, n: u64) -> Self {
127        self.sample_every = n.max(1);
128        self
129    }
130
131    /// Frames dropped because the writer thread lagged.
132    pub fn dropped(&self) -> u64 {
133        self.dropped.load(Ordering::Relaxed)
134    }
135}
136
137impl DebugSink for PngDump {
138    fn write(&mut self, tick: u64, stage: DebugStage, frame: &Frame) {
139        if !tick.is_multiple_of(self.sample_every) {
140            return;
141        }
142        let Some(tx) = &self.tx else { return };
143        try_dispatch(
144            tx,
145            &self.dropped,
146            Job {
147                tick,
148                stage,
149                label: self.label.clone(),
150                width: frame.width(),
151                height: frame.height(),
152                data: frame.data().to_vec(),
153            },
154        );
155    }
156
157    fn set_label(&mut self, label: &str) {
158        self.label = label.to_string();
159    }
160}
161
162impl Drop for PngDump {
163    fn drop(&mut self) {
164        drop(self.tx.take()); // close the channel so the writer drains and exits
165        if let Some(w) = self.writer.take() {
166            let _ = w.join();
167        }
168    }
169}
170
171#[cfg(test)]
172mod tests {
173    use super::*;
174
175    #[test]
176    fn full_channel_drops_and_counts_instead_of_blocking() {
177        let (tx, _rx) = sync_channel::<Job>(2); // never drained
178        let dropped = AtomicU64::new(0);
179        for tick in 0..5 {
180            try_dispatch(
181                &tx,
182                &dropped,
183                Job {
184                    tick,
185                    stage: DebugStage::Output,
186                    label: "w".into(),
187                    width: 1,
188                    height: 1,
189                    data: vec![0, 0, 0, 255],
190                },
191            );
192        }
193        assert_eq!(dropped.load(Ordering::Relaxed), 3, "2 buffered, 3 dropped");
194    }
195
196    #[test]
197    fn writes_sampled_stills_under_the_watcher_label() {
198        let tmp = tempfile::tempdir().unwrap();
199        let mut dump = PngDump::dir(tmp.path()).sample_every(2);
200        dump.set_label("tooltip");
201
202        let frame = Frame::solid(2, 2, [10, 20, 30, 255]); // BGRA
203        for tick in 0..3 {
204            dump.write(tick, DebugStage::Output, &frame);
205        }
206        drop(dump); // joins the writer: all accepted jobs are on disk
207
208        let dir = tmp.path().join("tooltip");
209        assert!(dir.join("000000-output.png").exists());
210        assert!(
211            !dir.join("000001-output.png").exists(),
212            "sample_every(2) must skip odd ticks"
213        );
214        assert!(dir.join("000002-output.png").exists());
215
216        // BGRA -> RGBA swizzle correctness.
217        let img = image::open(dir.join("000000-output.png"))
218            .unwrap()
219            .to_rgba8();
220        assert_eq!(img.get_pixel(0, 0).0, [30, 20, 10, 255]);
221    }
222}
visual_cortex_vision/debug.rs

visual_cortex_vision/
debug.rs
