use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
use meet_audio::pcm::{self, WavWriter};
use meet_audio::source::{Capture, CaptureError};
use crate::transcript::Speaker;
pub struct Track {
pub speaker: Speaker,
pub capture: Box<dyn Capture>,
pub path: Option<PathBuf>,
}
pub const SILENCE_DBFS: f32 = -60.0;
pub const SILENCE_AFTER: Duration = Duration::from_secs(60);
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum Event {
Level { speaker: Speaker, dbfs: f32 },
TrackFailed { speaker: Speaker, error: String },
Silent { speaker: Speaker, seconds: u64 },
Heard { speaker: Speaker },
}
pub type Emit = Arc<dyn Fn(Event) + Send + Sync>;
#[derive(Debug, Clone, PartialEq)]
pub struct Written {
pub speaker: Speaker,
pub duration_ms: u64,
}
pub struct Session {
stop: Arc<AtomicBool>,
paused: Arc<AtomicBool>,
threads: Vec<JoinHandle<Option<Written>>>,
pub started: Instant,
}
impl Session {
pub fn start(tracks: Vec<Track>, emit: Emit) -> Result<Self, String> {
Self::start_with(tracks, emit, SILENCE_AFTER)
}
pub fn start_with(tracks: Vec<Track>, emit: Emit, silence_after: Duration) -> Result<Self, String> {
let stop = Arc::new(AtomicBool::new(false));
let paused = Arc::new(AtomicBool::new(false));
let mut threads = Vec::new();
for track in tracks {
let writer = match &track.path {
Some(path) => Some(WavWriter::create(path).map_err(|e| format!("{}: {e}", path.display()))?),
None => None,
};
let (stop, paused, emit) = (stop.clone(), paused.clone(), emit.clone());
threads.push(
std::thread::Builder::new()
.name(format!("meet-track-{}", track.speaker.track_file()))
.spawn(move || run_track(track, writer, stop, paused, emit, silence_after))
.map_err(|e| e.to_string())?,
);
}
Ok(Self {
stop,
paused,
threads,
started: Instant::now(),
})
}
pub fn set_paused(&self, paused: bool) {
self.paused.store(paused, Ordering::Relaxed);
}
pub fn is_paused(&self) -> bool {
self.paused.load(Ordering::Relaxed)
}
pub fn stop(self) -> Vec<Written> {
self.stop.store(true, Ordering::Relaxed);
self.threads
.into_iter()
.filter_map(|t| t.join().ok().flatten())
.collect()
}
}
fn run_track(
mut track: Track,
mut writer: Option<WavWriter>,
stop: Arc<AtomicBool>,
paused: Arc<AtomicBool>,
emit: Emit,
silence_after: Duration,
) -> Option<Written> {
let speaker = track.speaker.clone();
let mut level_buf: Vec<i16> = Vec::new();
let mut last_level = Instant::now();
let mut last_flush = Instant::now();
let mut last_heard = Instant::now();
let mut warned = false;
while !stop.load(Ordering::Relaxed) {
let samples = match track.capture.read() {
Ok(s) => s,
Err(CaptureError::Ended) => break,
Err(e) => {
emit(Event::TrackFailed {
speaker: speaker.clone(),
error: e.to_string(),
});
break;
}
};
if paused.load(Ordering::Relaxed) {
last_heard = Instant::now();
continue;
}
if !samples.is_empty() {
if let Some(w) = writer.as_mut() {
if w.write(&samples).is_err() {
emit(Event::TrackFailed {
speaker: speaker.clone(),
error: "не удалось записать на диск".into(),
});
break;
}
if last_flush.elapsed() >= Duration::from_secs(5) {
let _ = w.flush();
last_flush = Instant::now();
}
}
level_buf.extend_from_slice(&samples);
}
if last_level.elapsed() >= Duration::from_millis(200) {
let dbfs = if level_buf.is_empty() {
-90.0
} else {
pcm::level_dbfs(&level_buf)
};
emit(Event::Level {
speaker: speaker.clone(),
dbfs,
});
if dbfs > SILENCE_DBFS {
last_heard = Instant::now();
if warned {
warned = false;
emit(Event::Heard {
speaker: speaker.clone(),
});
}
} else if !warned && last_heard.elapsed() >= silence_after {
warned = true;
emit(Event::Silent {
speaker: speaker.clone(),
seconds: last_heard.elapsed().as_secs(),
});
}
level_buf.clear();
last_level = Instant::now();
}
}
emit(Event::Level {
speaker: speaker.clone(),
dbfs: -90.0,
});
let mut writer = writer?;
let _ = writer.flush();
Some(Written {
speaker,
duration_ms: writer.duration_ms(),
})
}
#[cfg(test)]
mod tests {
use super::*;
use meet_audio::pcm::{decode_wav, ms_to_samples};
use meet_audio::source::Replay;
use std::sync::Mutex;
struct Broken;
impl Capture for Broken {
fn read(&mut self) -> Result<Vec<i16>, CaptureError> {
Err(CaptureError::Loopback("устройство пропало".into()))
}
}
#[test]
fn tracks_are_written_to_their_files() {
let dir = tempfile::tempdir().unwrap();
let events = Arc::new(Mutex::new(Vec::new()));
let sink = events.clone();
let me = dir.path().join("me.wav");
let session = Session::start(
vec![
Track {
speaker: Speaker::Me,
capture: Box::new(Replay::new(vec![500; ms_to_samples(2_000)], 40.0)),
path: Some(me.clone()),
},
Track {
speaker: Speaker::Others,
capture: Box::new(Broken),
path: Some(dir.path().join("others.wav")),
},
],
Arc::new(move |e| sink.lock().unwrap().push(e)),
)
.unwrap();
std::thread::sleep(Duration::from_millis(400));
let written = session.stop();
let me_written = written.iter().find(|w| w.speaker == Speaker::Me).unwrap();
assert_eq!(me_written.duration_ms, 2_000);
assert_eq!(
decode_wav(&std::fs::read(&me).unwrap()).unwrap().len(),
ms_to_samples(2_000)
);
let events = events.lock().unwrap();
assert!(events.iter().any(|e| matches!(
e,
Event::TrackFailed {
speaker: Speaker::Others,
..
}
)));
assert!(events.iter().any(|e| matches!(
e,
Event::Level {
speaker: Speaker::Me,
..
}
)));
}
#[test]
fn paused_audio_is_not_written() {
let dir = tempfile::tempdir().unwrap();
let session = Session::start(
vec![Track {
speaker: Speaker::Me,
capture: Box::new(Replay::new(vec![500; ms_to_samples(60_000)], 1.0)),
path: Some(dir.path().join("me.wav")),
}],
Arc::new(|_| {}),
)
.unwrap();
session.set_paused(true);
std::thread::sleep(Duration::from_millis(300));
assert_eq!(session.stop()[0].duration_ms, 0, "на паузе ничего не записано");
}
#[test]
fn long_silence_is_reported_and_cleared_when_sound_returns() {
let mut audio = vec![0i16; ms_to_samples(600)];
audio.extend(vec![3_000i16; ms_to_samples(600)]);
let events = Arc::new(Mutex::new(Vec::new()));
let sink = events.clone();
let session = Session::start_with(
vec![Track {
speaker: Speaker::Others,
capture: Box::new(Replay::new(audio, 1.0)),
path: None,
}],
Arc::new(move |e| sink.lock().unwrap().push(e)),
Duration::from_millis(300),
)
.unwrap();
std::thread::sleep(Duration::from_millis(1_400));
let written = session.stop();
assert!(written.is_empty(), "проверка звука ничего не пишет");
let events = events.lock().unwrap();
let silent = events.iter().position(|e| matches!(e, Event::Silent { .. }));
let heard = events.iter().position(|e| matches!(e, Event::Heard { .. }));
assert!(silent.is_some() && heard > silent, "{events:?}");
}
}