]> git.nihav.org Git - nihav-player.git/blob - videoplayer/src/audiodec.rs
a better video player
[nihav-player.git] / videoplayer / src / audiodec.rs
1 use std::time::Duration;
2 use std::thread::JoinHandle;
3 use std::sync::{Arc, Mutex};
4 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
5 use std::sync::mpsc::{Receiver, SyncSender, TrySendError};
6 use std::str::FromStr;
7
8 use sdl2::AudioSubsystem;
9 use sdl2::audio::{AudioDevice, AudioCallback, AudioSpecDesired};
10
11 use nihav_core::formats::*;
12 use nihav_core::codecs::*;
13 use nihav_core::soundcvt::*;
14
15 use super::{DecoderStuff, PktSendEvent};
16
17 static SKIP_ADECODING: AtomicBool = AtomicBool::new(false);
18 static AUDIO_VOLUME: AtomicUsize = AtomicUsize::new(100);
19 static AUDIO_END: AtomicBool = AtomicBool::new(false);
20 static CUR_QUEUE_FILL: AtomicUsize = AtomicUsize::new(0);
21 static SAMPLE_RATE: AtomicUsize = AtomicUsize::new(0);
22 static CHANNELS: AtomicUsize = AtomicUsize::new(0);
23
24 static CURRENT_TIME: AtomicUsize = AtomicUsize::new(0);
25 static CURRENT_TIME_SET: AtomicBool = AtomicBool::new(false);
26
27 const QUEUE_INITIAL_SIZE: usize = 16384;
28 const QUEUE_REFILL_LIMIT: usize = 262144;
29
30 struct AudioQueue {
31 queue: Vec<i16>,
32 start: usize,
33 end: usize,
34 srate: usize,
35 chans: usize,
36 spos: usize,
37 }
38
39 impl AudioQueue {
40 fn new(srate: usize, chans: usize) -> Self {
41 SAMPLE_RATE.store(srate, Ordering::Relaxed);
42 CHANNELS.store(chans, Ordering::Relaxed);
43 CUR_QUEUE_FILL.store(0, Ordering::Relaxed);
44 Self {
45 queue: Vec::with_capacity(QUEUE_INITIAL_SIZE),
46 start: 0,
47 end: 0,
48 srate, chans,
49 spos: 0,
50 }
51 }
52 fn home(&mut self) {
53 if self.start == 0 { return; }
54 let fill = self.fill();
55 if fill > 0 {
56 if fill < self.start {
57 let (dst, src) = self.queue.split_at_mut(self.start);
58 dst[..fill].copy_from_slice(&src[..fill]);
59 } else {
60 for i in 0..fill {
61 self.queue[i] = self.queue[self.start + i];
62 }
63 }
64 }
65 self.start = 0;
66 self.end = fill;
67 }
68 fn set_time(&mut self) {
69 let fill = self.fill();
70 let samp_time = self.spos.saturating_sub(fill / self.chans);
71 CURRENT_TIME.store(samp_time * 1000 / self.srate, Ordering::Relaxed);
72 CURRENT_TIME_SET.store(true, Ordering::Relaxed);
73 CUR_QUEUE_FILL.store(self.fill(), Ordering::Relaxed);
74 }
75 fn add(&mut self, src: &[i16], samplepos: usize) {
76 if self.end + src.len() > self.queue.len() {
77 self.home();
78 }
79 if self.end + src.len() > self.queue.len() {
80 self.queue.resize(self.end + src.len(), 0);
81 }
82 self.queue[self.end..][..src.len()].copy_from_slice(&src);
83 self.end += src.len();
84 self.spos = samplepos;
85 self.set_time();
86 }
87 fn add_bytes(&mut self, src: &[u8], samplepos: usize) {
88 let srclen = src.len() / 2;
89 if self.end + srclen > self.queue.len() {
90 self.home();
91 }
92 if self.end + srclen > self.queue.len() {
93 self.queue.resize(self.end + srclen, 0);
94 }
95 for (dst, src) in self.queue[self.end..][..srclen].iter_mut().zip(src.chunks_exact(2)) {
96 *dst = (u16::from(src[0]) + u16::from(src[1]) * 256) as i16;
97 }
98 self.end += srclen;
99 self.spos = samplepos;
100 self.set_time();
101 }
102 fn drain(&mut self, size: usize) {
103 let fill = self.fill();
104 if size >= fill {
105 self.flush();
106 } else {
107 self.start += size;
108 }
109 self.set_time();
110 }
111 fn fill(&self) -> usize { self.end - self.start }
112 fn flush(&mut self) {
113 self.start = 0;
114 self.end = 0;
115 }
116 }
117
118 pub struct AudioOutput {
119 queue: Arc<Mutex<AudioQueue>>,
120 }
121
122 impl AudioCallback for AudioOutput {
123 type Channel = i16;
124
125 fn callback(&mut self, out: &mut [Self::Channel]) {
126 let mut queue = self.queue.lock().unwrap();
127 let dstlen = out.len();
128 let copylen = queue.fill().min(dstlen);
129 let volume = AUDIO_VOLUME.load(Ordering::Relaxed) as i32;
130 if volume == 100 {
131 out[..copylen].copy_from_slice(&queue.queue[queue.start..][..copylen]);
132 } else {
133 for (dst, &src) in out[..copylen].iter_mut().zip(queue.queue[queue.start..].iter()) {
134 *dst = (i32::from(src) * volume / 100).max(-32768).min(32767) as i16;
135 }
136 }
137 queue.drain(copylen);
138 for el in out[copylen..].iter_mut() { *el = 0; }
139 }
140 }
141
142 fn dummy_audio_thread(aprecv: Receiver<PktSendEvent>) -> JoinHandle<()> {
143 std::thread::Builder::new().name("acontrol-dummy".to_string()).spawn(move ||{
144 loop {
145 match aprecv.recv() {
146 Ok(PktSendEvent::End) => break,
147 Ok(PktSendEvent::ImmediateEnd) => break,
148 Err(_) => {
149 break;
150 },
151 _ => {},
152 };
153 }
154 }).unwrap()
155 }
156
157 type AudioPlaybackType = Option<AudioDevice<AudioOutput>>;
158
159 fn start_audio_decoding(asystem: &AudioSubsystem, ainfo: NAAudioInfo, mut audio_dec: DecoderStuff, aprecv: Receiver<PktSendEvent>) -> (AudioPlaybackType, JoinHandle<()>) {
160 let ch = ainfo.channels.max(2);
161 let desired_spec = AudioSpecDesired {
162 freq: Some(ainfo.sample_rate as i32),
163 channels: Some(ch),
164 samples: None
165 };
166 let dst_info = NAAudioInfo {
167 sample_rate: ainfo.sample_rate,
168 channels: ch,
169 format: SND_S16_FORMAT,
170 block_len: 0,
171 };
172 let queue = Arc::new(Mutex::new(AudioQueue::new(ainfo.sample_rate as usize, ch as usize)));
173 let qclone = queue.clone();
174 let ret = asystem.open_playback(None, &desired_spec, |_spec| {
175 AudioOutput {
176 queue: qclone,
177 }
178 });
179 if ret.is_err() {
180 return (None, dummy_audio_thread(aprecv))
181 }
182 let adevice = ret.unwrap();
183 (Some(adevice), std::thread::Builder::new().name("acontrol".to_string()).spawn(move ||{
184 let mut samplepos = 0usize;
185 let dst_chmap = if dst_info.channels == 2 {
186 NAChannelMap::from_str("L,R").unwrap()
187 } else {
188 NAChannelMap::from_str("C").unwrap()
189 };
190 SKIP_ADECODING.store(false, Ordering::Relaxed);
191 loop {
192 match aprecv.recv() {
193 Ok(PktSendEvent::Packet(pkt)) => {
194 loop {
195 if CUR_QUEUE_FILL.load(Ordering::Relaxed)
196 < QUEUE_REFILL_LIMIT || SKIP_ADECODING.load(Ordering::Relaxed) {
197 break;
198 }
199 std::thread::sleep(Duration::from_millis(100));
200 }
201 if !SKIP_ADECODING.load(Ordering::Relaxed) {
202 if let Ok(frm) = audio_dec.dec.decode(&mut audio_dec.dsupp, &pkt) {
203 let buf = frm.get_buffer();
204 if let Some(pts) = frm.get_pts() {
205 samplepos = NATimeInfo::ts_to_time(pts, u64::from(dst_info.sample_rate), frm.ts.tb_num, frm.ts.tb_den) as usize;
206 }
207 samplepos += buf.get_audio_length();
208 if let Ok(out_buf) = convert_audio_frame(&buf, &dst_info, &dst_chmap) {
209 match out_buf {
210 NABufferType::AudioI16(abuf) => {
211 let mut qdata = queue.lock().unwrap();
212 qdata.add(abuf.get_data(), samplepos);
213 drop(qdata);
214 },
215 NABufferType::AudioPacked(abuf) => {
216 let mut qdata = queue.lock().unwrap();
217 qdata.add_bytes(abuf.get_data(), samplepos);
218 drop(qdata);
219 },
220 _ => {},
221 };
222 }
223 }
224 }
225 },
226 Ok(PktSendEvent::Flush) => {
227 audio_dec.dec.flush();
228 let mut qdata = queue.lock().unwrap();
229 qdata.flush();
230 SKIP_ADECODING.store(false, Ordering::Relaxed);
231 },
232 Ok(PktSendEvent::End) => break,
233 Ok(PktSendEvent::ImmediateEnd) => {
234 let mut qdata = queue.lock().unwrap();
235 qdata.flush();
236 break;
237 },
238 Ok(PktSendEvent::HurryUp) => {},
239 Err(_) => {
240 break;
241 },
242 };
243 }
244 loop {
245 let qdata = queue.lock().unwrap();
246 if qdata.fill() == 0 || SKIP_ADECODING.load(Ordering::Relaxed) {
247 break;
248 }
249 }
250 AUDIO_END.store(true, Ordering::Relaxed);
251 }).unwrap())
252 }
253
254 pub struct AudioControl {
255 aqueue: Vec<PktSendEvent>,
256 apsend: SyncSender<PktSendEvent>,
257 adevice: AudioPlaybackType,
258 athread: JoinHandle<()>,
259 }
260
261 impl AudioControl {
262 pub fn new(audio_dec: Option<DecoderStuff>, ainfo: Option<NAAudioInfo>, asystem: &AudioSubsystem) -> Self {
263 let (apsend, aprecv) = std::sync::mpsc::sync_channel::<PktSendEvent>(20);
264 let (adevice, athread) = if let Some(audio_dec) = audio_dec {
265 start_audio_decoding(asystem, ainfo.unwrap(), audio_dec, aprecv)
266 } else {
267 (None, dummy_audio_thread(aprecv))
268 };
269 AUDIO_END.store(false, Ordering::Relaxed);
270
271 Self {
272 aqueue: Vec::new(),
273 apsend,
274 adevice,
275 athread,
276 }
277 }
278 pub fn has_audio(&self) -> bool { self.adevice.is_some() }
279 pub fn pause(&mut self) {
280 if let Some(ref device) = self.adevice {
281 device.pause();
282 }
283 }
284 pub fn resume(&mut self) {
285 if let Some(ref device) = self.adevice {
286 device.resume();
287 }
288 }
289 pub fn set_volume(&mut self, volume: usize) {
290 AUDIO_VOLUME.store(volume, Ordering::Relaxed);
291 }
292 pub fn get_volume(&self) -> usize {
293 AUDIO_VOLUME.load(Ordering::Relaxed)
294 }
295 pub fn is_audio_end(&self) -> bool {
296 AUDIO_END.load(Ordering::Relaxed)
297 }
298 pub fn get_fill(&self) -> usize { CUR_QUEUE_FILL.load(Ordering::Relaxed) }
299 pub fn get_time(&self) -> Option<u64> {
300 if CURRENT_TIME_SET.load(Ordering::Relaxed) {
301 Some(CURRENT_TIME.load(Ordering::Relaxed) as u64)
302 } else {
303 None
304 }
305 }
306 pub fn get_time_left(&self) -> u64 {
307 let srate = SAMPLE_RATE.load(Ordering::Relaxed);
308 let chans = CHANNELS.load(Ordering::Relaxed);
309 if srate != 0 && chans != 0{
310 let fill = self.get_fill();
311 (fill * 1000 / srate / chans) as u64
312 } else {
313 0
314 }
315 }
316
317 pub fn get_queue_size(&self) -> usize { self.aqueue.len() }
318 pub fn try_send_audio(&mut self, evt: PktSendEvent) -> bool {
319 if self.aqueue.len() > 0 {
320 self.aqueue.push(evt);
321 false
322 } else {
323 self.try_send_event(evt)
324 }
325 }
326 fn try_send_event(&mut self, evt: PktSendEvent) -> bool {
327 if let Err(TrySendError::Full(evt)) = self.apsend.try_send(evt) {
328 self.aqueue.insert(0, evt);
329 false
330 } else {
331 true
332 }
333 }
334 pub fn try_send_queued(&mut self) -> bool {
335 while !self.aqueue.is_empty() {
336 let pkt = self.aqueue.remove(0);
337 if !self.try_send_event(pkt) {
338 return false;
339 }
340 }
341 true
342 }
343
344 pub fn flush(&mut self) {
345 self.pause();
346 self.aqueue.truncate(0);
347 SKIP_ADECODING.store(true, Ordering::Release);
348 CURRENT_TIME_SET.store(false, Ordering::Release);
349 let _ = self.apsend.send(PktSendEvent::Flush);
350 }
351 pub fn finish(self) {
352 SKIP_ADECODING.store(true, Ordering::Release);
353 let _ = self.apsend.send(PktSendEvent::ImmediateEnd);
354 self.athread.join().unwrap();
355 }
356 }