decode/
editor_stream.rs

1//! Random-access decode for the editor (ED.3 / M-EDIT).
2//!
3//! [`GstreamerPipeStream`](crate::gstreamer_pipe::GstreamerPipeStream)
4//! decodes forward only — fine for the recorder's playback, useless for an
5//! editor that scrubs to arbitrary frames. [`EditorVideoStream`] wraps it
6//! with frame-indexed seeking plus a bounded decoded-frame cache.
7//!
8//! ## Seek strategy (CLI-pipe constraint)
9//!
10//! `gst-launch-1.0` exposes no command-line seek, so a precise jump to a
11//! keyframe isn't available the way `gstreamer-rs`'s `seek_simple` would
12//! be. The honest v1 therefore **forward-decodes**: a seek forward keeps
13//! pulling from the live pipe; a seek *backward* (before the pipe's
14//! current position) re-spawns the pipe from frame 0 and decodes up to the
15//! target. A bounded LRU of decoded frames makes local scrubbing and
16//! repeated access cheap — and export, which walks frames in order, never
17//! re-spawns. Swapping in a `gstreamer-rs` `ACCURATE` seek later is a
18//! one-site change behind this type (see M-DEC.3+); [`spawn_count`] exists
19//! so tests can assert the cache is doing its job.
20//!
21//! [`spawn_count`]: EditorVideoStream::spawn_count
22
23use std::collections::HashMap;
24use std::path::{Path, PathBuf};
25use std::time::Duration;
26
27use crate::gstreamer_pipe::{GstreamerPipeStream, Result, VideoMetadata};
28use crate::{VideoFrame, VideoStream};
29
30/// Default decoded-frame cache capacity (~10 s at 30 fps).
31pub const DEFAULT_CACHE_FRAMES: usize = 300;
32
33/// A small LRU cache of decoded frames keyed by frame index.
34struct FrameCache {
35    map: HashMap<u64, VideoFrame>,
36    /// Frame indices in least-recently-used order (LRU at the front).
37    order: Vec<u64>,
38    capacity: usize,
39}
40
41impl FrameCache {
42    fn new(capacity: usize) -> Self {
43        Self {
44            map: HashMap::new(),
45            order: Vec::new(),
46            capacity: capacity.max(1),
47        }
48    }
49
50    fn get(&mut self, index: u64) -> Option<VideoFrame> {
51        let frame = self.map.get(&index).cloned();
52        if frame.is_some() {
53            self.touch(index);
54        }
55        frame
56    }
57
58    fn touch(&mut self, index: u64) {
59        if let Some(pos) = self.order.iter().position(|&i| i == index) {
60            self.order.remove(pos);
61        }
62        self.order.push(index);
63    }
64
65    fn put(&mut self, index: u64, frame: VideoFrame) {
66        self.map.insert(index, frame);
67        self.touch(index);
68        while self.order.len() > self.capacity {
69            let evicted = self.order.remove(0);
70            self.map.remove(&evicted);
71        }
72    }
73}
74
75/// A frame-indexed, seekable view over a video file, backed by the
76/// forward-only [`GstreamerPipeStream`] plus an LRU frame cache.
77///
78/// Construct with [`EditorVideoStream::open`], then pull any frame by
79/// index with [`EditorVideoStream::frame`] (or seek by time with
80/// [`EditorVideoStream::seek_to_time`]). Out-of-range indices clamp to the
81/// last frame.
82pub struct EditorVideoStream {
83    path: PathBuf,
84    meta: VideoMetadata,
85    stream: Option<GstreamerPipeStream>,
86    /// Index the underlying forward stream will produce next.
87    next_index: u64,
88    cache: FrameCache,
89    spawn_count: u64,
90}
91
92impl std::fmt::Debug for EditorVideoStream {
93    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
94        f.debug_struct("EditorVideoStream")
95            .field("path", &self.path)
96            .field("meta", &self.meta)
97            .field("next_index", &self.next_index)
98            .field("spawn_count", &self.spawn_count)
99            .finish_non_exhaustive()
100    }
101}
102
103impl EditorVideoStream {
104    /// Probe `path` and prepare a seekable stream with the default cache
105    /// size. No decode pipe is spawned until the first [`frame`] call.
106    ///
107    /// # Errors
108    ///
109    /// Returns the underlying probe error if `gst-discoverer-1.0` can't
110    /// read the file (missing tool, unreadable media).
111    ///
112    /// [`frame`]: Self::frame
113    pub fn open(path: &Path) -> Result<Self> {
114        Self::open_with_cache(path, DEFAULT_CACHE_FRAMES)
115    }
116
117    /// As [`open`](Self::open) with an explicit cache capacity (frames).
118    ///
119    /// # Errors
120    ///
121    /// See [`open`](Self::open).
122    pub fn open_with_cache(path: &Path, cache_frames: usize) -> Result<Self> {
123        let meta = GstreamerPipeStream::probe(path)?;
124        Ok(Self {
125            path: path.to_path_buf(),
126            meta,
127            stream: None,
128            next_index: 0,
129            cache: FrameCache::new(cache_frames),
130            spawn_count: 0,
131        })
132    }
133
134    /// Frame width in pixels.
135    #[must_use]
136    pub fn width(&self) -> u32 {
137        self.meta.width
138    }
139
140    /// Frame height in pixels.
141    #[must_use]
142    pub fn height(&self) -> u32 {
143        self.meta.height
144    }
145
146    /// Source frame rate (fps).
147    #[must_use]
148    pub fn frame_rate(&self) -> f32 {
149        self.meta.frame_rate
150    }
151
152    /// Total frame count, if the container reported a duration.
153    #[must_use]
154    pub fn frame_count(&self) -> Option<u64> {
155        self.meta.frame_count
156    }
157
158    /// How many times the underlying decode pipe has been (re)spawned.
159    /// Diagnostic — used by tests to assert the cache avoids re-spawns.
160    #[must_use]
161    pub fn spawn_count(&self) -> u64 {
162        self.spawn_count
163    }
164
165    fn last_index(&self) -> Option<u64> {
166        self.meta.frame_count.map(|count| count.saturating_sub(1))
167    }
168
169    /// Return the frame at `index`, decoding (and re-spawning) as needed.
170    /// Out-of-range indices clamp to the last frame. Returns `None` only
171    /// when the decode pipe can't be spawned or the stream ends early.
172    #[allow(
173        clippy::must_use_candidate,
174        reason = "frame is often called to position the stream and the returned frame ignored; the caller decides whether to use it"
175    )]
176    pub fn frame(&mut self, index: u64) -> Option<VideoFrame> {
177        let index = match self.last_index() {
178            Some(last) => index.min(last),
179            None => index,
180        };
181
182        if let Some(frame) = self.cache.get(index) {
183            return Some(frame);
184        }
185
186        // Re-spawn when there's no live pipe, or we've already decoded past
187        // the target (can't rewind a forward-only pipe).
188        if self.stream.is_none() || self.next_index > index {
189            self.respawn()?;
190        }
191
192        while self.next_index <= index {
193            let mut frame = self.stream.as_mut()?.next_frame()?;
194            let decoded = self.next_index;
195            // Re-stamp index/pts against our own clock (the underlying
196            // stream restarts at 0 on each re-spawn).
197            frame.frame_index = decoded;
198            frame.pts_seconds = f64::from(u32::try_from(decoded).unwrap_or(u32::MAX))
199                / f64::from(self.meta.frame_rate);
200            self.next_index += 1;
201            self.cache.put(decoded, frame.clone());
202            if decoded == index {
203                return Some(frame);
204            }
205        }
206        self.cache.get(index)
207    }
208
209    /// Seek to the frame nearest `time` and return it.
210    #[allow(
211        clippy::must_use_candidate,
212        reason = "seek may be called purely to position the stream"
213    )]
214    pub fn seek_to_time(&mut self, time: Duration) -> Option<VideoFrame> {
215        let raw = (time.as_secs_f64() * f64::from(self.meta.frame_rate)).round();
216        #[allow(
217            clippy::cast_possible_truncation,
218            clippy::cast_sign_loss,
219            reason = "raw is clamped non-negative and frame indices fit u64"
220        )]
221        let index = if raw < 0.0 { 0 } else { raw as u64 };
222        self.frame(index)
223    }
224
225    fn respawn(&mut self) -> Option<()> {
226        let stream = GstreamerPipeStream::open(&self.path).ok()?;
227        self.stream = Some(stream);
228        self.next_index = 0;
229        self.spawn_count += 1;
230        Some(())
231    }
232}
decode/editor_stream.rs

decode/
editor_stream.rs
