1use crate::{
2 audio::{
3 AudioFrame,
4 constants::{MIXER_CHANNELS, TARGET_SAMPLE_RATE},
5 demux::{DemuxResult, open_format},
6 engine::{BoxedEngine, StandardEngine},
7 resample::Resampler,
8 },
9 common::types::AudioFormat,
10 config::player::{PlayerConfig, ResamplingQuality},
11};
12use flume::Receiver;
13use std::io::ErrorKind;
14use symphonia::core::{
15 audio::SampleBuffer,
16 codecs::Decoder,
17 errors::Error,
18 formats::{FormatReader, SeekMode, SeekTo},
19 io::MediaSource,
20 units::Time,
21};
22use tracing::{Level, debug, span, warn};
23#[derive(Debug, Clone, PartialEq)]
24pub enum DecoderCommand {
25 Seek(u64),
26 Stop,
27}
28#[derive(Debug, PartialEq)]
29pub enum CommandOutcome {
30 Stop,
31 Seeked,
32 SeekFailed,
33 None,
34}
35pub struct AudioProcessor {
36 format: Box<dyn FormatReader>,
37 decoder: Box<dyn Decoder>,
38 resampler: Resampler,
39 track_id: u32,
40 engine: BoxedEngine,
41 cmd_rx: Receiver<DecoderCommand>,
42 error_tx: Option<flume::Sender<String>>,
43 sample_buf: Option<SampleBuffer<i16>>,
44 source_rate: u32,
45 channels: usize,
46 config: PlayerConfig,
47 recoverable_errors: u32,
48 downmix_buf: Vec<i16>,
49}
50impl AudioProcessor {
51 pub fn new(
52 source: Box<dyn MediaSource>,
53 kind: Option<AudioFormat>,
54 frame_tx: flume::Sender<AudioFrame>,
55 cmd_rx: Receiver<DecoderCommand>,
56 error_tx: Option<flume::Sender<String>>,
57 config: PlayerConfig,
58 ) -> Result<Self, Error> {
59 Self::with_engine(
60 source,
61 kind,
62 Box::new(StandardEngine::new(frame_tx)),
63 cmd_rx,
64 error_tx,
65 config,
66 )
67 }
68 pub fn with_engine(
69 source: Box<dyn MediaSource>,
70 kind: Option<AudioFormat>,
71 engine: BoxedEngine,
72 cmd_rx: Receiver<DecoderCommand>,
73 error_tx: Option<flume::Sender<String>>,
74 config: PlayerConfig,
75 ) -> Result<Self, Error> {
76 let DemuxResult::Transcode {
77 format,
78 track_id,
79 decoder,
80 sample_rate,
81 channels,
82 } = open_format(source, kind)?;
83 debug!(
84 "AudioProcessor: opened format — {}Hz {}ch",
85 sample_rate, channels
86 );
87 let resampler = Self::make_resampler(sample_rate, &config);
88 Ok(Self {
89 format,
90 decoder,
91 resampler,
92 track_id,
93 engine,
94 cmd_rx,
95 error_tx,
96 sample_buf: None,
97 source_rate: sample_rate,
98 channels,
99 config,
100 recoverable_errors: 0,
101 downmix_buf: Vec::with_capacity(1920),
102 })
103 }
104}
105impl AudioProcessor {
106 pub fn run(&mut self) -> Result<(), Error> {
107 self.run_inner(false)
108 }
109 pub fn run_with_seek(&mut self) -> Result<(), Error> {
110 self.run_inner(true)
111 }
112}
113impl AudioProcessor {
114 fn run_inner(&mut self, seek_enabled: bool) -> Result<(), Error> {
115 let _span = span!(Level::DEBUG, "audio_processor").entered();
116 debug!(
117 "Starting transcode loop (seek={}): {}Hz {}ch -> {}Hz",
118 seek_enabled, self.source_rate, self.channels, TARGET_SAMPLE_RATE
119 );
120 let mut packet_count = 0u64;
121 loop {
122 packet_count += 1;
123 match self.check_commands() {
124 CommandOutcome::Stop => break,
125 CommandOutcome::Seeked | CommandOutcome::SeekFailed if seek_enabled => continue,
126 _ => {}
127 }
128 let packet = match self.format.next_packet() {
129 Ok(p) => p,
130 Err(Error::IoError(e)) if e.kind() == ErrorKind::UnexpectedEof => break,
131 Err(e) => {
132 self.send_error(format!("Packet read error: {e}"));
133 return Err(e);
134 }
135 };
136 if packet.track_id() != self.track_id {
137 continue;
138 }
139 match self.decoder.decode(&packet) {
140 Ok(decoded) => {
141 self.recoverable_errors = 0;
142 let spec = *decoded.spec();
143 let mut buf = self.sample_buf.take().unwrap_or_else(|| {
144 SampleBuffer::<i16>::new(decoded.capacity() as u64, spec)
145 });
146 buf.copy_interleaved_ref(decoded);
147 let samples = buf.samples();
148 if !samples.is_empty() {
149 let frame_channels = spec.channels.count();
150 let frame_rate = spec.rate;
151 if frame_rate != self.source_rate {
152 debug!(
153 "AudioProcessor: frame rate mismatch ({}Hz vs {}Hz) — re-initializing resampler",
154 frame_rate, self.source_rate
155 );
156 self.source_rate = frame_rate;
157 self.resampler = Self::make_resampler(self.source_rate, &self.config);
158 }
159 let source_rate = self.source_rate;
160 let pcm_data = if frame_channels == MIXER_CHANNELS {
161 samples
162 } else {
163 Self::downmix_internal(
164 &mut self.downmix_buf,
165 samples,
166 frame_channels,
167 packet_count,
168 )
169 };
170 let capacity = (pcm_data.len() as f64 * TARGET_SAMPLE_RATE as f64
171 / source_rate as f64)
172 .ceil() as usize
173 + 32;
174 let mut resampled = crate::audio::buffer::acquire_buffer(capacity);
175 if self.resampler.is_passthrough() {
176 resampled.extend_from_slice(pcm_data);
177 } else {
178 self.resampler.process(pcm_data, &mut resampled);
179 }
180 if !resampled.is_empty() {
181 if packet_count == 1 {
182 debug!(
183 "AudioProcessor: Sending first frame to engine (capacity={})",
184 resampled.capacity()
185 );
186 }
187 if !self.engine.push(AudioFrame::Pcm(resampled)) {
188 return Ok(());
189 }
190 }
191 }
192 self.sample_buf = Some(buf);
193 }
194 Err(Error::IoError(e)) if e.kind() == ErrorKind::UnexpectedEof => break,
195 Err(Error::DecodeError(e)) | Err(Error::Unsupported(e)) => {
196 self.recoverable_errors += 1;
197 if e.contains("main_data_begin") {
198 continue;
199 }
200 if self.recoverable_errors == 1 {
201 warn!("Decode error (recoverable): {e}");
202 } else if self.recoverable_errors.is_multiple_of(100) {
203 warn!(
204 "Decode error (recoverable, x{}): {e}",
205 self.recoverable_errors
206 );
207 }
208 }
209 Err(Error::ResetRequired) => {
210 self.decoder.reset();
211 self.resampler.reset();
212 self.sample_buf = None;
213 warn!("Decoder reset required — resetting state and continuing");
214 }
215 Err(e) => {
216 self.send_error(format!("Decode error: {e}"));
217 return Err(e);
218 }
219 }
220 }
221 debug!("Transcode loop finished");
222 Ok(())
223 }
224}
225impl AudioProcessor {
226 fn check_commands(&mut self) -> CommandOutcome {
227 match self.cmd_rx.try_recv() {
228 Ok(DecoderCommand::Seek(ms)) => {
229 let time = Time::from(ms as f64 / 1000.0);
230 if self
231 .format
232 .seek(
233 SeekMode::Accurate,
234 SeekTo::Time {
235 time,
236 track_id: Some(self.track_id),
237 },
238 )
239 .is_ok()
240 {
241 self.resampler.reset();
242 self.decoder.reset();
243 self.sample_buf = None;
244 let _ = self.engine.push(AudioFrame::Pcm(Vec::new()));
245 CommandOutcome::Seeked
246 } else {
247 warn!("AudioProcessor: seek to {}ms failed", ms);
248 CommandOutcome::SeekFailed
249 }
250 }
251 Ok(DecoderCommand::Stop) | Err(flume::TryRecvError::Disconnected) => {
252 CommandOutcome::Stop
253 }
254 _ => CommandOutcome::None,
255 }
256 }
257 fn send_error(&self, msg: String) {
258 if let Some(tx) = &self.error_tx {
259 let _ = tx.send(msg);
260 }
261 }
262 fn make_resampler(sample_rate: u32, config: &PlayerConfig) -> Resampler {
263 if sample_rate == TARGET_SAMPLE_RATE {
264 return Resampler::linear(sample_rate, TARGET_SAMPLE_RATE, MIXER_CHANNELS);
265 }
266 match config.resampling_quality {
267 ResamplingQuality::Low => {
268 Resampler::linear(sample_rate, TARGET_SAMPLE_RATE, MIXER_CHANNELS)
269 }
270 ResamplingQuality::Medium => {
271 Resampler::hermite(sample_rate, TARGET_SAMPLE_RATE, MIXER_CHANNELS)
272 }
273 ResamplingQuality::High => {
274 Resampler::sinc(sample_rate, TARGET_SAMPLE_RATE, MIXER_CHANNELS)
275 }
276 }
277 }
278 fn downmix_internal<'a>(
279 downmix_buf: &'a mut Vec<i16>,
280 samples: &[i16],
281 frame_channels: usize,
282 packet_count: u64,
283 ) -> &'a [i16] {
284 if packet_count.is_multiple_of(100) {
285 debug!(
286 "AudioProcessor: Downmixing {}ch -> {}ch (samples: {})",
287 frame_channels,
288 MIXER_CHANNELS,
289 samples.len()
290 );
291 }
292 let num_frames = samples.len() / frame_channels;
293 downmix_buf.clear();
294 downmix_buf.reserve(num_frames * MIXER_CHANNELS);
295 for i in 0..num_frames {
296 let frame = &samples[i * frame_channels..(i + 1) * frame_channels];
297 let mut l = 0i32;
298 let mut r = 0i32;
299 for (ch, &sample) in frame.iter().enumerate() {
300 if ch % 2 == 0 {
301 l += sample as i32;
302 } else {
303 r += sample as i32;
304 }
305 }
306 let left_count = frame_channels.div_ceil(2);
307 let right_count = frame_channels / 2;
308 downmix_buf.push((l / left_count as i32) as i16);
309 if right_count > 0 {
310 downmix_buf.push((r / right_count as i32) as i16);
311 } else {
312 downmix_buf.push((l / left_count as i32) as i16);
313 }
314 }
315 &downmix_buf[..]
316 }
317}