diff --git a/Cargo.lock b/Cargo.lock index 3d36df75..710f8f78 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,16 @@ dependencies = [ "memchr", ] +[[package]] +name = "annotate-snippets" +version = "0.11.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "710e8eae58854cdc1790fcb56cca04d712a17be849eeb81da2a724bf4bae2bc4" +dependencies = [ + "anstyle", + "unicode-width", +] + [[package]] name = "anstream" version = "1.0.0" @@ -61,6 +71,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + [[package]] name = "appendlist" version = "1.4.0" @@ -88,6 +104,25 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "bindgen" +version = "0.72.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895" +dependencies = [ + "annotate-snippets", + "bitflags", + "cexpr", + "clang-sys", + "itertools", + "proc-macro2", + "quote", + "regex", + "rustc-hash", + "shlex 1.3.0", + "syn 2.0.119", +] + [[package]] name = "bitflags" version = "2.13.1" @@ -144,7 +179,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" dependencies = [ "find-msvc-tools", - "shlex", + "shlex 2.0.1", +] + +[[package]] +name = "cexpr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" +dependencies = [ + "nom 7.1.3", +] + +[[package]] +name = "cfg-expr" +version = "0.20.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe4ece8474b5f766c63426647e7b4b316b67431ade1036a8313cee24a03ae917" +dependencies = [ + "smallvec", + "target-lexicon", ] [[package]] @@ -169,6 +223,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "clang-sys" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a" +dependencies = [ + "glob", + "libc", + "libloading", +] + [[package]] name = "clap" version = "4.6.6" @@ -209,6 +274,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + [[package]] name = "colorchoice" version = "1.0.5" @@ -224,6 +298,12 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "cookie-factory" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9885fa71e26b8ab7855e2ec7cae6e9b380edff76cd052e07c683a0319d51b3a2" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -233,6 +313,15 @@ dependencies = [ "libc", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.22" @@ -311,6 +400,12 @@ dependencies = [ "linux-raw-sys 0.9.4", ] +[[package]] +name = "either" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" + [[package]] name = "encoding_rs" version = "0.8.35" @@ -500,6 +595,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "khronos_api" version = "3.1.0" @@ -528,6 +632,33 @@ dependencies = [ "windows-link", ] +[[package]] +name = "libspa" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "882f7427e7989dcc9d388b7f05c4630390a1d7696f9ffa469cd4a7a48f0b4c40" +dependencies = [ + "bitflags", + "cc", + "cookie-factory", + "libc", + "libspa-sys", + "nom 8.0.0", + "rustix 1.1.4", + "system-deps", +] + +[[package]] +name = "libspa-sys" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b6e17bdaf63ed0d5e4144022624032b41fd9733112e8c74ac26fc9bf1291924" +dependencies = [ + "bindgen", + "cc", + "system-deps", +] + [[package]] name = "libudev-sys" version = "0.1.4" @@ -586,6 +717,12 @@ dependencies = [ "libc", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "nescope" version = "0.1.0" @@ -609,6 +746,21 @@ dependencies = [ name = "nesprotocol" version = "0.1.0" +[[package]] +name = "neswire" +version = "0.1.0" +dependencies = [ + "anyhow", + "bytemuck", + "clap", + "crossbeam-channel", + "nesprotocol", + "opus-head-sys", + "pipewire", + "tracing", + "tracing-subscriber", +] + [[package]] name = "nix" version = "0.31.3" @@ -621,6 +773,25 @@ dependencies = [ "libc", ] +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + +[[package]] +name = "nom" +version = "8.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405" +dependencies = [ + "memchr", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -651,6 +822,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "opus-head-sys" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df99c7905445bdbb5c3e74161d4808625af15dd59a52427c7eeaed70c5c1c7f4" +dependencies = [ + "cmake", +] + [[package]] name = "paste" version = "1.0.15" @@ -663,6 +843,31 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pipewire" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bde71084c4e25959d68f1ea54daa75e5ecdb338e5caf0b5510143b79baa32d5c" +dependencies = [ + "bitflags", + "libc", + "libspa", + "libspa-sys", + "pipewire-sys", + "rustix 1.1.4", +] + +[[package]] +name = "pipewire-sys" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ce653f53e63e5b93853218092ee9a8906a5d082c92f3f1db26316955dd63ce0" +dependencies = [ + "bindgen", + "libspa-sys", + "system-deps", +] + [[package]] name = "pixman" version = "0.2.1" @@ -797,6 +1002,18 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "regex" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + [[package]] name = "regex-automata" version = "0.4.18" @@ -814,6 +1031,12 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + [[package]] name = "rustix" version = "0.38.44" @@ -846,6 +1069,35 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "sha2" version = "0.10.9" @@ -866,6 +1118,12 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + [[package]] name = "shlex" version = "2.0.1" @@ -956,6 +1214,25 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "system-deps" +version = "7.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "396a35feb67335377e0251fcbc1092fc85c484bd4e3a7a54319399da127796e7" +dependencies = [ + "cfg-expr", + "heck", + "pkg-config", + "toml", + "version-compare", +] + +[[package]] +name = "target-lexicon" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adb6935a6f5c20170eeceb1a3835a49e12e19d792f6dd344ccc76a985ca5a6ca" + [[package]] name = "tempfile" version = "3.27.0" @@ -1018,6 +1295,45 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "toml" +version = "1.1.4+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3aace63f4bbcdfc2c965b059de67119c89c4017a70d633be6c104910f67056f5" +dependencies = [ + "indexmap", + "serde_core", + "serde_spanned", + "toml_datetime", + "toml_parser", + "toml_writer", + "winnow", +] + +[[package]] +name = "toml_datetime" +version = "1.1.1+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7" +dependencies = [ + "serde_core", +] + +[[package]] +name = "toml_parser" +version = "1.1.3+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d38ac1cf9b95face32296c0a3ede1fdc270627c9d9c02a7274dd6d960dc4d56" +dependencies = [ + "winnow", +] + +[[package]] +name = "toml_writer" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" + [[package]] name = "tracing" version = "0.1.44" @@ -1104,6 +1420,12 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-width" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" + [[package]] name = "utf8parse" version = "0.2.2" @@ -1116,6 +1438,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "version-compare" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03c2856837ef78f57382f06b2b8563a2f512f7185d732608fd9176cb3b8edf0e" + [[package]] name = "version_check" version = "0.9.5" @@ -1382,6 +1710,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "winnow" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" + [[package]] name = "wit-bindgen" version = "0.57.1" diff --git a/Cargo.toml b/Cargo.toml index 4bd013d3..3388c06c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ resolver = "3" members = [ "apps/nescope", + "apps/neswire", "crates/nesprotocol", ] diff --git a/apps/neswire/.gitignore b/apps/neswire/.gitignore new file mode 100644 index 00000000..6af16d4c --- /dev/null +++ b/apps/neswire/.gitignore @@ -0,0 +1,45 @@ +``` +# Rust/Cargo specific +/target/ +Cargo.lock +**/Cargo.lock +.cargo/ +cargo_home/ + +# Git related +.git/ + +# Temporary files +*.tmp +*.swp +*~ + +# Logs +*.log + +# OS generated files +.DS_Store +Thumbs.db + +# Environment files +.env +.env.local +*.env.* + +# IDE files +.vscode/ +.idea/ +*.swp +*.swo + +# Build artifacts +build/ +dist/ +*.o +*.obj +*.exe +*.dll +*.so +*.a +*.out +``` \ No newline at end of file diff --git a/apps/neswire/Cargo.toml b/apps/neswire/Cargo.toml new file mode 100644 index 00000000..045886c3 --- /dev/null +++ b/apps/neswire/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "neswire" +version = "0.1.0" +edition.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +pipewire = "0.10" +crossbeam-channel = "0.5" +anyhow = "1" +opus-head-sys = "0.3" +bytemuck = { version = "1", features = ["extern_crate_alloc"] } +clap = { version = "4", features = ["derive", "env"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +nesprotocol = { path = "../../crates/nesprotocol" } diff --git a/apps/neswire/README.md b/apps/neswire/README.md new file mode 100644 index 00000000..b4c86b89 --- /dev/null +++ b/apps/neswire/README.md @@ -0,0 +1,36 @@ +## neswire + +A small custom PipeWire sink for cloud gaming audio capture. + +Currently for debugging uses RTP to send Opus (with FEC enabled by default) over to target address. + + +### Testing + +#### Mono/Stereo + +Launch gstreamer pipeline to receive audio as so (will save incoming RTP audio into test.mkv): +```bash +gst-launch-1.0 udpsrc port=12345 caps="application/x-rtp,media=audio,encoding-name=OPUS,clock-rate=48000,payload=111" ! rtpopusdepay2 ! opusdec ! matroskamux ! filesink location=test.mkv sync=false +``` + +Then run neswire like so for example: +```bash +cargo run --release --bin neswire -- --rtp-addr 127.0.0.1:12345 +``` + +#### Surround + +Launch gstreamer pipeline to receive audio as so (will save incoming RTP audio into test_multi.mkv): +```bash +gst-launch-1.0 udpsrc port=12345 caps='application/x-rtp,media=audio,encoding-name=MULTIOPUS,clock-rate=48000,payload=111,encoding-params=(string)8,num_streams=(string)5,coupled_streams=(string)3,channel_mapping=(string)"0,6,1,2,3,4,5,7"' ! rtpopusdepay2 ! opusdec ! matroskamux ! filesink location=test.mkv sync=false +``` + +Then run neswire like so for example: +```bash +cargo run --release --bin neswire -- --rtp-addr 127.0.0.1:12345 --channels 8 +``` + + +For both afterwards, set the neswire sink as audio output source in your system (or specify it as output in some app/game), + then play some audio, stop gst pipeline and sink, listen to results after. diff --git a/apps/neswire/src/bin/hub-stub.rs b/apps/neswire/src/bin/hub-stub.rs new file mode 100644 index 00000000..1ce15bb3 --- /dev/null +++ b/apps/neswire/src/bin/hub-stub.rs @@ -0,0 +1,205 @@ +//! A stand-in for nestri-guest-hub's audio IPC listener. +//! +//! neswire's only output is a Unix datagram socket that the hub binds, so +//! running it outside a guest means having nothing to talk to: it retries +//! `connect` forever and there is no way to see what it would have sent. This +//! binds that socket and reports what arrives. +//! +//! It decodes the Opus rather than only counting bytes, because byte counts +//! cannot tell the two interesting failures apart. Opus codes digital silence +//! in about two bytes a packet, so a sink receiving nothing but zeros still +//! produces a steady ~3 kbps and looks, from every meter downstream, exactly +//! like one that is working. Peak amplitude is what distinguishes them. +//! +//! Mirrors `ipc_listener::run_audio_listener` in the hub: same bind, same 0o666, +//! same stream-type and codec checks. Where it differs from the hub, the hub is +//! right and this should be corrected. + +use std::os::unix::net::UnixDatagram; +use std::time::{Duration, Instant}; + +use anyhow::{Result, bail}; +use clap::Parser; +use opus_head_sys::*; + +use nesprotocol::{CODEC_OPUS, STREAM_AUDIO, decode_ipc_frame}; + +#[derive(Parser, Debug)] +#[command(about = "Receive and measure neswire's audio IPC stream")] +struct Args { + /// Path to bind, matching neswire's --ipc-path. + #[arg(long, default_value = "/tmp/nestri-audio.sock")] + ipc_path: String, + + /// Channels neswire is configured for. The decoder has to agree with the + /// encoder; a mismatch here reads as garbage, not as an error. + #[arg(long, default_value_t = 2)] + channels: u32, + + /// How often to print a line. + #[arg(long, default_value_t = 1.0)] + interval_secs: f64, +} + +struct MsDecoder { + ptr: *mut OpusMSDecoder, + channels: usize, +} + +impl MsDecoder { + /// Built to match `MsEncoder::create_surround` for the same channel count: + /// stereo is mapping family 0, one stream, one coupled pair. + fn new(sample_rate: u32, channels: u32) -> Result { + let (streams, coupled, mapping): (i32, i32, Vec) = match channels { + 2 => (1, 1, vec![0, 1]), + 6 => (4, 2, vec![0, 4, 1, 2, 3, 5]), + 8 => (5, 3, vec![0, 6, 1, 2, 3, 4, 5, 7]), + n => bail!("unsupported channel count: {n}"), + }; + + let mut error: i32 = 0; + let ptr = unsafe { + opus_multistream_decoder_create( + sample_rate as i32, + channels as i32, + streams, + coupled, + mapping.as_ptr(), + &mut error, + ) + }; + if error != OPUS_OK as i32 || ptr.is_null() { + bail!("opus_multistream_decoder_create failed: {error}"); + } + Ok(Self { + ptr, + channels: channels as usize, + }) + } + + /// Returns the decoded samples, interleaved. + fn decode(&self, packet: &[u8], pcm: &mut [f32]) -> Result { + let frame_size = (pcm.len() / self.channels) as i32; + let decoded = unsafe { + opus_multistream_decode_float( + self.ptr, + packet.as_ptr(), + packet.len() as i32, + pcm.as_mut_ptr(), + frame_size, + 0, + ) + }; + if decoded < 0 { + bail!("opus decode failed: {decoded}"); + } + Ok(decoded as usize * self.channels) + } +} + +impl Drop for MsDecoder { + fn drop(&mut self) { + unsafe { opus_multistream_decoder_destroy(self.ptr) }; + } +} + +fn main() -> Result<()> { + let args = Args::parse(); + + // Same sequence as the hub: clear a stale socket, bind, widen the mode. + // neswire runs as a different user there, and 0o666 is what makes that + // work; keeping it here means this stub cannot pass a case the hub fails. + let _ = std::fs::remove_file(&args.ipc_path); + let socket = UnixDatagram::bind(&args.ipc_path)?; + std::fs::set_permissions( + &args.ipc_path, + std::os::unix::fs::PermissionsExt::from_mode(0o666), + )?; + socket.set_read_timeout(Some(Duration::from_millis(200)))?; + + println!("listening on {}", args.ipc_path); + println!("waiting for neswire..."); + + let decoder = MsDecoder::new(48_000, args.channels)?; + // 120ms at 48kHz is the largest frame Opus can produce, so nothing that + // decodes at all can overrun this. + let mut pcm = vec![0f32; 5760 * args.channels as usize]; + let mut buf = vec![0u8; 65536]; + + let interval = Duration::from_secs_f64(args.interval_secs); + let mut window_start = Instant::now(); + let mut packets = 0u64; + let mut bytes = 0u64; + let mut peak = 0f32; + let mut samples = 0u64; + let mut seen_anything = false; + + loop { + match socket.recv(&mut buf) { + Ok(n) => { + let Some(frame) = decode_ipc_frame(&buf[..n]) else { + eprintln!("invalid IPC frame ({n} bytes)"); + continue; + }; + if frame.stream_type != STREAM_AUDIO { + eprintln!("unexpected stream type: {}", frame.stream_type); + continue; + } + if frame.codec != CODEC_OPUS { + eprintln!("unexpected codec: {}", frame.codec); + continue; + } + + packets += 1; + bytes += frame.data.len() as u64; + seen_anything = true; + + match decoder.decode(frame.data, &mut pcm) { + Ok(count) => { + samples += count as u64; + for sample in &pcm[..count] { + peak = peak.max(sample.abs()); + } + } + Err(e) => eprintln!("{e}"), + } + } + Err(ref e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => {} + Err(e) => return Err(e.into()), + } + + let elapsed = window_start.elapsed(); + if elapsed >= interval { + let secs = elapsed.as_secs_f64(); + let kbps = bytes as f64 * 8.0 / 1000.0 / secs; + + if packets == 0 { + println!( + "{}", + if seen_anything { + "no packets — neswire stopped sending" + } else { + "no packets yet" + } + ); + } else if peak == 0.0 { + println!( + "{packets:>4} pkt {kbps:>6.1} kbps {samples:>6} samples \ + SILENT — decodes cleanly, every sample is zero" + ); + } else { + println!( + "{packets:>4} pkt {kbps:>6.1} kbps {samples:>6} samples peak {peak:.4}" + ); + } + + window_start = Instant::now(); + packets = 0; + bytes = 0; + samples = 0; + peak = 0.0; + } + } +} diff --git a/apps/neswire/src/encoder.rs b/apps/neswire/src/encoder.rs new file mode 100644 index 00000000..1f5509ef --- /dev/null +++ b/apps/neswire/src/encoder.rs @@ -0,0 +1,421 @@ +use anyhow::{Result, bail}; +use crossbeam_channel::Receiver; +use opus_head_sys::*; +use std::os::unix::net::UnixDatagram; +use std::time::Instant; + +use nesprotocol::{CODEC_OPUS, STREAM_AUDIO, encode_ipc_frame}; + +pub struct EncoderConfig { + pub channels: u32, + pub sample_rate: u32, + pub frame_size: u32, + pub ipc_path: String, + pub bitrate_per_channel: u32, +} + +struct MsEncoder { + ptr: *mut OpusMSEncoder, +} +impl MsEncoder { + fn create_surround(config: &EncoderConfig) -> Result { + let mut error: i32 = 0; + let mut streams: i32 = 0; + let mut coupled_streams: i32 = 0; + let mut mapping = [0u8; 255]; + + let mapping_family = if config.channels > 2 { 1 } else { 0 }; + + let encoder = unsafe { + opus_multistream_surround_encoder_create( + config.sample_rate as i32, + config.channels as i32, + mapping_family, + &mut streams, + &mut coupled_streams, + mapping.as_mut_ptr(), + OPUS_APPLICATION_AUDIO as i32, + &mut error, + ) + }; + + if error != OPUS_OK as i32 || encoder.is_null() { + bail!( + "opus_multistream_surround_encoder_create failed: {}", + opus_error(error) + ); + } + + tracing::info!( + "opus multistream mapping: streams={}, coupled={}, mapping={:?}", + streams, + coupled_streams, + &mapping[..config.channels as usize], + ); + + let bitrate = (config.bitrate_per_channel * config.channels) * 1000; + + let ret = unsafe { + opus_multistream_encoder_ctl(encoder, OPUS_SET_BITRATE_REQUEST as i32, bitrate) + }; + if ret != OPUS_OK as i32 { + unsafe { opus_multistream_encoder_destroy(encoder) }; + bail!("failed to set bitrate: {}", opus_error(ret)); + } + + tracing::info!("opus bitrate: {}kbps", bitrate / 1000); + + let ret = unsafe { + opus_multistream_encoder_ctl(encoder, OPUS_SET_COMPLEXITY_REQUEST as i32, 10 as i32) + }; + if ret != OPUS_OK as i32 { + unsafe { opus_multistream_encoder_destroy(encoder) }; + bail!("failed to set complexity: {}", opus_error(ret)); + } + + tracing::info!("opus complexity set to 10 (max quality)"); + + let ret = unsafe { + opus_multistream_encoder_ctl( + encoder, + OPUS_SET_SIGNAL_REQUEST as i32, + OPUS_SIGNAL_MUSIC as i32, + ) + }; + if ret != OPUS_OK as i32 { + unsafe { opus_multistream_encoder_destroy(encoder) }; + bail!("failed to set signal type: {}", opus_error(ret)); + } + + tracing::info!("opus signal type set to MUSIC"); + + Ok(Self { ptr: encoder }) + } + + fn encode_float(&self, pcm: &[f32], frame_size: u32, output: &mut [u8]) -> Result { + let ret = unsafe { + opus_multistream_encode_float( + self.ptr, + pcm.as_ptr(), + frame_size as i32, + output.as_mut_ptr(), + output.len() as i32, + ) + }; + + if ret < 0 { + bail!("opus encode failed: {}", opus_error(ret)); + } + + Ok(ret as usize) + } +} +impl Drop for MsEncoder { + fn drop(&mut self) { + unsafe { + opus_multistream_encoder_destroy(self.ptr); + } + } +} + +fn opus_error(code: i32) -> String { + unsafe { + let ptr = opus_strerror(code); + if ptr.is_null() { + format!("unknown error {}", code) + } else { + std::ffi::CStr::from_ptr(ptr).to_string_lossy().into_owned() + } + } +} + +fn send_frame( + encoder: &MsEncoder, + socket: &UnixDatagram, + config: &EncoderConfig, + frame: &[f32], + start: &Instant, + opus_buf: &mut Vec, +) -> Result<()> { + let encoded_len = encoder.encode_float(frame, config.frame_size, opus_buf)?; + let encoded_data = &opus_buf[..encoded_len]; + let timestamp_ms = start.elapsed().as_millis() as u32; + let ipc_frame = encode_ipc_frame( + STREAM_AUDIO, + CODEC_OPUS, + 0, + timestamp_ms, + 0, + 0, + encoded_data, + ); + socket.send(&ipc_frame)?; + Ok(()) +} + +pub fn run(config: EncoderConfig, rx: Receiver>) -> Result<()> { + tracing::info!( + "encoder started — IPC {}, {}ch, {} samples/frame", + config.ipc_path, + config.channels, + config.frame_size, + ); + + let socket = UnixDatagram::unbound()?; + let encoder = MsEncoder::create_surround(&config)?; + + let mut opus_buf = vec![0u8; 4000]; + let start = Instant::now(); + + // Connect to hub with retry + loop { + match socket.connect(&config.ipc_path) { + Ok(()) => { + tracing::info!("IPC connected to {}", config.ipc_path); + break; + } + Err(e) => { + tracing::warn!( + "IPC connect to {} failed (retrying in 2s): {e}", + config.ipc_path + ); + std::thread::sleep(std::time::Duration::from_secs(2)); + } + } + } + + // Send loop + while let Ok(frame) = rx.recv() { + send_frame(&encoder, &socket, &config, &frame, &start, &mut opus_buf)?; + } + + tracing::info!("encoder shut down"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use nesprotocol::decode_ipc_frame; + use std::os::unix::net::UnixDatagram; + + const SAMPLE_RATE: u32 = 48_000; + const CHANNELS: u32 = 2; + /// 5ms, the default. 240 samples per channel. + const FRAME_SIZE: u32 = 240; + + fn config(ipc_path: &str) -> EncoderConfig { + EncoderConfig { + channels: CHANNELS, + sample_rate: SAMPLE_RATE, + frame_size: FRAME_SIZE, + ipc_path: ipc_path.to_string(), + bitrate_per_channel: 64, + } + } + + fn silence() -> Vec { + vec![0.0; (FRAME_SIZE * CHANNELS) as usize] + } + + /// One frame of a 440Hz tone, interleaved stereo, starting at `frame_index` + /// so consecutive frames form a continuous wave rather than restarting. + fn tone(frame_index: usize) -> Vec { + let start = frame_index * FRAME_SIZE as usize; + (0..FRAME_SIZE as usize) + .flat_map(|i| { + let t = (start + i) as f32 / SAMPLE_RATE as f32; + let v = (t * 440.0 * std::f32::consts::TAU).sin() * 0.5; + [v, v] + }) + .collect() + } + + /// The decoder the hub's client ends up using, in miniature: same + /// parameters the desktop app builds for stereo. + struct Decoder(*mut OpusMSDecoder); + + impl Decoder { + fn new() -> Self { + let mut error = 0i32; + let mapping = [0u8, 1]; + let ptr = unsafe { + opus_multistream_decoder_create( + SAMPLE_RATE as i32, + CHANNELS as i32, + 1, + 1, + mapping.as_ptr(), + &mut error, + ) + }; + assert_eq!(error, OPUS_OK as i32, "decoder create failed"); + Self(ptr) + } + + fn decode(&self, packet: &[u8]) -> Vec { + let mut pcm = vec![0f32; (FRAME_SIZE * CHANNELS) as usize]; + let n = unsafe { + opus_multistream_decode_float( + self.0, + packet.as_ptr(), + packet.len() as i32, + pcm.as_mut_ptr(), + FRAME_SIZE as i32, + 0, + ) + }; + assert!(n > 0, "decode failed: {n}"); + pcm.truncate(n as usize * CHANNELS as usize); + pcm + } + } + + impl Drop for Decoder { + fn drop(&mut self) { + unsafe { opus_multistream_decoder_destroy(self.0) }; + } + } + + fn peak(pcm: &[f32]) -> f32 { + pcm.iter().fold(0f32, |acc, s| acc.max(s.abs())) + } + + /// Silence is nearly free to encode, and that is a trap worth pinning down. + /// + /// The configured bitrate is a ceiling, not a floor. At 200 packets a second + /// these two-byte packets come to roughly 3kbps of perfectly valid, + /// perfectly silent Opus — which looks, to every byte counter downstream, + /// like a working stream. A whole debugging session was spent on the + /// difference. If this test starts failing because silence got expensive, + /// that reasoning needs revisiting. + #[test] + fn silence_costs_almost_nothing_to_encode() { + let config = config("/nonexistent"); + let encoder = MsEncoder::create_surround(&config).expect("encoder"); + let mut buf = vec![0u8; 4000]; + + // Past the encoder's warm-up, where the first packets are larger. + for _ in 0..10 { + encoder + .encode_float(&silence(), FRAME_SIZE, &mut buf) + .expect("encode"); + } + + let len = encoder + .encode_float(&silence(), FRAME_SIZE, &mut buf) + .expect("encode"); + assert!( + len <= 8, + "silence took {len} bytes; the ~3kbps silent-stream signature no \ + longer holds and the diagnostics that rely on it are wrong" + ); + } + + /// The other half of the above: real audio must cost real bytes, or the + /// test before this one would pass on a permanently broken encoder. + #[test] + fn a_tone_costs_far_more_than_silence() { + let config = config("/nonexistent"); + // One encoder each. Opus codes a stream, not isolated frames, so + // feeding both signals to one encoder makes every "silent" frame the + // tail of a tone and costs it accordingly -- which is a measurement of + // nothing. + let quiet_encoder = MsEncoder::create_surround(&config).expect("encoder"); + let loud_encoder = MsEncoder::create_surround(&config).expect("encoder"); + let mut buf = vec![0u8; 4000]; + + let mut quiet = 0; + let mut loud = 0; + for i in 0..40 { + quiet += quiet_encoder + .encode_float(&silence(), FRAME_SIZE, &mut buf) + .expect("encode"); + loud += loud_encoder + .encode_float(&tone(i), FRAME_SIZE, &mut buf) + .expect("encode"); + } + + assert!( + loud > quiet * 4, + "a tone encoded to {loud} bytes against {quiet} for silence; the \ + encoder is not responding to its input" + ); + } + + /// A tone goes in and comes back out, through the exact encoder the guest + /// runs and the exact decoder parameters the desktop client builds. + /// + /// This is the one that would have caught a channel-mapping or sample-rate + /// disagreement between the two ends, which no byte count can see. + #[test] + fn a_tone_survives_the_round_trip() { + let config = config("/nonexistent"); + let encoder = MsEncoder::create_surround(&config).expect("encoder"); + let decoder = Decoder::new(); + let mut buf = vec![0u8; 4000]; + + // Opus needs a few frames before its output is representative, so the + // assertion is on the tail rather than the first packet. + let mut last = Vec::new(); + for i in 0..40 { + let len = encoder + .encode_float(&tone(i), FRAME_SIZE, &mut buf) + .expect("encode"); + last = decoder.decode(&buf[..len]); + } + + assert_eq!( + last.len(), + (FRAME_SIZE * CHANNELS) as usize, + "decoded frame is the wrong length" + ); + assert!( + peak(&last) > 0.1, + "a 0.5-amplitude tone decoded to a peak of {:.4}", + peak(&last) + ); + } + + /// What actually crosses the socket is what the hub knows how to read. + /// + /// The hub rejects any frame whose stream type is not `STREAM_AUDIO` or + /// whose codec is not `CODEC_OPUS`, and drops it with a warning nobody + /// reads. This asserts the contract from the sending side. + #[test] + fn the_hub_receives_a_frame_it_can_parse() { + let path = std::env::temp_dir().join(format!( + "neswire-test-{}-{}.sock", + std::process::id(), + line!() + )); + let _ = std::fs::remove_file(&path); + + // Bound first: the hub binds and neswire connects, so a test that did + // this the other way round would not be testing the real sequence. + let listener = UnixDatagram::bind(&path).expect("bind"); + let sender = UnixDatagram::unbound().expect("socket"); + sender.connect(&path).expect("connect"); + + let config = config(path.to_str().expect("utf-8 path")); + let encoder = MsEncoder::create_surround(&config).expect("encoder"); + let start = Instant::now(); + let mut opus_buf = vec![0u8; 4000]; + + send_frame(&encoder, &sender, &config, &tone(0), &start, &mut opus_buf) + .expect("send"); + + let mut buf = vec![0u8; 65536]; + let n = listener.recv(&mut buf).expect("recv"); + let frame = decode_ipc_frame(&buf[..n]).expect("hub could not parse the frame"); + + assert_eq!(frame.stream_type, STREAM_AUDIO); + assert_eq!(frame.codec, CODEC_OPUS); + assert!(!frame.data.is_empty(), "frame carried no payload"); + + // And the payload is Opus the far end can actually decode. + Decoder::new().decode(frame.data); + + let _ = std::fs::remove_file(&path); + } +} diff --git a/apps/neswire/src/main.rs b/apps/neswire/src/main.rs new file mode 100644 index 00000000..7a719c95 --- /dev/null +++ b/apps/neswire/src/main.rs @@ -0,0 +1,58 @@ +mod encoder; +mod sink; + +use clap::Parser; + +#[derive(Parser, Debug)] +struct Args { + /// Path for the audio IPC socket (neswire → nestri-guest-hub) + #[arg( + long, + env = "NESWIRE_IPC_PATH", + default_value = "/tmp/nestri-audio.sock" + )] + ipc_path: String, + + /// Output channels: 2, 6, or 8 + #[arg(long, env = "NESWIRE_CHANNELS", default_value_t = 2)] + channels: u32, + + /// Packet duration in ms (5, 10..) + #[arg(long, env = "NESWIRE_PACKET_DURATION_MS", default_value_t = 5)] + packet_duration_ms: u32, + + /// Bitrate per channel in kbps + #[arg(long, env = "NESWIRE_BITRATE_PER_CHANNEL", default_value_t = 64)] + bitrate_per_channel: u32, +} + +fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let args = Args::parse(); + + let sample_rate = 48_000u32; + let frame_size = (sample_rate * args.packet_duration_ms) / 1000; + + let (tx, rx) = crossbeam_channel::bounded::>(256); + + let encoder_config = encoder::EncoderConfig { + channels: args.channels, + sample_rate, + frame_size, + ipc_path: args.ipc_path, + bitrate_per_channel: args.bitrate_per_channel, + }; + + std::thread::spawn(move || { + if let Err(e) = encoder::run(encoder_config, rx) { + tracing::error!("encoder thread died: {e:#}"); + } + }); + + sink::run(sample_rate, args.channels, frame_size, tx)?; + + Ok(()) +} diff --git a/apps/neswire/src/sink.rs b/apps/neswire/src/sink.rs new file mode 100644 index 00000000..0f976bee --- /dev/null +++ b/apps/neswire/src/sink.rs @@ -0,0 +1,183 @@ +use anyhow::Result; +use crossbeam_channel::Sender; +use pipewire::{self as pw, loop_::Signal, spa, stream::*}; +use spa::param::audio::{AudioFormat, AudioInfoRaw}; + +pub fn run(sample_rate: u32, channels: u32, frame_size: u32, tx: Sender>) -> Result<()> { + pw::init(); + tracing::info!("pipewire init ok"); + + let mainloop = + pw::main_loop::MainLoopRc::new(None).map_err(|e| anyhow::anyhow!("mainloop: {e}"))?; + tracing::info!("mainloop created"); + + let context = pw::context::ContextRc::new(&mainloop, None) + .map_err(|e| anyhow::anyhow!("context: {e}"))?; + tracing::info!("context created"); + + let core = context + .connect_rc(None) + .map_err(|e| anyhow::anyhow!("core connect: {e}"))?; + tracing::info!("core connected"); + + let stream = pw::stream::StreamRc::new( + core, + "neswire", + pw::properties::properties! { + *pw::keys::MEDIA_TYPE => "Audio", + *pw::keys::MEDIA_CATEGORY => "Capture", + *pw::keys::MEDIA_CLASS => "Audio/Sink", + *pw::keys::NODE_NAME => "neswire", + *pw::keys::NODE_DESCRIPTION => "Neswire Cloud Gaming Audio Sink", + // Prevent session manager from suspending us + "node.always-process" => "true", + // Desired latency in samples + "node.latency" => format!("{}/{}", frame_size, sample_rate), + }, + ) + .map_err(|e| anyhow::anyhow!("stream create: {e}"))?; + + // Accumulation buffer for collecting enough samples before sending a frame + let samples_per_frame = (frame_size * channels) as usize; + + struct State { + tx: Sender>, + accum: Vec, + samples_per_frame: usize, + } + + let state = State { + tx, + accum: Vec::with_capacity(samples_per_frame * 2), + samples_per_frame, + }; + + let _listener = stream + .add_local_listener_with_user_data(state) + .param_changed(move |_, _state, id, pod| { + if id != spa::param::ParamType::Format.as_raw() { + return; + } + // You can parse the negotiated format here if needed + // For now we're requesting a fixed format so it should match + if let Some(_pod) = pod { + tracing::info!("format negotiated"); + } + }) + .process(move |stream, state| { + // Dequeue the buffer from PipeWire + if let Some(mut buffer) = stream.dequeue_buffer() { + let datas = buffer.datas_mut(); + if let Some(data) = datas.first_mut() { + let chunk = data.chunk(); + let offset = chunk.offset() as usize; + let size = chunk.size() as usize; + + if let Some(slice) = data.data() { + let audio_bytes = &slice[offset..offset + size]; + + // Reinterpret as f32 samples (we requested F32LE) + let samples: &[f32] = bytemuck::cast_slice(audio_bytes); + + state.accum.extend_from_slice(samples); + + tracing::trace!( + "pw delivered {} samples, accum now {}, frame size {}", + samples.len(), + state.accum.len(), + state.samples_per_frame, + ); + + // Drain complete frames + while state.accum.len() >= state.samples_per_frame { + let frame: Vec = + state.accum.drain(..state.samples_per_frame).collect(); + + if state.tx.try_send(frame).is_err() { + tracing::warn!("encoder falling behind, dropping frame"); + } + } + } + } + } + }) + .register() + .map_err(|e| anyhow::anyhow!("stream register: {e}"))?; + + let mut position = [0u32; 64]; + let channel_map: &[u32] = match channels { + 2 => &[ + spa::sys::SPA_AUDIO_CHANNEL_FL, + spa::sys::SPA_AUDIO_CHANNEL_FR, + ], + 6 => &[ + spa::sys::SPA_AUDIO_CHANNEL_FL, + spa::sys::SPA_AUDIO_CHANNEL_FC, + spa::sys::SPA_AUDIO_CHANNEL_FR, + spa::sys::SPA_AUDIO_CHANNEL_RL, + spa::sys::SPA_AUDIO_CHANNEL_RR, + spa::sys::SPA_AUDIO_CHANNEL_LFE, + ], + 8 => &[ + spa::sys::SPA_AUDIO_CHANNEL_FL, + spa::sys::SPA_AUDIO_CHANNEL_FC, + spa::sys::SPA_AUDIO_CHANNEL_FR, + spa::sys::SPA_AUDIO_CHANNEL_SL, + spa::sys::SPA_AUDIO_CHANNEL_SR, + spa::sys::SPA_AUDIO_CHANNEL_RL, + spa::sys::SPA_AUDIO_CHANNEL_RR, + spa::sys::SPA_AUDIO_CHANNEL_LFE, + ], + _ => anyhow::bail!("unsupported channel count: {channels}"), + }; + position[..channel_map.len()].copy_from_slice(channel_map); + + // Build the format we want: f32le, 48kHz, N channels + let mut audio_info = AudioInfoRaw::new(); + audio_info.set_format(AudioFormat::F32LE); + audio_info.set_rate(sample_rate); + audio_info.set_channels(channels); + audio_info.set_position(position); + + let values: Vec = pw::spa::pod::serialize::PodSerializer::serialize( + std::io::Cursor::new(Vec::new()), + &pw::spa::pod::Value::Object(pw::spa::pod::Object { + type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(), + id: pw::spa::param::ParamType::EnumFormat.as_raw(), + properties: audio_info.into(), + }), + )? + .0 + .into_inner(); + + let mut params = [pw::spa::pod::Pod::from_bytes(&values).unwrap()]; + + stream + .connect( + spa::utils::Direction::Input, // We receive audio (we're a sink) + Some(pw::constants::ID_ANY), + StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS | StreamFlags::RT_PROCESS, + &mut params, + ) + .map_err(|e| anyhow::anyhow!("stream connect: {e}"))?; + tracing::info!("stream connected"); + + tracing::info!( + "neswire sink running — {}ch, {}Hz, {} samples/frame", + channels, + sample_rate, + frame_size + ); + + let weak = mainloop.downgrade(); + let _sigint = mainloop.loop_().add_signal_local(Signal::INT, move || { + if let Some(mainloop) = weak.upgrade() { + mainloop.quit(); + } + }); + + mainloop.run(); + + tracing::info!("shutting down"); + Ok(()) +}