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    /// False-color per-block state (idle/kept/holdover/re-seed) — shows WHY
38    /// each block did what it did.
39    State,
40    /// Exactly what the rest of the pipeline sees.
41    Output,
42}
43
44impl DebugStage {
45    fn suffix(self) -> &'static str {
46        match self {
47            DebugStage::Input => "input",
48            DebugStage::Baseline => "baseline",
49            DebugStage::Overlay => "overlay",
50            DebugStage::State => "state",
51            DebugStage::Output => "output",
52        }
53    }
54}
55
56struct Job {
57    tick: u64,
58    stage: DebugStage,
59    label: String,
60    width: u32,
61    height: u32,
62    /// BGRA, cloned off the tick path.
63    data: Vec<u8>,
64}
65
66/// Push a job without ever blocking: a full channel means the writer is
67/// lagging, and debug output loses to the watcher — drop and count.
68fn try_dispatch(tx: &SyncSender<Job>, dropped: &AtomicU64, job: Job) {
69    match tx.try_send(job) {
70        Ok(()) => {}
71        Err(TrySendError::Full(_)) => {
72            let n = dropped.fetch_add(1, Ordering::Relaxed) + 1;
73            tracing::debug!(dropped = n, "PngDump lagging; dropping debug frame");
74        }
75        Err(TrySendError::Disconnected(_)) => {}
76    }
77}
78
79/// PNG stills per sampled tick:
80/// `<dir>/<watcher>/NNNNNN-{input,baseline,overlay,output}.png`.
81///
82/// Encoding and disk I/O happen on a dedicated writer thread behind a
83/// bounded channel; frames are dropped (and counted) if the writer lags.
84pub struct PngDump {
85    tx: Option<SyncSender<Job>>,
86    writer: Option<JoinHandle<()>>,
87    dropped: Arc<AtomicU64>,
88    sample_every: u64,
89    label: String,
90}
91
92impl PngDump {
93    /// Write stills under `dir` (created on demand).
94    pub fn dir(dir: impl Into<PathBuf>) -> Self {
95        let dir = dir.into();
96        let (tx, rx) = sync_channel::<Job>(8);
97        let writer = std::thread::spawn(move || {
98            for job in rx {
99                let sub = dir.join(&job.label);
100                if let Err(e) = std::fs::create_dir_all(&sub) {
101                    tracing::warn!(dir = %sub.display(), error = %e, "PngDump mkdir failed");
102                    continue;
103                }
104                let path = sub.join(format!("{:06}-{}.png", job.tick, job.stage.suffix()));
105                // BGRA -> RGBA for the PNG encoder.
106                let mut rgba = job.data;
107                for px in rgba.chunks_exact_mut(4) {
108                    px.swap(0, 2);
109                }
110                match image::RgbaImage::from_raw(job.width, job.height, rgba) {
111                    Some(img) => {
112                        if let Err(e) = img.save(&path) {
113                            tracing::warn!(path = %path.display(), error = %e, "PngDump write failed");
114                        }
115                    }
116                    None => tracing::warn!("PngDump: frame buffer size mismatch"),
117                }
118            }
119        });
120        Self {
121            tx: Some(tx),
122            writer: Some(writer),
123            dropped: Arc::default(),
124            sample_every: 1,
125            label: "preprocessor".into(),
126        }
127    }
128
129    /// Only write every Nth tick (default 1 = every tick).
130    pub fn sample_every(mut self, n: u64) -> Self {
131        self.sample_every = n.max(1);
132        self
133    }
134
135    /// Frames dropped because the writer thread lagged.
136    pub fn dropped(&self) -> u64 {
137        self.dropped.load(Ordering::Relaxed)
138    }
139}
140
141impl DebugSink for PngDump {
142    fn write(&mut self, tick: u64, stage: DebugStage, frame: &Frame) {
143        if !tick.is_multiple_of(self.sample_every) {
144            return;
145        }
146        let Some(tx) = &self.tx else { return };
147        try_dispatch(
148            tx,
149            &self.dropped,
150            Job {
151                tick,
152                stage,
153                label: self.label.clone(),
154                width: frame.width(),
155                height: frame.height(),
156                data: frame.data().to_vec(),
157            },
158        );
159    }
160
161    fn set_label(&mut self, label: &str) {
162        self.label = label.to_string();
163    }
164}
165
166impl Drop for PngDump {
167    fn drop(&mut self) {
168        drop(self.tx.take()); // close the channel so the writer drains and exits
169        if let Some(w) = self.writer.take() {
170            let _ = w.join();
171        }
172    }
173}
174
175#[cfg(test)]
176mod tests {
177    use super::*;
178
179    #[test]
180    fn full_channel_drops_and_counts_instead_of_blocking() {
181        let (tx, _rx) = sync_channel::<Job>(2); // never drained
182        let dropped = AtomicU64::new(0);
183        for tick in 0..5 {
184            try_dispatch(
185                &tx,
186                &dropped,
187                Job {
188                    tick,
189                    stage: DebugStage::Output,
190                    label: "w".into(),
191                    width: 1,
192                    height: 1,
193                    data: vec![0, 0, 0, 255],
194                },
195            );
196        }
197        assert_eq!(dropped.load(Ordering::Relaxed), 3, "2 buffered, 3 dropped");
198    }
199
200    #[test]
201    fn writes_sampled_stills_under_the_watcher_label() {
202        let tmp = tempfile::tempdir().unwrap();
203        let mut dump = PngDump::dir(tmp.path()).sample_every(2);
204        dump.set_label("tooltip");
205
206        let frame = Frame::solid(2, 2, [10, 20, 30, 255]); // BGRA
207        for tick in 0..3 {
208            dump.write(tick, DebugStage::Output, &frame);
209        }
210        drop(dump); // joins the writer: all accepted jobs are on disk
211
212        let dir = tmp.path().join("tooltip");
213        assert!(dir.join("000000-output.png").exists());
214        assert!(
215            !dir.join("000001-output.png").exists(),
216            "sample_every(2) must skip odd ticks"
217        );
218        assert!(dir.join("000002-output.png").exists());
219
220        // BGRA -> RGBA swizzle correctness.
221        let img = image::open(dir.join("000000-output.png"))
222            .unwrap()
223            .to_rgba8();
224        assert_eq!(img.get_pixel(0, 0).0, [30, 20, 10, 255]);
225    }
226}
visual_cortex_vision/debug.rs

visual_cortex_vision/
debug.rs
