Introduce initial implementation of the vmic CLI with PipeWire integration

- Added `Cargo.toml` to define dependencies and project metadata.
- Implemented core CLI functionality for managing virtual microphones (`create`, `edit`, `delete`, `list`, `route`, and `wipe` commands) using `clap`.
- Integrated `rusqlite` for persistent state storage and `pipewire` to manage PipeWire nodes.
- Ensured graceful handling of feedback loops and system defaults during virtual mic creation and deletion.
- Added error handling via `thiserror` for cleaner error definitions.
This commit is contained in:
2026-08-11 13:28:33 +02:00
parent 76c129a5dc
commit e19f398397
18 changed files with 2151 additions and 1 deletions

27
Cargo.toml Normal file
View File

@@ -0,0 +1,27 @@
[package]
name = "vmic"
version = "0.1.0"
edition = "2021"
description = "Create and manage PipeWire virtual microphones."
license = "AGPL-3+"
[[bin]]
name = "vmic"
path = "src/main.rs"
[dependencies]
clap = { version = "4", features = ["derive"] }
clap_complete = "4"
thiserror = "^2"
dirs = "6"
rusqlite = { version = "^0", features = ["bundled"] }
pipewire = "=0.8.0"
libc = "0.2"
# Keep panics from corrupting the sqlite connection mid-transaction; abort
# is fine for a CLI as there is no long-lived state to unwind for.
[profile.release]
panic = "abort"
lto = true

View File

@@ -1,3 +1,80 @@
# vmic
Create and manage PipeWire virtual microphones easily.
Create and manage PipeWire virtual microphones easily, from the command line.
## What it does
A vmic is two chained loopback stages:
```
apps -> vmic_X_sink =(stage 1)=> vmic_X_mid -> vmic_X_mix =(stage 2)=> vmic_X_mic -> recorders
^ hardware mic joins here (route -s)
```
Point an app's output device at `vmic_X_sink`, and a recording app's input device at
`vmic_X_mic`. Optionally mix in a real microphone downstream, and optionally hear your
own output via a self-monitor loopback (which never carries the mixed-in mic).
## Requirements
- PipeWire + pipewire-pulse (`pactl`, `pw-loopback`, `pw-link` on `PATH`)
- Rust (edition 2021)
## Build
```
cargo build --release
```
If the build fails inside `libspa-sys` (e.g. `no field 'data' on type 'spa_pod_builder'`),
your system clang is newer than the pinned `bindgen` version supports. Point `LIBCLANG_PATH`
at an older libclang - a quick fix is a pip-vendored one:
```
python3 -m venv ~/.local/share/vmic-build/libclang-venv
~/.local/share/vmic-build/libclang-venv/bin/pip install libclang
```
then add to `.cargo/config.toml`:
```toml
[env]
LIBCLANG_PATH = "/home/<you>/.local/share/vmic-build/libclang-venv/lib/python3.*/site-packages/clang/native"
```
## Usage
```
vmic create <name> [-l] # create a vmic, optionally with self-monitor
vmic route <name> -i <app[:media]> # move an app's playback into the vmic
vmic route <name> -o <app[:media]> # move an app's recording off the vmic
vmic route <name> -s <source|off> # mix a hardware mic in, or remove it
vmic edit <name> -l <true|false> # toggle the self-monitor loopback
vmic edit <name> -v <0.8|80> # self-monitor loopback volume
vmic edit <name> -sv <0.8|80> # mixed-in source volume
vmic delete <name> # tear down one vmic
vmic list # show all vmics and their status
vmic wipe # tear down every vmic, tracked or not
```
Aliases: `create` = `mk`/`make`, `delete` = `rm`/`remove`, `list` = `ls`, `wipe` = `reset`.
Shell completions: `vmic completions <bash|zsh|fish|...>`.
## Example
```
vmic create podcast -l
# set your app's output device to vmic_podcast_sink
# set your recorder's input device to vmic_podcast_mic
vmic route podcast -s "USB Microphone"
```
## State
Tracked in a SQLite database at `~/.config/vmic/vmic.db` (override the directory with
`VMIC_STATE_DIR_OVERRIDE`).
## License
APGL-3

90
src/cli.rs Normal file
View File

@@ -0,0 +1,90 @@
use clap::{Args, Parser, Subcommand};
use clap_complete::Shell;
/// vmic - create and manage PipeWire virtual microphones.
#[derive(Parser)]
#[command(name = "vmic", version, about)]
pub struct Cli {
#[command(subcommand)]
pub command: Commands,
}
#[derive(Subcommand)]
pub enum Commands {
/// Create a new virtual microphone.
#[command(visible_alias = "mk", visible_alias = "make")]
Create(CreateArgs),
/// Move app streams and mix a hardware source into a vmic.
Route(RouteArgs),
/// Change loopback and volume settings on a vmic.
Edit(EditArgs),
/// Delete a virtual microphone.
#[command(visible_alias = "rm", visible_alias = "remove")]
Delete(DeleteArgs),
/// List all virtual microphones.
#[command(visible_alias = "ls")]
List,
/// Tear down every vmic, tracked or not.
#[command(visible_alias = "reset")]
Wipe,
/// Generate a shell completion script.
Completions { shell: Shell },
}
#[derive(Args)]
pub struct CreateArgs {
/// Name for the new vmic.
pub name: String,
/// Enable a self-monitor loopback.
#[arg(short, long)]
pub loopback: bool,
}
#[derive(Args)]
pub struct RouteArgs {
/// Name of the vmic to route into.
pub name: String,
/// Move matching sink-inputs into this vmic's sink.
#[arg(short, long = "input", value_name = "APP[:MEDIA]")]
pub inputs: Vec<String>,
/// Move matching source-outputs off this vmic's source.
#[arg(short, long = "output", value_name = "APP[:MEDIA]")]
pub outputs: Vec<String>,
/// Mix a hardware source into the vmic, or "off" to remove it.
#[arg(short, long = "source", value_name = "SOURCE|off")]
pub source: Option<String>,
}
#[derive(Args)]
pub struct EditArgs {
/// Name of the vmic to edit.
pub name: String,
/// Enable or disable the self-monitor loopback.
#[arg(short, long)]
pub loopback: Option<bool>,
/// Loopback volume: fraction (0.8) or percent (80).
#[arg(short, long)]
pub volume: Option<f32>,
/// Mixed-in source volume: fraction (0.8) or percent (80).
#[arg(long = "source-volume", visible_alias = "sv")]
pub source_volume: Option<f32>,
}
#[derive(Args)]
pub struct DeleteArgs {
/// Name of the vmic to delete.
pub name: String,
}

77
src/commands/create.rs Normal file
View File

@@ -0,0 +1,77 @@
use crate::cli::CreateArgs;
use crate::error::{Result, VmicError};
use crate::pw::{loopback, PwGraph};
use crate::state::{self, VmicState};
use crate::ui;
use std::process::Command;
pub fn run(args: CreateArgs) -> Result<()> {
state::require_valid_name(&args.name)?;
let name = state::normalize(&args.name);
let conn = state::open_db()?;
if VmicState::exists(&conn, &name)? {
return Err(VmicError::AlreadyExists(name));
}
let names = loopback::NodeNames::for_vmic(&name);
let graph = PwGraph::connect()?;
// pw-loopback may steal the default sink/source; restore them afterward.
let orig_sink = default_output("get-default-sink");
let orig_source = default_output("get-default-source");
let stages = loopback::create_stages(&graph, &names)?;
if let Some(sink) = &orig_sink {
let _ = Command::new("pactl").args(["set-default-sink", sink]).status();
}
if let Some(source) = &orig_source {
let _ = Command::new("pactl").args(["set-default-source", source]).status();
}
// Insurance: a sink monitor feeding any vmic node is a feedback ring;
// it should never exist, but tear it down if one appears anyway.
let prefix = format!("vmic_{name}_");
let removed = graph.break_feedback_rings(&prefix)?;
if removed > 0 {
ui::warn(&format!("removed {removed} unexpected monitor feedback link(s)."));
}
let mut state = VmicState {
name: name.clone(),
sink_name: names.sink.clone(),
mid_name: names.mid,
mix_name: names.mix,
source_name: names.source.clone(),
sink_node_id: Some(stages.sink_node_id),
mix_node_id: Some(stages.mix_node_id),
sink_pid: Some(stages.sink_pid as i64),
mix_pid: Some(stages.mix_pid as i64),
..Default::default()
};
if args.loopback {
let loop_id = loopback::enable_self_monitor(&graph, &names.sink)?;
state.loopback_id = Some(loop_id);
}
state.save(&conn)?;
ui::ok(&format!("Created virtual microphone '{name}'."));
if let (Some(sink), Some(source)) = (&orig_sink, &orig_source) {
ui::info(&format!(" Default Output/Input kept intact ('{sink}' / '{source}')."));
}
println!(" In the source app, set output device to:");
println!(" '{}'", ui::blue(&names.sink));
println!(" In the receiving app, set input/mic device to:");
println!(" '{}'", ui::blue(&names.source));
Ok(())
}
fn default_output(subcommand: &str) -> Option<String> {
let out = Command::new("pactl").arg(subcommand).output().ok()?;
let s = String::from_utf8_lossy(&out.stdout).trim().to_string();
(!s.is_empty()).then_some(s)
}

31
src/commands/delete.rs Normal file
View File

@@ -0,0 +1,31 @@
use crate::cli::DeleteArgs;
use crate::error::Result;
use crate::pw::{loopback, PwGraph};
use crate::state::{self, VmicState};
use crate::ui;
pub fn run(args: DeleteArgs) -> Result<()> {
let conn = state::open_db()?;
let name = state::normalize(&args.name);
let mut state = VmicState::load(&conn, &name)?;
let graph = PwGraph::connect()?;
if let Some(loopback_id) = state.loopback_id {
loopback::unload_pulse_module(loopback_id)?;
}
// Unlink a mixed-in hardware source explicitly; the links would also
// die with the nodes below.
super::unsource(&graph, &mut state)?;
if let Some(pid) = state.mix_pid {
loopback::terminate_stage(pid, &state.mix_name)?;
}
if let Some(pid) = state.sink_pid {
loopback::terminate_stage(pid, &state.sink_name)?;
}
VmicState::delete(&conn, &name)?;
ui::ok(&format!("Deleted virtual mic '{name}'."));
Ok(())
}

109
src/commands/edit.rs Normal file
View File

@@ -0,0 +1,109 @@
use crate::cli::EditArgs;
use crate::commands::to_pct;
use crate::error::{Result, VmicError};
use crate::pw::{loopback, PwGraph};
use crate::state::{self, VmicState};
use crate::ui;
pub fn run(args: EditArgs) -> Result<()> {
if args.loopback.is_none() && args.volume.is_none() && args.source_volume.is_none() {
return Err(VmicError::NothingToDo(
"pass --loopback, --volume and/or --source-volume.".into(),
));
}
let conn = state::open_db()?;
let name = state::normalize(&args.name);
let mut state = VmicState::load(&conn, &name)?;
let graph = PwGraph::connect()?;
if let Some(enable) = args.loopback {
edit_loopback(&graph, &mut state, enable)?;
}
if let Some(volume) = args.volume {
edit_volume(&graph, &mut state, volume)?;
}
if let Some(source_volume) = args.source_volume {
edit_source_volume(&mut state, source_volume)?;
}
state.save(&conn)
}
fn edit_loopback(graph: &PwGraph, state: &mut VmicState, enable: bool) -> Result<()> {
match (enable, state.loopback_id) {
(true, Some(id)) => {
ui::warn(&format!(
"self-monitor loopback already enabled for '{}' (module {id}).",
state.name
));
}
(true, None) => {
let id = loopback::enable_self_monitor(graph, &state.sink_name)?;
state.loopback_id = Some(id);
ui::ok(&format!("Self-monitor loopback enabled (module {id})."));
// Re-apply a previously stored loopback volume, if any. A
// failure here is a warning, not a hard error - the loopback
// itself was still created successfully.
if let Some(pct) = state.volume_pct {
if loopback::set_loopback_volume(graph, id, pct).is_ok() {
ui::info(&format!("Loopback volume restored to {pct}%."));
} else {
ui::warn(&format!("could not apply stored loopback volume ({pct}%)."));
}
}
}
(false, None) => {
ui::warn(&format!("self-monitor loopback is not enabled for '{}'.", state.name));
}
(false, Some(id)) => {
// A failed unload propagates as a real error rather than being
// silently swallowed.
loopback::unload_pulse_module(id)?;
state.loopback_id = None;
ui::ok(&format!("Self-monitor loopback disabled (module {id} unloaded)."));
}
}
Ok(())
}
fn edit_volume(graph: &PwGraph, state: &mut VmicState, value: f32) -> Result<()> {
let pct = to_pct(value).ok_or_else(|| VmicError::InvalidVolume(value.to_string()))?;
state.volume_pct = Some(pct);
let Some(id) = state.loopback_id else {
ui::ok(&format!(
"Loopback volume for '{}' saved ({pct}%); applies when the loopback is enabled.",
state.name
));
return Ok(());
};
if loopback::set_loopback_volume(graph, id, pct).is_ok() {
ui::ok(&format!("Loopback volume for '{}' set to {pct}%.", state.name));
} else {
ui::warn(&format!("could not find the loopback playback stream (module {id})."));
}
Ok(())
}
fn edit_source_volume(state: &mut VmicState, value: f32) -> Result<()> {
let pct = to_pct(value).ok_or_else(|| VmicError::InvalidVolume(value.to_string()))?;
state.mic_volume_pct = Some(pct);
let Some(mic) = state.mic_source.clone() else {
ui::ok(&format!("Source volume for '{}' saved ({pct}%).", state.name));
return Ok(());
};
let status = std::process::Command::new("pactl")
.args(["set-source-volume", &mic, &format!("{pct}%")])
.status()?;
if status.success() {
ui::ok(&format!("Source volume for '{}' set to {pct}% ('{mic}').", state.name));
} else {
ui::warn(&format!("could not set volume on '{mic}'."));
}
Ok(())
}

62
src/commands/list.rs Normal file
View File

@@ -0,0 +1,62 @@
use crate::error::Result;
use crate::pw::PwGraph;
use crate::state::{self, VmicState};
use crate::ui;
pub fn run() -> Result<()> {
let conn = state::open_db()?;
let states = VmicState::list_all(&conn)?;
if states.is_empty() {
ui::info("No virtual mics created.");
return Ok(());
}
let graph = PwGraph::connect().ok();
for (i, state) in states.iter().enumerate() {
let status = liveness(graph.as_ref(), state);
let status_colored = if status == "active" { ui::green(&status) } else { ui::yellow(&status) };
println!("{}", ui::blue(&state.name));
println!(" output device: {}", state.sink_name);
println!(" mic device: {}", state.source_name);
println!(" status: {status_colored}");
println!(
" self-monitor loopback: {}",
state
.loopback_id
.map(|id| format!("yes (module {id})"))
.unwrap_or_else(|| "no".into())
);
if let Some(pct) = state.volume_pct {
println!(" loopback volume: {}", ui::blue(&format!("{pct}%")));
}
if let Some(mic) = &state.mic_source {
println!(" mixed source: {mic}");
if let Some(pct) = state.mic_volume_pct {
println!(" source volume: {}", ui::blue(&format!("{pct}%")));
}
}
if i + 1 < states.len() {
println!();
}
}
Ok(())
}
/// Checks `sink_node_id`/`mix_node_id` against the live graph snapshot.
fn liveness(graph: Option<&PwGraph>, state: &VmicState) -> String {
let Some(graph) = graph else {
return "unknown (could not connect to PipeWire)".to_string();
};
let nodes = graph.nodes();
let sink_alive = state.sink_node_id.is_some_and(|id| nodes.iter().any(|n| n.id == id));
let mix_alive = state.mix_node_id.is_some_and(|id| nodes.iter().any(|n| n.id == id));
match (sink_alive, mix_alive) {
(true, true) => "active".to_string(),
(true, false) => "degraded (mix loopback process not found)".to_string(),
(false, true) => "degraded (sink loopback process not found)".to_string(),
(false, false) => "stale (matching pw-loopback process not found)".to_string(),
}
}

35
src/commands/mod.rs Normal file
View File

@@ -0,0 +1,35 @@
pub mod create;
pub mod delete;
pub mod edit;
pub mod list;
pub mod route;
pub mod wipe;
use crate::error::Result;
use crate::pw::{loopback, PwGraph};
use crate::state::VmicState;
use crate::ui;
/// Normalizes a fraction (0.8) or percentage (80) to an integer percent,
/// clamped to the storable range (0..=255 - `volume_pct`/`mic_volume_pct`
/// are `u8`). Warns rather than silently dropping precision when clamped.
pub fn to_pct(value: f32) -> Option<u8> {
if !value.is_finite() || value < 0.0 {
return None;
}
let pct = if value <= 1.0 { value * 100.0 } else { value };
if pct > 255.0 {
ui::warn(&format!("volume {}% exceeds the maximum of 255%; clamped.", pct.round()));
}
Some(pct.round().clamp(0.0, u8::MAX as f32) as u8)
}
/// Unlinks a mixed-in hardware source and clears `state.mic_source`.
/// Shared by `route -s off` and `delete`.
pub(crate) fn unsource(graph: &PwGraph, state: &mut VmicState) -> Result<()> {
let Some(mic) = state.mic_source.take() else {
return Ok(());
};
loopback::unlink_stage(graph, &mic, &state.mix_name)?;
Ok(())
}

102
src/commands/route.rs Normal file
View File

@@ -0,0 +1,102 @@
use crate::cli::RouteArgs;
use crate::error::{Result, VmicError};
use crate::pw::{loopback, text, PwGraph};
use crate::state::{self, VmicState};
use crate::ui;
use std::time::Duration;
/// Shorter than stage-creation's timeout since the source already exists -
/// this is just retrying while its ports enumerate.
const SOURCE_LINK_TIMEOUT: Duration = Duration::from_secs(2);
pub fn run(args: RouteArgs) -> Result<()> {
let conn = state::open_db()?;
let name = state::normalize(&args.name);
let mut state = VmicState::load(&conn, &name)?;
if args.inputs.is_empty() && args.outputs.is_empty() && args.source.is_none() {
return Err(VmicError::NothingToDo(
"pass --input, --output and/or --source.".into(),
));
}
let graph = PwGraph::connect()?;
for filter in &args.inputs {
move_stream("sink-inputs", "move-sink-input", &state.sink_name, filter)?;
}
for filter in &args.outputs {
move_stream("source-outputs", "move-source-output", &state.source_name, filter)?;
}
if let Some(source) = &args.source {
route_source(&graph, &mut state, source)?;
state.save(&conn)?;
}
Ok(())
}
/// Matches streams via `pw::text` (name/description matching - see that
/// module for why this stays text-based) then moves each with `pactl`.
fn move_stream(kind: &str, move_cmd: &str, target: &str, filter: &str) -> Result<()> {
let matches = text::match_streams(kind, filter)?;
if matches.is_empty() {
ui::warn(&format!("'{filter}': matched 0 streams, nothing moved."));
return Ok(());
}
for m in &matches {
std::process::Command::new("pactl")
.args([move_cmd, &m.id, target])
.status()?;
}
ui::info(&format!("'{filter}': matched {} stream(s) -> moved to '{target}'.", matches.len()));
Ok(())
}
fn route_source(graph: &PwGraph, state: &mut VmicState, value: &str) -> Result<()> {
let lowered = value.to_lowercase();
if matches!(lowered.as_str(), "off" | "none" | "disable") {
let Some(mic) = state.mic_source.clone() else {
ui::warn(&format!("no source is mixed into '{}'.", state.name));
return Ok(());
};
super::unsource(graph, state)?;
ui::ok(&format!("Removed source mix '{mic}' from '{}'.", state.name));
return Ok(());
}
let matches = text::match_source(value)?;
match matches.len() {
0 => return Err(VmicError::NoMatchingSource(value.to_string())),
1 => {}
_ => {
let names: Vec<_> = matches.iter().map(|m| m.name.clone()).collect();
return Err(VmicError::AmbiguousSource {
value: value.to_string(),
matches: names.join(", "),
});
}
}
let mic = matches[0].name.clone();
// Replace a different currently-mixed source first.
if state.mic_source.as_deref().is_some_and(|old| old != mic) {
super::unsource(graph, state)?;
}
let linked = loopback::link_stage(graph, &mic, &state.mix_name, SOURCE_LINK_TIMEOUT)?;
state.mic_source = Some(mic.clone());
// Insurance: make sure the new links did not create a monitor ring.
let prefix = format!("vmic_{}_", state.name.to_lowercase());
let rings = graph.break_feedback_rings(&prefix)?;
if rings > 0 {
ui::warn(&format!("removed {rings} unexpected monitor feedback link(s)."));
}
ui::ok(&format!(
"Mixed '{mic}' into the mix stage of '{}' ({linked} link(s); not audible in self-monitor).",
state.name
));
Ok(())
}

103
src/commands/wipe.rs Normal file
View File

@@ -0,0 +1,103 @@
use crate::error::Result;
use crate::pw::{loopback, text, PwGraph};
use crate::state;
use crate::ui;
use std::process::Command;
pub fn run() -> Result<()> {
let graph = PwGraph::connect()?;
let mods_removed = wipe_loopback_modules(&graph)?;
let links_removed = graph.unlink_all_into_prefix("vmic_")?;
let procs_killed = wipe_pw_loopback_processes();
let conn = state::open_db()?;
let rows_removed = conn.execute("DELETE FROM vmics", [])? as u32;
restore_defaults();
if mods_removed + links_removed + procs_killed + rows_removed == 0 {
ui::info("Nothing to wipe: no virtual mics or vmic loopbacks found.");
} else {
ui::ok("Wiped all virtual mics:");
println!(" loopback modules unloaded: {mods_removed}");
println!(" links removed: {links_removed}");
println!(" pw-loopback processes killed: {procs_killed}");
println!(" state rows removed: {rows_removed}");
}
Ok(())
}
/// Unloads every loopback module referencing a vmic_* node, tracked or not.
fn wipe_loopback_modules(_graph: &PwGraph) -> Result<u32> {
let ids = text::list_vmic_loopback_module_ids()?;
let mut removed = 0;
for id in ids {
if loopback::unload_pulse_module(id).is_ok() {
removed += 1;
}
}
Ok(removed)
}
/// Kills every pw-loopback process whose cmdline mentions a vmic_* node,
/// tracked or not. SIGTERM only - no SIGKILL escalation, unlike
/// `terminate_stage`.
fn wipe_pw_loopback_processes() -> u32 {
let mut killed = 0;
let Ok(entries) = std::fs::read_dir("/proc") else {
return 0;
};
for entry in entries.flatten() {
let Ok(pid) = entry.file_name().to_string_lossy().parse::<libc::pid_t>() else {
continue;
};
let comm = std::fs::read_to_string(format!("/proc/{pid}/comm")).unwrap_or_default();
if comm.trim() != "pw-loopback" {
continue;
}
let Ok(cmdline) = std::fs::read(format!("/proc/{pid}/cmdline")) else {
continue;
};
if String::from_utf8_lossy(&cmdline).contains("vmic_")
&& unsafe { libc::kill(pid, libc::SIGTERM) } == 0
{
killed += 1;
}
}
killed
}
/// If the default sink/source pointed at a vmic node that just got wiped,
/// point it back at the first non-vmic device instead of leaving it
/// dangling.
fn restore_defaults() {
if default_output("get-default-sink").is_some_and(|s| s.starts_with("vmic_")) {
if let Some(new_sink) = first_non_vmic("sinks", false) {
let _ = Command::new("pactl").args(["set-default-sink", &new_sink]).status();
ui::info(&format!("Default sink restored to '{new_sink}'."));
}
}
if default_output("get-default-source").is_some_and(|s| s.starts_with("vmic_")) {
if let Some(new_source) = first_non_vmic("sources", true) {
let _ = Command::new("pactl").args(["set-default-source", &new_source]).status();
ui::info(&format!("Default source restored to '{new_source}'."));
}
}
}
fn default_output(subcommand: &str) -> Option<String> {
let out = Command::new("pactl").arg(subcommand).output().ok()?;
let s = String::from_utf8_lossy(&out.stdout).trim().to_string();
(!s.is_empty()).then_some(s)
}
fn first_non_vmic(kind: &str, exclude_monitors: bool) -> Option<String> {
let out = Command::new("pactl").args(["list", "short", kind]).output().ok()?;
String::from_utf8_lossy(&out.stdout)
.lines()
.filter_map(|l| l.split_whitespace().nth(1))
.find(|name| !name.starts_with("vmic_") && !(exclude_monitors && name.ends_with(".monitor")))
.map(str::to_string)
}

45
src/error.rs Normal file
View File

@@ -0,0 +1,45 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum VmicError {
#[error(
"invalid name '{0}' (use 1-64 chars: letters, digits, '_' or '-'; \
must start with a letter or digit)"
)]
InvalidName(String),
#[error("no vmic named '{0}'")]
NotFound(String),
#[error("vmic '{0}' already exists")]
AlreadyExists(String),
#[error("default output is the vmic sink itself; self-monitor would create a feedback loop")]
WouldFeedback,
#[error("could not resolve ports for '{from}' -> '{to}'")]
PortResolution { from: String, to: String },
#[error("-s '{0}': no matching source (monitors and vmic nodes are excluded)")]
NoMatchingSource(String),
#[error("-s '{value}': ambiguous, matches: {matches}")]
AmbiguousSource { value: String, matches: String },
#[error("invalid volume '{0}' (expected 0..1 or a percentage, e.g. 0.8 or 80)")]
InvalidVolume(String),
#[error("nothing to do: {0}")]
NothingToDo(String),
#[error("PipeWire graph error: {0}")]
PipeWire(String),
#[error(transparent)]
Io(#[from] std::io::Error),
#[error(transparent)]
Sqlite(#[from] rusqlite::Error),
}
pub type Result<T> = std::result::Result<T, VmicError>;

194
src/main.rs Normal file
View File

@@ -0,0 +1,194 @@
mod cli;
mod commands;
mod error;
mod pw;
mod state;
mod ui;
use clap::{CommandFactory, Parser};
use clap_complete::{generate, Shell};
use cli::{Cli, Commands};
fn main() {
let args = normalize_args(std::env::args().collect());
// Plain `vmic`, `-h`/`--help`, or `help` at the top level: show every
// subcommand's own flags inline instead of making the user drill into
// each one with its own `--help`.
if wants_top_level_help(&args) {
print_full_help();
std::process::exit(if args.len() <= 1 { 2 } else { 0 });
}
let cli = match Cli::try_parse_from(&args) {
Ok(cli) => cli,
Err(e) => e.exit(),
};
let result = match cli.command {
Commands::Create(args) => commands::create::run(args),
Commands::Route(args) => commands::route::run(args),
Commands::Edit(args) => commands::edit::run(args),
Commands::Delete(args) => commands::delete::run(args),
Commands::List => commands::list::run(),
Commands::Wipe => commands::wipe::run(),
Commands::Completions { shell } => {
print_completions(shell);
Ok(())
}
};
if let Err(e) = result {
ui::err(&e.to_string());
std::process::exit(1);
}
}
/// clap's `#[arg(short)]` only supports single-character short flags, so
/// `-sv` can't be a real short flag. Rewrite it to `--source-volume` before
/// clap ever sees it.
fn normalize_args(args: Vec<String>) -> Vec<String> {
args.into_iter()
.map(|a| if a == "-sv" { "--source-volume".to_string() } else { a })
.collect()
}
fn print_completions(shell: Shell) {
let mut cmd = Cli::command();
let name = cmd.get_name().to_string();
generate(shell, &mut cmd, name, &mut std::io::stdout());
}
/// True for a bare `vmic` invocation, or top-level `-h`/`--help`/`help` -
/// i.e. anything that should show the expanded help rather than being
/// handled (or rejected) by a specific subcommand.
fn wants_top_level_help(args: &[String]) -> bool {
match args.get(1..) {
Some([]) => true,
Some([a]) => a == "-h" || a == "--help" || a == "help",
_ => false,
}
}
/// Prints one screen of help: every subcommand's positional args and flags
/// summarized inline (e.g. `<name> [-l/--loopback]`), so nothing requires
/// drilling into a subcommand's own `--help` just to see what it takes.
fn print_full_help() {
let mut cmd = Cli::command();
cmd.build(); // resolve default value names etc. before introspecting
if let Some(about) = cmd.get_about() {
println!("{about}");
println!();
}
println!("{} vmic {} {}", ui::blue("Usage:"), ui::cyan("<COMMAND>"), ui::yellow("[ARGS]"));
println!();
println!("{}", ui::blue("Commands:"));
let rows: Vec<(String, String, String, String)> = cmd
.get_subcommands()
.filter(|s| s.get_name() != "help")
.map(|s| {
let aliases = s.get_visible_aliases().collect::<Vec<_>>().join("/");
let about = s.get_about().map(|a| a.to_string()).unwrap_or_default();
(s.get_name().to_string(), compact_args(s), about, aliases)
})
.collect();
print_table(&rows);
println!();
println!("{}", ui::blue("Options:"));
let opt_rows: Vec<(String, String)> = cmd
.get_arguments()
.filter(|a| !a.is_positional())
.filter_map(|a| Some((flag_names(a)?, a.get_help().map(|h| h.to_string()).unwrap_or_default())))
.collect();
let flag_w = opt_rows.iter().map(|(f, _)| f.len()).max().unwrap_or(0);
for (flag, help) in &opt_rows {
println!(" {flag:flag_w$} {help}");
}
}
/// `<name>` for each positional, `[-x/--long <VALUE>]` or `[-x/--long]` for
/// each flag, in one space-joined line - a compact stand-in for the
/// `[OPTIONS] <NAME>` a normal usage line would collapse this to.
///
/// Colors match the original fish script's `vmic_usage_line`: positional
/// `<...>` args in cyan, optional `[...]` flag groups (whole bracket,
/// including any value placeholder inside) in yellow.
fn compact_args(cmd: &clap::Command) -> String {
let mut parts: Vec<String> = cmd
.get_positionals()
.map(|a| ui::cyan(&format!("<{}>", a.get_id().as_str())))
.collect();
for arg in cmd.get_arguments() {
if arg.is_positional() || arg.get_id().as_str() == "help" {
continue;
}
let Some(flag) = flag_names(arg) else { continue };
let group = if matches!(arg.get_action(), clap::ArgAction::Set | clap::ArgAction::Append) {
let value = arg
.get_value_names()
.and_then(|v| v.first())
.map(|v| v.to_string())
.unwrap_or_else(|| arg.get_id().as_str().to_uppercase());
format!("[{flag} <{value}>]")
} else {
format!("[{flag}]")
};
parts.push(ui::yellow(&group));
}
parts.join(" ")
}
/// `-x, --long` / `-x` / `--long` for a non-positional arg, or `None` for
/// one with no visible flag at all.
fn flag_names(arg: &clap::Arg) -> Option<String> {
match (arg.get_short(), arg.get_long()) {
(Some(s), Some(l)) => Some(format!("-{s}/--{l}")),
(Some(s), None) => Some(format!("-{s}")),
(None, Some(l)) => Some(format!("--{l}")),
(None, None) => None,
}
}
/// Prints `(name, args, about, aliases)` rows with the name and args
/// columns padded to their widest entry, so the `about` column lines up.
/// `args` carries ANSI color codes (see `compact_args`), so padding by
/// `.len()`/`{:width$}` would count invisible escape bytes as columns and
/// misalign everything - `pad_visual` pads by visible width instead.
/// `aliases`, if non-empty, trails after `| `.
fn print_table(rows: &[(String, String, String, String)]) {
let name_w = rows.iter().map(|(n, ..)| n.len()).max().unwrap_or(0);
let args_w = rows.iter().map(|(_, a, ..)| visual_width(a)).max().unwrap_or(0);
for (name, args, about, aliases) in rows {
print!(" {name:name_w$} {} {about}", pad_visual(args, args_w));
if !aliases.is_empty() {
print!(" | {aliases}");
}
println!();
}
}
/// Number of visible columns in `s`, skipping any `\x1b[...m` ANSI SGR
/// escape sequences it contains.
fn visual_width(s: &str) -> usize {
let mut width = 0;
let mut in_escape = false;
for c in s.chars() {
if in_escape {
in_escape = c != 'm';
} else if c == '\x1b' {
in_escape = true;
} else {
width += 1;
}
}
width
}
/// Right-pads `s` with spaces to `width` visible columns, per `visual_width`.
fn pad_visual(s: &str, width: usize) -> String {
format!("{s}{}", " ".repeat(width.saturating_sub(visual_width(s))))
}

342
src/pw/graph.rs Normal file
View File

@@ -0,0 +1,342 @@
use crate::error::{Result, VmicError};
use pipewire as pw;
use pw::types::ObjectType;
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::rc::Rc;
use std::time::{Duration, Instant};
/// A node in the live PipeWire graph, as seen via the registry.
#[derive(Debug, Clone)]
pub struct Node {
pub id: u32,
pub name: String, // always one of this program's own ASCII vmic_* names
#[allow(dead_code)]
pub media_class: String, // "Audio/Sink", "Audio/Source", ...
}
/// A single input/output port on a node.
#[derive(Debug, Clone)]
pub struct Port {
pub id: u32,
pub node_id: u32,
pub name: String, // e.g. "capture_FL", "playback_FL", "monitor_FL"
pub is_input: bool,
}
/// A live link between two ports.
#[derive(Debug, Clone)]
pub struct Link {
pub id: u32,
pub output_port: u32,
pub input_port: u32,
}
/// The standard core-registered factory that creates `PipeWire:Interface:Link`
/// objects. Confirmed via `pw-dump` (`factory.name = "link-factory"`) - the
/// same name the real `pw-link` CLI hardcodes, not something that varies
/// per system.
const LINK_FACTORY: &str = "link-factory";
const ROUNDTRIP_TIMEOUT: Duration = Duration::from_secs(5);
/// Bound on one polling tick while waiting for a freshly spawned node/port
/// to appear. Short and retried, unlike `ROUNDTRIP_TIMEOUT`, which is a
/// hard failure.
const POLL_TICK: Duration = Duration::from_millis(200);
#[derive(Default)]
struct GraphState {
nodes: HashMap<u32, Node>,
ports: HashMap<u32, Port>,
links: HashMap<u32, Link>,
}
/// Handle to a running PipeWire main loop + registry snapshot, used only for
/// operations on this program's own nodes (known ASCII names, numeric ids).
/// Anything involving third-party device names/descriptions goes through
/// `pw::text` instead.
///
/// Field order matters: Rust drops struct fields in declaration order, and
/// `_listener`/`registry` must be torn down before `core`, which must be
/// torn down before `_context`, which must be torn down before `mainloop`.
pub struct PwGraph {
_listener: pw::registry::Listener,
registry: pw::registry::Registry,
core: pw::core::Core,
_context: pw::context::Context,
mainloop: pw::main_loop::MainLoop,
state: Rc<RefCell<GraphState>>,
}
impl PwGraph {
pub fn connect() -> Result<Self> {
let mainloop = pw::main_loop::MainLoop::new(None)
.map_err(|e| VmicError::PipeWire(format!("main loop: {e}")))?;
let context = pw::context::Context::new(&mainloop)
.map_err(|e| VmicError::PipeWire(format!("context: {e}")))?;
let core = context
.connect(None)
.map_err(|e| VmicError::PipeWire(format!("connect: {e} (is PipeWire running?)")))?;
let registry = core
.get_registry()
.map_err(|e| VmicError::PipeWire(format!("get registry: {e}")))?;
let state = Rc::new(RefCell::new(GraphState::default()));
let state_add = state.clone();
let state_rm = state.clone();
let listener = registry
.add_listener_local()
.global(move |g| {
let mut s = state_add.borrow_mut();
match &g.type_ {
ObjectType::Node => {
let name = g
.props
.and_then(|p| p.get(*pw::keys::NODE_NAME))
.unwrap_or_default()
.to_string();
let media_class = g
.props
.and_then(|p| p.get(*pw::keys::MEDIA_CLASS))
.unwrap_or_default()
.to_string();
s.nodes.insert(g.id, Node { id: g.id, name, media_class });
}
ObjectType::Port => {
let node_id = g
.props
.and_then(|p| p.get(*pw::keys::NODE_ID))
.and_then(|v| v.parse().ok())
.unwrap_or(0);
let name = g
.props
.and_then(|p| p.get(*pw::keys::PORT_NAME))
.unwrap_or_default()
.to_string();
let is_input =
g.props.and_then(|p| p.get(*pw::keys::PORT_DIRECTION)) == Some("in");
s.ports.insert(g.id, Port { id: g.id, node_id, name, is_input });
}
ObjectType::Link => {
let output_port = g
.props
.and_then(|p| p.get(*pw::keys::LINK_OUTPUT_PORT))
.and_then(|v| v.parse().ok())
.unwrap_or(0);
let input_port = g
.props
.and_then(|p| p.get(*pw::keys::LINK_INPUT_PORT))
.and_then(|v| v.parse().ok())
.unwrap_or(0);
s.links.insert(g.id, Link { id: g.id, output_port, input_port });
}
_ => {}
}
})
.global_remove(move |id| {
let mut s = state_rm.borrow_mut();
s.nodes.remove(&id);
s.ports.remove(&id);
s.links.remove(&id);
})
.register();
do_roundtrip(&mainloop, &core, ROUNDTRIP_TIMEOUT)?;
Ok(Self { _listener: listener, registry, core, _context: context, mainloop, state })
}
fn roundtrip(&self) -> Result<()> {
do_roundtrip(&self.mainloop, &self.core, ROUNDTRIP_TIMEOUT)
}
pub(crate) fn poll_tick(&self) {
let _ = do_roundtrip(&self.mainloop, &self.core, POLL_TICK);
}
pub fn nodes(&self) -> Vec<Node> {
self.state.borrow().nodes.values().cloned().collect()
}
pub fn ports_for(&self, node_name: &str) -> Vec<Port> {
let s = self.state.borrow();
let Some(node) = s.nodes.values().find(|n| n.name == node_name) else {
return Vec::new();
};
s.ports.values().filter(|p| p.node_id == node.id).cloned().collect()
}
pub fn links(&self) -> Vec<Link> {
self.state.borrow().links.values().cloned().collect()
}
/// Links two ports if not already linked (idempotent).
pub fn link_ports(&self, output_port: u32, input_port: u32) -> Result<()> {
if self.is_linked(output_port, input_port) {
return Ok(());
}
let props = pw::properties::properties! {
*pw::keys::LINK_OUTPUT_PORT => output_port.to_string(),
*pw::keys::LINK_INPUT_PORT => input_port.to_string(),
// Don't destroy the link on the remote when our local proxy for
// it is dropped at the end of this function - links should
// outlive this one-shot CLI invocation.
"object.linger" => "1",
};
if let Ok(link) = self.core.create_object::<pw::link::Link>(LINK_FACTORY, &props) {
let _ = self.roundtrip();
drop(link);
}
// Defensive re-check: a create_object error doesn't necessarily mean
// no link exists.
if self.is_linked(output_port, input_port) {
Ok(())
} else {
Err(VmicError::PipeWire(format!("failed to link ports {output_port} -> {input_port}")))
}
}
pub fn unlink_ports(&self, output_port: u32, input_port: u32) -> Result<()> {
let id = self
.links()
.into_iter()
.find(|l| l.output_port == output_port && l.input_port == input_port)
.map(|l| l.id);
let Some(id) = id else {
return Ok(()); // already gone
};
self.registry
.destroy_global(id)
.into_result()
.map_err(|e| VmicError::PipeWire(format!("unlink {output_port}->{input_port}: {e}")))?;
self.roundtrip()
}
fn is_linked(&self, output_port: u32, input_port: u32) -> bool {
self.links()
.iter()
.any(|l| l.output_port == output_port && l.input_port == input_port)
}
/// Collects (output_port, input_port) pairs for links whose endpoint
/// node/port names satisfy `pred(out_node, out_port, in_node, in_port)`.
fn matching_link_ports(
&self,
pred: impl Fn(&str, &str, &str, &str) -> bool,
) -> Vec<(u32, u32)> {
let s = self.state.borrow();
let mut out = Vec::new();
for link in s.links.values() {
let (Some(out_port), Some(in_port)) =
(s.ports.get(&link.output_port), s.ports.get(&link.input_port))
else {
continue;
};
let (Some(out_node), Some(in_node)) =
(s.nodes.get(&out_port.node_id), s.nodes.get(&in_port.node_id))
else {
continue;
};
if pred(&out_node.name, &out_port.name, &in_node.name, &in_port.name) {
out.push((link.output_port, link.input_port));
}
}
out
}
fn unlink_all(&self, pairs: Vec<(u32, u32)>) -> Result<u32> {
let mut removed = 0;
for (o, i) in pairs {
if self.unlink_ports(o, i).is_ok() {
removed += 1;
}
}
Ok(removed)
}
/// Destroys any link from a vmic sink's monitor back into a vmic node
/// (a feedback ring). Returns the number removed.
pub fn break_feedback_rings(&self, prefix: &str) -> Result<u32> {
let sink_name = format!("{prefix}sink");
let pairs = self.matching_link_ports(|out_node, out_port, in_node, _in_port| {
out_node == sink_name && out_port.starts_with("monitor") && in_node.starts_with(prefix)
});
self.unlink_all(pairs)
}
/// Destroys every link whose input lands on a node under `prefix`,
/// regardless of the output side - broader than `break_feedback_rings`,
/// used when tearing down everything.
pub fn unlink_all_into_prefix(&self, prefix: &str) -> Result<u32> {
let pairs = self.matching_link_ports(|_out_node, _out_port, in_node, _in_port| {
in_node.starts_with(prefix)
});
self.unlink_all(pairs)
}
/// Waits (bounded) for a node to appear in the graph.
pub fn wait_for_node(&self, name: &str, timeout: Duration) -> Result<Node> {
let deadline = Instant::now() + timeout;
loop {
if let Some(n) = self.nodes().into_iter().find(|n| n.name == name) {
return Ok(n);
}
if Instant::now() >= deadline {
return Err(VmicError::PipeWire(format!(
"node '{name}' did not appear within {timeout:?}"
)));
}
self.poll_tick();
}
}
}
/// Triggers `core.sync` and pumps the main loop until the server
/// acknowledges it (delivering any pending `global`/`global_remove` events
/// along the way), or until `timeout` elapses (a watchdog, so a dead
/// daemon can't hang the CLI).
fn do_roundtrip(mainloop: &pw::main_loop::MainLoop, core: &pw::core::Core, timeout: Duration) -> Result<()> {
let done = Rc::new(Cell::new(false));
let done_clone = done.clone();
let loop_done = mainloop.clone();
let pending = core
.sync(0)
.map_err(|e| VmicError::PipeWire(format!("sync: {e}")))?;
let _core_listener = core
.add_listener_local()
.done(move |id, seq| {
if id == pw::core::PW_ID_CORE && seq == pending {
done_clone.set(true);
loop_done.quit();
}
})
.register();
let timed_out = Rc::new(Cell::new(false));
let timed_out_clone = timed_out.clone();
let loop_timeout = mainloop.clone();
let timer = mainloop.loop_().add_timer(move |_expirations| {
timed_out_clone.set(true);
loop_timeout.quit();
});
timer
.update_timer(Some(timeout), None)
.into_result()
.map_err(|e| VmicError::PipeWire(format!("arm watchdog timer: {e}")))?;
while !done.get() && !timed_out.get() {
mainloop.run();
}
if !done.get() {
return Err(VmicError::PipeWire("timed out waiting for the PipeWire daemon".into()));
}
Ok(())
}

326
src/pw/loopback.rs Normal file
View File

@@ -0,0 +1,326 @@
use crate::error::{Result, VmicError};
use crate::pw::graph::{Port, PwGraph};
use crate::pw::text;
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
pub const LATENCY_MS: u32 = 32;
pub const SAMPLE_RATE: u32 = 48_000;
const NODE_WAIT_TIMEOUT: Duration = Duration::from_secs(3);
const LINK_RETRY_TIMEOUT: Duration = Duration::from_secs(3);
const STAGE_STARTUP_CHECK: Duration = Duration::from_millis(500);
/// Node names derived for a vmic. Always ASCII, since they come from a name
/// that already passed `state::valid_name`.
pub struct NodeNames {
pub sink: String, // stage 1 capture (apps play into this)
pub mid: String, // stage 1 playback (internal)
pub mix: String, // stage 2 capture (internal, hw mic joins here)
pub source: String, // stage 2 playback (recorders read from this)
pub sink_desc: String,
pub mid_desc: String,
pub mix_desc: String,
pub source_desc: String,
}
impl NodeNames {
pub fn for_vmic(name: &str) -> Self {
let lower = name.to_lowercase();
let upper = name.to_uppercase();
Self {
sink: format!("vmic_{lower}_sink"),
mid: format!("vmic_{lower}_mid"),
mix: format!("vmic_{lower}_mix"),
source: format!("vmic_{lower}_mic"),
sink_desc: format!("VIRTUAL_MIC_{upper}_OUTPUT"),
mid_desc: format!("VIRTUAL_MIC_{upper}_MID_INTERNAL"),
mix_desc: format!("VIRTUAL_MIC_{upper}_MIX_INTERNAL"),
source_desc: format!("VIRTUAL_MIC_{upper}_INPUT"),
}
}
}
/// OS pids and PipeWire node ids produced by [`create_stages`], persisted in
/// `VmicState` so later commands can terminate the processes and check
/// liveness against the live graph.
pub struct StageHandles {
pub sink_pid: u32,
pub mix_pid: u32,
pub sink_node_id: u32,
pub mix_node_id: u32,
}
/// Creates the two chained loopback stages as detached `pw-loopback`
/// subprocesses (stdin is `/dev/null`, never waited on, so the OS
/// reparents them to init on exit), then links stage 1 into stage 2
/// (mid -> mix).
///
/// apps -> sink =(stage1)=> mid -> mix =(stage2)=> source -> recorders
/// ^ hw mic joins here (route --source)
pub fn create_stages(graph: &PwGraph, names: &NodeNames) -> Result<StageHandles> {
let fmt = format!("audio.rate={SAMPLE_RATE} audio.position=[FL FR]");
let log_path = std::env::temp_dir().join(format!(
"vmic_{}.log",
names.sink.trim_start_matches("vmic_").trim_end_matches("_sink")
));
let log_file = std::fs::OpenOptions::new().create(true).append(true).open(&log_path)?;
let mut sink_child = spawn_loopback_stage(
&format!(
"media.class=Audio/Sink node.name={} node.description={} node.virtual=false {fmt}",
names.sink, names.sink_desc
),
&format!(
"media.class=Audio/Source node.name={} node.description={} node.virtual=false {fmt}",
names.mid, names.mid_desc
),
&log_file,
)?;
let mut mix_child = spawn_loopback_stage(
&format!(
"media.class=Audio/Sink node.name={} node.description={} node.virtual=false {fmt}",
names.mix, names.mix_desc
),
&format!(
"media.class=Audio/Source node.name={} node.description={} node.virtual=false {fmt}",
names.source, names.source_desc
),
&log_file,
)?;
// Give both processes a moment to fail fast (e.g. a name collision)
// before trusting them.
std::thread::sleep(STAGE_STARTUP_CHECK);
if sink_child.try_wait()?.is_some() || mix_child.try_wait()?.is_some() {
let _ = sink_child.kill();
let _ = mix_child.kill();
return Err(VmicError::PipeWire(format!(
"pw-loopback exited immediately, check '{}'.",
log_path.display()
)));
}
let sink_pid = sink_child.id();
let mix_pid = mix_child.id();
let stage_ids = (|| -> Result<(u32, u32)> {
let sink_node = graph.wait_for_node(&names.sink, NODE_WAIT_TIMEOUT)?;
let mix_node = graph.wait_for_node(&names.mix, NODE_WAIT_TIMEOUT)?;
link_stage(graph, &names.mid, &names.mix, LINK_RETRY_TIMEOUT)?;
Ok((sink_node.id, mix_node.id))
})();
let (sink_node_id, mix_node_id) = match stage_ids {
Ok(ids) => ids,
Err(e) => {
let _ = sink_child.kill();
let _ = mix_child.kill();
return Err(e);
}
};
Ok(StageHandles { sink_pid, mix_pid, sink_node_id, mix_node_id })
}
fn spawn_loopback_stage(capture_props: &str, playback_props: &str, log_file: &std::fs::File) -> Result<Child> {
Command::new("pw-loopback")
.arg(format!("--latency={LATENCY_MS}"))
.args(["-m", "[ FL FR ]"])
.arg(format!("--capture-props={capture_props}"))
.arg(format!("--playback-props={playback_props}"))
.stdin(Stdio::null())
.stdout(log_file.try_clone()?)
.stderr(log_file.try_clone()?)
.spawn()
.map_err(VmicError::from)
}
/// Resolves and links `from_node`'s output ports into `target_node`'s input
/// ports, retrying briefly while ports may still be appearing. Shared by
/// stage creation and `route -s`.
pub fn link_stage(graph: &PwGraph, from_node: &str, target_node: &str, timeout: Duration) -> Result<u32> {
let deadline = Instant::now() + timeout;
loop {
let outs: Vec<Port> = graph.ports_for(from_node).into_iter().filter(|p| !p.is_input).collect();
let ins: Vec<Port> = graph.ports_for(target_node).into_iter().filter(|p| p.is_input).collect();
if let Some(pairs) = resolve_link_pairs(&outs, &ins) {
let linked = pairs.iter().filter(|(o, i)| graph.link_ports(*o, *i).is_ok()).count() as u32;
if linked > 0 {
return Ok(linked);
}
}
if Instant::now() >= deadline {
return Err(VmicError::PortResolution {
from: from_node.to_string(),
to: target_node.to_string(),
});
}
graph.poll_tick();
}
}
/// Resolves the same port pairs `link_stage` would use and unlinks them,
/// best-effort. Used to tear down a mixed-in hardware source.
pub fn unlink_stage(graph: &PwGraph, from_node: &str, target_node: &str) -> Result<u32> {
let outs: Vec<Port> = graph.ports_for(from_node).into_iter().filter(|p| !p.is_input).collect();
let ins: Vec<Port> = graph.ports_for(target_node).into_iter().filter(|p| p.is_input).collect();
let Some(pairs) = resolve_link_pairs(&outs, &ins) else {
return Ok(0);
};
let mut removed = 0;
for (o, i) in pairs {
if graph.unlink_ports(o, i).is_ok() {
removed += 1;
}
}
Ok(removed)
}
fn find_port<'a>(ports: &'a [Port], name: &str) -> Option<&'a Port> {
ports.iter().find(|p| p.name == name)
}
/// Resolves link pairs for feeding `outs` into `ins`. A mono source is
/// linked into both input channels.
fn resolve_link_pairs(outs: &[Port], ins: &[Port]) -> Option<Vec<(u32, u32)>> {
let mut fl = find_port(outs, "capture_FL");
let mut fr = find_port(outs, "capture_FR");
// Sinks name their input ports playback_*; some name them input_*.
let mut ifl = find_port(ins, "input_FL").or_else(|| find_port(ins, "playback_FL"));
let mut ifr = find_port(ins, "input_FR").or_else(|| find_port(ins, "playback_FR"));
// Fall back to first/second port for exotic port names.
if fl.is_none() {
fl = outs.first();
}
if fr.is_none() {
fr = outs.get(1);
}
if ifl.is_none() {
ifl = ins.first();
}
if ifr.is_none() {
ifr = ins.get(1);
}
let fl = fl?;
let ifl = ifl?;
let mut pairs = vec![(fl.id, ifl.id)];
match (fr, ifr) {
(Some(fr), Some(ifr)) => pairs.push((fr.id, ifr.id)),
(None, Some(ifr)) => pairs.push((fl.id, ifr.id)), // mono source: feed both channels
_ => {}
}
Some(pairs)
}
/// SIGTERM a verified pw-loopback pid, escalating to SIGKILL if it lingers.
/// `expected_name_fragment` must appear in the process's cmdline first, as
/// insurance against a stale/reused pid.
pub fn terminate_stage(pid: i64, expected_name_fragment: &str) -> Result<()> {
let pid = pid as libc::pid_t;
if !pid_matches(pid, expected_name_fragment) {
return Ok(());
}
unsafe { libc::kill(pid, libc::SIGTERM) };
for _ in 0..10 {
if !process_alive(pid) {
return Ok(());
}
std::thread::sleep(Duration::from_millis(100));
}
if process_alive(pid) && pid_matches(pid, expected_name_fragment) {
unsafe { libc::kill(pid, libc::SIGKILL) };
}
Ok(())
}
fn process_alive(pid: libc::pid_t) -> bool {
unsafe { libc::kill(pid, 0) == 0 }
}
/// True only if `pid` is a live `pw-loopback` process whose cmdline
/// mentions `expected_name_fragment`.
fn pid_matches(pid: libc::pid_t, expected_name_fragment: &str) -> bool {
let Ok(cmdline) = std::fs::read(format!("/proc/{pid}/cmdline")) else {
return false;
};
let text = String::from_utf8_lossy(&cmdline);
text.contains("pw-loopback") && text.contains(expected_name_fragment)
}
/// Unloads a pactl-hosted module (the self-monitor loopback).
pub fn unload_pulse_module(module_id: u32) -> Result<()> {
let status = Command::new("pactl")
.args(["unload-module", &module_id.to_string()])
.status()?;
if !status.success() {
return Err(VmicError::PipeWire(format!("failed to unload module {module_id}")));
}
Ok(())
}
/// Loads a self-monitor loopback: sink.monitor -> default sink. `graph` is
/// unused (module loading goes through `pactl`, see `pw/mod.rs`) but kept
/// so the signature matches its call sites.
pub fn enable_self_monitor(_graph: &PwGraph, sink_name: &str) -> Result<u32> {
let default_sink = String::from_utf8_lossy(&Command::new("pactl").arg("get-default-sink").output()?.stdout)
.trim()
.to_string();
if default_sink == sink_name {
return Err(VmicError::WouldFeedback);
}
let output = Command::new("pactl")
.args([
"load-module",
"module-loopback",
&format!("source={sink_name}.monitor"),
"sink=@DEFAULT_SINK@",
])
.output()?;
if !output.status.success() {
return Err(VmicError::PipeWire("failed to create self-monitor loopback".into()));
}
String::from_utf8_lossy(&output.stdout)
.trim()
.parse()
.map_err(|_| VmicError::PipeWire("failed to create self-monitor loopback".into()))
}
/// Sets the volume of the self-monitor loopback's playback stream,
/// retrying briefly while it may still be appearing.
pub fn set_loopback_volume(_graph: &PwGraph, loopback_id: u32, pct: u8) -> Result<()> {
let mut sink_input_id = None;
for _ in 0..5 {
if let Some(found) = text::find_sink_input_for_module(loopback_id)? {
sink_input_id = Some(found);
break;
}
std::thread::sleep(Duration::from_millis(200));
}
let Some(sink_input_id) = sink_input_id else {
return Err(VmicError::PipeWire("loopback playback stream not found".into()));
};
let status = Command::new("pactl")
.args(["set-sink-input-volume", &sink_input_id.to_string(), &format!("{pct}%")])
.status()?;
if !status.success() {
return Err(VmicError::PipeWire("pactl set-sink-input-volume failed".into()));
}
Ok(())
}

28
src/pw/mod.rs Normal file
View File

@@ -0,0 +1,28 @@
//! PipeWire control, split into three modules:
//!
//! - `graph` - live graph control (link/unlink ports, break feedback
//! rings, node/port/link introspection) via `pipewire-rs`,
//! using only numeric object ids and this program's own
//! ASCII node names.
//! - `loopback` - creates/destroys the two chained loopback stages and the
//! self-monitor loopback, built on top of `graph`.
//! - `text` - byte-safe text matching (`pactl` subprocess output) for
//! anything that has to match against human-authored,
//! possibly-multibyte device names/descriptions.
//!
//! Module *lifecycle* (spawning/loading the loopback stages and the
//! self-monitor) can't go through `graph`: `pw_context_load_module` loads a
//! module into the *calling process's own* context, which then has to keep
//! running its main loop forever for the module's nodes to keep producing
//! audio - that's what `pw-loopback` itself is. A one-shot CLI invocation
//! can't be that persistent host, so the two chained stages stay
//! `pw-loopback` subprocesses (tracked by pid), and the self-monitor stays a
//! `pactl load-module`/`unload-module` call. Everything downstream of
//! that - linking, unlinking, ring-breaking, liveness - is native
//! `pipewire-rs` via `graph`.
pub mod graph;
pub mod loopback;
pub mod text;
pub use graph::PwGraph;

256
src/pw/text.rs Normal file
View File

@@ -0,0 +1,256 @@
//! Byte-safe text scraping for anything that has to match against
//! human-authored device names/descriptions.
//!
//! PipeWire's structured JSON dump (`pw-dump --json`) has had real problems
//! serializing certain multibyte UTF-8 sequences in node/device properties -
//! a hardware mic whose description contains a "TM"-style trademark symbol
//! is a documented case that can produce JSON a strict deserializer refuses
//! to parse. This module shells out to `pactl`'s plain-text listings
//! instead, and reads output with `String::from_utf8_lossy`, so a stray or
//! unexpected byte sequence degrades to `\u{FFFD}` instead of aborting the
//! whole match or corrupting a JSON parse.
//!
//! This module owns ONLY name/description matching. Actually creating and
//! linking nodes goes through `pw::graph`'s numeric-id API once a match is
//! resolved to a concrete node name.
use crate::error::Result;
use std::process::Command;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SourceMatch {
pub name: String,
pub description: String,
}
/// Runs `pactl list sources` and parses Name/Description pairs.
pub fn list_sources() -> Result<Vec<SourceMatch>> {
let output = Command::new("pactl").args(["list", "sources"]).output()?;
Ok(parse_pactl_sources(&String::from_utf8_lossy(&output.stdout)))
}
fn parse_pactl_sources(text: &str) -> Vec<SourceMatch> {
let mut sources = Vec::new();
let mut current_name: Option<String> = None;
for line in text.lines() {
let trimmed = line.trim_start();
if let Some(rest) = trimmed.strip_prefix("Name: ") {
current_name = Some(rest.trim().to_string());
} else if let Some(rest) = trimmed.strip_prefix("Description: ") {
if let Some(name) = current_name.take() {
sources.push(SourceMatch {
name,
description: rest.trim().to_string(),
});
}
}
}
sources
}
/// Excludes monitor sources and vmic's own nodes (a vmic node routed as its
/// own hardware source would be an unguarded feedback loop - see
/// `PwGraph::break_feedback_rings`, which doesn't catch this case since it
/// only watches for rings through a sink's monitor port). Case-insensitive
/// substring match against "name description", or an explicit numeric
/// index into `pactl list short sources` (same exclusions apply there too).
pub fn match_source(filter: &str) -> Result<Vec<SourceMatch>> {
if let Ok(idx) = filter.parse::<u32>() {
let output = Command::new("pactl")
.args(["list", "short", "sources"])
.output()?;
let text = String::from_utf8_lossy(&output.stdout);
return Ok(text
.lines()
.filter_map(|l| {
let mut cols = l.split_whitespace();
let id = cols.next()?;
let name = cols.next()?;
(id == idx.to_string()).then(|| SourceMatch {
name: name.to_string(),
description: String::new(),
})
})
.filter(|s| !s.name.ends_with(".monitor") && !s.name.starts_with("vmic_"))
.collect());
}
let needle = filter.to_lowercase();
Ok(list_sources()?
.into_iter()
.filter(|s| !s.name.ends_with(".monitor") && !s.name.starts_with("vmic_"))
.filter(|s| format!("{} {}", s.name, s.description).to_lowercase().contains(&needle))
.collect())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamMatch {
pub id: String,
pub application: String,
pub media_name: String,
}
/// Matches a numeric id, "app", or "app:media" (case-insensitive
/// substring) against sink-inputs or source-outputs.
pub fn match_streams(kind: &str, filter: &str) -> Result<Vec<StreamMatch>> {
let output = Command::new("pactl").args(["list", kind]).output()?;
let text = String::from_utf8_lossy(&output.stdout);
let streams = parse_pactl_streams(&text, kind);
if let Ok(_) = filter.parse::<u32>() {
return Ok(streams.into_iter().filter(|s| s.id == filter).collect());
}
let mut parts = filter.splitn(2, ':');
let app = parts.next().unwrap_or("").to_lowercase();
let media = parts.next().unwrap_or("").to_lowercase();
Ok(streams
.into_iter()
.filter(|s| app.is_empty() || s.application.to_lowercase().contains(&app))
.filter(|s| media.is_empty() || s.media_name.to_lowercase().contains(&media))
.collect())
}
fn parse_pactl_streams(text: &str, kind: &str) -> Vec<StreamMatch> {
let header_prefix = if kind == "sink-inputs" {
"Sink Input #"
} else {
"Source Output #"
};
let mut streams = Vec::new();
let mut id = String::new();
let mut application = String::new();
let mut media_name = String::new();
let mut started = false;
let flush = |streams: &mut Vec<StreamMatch>, id: &str, application: &str, media_name: &str| {
if !id.is_empty() {
streams.push(StreamMatch {
id: id.to_string(),
application: application.to_string(),
media_name: media_name.to_string(),
});
}
};
for line in text.lines() {
if let Some(rest) = line.strip_prefix(header_prefix) {
flush(&mut streams, &id, &application, &media_name);
id = rest.trim_start_matches('#').trim().to_string();
application.clear();
media_name.clear();
started = true;
continue;
}
if !started {
continue;
}
let trimmed = line.trim_start();
if let Some(rest) = trimmed.strip_prefix("application.name = ") {
application = unquote(rest);
} else if let Some(rest) = trimmed.strip_prefix("media.name = ") {
media_name = unquote(rest);
}
}
flush(&mut streams, &id, &application, &media_name);
streams
}
fn unquote(s: &str) -> String {
s.trim_matches('"').to_string()
}
/// The sink-input (playback stream) owned by the given `module-loopback`
/// id, if it has appeared yet. Callers retry this briefly - see
/// `pw::loopback::set_loopback_volume`.
pub fn find_sink_input_for_module(module_id: u32) -> Result<Option<u32>> {
let output = Command::new("pactl").args(["list", "sink-inputs"]).output()?;
Ok(parse_sink_input_for_module(&String::from_utf8_lossy(&output.stdout), module_id))
}
fn parse_sink_input_for_module(text: &str, module_id: u32) -> Option<u32> {
let mut current_id: Option<u32> = None;
for line in text.lines() {
if let Some(rest) = line.strip_prefix("Sink Input #") {
current_id = rest.trim().parse().ok();
continue;
}
let trimmed = line.trim_start();
if let Some(rest) = trimmed.strip_prefix("Owner Module: ") {
if rest.trim().parse::<u32>().ok() == Some(module_id) {
return current_id;
}
}
}
None
}
/// Every `module-loopback` whose `Argument:` line mentions a `vmic_` node,
/// tracked in state or not.
pub fn list_vmic_loopback_module_ids() -> Result<Vec<u32>> {
let output = Command::new("pactl").args(["list", "modules"]).output()?;
Ok(parse_vmic_loopback_module_ids(&String::from_utf8_lossy(&output.stdout)))
}
fn parse_vmic_loopback_module_ids(text: &str) -> Vec<u32> {
let mut ids = Vec::new();
let mut current_id: Option<u32> = None;
let mut is_loopback = false;
for line in text.lines() {
if let Some(rest) = line.strip_prefix("Module #") {
current_id = rest.trim().parse().ok();
is_loopback = false;
continue;
}
let trimmed = line.trim_start();
if let Some(rest) = trimmed.strip_prefix("Name: ") {
is_loopback = rest.trim() == "module-loopback";
} else if let Some(rest) = trimmed.strip_prefix("Argument:") {
if is_loopback && rest.contains("vmic_") {
if let Some(id) = current_id {
ids.push(id);
}
}
}
}
ids
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn finds_sink_input_owned_by_module() {
let text = "\
Sink Input #10
\tOwner Module: n/a
Sink Input #11
\tOwner Module: 42
Sink Input #12
\tOwner Module: 7
";
assert_eq!(parse_sink_input_for_module(text, 42), Some(11));
assert_eq!(parse_sink_input_for_module(text, 99), None);
}
#[test]
fn lists_only_vmic_loopback_modules() {
let text = "\
Module #1
\tName: module-loopback
\tArgument: source=vmic_test_sink.monitor sink=@DEFAULT_SINK@
Module #2
\tName: module-loopback
\tArgument: source=alsa_input.usb-mic sink=@DEFAULT_SINK@
Module #3
\tName: libpipewire-module-rt
\tArgument: vmic_should_not_match
";
assert_eq!(parse_vmic_loopback_module_ids(text), vec![1]);
}
}

187
src/state.rs Normal file
View File

@@ -0,0 +1,187 @@
use crate::error::{Result, VmicError};
use rusqlite::{params, Connection, OptionalExtension};
use std::path::PathBuf;
/// Persisted state for one vmic, stored in `~/.config/vmic/vmic.db`.
#[derive(Debug, Clone, Default)]
pub struct VmicState {
pub name: String,
pub sink_name: String,
pub mid_name: String,
pub mix_name: String,
pub source_name: String,
// PipeWire node ids, used to check liveness against the live graph.
pub sink_node_id: Option<u32>,
pub mix_node_id: Option<u32>,
// pw-loopback pids, needed alongside the node ids: a client can't
// destroy another client's node, so teardown happens by signal instead
// (see pw::loopback::terminate_stage).
pub sink_pid: Option<i64>,
pub mix_pid: Option<i64>,
pub loopback_id: Option<u32>,
pub volume_pct: Option<u8>,
// May contain arbitrary multibyte text; SQLite TEXT has no encoding
// pitfalls here, unlike PipeWire's JSON dump (see pw::text).
pub mic_source: Option<String>,
pub mic_volume_pct: Option<u8>,
}
fn db_path() -> PathBuf {
if let Ok(dir) = std::env::var("VMIC_STATE_DIR_OVERRIDE") {
return PathBuf::from(dir).join("vmic.db");
}
dirs::config_dir()
.expect("could not resolve config dir")
.join("vmic")
.join("vmic.db")
}
pub fn valid_name(name: &str) -> bool {
let mut chars = name.chars();
let Some(first) = chars.next() else { return false };
if !first.is_ascii_alphanumeric() {
return false;
}
name.len() <= 64
&& name
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
}
pub fn require_valid_name(name: &str) -> Result<()> {
if valid_name(name) {
Ok(())
} else {
Err(VmicError::InvalidName(name.to_string()))
}
}
/// Canonical form of a vmic name, used as the state-table key and for
/// deriving PipeWire node names. Names are lowercased so e.g. "Test" and
/// "test" can't collide on the underlying (already-lowercased) node names
/// while being treated as distinct rows in the state table.
pub fn normalize(name: &str) -> String {
name.to_lowercase()
}
/// Opens (or creates) the database and ensures the schema exists.
pub fn open_db() -> Result<Connection> {
let path = db_path();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let conn = Connection::open(path)?;
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS vmics (
name TEXT PRIMARY KEY,
sink_name TEXT NOT NULL,
mid_name TEXT NOT NULL,
mix_name TEXT NOT NULL,
source_name TEXT NOT NULL,
sink_node_id INTEGER,
mix_node_id INTEGER,
sink_pid INTEGER,
mix_pid INTEGER,
loopback_id INTEGER,
volume_pct INTEGER,
mic_source TEXT,
mic_volume_pct INTEGER
);",
)?;
Ok(conn)
}
impl VmicState {
fn from_row(row: &rusqlite::Row) -> rusqlite::Result<Self> {
Ok(Self {
name: row.get("name")?,
sink_name: row.get("sink_name")?,
mid_name: row.get("mid_name")?,
mix_name: row.get("mix_name")?,
source_name: row.get("source_name")?,
sink_node_id: row.get("sink_node_id")?,
mix_node_id: row.get("mix_node_id")?,
sink_pid: row.get("sink_pid")?,
mix_pid: row.get("mix_pid")?,
loopback_id: row.get("loopback_id")?,
volume_pct: row.get("volume_pct")?,
mic_source: row.get("mic_source")?,
mic_volume_pct: row.get("mic_volume_pct")?,
})
}
pub fn load(conn: &Connection, name: &str) -> Result<Self> {
require_valid_name(name)?;
conn.query_row("SELECT * FROM vmics WHERE name = ?1", params![name], Self::from_row)
.optional()?
.ok_or_else(|| VmicError::NotFound(name.to_string()))
}
pub fn exists(conn: &Connection, name: &str) -> Result<bool> {
Ok(conn
.query_row("SELECT 1 FROM vmics WHERE name = ?1", params![name], |_| Ok(()))
.optional()?
.is_some())
}
/// Upsert: replaces the row for `self.name` if it exists.
pub fn save(&self, conn: &Connection) -> Result<()> {
conn.execute(
"INSERT INTO vmics (
name, sink_name, mid_name, mix_name, source_name,
sink_node_id, mix_node_id, sink_pid, mix_pid, loopback_id, volume_pct,
mic_source, mic_volume_pct
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
ON CONFLICT(name) DO UPDATE SET
sink_name = excluded.sink_name,
mid_name = excluded.mid_name,
mix_name = excluded.mix_name,
source_name = excluded.source_name,
sink_node_id = excluded.sink_node_id,
mix_node_id = excluded.mix_node_id,
sink_pid = excluded.sink_pid,
mix_pid = excluded.mix_pid,
loopback_id = excluded.loopback_id,
volume_pct = excluded.volume_pct,
mic_source = excluded.mic_source,
mic_volume_pct = excluded.mic_volume_pct",
params![
self.name,
self.sink_name,
self.mid_name,
self.mix_name,
self.source_name,
self.sink_node_id,
self.mix_node_id,
self.sink_pid,
self.mix_pid,
self.loopback_id,
self.volume_pct,
self.mic_source,
self.mic_volume_pct,
],
)?;
Ok(())
}
pub fn delete(conn: &Connection, name: &str) -> Result<()> {
conn.execute("DELETE FROM vmics WHERE name = ?1", params![name])?;
Ok(())
}
/// Lists every persisted vmic, sorted by name.
pub fn list_all(conn: &Connection) -> Result<Vec<Self>> {
let mut stmt = conn.prepare("SELECT * FROM vmics ORDER BY name")?;
let rows = stmt.query_map([], Self::from_row)?;
let mut out = Vec::new();
for row in rows {
out.push(row?);
}
Ok(out)
}
}

59
src/ui.rs Normal file
View File

@@ -0,0 +1,59 @@
//! Colorized status output (NO_COLOR-aware).
use std::io::IsTerminal;
use std::sync::OnceLock;
/// Honors NO_COLOR (<https://no-color.org>), dumb terminals, and redirected output.
fn color_enabled() -> bool {
static ENABLED: OnceLock<bool> = OnceLock::new();
*ENABLED.get_or_init(|| {
std::env::var_os("NO_COLOR").is_none()
&& std::env::var("TERM").map(|t| t != "dumb").unwrap_or(true)
&& std::io::stdout().is_terminal()
&& std::io::stderr().is_terminal()
})
}
fn paint(code: &str, s: &str) -> String {
if color_enabled() {
format!("\x1b[{code}m{s}\x1b[0m")
} else {
s.to_string()
}
}
pub fn red(s: &str) -> String {
paint("31", s)
}
pub fn yellow(s: &str) -> String {
paint("33", s)
}
pub fn green(s: &str) -> String {
paint("32", s)
}
pub fn blue(s: &str) -> String {
paint("34", s)
}
pub fn cyan(s: &str) -> String {
paint("36", s)
}
pub fn err(msg: &str) {
eprintln!("{}", red(&format!("Error: {msg}")));
}
pub fn warn(msg: &str) {
eprintln!("{}", yellow(&format!("Warning: {msg}")));
}
pub fn info(msg: &str) {
println!("{}", blue(msg));
}
pub fn ok(msg: &str) {
println!("{}", green(msg));
}