1use super::backend::{PlayerBackend, PlayerError};
2use super::events::{PlayerEventEmitter, PlayerStatusEvent};
3use super::media::{MediaItem, MediaSource};
4use super::state::PlayerState;
5use super::stream_end::ObservedTime;
6use crate::playback_reporting::{EventThrottler, PlaybackOperation, PlaybackReporter};
7use crate::settings::{AudioSettings, VolumeLevel, EQ_BANDS};
8use crate::utils::conversions::{seconds_to_ticks, volume_to_percent};
9use crate::utils::lock::MutexSafe;
10use libmpv::Mpv;
11use log::{debug, error, info, warn};
12use std::process::Command;
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::sync::{Arc, Mutex};
15use std::time::{Duration, SystemTime, UNIX_EPOCH};
16use tokio::sync::Mutex as TokioMutex;
17
18pub struct MpvBackend {
23 mpv: Arc<Mpv>,
24 state: Arc<Mutex<InternalState>>,
25 event_emitter: Option<Arc<dyn PlayerEventEmitter>>,
26 audio_settings: AudioSettings,
27 playback_reporter: Arc<TokioMutex<Option<PlaybackReporter>>>,
28 position_throttler: Arc<EventThrottler>,
29 last_seek_time: Arc<AtomicU64>,
30 observed: Arc<Mutex<ObservedTime>>,
37}
38
39struct InternalState {
40 current_media: Option<MediaItem>,
41 volume: f32,
42}
43
44fn detect_audio_system() -> String {
46 info!("[MpvBackend] Detecting audio system...");
47
48 if let Ok(output) = Command::new("pactl").arg("info").output() {
50 if output.status.success() {
51 let stdout = String::from_utf8_lossy(&output.stdout);
52 if stdout.contains("PipeWire") {
53 info!("[MpvBackend] Detected PipeWire (with PulseAudio compatibility)");
54 return "pulse".to_string();
55 } else if stdout.contains("PulseAudio") {
56 info!("[MpvBackend] Detected PulseAudio");
57 return "pulse".to_string();
58 }
59 }
60 }
61
62 if let Ok(output) = Command::new("pw-cli").arg("info").arg("0").output() {
64 if output.status.success() {
65 info!("[MpvBackend] Detected PipeWire");
66 return "pulse".to_string(); }
68 }
69
70 if std::path::Path::new("/proc/asound/cards").exists() {
72 info!("[MpvBackend] Falling back to ALSA");
73 return "alsa".to_string();
74 }
75
76 warn!("[MpvBackend] Could not detect audio system, using 'auto'");
78 "auto".to_string()
79}
80
81fn get_stream_url(media: &MediaItem) -> String {
83 match &media.source {
84 MediaSource::Remote { stream_url, .. } => stream_url.clone(),
85 MediaSource::Local { file_path, .. } => {
86 format!("file://{}", file_path.to_string_lossy())
87 }
88 MediaSource::DirectUrl { url } => url.clone(),
89 }
90}
91
92impl MpvBackend {
93 pub fn new(
95 event_emitter: Option<Arc<dyn PlayerEventEmitter>>,
96 playback_reporter: Arc<TokioMutex<Option<PlaybackReporter>>>,
97 position_throttler: Arc<EventThrottler>,
98 ) -> Result<Self, PlayerError> {
99 info!("[MpvBackend] Initializing MPV backend...");
100
101 use std::ffi::CString;
104 unsafe {
105 let c_locale = CString::new("C").unwrap();
106 libc::setlocale(libc::LC_NUMERIC, c_locale.as_ptr());
107 }
108
109 let mpv = Mpv::new().map_err(|e| PlayerError {
110 message: format!("Failed to initialize MPV: {:?}", e),
111 })?;
112
113 let audio_driver = detect_audio_system();
115 info!(
116 "[MpvBackend] Configuring audio output driver: {}",
117 audio_driver
118 );
119
120 mpv.set_property("ao", audio_driver.as_str())
121 .map_err(|e| PlayerError {
122 message: format!(
123 "Failed to set audio output to '{}': {:?}. Make sure audio system is working.",
124 audio_driver, e
125 ),
126 })?;
127
128 mpv.set_property("msg-level", "all=warn,ao=debug")
130 .unwrap_or_else(|e| {
131 warn!("[MpvBackend] Warning: Could not set MPV log level: {:?}", e);
132 });
133
134 mpv.set_property("audio-display", "no")
136 .map_err(|e| PlayerError {
137 message: format!("Failed to configure MPV audio-display: {:?}", e),
138 })?;
139
140 mpv.set_property("video", "no").map_err(|e| PlayerError {
141 message: format!("Failed to configure MPV video: {:?}", e),
142 })?;
143
144 mpv.set_property("volume", 100i64)
146 .map_err(|e| PlayerError {
147 message: format!("Failed to set initial volume: {:?}", e),
148 })?;
149
150 mpv.set_property(
160 "stream-lavf-o",
161 "reconnect=1,reconnect_streamed=1,reconnect_on_network_error=1,reconnect_delay_max=5",
162 )
163 .unwrap_or_else(|e| {
164 warn!(
165 "[MpvBackend] Could not enable stream reconnection: {:?} — \
166 playback will not survive network interruptions",
167 e
168 );
169 });
170 mpv.set_property("network-timeout", 15i64)
171 .unwrap_or_else(|e| {
172 warn!("[MpvBackend] Could not set network timeout: {:?}", e);
173 });
174
175 let state = Arc::new(Mutex::new(InternalState {
176 current_media: None,
177 volume: 1.0,
178 }));
179
180 let backend = MpvBackend {
181 mpv: Arc::new(mpv),
182 state,
183 event_emitter,
184 audio_settings: AudioSettings::default(),
185 playback_reporter,
186 position_throttler,
187 last_seek_time: Arc::new(AtomicU64::new(0)),
188 observed: Arc::new(Mutex::new(ObservedTime::default())),
189 };
190
191 backend.start_event_loop();
193
194 info!("[MpvBackend] Initialized successfully");
195 Ok(backend)
196 }
197
198 fn start_event_loop(&self) {
200 let mpv = self.mpv.clone();
201 let event_emitter = self.event_emitter.clone();
202 let state = self.state.clone();
203 let reporter = self.playback_reporter.clone();
204 let throttler = self.position_throttler.clone();
205
206 std::thread::spawn(move || {
207 info!("[MpvBackend] Event loop started");
208
209 let mut ev_ctx = mpv.create_event_context();
210 ev_ctx.disable_deprecated_events().unwrap_or_else(|e| {
211 error!("[MpvBackend] Failed to disable deprecated events: {:?}", e);
212 });
213
214 loop {
215 match ev_ctx.wait_event(1.0) {
216 Some(Ok(event)) => match event {
217 libmpv::events::Event::StartFile => {
218 debug!("[MpvBackend] Starting file");
219 }
220 libmpv::events::Event::FileLoaded => {
221 info!("[MpvBackend] File loaded");
222
223 if let Ok(duration) = mpv.get_property::<f64>("duration") {
225 if let Some(emitter) = &event_emitter {
226 emitter.emit(PlayerStatusEvent::MediaLoaded { duration });
227 }
228 }
229 }
230 libmpv::events::Event::PlaybackRestart => {
231 debug!("[MpvBackend] Playback started/resumed");
232
233 let media_id = state
234 .lock_safe()
235 .current_media
236 .as_ref()
237 .map(|m| m.id.clone());
238
239 if let Some(emitter) = &event_emitter {
240 emitter.emit(PlayerStatusEvent::StateChanged {
241 state: "playing".to_string(),
242 media_id,
243 });
244 }
245 }
246 libmpv::events::Event::PropertyChange { name: "pause", .. } => {
247 if let Ok(is_paused) = mpv.get_property::<bool>("pause") {
249 let media_id = state
250 .lock_safe()
251 .current_media
252 .as_ref()
253 .map(|m| m.id.clone());
254
255 if let Some(emitter) = &event_emitter {
256 emitter.emit(PlayerStatusEvent::StateChanged {
257 state: if is_paused { "paused" } else { "playing" }
258 .to_string(),
259 media_id,
260 });
261 }
262 }
263 }
264 libmpv::events::Event::EndFile(reason) => {
265 debug!("[MpvBackend] End file with reason: {}", reason);
266
267 const MPV_END_FILE_REASON_EOF: u32 = 0;
271 const MPV_END_FILE_REASON_STOP: u32 = 2;
272 const MPV_END_FILE_REASON_QUIT: u32 = 3;
273 const MPV_END_FILE_REASON_ERROR: u32 = 4;
274
275 if reason == MPV_END_FILE_REASON_EOF {
276 debug!("[MpvBackend] Track finished naturally (EOF), emitting PlaybackEnded");
277 if let Some(emitter) = &event_emitter {
278 emitter.emit(PlayerStatusEvent::PlaybackEnded);
279 }
280 } else if reason == MPV_END_FILE_REASON_STOP {
281 debug!("[MpvBackend] Track stopped (loading new track), NOT emitting PlaybackEnded");
282 } else if reason == MPV_END_FILE_REASON_QUIT {
284 debug!("[MpvBackend] Player quitting, NOT emitting PlaybackEnded");
285 } else if reason == MPV_END_FILE_REASON_ERROR {
287 warn!("[MpvBackend] Track ended with an error — reporting as recoverable");
297 if let Some(emitter) = &event_emitter {
298 emitter.emit(PlayerStatusEvent::Error {
299 message: "Playback stream failed".to_string(),
300 recoverable: true,
301 });
302 }
303 } else {
304 debug!("[MpvBackend] Unknown end file reason {}, NOT emitting PlaybackEnded", reason);
305 }
306 }
307 libmpv::events::Event::Shutdown => {
308 info!("[MpvBackend] Shutdown event received");
309 break;
310 }
311 _ => {}
312 },
313 Some(Err(e)) => {
314 error!("[MpvBackend] Event error: {:?}", e);
315 }
316 None => {
317 }
319 }
320
321 std::thread::sleep(Duration::from_millis(10));
322 }
323
324 info!("[MpvBackend] Event loop ended");
325 });
326
327 let mpv_for_position = self.mpv.clone();
329 let emitter_for_position = self.event_emitter.clone();
330 let state_for_position = self.state.clone();
331 let reporter_for_position = reporter.clone();
332 let throttler_for_position = throttler.clone();
333 let last_seek_time_for_position = self.last_seek_time.clone();
334 let observed_for_position = self.observed.clone();
335
336 std::thread::spawn(move || {
337 loop {
338 std::thread::sleep(Duration::from_millis(250));
339
340 if let (Ok(pos), Ok(dur)) = (
343 mpv_for_position.get_property::<f64>("time-pos"),
344 mpv_for_position.get_property::<f64>("duration"),
345 ) {
346 observed_for_position.lock_safe().record(pos, dur);
352
353 let now = SystemTime::now()
356 .duration_since(UNIX_EPOCH)
357 .unwrap()
358 .as_millis() as u64;
359 let last_seek = last_seek_time_for_position.load(Ordering::Relaxed);
360 let time_since_seek = now.saturating_sub(last_seek);
361
362 if time_since_seek < 150 {
364 continue;
365 }
366
367 if let Some(emitter) = &emitter_for_position {
369 emitter.emit(PlayerStatusEvent::PositionUpdate {
370 position: pos,
371 duration: dur,
372 });
373 }
374
375 let is_paused = mpv_for_position
377 .get_property::<bool>("pause")
378 .unwrap_or(true);
379
380 if !is_paused {
382 let jellyfin_id = {
384 let state = state_for_position.lock_safe();
385 state
386 .current_media
387 .as_ref()
388 .and_then(|m| m.jellyfin_id().map(|s| s.to_string()))
389 };
390
391 if let Some(item_id) = jellyfin_id {
392 if throttler_for_position.should_report(&item_id) {
393 let position_ticks = seconds_to_ticks(pos);
394 let reporter_clone = reporter_for_position.clone();
395 let item_id_clone = item_id.clone();
396
397 if let Ok(handle) = tokio::runtime::Handle::try_current() {
400 handle.spawn(async move {
401 let reporter_guard = reporter_clone.lock().await;
402 if let Some(reporter_instance) = reporter_guard.as_ref() {
403 let operation = PlaybackOperation::Progress {
404 item_id: item_id_clone.clone(),
405 position_ticks,
406 is_paused: false,
407 };
408
409 match reporter_instance.report(operation, true).await {
410 Ok(_) => debug!(
411 "[MpvBackend] Reported progress for {}",
412 item_id_clone
413 ),
414 Err(e) => warn!(
415 "[MpvBackend] Failed to report progress: {}",
416 e
417 ),
418 }
419 }
420 });
421 } else {
422 std::thread::spawn(move || {
424 let rt = tokio::runtime::Runtime::new().unwrap();
425 rt.block_on(async move {
426 let reporter_guard = reporter_clone.lock().await;
427 if let Some(reporter_instance) = reporter_guard.as_ref() {
428 let operation = PlaybackOperation::Progress {
429 item_id: item_id_clone.clone(),
430 position_ticks,
431 is_paused: false,
432 };
433
434 match reporter_instance.report(operation, true).await {
435 Ok(_) => debug!("[MpvBackend] Reported progress for {}", item_id_clone),
436 Err(e) => warn!("[MpvBackend] Failed to report progress: {}", e),
437 }
438 }
439 });
440 });
441 }
442
443 throttler_for_position.mark_reported(&item_id);
444 }
445 }
446 }
447 }
448 }
449 });
450 }
451}
452
453impl PlayerBackend for MpvBackend {
454 fn load(&mut self, media: &MediaItem) -> Result<(), PlayerError> {
455 let stream_url = get_stream_url(media);
456 info!("[MpvBackend] Loading: {} - {}", media.title, stream_url);
457
458 {
460 let mut state = self.state.lock_safe();
461 state.current_media = Some(media.clone());
462 }
463 self.observed.lock_safe().reset();
466
467 self.mpv
469 .command("loadfile", &[&stream_url])
470 .map_err(|e| PlayerError {
471 message: format!("Failed to load file: {:?}", e),
472 })?;
473
474 debug!("[MpvBackend] Load command sent successfully");
475 Ok(())
476 }
477
478 fn play(&mut self) -> Result<(), PlayerError> {
479 debug!("[MpvBackend] Play command");
480
481 self.mpv
482 .set_property("pause", false)
483 .map_err(|e| PlayerError {
484 message: format!("Failed to play: {:?}", e),
485 })?;
486
487 Ok(())
488 }
489
490 fn pause(&mut self) -> Result<(), PlayerError> {
491 debug!("[MpvBackend] Pause command");
492
493 self.mpv
494 .set_property("pause", true)
495 .map_err(|e| PlayerError {
496 message: format!("Failed to pause: {:?}", e),
497 })?;
498
499 Ok(())
500 }
501
502 fn stop(&mut self) -> Result<(), PlayerError> {
503 debug!("[MpvBackend] Stop command");
504
505 self.mpv.command("stop", &[]).map_err(|e| PlayerError {
506 message: format!("Failed to stop: {:?}", e),
507 })?;
508
509 let mut state = self.state.lock_safe();
510 state.current_media = None;
511
512 Ok(())
513 }
514
515 fn seek(&mut self, position: f64) -> Result<(), PlayerError> {
516 debug!("[MpvBackend] Seek to {} seconds", position);
517
518 let now = SystemTime::now()
520 .duration_since(UNIX_EPOCH)
521 .unwrap()
522 .as_millis() as u64;
523 self.last_seek_time.store(now, Ordering::Relaxed);
524
525 self.mpv
526 .set_property("time-pos", position)
527 .map_err(|e| PlayerError {
528 message: format!("Failed to seek: {:?}", e),
529 })?;
530
531 self.observed.lock_safe().record_position(position);
534
535 Ok(())
536 }
537
538 fn set_volume(&mut self, volume: f32) -> Result<(), PlayerError> {
539 let clamped = volume.clamp(0.0, 1.0);
540 debug!("[MpvBackend] Set volume to {}", clamped);
541
542 let mpv_volume = volume_to_percent(clamped as f64) as i64;
544
545 self.mpv
546 .set_property("volume", mpv_volume)
547 .map_err(|e| PlayerError {
548 message: format!("Failed to set volume: {:?}", e),
549 })?;
550
551 let mut state = self.state.lock_safe();
552 state.volume = clamped;
553
554 Ok(())
555 }
556
557 fn position(&self) -> f64 {
566 let live = self.mpv.get_property::<f64>("time-pos").ok();
567 self.observed.lock_safe().position_or_last(live)
568 }
569
570 fn duration(&self) -> Option<f64> {
575 let live = self.mpv.get_property::<f64>("duration").ok();
576 self.observed.lock_safe().duration_or_last(live)
577 }
578
579 fn state(&self) -> PlayerState {
580 let state = self.state.lock_safe();
581
582 if let Some(ref media) = state.current_media {
583 let is_paused = self.mpv.get_property::<bool>("pause").unwrap_or(true);
584 let position = self.position();
585 let duration = self.duration().unwrap_or(0.0);
586
587 if is_paused {
588 PlayerState::Paused {
589 media: media.clone(),
590 position,
591 duration,
592 }
593 } else {
594 PlayerState::Playing {
595 media: media.clone(),
596 position,
597 duration,
598 }
599 }
600 } else {
601 PlayerState::Idle
602 }
603 }
604
605 fn volume(&self) -> f32 {
606 let state = self.state.lock_safe();
607 state.volume
608 }
609
610 fn set_audio_settings(&mut self, settings: &AudioSettings) -> Result<(), PlayerError> {
611 info!("[MpvBackend] Applying audio settings");
612 self.audio_settings = settings.clone();
613
614 if settings.gapless_playback {
616 self.mpv
617 .set_property("gapless-audio", "yes")
618 .map_err(|e| PlayerError {
619 message: format!("Failed to enable gapless: {:?}", e),
620 })?;
621 } else {
622 self.mpv
623 .set_property("gapless-audio", "no")
624 .map_err(|e| PlayerError {
625 message: format!("Failed to disable gapless: {:?}", e),
626 })?;
627 }
628
629 let af = build_af_filter(settings);
635 self.mpv
636 .set_property("af", af.as_str())
637 .map_err(|e| PlayerError {
638 message: format!("Failed to set audio filters: {:?}", e),
639 })?;
640
641 Ok(())
644 }
645
646 fn audio_settings(&self) -> AudioSettings {
647 self.audio_settings.clone()
648 }
649}
650
651fn build_af_filter(settings: &AudioSettings) -> String {
660 let mut entries = eq_filter_entries(settings.equalizer_enabled, &settings.equalizer_bands);
661 if let Some(norm) = normalize_filter_entry(settings.normalize_volume, settings.volume_level) {
662 entries.push(norm);
663 }
664
665 if entries.is_empty() {
666 return String::new();
667 }
668 format!("lavfi=[{}]", entries.join(","))
669}
670
671fn eq_filter_entries(enabled: bool, bands: &[f32]) -> Vec<String> {
680 if !enabled {
681 return Vec::new();
682 }
683 bands
684 .iter()
685 .zip(EQ_BANDS.iter())
686 .filter(|(gain, _)| gain.abs() >= 0.05) .map(|(gain, freq)| {
688 format!("equalizer=f={}:width_type=o:width=1:g={}", freq, gain)
690 })
691 .collect()
692}
693
694const NORMALIZE_REF_PEAK: f32 = 0.87;
697const NORMALIZE_REF_LUFS: f32 = -14.0;
699
700fn normalize_filter_entry(enabled: bool, level: VolumeLevel) -> Option<String> {
714 if !enabled {
715 return None;
716 }
717 let db_offset = level.target_lufs() - NORMALIZE_REF_LUFS;
719 let peak = (NORMALIZE_REF_PEAK * 10f32.powf(db_offset / 20.0)).clamp(0.10, 0.99);
720 Some(format!("dynaudnorm=p={:.3}:g=15", peak))
722}
723
724impl Drop for MpvBackend {
725 fn drop(&mut self) {
726 info!("[MpvBackend] Shutting down");
727 }
729}
730
731#[cfg(test)]
732mod af_filter_tests {
733 use super::{build_af_filter, eq_filter_entries, normalize_filter_entry};
734 use crate::settings::{AudioSettings, VolumeLevel};
735
736 fn settings() -> AudioSettings {
737 AudioSettings {
738 equalizer_enabled: false,
739 equalizer_bands: vec![0.0; 10],
740 normalize_volume: false,
741 ..AudioSettings::default()
742 }
743 }
744
745 #[test]
749 fn test_eq_entries_empty_when_disabled_or_flat() {
750 assert!(eq_filter_entries(false, &[5.0, -3.0, 2.0]).is_empty());
751 assert!(eq_filter_entries(true, &[0.0; 10]).is_empty());
752 assert!(eq_filter_entries(true, &[0.01, -0.02]).is_empty());
754 }
755
756 #[test]
761 fn test_eq_filter_builds_lavfi_chain() {
762 let mut s = settings();
764 s.equalizer_enabled = true;
765 s.equalizer_bands = vec![5.0, 0.0, -2.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0];
766 let af = build_af_filter(&s);
767 assert!(af.starts_with("lavfi=["), "wrapped in lavfi: {af}");
768 assert!(af.ends_with("]"));
769 assert!(af.contains("equalizer=f=31:width_type=o:width=1:g=5"));
770 assert!(af.contains("equalizer=f=125:width_type=o:width=1:g=-2"));
771 assert_eq!(af.matches("equalizer=").count(), 2);
773 }
774
775 #[test]
780 fn test_normalize_disabled_produces_no_filter() {
781 assert!(normalize_filter_entry(false, VolumeLevel::Normal).is_none());
782 assert_eq!(build_af_filter(&settings()), "");
783 }
784
785 #[test]
790 fn test_normalize_peak_preserves_preset_ordering() {
791 fn peak_of(entry: &str) -> f32 {
792 entry
794 .split("p=")
795 .nth(1)
796 .and_then(|s| s.split(':').next())
797 .and_then(|s| s.parse().ok())
798 .expect("parseable peak")
799 }
800
801 let loud = normalize_filter_entry(true, VolumeLevel::Loud).unwrap();
802 let normal = normalize_filter_entry(true, VolumeLevel::Normal).unwrap();
803 let quiet = normalize_filter_entry(true, VolumeLevel::Quiet).unwrap();
804 for entry in [&loud, &normal, &quiet] {
805 assert!(
806 entry.starts_with("dynaudnorm="),
807 "dynaudnorm filter: {entry}"
808 );
809 }
810 assert!(
811 peak_of(&loud) > peak_of(&normal) && peak_of(&normal) > peak_of(&quiet),
812 "Loud {} > Normal {} > Quiet {}",
813 peak_of(&loud),
814 peak_of(&normal),
815 peak_of(&quiet),
816 );
817 for p in [peak_of(&loud), peak_of(&normal), peak_of(&quiet)] {
819 assert!(p > 0.0 && p <= 0.99, "peak in range: {p}");
820 }
821
822 let mut s = settings();
823 s.normalize_volume = true;
824 s.volume_level = VolumeLevel::Quiet;
825 let af = build_af_filter(&s);
826 assert!(af.starts_with("lavfi=["));
827 assert!(af.contains("dynaudnorm=p="));
828 }
829
830 #[test]
835 fn test_eq_and_normalize_combine_in_order() {
836 let mut s = settings();
837 s.equalizer_enabled = true;
838 s.equalizer_bands = vec![5.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0];
839 s.normalize_volume = true;
840 s.volume_level = VolumeLevel::Normal;
841 let af = build_af_filter(&s);
842
843 let eq_pos = af.find("equalizer=").expect("has EQ");
844 let norm_pos = af.find("dynaudnorm=").expect("has normalizer");
845 assert!(eq_pos < norm_pos, "normalizer runs after EQ: {af}");
846 }
847}