pipewire-soundpad/pwsp-lib/src/utils/pipewire.rs

use crate::types::pipewire::{AudioDevice, DeviceType, Port, Terminate};
use anyhow::{Result, anyhow};
use pipewire::{
    context::ContextRc, link::Link, main_loop::MainLoopRc, properties::properties,
    registry::GlobalObject, spa::utils::dict::DictRef,
};
use std::{collections::HashMap, thread};
use tokio::{
    sync::mpsc,
    time::{Duration, timeout},
};

pub fn setup_pipewire_context() -> Result<(MainLoopRc, ContextRc), String> {
    pipewire::init();
    let main_loop = MainLoopRc::new(None).map_err(|e| e.to_string())?;
    let context = ContextRc::new(&main_loop, None).map_err(|e| e.to_string())?;
    Ok((main_loop, context))
}

fn parse_global_object(
    global_object: &GlobalObject<&DictRef>,
) -> (Option<AudioDevice>, Option<Port>) {
    let props = match global_object.props {
        Some(p) => p,
        None => return (None, None),
    };

    if let Some(media_class) = props.get("media.class") {
        let node_id = global_object.id;
        let node_nick = props.get("node.nick");
        let node_name = props.get("node.name");
        let node_description = props.get("node.description");

        if media_class.starts_with("Audio/Source") {
            let input_device = AudioDevice::new(
                node_id,
                node_nick,
                node_description,
                node_name,
                DeviceType::Input,
            );
            return (Some(input_device), None);
        } else if media_class.starts_with("Stream/Output/Audio") {
            let output_device = AudioDevice::new(
                node_id,
                node_nick,
                node_description,
                node_name,
                DeviceType::Output,
            );
            return (Some(output_device), None);
        }
        return (None, None);
    }

    if props.get("port.direction").is_some()
        && let (Some(node_id), Some(port_id), Some(port_name)) = (
            props.get("node.id").and_then(|id| id.parse::<u32>().ok()),
            props.get("port.id").and_then(|id| id.parse::<u32>().ok()),
            props.get("port.name"),
        )
    {
        let port = Port {
            node_id,
            port_id,
            name: port_name.to_string(),
        };
        return (None, Some(port));
    }

    (None, None)
}

async fn pw_get_global_objects_thread(
    main_sender: mpsc::Sender<(Option<AudioDevice>, Option<Port>)>,
    pw_receiver: pipewire::channel::Receiver<Terminate>,
    init_sender: tokio::sync::oneshot::Sender<Result<(), String>>,
) {
    let (main_loop, context) = match setup_pipewire_context() {
        Ok(res) => res,
        Err(e) => {
            let _ = init_sender.send(Err(e));
            return;
        }
    };

    // Stop main loop on Terminate message
    let _receiver = pw_receiver.attach(main_loop.loop_(), {
        let _main_loop = main_loop.clone();
        move |_| _main_loop.quit()
    });

    let core = match context.connect(None) {
        Ok(core) => core,
        Err(e) => {
            let _ = init_sender.send(Err(format!("Failed to connect to pipewire context: {}", e)));
            return;
        }
    };

    let registry = match core.get_registry() {
        Ok(registry) => registry,
        Err(e) => {
            let _ = init_sender.send(Err(format!(
                "Failed to get registry from pipewire context: {}",
                e
            )));
            return;
        }
    };

    let _listener = registry
        .add_listener_local()
        .global(move |global| {
            // Try to parse every global object pipewire finds
            let (device, port) = parse_global_object(global);

            // Send message to the main thread
            let sender_clone = main_sender.clone();
            tokio::task::spawn(async move {
                sender_clone.send((device, port)).await.ok();
            });
        })
        .register();

    // Signal successful initialization
    if init_sender.send(Ok(())).is_err() {
        return;
    }

    main_loop.run();
}

pub async fn get_all_devices() -> Result<(Vec<AudioDevice>, Vec<AudioDevice>)> {
    // Channels to communicate with pipewire thread
    let (main_sender, mut main_receiver) = mpsc::channel(10);
    let (pw_sender, pw_receiver) = pipewire::channel::channel();
    let (init_sender, init_receiver) = tokio::sync::oneshot::channel();

    // Spawn pipewire thread in background
    let _pw_thread = tokio::spawn(async move {
        pw_get_global_objects_thread(main_sender, pw_receiver, init_sender).await
    });

    // Wait for initialization to complete
    if let Err(e) = init_receiver.await {
        return Err(anyhow!(e));
    }

    let mut input_devices: HashMap<u32, AudioDevice> = HashMap::new();
    let mut output_devices: HashMap<u32, AudioDevice> = HashMap::new();
    let mut ports: Vec<Port> = vec![];

    loop {
        // If we don't receive a message in 100ms, we can assume that pipewire thread is finished
        match timeout(Duration::from_millis(100), main_receiver.recv()).await {
            Ok(Some((device, port))) => {
                if let Some(device) = device {
                    match device.device_type {
                        DeviceType::Input => {
                            input_devices.insert(device.id, device);
                        }
                        DeviceType::Output => {
                            output_devices.insert(device.id, device);
                        }
                    }
                } else if let Some(port) = port {
                    ports.push(port);
                }
            }
            Ok(None) | Err(_) => {
                // Pipewire thread is finished and we can collect our devices
                let _ = pw_sender.send(Terminate {});

                for port in ports {
                    let node_id = port.node_id;

                    if let Some(input_device) = input_devices.get_mut(&node_id) {
                        input_device.add_port(port);
                    } else if let Some(output_device) = output_devices.get_mut(&node_id) {
                        output_device.add_port(port);
                    }
                }

                let mut input_devices: Vec<AudioDevice> = input_devices.into_values().collect();
                let mut output_devices: Vec<AudioDevice> = output_devices.into_values().collect();

                input_devices.sort_by_key(|a| a.id);
                output_devices.sort_by_key(|a| a.id);

                return Ok((input_devices, output_devices));
            }
        }
    }
}

pub async fn get_device(device_name: &str) -> Result<AudioDevice> {
    let (input_devices, output_devices) = get_all_devices().await?;

    input_devices
        .into_iter()
        .chain(output_devices)
        .find(|device| {
            device.name == device_name
                || device.nick == device_name
                || device.name.contains(device_name)
                || device.nick.contains(device_name)
        })
        .ok_or_else(|| anyhow!("Device not found"))
}

pub fn create_virtual_mic() -> Result<pipewire::channel::Sender<Terminate>> {
    let (pw_sender, pw_receiver) = pipewire::channel::channel::<Terminate>();
    let (init_sender, init_receiver) = std::sync::mpsc::sync_channel(0);

    let _pw_thread = thread::spawn(move || {
        let (main_loop, context) = match setup_pipewire_context() {
            Ok(res) => res,
            Err(e) => {
                let _ = init_sender.send(Err(e));
                return;
            }
        };
        let core = match context.connect(None) {
            Ok(core) => core,
            Err(e) => {
                let _ =
                    init_sender.send(Err(format!("Failed to connect to pipewire context: {}", e)));
                return;
            }
        };

        let props = properties!(
            "factory.name" => "support.null-audio-sink",
            "node.name" => "pwsp-virtual-mic",
            "node.description" => "PWSP Virtual Mic",
            "media.class" => "Audio/Source/Virtual",
            "audio.position" => "[ FL FR ]",
            "audio.channels" => "2",
            "object.linger" => "false", // Destroy the node on app exit
        );

        let _node = match core.create_object::<pipewire::node::Node>("adapter", &props) {
            Ok(node) => node,
            Err(e) => {
                let _ = init_sender.send(Err(format!("Failed to create virtual mic: {}", e)));
                return;
            }
        };

        let _receiver = pw_receiver.attach(main_loop.loop_(), {
            let _main_loop = main_loop.clone();
            move |_| _main_loop.quit()
        });

        println!("Virtual mic created");
        if init_sender.send(Ok(())).is_err() {
            return;
        }
        main_loop.run();
    });

    if let Err(e) = init_receiver.recv()? {
        return Err(anyhow!(e));
    }

    Ok(pw_sender)
}

pub async fn link_player_to_virtual_mic() -> Result<pipewire::channel::Sender<Terminate>> {
    let pwsp_daemon_output = match get_device("pwsp-daemon").await {
        Ok(device) => device,
        Err(_) => {
            return Err(anyhow!(
                "Could not find alsa_playback.pwsp-daemon device, skipping device linking"
            ));
        }
    };

    let pwsp_daemon_input = match get_device("pwsp-virtual-mic").await {
        Ok(device) => device,
        Err(_) => {
            return Err(anyhow!(
                "Could not find pwsp-virtual-mic device, skipping device linking"
            ));
        }
    };

    let output_fl = match pwsp_daemon_output.output_fl {
        Some(port) => port,
        None => return Err(anyhow!("Failed to get pwsp-daemon output_fl")),
    };
    let output_fr = match pwsp_daemon_output.output_fr {
        Some(port) => port,
        None => return Err(anyhow!("Failed to get pwsp-daemon output_fr")),
    };
    let input_fl = match pwsp_daemon_input.input_fl {
        Some(port) => port,
        None => return Err(anyhow!("Failed to get pwsp-virtual-mic input_fl")),
    };
    let input_fr = match pwsp_daemon_input.input_fr {
        Some(port) => port,
        None => return Err(anyhow!("Failed to get pwsp-virtual-mic input_fr")),
    };

    create_link(output_fl, output_fr, input_fl, input_fr)
}

pub fn create_link(
    output_fl: Port,
    output_fr: Port,
    input_fl: Port,
    input_fr: Port,
) -> Result<pipewire::channel::Sender<Terminate>> {
    let (pw_sender, pw_receiver) = pipewire::channel::channel::<Terminate>();
    let (init_sender, init_receiver) = std::sync::mpsc::sync_channel(0);

    let _pw_thread = thread::spawn(move || {
        let (main_loop, context) = match setup_pipewire_context() {
            Ok(res) => res,
            Err(e) => {
                let _ = init_sender.send(Err(e));
                return;
            }
        };
        let core = match context.connect(None) {
            Ok(core) => core,
            Err(e) => {
                let _ =
                    init_sender.send(Err(format!("Failed to connect to pipewire context: {}", e)));
                return;
            }
        };

        let props_fl = properties! {
            "link.output.node" => format!("{}", output_fl.node_id).as_str(),
            "link.output.port" => format!("{}", output_fl.port_id).as_str(),
            "link.input.node"  => format!("{}", input_fl.node_id).as_str(),
            "link.input.port"  => format!("{}", input_fl.port_id).as_str(),
        };
        let props_fr = properties! {
            "link.output.node" => format!("{}", output_fr.node_id).as_str(),
            "link.output.port" => format!("{}", output_fr.port_id).as_str(),
            "link.input.node"  => format!("{}", input_fr.node_id).as_str(),
            "link.input.port"  => format!("{}", input_fr.port_id).as_str(),
        };

        let _link_fl = match core.create_object::<Link>("link-factory", &props_fl) {
            Ok(link) => link,
            Err(e) => {
                let _ = init_sender.send(Err(format!("Failed to create link FL: {}", e)));
                return;
            }
        };
        let _link_fr = match core.create_object::<Link>("link-factory", &props_fr) {
            Ok(link) => link,
            Err(e) => {
                let _ = init_sender.send(Err(format!("Failed to create link FR: {}", e)));
                return;
            }
        };

        let _receiver = pw_receiver.attach(main_loop.loop_(), {
            let _main_loop = main_loop.clone();
            move |_| _main_loop.quit()
        });

        println!(
            "Link created: FL: {}-{} FR: {}-{}",
            output_fl.node_id, input_fl.node_id, output_fr.node_id, input_fr.node_id
        );
        if init_sender.send(Ok(())).is_err() {
            return;
        }
        main_loop.run();
    });

    if let Err(e) = init_receiver.recv()? {
        return Err(anyhow!(e));
    }

    Ok(pw_sender)
}