katui_rt/
media.rs

1//! Raw video from a child process: a decoder writes RGBA frames to its
2//! standard output, a reader thread queues them, and the stream hands out
3//! the newest frame its clock has reached.
4
5use std::ffi::OsStr;
6use std::io::{self, Read as _};
7use std::process::{Child, Command, Stdio};
8use std::sync::mpsc::{Receiver, TryRecvError, sync_channel};
9use std::thread;
10use std::time::Instant;
11
12/// A running decoder and the frames it produced. Dropping it kills the
13/// decoder.
14#[derive(Debug)]
15pub struct VideoStream {
16    frames: Receiver<Vec<u8>>,
17    decoder: Child,
18    frames_per_second: u32,
19    started: Instant,
20    shown: u32,
21    ended: bool,
22}
23
24impl VideoStream {
25    /// Run `command`, which must write `width` x `height` RGBA frames
26    /// back to back on its standard output at `frames_per_second`.
27    pub fn spawn(
28        mut command: Command,
29        (width, height): (u32, u32),
30        frames_per_second: u32,
31    ) -> io::Result<Self> {
32        let mut decoder = command
33            .stdin(Stdio::null())
34            .stdout(Stdio::piped())
35            .stderr(Stdio::null())
36            .spawn()?;
37        let mut output = decoder
38            .stdout
39            .take()
40            .ok_or_else(|| io::Error::other("the decoder gave no output pipe"))?;
41        let (send, frames) = sync_channel(frames_per_second.max(4) as usize / 4);
42        let frame_bytes = width as usize * height as usize * 4;
43        thread::spawn(move || {
44            loop {
45                let mut frame = vec![0; frame_bytes];
46                if output.read_exact(&mut frame).is_err() || send.send(frame).is_err() {
47                    break;
48                }
49            }
50        });
51        Ok(Self {
52            frames,
53            decoder,
54            frames_per_second: frames_per_second.max(1),
55            started: Instant::now(),
56            shown: 0,
57            ended: false,
58        })
59    }
60
61    /// Decode with ffmpeg. `input` holds ffmpeg's input arguments, such as
62    /// `["-ss", "12", "-i", "clip.mp4"]`. We letterbox the picture to
63    /// `size` and drop the sound.
64    pub fn ffmpeg<S: AsRef<OsStr>>(
65        input: &[S],
66        size: (u32, u32),
67        frames_per_second: u32,
68    ) -> io::Result<Self> {
69        let mut command = Command::new("ffmpeg");
70        command
71            .args(["-hide_banner", "-loglevel", "error", "-nostdin"])
72            .args(input)
73            .args(["-an", "-vf", &fit_filter(size, frames_per_second)])
74            .args(["-f", "rawvideo", "-pix_fmt", "rgba", "-"]);
75        Self::spawn(command, size, frames_per_second)
76    }
77
78    /// The newest frame the clock has passed since the last call, or
79    /// `None` when no new frame is due or decoded yet. We skip the frames
80    /// in between, so a slow terminal drops frames and the picture keeps
81    /// pace with the clock.
82    pub fn latest(&mut self) -> Option<Vec<u8>> {
83        let passed = self.started.elapsed().as_millis() * u128::from(self.frames_per_second) / 1000;
84        let due = u32::try_from(passed).unwrap_or(u32::MAX).saturating_add(1);
85        let mut newest = None;
86        while self.shown < due {
87            match self.frames.try_recv() {
88                Ok(frame) => {
89                    newest = Some(frame);
90                    self.shown += 1;
91                }
92                Err(TryRecvError::Empty) => break,
93                Err(TryRecvError::Disconnected) => {
94                    self.ended = true;
95                    break;
96                }
97            }
98        }
99        newest
100    }
101
102    /// How many frames the stream has handed out or skipped.
103    #[must_use]
104    pub fn shown(&self) -> u32 {
105        self.shown
106    }
107
108    /// Whether the decoder finished and every frame went out.
109    #[must_use]
110    pub fn ended(&self) -> bool {
111        self.ended
112    }
113}
114
115impl Drop for VideoStream {
116    fn drop(&mut self) {
117        let _ = self.decoder.kill();
118        let _ = self.decoder.wait();
119    }
120}
121
122/// The ffmpeg filter that fits a clip inside `width` x `height` square
123/// pixels, letterboxed, at `frames_per_second`.
124fn fit_filter((width, height): (u32, u32), frames_per_second: u32) -> String {
125    format!(
126        "fps={frames_per_second},scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:-1:-1:black"
127    )
128}
129
130#[cfg(test)]
131mod tests {
132    use std::process::Command;
133    use std::time::{Duration, Instant};
134
135    use super::{VideoStream, fit_filter};
136
137    #[test]
138    fn the_filter_letterboxes_to_the_frame_at_the_frame_rate() {
139        assert_eq!(
140            fit_filter((320, 180), 30),
141            "fps=30,scale=320:180:force_original_aspect_ratio=decrease,pad=320:180:-1:-1:black"
142        );
143    }
144
145    #[test]
146    fn a_stream_hands_out_the_newest_due_frame_then_ends() {
147        let mut command = Command::new("sh");
148        command.args(["-c", "printf 'aaaabbbbccccdddd'"]);
149        let mut stream = VideoStream::spawn(command, (2, 1), 1000).expect("sh runs");
150        let deadline = Instant::now() + Duration::from_secs(5);
151        let mut last = None;
152        while !stream.ended() && Instant::now() < deadline {
153            if let Some(frame) = stream.latest() {
154                last = Some(frame);
155            }
156            std::thread::sleep(Duration::from_millis(5));
157        }
158        assert!(stream.ended(), "the decoder exited");
159        assert_eq!(stream.shown(), 2);
160        assert_eq!(last.as_deref(), Some(&b"ccccdddd"[..]));
161    }
162}
katui_rt/media.rs

katui_rt/
media.rs
