docs: publish desktop audio with PocketStation

README.md

@@ -341,6 +341,7 @@ Software encoders (libvpx for VP8/VP9, libaom for AV1, OpenH264 for H264) are us
 - [Basic room](https://github.com/livekit/python-sdks/blob/main/examples/basic_room.py): Connect to a room
 - [Publish hue](https://github.com/livekit/python-sdks/blob/main/examples/publish_hue.py): Publish a rainbow video track
 - [Publish wave](https://github.com/livekit/python-sdks/blob/main/examples/publish_wave.py): Publish a sine wave
+- [Publish desktop audio](examples/desktop_audio/): Publish application, system, or microphone audio with PocketStation
 
 ## Getting help / Contributing
 

examples/README.md

@@ -26,6 +26,7 @@ export LIVEKIT_API_SECRET=secret
 | [publish_wave.py](#publish_wavepy) | Publish sine wave audio |
 | [publish_hue.py](#publish_huepy) | Publish color-cycling video |
 | [play_audio_stream.py](#play_audio_streampy) | Play received audio with sounddevice |
+| [desktop_audio/](desktop_audio/) | Publish application, system, or microphone audio with PocketStation |
 | [webhook.py](#webhookpy) | Webhook event handling |
 | [agent_dispatch.py](#agent_dispatchpy) | Manual agent dispatch |
 
@@ -282,6 +283,14 @@ python play_audio_stream.py
 
 ---
 
+## desktop_audio/
+
+Publish one running application, the complete desktop output mix, or an
+explicitly selected microphone as a LiveKit audio track. See the
+[desktop audio example](desktop_audio/).
+
+---
+
 ## webhook.py
 
 Handle LiveKit webhook events using aiohttp.

examples/desktop_audio/README.md

@@ -0,0 +1,81 @@
+# Publish desktop audio with PocketStation
+
+Publish one running application, the complete desktop output mix, or a
+microphone as a LiveKit audio track. PocketStation captures the local source;
+LiveKit publishes it to the room.
+
+## Install
+
+Use Python 3.11 or newer:
+
+```bash
+python -m venv .venv
+source .venv/bin/activate
+python -m pip install -r requirements.txt
+```
+
+Set the credentials for a LiveKit Cloud project or self-hosted server:
+
+```bash
+export LIVEKIT_URL="wss://your-project.livekit.cloud"
+export LIVEKIT_API_KEY="your-api-key"
+export LIVEKIT_API_SECRET="your-api-secret"
+```
+
+## Publish one application
+
+Start the application and make sure it is producing audio. Replace `Zoom`
+with its display name or application identifier:
+
+```bash
+python publish.py --application Zoom
+```
+
+Join the `desktop-audio` room from another participant to receive the
+`application-audio` track. PocketStation fails before capture begins if the
+selection matches no application or more than one application.
+
+## Other inputs
+
+Capture every sound playing through the desktop:
+
+```bash
+python publish.py --system-audio
+```
+
+Capture the default microphone:
+
+```bash
+python publish.py --microphone
+```
+
+Microphone capture is never enabled automatically. This example publishes raw
+microphone audio without echo cancellation, noise suppression, or automatic
+gain control. Use LiveKit `PlatformAudio` when an interactive call needs those
+voice-processing features.
+
+Pass `--room NAME` to use another room. Pass `--duration 30` for a finite run;
+otherwise press Control-C to stop. PocketStation stops capture and drains
+accepted frames before LiveKit disconnects.
+
+## How it works
+
+PocketStation captures 48 kHz audio in 10 ms frames. A PocketStation Connector
+downmixes each source frame to mono PCM16 and publishes it through a LiveKit
+`AudioSource` with a 100 ms queue. Provider delivery runs outside the
+operating-system audio callback, so a slow room connection does not block
+native capture.
+
+Application and system-audio tracks use LiveKit's
+`SOURCE_SCREENSHARE_AUDIO` classification. The explicit microphone option uses
+`SOURCE_MICROPHONE`.
+
+The example uses PocketStation `0.1.4`, LiveKit `1.1.17`, and LiveKit API
+`1.2.1` from PyPI. Platform permissions and native prerequisites are
+documented in the [PocketStation platform guide](https://github.com/pocketstation-io/sdk-python/blob/main/docs/operations/platform-support.md).
+
+Run the conversion test without connecting to a room:
+
+```bash
+python -m unittest discover tests
+```

examples/desktop_audio/publish.py

@@ -0,0 +1,225 @@
+"""Publish desktop or microphone audio from PocketStation to LiveKit."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import os
+from typing import Protocol
+
+import numpy as np
+import pocketstation as pks
+import pocketstation.aio as pks_aio
+from livekit import api, rtc
+
+SAMPLE_RATE_HZ = 48_000
+CHANNEL_COUNT = 1
+FRAME_DURATION_MS = 10
+LIVEKIT_QUEUE_MS = 100
+LOGGER = logging.getLogger(__name__)
+
+
+class SessionState(Protocol):
+    @property
+    def is_stopped(self) -> bool: ...
+
+
+class LiveKitConnector(pks_aio.Connector):
+    """Publish PocketStation frames through one LiveKit AudioSource."""
+
+    def __init__(self, source: rtc.AudioSource) -> None:
+        self.source = source
+        self.frames_sent = 0
+        self.closed = False
+
+    async def send(self, frame: pks.AudioFrame) -> None:
+        try:
+            await self.source.capture_frame(to_livekit_frame(frame))
+        except Exception:
+            LOGGER.exception(
+                "LiveKit rejected a %d Hz, %d-channel audio frame",
+                frame.sample_rate_hz,
+                frame.channel_count,
+            )
+            raise
+        self.frames_sent += 1
+
+    async def stop(self) -> None:
+        if self.closed:
+            return
+        await self.source.wait_for_playout()
+        await self.source.aclose()
+        self.closed = True
+
+
+def to_livekit_frame(frame: pks.AudioFrame) -> rtc.AudioFrame:
+    """Downmix one PocketStation float32 frame to mono PCM16."""
+    channels = np.asarray(frame.samples, dtype=np.float32).reshape((-1, frame.channel_count))
+    mono = np.mean(channels, axis=1, dtype=np.float32)
+    pcm = np.rint(np.clip(mono, -1.0, 1.0) * 32_767).astype("<i2")
+    return rtc.AudioFrame(
+        data=pcm.tobytes(),
+        sample_rate=frame.sample_rate_hz,
+        num_channels=CHANNEL_COUNT,
+        samples_per_channel=frame.sample_count // frame.channel_count,
+    )
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Publish one local audio source to a LiveKit room")
+    source = parser.add_mutually_exclusive_group(required=True)
+    source.add_argument("--application", help="Running application name or identifier")
+    source.add_argument(
+        "--system-audio",
+        action="store_true",
+        help="Capture the complete desktop output mix",
+    )
+    source.add_argument(
+        "--microphone",
+        action="store_true",
+        help="Capture the default microphone without voice processing",
+    )
+    parser.add_argument("--room", default="desktop-audio", help="LiveKit room name")
+    parser.add_argument(
+        "--duration",
+        type=float,
+        default=0.0,
+        help="Stop after this many seconds; zero runs until interrupted",
+    )
+    args = parser.parse_args()
+    if args.duration < 0:
+        parser.error("--duration must be zero or greater")
+    return args
+
+
+def selected_source(
+    args: argparse.Namespace,
+) -> tuple[pks.Source, str, rtc.TrackSource.ValueType]:
+    if args.application is not None:
+        selected = (
+            pks.Source.application_process_id(int(args.application))
+            if args.application.isdecimal()
+            else pks.Source.application(args.application)
+        )
+        return (
+            selected,
+            "application-audio",
+            rtc.TrackSource.SOURCE_SCREENSHARE_AUDIO,
+        )
+    if args.system_audio:
+        return (
+            pks.Source.system_audio(),
+            "system-audio",
+            rtc.TrackSource.SOURCE_SCREENSHARE_AUDIO,
+        )
+    return (
+        pks.Source.microphone_default(),
+        "microphone",
+        rtc.TrackSource.SOURCE_MICROPHONE,
+    )
+
+
+def livekit_credentials() -> tuple[str, str, str]:
+    names = ("LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET")
+    values = tuple(os.getenv(name) for name in names)
+    missing = [name for name, value in zip(names, values, strict=True) if not value]
+    if missing:
+        raise RuntimeError(f"Set {', '.join(missing)} before running this example")
+    return (
+        os.environ["LIVEKIT_URL"],
+        os.environ["LIVEKIT_API_KEY"],
+        os.environ["LIVEKIT_API_SECRET"],
+    )
+
+
+async def wait_until_done(
+    disconnected: asyncio.Event,
+    running: SessionState,
+    *,
+    duration: float,
+) -> None:
+    """Return when the room disconnects, capture stops, or time expires."""
+
+    async def wait_for_session() -> None:
+        while not running.is_stopped:
+            await asyncio.sleep(0.05)
+
+    room_wait = asyncio.create_task(disconnected.wait())
+    session_wait = asyncio.create_task(wait_for_session())
+    try:
+        await asyncio.wait(
+            {room_wait, session_wait},
+            timeout=duration or None,
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+    finally:
+        for task in (room_wait, session_wait):
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(room_wait, session_wait, return_exceptions=True)
+
+
+async def run(args: argparse.Namespace) -> None:
+    url, api_key, api_secret = livekit_credentials()
+    source, track_name, track_source = selected_source(args)
+
+    room = rtc.Room()
+    livekit_source = rtc.AudioSource(
+        SAMPLE_RATE_HZ,
+        CHANNEL_COUNT,
+        queue_size_ms=LIVEKIT_QUEUE_MS,
+    )
+    connector = LiveKitConnector(livekit_source)
+    disconnected = asyncio.Event()
+
+    @room.on("disconnected")
+    def on_disconnected(*_: object) -> None:
+        disconnected.set()
+
+    session = pks_aio.Session(
+        sample_rate_hz=SAMPLE_RATE_HZ,
+        channels=CHANNEL_COUNT,
+        frame_duration_ms=FRAME_DURATION_MS,
+    )
+    session.capture(source).send_to(connector)
+
+    token = (
+        api.AccessToken(api_key, api_secret)
+        .with_identity("pocketstation-publisher")
+        .with_name("PocketStation Publisher")
+        .with_grants(api.VideoGrants(room_join=True, room=args.room))
+        .to_jwt()
+    )
+
+    try:
+        await room.connect(url, token)
+        track = rtc.LocalAudioTrack.create_audio_track(track_name, livekit_source)
+        options = rtc.TrackPublishOptions()
+        options.source = track_source
+        publication = await room.local_participant.publish_track(track, options)
+        LOGGER.info("published %s as track %s", track_name, publication.sid)
+
+        running = await session.start()
+        async with running:
+            await wait_until_done(disconnected, running, duration=args.duration)
+        result = running.stop_result
+        if result is None or not result.success:
+            raise RuntimeError("PocketStation stopped after a capture or delivery failure")
+    finally:
+        if not connector.closed:
+            await connector.stop()
+        await room.disconnect()
+        LOGGER.info("published %d audio frames", connector.frames_sent)
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+    asyncio.run(run(parse_args()))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        pass

examples/desktop_audio/requirements.txt

@@ -0,0 +1,3 @@
+livekit==1.1.17
+livekit-api==1.2.1
+pocketstation==0.1.4

examples/desktop_audio/tests/test_publish.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from typing import cast
+
+import pocketstation as pks
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from publish import to_livekit_frame, wait_until_done
+
+
+class AudioConversionTest(unittest.TestCase):
+    def test_stereo_float32_samples_become_mono_pcm16(self) -> None:
+        source = SimpleNamespace(
+            samples=memoryview(bytearray()),
+            sample_rate_hz=48_000,
+            channel_count=2,
+            sample_count=4,
+        )
+        source.samples = [-1.5, -0.5, 0.5, 1.5]
+
+        frame = to_livekit_frame(cast(pks.AudioFrame, source))
+
+        self.assertEqual(frame.sample_rate, 48_000)
+        self.assertEqual(frame.num_channels, 1)
+        self.assertEqual(frame.samples_per_channel, 2)
+        self.assertEqual(list(frame.data), [-32_767, 32_767])
+
+
+class ActiveWaitTest(unittest.IsolatedAsyncioTestCase):
+    async def test_returns_when_pocketstation_stops(self) -> None:
+        disconnected = asyncio.Event()
+        running = SimpleNamespace(is_stopped=False)
+
+        async def stop_capture() -> None:
+            await asyncio.sleep(0.01)
+            running.is_stopped = True
+
+        stop_task = asyncio.create_task(stop_capture())
+        await asyncio.wait_for(
+            wait_until_done(disconnected, running, duration=0.0),
+            timeout=0.2,
+        )
+        await stop_task
+
+        self.assertFalse(disconnected.is_set())
+
+
+if __name__ == "__main__":
+    unittest.main()