Add bounded session playback, synced audio and frame thumbnails

This commit is contained in:
2026-10-08 19:20:44 +02:00
parent f6cdef5dd2
commit 0a7d62a807
24 changed files with 2566 additions and 55 deletions
+2
View File
@@ -10,6 +10,8 @@ publish.workspace = true
[dependencies]
clipbin-core.workspace = true
clipbin-media.workspace = true
image = { version = "0.25.10", default-features = false }
clipbin-obs.workspace = true
gpui.workspace = true
+22 -2
View File
@@ -62,6 +62,7 @@ pub struct ClipbinApp {
pub settings: Settings,
pub recorder: Recorder,
pub editor: Editor,
pub playback: crate::playback::Playback,
/// A short message for actions that are not connected yet.
pub notice: Option<SharedString>,
pub drag: Option<Drag>,
@@ -201,7 +202,7 @@ impl ClipbinApp {
let library = if sample {
Library::sample(SystemTime::now())
} else {
Library::empty(settings.unprotected_session_limit_bytes)
service.snapshot().library
};
let recorder = match (sample, std::env::var("CLIPBIN_SAMPLE_STATUS").as_deref()) {
(_, Ok("failed")) => Recorder {
@@ -234,6 +235,7 @@ impl ClipbinApp {
selection: (Duration::ZERO, Duration::ZERO),
playing: false,
},
playback: crate::playback::Playback::default(),
notice: None,
drag: None,
timeline_bounds: Rc::default(),
@@ -287,6 +289,7 @@ impl ClipbinApp {
})
.detach();
}
Self::start_playback_updates(window, cx);
app
}
@@ -298,7 +301,11 @@ impl ClipbinApp {
self.notice = None;
match route {
Route::Session(id) => self.open_session(id),
route => self.route = route,
route => {
self.playback.player.close();
self.editor.playing = false;
self.route = route;
}
}
cx.notify();
}
@@ -322,6 +329,19 @@ impl ClipbinApp {
selection,
playing: false,
};
if let Ok(store) = clipbin_core::store::Store::discover() {
let preview_time = playhead.min(
session
.length
.saturating_sub(Duration::from_nanos(16_666_667)),
);
self.editor.playhead = preview_time;
if session.bookmarks.is_empty() {
self.playback.player.open_latest(store, id);
} else {
self.playback.player.open(store, id, preview_time);
}
}
self.route = Route::Session(id);
}
+9 -4
View File
@@ -6,6 +6,7 @@ use gpui::{
mod app;
mod assets;
mod components;
mod playback;
mod recorder_service;
mod screens;
mod text_field;
@@ -47,6 +48,7 @@ fn main() {
});
cx.bind_keys([KeyBinding::new("ctrl-q", Quit, None)]);
text_field::bind_keys(cx);
playback::bind_keys(cx);
let bounds = Bounds::centered(None, size(px(1360.), px(860.)), cx);
let result = cx.open_window(
@@ -65,10 +67,13 @@ fn main() {
},
);
if let Err(error) = result {
eprintln!("Could not open Clipbin: {error}");
cx.quit();
return;
match result {
Ok(window) => playback::bind_actions(window, cx),
Err(error) => {
eprintln!("Could not open Clipbin: {error}");
cx.quit();
return;
}
}
cx.activate(true);
+279
View File
@@ -0,0 +1,279 @@
//! GPUI image upload and controls; file I/O and decoding belong to the player.
use crate::app::{ClipbinApp, Route};
use clipbin_core::store::Store;
use clipbin_media::{
player::{Picture, Player},
thumbnail::Thumbnails,
};
use gpui::{Context, RenderImage};
use std::{collections::HashMap, path::PathBuf, sync::Arc, time::Duration};
pub struct Playback {
pub player: Player,
pub image: Option<Arc<RenderImage>>,
pub message: Option<String>,
revision: u64,
last_frame: Option<Arc<clipbin_media::decode::Frame>>,
pub muted: bool,
pub volume: f32,
pub audio_error: Option<String>,
pub preview_height: f32,
pub thumbnails: HashMap<u64, PathBuf>,
pub clip_thumbnails: HashMap<u64, PathBuf>,
}
impl Default for Playback {
fn default() -> Self {
Self {
player: Player::new(),
image: None,
message: Some("Loading footage…".into()),
revision: 0,
last_frame: None,
muted: false,
volume: 1.,
audio_error: None,
preview_height: 416.,
thumbnails: HashMap::new(),
clip_thumbnails: HashMap::new(),
}
}
}
impl ClipbinApp {
pub fn seek_playback(&mut self, time: Duration) {
self.editor.playhead = time;
self.playback.player.seek(time);
}
pub fn toggle_playback(&mut self) {
if let Route::Session(id) = self.route {
if self
.library
.session(id)
.is_some_and(|session| self.editor.playhead >= session.length)
{
self.seek_playback(Duration::ZERO);
}
self.editor.playing = !self.editor.playing;
self.playback.player.play(self.editor.playing);
}
}
pub fn start_playback_updates(window: &gpui::Window, cx: &mut Context<Self>) {
cx.spawn_in(window, async |entity, cx| {
loop {
gpui::Timer::after(Duration::from_millis(16)).await;
if entity
.update_in(cx, |app, window, cx| {
if !matches!(app.route, Route::Session(_)) {
if let Some(image) = app.playback.image.take() {
let _ = window.drop_image(image);
}
app.playback.last_frame = None;
return;
}
let snapshot = app.playback.player.snapshot();
if snapshot.revision == app.playback.revision {
return;
}
app.playback.revision = snapshot.revision;
app.editor.playhead = snapshot.time;
app.editor.playing = snapshot.playing;
app.playback.audio_error = snapshot.audio_error;
match snapshot.picture {
Picture::Frame(frame) => {
// Swscale already produced GPUI-native BGRA off the UI thread.
if app
.playback
.last_frame
.as_ref()
.is_none_or(|last| !Arc::ptr_eq(last, &frame))
{
let upload_started = std::time::Instant::now();
let bytes = frame.bgra.clone();
if let Some(buffer) =
image::RgbaImage::from_raw(frame.width, frame.height, bytes)
{
let image =
Arc::new(RenderImage::new(vec![image::Frame::new(
buffer,
)]));
if let Some(old) = app.playback.image.replace(image) {
let _ = window.drop_image(old);
}
if std::env::var_os("CLIPBIN_TRACE_FRAMES").is_some() {
eprintln!(
"playback_ui upload_ms={:.3} width={} height={}",
upload_started.elapsed().as_secs_f64() * 1000.,
frame.width,
frame.height
);
}
}
}
app.playback.last_frame = Some(frame);
app.playback.message = None;
}
Picture::Loading => {
if let Some(old) = app.playback.image.take() {
let _ = window.drop_image(old);
}
app.playback.last_frame = None;
app.playback.message = Some("Loading footage…".into());
}
Picture::Gap => {
if let Some(old) = app.playback.image.take() {
let _ = window.drop_image(old);
}
app.playback.last_frame = None;
app.playback.message =
Some("No footage (not captured / cleaned up)".into());
}
Picture::Error(error) => {
if let Some(old) = app.playback.image.take() {
let _ = window.drop_image(old);
}
app.playback.last_frame = None;
app.playback.message = Some(error);
}
}
cx.notify();
})
.is_err()
{
break;
}
}
})
.detach();
// Discovery and thumbnail work never touch the UI thread's filesystem.
let (sender, receiver) = std::sync::mpsc::sync_channel(1);
std::thread::spawn(move || {
let thumbnails = Thumbnails::new();
loop {
let mut ready = HashMap::new();
let mut clips = HashMap::new();
if let Ok(store) = Store::discover()
&& let Ok(metadata) = store.read()
{
for metadata in metadata {
if let Some(segment) = metadata
.segments
.iter()
.find(|segment| segment.complete && segment.available)
&& let Some(path) = thumbnails
.request(&store.directory(metadata.session.id).join(&segment.path))
{
ready.insert(metadata.session.id, path);
}
}
}
if let Ok(store) = Store::discover()
&& let Some(root) = store.root.parent()
&& let Ok(entries) = std::fs::read_dir(root.join("clips"))
{
for entry in entries.flatten() {
let path = entry.path();
if path.extension().is_some_and(|extension| extension == "mp4")
&& let Some(id) = path
.file_stem()
.and_then(|id| id.to_str())
.and_then(|id| id.parse().ok())
&& let Some(thumbnail) = thumbnails.request(&path)
{
clips.insert(id, thumbnail);
}
}
}
if matches!(
sender.try_send((ready, clips)),
Err(std::sync::mpsc::TrySendError::Disconnected(_))
) {
break;
}
std::thread::sleep(Duration::from_secs(1));
}
});
cx.spawn(async move |entity, cx| {
loop {
gpui::Timer::after(Duration::from_millis(250)).await;
let ready = receiver.try_recv().ok();
if entity
.update(cx, |app, cx| {
if let Some((sessions, clips)) = ready
&& (app.playback.thumbnails != sessions
|| app.playback.clip_thumbnails != clips)
{
app.playback.thumbnails = sessions;
app.playback.clip_thumbnails = clips;
cx.notify();
}
})
.is_err()
{
break;
}
}
})
.detach();
}
}
gpui::actions!(playback, [TogglePlayback, StepBack, StepForward]);
pub fn bind_keys(cx: &mut gpui::App) {
cx.bind_keys([
gpui::KeyBinding::new("space", TogglePlayback, None),
gpui::KeyBinding::new("k", TogglePlayback, None),
gpui::KeyBinding::new("left", StepBack, None),
gpui::KeyBinding::new("j", StepBack, None),
gpui::KeyBinding::new("right", StepForward, None),
gpui::KeyBinding::new("l", StepForward, None),
]);
cx.bind_keys(
["space", "k", "j", "l"]
.map(|key| gpui::KeyBinding::new(key, gpui::NoAction {}, Some("TextField"))),
);
}
/// Window actions also work when a mouse click leaves no focused GPUI element.
/// Text-field contexts keep their own typing and arrow bindings.
pub fn bind_actions(window: gpui::WindowHandle<ClipbinApp>, cx: &mut gpui::App) {
// The event dispatcher already borrows the window; update its entity directly.
let editor = window.entity(cx).expect("editor window").downgrade();
let toggle = editor.clone();
cx.on_action(move |_: &TogglePlayback, cx| {
let _ = toggle.update(cx, |app, cx| {
app.toggle_playback();
cx.notify();
});
});
let back = editor.clone();
cx.on_action(move |_: &StepBack, cx| {
let _ = back.update(cx, |app, cx| {
app.step_playback(false);
cx.notify();
});
});
cx.on_action(move |_: &StepForward, cx| {
let _ = editor.update(cx, |app, cx| {
app.step_playback(true);
cx.notify();
});
});
}
impl ClipbinApp {
pub fn step_playback(&mut self, forward: bool) {
if !matches!(self.route, Route::Session(_)) {
return;
}
self.editor.playing = false;
self.playback.player.play(false);
let snapshot = self.playback.player.snapshot();
let step = Duration::from_secs_f64(1. / snapshot.frames_per_second.max(1) as f64);
let time = snapshot.video_time.unwrap_or(self.editor.playhead);
// Round towards the selected frame when nanoseconds cannot represent a frame period.
let at = if forward {
time + step
} else {
time.saturating_sub(step)
} + Duration::from_nanos(2);
self.seek_playback(at);
}
}
+1 -1
View File
@@ -61,7 +61,7 @@ pub fn render(app: &ClipbinApp, cx: &mut Context<ClipbinApp>) -> AnyElement {
.flex_col()
.gap_3()
.child(
thumbnail(&clip.game).child(
thumbnail(&clip.game, app.playback.clip_thumbnails.get(&clip.id)).child(
corner()
.bottom_2p5()
.right_2p5()
+118 -34
View File
@@ -6,15 +6,14 @@ use clipbin_core::library::{
Session, SpanKind, export_range, format_ago, format_clock, format_length,
};
use gpui::{
AnyElement, Bounds, ClickEvent, Context, FontWeight, Hsla, IntoElement, MouseButton,
MouseDownEvent, MouseMoveEvent, MouseUpEvent, Pixels, Point, ScrollWheelEvent, canvas, div,
pattern_slash, prelude::*, px, relative,
AnyElement, Bounds, ClickEvent, Context, FontWeight, Hsla, ImageSource, IntoElement,
MouseButton, MouseDownEvent, MouseMoveEvent, MouseUpEvent, ObjectFit, Pixels, Point,
ScrollWheelEvent, canvas, div, img, pattern_slash, prelude::*, px, relative,
};
use crate::{
app::{ClipbinApp, Drag, Route},
components::{ButtonStyle, aspect_ratio, button, chip, icon, panel, row},
screens::backdrop,
theme,
timeline_view::TimelineView,
};
@@ -73,37 +72,54 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
);
let preview = aspect_ratio(div(), 16. / 9.)
.h(px(app.playback.preview_height))
.flex_shrink_0()
.relative()
.w_full()
.overflow_hidden()
.rounded_lg()
.bg(gpui::black())
.child(backdrop(&session.game).absolute().size_full().opacity(0.55))
.child(
.children(app.playback.image.clone().map(|image| {
img(ImageSource::Render(image))
.absolute()
.size_full()
.object_fit(ObjectFit::Contain)
}))
.children(app.playback.message.clone().map(|message| {
div()
.absolute()
.size_full()
.flex()
.flex_col()
.items_center()
.justify_center()
.gap_3()
.child(
div()
.flex()
.items_center()
.justify_center()
.size(px(56.))
.rounded_full()
.bg(theme::overlay())
.child(icon("play").size_6().text_color(theme::foreground())),
)
.child(
div()
.text_xs()
.text_color(theme::muted())
.child("Preview arrives with playback"),
),
.text_xs()
.text_color(theme::muted())
.child(message)
}))
.child(
canvas(
{
let player = cx.entity().downgrade();
move |bounds, window, cx| {
let scale = window.scale_factor();
let size = (
(f32::from(bounds.size.width) * scale) as u32,
(f32::from(bounds.size.height) * scale) as u32,
);
let _ = player.update(cx, |app, cx| {
app.playback.player.resize(size);
let height = f32::from(bounds.size.width) * 9. / 16.;
if (app.playback.preview_height - height).abs() > 1. {
app.playback.preview_height = height;
cx.notify();
}
});
}
},
|_, _, _, _| {},
)
.absolute()
.size_full(),
)
.child(
div()
@@ -126,6 +142,7 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
.map(|b| b.at);
let transport = div()
.flex()
.flex_wrap()
.items_center()
.gap_1()
.child(
@@ -136,8 +153,7 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
None,
)
.on_click(cx.listener(|app, _: &ClickEvent, _, cx| {
app.notice =
Some("Playback isn't connected yet; it comes with the decoder.".into());
app.toggle_playback();
cx.notify();
})),
)
@@ -150,7 +166,7 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
)
.on_click(cx.listener(move |app, _: &ClickEvent, _, cx| {
if let Some(at) = previous_bookmark {
app.editor.playhead = at;
app.seek_playback(at);
cx.notify();
}
})),
@@ -164,7 +180,7 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
)
.on_click(cx.listener(move |app, _: &ClickEvent, _, cx| {
if let Some(at) = next_bookmark {
app.editor.playhead = at;
app.seek_playback(at);
cx.notify();
}
})),
@@ -187,6 +203,52 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
.child(format!("/ {}", format_clock(length))),
),
)
.child(
button(
"mute",
ButtonStyle::Ghost,
Some(if app.playback.muted {
"volume-x"
} else {
"volume-2"
}),
None,
)
.on_click(cx.listener(|app, _: &ClickEvent, _, cx| {
app.playback.muted = !app.playback.muted;
app.playback
.player
.audio(app.playback.muted, app.playback.volume);
cx.notify();
})),
)
.child(
button("volume-down", ButtonStyle::Ghost, None, Some("−")).on_click(cx.listener(
|app, _: &ClickEvent, _, cx| {
app.playback.volume = (app.playback.volume - 0.1).max(0.);
app.playback
.player
.audio(app.playback.muted, app.playback.volume);
cx.notify();
},
)),
)
.child(
div()
.text_xs()
.child(format!("{:.0}%", app.playback.volume * 100.)),
)
.child(
button("volume-up", ButtonStyle::Ghost, None, Some("+")).on_click(cx.listener(
|app, _: &ClickEvent, _, cx| {
app.playback.volume = (app.playback.volume + 0.1).min(1.);
app.playback
.player
.audio(app.playback.muted, app.playback.volume);
cx.notify();
},
)),
)
.child(div().flex_1())
.child(
div()
@@ -210,6 +272,12 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
.gap_3()
.child(preview)
.child(transport)
.children(
app.playback
.audio_error
.clone()
.map(|error| div().text_xs().text_color(theme::warning()).child(error)),
)
.child(timeline(app, session, cx));
let (export_start, export_end) = export_range(start, end, KEYFRAME_INTERVAL, length);
@@ -320,7 +388,7 @@ pub fn render(app: &ClipbinApp, id: u64, cx: &mut Context<ClipbinApp>) -> AnyEle
}),
)
.on_click(cx.listener(move |app, _: &ClickEvent, _, cx| {
app.editor.playhead = at;
app.seek_playback(at);
app.editor.selection = range;
cx.notify();
}))
@@ -529,9 +597,25 @@ fn timeline(app: &ClipbinApp, session: &Session, cx: &mut Context<ClipbinApp>) -
.size_full()
.overflow_hidden()
.child(
canvas(move |area, _, _| bounds.set(Some(area)), |_, _, _, _| {})
.absolute()
.size_full(),
canvas(
move |area, _, _| {
if bounds.get() != Some(area)
&& std::env::var_os("CLIPBIN_TRACE_FRAMES").is_some()
{
eprintln!(
"playback_timeline x={} y={} width={} height={}",
f32::from(area.left()),
f32::from(area.top()),
f32::from(area.size.width),
f32::from(area.size.height)
);
}
bounds.set(Some(area));
},
|_, _, _, _| {},
)
.absolute()
.size_full(),
)
.child(
div()
@@ -565,7 +649,7 @@ fn timeline(app: &ClipbinApp, session: &Session, cx: &mut Context<ClipbinApp>) -
} else if near(end) {
Drag::SelectionEnd
} else {
app.editor.playhead = time;
app.seek_playback(time);
Drag::Playhead
});
cx.notify();
@@ -583,7 +667,7 @@ fn timeline(app: &ClipbinApp, session: &Session, cx: &mut Context<ClipbinApp>) -
let (start, end) = app.editor.selection;
let minimum = Duration::from_secs(1);
match drag {
Drag::Playhead => app.editor.playhead = time,
Drag::Playhead => app.seek_playback(time),
Drag::SelectionStart => {
app.editor.selection.0 = time.min(end.saturating_sub(minimum))
}
+14 -1
View File
@@ -9,7 +9,20 @@ use crate::{components::aspect_ratio, theme};
/// A 16:9 thumbnail. Until frames are decoded, each game gets its own colour
/// with its initials, so cards stay recognisable.
pub fn thumbnail(game: &str) -> Div {
pub fn thumbnail(game: &str, path: Option<&std::path::PathBuf>) -> Div {
if let Some(path) = path {
return aspect_ratio(div(), 16. / 9.)
.relative()
.w_full()
.overflow_hidden()
.rounded_xl()
.child(
gpui::img(path.clone())
.absolute()
.size_full()
.object_fit(gpui::ObjectFit::Cover),
);
}
let tint = game_tint(game);
let initials: String = game
.split_whitespace()
@@ -61,7 +61,7 @@ pub fn render(app: &ClipbinApp, cx: &mut Context<ClipbinApp>) -> AnyElement {
.cursor_pointer()
.group("card")
.child(
thumbnail(&session.game)
thumbnail(&session.game, app.playback.thumbnails.get(&session.id))
.border_1()
.border_color(gpui::transparent_black())
.group_hover("card", |style| {
+10 -2
View File
@@ -11,6 +11,14 @@ publish.workspace = true
[dependencies]
clipbin-core.workspace = true
ffmpeg-next = { version = "9", default-features = false, features = ["codec", "format", "software-scaling", "software-resampling"] }
image = { version = "0.25.10", default-features = false, features = ["png"] }
cpal = { version = "0.18.2", features = ["pulseaudio"] }
[lints]
workspace = true
[target.'cfg(target_os = "linux")'.dependencies]
rustix = { version = "1", features = ["process", "thread"] }
[target.'cfg(windows)'.dependencies]
windows-sys = { version = "0.61", features = ["Win32_System_Threading"] }
[lints.rust]
unsafe_code = "deny"
@@ -0,0 +1,67 @@
//! Headless timing check against real segments; run inside the rig for device audio.
use clipbin_core::store::Store;
use clipbin_media::player::{Picture, Player};
use std::{
thread,
time::{Duration, Instant},
};
fn main() {
let directory = std::env::args().nth(1).expect("session root");
let id = std::env::args()
.nth(2)
.unwrap_or_else(|| "1".into())
.parse()
.unwrap();
let player = Player::new();
player.resize((800, 450));
player.open(
Store {
root: directory.into(),
},
id,
Duration::ZERO,
);
thread::sleep(Duration::from_millis(250));
player.play(true);
let begin = Instant::now();
let mut last = None;
let mut frames = 0;
for _ in 0..700 {
thread::sleep(Duration::from_millis(10));
let snapshot = player.snapshot();
assert!(snapshot.audio_error.is_none(), "{:?}", snapshot.audio_error);
if let Picture::Frame(frame) = snapshot.picture {
if let Some(video_time) = snapshot.video_time {
assert!(
video_time.abs_diff(snapshot.time) <= Duration::from_millis(20),
"video out of sync: {video_time:?} vs {:?}",
snapshot.time
);
}
if last
.as_ref()
.is_none_or(|previous| !std::sync::Arc::ptr_eq(previous, &frame))
{
frames += 1;
}
last = Some(frame);
}
}
let snapshot = player.snapshot();
println!(
"playback_check wall={:.3} session={:.3} displayed_frames={frames}",
begin.elapsed().as_secs_f64(),
snapshot.time.as_secs_f64()
);
assert!(
snapshot.time >= Duration::from_secs(6),
"audio master clock stalled: {snapshot:?}"
);
player.audio(false, 0.5);
thread::sleep(Duration::from_secs(1));
player.audio(true, 0.5);
let before = snapshot.time;
thread::sleep(Duration::from_secs(1));
assert!(player.snapshot().time > before, "muted wall clock stalled");
player.close();
}
+454
View File
@@ -0,0 +1,454 @@
//! Combined-track decoding and bounded device output. No decoder runs in a callback.
use crate::ffmpeg::{self, format, media, software::resampling, util::frame::Audio as AudioFrame};
use clipbin_core::store::Metadata;
use cpal::{
FromSample, SizedSample,
traits::{DeviceTrait, HostTrait, StreamTrait},
};
use std::{
collections::VecDeque,
path::{Path, PathBuf},
sync::{Arc, Mutex},
time::{Duration, Instant},
};
struct Block {
time: Duration,
samples: Vec<f32>,
consumed: usize,
}
struct Buffer {
blocks: VecDeque<Block>,
queued: usize,
playing: bool,
volume: f32,
// The time of the first sample actually scheduled for the device.
anchors: VecDeque<(Duration, Instant, Duration)>,
error: Option<String>,
}
impl Buffer {
fn time(&mut self, now: Instant) -> Option<Duration> {
while self.anchors.len() > 1 && self.anchors[1].1 <= now {
self.anchors.pop_front();
}
self.anchors.front().and_then(|(time, begins, span)| {
(now <= *begins + *span + Duration::from_millis(100))
.then(|| *time + now.saturating_duration_since(*begins).min(*span))
})
}
}
pub struct Audio {
_stream: cpal::Stream,
buffer: Arc<Mutex<Buffer>>,
rate: u32,
channels: u16,
decoder: Option<(PathBuf, AudioDecoder)>,
next: Duration,
}
impl Audio {
pub fn new() -> Result<Self, String> {
#[cfg(target_os = "linux")]
if let Ok(host) = cpal::host_from_id(cpal::HostId::PulseAudio) {
match Self::with_host(host) {
Ok(audio) => return Ok(audio),
Err(error) => eprintln!("PulseAudio output unavailable: {error}; trying ALSA"),
}
}
Self::with_host(cpal::default_host())
}
fn with_host(host: cpal::Host) -> Result<Self, String> {
let device = host
.default_output_device()
.ok_or("No audio output device")?;
let supported = device.default_output_config().map_err(|e| e.to_string())?;
let format = supported.sample_format();
let config: cpal::StreamConfig = supported.into();
let rate = config.sample_rate;
let channels = config.channels;
let buffer = Arc::new(Mutex::new(Buffer {
blocks: VecDeque::new(),
queued: 0,
playing: false,
volume: 1.,
anchors: VecDeque::new(),
error: None,
}));
let stream = match format {
cpal::SampleFormat::F32 => output::<f32>(&device, config, buffer.clone()),
cpal::SampleFormat::I16 => output::<i16>(&device, config, buffer.clone()),
cpal::SampleFormat::U16 => output::<u16>(&device, config, buffer.clone()),
cpal::SampleFormat::I32 => output::<i32>(&device, config, buffer.clone()),
cpal::SampleFormat::F64 => output::<f64>(&device, config, buffer.clone()),
_ => return Err(format!("Unsupported audio device format: {format}")),
}
.map_err(|e| e.to_string())?;
stream.play().map_err(|e| e.to_string())?;
eprintln!("playback_audio rate={rate} channels={channels} format={format}");
Ok(Self {
_stream: stream,
buffer,
rate,
channels,
decoder: None,
next: Duration::ZERO,
})
}
pub fn reset(&mut self, time: Duration) {
let mut buffer = self.buffer.lock().unwrap();
buffer.blocks.clear();
buffer.queued = 0;
buffer.anchors.clear();
self.decoder = None;
self.next = time;
}
pub fn controls(&self, playing: bool, volume: f32) {
let mut buffer = self.buffer.lock().unwrap();
buffer.playing = playing;
buffer.volume = volume.clamp(0., 1.);
if !playing {
buffer.anchors.clear();
}
}
/// Device time, corrected for callback-to-DAC latency; bounded by scheduled samples.
pub fn time(&self, now: Instant) -> Option<Duration> {
self.buffer.lock().unwrap().time(now)
}
pub fn error(&self) -> Option<String> {
self.buffer.lock().unwrap().error.clone()
}
pub fn catch_up(&mut self, time: Duration) {
if self.next + Duration::from_millis(50) < time && self.buffer.lock().unwrap().queued == 0 {
self.reset(time);
}
}
pub fn fill(&mut self, metadata: &Metadata) -> Result<(), ffmpeg::Error> {
// At most 250 ms buffered. All allocations/decoding happen on the player thread.
let limit = self.rate as usize * self.channels as usize / 4;
for _ in 0..16 {
if self.buffer.lock().unwrap().queued >= limit {
break;
}
let Some(at) = metadata.locate(self.next).filter(|at| at.segment.complete) else {
break;
};
if self
.decoder
.as_ref()
.is_none_or(|(path, _)| path != &at.segment.path)
{
self.decoder = Some((
at.segment.path.clone(),
AudioDecoder::open(&at.segment.path, self.rate, self.channels, at.offset)?,
));
}
let (_, decoder) = self.decoder.as_mut().unwrap();
let Some((offset, mut samples)) = decoder.next()? else {
self.next = at.segment.end;
self.decoder = None;
continue;
};
let begin = at.segment.start + offset;
let skip = ((self.next.saturating_sub(begin).as_secs_f64() * self.rate as f64).round()
as usize)
.min(samples.len() / self.channels as usize);
samples.drain(..skip * self.channels as usize);
let time = begin + Duration::from_secs_f64(skip as f64 / self.rate as f64);
let keep = ((at.segment.end.saturating_sub(time).as_secs_f64() * self.rate as f64)
.round() as usize)
* self.channels as usize;
samples.truncate(keep);
if samples.is_empty() {
continue;
}
self.next = time
+ Duration::from_secs_f64(
(samples.len() / self.channels as usize) as f64 / self.rate as f64,
);
if at.segment.end.saturating_sub(self.next)
< Duration::from_nanos(1_000_000_000 / self.rate as u64)
{
self.next = at.segment.end;
}
let mut buffer = self.buffer.lock().unwrap();
buffer.queued += samples.len();
buffer.blocks.push_back(Block {
time,
samples,
consumed: 0,
});
}
Ok(())
}
}
fn output<T: SizedSample + FromSample<f32>>(
device: &cpal::Device,
config: cpal::StreamConfig,
buffer: Arc<Mutex<Buffer>>,
) -> Result<cpal::Stream, cpal::Error> {
let errors = buffer.clone();
let rate = config.sample_rate;
let channels = config.channels as usize;
device.build_output_stream(
config,
move |data: &mut [T], info: &cpal::OutputCallbackInfo| {
// try_lock guarantees UI/decoder contention cannot block the device callback.
data.fill(T::from_sample(0.));
let Ok(mut buffer) = buffer.try_lock() else {
return;
};
if !buffer.playing {
return;
}
let latency = info
.timestamp()
.playback
.duration_since(info.timestamp().callback);
let mut scheduled = None;
for (frame_number, output) in data.chunks_exact_mut(channels).enumerate() {
let volume = buffer.volume;
let Some(block) = buffer.blocks.front_mut() else {
break;
};
let time = block.time
+ Duration::from_secs_f64((block.consumed / channels) as f64 / rate as f64);
scheduled.get_or_insert((
time,
Instant::now()
+ latency
+ Duration::from_secs_f64(frame_number as f64 / rate as f64),
Duration::ZERO,
));
for sample in output {
*sample =
T::from_sample((block.samples[block.consumed] * volume).clamp(-1., 1.));
block.consumed += 1;
}
let done = block.consumed == block.samples.len();
buffer.queued -= channels;
if let Some((_, _, span)) = scheduled.as_mut() {
*span += Duration::from_secs_f64(1. / rate as f64);
}
if done {
buffer.blocks.pop_front();
}
}
if let Some(scheduled) = scheduled {
if buffer.anchors.len() == 32 {
buffer.anchors.pop_front();
}
buffer.anchors.push_back(scheduled);
}
},
move |error| {
if let Ok(mut buffer) = errors.lock() {
buffer.error = Some(format!("Audio output failed: {error}"));
}
},
None,
)
}
struct AudioDecoder {
input: format::context::Input,
decoder: ffmpeg::decoder::Audio,
resampler: resampling::Context,
stream: usize,
time_base: f64,
origin: i64,
rate: u32,
channels: u16,
eof: bool,
next_time: Duration,
initialised_time: bool,
drained: bool,
}
impl AudioDecoder {
fn open(path: &Path, rate: u32, channels: u16, seek: Duration) -> Result<Self, ffmpeg::Error> {
crate::init();
let mut input = format::input(path)?;
// OBS writes system, mic, combined, in that order; a single track is already mixed.
let tracks: Vec<_> = input
.streams()
.filter(|stream| stream.parameters().medium() == media::Type::Audio)
.collect();
let track = if tracks.len() >= 3 {
&tracks[2]
} else {
tracks.first().ok_or(ffmpeg::Error::StreamNotFound)?
};
let stream = track.index();
let time_base = f64::from(track.time_base());
let origin = track.start_time().max(0);
let decoder = ffmpeg::codec::context::Context::from_parameters(track.parameters())?
.decoder()
.audio()?;
let layout = if decoder.channel_layout().channels() == 0 {
ffmpeg::ChannelLayout::default(decoder.channels() as i32)
} else {
decoder.channel_layout()
};
let resampler = resampling::Context::get(
decoder.format(),
layout,
decoder.rate(),
format::Sample::F32(format::sample::Type::Packed),
ffmpeg::ChannelLayout::default(channels as i32),
rate,
)?;
if seek > Duration::ZERO {
let timestamp = (seek.as_secs_f64() * 1_000_000.) as i64;
input.seek(timestamp, ..timestamp)?;
}
Ok(Self {
input,
decoder,
resampler,
stream,
time_base,
origin,
rate,
channels,
eof: false,
next_time: seek,
initialised_time: false,
drained: false,
})
}
fn next(&mut self) -> Result<Option<(Duration, Vec<f32>)>, ffmpeg::Error> {
loop {
let mut frame = AudioFrame::empty();
match self.decoder.receive_frame(&mut frame) {
Ok(()) => {
let timestamp = frame.timestamp().map_or(self.next_time, |pts| {
Duration::from_secs_f64((pts - self.origin).max(0) as f64 * self.time_base)
});
let time = if self.initialised_time {
self.next_time
} else {
timestamp
};
self.initialised_time = true;
// Allocate enough output for upsampling; ffmpeg-next's empty output uses input count.
let count = ((frame.samples() as u64 * self.rate as u64)
.div_ceil(frame.rate() as u64)
+ 256) as usize;
let mut converted = AudioFrame::new(
format::Sample::F32(format::sample::Type::Packed),
count,
ffmpeg::ChannelLayout::default(self.channels as i32),
);
converted.set_rate(self.rate);
self.resampler.run(&frame, &mut converted)?;
let samples = converted.plane::<f32>(0).to_vec();
self.next_time = time
+ Duration::from_secs_f64(converted.samples() as f64 / self.rate as f64);
return Ok(Some((time, samples)));
}
Err(ffmpeg::Error::Eof) => {
if self.drained {
return Ok(None);
}
self.drained = true;
let mut converted = AudioFrame::new(
format::Sample::F32(format::sample::Type::Packed),
256,
ffmpeg::ChannelLayout::default(self.channels as i32),
);
converted.set_rate(self.rate);
self.resampler.flush(&mut converted)?;
return if converted.samples() == 0 {
Ok(None)
} else {
Ok(Some((self.next_time, converted.plane::<f32>(0).to_vec())))
};
}
Err(ffmpeg::Error::Other { errno }) if errno == ffmpeg::error::EAGAIN => {}
Err(error) => return Err(error),
}
if self.eof {
return Ok(None);
}
let mut packet = ffmpeg::Packet::empty();
loop {
match packet.read(&mut self.input) {
Ok(()) if packet.stream() == self.stream => {
self.decoder.send_packet(&packet)?;
break;
}
Ok(()) => {}
Err(ffmpeg::Error::Eof) => {
self.eof = true;
self.decoder.send_eof()?;
break;
}
Err(error) => return Err(error),
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn device_clock_waits_for_latency_and_stops_at_scheduled_samples() {
let now = Instant::now();
let mut buffer = Buffer {
blocks: VecDeque::new(),
queued: 0,
playing: true,
volume: 1.,
anchors: VecDeque::new(),
error: None,
};
buffer.anchors.push_back((
Duration::from_secs(2),
now + Duration::from_millis(100),
Duration::from_millis(50),
));
buffer.anchors.push_back((
Duration::from_millis(2050),
now + Duration::from_millis(150),
Duration::from_millis(50),
));
assert_eq!(buffer.time(now), Some(Duration::from_secs(2)));
assert_eq!(
buffer.time(now + Duration::from_millis(125)),
Some(Duration::from_millis(2025))
);
assert_eq!(
buffer.time(now + Duration::from_millis(200)),
Some(Duration::from_millis(2100))
);
assert_eq!(buffer.anchors.len(), 1);
assert_eq!(buffer.time(now + Duration::from_secs(5)), None);
}
#[test]
fn resampling_selects_combined_track_and_preserves_its_tone() {
let Some(directory) = std::env::var_os("CLIPBIN_TEST_MEDIA_DIR") else {
return;
};
let mut decoder = AudioDecoder::open(
&PathBuf::from(directory).join("combined.mkv"),
44_100,
1,
Duration::ZERO,
)
.unwrap();
let mut samples = Vec::new();
while let Some((_, block)) = decoder.next().unwrap() {
samples.extend(block);
}
assert!(samples.len() > 80_000 && samples.len() < 95_000);
let crossings = samples
.windows(2)
.filter(|pair| pair[0] <= 0. && pair[1] > 0.)
.count();
let hz = crossings as f64 / (samples.len() as f64 / 44_100.);
assert!(
(435. ..445.).contains(&hz),
"combined track was not the 440 Hz tone: {hz}"
);
}
}
+203
View File
@@ -0,0 +1,203 @@
//! Single-threaded file decoding. Scale only frames which will be displayed.
use crate::ffmpeg::{self, codec, format, media, software::scaling, util::frame::video::Video};
use std::{path::Path, time::Duration};
#[derive(Clone, Debug)]
pub struct Frame {
pub width: u32,
pub height: u32,
/// Packed BGRA, without FFmpeg's row padding.
pub bgra: Vec<u8>,
}
pub struct Decoder {
input: format::context::Input,
decoder: ffmpeg::decoder::Video,
stream: usize,
time_base: f64,
origin: i64,
fps: u32,
next_index: u64,
eof: bool,
scaler: Option<(u32, u32, format::Pixel, u32, u32, scaling::context::Context)>,
}
impl Decoder {
pub fn open(path: &Path, fps: u32) -> Result<Self, ffmpeg::Error> {
crate::init();
let input = format::input(path)?;
let video = input
.streams()
.best(media::Type::Video)
.ok_or(ffmpeg::Error::StreamNotFound)?;
let stream = video.index();
let time_base = f64::from(video.time_base());
let origin = video.start_time().max(0);
let mut context = codec::context::Context::from_parameters(video.parameters())?;
// Never let FFmpeg create a pool competing with the recorder.
context.set_threading(codec::threading::Config::count(1));
let decoder = context.decoder().video()?;
Ok(Self {
input,
decoder,
stream,
time_base,
origin,
fps: fps.max(1),
next_index: 0,
eof: false,
scaler: None,
})
}
/// Reuse swscale's setup while the preview dimensions stay unchanged.
pub fn scale(&mut self, video: &Video, target: (u32, u32)) -> Result<Frame, ffmpeg::Error> {
let (width, height) = fitted_size((video.width(), video.height()), target);
if self
.scaler
.as_ref()
.is_none_or(|(w, h, pixel, source_w, source_h, _)| {
(*w, *h) != (width, height)
|| *pixel != video.format()
|| (*source_w, *source_h) != (video.width(), video.height())
})
{
self.scaler = Some((
width,
height,
video.format(),
video.width(),
video.height(),
scaling::context::Context::get(
video.format(),
video.width(),
video.height(),
format::Pixel::BGRA,
width,
height,
scaling::flag::Flags::FAST_BILINEAR,
)?,
));
}
let (_, _, _, _, _, scaler) = self.scaler.as_mut().unwrap();
packed_frame(scaler, video, width, height)
}
/// FFmpeg seeks backwards to a keyframe; callers decode forward by frame index.
pub fn seek(&mut self, time: Duration) -> Result<(), ffmpeg::Error> {
let timestamp = (time.as_secs_f64() * 1_000_000.
+ self.origin as f64 * self.time_base * 1_000_000.) as i64;
self.input.seek(timestamp, ..timestamp)?;
self.decoder.flush();
self.next_index = frame_index(time, self.fps);
self.eof = false;
Ok(())
}
pub fn next_frame(&mut self) -> Result<Option<(u64, Video)>, ffmpeg::Error> {
loop {
let mut frame = Video::empty();
match self.decoder.receive_frame(&mut frame) {
Ok(()) => {
let index = frame.timestamp().map_or(self.next_index, |pts| {
(((pts - self.origin).max(0) as f64 * self.time_base * self.fps as f64)
.round()) as u64
});
self.next_index = index + 1;
return Ok(Some((index, frame)));
}
Err(ffmpeg::Error::Eof) => return Ok(None),
Err(ffmpeg::Error::Other { errno }) if errno == ffmpeg::error::EAGAIN => {}
Err(error) => return Err(error),
}
if self.eof {
return Ok(None);
}
let mut packet = ffmpeg::Packet::empty();
loop {
match packet.read(&mut self.input) {
Ok(()) if packet.stream() == self.stream => {
self.decoder.send_packet(&packet)?;
break;
}
Ok(()) => {}
Err(ffmpeg::Error::Eof) => {
self.eof = true;
self.decoder.send_eof()?;
break;
}
Err(error) => return Err(error),
}
}
}
}
}
pub fn frame_index(time: Duration, fps: u32) -> u64 {
(time.as_nanos() * u128::from(fps.max(1)) / 1_000_000_000) as u64
}
/// Preserve aspect ratio and cap allocation even on a very large display.
pub fn fitted_size(source: (u32, u32), target: (u32, u32)) -> (u32, u32) {
let ratio = (target.0.clamp(1, 1920) as f64 / source.0.max(1) as f64)
.min(target.1.clamp(1, 1080) as f64 / source.1.max(1) as f64)
.min(1.);
(
(source.0 as f64 * ratio).round().max(1.) as u32,
(source.1 as f64 * ratio).round().max(1.) as u32,
)
}
pub fn scale(video: &Video, target: (u32, u32)) -> Result<Frame, ffmpeg::Error> {
let (width, height) = fitted_size((video.width(), video.height()), target);
let mut scaler = scaling::context::Context::get(
video.format(),
video.width(),
video.height(),
format::Pixel::BGRA,
width,
height,
scaling::flag::Flags::FAST_BILINEAR,
)?;
packed_frame(&mut scaler, video, width, height)
}
fn packed_frame(
scaler: &mut scaling::context::Context,
video: &Video,
width: u32,
height: u32,
) -> Result<Frame, ffmpeg::Error> {
let mut output = Video::empty();
scaler.run(video, &mut output)?;
let mut bgra = Vec::with_capacity(width as usize * height as usize * 4);
for row in 0..height as usize {
let start = row * output.stride(0);
bgra.extend_from_slice(&output.data(0)[start..start + width as usize * 4]);
}
Ok(Frame {
width,
height,
bgra,
})
}
/// A thumbnail from the first decoded frame of any MP4/MKV path.
pub fn first_frame(path: &Path, target: (u32, u32)) -> Result<Frame, ffmpeg::Error> {
let mut decoder = Decoder::open(path, 60)?;
let (_, frame) = decoder.next_frame()?.ok_or(ffmpeg::Error::Eof)?;
scale(&frame, target)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn frame_exact_seek_and_bounded_aspect_ratio() {
assert_eq!(frame_index(Duration::from_millis(1999), 60), 119);
assert_eq!(frame_index(Duration::from_secs(2), 60), 120);
assert_eq!(fitted_size((2560, 1440), (800, 450)), (800, 450));
assert_eq!(fitted_size((1440, 2560), (800, 450)), (253, 450));
assert_eq!(fitted_size((8000, 8000), (8000, 8000)), (1080, 1080));
}
}
+10
View File
@@ -67,3 +67,13 @@ mod tests {
println!("{probe:?}");
}
}
pub mod audio;
pub mod decode;
pub mod player;
pub mod thumbnail;
mod priority;
#[cfg(windows)]
#[allow(unsafe_code)]
mod windows_priority;
+459
View File
@@ -0,0 +1,459 @@
//! On-demand background playback with a one-frame mailbox and coalesced controls.
use crate::decode::{self, Decoder, Frame};
use clipbin_core::store::{Metadata, Store};
use std::{
path::PathBuf,
sync::{Arc, Condvar, Mutex},
thread,
time::{Duration, Instant},
};
#[derive(Clone, Debug)]
pub enum Picture {
Loading,
Frame(Arc<Frame>),
Gap,
Error(String),
}
#[derive(Clone, Debug)]
pub struct Snapshot {
pub time: Duration,
pub playing: bool,
pub picture: Picture,
pub revision: u64,
pub audio_error: Option<String>,
pub video_time: Option<Duration>,
pub frames_per_second: u32,
}
struct Controls {
session: Option<(Store, u64)>,
seek: Option<Duration>,
latest: bool,
playing: bool,
size: (u32, u32),
generation: u64,
muted: bool,
volume: f32,
quit: bool,
snapshot: Snapshot,
}
struct Shared {
state: Mutex<Controls>,
wake: Condvar,
}
pub struct Player {
shared: Arc<Shared>,
}
impl Default for Player {
fn default() -> Self {
Self::new()
}
}
impl Player {
pub fn new() -> Self {
let shared = Arc::new(Shared {
state: Mutex::new(Controls {
session: None,
seek: None,
latest: false,
playing: false,
size: (800, 450),
generation: 0,
muted: false,
volume: 1.,
quit: false,
snapshot: Snapshot {
time: Duration::ZERO,
playing: false,
picture: Picture::Loading,
revision: 0,
audio_error: None,
video_time: None,
frames_per_second: 60,
},
}),
wake: Condvar::new(),
});
let worker = shared.clone();
thread::Builder::new()
.name("clipbin-player".into())
.spawn(move || run(worker))
.expect("player thread");
Self { shared }
}
pub fn open(&self, store: Store, id: u64, time: Duration) {
self.open_session(store, id, time, false);
}
pub fn open_latest(&self, store: Store, id: u64) {
self.open_session(store, id, Duration::ZERO, true);
}
fn open_session(&self, store: Store, id: u64, time: Duration, latest: bool) {
let mut state = self.shared.state.lock().unwrap();
state.session = Some((store, id));
state.latest = latest;
state.seek = Some(time);
state.playing = false;
state.generation += 1;
state.snapshot.picture = Picture::Loading;
state.snapshot.revision += 1;
self.shared.wake.notify_one();
}
pub fn close(&self) {
let mut state = self.shared.state.lock().unwrap();
state.session = None;
state.playing = false;
state.generation += 1;
self.shared.wake.notify_one();
}
pub fn seek(&self, time: Duration) {
let mut state = self.shared.state.lock().unwrap();
state.seek = Some(time);
state.snapshot.time = time;
state.snapshot.picture = Picture::Loading;
state.snapshot.revision += 1;
state.generation += 1;
self.shared.wake.notify_one();
}
pub fn play(&self, playing: bool) {
self.shared.state.lock().unwrap().playing = playing;
self.shared.wake.notify_one();
}
pub fn audio(&self, muted: bool, volume: f32) {
let mut state = self.shared.state.lock().unwrap();
if state.muted != muted {
state.seek = Some(state.snapshot.time);
state.generation += 1;
}
state.muted = muted;
state.volume = volume.clamp(0., 1.);
self.shared.wake.notify_one();
}
pub fn resize(&self, size: (u32, u32)) {
let mut state = self.shared.state.lock().unwrap();
if state.size != size {
state.size = size;
state.generation += 1;
self.shared.wake.notify_one();
}
}
pub fn snapshot(&self) -> Snapshot {
self.shared.state.lock().unwrap().snapshot.clone()
}
}
impl Drop for Player {
fn drop(&mut self) {
self.shared.state.lock().unwrap().quit = true;
self.shared.wake.notify_one();
}
}
/// Monotonic session clock: decoding delays never accumulate into playback time.
#[derive(Debug)]
pub struct Clock {
base: Duration,
started: Instant,
playing: bool,
}
impl Clock {
pub fn new(now: Instant) -> Self {
Self {
base: Duration::ZERO,
started: now,
playing: false,
}
}
pub fn time(&self, now: Instant) -> Duration {
self.base
+ if self.playing {
now.saturating_duration_since(self.started)
} else {
Duration::ZERO
}
}
pub fn set(&mut self, time: Duration, playing: bool, now: Instant) {
self.base = time;
self.started = now;
self.playing = playing;
}
}
fn run(shared: Arc<Shared>) {
crate::priority::lower(false);
let mut clock = Clock::new(Instant::now());
let mut generation = u64::MAX;
let mut metadata: Option<Metadata> = None;
let mut refreshed = Instant::now();
let mut decoder: Option<(PathBuf, Decoder, u64)> = None;
let mut audio: Option<crate::audio::Audio> = None;
let mut audio_error = None;
let mut audio_attempted = false;
loop {
let (session, seek, playing, size, requested, muted, volume, latest) = {
let mut state = shared.state.lock().unwrap();
if state.quit {
break;
}
(
state.session.clone(),
state.seek.take(),
state.playing,
state.size,
state.generation,
state.muted,
state.volume,
std::mem::take(&mut state.latest),
)
};
let changed = requested != generation;
if changed {
generation = requested;
decoder = None;
}
let now = Instant::now();
if changed || now.duration_since(refreshed) >= Duration::from_millis(250) {
metadata = session
.as_ref()
.and_then(|(store, id)| store.metadata(*id).ok());
refreshed = now;
}
let seek = if latest {
metadata
.as_ref()
.and_then(|metadata| {
metadata
.segments
.iter()
.rev()
.find(|segment| {
segment.complete && segment.available && segment.end > segment.start
})
.map(|segment| {
segment.end.saturating_sub(Duration::from_secs_f64(
1. / segment.frames_per_second.max(1) as f64,
))
})
})
.or(Some(Duration::ZERO))
} else {
seek.map(|time| {
metadata
.as_ref()
.map_or(time, |metadata| time.min(metadata.session.length))
})
};
let transport_changed = playing != clock.playing;
if seek.is_some() || transport_changed {
clock.set(seek.unwrap_or_else(|| clock.time(now)), playing, now);
}
if session.is_none() {
audio = None;
audio_attempted = false;
audio_error = None;
}
if session.is_some() && !audio_attempted && !muted {
audio_attempted = true;
match crate::audio::Audio::new() {
Ok(output) => audio = Some(output),
Err(error) => audio_error = Some(error),
}
}
if let Some(audio) = audio.as_mut() {
if changed || seek.is_some() || transport_changed {
audio.reset(clock.time(now));
}
audio.controls(playing && !muted, volume);
}
let Some(metadata) = metadata.as_ref().filter(|_| session.is_some()) else {
wait(&shared, Duration::from_millis(100));
continue;
};
let mut time = clock.time(now).min(metadata.session.length);
if playing
&& !muted
&& let Some(audio) = audio.as_mut()
{
if let Some(device_time) = audio.time(now)
&& metadata
.locate(device_time)
.is_some_and(|at| at.segment.complete)
{
time = device_time.min(metadata.session.length);
clock.set(time, true, now);
}
audio.catch_up(time);
if let Err(error) = audio.fill(metadata) {
audio_error = Some(format!("Cannot play audio: {error}"));
}
if let Some(error) = audio.error() {
audio_error = Some(error);
}
}
if time >= metadata.session.length && playing {
clock.set(time, false, now);
shared.state.lock().unwrap().playing = false;
}
// Completed files only: active files may still lack seekable cues.
let located = metadata.locate(time).filter(|at| at.segment.complete);
let start = located
.as_ref()
.map_or(Duration::ZERO, |at| at.segment.start);
let frames_per_second = located
.as_ref()
.map_or(60, |at| at.segment.frames_per_second);
let mut picture = Picture::Gap;
let mut presented = false;
let started = Instant::now();
if let Some(at) = located {
let target = decode::frame_index(at.offset, at.segment.frames_per_second);
let result = (|| {
if decoder.as_ref().is_none_or(|(path, _, index)| {
path != &at.segment.path
|| target < *index
|| target > index.saturating_add(120)
}) {
let mut opened = Decoder::open(&at.segment.path, at.segment.frames_per_second)?;
if at.offset > Duration::ZERO {
opened.seek(at.offset)?;
}
decoder = Some((at.segment.path.clone(), opened, u64::MAX));
}
let (_, active, last) = decoder.as_mut().unwrap();
if *last == target && !changed {
return Ok(None);
}
loop {
// A drag or navigation cancels decode-forward work between frames.
let state = shared.state.lock().unwrap();
if state.quit || state.generation != generation {
return Ok(None);
}
drop(state);
let Some((index, frame)) = active.next_frame()? else {
return Ok(Some(Picture::Gap));
};
if index >= target {
*last = index;
return active
.scale(&frame, size)
.map(|frame| Some(Picture::Frame(Arc::new(frame))));
}
}
})();
match result {
Ok(Some(frame)) => {
presented = matches!(frame, Picture::Frame(_));
picture = frame;
}
Ok(None) => picture = shared.state.lock().unwrap().snapshot.picture.clone(),
Err(error) => {
picture = Picture::Error(format!("Cannot read footage: {error}"));
decoder = None;
}
}
} else {
decoder = None;
}
let mut state = shared.state.lock().unwrap();
if state.generation == generation {
let updated = time != state.snapshot.time
|| clock.playing != state.snapshot.playing
|| presented
|| std::mem::discriminant(&picture)
!= std::mem::discriminant(&state.snapshot.picture)
|| audio_error != state.snapshot.audio_error;
let revision = state.snapshot.revision + u64::from(updated);
let video_time = matches!(picture, Picture::Frame(_))
.then(|| {
decoder.as_ref().map(|(_, _, index)| {
start + clipbin_core::store::frame_offset(*index, frames_per_second)
})
})
.flatten();
state.snapshot = Snapshot {
time,
playing: clock.playing,
picture,
revision,
audio_error: audio_error.clone(),
video_time,
frames_per_second,
};
if std::env::var_os("CLIPBIN_TRACE_FRAMES").is_some() && (playing || changed) {
eprintln!(
"playback time={:.3} decode_ms={:.3} generation={generation} presented={presented} state={} audio_time={:.3} video_time={:.3}",
time.as_secs_f64(),
started.elapsed().as_secs_f64() * 1000.,
match &state.snapshot.picture {
Picture::Frame(_) => "frame",
Picture::Gap => "gap",
Picture::Loading => "loading",
Picture::Error(_) => "error",
},
audio
.as_ref()
.and_then(|audio| audio.time(Instant::now()))
.map_or(-1., |time| time.as_secs_f64()),
video_time.map_or(-1., |time| time.as_secs_f64())
);
}
}
drop(state);
wait(
&shared,
if clock.playing {
Duration::from_millis(4)
} else {
Duration::from_millis(100)
},
);
}
}
fn wait(shared: &Shared, timeout: Duration) {
let state = shared.state.lock().unwrap();
if !state.quit && state.seek.is_none() {
drop(shared.wake.wait_timeout(state, timeout).unwrap());
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_decoder_stall_skips_overdue_frames_without_shifting_the_clock() {
let now = Instant::now();
let mut clock = Clock::new(now);
clock.set(Duration::ZERO, true, now);
let first_target = decode::frame_index(clock.time(now + Duration::from_millis(16)), 60);
let after_stall = decode::frame_index(clock.time(now + Duration::from_millis(250)), 60);
assert_eq!(first_target, 0);
assert_eq!(after_stall, 15);
// Scaling only the target discards the intervening frames; their decoding
// cost must never be added to the clock's next presentation deadline.
assert_eq!(
decode::frame_index(clock.time(now + Duration::from_secs(1)), 60),
60
);
}
#[test]
fn clock_pauses_and_seek_does_not_accumulate_decode_delay() {
let now = Instant::now();
let mut clock = Clock::new(now);
clock.set(Duration::from_secs(2), true, now);
assert_eq!(
clock.time(now + Duration::from_secs(3)),
Duration::from_secs(5)
);
clock.set(Duration::from_secs(5), false, now);
assert_eq!(
clock.time(now + Duration::from_secs(100)),
Duration::from_secs(5)
);
clock.set(Duration::from_millis(1500), true, now);
assert_eq!(
decode::frame_index(clock.time(now + Duration::from_millis(100)), 60),
96
);
}
}
+14
View File
@@ -0,0 +1,14 @@
//! Browsing runs below the recorder's normal scheduling priority.
pub fn lower(thumbnail: bool) {
#[cfg(target_os = "linux")]
if let Err(error) = rustix::process::setpriority_process(
Some(rustix::thread::gettid()),
if thumbnail { 10 } else { 3 },
) {
eprintln!("Cannot lower browsing thread priority: {error}");
}
#[cfg(windows)]
crate::windows_priority::lower(thumbnail);
#[cfg(not(any(windows, target_os = "linux")))]
let _ = thumbnail;
}
+81
View File
@@ -0,0 +1,81 @@
//! One background worker and a bounded queue for persistent PNG thumbnails.
use crate::decode;
use std::{
collections::HashSet,
path::{Path, PathBuf},
sync::{Arc, Mutex, mpsc},
thread,
time::Duration,
};
/// Deterministic cache path beside the source; exported MP4s use the same API.
pub fn cache_path(path: &Path) -> PathBuf {
path.with_extension("thumbnail.png")
}
/// Decode any MP4/MKV and atomically publish a small thumbnail.
pub fn generate(path: &Path) -> Result<PathBuf, String> {
let cache = cache_path(path);
if cache.is_file() {
return Ok(cache);
}
let mut frame = decode::first_frame(path, (320, 180)).map_err(|error| error.to_string())?;
let temporary = cache.with_extension("png.part");
for pixel in frame.bgra.as_chunks_mut::<4>().0 {
pixel.swap(0, 2);
}
image::save_buffer_with_format(
&temporary,
&frame.bgra,
frame.width,
frame.height,
image::ColorType::Rgba8,
image::ImageFormat::Png,
)
.map_err(|error| error.to_string())?;
std::fs::rename(&temporary, &cache).map_err(|error| error.to_string())?;
Ok(cache)
}
pub struct Thumbnails {
sender: mpsc::SyncSender<PathBuf>,
pending: Arc<Mutex<HashSet<PathBuf>>>,
}
impl Default for Thumbnails {
fn default() -> Self {
Self::new()
}
}
impl Thumbnails {
pub fn new() -> Self {
let (sender, receiver) = mpsc::sync_channel::<PathBuf>(16);
let pending = Arc::new(Mutex::new(HashSet::new()));
let work = pending.clone();
thread::Builder::new()
.name("clipbin-thumbnails".into())
.spawn(move || {
crate::priority::lower(true);
for path in receiver {
// One decoder thread, one frame, and a duty-cycle cap leave capture priority.
thread::sleep(Duration::from_millis(150));
if let Err(error) = generate(&path) {
eprintln!("thumbnail {}: {error}", path.display());
}
work.lock().unwrap().remove(&path);
}
})
.expect("thumbnail thread");
Self { sender, pending }
}
pub fn request(&self, path: &Path) -> Option<PathBuf> {
let cache = cache_path(path);
if cache.is_file() {
return Some(cache);
}
let mut pending = self.pending.lock().unwrap();
if pending.insert(path.to_owned()) && self.sender.try_send(path.to_owned()).is_err() {
pending.remove(path);
}
None
}
}
@@ -0,0 +1,19 @@
//! The only raw Windows calls in the media crate; no handles escape this bridge.
use windows_sys::Win32::System::Threading::{
GetCurrentThread, SetThreadPriority, THREAD_PRIORITY_BELOW_NORMAL, THREAD_PRIORITY_LOWEST,
};
pub fn lower(thumbnail: bool) {
let priority = if thumbnail {
THREAD_PRIORITY_LOWEST
} else {
THREAD_PRIORITY_BELOW_NORMAL
};
// GetCurrentThread returns a borrowed pseudo-handle, valid for this thread.
if unsafe { SetThreadPriority(GetCurrentThread(), priority) } == 0 {
eprintln!(
"Cannot lower browsing thread priority: {}",
std::io::Error::last_os_error()
);
}
}
+106
View File
@@ -0,0 +1,106 @@
//! Run through scripts/validate/playback-fixture.sh for real FFmpeg seek coverage.
use clipbin_core::store::Store;
use clipbin_media::{
decode::{Decoder, scale},
player::{Picture, Player},
};
use std::{
path::PathBuf,
thread,
time::{Duration, Instant},
};
fn fixture() -> Option<PathBuf> {
std::env::var_os("CLIPBIN_TEST_MEDIA_DIR").map(PathBuf::from)
}
#[test]
fn seeking_matches_sequential_frames_in_each_independent_segment() {
let Some(directory) = fixture() else {
return;
};
for segment in 0..3 {
let path = directory.join(format!("1/segment-{segment:03}.mkv"));
let mut sequential = Decoder::open(&path, 60).unwrap();
let mut expected = Vec::new();
while let Some((index, frame)) = sequential.next_frame().unwrap() {
expected.push((index, scale(&frame, (160, 90)).unwrap().bgra));
}
assert_eq!(expected.len(), 120);
for target in [0, 1, 59, 60, 61, 119] {
let mut decoder = Decoder::open(&path, 60).unwrap();
decoder
.seek(Duration::from_secs_f64(target as f64 / 60.))
.unwrap();
loop {
let (index, frame) = decoder.next_frame().unwrap().unwrap();
if index >= target {
assert_eq!(index, target);
assert_eq!(
scale(&frame, (160, 90)).unwrap().bgra,
expected[target as usize].1
);
break;
}
}
}
}
}
fn wait(player: &Player, at: Duration) -> clipbin_media::player::Snapshot {
let until = Instant::now() + Duration::from_secs(5);
loop {
let snapshot = player.snapshot();
if snapshot.time == at && !matches!(snapshot.picture, Picture::Loading) {
return snapshot;
}
assert!(Instant::now() < until, "seek timed out: {snapshot:?}");
thread::sleep(Duration::from_millis(10));
}
}
#[test]
fn player_crosses_boundaries_and_gaps_and_discards_superseded_seeks() {
let Some(directory) = fixture() else {
return;
};
let player = Player::new();
player.audio(true, 1.);
player.open(Store { root: directory }, 1, Duration::ZERO);
assert!(matches!(
wait(&player, Duration::ZERO).picture,
Picture::Frame(_)
));
for millis in [1999, 2000, 2500, 4000, 5999, 6000, 7000] {
let at = Duration::from_millis(millis);
player.seek(at);
let snapshot = wait(&player, at);
assert_eq!(
matches!(snapshot.picture, Picture::Gap),
(4000..6000).contains(&millis),
"at {millis}"
);
}
for i in 0..50 {
player.seek(Duration::from_millis(i * 40));
}
player.seek(Duration::from_millis(6500));
assert!(matches!(
wait(&player, Duration::from_millis(6500)).picture,
Picture::Frame(_)
));
player.seek(Duration::from_millis(1900));
wait(&player, Duration::from_millis(1900));
player.play(true);
thread::sleep(Duration::from_millis(400));
player.play(false);
let snapshot = player.snapshot();
assert!((Duration::from_millis(2250)..Duration::from_millis(2500)).contains(&snapshot.time));
assert!(matches!(snapshot.picture, Picture::Frame(_)));
assert!(snapshot.video_time.unwrap().abs_diff(snapshot.time) < Duration::from_millis(20));
player.seek(Duration::from_millis(3900));
wait(&player, Duration::from_millis(3900));
player.play(true);
thread::sleep(Duration::from_millis(350));
let snapshot = player.snapshot();
assert!(snapshot.time >= Duration::from_millis(4200));
assert!(matches!(snapshot.picture, Picture::Gap));
player.close();
}