rays
e5b9b0cfdb
A 47 s reap had a trace of one span, so whether the time went on checking every file on disk, the database or deleting could not be told. The reap's steps are spans now, with what it removed on the reap span; an episode list's query, sanitising, DeadAir and count are steps; and DeadAir's work on a file is a trace of its own (probe, silencedetect, frame_bytes, store_plans), only for files it looks at, not the idle passes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
745 lines
33 KiB
Rust
745 lines
33 KiB
Rust
//! DeadAir (issue #171): podcasts play with their long silences shortened, so they finish
|
|
//! sooner without sounding sped up.
|
|
//!
|
|
//! Nothing is copied (#181). DeadAir finds a file's silences once, in the background, and keeps a
|
|
//! `Plan` for each pause length a person can choose (#182): the frames kept, as byte ranges of
|
|
//! the original. The player gets an HLS playlist of those ranges (`playlist`), which Safari plays
|
|
//! itself and other browsers through hls.js, with no gap at the cuts. Seeking past each silence
|
|
//! instead left about 1.3 s of nothing at every one (#178); a copy made with ffmpeg cost as much
|
|
//! disk again as the original and could only have one pause length.
|
|
//!
|
|
//! Positions stay in the original's time. What a plan leaves out is stored as `Cut`s of the
|
|
//! original's time, and the server converts both ways (`to_copy`, `to_original`): the iOS and
|
|
//! Android apps save positions themselves, from their own player, so converting in the page
|
|
//! would miss them.
|
|
|
|
use std::collections::{BTreeMap, HashMap};
|
|
use std::fmt::Write as _;
|
|
use std::path::Path;
|
|
use std::sync::Arc;
|
|
|
|
use tracing::Instrument;
|
|
|
|
use crate::db::step;
|
|
|
|
use anyhow::{Context, Result, bail};
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use crate::Ctx;
|
|
|
|
/// Quieter than this for longer than `MIN_SILENCE` is a silence: what the measurements on the
|
|
/// issue used. A podcast's room tone sits well below it and speech well above.
|
|
const THRESHOLD: &str = "-38dB";
|
|
const MIN_SILENCE: f64 = 0.35;
|
|
/// The pauses a person can have DeadAir leave where it cuts a silence, in ms (#182). Half of it
|
|
/// is kept at each end of the silence, so the frames on both sides of a cut are silent. Decoding
|
|
/// blends a frame with the one before it, and an MP3 frame can begin in the bytes of the frame
|
|
/// before (the bit reservoir): the first frame after a cut decodes against the wrong neighbour,
|
|
/// which nobody hears between two silent frames and everyone hears at the start of a word.
|
|
pub const PAUSES: [i64; 4] = [200, 300, 400, 500];
|
|
/// 0.2 s, the first DeadAir's, made speech sound rushed.
|
|
pub const DEFAULT_PAUSE: i64 = 300;
|
|
/// The longest piece of a playlist, in seconds: about what HLS players expect, so a long stretch
|
|
/// of talk is not one request for megabytes.
|
|
const MAX_PIECE: f64 = 10.0;
|
|
/// How long the maker waits when there is nothing to do: a download or a feed turned on is
|
|
/// picked up this long after, at most.
|
|
const IDLE: std::time::Duration = std::time::Duration::from_secs(120);
|
|
|
|
/// A span of the original, in milliseconds, that DeadAir leaves out.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
|
pub struct Cut(pub i64, pub i64);
|
|
|
|
/// What one pause length leaves of a file: what it cuts, and what it keeps as pieces of the
|
|
/// original, (byte offset, bytes, seconds).
|
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
|
pub struct Plan {
|
|
pub cuts: Vec<Cut>,
|
|
pub pieces: Vec<(u64, u64, f64)>,
|
|
}
|
|
|
|
impl Plan {
|
|
pub fn saved_ms(&self) -> i64 {
|
|
saved(&self.cuts)
|
|
}
|
|
}
|
|
|
|
/// The length of the ID3v2 tag a file starts with, 0 if it has none.
|
|
pub fn id3_len(head: &[u8]) -> u64 {
|
|
if head.len() < 10 || &head[..3] != b"ID3" {
|
|
return 0;
|
|
}
|
|
// Seven bits a byte, so no byte of the size looks like the start of a frame.
|
|
let size = head[6..10].iter().fold(0u64, |n, b| n << 7 | u64::from(b & 0x7f));
|
|
let footer = if head[5] & 0x10 != 0 { 10 } else { 0 };
|
|
10 + size + footer
|
|
}
|
|
|
|
/// A file as `plan` plays it, to save (#186): its ID3 tag, so the title and artwork come too, then
|
|
/// the pieces one after another. MP3 and ADTS frames each stand alone, so that is a file any
|
|
/// player plays. Read as it is sent and never stored, as the playlist is: (its length, its bytes).
|
|
pub async fn kept(
|
|
path: &Path,
|
|
plan: &Plan,
|
|
) -> std::io::Result<(u64, impl futures_util::Stream<Item = std::io::Result<axum::body::Bytes>> + use<>)> {
|
|
use tokio::io::{AsyncReadExt, AsyncSeekExt};
|
|
let mut file = tokio::fs::File::open(path).await?;
|
|
let mut head = [0u8; 10];
|
|
let n = file.read(&mut head).await?;
|
|
// Only up to the first frame kept: past the tag can be the Xing frame, which gives the
|
|
// original's length.
|
|
let first = plan.pieces.first().map_or(0, |p| p.0);
|
|
let tag = id3_len(&head[..n]).min(first);
|
|
let ranges: std::collections::VecDeque<(u64, u64)> =
|
|
std::iter::once((0, tag)).chain(plan.pieces.iter().map(|&(o, n, _)| (o, n))).filter(|r| r.1 > 0).collect();
|
|
let len = ranges.iter().map(|r| r.1).sum();
|
|
let body = futures_util::stream::try_unfold((file, ranges), |(mut f, mut left)| async move {
|
|
let Some((off, n)) = left.pop_front() else { return Ok(None) };
|
|
let mut buf = vec![0; n.min(1 << 16) as usize];
|
|
f.seek(std::io::SeekFrom::Start(off)).await?;
|
|
f.read_exact(&mut buf).await?;
|
|
let got = buf.len() as u64;
|
|
if got < n {
|
|
left.push_front((off + got, n - got));
|
|
}
|
|
Ok(Some((axum::body::Bytes::from(buf), (f, left))))
|
|
});
|
|
Ok((len, body))
|
|
}
|
|
|
|
/// The silences silencedetect reported, as (start, end) seconds. One still open at the end of
|
|
/// the file has no end and is left alone.
|
|
pub fn parse_silences(log: &str) -> Vec<(f64, f64)> {
|
|
let mut out = Vec::new();
|
|
let mut start = None;
|
|
for line in log.lines() {
|
|
if let Some(v) = after(line, "silence_start: ") {
|
|
start = Some(v);
|
|
} else if let Some(v) = after(line, "silence_end: ")
|
|
&& let Some(s) = start.take()
|
|
{
|
|
out.push((s, v));
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
fn after(line: &str, key: &str) -> Option<f64> {
|
|
let rest = &line[line.find(key)? + key.len()..];
|
|
rest.split_whitespace().next()?.parse().ok()
|
|
}
|
|
|
|
/// Which frames to drop for these silences, as inclusive ranges of frame numbers, and the same
|
|
/// as `Cut`s. Frame `n` plays from `first + n * frame` seconds. Only frames wholly inside a
|
|
/// silence, `keep` seconds in from each end, go.
|
|
pub fn plan(silences: &[(f64, f64)], first: f64, frame: f64, keep: f64) -> (Vec<(u64, u64)>, Vec<Cut>) {
|
|
let spans: Vec<(u64, u64)> = silences
|
|
.iter()
|
|
.filter(|(s, e)| e - s >= MIN_SILENCE)
|
|
.filter_map(|&(s, e)| {
|
|
let a = ((s + keep - first) / frame).ceil().max(0.0) as u64;
|
|
let b = ((e - keep - first) / frame).floor() as i64 - 1;
|
|
(b >= a as i64).then_some((a, b as u64))
|
|
})
|
|
.collect();
|
|
let ms = |n: u64| ((first + n as f64 * frame) * 1000.0).round() as i64;
|
|
let cuts = spans.iter().map(|&(a, b)| Cut(ms(a), ms(b + 1))).collect();
|
|
(spans, cuts)
|
|
}
|
|
|
|
/// The frames kept, as pieces of the file: runs of frames lying end to end in it, none past
|
|
/// `MAX_PIECE`. `frames` is each frame's (byte offset, bytes); `drop` is from `plan`.
|
|
pub fn pieces(frames: &[(u64, u64)], drop: &[(u64, u64)], frame: f64) -> Vec<(u64, u64, f64)> {
|
|
let most = ((MAX_PIECE / frame).floor() as usize).max(1);
|
|
let mut out = Vec::new();
|
|
let push = |out: &mut Vec<(u64, u64, f64)>, (first, n): (usize, usize)| {
|
|
let (start, (pos, size)) = (frames[first].0, frames[first + n - 1]);
|
|
out.push((start, pos + size - start, n as f64 * frame));
|
|
};
|
|
let mut drops = drop.iter().peekable();
|
|
let mut run: Option<(usize, usize)> = None;
|
|
for (n, &(pos, _)) in frames.iter().enumerate() {
|
|
while drops.peek().is_some_and(|&&(_, b)| (b as usize) < n) {
|
|
drops.next();
|
|
}
|
|
let dropped = drops.peek().is_some_and(|&&(a, _)| a as usize <= n);
|
|
// Something between two frames (an ID3 tag partway through) ends a piece too.
|
|
let joins = !dropped
|
|
&& run.is_some_and(|(f, c)| c < most && frames[f + c - 1].0 + frames[f + c - 1].1 == pos);
|
|
if !joins && let Some(r) = run.take() {
|
|
push(&mut out, r);
|
|
}
|
|
if !dropped {
|
|
run = Some(run.map_or((n, 1), |(f, c)| (f, c + 1)));
|
|
}
|
|
}
|
|
if let Some(r) = run {
|
|
push(&mut out, r);
|
|
}
|
|
out
|
|
}
|
|
|
|
/// The HLS playlist for a plan: each piece a byte range of `uri`, the original.
|
|
pub fn playlist(plan: &Plan, uri: &str) -> String {
|
|
let longest = plan.pieces.iter().map(|p| p.2).fold(0.0, f64::max);
|
|
let mut s = format!(
|
|
"#EXTM3U\n#EXT-X-VERSION:4\n#EXT-X-TARGETDURATION:{}\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-MEDIA-SEQUENCE:0\n",
|
|
longest.ceil() as u64
|
|
);
|
|
for &(off, len, secs) in &plan.pieces {
|
|
let _ = write!(s, "#EXTINF:{secs:.6},\n#EXT-X-BYTERANGE:{len}@{off}\n{uri}\n");
|
|
}
|
|
s + "#EXT-X-ENDLIST\n"
|
|
}
|
|
|
|
/// A time in the original, in ms, as the same moment in the shortened version. A time inside a
|
|
/// cut is where the cut was made.
|
|
pub fn to_copy(cuts: &[Cut], ms: i64) -> i64 {
|
|
let mut out = ms;
|
|
for &Cut(s, e) in cuts {
|
|
if ms >= e {
|
|
out -= e - s;
|
|
} else {
|
|
if ms > s {
|
|
out -= ms - s;
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
/// A time in the shortened version, in ms, as the same moment in the original.
|
|
pub fn to_original(cuts: &[Cut], ms: i64) -> i64 {
|
|
let mut out = ms;
|
|
for &Cut(s, e) in cuts {
|
|
if out >= s {
|
|
out += e - s;
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
pub fn saved(cuts: &[Cut]) -> i64 {
|
|
cuts.iter().map(|Cut(s, e)| e - s).sum()
|
|
}
|
|
|
|
/// Whether someone with these settings plays a feed shortened. Off for every feed until they
|
|
/// turn it on for that one, or for the group it came in (an OPML or a Patreon creator), which a
|
|
/// feed of the group can still say otherwise to.
|
|
pub fn wants(prefs: &HashMap<String, bool>, feed: &str, group: Option<&str>) -> bool {
|
|
prefs.get(feed).or_else(|| group.and_then(|g| prefs.get(g))).copied().unwrap_or(false)
|
|
}
|
|
|
|
/// The settings of each person with DeadAir on for something, and each managed feed's group,
|
|
/// which is what `wants` and `pause` need.
|
|
pub struct Wanting {
|
|
pub people: HashMap<i64, HashMap<String, bool>>,
|
|
pub pauses: HashMap<(i64, String), i64>,
|
|
pub groups: HashMap<String, String>,
|
|
}
|
|
|
|
impl Wanting {
|
|
pub async fn load(ctx: &Ctx, only: Option<i64>) -> Result<Self> {
|
|
let mut people: HashMap<i64, HashMap<String, bool>> = HashMap::new();
|
|
let mut pauses = HashMap::new();
|
|
for (user, feed, wanted, pause) in ctx.db.deadair_prefs(only).await? {
|
|
if let Some(p) = pause {
|
|
pauses.insert((user, feed.clone()), p);
|
|
}
|
|
people.entry(user).or_default().insert(feed, wanted);
|
|
}
|
|
people.retain(|_, p| p.values().any(|&w| w));
|
|
let groups = if people.is_empty() {
|
|
HashMap::new()
|
|
} else {
|
|
ctx.db.managed_feeds().await?.into_iter().map(|m| (m.id, m.group_id)).collect()
|
|
};
|
|
Ok(Self { people, pauses, groups })
|
|
}
|
|
|
|
pub fn wants(&self, user: i64, feed: &str) -> bool {
|
|
self.people
|
|
.get(&user)
|
|
.is_some_and(|p| wants(p, feed, self.groups.get(feed).map(String::as_str)))
|
|
}
|
|
|
|
/// The pause this person has DeadAir leave in a feed: the feed's, else its group's.
|
|
pub fn pause(&self, user: i64, feed: &str) -> i64 {
|
|
let group = self.groups.get(feed);
|
|
self.pauses
|
|
.get(&(user, feed.to_owned()))
|
|
.or_else(|| group.and_then(|g| self.pauses.get(&(user, g.clone()))))
|
|
.copied()
|
|
.unwrap_or(DEFAULT_PAUSE)
|
|
}
|
|
}
|
|
|
|
/// The plan someone plays a file with: their pause's, for a file in a feed they have DeadAir on
|
|
/// for, once its silences are found. None if it cuts nothing.
|
|
fn chosen(wanting: &Wanting, user: i64, feed: &str, plans: &BTreeMap<i64, Plan>) -> Option<Plan> {
|
|
if !wanting.wants(user, feed) {
|
|
return None;
|
|
}
|
|
plans.get(&wanting.pause(user, feed)).filter(|p| !p.cuts.is_empty()).cloned()
|
|
}
|
|
|
|
/// What each person is playing shortened, by (person, feed, guid): set when they are handed the
|
|
/// playlist and cleared when `/media` serves them the original, so a saved position is converted
|
|
/// from the clock it was measured in, whether the page saved it or a native app did.
|
|
// shortcut: in memory, so a restart forgets it. A position saved before the player next asks
|
|
// for the playlist is then taken as the original's, which resumes a little early, never late.
|
|
pub type Playing = Arc<std::sync::Mutex<HashMap<(i64, String, String), Arc<Plan>>>>;
|
|
|
|
/// The plan for one file as this person would play it, for the playlist route.
|
|
pub async fn plan_for(ctx: &Ctx, user: i64, feed: &str, enclosure: i64) -> Result<Option<Plan>> {
|
|
let wanting = Wanting::load(ctx, Some(user)).await?;
|
|
let plans = ctx.db.plans(&[enclosure]).await?;
|
|
Ok(plans.get(&enclosure).and_then(|p| chosen(&wanting, user, feed, p)))
|
|
}
|
|
|
|
/// Puts a page of entries in the clock of what this person will play: for each file they play
|
|
/// shortened, `deadair` is the seconds it saves, and the entry's position and length are the
|
|
/// shortened version's. The page plays `/media/{id}/deadair.m3u8` for a file that has it.
|
|
pub async fn for_reader(ctx: &Ctx, user: i64, rows: &mut [crate::db::EntryRow]) -> Result<()> {
|
|
let wanting = Wanting::load(ctx, Some(user)).await?;
|
|
if wanting.people.is_empty() {
|
|
return Ok(());
|
|
}
|
|
let ids: Vec<i64> = rows
|
|
.iter()
|
|
.filter(|r| wanting.wants(user, &r.feed_id))
|
|
.flat_map(|r| r.enclosures.iter().filter(|e| e.path.is_some()).map(|e| e.id))
|
|
.collect();
|
|
let plans = ctx.db.plans(&ids).await?;
|
|
for r in rows.iter_mut() {
|
|
// Heard through already: the maker passes it over, so nothing is coming.
|
|
let finished = r.duration.is_some_and(|d| d > 0 && r.position * 10 >= d * 9);
|
|
let mut converted = false;
|
|
for e in &mut r.enclosures {
|
|
e.deadair_pending = ids.contains(&e.id) && e.state == "done" && !plans.contains_key(&e.id) && !finished;
|
|
let Some(p) = plans.get(&e.id).and_then(|p| chosen(&wanting, user, &r.feed_id, p)) else { continue };
|
|
e.deadair = Some(p.saved_ms() / 1000);
|
|
// One position an item, however many files: the first played shortened sets the clock.
|
|
if !std::mem::replace(&mut converted, true) {
|
|
r.position = to_copy(&p.cuts, r.position * 1000) / 1000;
|
|
r.duration = r.duration.map(|d| to_copy(&p.cuts, d * 1000) / 1000);
|
|
}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// A position and length a player saved, in the clock of what it was playing, as the
|
|
/// original's. Past 75% of the shortened version, the episode also counts towards the time
|
|
/// DeadAir has saved this person.
|
|
pub async fn on_save(
|
|
ctx: &Ctx,
|
|
playing: &Playing,
|
|
user: i64,
|
|
feed: &str,
|
|
guid: &str,
|
|
secs: i64,
|
|
duration: Option<i64>,
|
|
) -> Result<(i64, Option<i64>)> {
|
|
let plan = playing.lock().unwrap().get(&(user, feed.to_owned(), guid.to_owned())).cloned();
|
|
let Some(p) = plan else { return Ok((secs, duration)) };
|
|
let original = |s: i64| to_original(&p.cuts, s * 1000) / 1000;
|
|
let (secs, duration) = (original(secs), duration.map(original));
|
|
let len = match duration {
|
|
Some(d) => Some(d),
|
|
None => ctx.db.entry_duration(user, feed, guid).await?,
|
|
};
|
|
if let Some(len) = len
|
|
&& len > 0
|
|
&& secs * 4 >= len * 3
|
|
{
|
|
ctx.db.count_deadair(user, feed, guid, p.saved_ms()).await?;
|
|
}
|
|
Ok((secs, duration))
|
|
}
|
|
|
|
/// Finds silences until stopped: in the next file anyone wants shortened, or a wait of `IDLE`
|
|
/// when there is none.
|
|
pub async fn run(ctx: Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
|
|
if !have_ffmpeg().await {
|
|
tracing::warn!("ffmpeg is not installed, so DeadAir shortens nothing");
|
|
return;
|
|
}
|
|
// Files ffmpeg failed on, passed over until the next start. Not stored: a failure recorded
|
|
// as the file having nothing to cut was never tried again, even once a fix was deployed (#174).
|
|
let mut failed = std::collections::HashSet::new();
|
|
loop {
|
|
// Dropping the pass on a signal kills the ffmpeg it is running (`kill_on_drop`).
|
|
let done = tokio::select! {
|
|
_ = stop.changed() => return,
|
|
done = next(&ctx, &mut failed) => done,
|
|
};
|
|
match done {
|
|
Ok(true) => continue,
|
|
Ok(false) => {}
|
|
Err(e) => tracing::warn!(error = %format!("{e:#}"), "DeadAir could not read a file"),
|
|
}
|
|
tokio::select! {
|
|
_ = stop.changed() => return,
|
|
_ = ctx.deadair.notified() => {}
|
|
_ = tokio::time::sleep(IDLE) => {}
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn have_ffmpeg() -> bool {
|
|
tokio::process::Command::new("ffmpeg")
|
|
.arg("-version")
|
|
.stdout(std::process::Stdio::null())
|
|
.status()
|
|
.await
|
|
.is_ok_and(|s| s.success())
|
|
}
|
|
|
|
/// Plans one file someone wants shortened, if there is one. A file everyone who would play it
|
|
/// shortened has already heard through gets none: finished episodes are rarely played again.
|
|
// shortcut: those stay on the list and are looked at again each pass, one query each; record
|
|
// them as passed over if a large library makes the pass slow.
|
|
async fn next(ctx: &Ctx, failed: &mut std::collections::HashSet<i64>) -> Result<bool> {
|
|
// Nobody with DeadAir on for anything, which is most passes: nothing more to ask the database.
|
|
let wanting = Wanting::load(ctx, None).await?;
|
|
if wanting.people.is_empty() {
|
|
return Ok(false);
|
|
}
|
|
for t in ctx.db.deadair_todo().await? {
|
|
if failed.contains(&t.id) {
|
|
continue;
|
|
}
|
|
let users: Vec<i64> =
|
|
wanting.people.keys().copied().filter(|&u| wanting.wants(u, &t.feed_id)).collect();
|
|
if users.is_empty() || !ctx.db.subscribed_any(&users, &t.feed_id).await? {
|
|
continue;
|
|
}
|
|
let finished = ctx.db.finished(&t.feed_id, &t.guid).await?;
|
|
if users.iter().all(|u| finished.contains(u)) {
|
|
continue;
|
|
}
|
|
let src = crate::download::playable(Path::new(&t.path));
|
|
let started = std::time::Instant::now();
|
|
// A trace of its own for each file looked at, not for the passes that find nothing to do,
|
|
// which are most of them.
|
|
let span = tracing::info_span!("deadair", feed = t.feed_id, enclosure = t.id, cuts = tracing::field::Empty);
|
|
let plans = match analyse(&src).instrument(span.clone()).await {
|
|
Ok(p) => p.unwrap_or_default(),
|
|
Err(e) => {
|
|
tracing::warn!(path = %src.display(), error = %format!("{e:#}"), "DeadAir could not read a file");
|
|
failed.insert(t.id);
|
|
return Ok(true);
|
|
}
|
|
};
|
|
ctx.db.store_plans(t.id, &serde_json::to_string(&plans)?).instrument(span.in_scope(|| step("store_plans"))).await?;
|
|
let usual = plans.get(&DEFAULT_PAUSE);
|
|
span.record("cuts", usual.map_or(0, |p| p.cuts.len()));
|
|
let saved_ms = usual.map_or(0, Plan::saved_ms);
|
|
ctx.out.emit(crate::ipc::Event::DeadairDone { feed: t.feed_id.clone(), enclosure: t.id, saved_s: saved_ms / 1000 });
|
|
// The event says it too (ev="deadair_done"); this line has how long it took and how many cuts.
|
|
tracing::info!(
|
|
feed = t.feed_id,
|
|
path = %src.display(),
|
|
cuts = usual.map_or(0, |p| p.cuts.len()),
|
|
saved_s = saved_ms / 1000,
|
|
took_ms = started.elapsed().as_millis() as u64,
|
|
"DeadAir found the silences"
|
|
);
|
|
return Ok(true);
|
|
}
|
|
Ok(false)
|
|
}
|
|
|
|
/// A plan for each of `PAUSES`, for one file. None for a file DeadAir does not apply to: video,
|
|
/// anything but MP3 or AAC on its own (ADTS; AAC in MP4 has no byte ranges a playlist can
|
|
/// use), or one with no silence long enough to cut.
|
|
pub async fn analyse(src: &Path) -> Result<Option<BTreeMap<i64, Plan>>> {
|
|
let Some((first, frame)) = probe(src).instrument(step("probe")).await? else { return Ok(None) };
|
|
let path = src.display().to_string();
|
|
let log = ffmpeg(&[
|
|
"-hide_banner", "-nostats", "-i", &path, "-map", "0:a:0",
|
|
"-af", &format!("silencedetect=n={THRESHOLD}:d={MIN_SILENCE}"), "-f", "null", "-",
|
|
])
|
|
.instrument(step("silencedetect"))
|
|
.await
|
|
.context("finding the silences")?;
|
|
let silences = parse_silences(&log);
|
|
let frames = frame_bytes(&path).instrument(step("frame_bytes")).await?;
|
|
let plans: BTreeMap<i64, Plan> = PAUSES
|
|
.iter()
|
|
.map(|&pause| {
|
|
let (drop, cuts) = plan(&silences, first, frame, pause as f64 / 2000.0);
|
|
(pause, Plan { cuts, pieces: pieces(&frames, &drop, frame) })
|
|
})
|
|
.collect();
|
|
Ok(plans.values().any(|p| !p.cuts.is_empty()).then_some(plans))
|
|
}
|
|
|
|
/// Where each audio frame is in the file, as (byte offset, bytes), in order.
|
|
async fn frame_bytes(path: &str) -> Result<Vec<(u64, u64)>> {
|
|
let out = run_out("ffprobe", &[
|
|
"-v", "error", "-select_streams", "a:0", "-show_entries", "packet=pos,size", "-of", "compact=p=0", path,
|
|
])
|
|
.await?;
|
|
// ffprobe writes the fields in its own order, not the order asked for: by name, then.
|
|
out.lines()
|
|
.map(|l| {
|
|
let field = |k: &str| {
|
|
l.split('|').find_map(|f| f.strip_prefix(k)?.strip_prefix('=')?.parse::<u64>().ok())
|
|
};
|
|
field("pos").zip(field("size")).with_context(|| format!("reading ffprobe's frame: {l}"))
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// When the first audio frame starts and how long each lasts, in seconds, for a file DeadAir
|
|
/// can shorten; None for one it cannot. A cover picture in an MP3 is not video.
|
|
async fn probe(src: &Path) -> Result<Option<(f64, f64)>> {
|
|
let path = src.display().to_string();
|
|
let out = run_out("ffprobe", &[
|
|
"-v", "error", "-show_entries",
|
|
"format=format_name:stream=codec_type,codec_name:stream_disposition=attached_pic", "-of", "json", &path,
|
|
])
|
|
.await?;
|
|
let v: serde_json::Value = serde_json::from_str(&out).context("reading ffprobe")?;
|
|
let streams = v["streams"].as_array().cloned().unwrap_or_default();
|
|
let video = streams
|
|
.iter()
|
|
.any(|s| s["codec_type"] == "video" && s["disposition"]["attached_pic"] != 1);
|
|
let audio = streams.iter().find(|s| s["codec_type"] == "audio");
|
|
let format = v["format"]["format_name"].as_str().unwrap_or("");
|
|
let plain = |codec: &str| format == codec && audio.is_some_and(|a| a["codec_name"] == codec);
|
|
if video || !(plain("mp3") || plain("aac")) {
|
|
return Ok(None);
|
|
}
|
|
let out = run_out("ffprobe", &[
|
|
"-v", "error", "-select_streams", "a:0", "-show_entries", "packet=pts_time,duration_time",
|
|
"-read_intervals", "%+#1", "-of", "csv=p=0", &path,
|
|
])
|
|
.await?;
|
|
let mut f = out.trim().split(',').map(|x| x.trim().parse::<f64>());
|
|
match (f.next(), f.next()) {
|
|
(Some(Ok(first)), Some(Ok(frame))) if frame > 0.0 => Ok(Some((first, frame))),
|
|
_ => bail!("no audio frame in {path}"),
|
|
}
|
|
}
|
|
|
|
/// Runs ffmpeg and gives back what it wrote to stderr, where silencedetect reports.
|
|
async fn ffmpeg(args: &[&str]) -> Result<String> {
|
|
let out = command("ffmpeg").args(["-nostdin", "-threads", "1"]).args(args).output().await?;
|
|
let log = String::from_utf8_lossy(&out.stderr).into_owned();
|
|
if !out.status.success() {
|
|
bail!("ffmpeg failed: {}", log.lines().rev().take(3).collect::<Vec<_>>().join(" / "));
|
|
}
|
|
Ok(log)
|
|
}
|
|
|
|
async fn run_out(prog: &str, args: &[&str]) -> Result<String> {
|
|
let out = command(prog).args(args).output().await.with_context(|| format!("running {prog}"))?;
|
|
if !out.status.success() {
|
|
bail!("{prog} failed: {}", String::from_utf8_lossy(&out.stderr).trim());
|
|
}
|
|
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
|
|
}
|
|
|
|
fn command(prog: &str) -> tokio::process::Command {
|
|
let mut c = tokio::process::Command::new(prog);
|
|
c.kill_on_drop(true).stdin(std::process::Stdio::null());
|
|
c
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn silences_are_read_from_the_log_and_an_open_one_is_left() {
|
|
let log = "[silencedetect @ 0x1] silence_start: 1\n\
|
|
[silencedetect @ 0x1] silence_end: 3.00002 | silence_duration: 2.00002\n\
|
|
[silencedetect @ 0x1] silence_start: 10.5\n";
|
|
assert_eq!(parse_silences(log), vec![(1.0, 3.00002)]);
|
|
}
|
|
|
|
#[test]
|
|
fn cuts_leave_silent_frames_either_side_and_skip_short_silences() {
|
|
// MP3 frames, 26.122 ms; one silence of 2 s, one too short to touch; a 0.3 s pause.
|
|
let frame = 1152.0 / 44100.0;
|
|
let (frames, cuts) = plan(&[(1.0, 3.0), (5.0, 5.3)], 0.0, frame, 0.15);
|
|
assert_eq!(frames, vec![(45, 108)]);
|
|
let Cut(s, e) = cuts[0];
|
|
assert!(s >= 1150 && e <= 2850, "{cuts:?}");
|
|
assert!(s - 1150 < 27 && 2850 - e < 27, "no more than a frame kept beyond half the pause: {cuts:?}");
|
|
// AAC starts a frame early, for the encoder's priming.
|
|
let (frames, _) = plan(&[(1.0, 3.0)], -0.02322, 1024.0 / 44100.0, 0.15);
|
|
assert_eq!(frames, vec![(51, 122)]);
|
|
// A silence the margins use up entirely is left alone.
|
|
assert!(plan(&[(1.0, 1.35)], 0.0, frame, 0.15).0.is_empty());
|
|
// A longer pause cuts less of the same silence.
|
|
assert!(saved(&plan(&[(1.0, 3.0)], 0.0, frame, 0.25).1) < saved(&cuts));
|
|
}
|
|
|
|
#[test]
|
|
fn pieces_are_the_frames_kept_end_to_end_and_never_too_long() {
|
|
// 1000 frames of 100 bytes after a 50-byte tag, an ID3 tag of 7 bytes before frame 600.
|
|
let frames: Vec<(u64, u64)> =
|
|
(0..1000u64).map(|n| (50 + n * 100 + if n >= 600 { 7 } else { 0 }, 100)).collect();
|
|
let frame = 0.026;
|
|
let p = pieces(&frames, &[(100, 199), (900, 999)], frame);
|
|
let at: Vec<u64> = p.iter().map(|x| x.0).collect();
|
|
// 0-99; 200-583 and 584-599, split at 384 frames (10 s); 600-899 after the tag.
|
|
assert_eq!(at, [50, 20_050, 58_450, 60_057]);
|
|
assert_eq!(p[0], (50, 10_000, 100.0 * frame));
|
|
assert_eq!(p[2].1, 1600, "frames 584-599");
|
|
assert!(p.iter().all(|x| x.2 <= MAX_PIECE));
|
|
let kept: u64 = p.iter().map(|x| x.1).sum();
|
|
assert_eq!(kept, 800 * 100, "every kept frame once, the tag in none");
|
|
}
|
|
|
|
#[test]
|
|
fn the_playlist_lists_each_piece_as_a_range_of_the_original() {
|
|
let plan = Plan { cuts: vec![Cut(1000, 2000)], pieces: vec![(50, 400, 1.0), (500, 4000, 9.5)] };
|
|
assert_eq!(
|
|
playlist(&plan, "deadair.mp3"),
|
|
"#EXTM3U\n#EXT-X-VERSION:4\n#EXT-X-TARGETDURATION:10\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-MEDIA-SEQUENCE:0\n\
|
|
#EXTINF:1.000000,\n#EXT-X-BYTERANGE:400@50\ndeadair.mp3\n\
|
|
#EXTINF:9.500000,\n#EXT-X-BYTERANGE:4000@500\ndeadair.mp3\n#EXT-X-ENDLIST\n"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn positions_convert_both_ways() {
|
|
let cuts = [Cut(10_000, 20_000), Cut(30_000, 35_000)];
|
|
assert_eq!(to_copy(&cuts, 5_000), 5_000);
|
|
assert_eq!(to_copy(&cuts, 15_000), 10_000, "inside a cut is where it was made");
|
|
assert_eq!(to_copy(&cuts, 25_000), 15_000);
|
|
assert_eq!(to_copy(&cuts, 40_000), 25_000);
|
|
for t in [0, 5_000, 10_000, 15_000, 19_999, 20_000, 25_000, 40_000] {
|
|
let back = to_original(&cuts, to_copy(&cuts, t));
|
|
// A time inside a cut comes back as the cut's end: the shortened version has nothing between.
|
|
let expect = if (10_000..20_000).contains(&t) { 20_000 } else { t };
|
|
assert_eq!(back, expect, "{t}");
|
|
}
|
|
assert_eq!(saved(&cuts), 15_000);
|
|
}
|
|
|
|
#[test]
|
|
fn off_for_every_feed_until_turned_on_for_it_or_its_group() {
|
|
let p = |v: &[(&str, bool)]| v.iter().map(|(k, w)| (k.to_string(), *w)).collect::<HashMap<_, _>>();
|
|
assert!(!wants(&p(&[]), "a", None), "off until turned on for the feed");
|
|
assert!(!wants(&p(&[("b", true)]), "a", None), "another feed's says nothing of this one");
|
|
assert!(wants(&p(&[("a", true)]), "a", None));
|
|
assert!(wants(&p(&[("g", true)]), "a", Some("g")), "on for its group");
|
|
assert!(!wants(&p(&[("g", true), ("a", false)]), "a", Some("g")), "kept out of its group's");
|
|
}
|
|
|
|
#[test]
|
|
fn the_pause_is_the_feeds_then_its_groups_then_the_default() {
|
|
let w = Wanting {
|
|
people: HashMap::new(),
|
|
pauses: [((1, "a".to_string()), 500), ((1, "g".to_string()), 400)].into(),
|
|
groups: [("b".to_string(), "g".to_string())].into(),
|
|
};
|
|
assert_eq!(w.pause(1, "a"), 500);
|
|
assert_eq!(w.pause(1, "b"), 400, "its group's");
|
|
assert_eq!(w.pause(1, "c"), DEFAULT_PAUSE);
|
|
assert_eq!(w.pause(2, "a"), DEFAULT_PAUSE, "someone else's says nothing of yours");
|
|
}
|
|
|
|
#[test]
|
|
fn id3_len_reads_the_tags_size() {
|
|
assert_eq!(id3_len(b"ID3\x04\x00\x00\x00\x00\x02\x01xx"), 10 + 257);
|
|
assert_eq!(id3_len(b"ID3\x04\x00\x10\x00\x00\x02\x01"), 10 + 257 + 10, "with a footer");
|
|
assert_eq!(id3_len(b"\xff\xfb\x90\x00\x00\x00\x00\x00\x00\x00"), 0);
|
|
assert_eq!(id3_len(b"ID3"), 0);
|
|
}
|
|
|
|
/// The whole of it, on a clip ffmpeg makes: a second of tone, two of silence, and so on,
|
|
/// played back through its playlist by ffmpeg's own HLS reader. Skipped where ffmpeg is not
|
|
/// installed.
|
|
#[tokio::test]
|
|
async fn the_playlist_of_a_file_is_shorter_by_what_was_cut_and_plays() {
|
|
if !have_ffmpeg().await {
|
|
eprintln!("no ffmpeg; skipped");
|
|
return;
|
|
}
|
|
let dir = std::env::temp_dir().join(format!("ipx-deadair-{}", std::process::id()));
|
|
std::fs::create_dir_all(&dir).unwrap();
|
|
for (ext, codec) in [("mp3", "libmp3lame"), ("aac", "aac"), ("m4a", "aac")] {
|
|
let src = dir.join(format!("t.{ext}"));
|
|
ffmpeg(&[
|
|
"-v", "error", "-y", "-f", "lavfi", "-i",
|
|
"aevalsrc='if(lt(mod(t,3),1),0.5*sin(2*PI*440*t),0)':d=12:s=44100",
|
|
"-c:a", codec, &src.display().to_string(),
|
|
])
|
|
.await
|
|
.unwrap();
|
|
let plans = analyse(&src).await.unwrap();
|
|
if ext == "m4a" {
|
|
assert!(plans.is_none(), "AAC in MP4 has no byte ranges to play");
|
|
continue;
|
|
}
|
|
let plans = plans.expect("plans");
|
|
assert_eq!(plans.keys().copied().collect::<Vec<_>>(), PAUSES);
|
|
let p = &plans[&DEFAULT_PAUSE];
|
|
assert_eq!(p.cuts.len(), 4, "{ext}: {:?}", p.cuts);
|
|
// About 1.7 s from each of the four 2 s silences, and less for a longer pause.
|
|
let cut = p.saved_ms();
|
|
assert!((6_500..6_900).contains(&cut), "{ext}: {cut}");
|
|
assert!(plans[&500].saved_ms() < cut && cut < plans[&200].saved_ms());
|
|
let list = dir.join(format!("{ext}.m3u8"));
|
|
std::fs::write(&list, playlist(p, &format!("t.{ext}"))).unwrap();
|
|
// Its frames added up: a raw AAC file's duration is only guessed from its bitrate.
|
|
let len = |p: &Path| {
|
|
let p = p.display().to_string();
|
|
async move {
|
|
run_out("ffprobe", &["-v", "error", "-select_streams", "a:0", "-show_entries", "packet=duration_time", "-of", "csv=p=0", &p])
|
|
.await
|
|
.unwrap()
|
|
.lines()
|
|
.filter_map(|l| l.trim().trim_end_matches(',').parse::<f64>().ok())
|
|
.sum::<f64>()
|
|
}
|
|
};
|
|
let short = len(&src).await - len(&list).await;
|
|
assert!((short * 1000.0 - cut as f64).abs() < 100.0, "{ext}: {short} s shorter, {cut} ms cut");
|
|
// Decodes without a complaint.
|
|
assert!(ffmpeg(&["-v", "error", "-i", &list.display().to_string(), "-f", "null", "-"]).await.unwrap().is_empty());
|
|
// Saved as a file of its own (#186): as long as the playlist, and it plays.
|
|
let (n, body) = kept(&src, p).await.unwrap();
|
|
let parts: Vec<_> = futures_util::TryStreamExt::try_collect(body).await.unwrap();
|
|
let bytes = parts.concat();
|
|
assert_eq!(bytes.len() as u64, n);
|
|
if ext == "mp3" {
|
|
assert!(bytes.starts_with(b"ID3"), "the original's tag comes too");
|
|
}
|
|
let saved = dir.join(format!("saved.{ext}"));
|
|
std::fs::write(&saved, &bytes).unwrap();
|
|
assert!((len(&saved).await - len(&list).await).abs() < 0.05, "{ext}");
|
|
assert!(ffmpeg(&["-v", "error", "-i", &saved.display().to_string(), "-f", "null", "-"]).await.unwrap().is_empty());
|
|
}
|
|
std::fs::remove_dir_all(&dir).unwrap();
|
|
}
|
|
}
|
|
|
|
/// A real episode, for trying the cut on something other than a test tone:
|
|
/// `IPX_DEADAIR_FILE=/path/ep.mp3 cargo test real_file -- --ignored --nocapture`. Each pause's
|
|
/// playlist is left beside it, `ep.mp3.300.m3u8`, to listen to.
|
|
#[cfg(test)]
|
|
#[tokio::test]
|
|
#[ignore]
|
|
async fn real_file() {
|
|
let src = std::path::PathBuf::from(std::env::var("IPX_DEADAIR_FILE").expect("IPX_DEADAIR_FILE"));
|
|
let t = std::time::Instant::now();
|
|
let plans = analyse(&src).await.unwrap().expect("plans");
|
|
eprintln!("analysed in {:.1?}", t.elapsed());
|
|
let name = src.file_name().unwrap().to_string_lossy().into_owned();
|
|
for (pause, p) in &plans {
|
|
std::fs::write(src.with_file_name(format!("{name}.{pause}.m3u8")), playlist(p, &name)).unwrap();
|
|
eprintln!("{pause} ms: {} cuts, {:.1} min saved, {} pieces", p.cuts.len(), p.saved_ms() as f64 / 60_000.0, p.pieces.len());
|
|
}
|
|
}
|