use serde::{Deserialize, Serialize}; use std::process::{Child, Command}; use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use tokio::sync::Mutex; use crate::config::PlayerConfig; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum PlaybackState { Stopped, Playing, Paused, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PlayerStatus { pub state: PlaybackState, pub media_type: String, pub title: String, pub artist: String, pub genre: String, pub stream_title: String, pub time_pos: f64, pub duration: f64, pub volume: f64, pub is_muted: bool, pub error_message: Option, } impl Default for PlayerStatus { fn default() -> Self { Self { state: PlaybackState::Stopped, media_type: String::new(), title: String::new(), artist: String::new(), genre: String::new(), stream_title: String::new(), time_pos: 0.0, duration: 0.0, volume: 100.0, is_muted: false, error_message: None, } } } pub struct PlayerController { config: PlayerConfig, child_process: Mutex>, pub status: Arc>, status_tx: tokio::sync::watch::Sender, status_rx: tokio::sync::watch::Receiver, pub window_id: Mutex>, pub ipc_tx: Mutex>>, } impl PlayerController { pub fn new(config: PlayerConfig) -> Arc { let initial_status = PlayerStatus::default(); let (status_tx, status_rx) = tokio::sync::watch::channel(initial_status.clone()); Arc::new(Self { config, child_process: Mutex::new(None), status: Arc::new(Mutex::new(initial_status)), status_tx, status_rx, window_id: Mutex::new(None), ipc_tx: Mutex::new(None), }) } pub fn status_watch_rx(&self) -> tokio::sync::watch::Receiver { self.status_rx.clone() } pub async fn notify_status(&self) { let st = self.status.lock().await.clone(); let _ = self.status_tx.send(st); } pub async fn set_window_id(&self, wid: Option) { let mut guard = self.window_id.lock().await; *guard = wid; if let Some(w) = wid { tracing::info!("Registered window handle for MPV embedding: {}", w); } } pub async fn play(&self, url: &str, title: &str) -> Result<(), Box> { let media_type = if url.contains("youtube.com") || url.contains("youtu.be") || (url.starts_with("http") && !url.contains("/Videos/") && !url.contains("/Audio/")) { "stream" } else { "video" }; self.play_media(url, title, media_type, None, None).await } pub async fn play_media( &self, url: &str, title: &str, media_type: &str, artist: Option<&str>, genre: Option<&str>, ) -> Result<(), Box> { tracing::info!("Playing media [type={}]: '{}' -> {}", media_type, title, url); // Update status immediately { let mut st = self.status.lock().await; st.state = PlaybackState::Playing; st.media_type = media_type.to_string(); st.title = title.to_string(); st.artist = artist.unwrap_or_default().to_string(); st.genre = genre.unwrap_or_default().to_string(); st.stream_title = if media_type == "stream" { title.to_string() } else { String::new() }; st.time_pos = 0.0; st.error_message = None; } self.notify_status().await; // Clean up socket file if exists let sock_path = std::path::Path::new(&self.config.ipc_socket); if sock_path.exists() { let _ = std::fs::remove_file(sock_path); } // Kill any existing instance { let mut proc_guard = self.child_process.lock().await; if let Some(mut child) = proc_guard.take() { let _ = child.kill(); } } // Build MPV command let mut cmd = Command::new(&self.config.mpv_path); cmd.arg(format!("--input-ipc-server={}", self.config.ipc_socket)); cmd.arg(format!("--hwdec={}", self.config.hwdec)); cmd.arg(format!("--vo={}", self.config.vo)); cmd.arg("--osd-bar=yes"); cmd.arg("--osd-level=1"); cmd.arg("--osd-font-size=28"); cmd.arg("--osd-on-seek=msg-bar"); cmd.arg("--input-vo-keyboard=no"); cmd.arg("--input-cursor=no"); cmd.arg("--input-cursor-passthrough=yes"); // Configure yt-dlp for YouTube streams let ytdlp_path = if std::path::Path::new("/home/bart/.local/bin/yt-dlp").exists() { "/home/bart/.local/bin/yt-dlp" } else { "yt-dlp" }; cmd.arg(format!("--script-opts=ytdl_hook-ytdl_path={}", ytdlp_path)); cmd.arg("--ytdl-format=bestvideo[height<=?1080]+bestaudio/best"); cmd.arg("--ytdl-raw-options=no-playlist="); let wid = *self.window_id.lock().await; if let Some(w) = wid { tracing::info!("Embedding MPV directly into window ID: {}", w); cmd.arg(format!("--wid={}", w)); } else if self.config.fullscreen { cmd.arg("--fs"); } cmd.arg(format!("--title={}", title)); // Strip playlist/index params from single YouTube video playback let play_url = if url.contains("youtube.com") || url.contains("youtu.be") { let mut u = url.to_string(); if let Some(pos) = u.find("&list=") { u.truncate(pos); } if let Some(pos) = u.find("&index=") { u.truncate(pos); } u } else { url.to_string() }; cmd.arg(play_url); match cmd.spawn() { Ok(child) => { let mut proc_guard = self.child_process.lock().await; *proc_guard = Some(child); } Err(e) => { let msg = format!("MPV niet gevonden ('{}')! Installeer MPV via: sudo apt install mpv", self.config.mpv_path); tracing::warn!("{}: {}", msg, e); let mut st = self.status.lock().await; st.state = PlaybackState::Stopped; st.error_message = Some(msg); return Err(format!("Could not launch MPV: {}", e).into()); } } // Create persistent IPC command channel let (ipc_tx, ipc_rx) = tokio::sync::mpsc::unbounded_channel::(); { let mut guard = self.ipc_tx.lock().await; *guard = Some(ipc_tx); } // Start background IPC monitoring task let sock_file = self.config.ipc_socket.clone(); let status_arc = self.status.clone(); let status_tx = self.status_tx.clone(); tokio::spawn(async move { let _ = monitor_mpv_ipc(sock_file, status_arc, status_tx, ipc_rx).await; }); Ok(()) } pub async fn play_pause(&self) { let current_state = { let st = self.status.lock().await; st.state }; let new_state = match current_state { PlaybackState::Playing => PlaybackState::Paused, PlaybackState::Paused => PlaybackState::Playing, PlaybackState::Stopped => PlaybackState::Stopped, }; let _ = self.send_ipc_cmd(&["osd-msg", "cycle", "pause"]).await; { let mut st = self.status.lock().await; st.state = new_state; } self.notify_status().await; } pub async fn stop(&self) { let _ = self.send_ipc_cmd(&["stop"]).await; { let mut guard = self.ipc_tx.lock().await; *guard = None; } let mut proc_guard = self.child_process.lock().await; if let Some(mut child) = proc_guard.take() { let _ = child.kill(); } { let mut st = self.status.lock().await; st.state = PlaybackState::Stopped; st.media_type.clear(); st.title.clear(); st.artist.clear(); st.genre.clear(); st.stream_title.clear(); st.time_pos = 0.0; st.duration = 0.0; } self.notify_status().await; } pub async fn seek(&self, seconds: f64) { let _ = self.send_ipc_cmd(&["osd-msg-bar", "seek", &seconds.to_string(), "relative"]).await; } pub async fn seek_percent(&self, percent: f64) { let p = percent.clamp(0.0, 100.0); let _ = self.send_ipc_cmd(&["osd-msg-bar", "seek", &p.to_string(), "absolute-percent"]).await; } pub async fn seek_time(&self, seconds: f64) { let s = seconds.max(0.0); let _ = self.send_ipc_cmd(&["osd-msg-bar", "seek", &s.to_string(), "absolute"]).await; } pub async fn toggle_mute(&self) { let new_muted = { let mut st = self.status.lock().await; st.is_muted = !st.is_muted; st.is_muted }; let _ = self.send_ipc_cmd(&["cycle", "mute"]).await; self.notify_status().await; tracing::info!("Player mute toggled to {}", new_muted); } pub async fn set_mute(&self, muted: bool) { { let mut st = self.status.lock().await; st.is_muted = muted; } let arg = if muted { "yes" } else { "no" }; let _ = self.send_ipc_cmd(&["set", "mute", arg]).await; self.notify_status().await; tracing::info!("Player mute set to {}", muted); } pub async fn set_volume(&self, volume: f64) { let vol = volume.clamp(0.0, 100.0); { let mut st = self.status.lock().await; st.volume = vol; } let _ = self.send_ipc_cmd(&["osd-msg-bar", "set", "volume", &vol.to_string()]).await; self.notify_status().await; tracing::info!("Player volume set to {}", vol); } pub async fn volume_up(&self) { let new_vol = { let mut st = self.status.lock().await; st.volume = (st.volume + 5.0).min(100.0); st.volume }; let _ = self.send_ipc_cmd(&["osd-msg-bar", "add", "volume", "5"]).await; self.notify_status().await; tracing::info!("Player volume increased to {}", new_vol); } pub async fn volume_down(&self) { let new_vol = { let mut st = self.status.lock().await; st.volume = (st.volume - 5.0).max(0.0); st.volume }; let _ = self.send_ipc_cmd(&["osd-msg-bar", "add", "volume", "-5"]).await; self.notify_status().await; tracing::info!("Player volume decreased to {}", new_vol); } pub async fn cycle_subtitle(&self) { let _ = self.send_ipc_cmd(&["cycle", "sub"]).await; let _ = self.send_ipc_cmd(&[ "expand-properties", "show-text", "💬 Ondertitels: ${current-tracks/sub/title:${current-tracks/sub/lang:${sub:Geen beschikbaar}}}", "2500", ]).await; tracing::info!("Cycled subtitle track"); } pub async fn cycle_audio(&self) { let _ = self.send_ipc_cmd(&["cycle", "audio"]).await; let _ = self.send_ipc_cmd(&[ "expand-properties", "show-text", "🎬 Audio: ${current-tracks/audio/title:${current-tracks/audio/lang:${audio:Standaard}}}", "2500", ]).await; tracing::info!("Cycled audio track"); } pub async fn set_video_visible(&self, visible: bool) { let arg = if visible { "auto" } else { "no" }; let _ = self.send_ipc_cmd(&["set", "vid", arg]).await; tracing::info!("Set MPV video visibility to {}", visible); } async fn send_ipc_cmd(&self, args: &[&str]) -> Result<(), Box> { let guard = self.ipc_tx.lock().await; if let Some(ref tx) = *guard { let cmd_json = serde_json::json!({ "command": args }); let _ = tx.send(cmd_json); return Ok(()); } Err("MPV IPC not connected".into()) } } async fn monitor_mpv_ipc( sock_path: String, status: Arc>, status_tx: tokio::sync::watch::Sender, mut ipc_rx: tokio::sync::mpsc::UnboundedReceiver, ) -> Result<(), Box> { // Wait for socket to become available for _ in 0..30 { if std::path::Path::new(&sock_path).exists() { break; } tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; } let stream = match UnixStream::connect(&sock_path).await { Ok(s) => s, Err(_) => return Ok(()), }; let (reader, mut writer) = stream.into_split(); let mut lines = BufReader::new(reader).lines(); // Observe properties let observe_cmds = [ r#"{"command":["observe_property",1,"time-pos"]}"#, r#"{"command":["observe_property",2,"duration"]}"#, r#"{"command":["observe_property",3,"pause"]}"#, r#"{"command":["observe_property",4,"volume"]}"#, r#"{"command":["observe_property",5,"media-title"]}"#, r#"{"command":["observe_property",6,"icy-title"]}"#, r#"{"command":["observe_property",7,"metadata"]}"#, r#"{"command":["observe_property",8,"mute"]}"#, ]; for c in &observe_cmds { let _ = writer.write_all(format!("{}\n", c).as_bytes()).await; } loop { tokio::select! { cmd_opt = ipc_rx.recv() => { match cmd_opt { Some(cmd_val) => { let payload = format!("{}\n", cmd_val); if let Err(e) = writer.write_all(payload.as_bytes()).await { tracing::warn!("MPV IPC write error: {}", e); break; } } None => break, } } line_res = lines.next_line() => { match line_res { Ok(Some(line)) => { if let Ok(v) = serde_json::from_str::(&line) { if let Some(event) = v.get("event").and_then(|e| e.as_str()) { if event == "property-change" { let name = v.get("name").and_then(|n| n.as_str()).unwrap_or(""); let data = v.get("data"); let mut st = status.lock().await; let mut notify = false; match name { "mute" => { if let Some(m) = data.and_then(|d| d.as_bool()) { if st.is_muted != m { st.is_muted = m; notify = true; } } } "time-pos" => { if let Some(pos) = data.and_then(|d| d.as_f64()) { st.time_pos = pos; } } "duration" => { if let Some(dur) = data.and_then(|d| d.as_f64()) { st.duration = dur; } } "pause" => { if let Some(paused) = data.and_then(|d| d.as_bool()) { let new_state = if paused { PlaybackState::Paused } else { PlaybackState::Playing }; if st.state != new_state { st.state = new_state; notify = true; } } } "volume" => { if let Some(vol) = data.and_then(|d| d.as_f64()) { if (st.volume - vol).abs() > 0.01 { st.volume = vol; notify = true; } } } "icy-title" => { if let Some(icy) = data.and_then(|d| d.as_str()) { let trimmed = icy.trim(); if !trimmed.is_empty() && st.stream_title != trimmed { tracing::info!("MPV ICY Stream title detected: {}", trimmed); st.stream_title = trimmed.to_string(); if st.media_type.is_empty() || st.media_type != "stream" { st.media_type = "stream".to_string(); } notify = true; } } } "media-title" => { if let Some(mt) = data.and_then(|d| d.as_str()) { let trimmed = mt.trim(); if !trimmed.is_empty() { if st.media_type == "stream" { if st.stream_title.is_empty() || st.stream_title != trimmed { tracing::info!("MPV Stream media-title: {}", trimmed); st.stream_title = trimmed.to_string(); notify = true; } } else if st.title.is_empty() { st.title = trimmed.to_string(); notify = true; } } } } "metadata" => { if let Some(meta_obj) = data.and_then(|d| d.as_object()) { let stream_t = meta_obj.get("icy-title") .or_else(|| meta_obj.get("icy_title")) .or_else(|| meta_obj.get("StreamTitle")) .or_else(|| meta_obj.get("title")) .or_else(|| meta_obj.get("Title")) .and_then(|val| val.as_str()); if let Some(s) = stream_t { let trimmed = s.trim(); if !trimmed.is_empty() && (st.media_type == "stream" || meta_obj.contains_key("icy-title")) { if st.stream_title != trimmed { st.stream_title = trimmed.to_string(); st.media_type = "stream".to_string(); notify = true; } } } if let Some(a) = meta_obj.get("artist").or_else(|| meta_obj.get("Artist")).and_then(|val| val.as_str()) { let trimmed = a.trim(); if !trimmed.is_empty() && st.artist != trimmed { st.artist = trimmed.to_string(); notify = true; } } if let Some(g) = meta_obj.get("genre").or_else(|| meta_obj.get("Genre")).or_else(|| meta_obj.get("icy-genre")).and_then(|val| val.as_str()) { let trimmed = g.trim(); if !trimmed.is_empty() && st.genre != trimmed { st.genre = trimmed.to_string(); notify = true; } } } } _ => {} } if notify { let _ = status_tx.send(st.clone()); } } else if event == "end-file" { let mut st = status.lock().await; st.state = PlaybackState::Stopped; st.media_type.clear(); st.title.clear(); st.artist.clear(); st.genre.clear(); st.stream_title.clear(); st.time_pos = 0.0; st.duration = 0.0; let _ = status_tx.send(st.clone()); break; } } } } _ => break, } } } } Ok(()) }