From b10350655c03bda7c8e6db116d59fe3ab844522e Mon Sep 17 00:00:00 2001 From: Rudy Date: Thu, 2 Jul 2026 13:04:45 +0200 Subject: [PATCH 1/6] feat(source): add source@v1 capture-and-stream client Add a 'sendspin source' command that runs a source client: capture audio from a local input (line-in/microphone) or a sine test tone, encode it (PCM, FLAC, or Opus via PyAV), and stream it to a server, which mixes and distributes it to players. The server controls when the source streams. - source_utils: PyAV SourceEncoder (exposes raw codec header) + RMS level - source_stream: SourceStreamer capture loop (sounddevice / sine), server start/stop handling, optional line-sense signal reporting - cli: 'source' subcommand and 'audio-devices inputs' listing - audio_devices: input-device enumeration - settings: SourceSettings persistence (settings-source.json) - README: source mode documentation --- README.md | 40 ++++++ sendspin/audio_devices.py | 32 +++++ sendspin/cli.py | 279 +++++++++++++++++++++++++++++++++++- sendspin/settings.py | 80 +++++++++++ sendspin/source_stream.py | 202 ++++++++++++++++++++++++++ sendspin/source_utils.py | 159 ++++++++++++++++++++ tests/test_source_stream.py | 120 ++++++++++++++++ tests/test_source_utils.py | 99 +++++++++++++ 8 files changed, 1010 insertions(+), 1 deletion(-) create mode 100644 sendspin/source_stream.py create mode 100644 sendspin/source_utils.py create mode 100644 tests/test_source_stream.py create mode 100644 tests/test_source_utils.py diff --git a/README.md b/README.md index c6049ed..2efe065 100644 --- a/README.md +++ b/README.md @@ -296,6 +296,46 @@ Use `--manufacturer` and `--product-name` to override the device identity report sendspin daemon --name "Living Room" --manufacturer "Acme" --product-name "Living Room Speaker" ``` +### Source Mode + +Run as a **source** client to capture audio from a local input (line-in, turntable, +Bluetooth receiver, or microphone) and stream it *into* Sendspin. The server mixes +and distributes it to players like any other audio, so an analog input on one machine +can play back synchronized across the whole group. + +```bash +# Capture the default input device and stream it to a discovered server +sendspin source + +# Pick a specific server, input device, and codec +sendspin source --url ws://192.168.1.50:8927/sendspin --device 2 --codec flac + +# Stream a 440 Hz sine test tone (no capture hardware needed) +sendspin source --input sine +``` + +List available capture devices: + +```bash +sendspin audio-devices inputs +``` + +Key options: + +- `--input {linein,sine}` — capture from a real input device, or generate a sine test tone. Defaults to `linein`; passing `--device` implies `linein`. +- `--device` — input device index or name (see `audio-devices inputs`). +- `--codec {pcm,opus,flac}` — codec used to encode captured audio before sending (default `pcm`). Capture is 16-bit. +- `--sample-rate` / `--channels` — capture format (default 48000 Hz, 2 channels). +- `--line-sense` — report input signal presence to the server via `client/state`; the server may use it to decide when to start/stop the source. + +The **server** decides when a source streams: a source stays idle until the server +sends a `start` command, and stops on `stop` or disconnect. A device may run both the +`source` and `player` roles; when it does, it never plays its captured input locally — +it only plays back what the server distributes, staying in sync with the group. + +Source-mode preferences (client id, last server, input/codec defaults) are persisted +to `~/.config/sendspin/settings-source.json`. + ### Hooks You can run external commands when audio streams start or stop. This is useful for controlling amplifiers, lighting, or other home automation: diff --git a/sendspin/audio_devices.py b/sendspin/audio_devices.py index a365188..45396b0 100644 --- a/sendspin/audio_devices.py +++ b/sendspin/audio_devices.py @@ -80,6 +80,38 @@ def query_devices() -> list[AudioDevice]: return result +@dataclass(slots=True) +class InputDevice: + """Represents an audio input (capture) device.""" + + index: int + name: str + input_channels: int + sample_rate: float + is_default: bool + + +def query_input_devices() -> list[InputDevice]: + """Query all available audio input (capture) devices.""" + devices = sounddevice.query_devices() + default_input = int(sounddevice.default.device[0]) + + result: list[InputDevice] = [] + for i in range(len(devices)): + dev = devices[i] + if dev["max_input_channels"] > 0: + result.append( + InputDevice( + index=i, + name=str(dev["name"]), + input_channels=int(dev["max_input_channels"]), + sample_rate=float(dev["default_samplerate"]), + is_default=(i == default_input), + ) + ) + return result + + def _check_format(device: AudioDevice, rate: int, channels: int, dtype: str) -> bool: """Check if a specific audio format is supported by the device.""" try: diff --git a/sendspin/cli.py b/sendspin/cli.py index 0bead5b..a1d3a24 100644 --- a/sendspin/cli.py +++ b/sendspin/cli.py @@ -27,9 +27,11 @@ from sendspin.volume_controller import VolumeController if TYPE_CHECKING: + from aiosendspin.client import SendspinClient from aiosendspin.models.player import SupportedAudioFormat from sendspin.audio_devices import AudioDevice + from sendspin.source_stream import SourceStreamer LOGGER = logging.getLogger(__name__) @@ -42,7 +44,7 @@ PLAYER_APP_SENTINEL = "player" EXPLICIT_APPS = frozenset( - {PLAYER_APP_SENTINEL, "daemon", "serve", "audio-devices", "servers", "clients"} + {PLAYER_APP_SENTINEL, "daemon", "serve", "source", "audio-devices", "servers", "clients"} ) TOP_LEVEL_ACTIONS = frozenset({"-h", "--help", "--version"}) @@ -139,6 +141,38 @@ def list_audio_devices() -> None: print(f" {name:<12} {description}") +def list_input_devices() -> None: + """List all available audio input (capture) devices.""" + try: + from sendspin.audio_devices import query_input_devices + except OSError as e: + if "PortAudio library not found" in str(e): + print(PORTAUDIO_NOT_FOUND_MESSAGE) + sys.exit(1) + raise + + try: + devices = query_input_devices() + except OSError as e: + if "PortAudio library not found" in str(e): + print(PORTAUDIO_NOT_FOUND_MESSAGE) + sys.exit(1) + raise + + print("Available audio input devices:") + print() + for device in devices: + default_marker = " (default)" if device.is_default else "" + print( + f" [{device.index}] {device.name}{default_marker}\n" + f" Channels: {device.input_channels}, " + f"Sample rate: {device.sample_rate} Hz" + ) + if devices: + default = next((d for d in devices if d.is_default), devices[0]) + print(f"\nTo capture from an input device:\n sendspin source --device {default.index}") + + def _add_player_runtime_options(target: ArgumentTarget, *, suppress_defaults: bool = False) -> None: """Add the interactive player's runtime options.""" default: str | float | None @@ -274,6 +308,92 @@ def _add_player_actions(target: ArgumentTarget, *, suppress_defaults: bool = Fal ) +def _add_source_parser(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) -> None: + """Add the ``source`` app parser (capture a local input into Sendspin).""" + source_parser = subparsers.add_parser( + "source", + help="Capture a local audio input and stream it to a server", + description=( + "Run as a Sendspin source client: capture audio from a local input " + "(line-in/microphone, or a synthetic sine test tone) and stream it to a " + "server, which mixes and distributes it to players. The server decides " + "when the source starts and stops streaming." + ), + ) + source_parser.add_argument( + "--url", + default=None, + help="WebSocket URL of the server. If omitted, the first discovered server is used.", + ) + source_parser.add_argument("--name", default=None, help="Friendly name for this source client") + source_parser.add_argument( + "--id", default=None, help="Unique identifier for this source client" + ) + source_parser.add_argument( + "--input", + dest="source_input", + choices=["sine", "linein"], + default=None, + help="Capture source: 'linein' (real input device) or 'sine' (test tone)", + ) + source_parser.add_argument( + "--device", + dest="source_device", + default=None, + help="Input device index or name (see 'sendspin audio-devices inputs')", + ) + source_parser.add_argument( + "--codec", + dest="source_codec", + choices=["pcm", "opus", "flac"], + default=None, + help="Codec to encode captured audio with (default: pcm)", + ) + source_parser.add_argument( + "--sample-rate", + dest="source_sample_rate", + type=int, + default=None, + help="Capture sample rate in Hz", + ) + source_parser.add_argument( + "--channels", dest="source_channels", type=int, default=None, help="Capture channel count" + ) + source_parser.add_argument( + "--frame-ms", dest="source_frame_ms", type=int, default=20, help="Capture frame size in ms" + ) + source_parser.add_argument( + "--sine-hz", + dest="source_sine_hz", + type=float, + default=440.0, + help="Sine test-tone frequency", + ) + source_parser.add_argument( + "--signal-threshold-db", + dest="source_signal_threshold_db", + type=float, + default=-50.0, + help="RMS threshold (dBFS) for line-sense signal detection", + ) + source_parser.add_argument( + "--line-sense", + action="store_true", + help="Report line-sensing signal presence to the server via client/state", + ) + source_parser.add_argument( + "--settings-dir", + default=None, + help="Directory to store settings (default: ~/.config/sendspin)", + ) + source_parser.add_argument( + "--log-level", + default=None, + choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], + help="Logging level to use (default: INFO)", + ) + + def _build_parser() -> argparse.ArgumentParser: """Build the top-level CLI parser.""" parser = argparse.ArgumentParser( @@ -483,6 +603,9 @@ def _build_parser() -> argparse.ArgumentParser: ), ) + # Source subcommand + _add_source_parser(subparsers) + # audio-devices subcommand audio_devices_parser = subparsers.add_parser( "audio-devices", @@ -499,6 +622,11 @@ def _build_parser() -> argparse.ArgumentParser: help="List available audio output devices", description="List all available audio output devices and exit.", ) + audio_devices_sub.add_parser( + "inputs", + help="List available audio input (capture) devices", + description="List all available audio input (capture) devices and exit.", + ) # servers subcommand servers_parser = subparsers.add_parser( @@ -693,6 +821,137 @@ async def _run_serve_mode(args: argparse.Namespace) -> int: return await run_server(serve_config) +async def _discover_first_server_url() -> str | None: + """Discover Sendspin servers and return the first URL, or None if none found.""" + from sendspin.discovery import discover_servers + + servers = await discover_servers(discovery_time=3.0) + if not servers: + return None + return servers[0].url + + +async def _source_connection_loop( + client: SendspinClient, + url: str, + streamer: SourceStreamer, +) -> None: + """Connect to the server with reconnect, resetting streaming on each drop.""" + from aiohttp import ClientError + + error_backoff = 1.0 + max_backoff = 300.0 + while True: + try: + await client.connect(url) + error_backoff = 1.0 + disconnect_event: asyncio.Event = asyncio.Event() + unsubscribe = client.add_disconnect_listener(disconnect_event.set) + await disconnect_event.wait() + unsubscribe() + streamer.reset() + LOGGER.info("Disconnected from server; reconnecting to %s", url) + except (TimeoutError, OSError, ClientError) as e: + LOGGER.warning( + "Connection error (%s), retrying in %.0fs", type(e).__name__, error_backoff + ) + await asyncio.sleep(error_backoff) + error_backoff = min(error_backoff * 2, max_backoff) + + +async def _run_source_mode(args: argparse.Namespace) -> int: + """Run as a source client: capture a local input and stream it to a server.""" + from aiosendspin.client import SendspinClient as _SendspinClient + from aiosendspin.models.source import ( + ClientHelloSourceSupport, + SourceFeatures, + SourceSupportedFormat, + ) + from aiosendspin.models.types import AudioCodec, Roles + + from sendspin.settings import get_source_settings + from sendspin.source_stream import SourceStreamConfig, SourceStreamer + + settings = await get_source_settings(args.settings_dir) + + url = args.url or settings.last_server_url + input_kind = args.source_input or settings.source_input + device = args.source_device or settings.source_device + codec_str = args.source_codec or settings.source_codec + sample_rate = args.source_sample_rate or settings.source_sample_rate + channels = args.source_channels or settings.source_channels + log_level = args.log_level or settings.log_level or "INFO" + logging.basicConfig(level=getattr(logging, log_level)) + + # A device implies real line-in capture unless the user asked for the sine tone. + if device is not None and args.source_input is None: + input_kind = "linein" + + if url is None: + LOGGER.info("No --url given; discovering servers...") + url = await _discover_first_server_url() + if url is None: + print("No Sendspin server found. Provide --url or start a server.") + return 1 + print(f"Using discovered server: {url}") + + client_id, client_name = _resolve_client_info(args.id or settings.client_id, args.name) + codec = AudioCodec(codec_str) + + config = SourceStreamConfig( + codec=codec, + input_kind=input_kind, + device=device, + sample_rate=sample_rate, + channels=channels, + frame_ms=args.source_frame_ms, + sine_hz=args.source_sine_hz, + signal_threshold_db=args.source_signal_threshold_db, + line_sense=args.line_sense, + ) + support = ClientHelloSourceSupport( + supported_formats=[ + SourceSupportedFormat( + codec=codec, channels=channels, sample_rate=sample_rate, bit_depth=16 + ) + ], + features=SourceFeatures(line_sense=args.line_sense), + ) + client = _SendspinClient( + client_id=client_id, + client_name=client_name, + roles=[Roles.SOURCE], + source_support=support, + ) + streamer = SourceStreamer(client, config) + client.add_source_command_listener(streamer.handle_source_command) + + settings.update( + client_id=client_id, + name=client_name, + last_server_url=url, + source_input=input_kind, + source_device=device, + source_codec=codec_str, + source_sample_rate=sample_rate, + source_channels=channels, + ) + + LOGGER.info("Source client '%s' -> %s (%s, %s)", client_id, url, input_kind, codec.value) + capture_task = asyncio.create_task(streamer.run()) + try: + await _source_connection_loop(client, url, streamer) + finally: + capture_task.cancel() + try: + await capture_task + except asyncio.CancelledError: + pass + await client.disconnect() + await settings.flush() + return 0 + + async def _run_daemon_mode( args: argparse.Namespace, settings: ClientSettings, @@ -743,11 +1002,29 @@ def main() -> int: traceback.print_exc() return 1 + # Handle source subcommand + if args.command == "source": + try: + return asyncio.run(_run_source_mode(args)) + except KeyboardInterrupt: + return 0 + except CLIError as e: + print(f"Error: {e}") + return e.exit_code + except OSError as e: + if "PortAudio library not found" in str(e): + print(PORTAUDIO_NOT_FOUND_MESSAGE) + return 1 + raise + # Handle utility subcommands if args.command == "audio-devices": if args.audio_devices_command == "list": list_audio_devices() return 0 + if args.audio_devices_command == "inputs": + list_input_devices() + return 0 if args.command == "servers": if args.servers_command == "list": diff --git a/sendspin/settings.py b/sendspin/settings.py index cb2f669..0a2098a 100644 --- a/sendspin/settings.py +++ b/sendspin/settings.py @@ -295,6 +295,86 @@ def _load(self) -> bool: return False +@dataclass +class SourceSettings(BaseSettings): + """Settings for source mode (capturing a local input into Sendspin).""" + + client_id: str | None = None + last_server_url: str | None = None + source_input: str = "linein" + source_device: str | None = None + source_codec: str = "pcm" + source_sample_rate: int = 48000 + source_channels: int = 2 + + def update( + self, + *, + name: str | None = None, + log_level: str | None = None, + client_id: str | None = None, + last_server_url: str | None = None, + source_input: str | None = None, + source_device: str | None = None, + source_codec: str | None = None, + source_sample_rate: int | None = None, + source_channels: int | None = None, + ) -> None: + """Update settings fields. Only changed fields trigger a save.""" + changed = self._update_fields( + { + "name": name, + "log_level": log_level, + "client_id": client_id, + "last_server_url": last_server_url, + "source_input": source_input, + "source_device": source_device, + "source_codec": source_codec, + "source_sample_rate": source_sample_rate, + "source_channels": source_channels, + } + ) + if changed: + self._schedule_save() + + def _load(self) -> bool: + """Load settings from the settings file (blocking I/O).""" + if self._settings_file is None or not self._settings_file.exists(): + logger.debug("Settings file does not exist: %s", self._settings_file) + return False + + try: + data = json.loads(self._settings_file.read_text()) + self.name = data.get("name") + self.log_level = data.get("log_level") + self.client_id = data.get("client_id") + self.last_server_url = data.get("last_server_url") + self.source_input = data.get("source_input", "linein") + self.source_device = data.get("source_device") + self.source_codec = data.get("source_codec", "pcm") + self.source_sample_rate = data.get("source_sample_rate", 48000) + self.source_channels = data.get("source_channels", 2) + logger.info("Loaded settings from %s", self._settings_file) + except (json.JSONDecodeError, OSError) as e: + logger.warning("Failed to load settings from %s: %s", self._settings_file, e) + return False + + +async def get_source_settings(config_dir: str | None = None) -> SourceSettings: + """Create and load source-mode settings. + + Args: + config_dir: Optional directory to store settings. Defaults to ~/.config/sendspin. + + Returns: + SourceSettings instance with settings loaded from disk. + """ + config_path = Path(config_dir) if config_dir else Path.home() / ".config" / "sendspin" + settings = SourceSettings(_settings_file=config_path / "settings-source.json") + await settings.load() + return settings + + async def get_client_settings( mode: Literal["tui", "daemon"], config_dir: str | None = None ) -> ClientSettings: diff --git a/sendspin/source_stream.py b/sendspin/source_stream.py new file mode 100644 index 0000000..8c85929 --- /dev/null +++ b/sendspin/source_stream.py @@ -0,0 +1,202 @@ +"""Audio capture and streaming for the Sendspin source role. + +``SourceStreamer`` captures 16-bit PCM from a local input (a synthetic sine test +tone or a real line-in/microphone via ``sounddevice``), encodes it with +``SourceEncoder``, and streams timestamped frames to the server. The server is +the sole initiator of streaming: capture flows to the server only after a +``server/command`` ``start`` and stops on ``stop`` (or disconnect). +""" + +from __future__ import annotations + +import asyncio +import logging +import math +import struct +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from aiosendspin.models.source import ClientStreamStartSource +from aiosendspin.models.types import AudioCodec, SourceCommand, SourceSignal + +from sendspin.source_utils import SourceEncoder, calc_level +from sendspin.utils import create_task + +if TYPE_CHECKING: + from aiosendspin.client import SendspinClient + from aiosendspin.models.source import SourceCommandPayload + +logger = logging.getLogger(__name__) + +_SINE_AMPLITUDE = 0.3 + + +@dataclass(slots=True) +class SourceStreamConfig: + """Configuration for a source capture session.""" + + codec: AudioCodec + input_kind: str # "sine" | "linein" + device: str | None + sample_rate: int + channels: int + frame_ms: int + sine_hz: float + signal_threshold_db: float + line_sense: bool + + @property + def samples_per_frame(self) -> int: + """Number of samples per captured frame.""" + return max(1, self.sample_rate * self.frame_ms // 1000) + + +class SourceStreamer: + """Captures audio and streams it to the server when the server requests it.""" + + def __init__(self, client: SendspinClient, config: SourceStreamConfig) -> None: + """Initialize the streamer for a client and capture configuration.""" + self._client = client + self._config = config + self._streaming = asyncio.Event() + self._encoder: SourceEncoder | None = None + self._last_signal: SourceSignal | None = None + + async def run(self) -> None: + """Run the capture loop until cancelled. + + Captured audio flows to the server only while streaming is active (after a + server ``start`` command); see :meth:`handle_source_command`. + """ + if self._config.input_kind == "sine": + await self._stream_sine() + else: + await self._stream_linein() + + @property + def streaming(self) -> bool: + """Whether the source is currently streaming to the server.""" + return self._streaming.is_set() + + def handle_source_command(self, payload: SourceCommandPayload) -> None: + """React to a server start/stop command.""" + if payload.command == SourceCommand.START: + create_task(self._begin_stream()) + elif payload.command == SourceCommand.STOP: + create_task(self._end_stream()) + + def reset(self) -> None: + """Clear streaming state (e.g., on disconnect).""" + self._streaming.clear() + self._encoder = None + self._last_signal = None + + async def _begin_stream(self) -> None: + if self._streaming.is_set(): + return + cfg = self._config + encoder = SourceEncoder( + codec=cfg.codec, + channels=cfg.channels, + sample_rate=cfg.sample_rate, + frame_samples=cfg.samples_per_frame, + ) + self._encoder = encoder + await self._client.send_client_stream_start( + ClientStreamStartSource( + codec=cfg.codec, + channels=cfg.channels, + sample_rate=cfg.sample_rate, + bit_depth=16, + codec_header=encoder.codec_header, + ) + ) + self._streaming.set() + logger.info("Source streaming started (%s, %d Hz)", cfg.codec.value, cfg.sample_rate) + + async def _end_stream(self) -> None: + if not self._streaming.is_set(): + return + self._streaming.clear() + if self._encoder is not None: + for tail in self._encoder.flush(): + await self._client.send_source_audio_chunk( + tail, capture_timestamp_us=self._client.now_us() + ) + self._encoder = None + await self._client.send_client_stream_end() + logger.info("Source streaming stopped") + + async def _send_frame(self, pcm: bytes) -> None: + """Report signal (if line sensing) and stream the frame when active.""" + if self._config.line_sense: + self._maybe_report_signal(pcm) + if not self._streaming.is_set() or self._encoder is None: + return + capture_us = self._client.now_us() + for encoded, frame_us in self._encoder.encode(pcm, capture_us): + await self._client.send_source_audio_chunk(encoded, capture_timestamp_us=frame_us) + + def _maybe_report_signal(self, pcm: bytes) -> None: + level = calc_level(pcm) + threshold = 10 ** (self._config.signal_threshold_db / 20) + signal = SourceSignal.PRESENT if level >= threshold else SourceSignal.ABSENT + if signal != self._last_signal: + self._last_signal = signal + create_task(self._client.send_source_state(signal=signal)) + + async def _stream_sine(self) -> None: + cfg = self._config + samples = cfg.samples_per_frame + phase = 0.0 + increment = 2 * math.pi * cfg.sine_hz / cfg.sample_rate + frame_seconds = cfg.frame_ms / 1000 + while True: + buffer = bytearray() + for _ in range(samples): + value = int(_SINE_AMPLITUDE * math.sin(phase) * 32767) + phase += increment + buffer.extend(struct.pack(" None: + import sounddevice # noqa: PLC0415 + + cfg = self._config + loop = asyncio.get_running_loop() + queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=32) + + def _callback(indata: object, _frames: int, _time: object, status: object) -> None: + if status: + logger.debug("Input stream status: %s", status) + data = bytes(indata) # type: ignore[call-overload] + try: + loop.call_soon_threadsafe(queue.put_nowait, data) + except asyncio.QueueFull: + logger.debug("Source capture queue full; dropping frame") + + device = cfg.device if cfg.device is None else _resolve_device(cfg.device) + stream = sounddevice.RawInputStream( + samplerate=cfg.sample_rate, + channels=cfg.channels, + dtype="int16", + blocksize=cfg.samples_per_frame, + device=device, + callback=_callback, + ) + with stream: + logger.info( + "Capturing from input device %s (%d Hz, %d ch)", + cfg.device or "default", + cfg.sample_rate, + cfg.channels, + ) + while True: + data = await queue.get() + await self._send_frame(data) + + +def _resolve_device(device: str) -> int | str: + """Resolve an input device argument to a sounddevice identifier.""" + return int(device) if device.isnumeric() else device diff --git a/sendspin/source_utils.py b/sendspin/source_utils.py new file mode 100644 index 0000000..d1000d6 --- /dev/null +++ b/sendspin/source_utils.py @@ -0,0 +1,159 @@ +"""Encoding and signal helpers for the Sendspin source role. + +A source client captures 16-bit PCM from a local input and streams it to the +server. ``SourceEncoder`` encodes that PCM to the negotiated codec (PCM +passthrough, FLAC, or Opus) using PyAV, exposing the raw codec header so the +server can initialize its decoder. ``calc_level`` provides a simple RMS level +used for optional line sensing. +""" + +from __future__ import annotations + +import array +import base64 +import logging +import sys +from fractions import Fraction +from types import ModuleType +from typing import TYPE_CHECKING + +from aiosendspin.models.types import AudioCodec + +if TYPE_CHECKING: + import av + +logger = logging.getLogger(__name__) + +# Source capture is fixed at 16-bit signed PCM (matches sounddevice int16 capture +# and keeps codec init simple). The server can still receive other depths from +# other source implementations. +BYTES_PER_SAMPLE = 2 +_MAX_INT16 = 32767.0 + + +def calc_level(pcm: bytes) -> float: + """Return a normalized RMS level (0.0-1.0) for 16-bit interleaved PCM.""" + if not pcm: + return 0.0 + samples = array.array("h") + samples.frombytes(pcm[: len(pcm) - (len(pcm) % 2)]) + if sys.byteorder != "little": + samples.byteswap() + if not samples: + return 0.0 + total = 0.0 + for sample in samples: + norm = sample / _MAX_INT16 + total += norm * norm + rms = (total / len(samples)) ** 0.5 + return float(min(1.0, rms)) + + +class SourceEncoder: + """Encode 16-bit interleaved PCM to the negotiated source codec via PyAV.""" + + def __init__( + self, + *, + codec: AudioCodec, + channels: int, + sample_rate: int, + frame_samples: int, + ) -> None: + """Create an encoder. + + Args: + codec: Target codec (pcm, flac, or opus). + channels: Number of channels. + sample_rate: Sample rate in Hz. + frame_samples: Preferred samples per frame (used for PCM and as a + fallback when the codec does not report its own frame size). + """ + self._codec = codec + self._channels = channels + self._sample_rate = sample_rate + self._stride = BYTES_PER_SAMPLE * channels + self._layout = "mono" if channels == 1 else "stereo" + self._buffer = bytearray() + self._buffer_head_us: int | None = None + self._pts = 0 + self._encoder: av.AudioCodecContext | None = None + self._codec_header: str | None = None + self._frame_samples = frame_samples + + if codec == AudioCodec.PCM: + return + + av_mod = _get_av() + codec_name = "libopus" if codec == AudioCodec.OPUS else "flac" + encoder = av_mod.AudioCodecContext.create(codec_name, "w") + encoder.sample_rate = sample_rate + encoder.format = "s16" + encoder.layout = self._layout + with av_mod.logging.Capture(): + encoder.open() + if encoder.frame_size: + self._frame_samples = encoder.frame_size + if encoder.extradata: + self._codec_header = base64.b64encode(bytes(encoder.extradata)).decode("ascii") + self._encoder = encoder + + @property + def codec_header(self) -> str | None: + """Base64 raw codec header (extradata) for client_stream/start, if any.""" + return self._codec_header + + @property + def frame_samples(self) -> int: + """Preferred number of samples per captured frame.""" + return self._frame_samples + + def encode(self, pcm: bytes, capture_timestamp_us: int) -> list[tuple[bytes, int]]: + """Encode captured PCM into ``(frame_bytes, capture_timestamp_us)`` pairs. + + For PCM the input passes through unchanged. For FLAC/Opus the PCM is + buffered and emitted in codec-sized frames; each emitted frame is stamped + with the capture time of its first sample. + """ + if self._encoder is None: + return [(pcm, capture_timestamp_us)] if pcm else [] + + if not self._buffer: + self._buffer_head_us = capture_timestamp_us + self._buffer.extend(pcm) + + results: list[tuple[bytes, int]] = [] + chunk_size = self._frame_samples * self._stride + while len(self._buffer) >= chunk_size: + block = bytes(self._buffer[:chunk_size]) + del self._buffer[:chunk_size] + assert self._buffer_head_us is not None + frame_ts = self._buffer_head_us + self._buffer_head_us += round(self._frame_samples * 1_000_000 / self._sample_rate) + for encoded in self._encode_block(block): + results.append((encoded, frame_ts)) + return results + + def flush(self) -> list[bytes]: + """Flush the codec, returning any trailing frames.""" + if self._encoder is None: + return [] + return [data for packet in self._encoder.encode(None) if (data := bytes(packet))] + + def _encode_block(self, block: bytes) -> list[bytes]: + assert self._encoder is not None + av_mod = _get_av() + frame = av_mod.AudioFrame(format="s16", layout=self._layout, samples=self._frame_samples) + frame.sample_rate = self._sample_rate + frame.pts = self._pts + frame.time_base = Fraction(1, self._sample_rate) + self._pts += self._frame_samples + frame.planes[0].update(block) + return [data for packet in self._encoder.encode(frame) if (data := bytes(packet))] + + +def _get_av() -> ModuleType: + """Import PyAV lazily so non-source commands do not require it eagerly.""" + import av # noqa: PLC0415 + + return av diff --git a/tests/test_source_stream.py b/tests/test_source_stream.py new file mode 100644 index 0000000..705b697 --- /dev/null +++ b/tests/test_source_stream.py @@ -0,0 +1,120 @@ +"""Tests for the source streamer command handling and framing.""" + +from __future__ import annotations + +import asyncio + +from aiosendspin.models.source import ClientStreamStartSource, SourceCommandPayload +from aiosendspin.models.types import AudioCodec, SourceCommand, SourceSignal + +from sendspin.source_stream import SourceStreamConfig, SourceStreamer + + +class _FakeClient: + """Records the source-related calls a SourceStreamer makes.""" + + def __init__(self) -> None: + self.stream_starts: list[ClientStreamStartSource] = [] + self.stream_ends = 0 + self.chunks: list[tuple[bytes, int]] = [] + self.signals: list[SourceSignal | None] = [] + self._clock = 1_000_000 + + def now_us(self) -> int: + self._clock += 1000 + return self._clock + + async def send_client_stream_start(self, source: ClientStreamStartSource) -> None: + self.stream_starts.append(source) + + async def send_client_stream_end(self) -> None: + self.stream_ends += 1 + + async def send_source_audio_chunk(self, data: bytes, *, capture_timestamp_us: int) -> bool: + self.chunks.append((data, capture_timestamp_us)) + return True + + async def send_source_state(self, *, signal: SourceSignal | None = None) -> None: + self.signals.append(signal) + + +def _config(*, codec: AudioCodec = AudioCodec.PCM, line_sense: bool = False) -> SourceStreamConfig: + return SourceStreamConfig( + codec=codec, + input_kind="sine", + device=None, + sample_rate=48000, + channels=2, + frame_ms=20, + sine_hz=440.0, + signal_threshold_db=-50.0, + line_sense=line_sense, + ) + + +def _make() -> tuple[SourceStreamer, _FakeClient]: + client = _FakeClient() + return SourceStreamer(client, _config()), client # type: ignore[arg-type] + + +async def test_begin_stream_announces_format_and_starts() -> None: + """Beginning a stream sends client_stream/start and marks streaming active.""" + streamer, client = _make() + await streamer._begin_stream() # noqa: SLF001 + assert len(client.stream_starts) == 1 + assert client.stream_starts[0].codec == AudioCodec.PCM + assert streamer._streaming.is_set() # noqa: SLF001 + + +async def test_end_stream_sends_end_and_stops() -> None: + """Ending a stream sends client_stream/end and clears streaming.""" + streamer, client = _make() + await streamer._begin_stream() # noqa: SLF001 + await streamer._end_stream() # noqa: SLF001 + assert client.stream_ends == 1 + assert not streamer._streaming.is_set() # noqa: SLF001 + + +async def test_send_frame_streams_only_when_active() -> None: + """Frames are streamed only after the stream has begun.""" + streamer, client = _make() + pcm = b"\x01\x02\x03\x04" * 16 + + await streamer._send_frame(pcm) # noqa: SLF001 (not started yet) + assert client.chunks == [] + + await streamer._begin_stream() # noqa: SLF001 + await streamer._send_frame(pcm) # noqa: SLF001 + assert len(client.chunks) == 1 + assert client.chunks[0][0] == pcm # PCM passthrough + + +async def test_line_sense_reports_signal_changes() -> None: + """With line sensing enabled, signal presence changes are reported once.""" + client = _FakeClient() + streamer = SourceStreamer(client, _config(line_sense=True)) # type: ignore[arg-type] + + loud = b"\x00\x40" * 64 # non-trivial amplitude + silence = b"\x00\x00" * 64 + + streamer._maybe_report_signal(loud) # noqa: SLF001 + streamer._maybe_report_signal(loud) # no change -> not re-reported + streamer._maybe_report_signal(silence) # noqa: SLF001 + await asyncio.sleep(0.05) # let the scheduled send_source_state tasks run + + assert client.signals == [SourceSignal.PRESENT, SourceSignal.ABSENT] + + +async def test_handle_source_command_dispatches_start_stop() -> None: + """A server start command begins streaming; a stop command ends it.""" + streamer, client = _make() + + streamer.handle_source_command(SourceCommandPayload(command=SourceCommand.START)) + await asyncio.sleep(0.05) + assert streamer._streaming.is_set() # noqa: SLF001 + assert len(client.stream_starts) == 1 + + streamer.handle_source_command(SourceCommandPayload(command=SourceCommand.STOP)) + await asyncio.sleep(0.05) + assert not streamer._streaming.is_set() # noqa: SLF001 + assert client.stream_ends == 1 diff --git a/tests/test_source_utils.py b/tests/test_source_utils.py new file mode 100644 index 0000000..0a42891 --- /dev/null +++ b/tests/test_source_utils.py @@ -0,0 +1,99 @@ +"""Tests for source encoding and signal helpers.""" + +from __future__ import annotations + +import math +import struct + +import pytest +from aiosendspin.models.source import ClientStreamStartSource +from aiosendspin.models.types import AudioCodec +from aiosendspin.server.roles.source.group import SourceDecoder + +from sendspin.source_utils import SourceEncoder, calc_level + +RATE = 48000 +CHANNELS = 2 +STRIDE = 2 * CHANNELS + + +def _sine_pcm(duration_ms: int, freq: float = 440.0) -> bytes: + samples = RATE * duration_ms // 1000 + buffer = bytearray() + for i in range(samples): + value = int(0.3 * math.sin(2 * math.pi * freq * i / RATE) * 32767) + buffer.extend(struct.pack(" None: + """Silence has zero level; empty input is safe.""" + assert calc_level(b"") == 0.0 + assert calc_level(b"\x00\x00" * 100) == 0.0 + + +def test_calc_level_signal_is_positive() -> None: + """A real signal produces a positive normalized level.""" + assert 0.0 < calc_level(_sine_pcm(20)) <= 1.0 + + +def test_pcm_encoder_passes_through() -> None: + """PCM encoding passes bytes through unchanged with no header.""" + encoder = SourceEncoder( + codec=AudioCodec.PCM, channels=CHANNELS, sample_rate=RATE, frame_samples=960 + ) + assert encoder.codec_header is None + pcm = _sine_pcm(20) + assert encoder.encode(pcm, 5000) == [(pcm, 5000)] + assert encoder.encode(b"", 6000) == [] + + +@pytest.mark.parametrize("codec", [AudioCodec.FLAC, AudioCodec.OPUS]) +def test_compressed_encoder_produces_header_and_frames(codec: AudioCodec) -> None: + """FLAC/Opus encoding yields a codec header and encoded frames.""" + encoder = SourceEncoder( + codec=codec, channels=CHANNELS, sample_rate=RATE, frame_samples=RATE * 20 // 1000 + ) + assert encoder.codec_header is not None + frames = encoder.encode(_sine_pcm(200), 1_000_000) + frames.extend((tail, 0) for tail in encoder.flush()) + assert frames + assert all(isinstance(data, bytes) and data for data, _ in frames) + + +@pytest.mark.parametrize("codec", [AudioCodec.PCM, AudioCodec.FLAC, AudioCodec.OPUS]) +def test_encode_round_trips_through_server_decoder(codec: AudioCodec) -> None: + """Frames encoded by the CLI decode back to PCM on the server side.""" + pcm = _sine_pcm(500) + encoder = SourceEncoder( + codec=codec, channels=CHANNELS, sample_rate=RATE, frame_samples=RATE * 20 // 1000 + ) + frame_bytes = RATE * 20 // 1000 * STRIDE + encoded: list[bytes] = [] + offset = 0 + ts = 1_000_000 + while offset < len(pcm): + block = pcm[offset : offset + frame_bytes] + offset += frame_bytes + encoded.extend(data for data, _ in encoder.encode(block, ts)) + ts += 20_000 + encoded.extend(encoder.flush()) + + decoder = SourceDecoder( + ClientStreamStartSource( + codec=codec, + channels=CHANNELS, + sample_rate=RATE, + bit_depth=16, + codec_header=encoder.codec_header, + ) + ) + decoded = bytearray() + for frame in encoded: + for chunk in decoder.decode(frame): + decoded += chunk + + in_samples = len(pcm) // STRIDE + out_samples = len(decoded) // STRIDE + # PCM is exact; FLAC/Opus carry small codec delay/padding, so allow a tolerance. + assert out_samples == pytest.approx(in_samples, abs=RATE // 10) From 05679c2418cefd223ed078c544ba5ba51bd60eeb Mon Sep 17 00:00:00 2001 From: Rudy Date: Fri, 7 Aug 2026 13:40:55 +0000 Subject: [PATCH 2/6] feat(source): drop client/hello format negotiation for sources Follow aiosendspin dropping supported_formats from source@v1_support (spec#113). A source announces its capture format per stream in client_stream/start, so advertising it up front negotiated nothing and the field being required made a conforming hello fail to parse. Also spell out in the README that starting a source is a policy the server application makes: the library no longer starts one on its own, so a source pointed at a server that never asks for audio stays idle by design. --- README.md | 9 ++++++--- sendspin/cli.py | 12 +++--------- 2 files changed, 9 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 2efe065..017a10d 100644 --- a/README.md +++ b/README.md @@ -329,9 +329,12 @@ Key options: - `--line-sense` — report input signal presence to the server via `client/state`; the server may use it to decide when to start/stop the source. The **server** decides when a source streams: a source stays idle until the server -sends a `start` command, and stops on `stop` or disconnect. A device may run both the -`source` and `player` roles; when it does, it never plays its captured input locally — -it only plays back what the server distributes, staying in sync with the group. +sends a `start` command, and stops on `stop` or disconnect. This is a policy the +server application makes for itself — the `aiosendspin` library never starts a +source on its own — so a source connected to a server that does not ask for audio +stays idle by design. A device may run both the `source` and `player` roles; when +it does, it never plays its captured input locally — it only plays back what the +server distributes, staying in sync with the group. Source-mode preferences (client id, last server, input/codec defaults) are persisted to `~/.config/sendspin/settings-source.json`. diff --git a/sendspin/cli.py b/sendspin/cli.py index a1d3a24..d79aae0 100644 --- a/sendspin/cli.py +++ b/sendspin/cli.py @@ -865,7 +865,6 @@ async def _run_source_mode(args: argparse.Namespace) -> int: from aiosendspin.models.source import ( ClientHelloSourceSupport, SourceFeatures, - SourceSupportedFormat, ) from aiosendspin.models.types import AudioCodec, Roles @@ -909,14 +908,9 @@ async def _run_source_mode(args: argparse.Namespace) -> int: signal_threshold_db=args.source_signal_threshold_db, line_sense=args.line_sense, ) - support = ClientHelloSourceSupport( - supported_formats=[ - SourceSupportedFormat( - codec=codec, channels=channels, sample_rate=sample_rate, bit_depth=16 - ) - ], - features=SourceFeatures(line_sense=args.line_sense), - ) + # The capture format is announced per stream in client_stream/start; there is + # no format negotiation in client/hello. + support = ClientHelloSourceSupport(features=SourceFeatures(line_sense=args.line_sense)) client = _SendspinClient( client_id=client_id, client_name=client_name, From 1898248820e60ef8ec60a69fd844573876df5df9 Mon Sep 17 00:00:00 2001 From: Jorge Pulido Date: Tue, 18 Aug 2026 08:55:53 +0200 Subject: [PATCH 3/6] Bump aiosendspin to 9.1.0 --- pyproject.toml | 2 +- sendspin/audio_connector.py | 4 +- sendspin/cli.py | 41 +++++++-- sendspin/daemon/daemon.py | 136 +++-------------------------- sendspin/serve/__init__.py | 9 +- sendspin/serve/worker.py | 8 +- sendspin/settings.py | 30 +++++++ sendspin/source_stream.py | 102 +++++++++++----------- sendspin/source_utils.py | 132 +--------------------------- sendspin/tui/app.py | 34 +++----- tests/daemon/test_daemon.py | 4 + tests/test_source_stream.py | 92 ++++++++++++------- tests/test_source_utils.py | 70 +-------------- tests/tui/test_role_negotiation.py | 69 --------------- tests/tui/test_volume_state.py | 4 + uv.lock | 91 +++++++++++++++++-- 16 files changed, 308 insertions(+), 520 deletions(-) delete mode 100644 tests/tui/test_role_negotiation.py diff --git a/pyproject.toml b/pyproject.toml index 094f33a..d37b23c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", ] dependencies = [ - "aiosendspin[server]~=6.0.1", + "aiosendspin[server,source]~=9.1.0", "aiosendspin-mpris~=2.1.1", "av>=15.0.0", "numpy>=1.26.0", diff --git a/sendspin/audio_connector.py b/sendspin/audio_connector.py index fd82602..84b078f 100644 --- a/sendspin/audio_connector.py +++ b/sendspin/audio_connector.py @@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, cast from aiosendspin.models.core import StreamStartMessage -from aiosendspin.models.types import AudioCodec, ClientStateType +from aiosendspin.models.types import AudioCodec from sendspin.audio import AudioPlayer from sendspin.audio_devices import AudioDevice @@ -454,7 +454,7 @@ def send_player_volume(self) -> None: if self._client is not None and self._client.connected: create_task( self._client.send_player_state( - state=ClientStateType.SYNCHRONIZED, + available=True, volume=self._volume, muted=self._muted, ) diff --git a/sendspin/cli.py b/sendspin/cli.py index d79aae0..5e0f69e 100644 --- a/sendspin/cli.py +++ b/sendspin/cli.py @@ -849,7 +849,7 @@ async def _source_connection_loop( unsubscribe = client.add_disconnect_listener(disconnect_event.set) await disconnect_event.wait() unsubscribe() - streamer.reset() + await streamer.reset() LOGGER.info("Disconnected from server; reconnecting to %s", url) except (TimeoutError, OSError, ClientError) as e: LOGGER.warning( @@ -862,13 +862,14 @@ async def _source_connection_loop( async def _run_source_mode(args: argparse.Namespace) -> int: """Run as a source client: capture a local input and stream it to a server.""" from aiosendspin.client import SendspinClient as _SendspinClient + from aiosendspin.client import PairingSupport from aiosendspin.models.source import ( ClientHelloSourceSupport, - SourceFeatures, + ClientHelloSourceFeatures, ) from aiosendspin.models.types import AudioCodec, Roles - from sendspin.settings import get_source_settings + from sendspin.settings import get_client_security, get_source_settings from sendspin.source_stream import SourceStreamConfig, SourceStreamer settings = await get_source_settings(args.settings_dir) @@ -894,7 +895,9 @@ async def _run_source_mode(args: argparse.Namespace) -> int: return 1 print(f"Using discovered server: {url}") - client_id, client_name = _resolve_client_info(args.id or settings.client_id, args.name) + _, client_name = _resolve_client_info(args.id or settings.client_id, args.name) + identity, pairing_store = await get_client_security(settings) + client_id = identity.peer_id codec = AudioCodec(codec_str) config = SourceStreamConfig( @@ -910,15 +913,25 @@ async def _run_source_mode(args: argparse.Namespace) -> int: ) # The capture format is announced per stream in client_stream/start; there is # no format negotiation in client/hello. - support = ClientHelloSourceSupport(features=SourceFeatures(line_sense=args.line_sense)) + support = ClientHelloSourceSupport( + features=ClientHelloSourceFeatures(line_sense=args.line_sense) + ) + + async def display_pin(pin: str | None) -> None: + if pin is not None: + print(f"Pairing PIN: {pin}") + client = _SendspinClient( - client_id=client_id, + identity=identity, client_name=client_name, roles=[Roles.SOURCE], + pairing_store=pairing_store, + pairing_support=PairingSupport(pin_display=display_pin, offer_static_pin=False), source_support=support, ) + client.open_pairing_window() streamer = SourceStreamer(client, config) - client.add_source_command_listener(streamer.handle_source_command) + client.add_server_command_listener(streamer.handle_source_command) settings.update( client_id=client_id, @@ -954,14 +967,19 @@ async def _run_daemon_mode( ) -> int: """Run the client in daemon mode (no UI).""" from sendspin.daemon.daemon import DaemonArgs, SendspinDaemon + from sendspin.settings import get_client_security - client_id, client_name = _resolve_client_info(args.id, args.name) + _, client_name = _resolve_client_info(args.id, args.name) + identity, pairing_store = await get_client_security(settings) + client_id = identity.peer_id daemon_args = DaemonArgs( audio_device=audio_device, url=args.url, client_id=client_id, client_name=client_name, + identity=identity, + pairing_store=pairing_store, settings=settings, static_delay_ms=args.static_delay_ms, listen_port=args.listen_port, @@ -1170,9 +1188,12 @@ async def _run_client_mode(args: argparse.Namespace) -> int: if args.command == "daemon": return await _run_daemon_mode(args, settings, audio_device, volume_controller) + from sendspin.settings import get_client_security from sendspin.tui.app import AppArgs, SendspinApp - client_id, client_name = _resolve_client_info(args.id, args.name) + _, client_name = _resolve_client_info(args.id, args.name) + identity, pairing_store = await get_client_security(settings) + client_id = identity.peer_id app_args = AppArgs( audio_device=audio_device, @@ -1180,6 +1201,8 @@ async def _run_client_mode(args: argparse.Namespace) -> int: url_from_settings=url_from_settings, client_id=client_id, client_name=client_name, + identity=identity, + pairing_store=pairing_store, settings=settings, static_delay_ms=args.static_delay_ms, use_mpris=args.use_mpris, diff --git a/sendspin/daemon/daemon.py b/sendspin/daemon/daemon.py index 2faa15c..6112e02 100644 --- a/sendspin/daemon/daemon.py +++ b/sendspin/daemon/daemon.py @@ -10,15 +10,12 @@ from collections.abc import Callable from typing import TYPE_CHECKING -from aiohttp import ClientError, web +from aiohttp import ClientError from aiosendspin.client import ClientListener, SendspinClient -from aiosendspin.models.core import GroupUpdateServerPayload, ServerCommandPayload +from aiosendspin.models.core import ServerCommandPayload from aiosendspin.models.player import ClientHelloPlayerSupport, SupportedAudioFormat from aiosendspin_mpris import MPRIS_AVAILABLE, SendspinMpris from aiosendspin.models.types import ( - ConnectionReason, - GoodbyeReason, - PlaybackStateType, PlayerCommand, Roles, ) @@ -30,6 +27,9 @@ from sendspin.utils import create_task, get_device_info if TYPE_CHECKING: + from aiosendspin.noise.keys import Identity + from aiosendspin.noise.trust_store import ClientPairingStore + from sendspin.volume_controller import VolumeController logger = logging.getLogger(__name__) @@ -42,6 +42,8 @@ class DaemonArgs: audio_device: AudioDevice client_id: str client_name: str + identity: Identity + pairing_store: ClientPairingStore settings: ClientSettings url: str | None = None static_delay_ms: float | None = None @@ -77,9 +79,7 @@ def __init__(self, args: DaemonArgs) -> None: # because CLI overrides aren't persisted to settings, so # `settings.static_delay_ms` can lag the value actually given to the client. self._static_delay_ms: float = 0.0 - self._connection_lock: asyncio.Lock | None = None self._server_url: str | None = None - self._group_update_unsubscribe: Callable[[], None] | None = None self._server_command_unsubscribe: Callable[[], None] | None = None def _create_client(self) -> SendspinClient: @@ -95,9 +95,10 @@ def _create_client(self) -> SendspinClient: supported_formats.insert(0, self._args.preferred_format) return SendspinClient( - client_id=self._args.client_id, + identity=self._args.identity, client_name=self._args.client_name, roles=client_roles, + pairing_store=self._args.pairing_store, device_info=get_device_info( manufacturer=self._args.manufacturer, product_name=self._args.product_name, @@ -202,11 +203,13 @@ async def _run_server_initiated(self) -> None: self._args.listen_port, ) - self._connection_lock = asyncio.Lock() + client = self._create_client() + self._attach_client(client) + client.add_disconnect_listener(lambda: create_task(self._handle_disconnect())) self._listener = ClientListener( client_id=self._args.client_id, - on_connection=self._handle_server_connection, + on_connection=client.attach_websocket, port=self._args.listen_port, client_name=self._args.client_name, host=self._args.interface if self._args.interface is not None else "0.0.0.0", @@ -225,7 +228,6 @@ def _attach_client(self, client: SendspinClient) -> None: self._server_command_unsubscribe = client.add_server_command_listener( self._handle_server_command ) - self._group_update_unsubscribe = client.add_group_update_listener(self._on_group_update) if MPRIS_AVAILABLE and self._args.use_mpris: self._mpris = SendspinMpris(client) self._mpris.start() @@ -235,9 +237,6 @@ def _detach_client(self) -> None: if self._server_command_unsubscribe is not None: self._server_command_unsubscribe() self._server_command_unsubscribe = None - if self._group_update_unsubscribe is not None: - self._group_update_unsubscribe() - self._group_update_unsubscribe = None if self._mpris is not None: self._mpris.stop() self._mpris = None @@ -249,115 +248,6 @@ async def _handle_disconnect(self) -> None: if self._audio_handler is not None: await self._audio_handler.handle_disconnect() - def _should_switch_to_new_server( - self, old_client: SendspinClient, new_client: SendspinClient - ) -> bool: - """Decide whether to switch to a new server per the multi-server spec. - - Assumes both clients have completed their handshake. - """ - assert new_client.server_info is not None - - # Old client may have disconnected before we acquired the lock. - if old_client.server_info is None: - return True - - if new_client.server_info.server_id == old_client.server_info.server_id: - return True - - new_reason = new_client.server_info.connection_reason - old_reason = old_client.server_info.connection_reason - - if new_reason == ConnectionReason.PLAYBACK: - return True - if old_reason == ConnectionReason.PLAYBACK: - return False - - # Both 'discovery' — prefer last played server. - if self._settings.last_played_server_id == new_client.server_info.server_id: - return True - - return False - - def _on_group_update(self, payload: GroupUpdateServerPayload) -> None: - """Track last played server for multi-server arbitration.""" - if payload.playback_state != PlaybackStateType.PLAYING: - return - if self._client is None or self._client.server_info is None: - return - server_id = self._client.server_info.server_id - if self._settings.last_played_server_id != server_id: - self._settings.update(last_played_server_id=server_id) - - async def _handle_server_connection(self, ws: web.WebSocketResponse) -> None: - """Handle an incoming server connection.""" - logger.info("Server connected") - assert self._audio_handler is not None - assert self._connection_lock is not None - assert self._settings is not None - - # Lock ensures we wait for any in-progress handshake to complete - # before disconnecting the previous server - async with self._connection_lock: - old_client = self._client - - # Per spec: always complete the handshake before deciding which - # server to keep. - client = self._create_client() - - try: - await client.attach_websocket(ws) - except TimeoutError: - logger.warning("Handshake with server timed out") - return - except Exception: - logger.exception("Error during server handshake") - return - - # Decide which server to keep. - if old_client is not None: - if self._should_switch_to_new_server(old_client, client): - assert client.server_info is not None - logger.info( - "Switching to server '%s' (%s)", - client.server_info.name, - client.server_info.connection_reason.value, - ) - self._detach_client() - await self._handle_disconnect() - await old_client.send_goodbye(GoodbyeReason.ANOTHER_SERVER) - await old_client.disconnect() - else: - assert old_client.server_info is not None - assert client.server_info is not None - logger.info( - "Keeping server '%s', rejecting '%s' (%s)", - old_client.server_info.name, - client.server_info.name, - client.server_info.connection_reason.value, - ) - await client.send_goodbye(GoodbyeReason.ANOTHER_SERVER) - await client.disconnect() - return - - self._attach_client(client) - - # Handshake complete, release lock so new connections can proceed - # Now wait for disconnect (outside the lock) - try: - disconnect_event = asyncio.Event() - unsubscribe = client.add_disconnect_listener(disconnect_event.set) - await disconnect_event.wait() - unsubscribe() - logger.info("Server disconnected") - except Exception: - logger.exception("Error waiting for server disconnect") - finally: - # Only cleanup if we're still the active client (not replaced by new connection) - if self._client is client: - self._detach_client() - await self._handle_disconnect() - async def _connection_loop(self, url: str) -> None: """Run the connection loop with automatic reconnection (client-initiated mode).""" assert self._client is not None diff --git a/sendspin/serve/__init__.py b/sendspin/serve/__init__.py index 785067a..3ab6a39 100644 --- a/sendspin/serve/__init__.py +++ b/sendspin/serve/__init__.py @@ -10,7 +10,6 @@ import signal import socket import sys -import uuid from contextlib import suppress from dataclasses import dataclass from typing import TYPE_CHECKING, Any @@ -24,6 +23,8 @@ SendspinGroup, ) from aiosendspin.server.push_stream import PushStream +from aiosendspin.noise.keys import Identity +from aiosendspin.noise.trust_store import InMemoryServerPairingStore from sendspin.utils import create_task @@ -120,12 +121,12 @@ async def run_server(config: ServeConfig) -> int: if sys.platform == "win32": event_loop.set_exception_handler(_windows_exception_handler) - server_id = f"sendspin-cli-{uuid.uuid4().hex[:8]}" - server = SendspinPlayerServer( loop=event_loop, - server_id=server_id, + identity=Identity.generate(), server_name=config.name, + pairing_store=InMemoryServerPairingStore(), + allow_unencrypted=True, ) client_connected = asyncio.Event() diff --git a/sendspin/serve/worker.py b/sendspin/serve/worker.py index 5bdbb84..4238030 100644 --- a/sendspin/serve/worker.py +++ b/sendspin/serve/worker.py @@ -10,9 +10,10 @@ import logging import multiprocessing as mp from multiprocessing.sharedctypes import Synchronized -import uuid from contextlib import suppress +from aiosendspin.noise.keys import Identity +from aiosendspin.noise.trust_store import InMemoryServerPairingStore from aiosendspin.server import ( ClientAddedEvent, ClientRemovedEvent, @@ -101,11 +102,12 @@ async def run(self) -> None: async def _start_server(self) -> None: """Start the SendspinPlayerServer on this worker's port.""" loop = asyncio.get_running_loop() - server_id = f"sendspin-worker-{self.worker_id}-{uuid.uuid4().hex[:8]}" self._server = SendspinPlayerServer( loop=loop, - server_id=server_id, + identity=Identity.generate(), server_name=f"Sendspin Worker {self.worker_id}", + pairing_store=InMemoryServerPairingStore(), + allow_unencrypted=True, total_listeners=self._total_listeners, ) self._server.add_event_listener(self._on_server_event) diff --git a/sendspin/settings.py b/sendspin/settings.py index 0a2098a..1b300e6 100644 --- a/sendspin/settings.py +++ b/sendspin/settings.py @@ -9,16 +9,46 @@ import asyncio import json import logging +import os from dataclasses import dataclass, field, fields from pathlib import Path from typing import Any, ClassVar, Literal +from aiosendspin.noise.keys import Identity, b64url_decode +from aiosendspin.noise.trust_store import FileClientPairingStore + logger = logging.getLogger(__name__) # Debounce delay for saving settings SAVE_DEBOUNCE_SECONDS = 60.0 +async def get_client_security( + settings: BaseSettings, +) -> tuple[Identity, FileClientPairingStore]: + """Load or create the identity and pairing store beside a settings file.""" + assert settings._settings_file is not None + directory = settings._settings_file.parent + suffix = settings._settings_file.stem.removeprefix("settings-") + identity_file = directory / f"identity-{suffix}.key" + + def load_identity() -> Identity: + try: + private_bytes = b64url_decode(identity_file.read_text()) + return Identity.from_private_bytes(private_bytes) + except FileNotFoundError: + identity = Identity.generate() + directory.mkdir(parents=True, exist_ok=True) + fd = os.open(identity_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="ascii") as file: + file.write(identity.private_b64u) + return identity + + identity = await asyncio.to_thread(load_identity) + pairing_store = await FileClientPairingStore.open(directory / f"pairing-{suffix}.json") + return identity, pairing_store + + @dataclass class BaseSettings: """Base class for settings with persistence support. diff --git a/sendspin/source_stream.py b/sendspin/source_stream.py index 8c85929..a819567 100644 --- a/sendspin/source_stream.py +++ b/sendspin/source_stream.py @@ -1,8 +1,8 @@ """Audio capture and streaming for the Sendspin source role. ``SourceStreamer`` captures 16-bit PCM from a local input (a synthetic sine test -tone or a real line-in/microphone via ``sounddevice``), encodes it with -``SourceEncoder``, and streams timestamped frames to the server. The server is +tone or a real line-in/microphone via ``sounddevice``) and feeds an SDK-managed +``SourceCapture``. The server is the sole initiator of streaming: capture flows to the server only after a ``server/command`` ``start`` and stops on ``stop`` (or disconnect). """ @@ -16,15 +16,16 @@ from dataclasses import dataclass from typing import TYPE_CHECKING -from aiosendspin.models.source import ClientStreamStartSource -from aiosendspin.models.types import AudioCodec, SourceCommand, SourceSignal +from aiosendspin.models.player import SupportedAudioFormat +from aiosendspin.models.types import AudioCodec, SignalState -from sendspin.source_utils import SourceEncoder, calc_level +from sendspin.source_utils import calc_level from sendspin.utils import create_task if TYPE_CHECKING: from aiosendspin.client import SendspinClient - from aiosendspin.models.source import SourceCommandPayload + from aiosendspin.client.source import SourceCapture + from aiosendspin.models.core import ServerCommandPayload logger = logging.getLogger(__name__) @@ -59,8 +60,9 @@ def __init__(self, client: SendspinClient, config: SourceStreamConfig) -> None: self._client = client self._config = config self._streaming = asyncio.Event() - self._encoder: SourceEncoder | None = None - self._last_signal: SourceSignal | None = None + self._capture: SourceCapture | None = None + self._last_signal: SignalState | None = None + self._command_lock = asyncio.Lock() async def run(self) -> None: """Run the capture loop until cancelled. @@ -78,72 +80,68 @@ def streaming(self) -> bool: """Whether the source is currently streaming to the server.""" return self._streaming.is_set() - def handle_source_command(self, payload: SourceCommandPayload) -> None: + def handle_source_command(self, payload: ServerCommandPayload) -> None: """React to a server start/stop command.""" - if payload.command == SourceCommand.START: + if payload.source is None: + return + logger.info("Received source %s command", payload.source.command) + if payload.source.command == "start": create_task(self._begin_stream()) - elif payload.command == SourceCommand.STOP: + elif payload.source.command == "stop": create_task(self._end_stream()) - def reset(self) -> None: + async def reset(self) -> None: """Clear streaming state (e.g., on disconnect).""" - self._streaming.clear() - self._encoder = None + await self._end_stream() self._last_signal = None async def _begin_stream(self) -> None: - if self._streaming.is_set(): - return - cfg = self._config - encoder = SourceEncoder( - codec=cfg.codec, - channels=cfg.channels, - sample_rate=cfg.sample_rate, - frame_samples=cfg.samples_per_frame, - ) - self._encoder = encoder - await self._client.send_client_stream_start( - ClientStreamStartSource( - codec=cfg.codec, - channels=cfg.channels, - sample_rate=cfg.sample_rate, - bit_depth=16, - codec_header=encoder.codec_header, + async with self._command_lock: + if self._streaming.is_set(): + return + cfg = self._config + capture = self._client.create_source_capture( + SupportedAudioFormat( + codec=cfg.codec, + sample_rate=cfg.sample_rate, + bit_depth=16, + channels=cfg.channels, + ) ) - ) - self._streaming.set() - logger.info("Source streaming started (%s, %d Hz)", cfg.codec.value, cfg.sample_rate) + await capture.start() + self._capture = capture + self._streaming.set() + logger.info("Source streaming started (%s, %d Hz)", cfg.codec.value, cfg.sample_rate) async def _end_stream(self) -> None: - if not self._streaming.is_set(): - return - self._streaming.clear() - if self._encoder is not None: - for tail in self._encoder.flush(): - await self._client.send_source_audio_chunk( - tail, capture_timestamp_us=self._client.now_us() - ) - self._encoder = None - await self._client.send_client_stream_end() - logger.info("Source streaming stopped") + async with self._command_lock: + if not self._streaming.is_set(): + return + self._streaming.clear() + capture = self._capture + self._capture = None + if capture is not None: + await capture.stop() + logger.info("Source streaming stopped") async def _send_frame(self, pcm: bytes) -> None: """Report signal (if line sensing) and stream the frame when active.""" if self._config.line_sense: self._maybe_report_signal(pcm) - if not self._streaming.is_set() or self._encoder is None: - return - capture_us = self._client.now_us() - for encoded, frame_us in self._encoder.encode(pcm, capture_us): - await self._client.send_source_audio_chunk(encoded, capture_timestamp_us=frame_us) + async with self._command_lock: + if not self._streaming.is_set() or self._capture is None: + return + await self._capture.feed(pcm) def _maybe_report_signal(self, pcm: bytes) -> None: level = calc_level(pcm) threshold = 10 ** (self._config.signal_threshold_db / 20) - signal = SourceSignal.PRESENT if level >= threshold else SourceSignal.ABSENT + signal = SignalState.PRESENT if level >= threshold else SignalState.ABSENT if signal != self._last_signal: self._last_signal = signal - create_task(self._client.send_source_state(signal=signal)) + connection = self._client._admitted_connection # noqa: SLF001 + if connection is not None: + create_task(connection.send_source_signal(signal)) async def _stream_sine(self) -> None: cfg = self._config diff --git a/sendspin/source_utils.py b/sendspin/source_utils.py index d1000d6..eae12ec 100644 --- a/sendspin/source_utils.py +++ b/sendspin/source_utils.py @@ -1,33 +1,13 @@ -"""Encoding and signal helpers for the Sendspin source role. - -A source client captures 16-bit PCM from a local input and streams it to the -server. ``SourceEncoder`` encodes that PCM to the negotiated codec (PCM -passthrough, FLAC, or Opus) using PyAV, exposing the raw codec header so the -server can initialize its decoder. ``calc_level`` provides a simple RMS level -used for optional line sensing. -""" +"""Signal helpers for the Sendspin source role.""" from __future__ import annotations import array -import base64 -import logging import sys -from fractions import Fraction -from types import ModuleType -from typing import TYPE_CHECKING - -from aiosendspin.models.types import AudioCodec - -if TYPE_CHECKING: - import av - -logger = logging.getLogger(__name__) # Source capture is fixed at 16-bit signed PCM (matches sounddevice int16 capture # and keeps codec init simple). The server can still receive other depths from # other source implementations. -BYTES_PER_SAMPLE = 2 _MAX_INT16 = 32767.0 @@ -47,113 +27,3 @@ def calc_level(pcm: bytes) -> float: total += norm * norm rms = (total / len(samples)) ** 0.5 return float(min(1.0, rms)) - - -class SourceEncoder: - """Encode 16-bit interleaved PCM to the negotiated source codec via PyAV.""" - - def __init__( - self, - *, - codec: AudioCodec, - channels: int, - sample_rate: int, - frame_samples: int, - ) -> None: - """Create an encoder. - - Args: - codec: Target codec (pcm, flac, or opus). - channels: Number of channels. - sample_rate: Sample rate in Hz. - frame_samples: Preferred samples per frame (used for PCM and as a - fallback when the codec does not report its own frame size). - """ - self._codec = codec - self._channels = channels - self._sample_rate = sample_rate - self._stride = BYTES_PER_SAMPLE * channels - self._layout = "mono" if channels == 1 else "stereo" - self._buffer = bytearray() - self._buffer_head_us: int | None = None - self._pts = 0 - self._encoder: av.AudioCodecContext | None = None - self._codec_header: str | None = None - self._frame_samples = frame_samples - - if codec == AudioCodec.PCM: - return - - av_mod = _get_av() - codec_name = "libopus" if codec == AudioCodec.OPUS else "flac" - encoder = av_mod.AudioCodecContext.create(codec_name, "w") - encoder.sample_rate = sample_rate - encoder.format = "s16" - encoder.layout = self._layout - with av_mod.logging.Capture(): - encoder.open() - if encoder.frame_size: - self._frame_samples = encoder.frame_size - if encoder.extradata: - self._codec_header = base64.b64encode(bytes(encoder.extradata)).decode("ascii") - self._encoder = encoder - - @property - def codec_header(self) -> str | None: - """Base64 raw codec header (extradata) for client_stream/start, if any.""" - return self._codec_header - - @property - def frame_samples(self) -> int: - """Preferred number of samples per captured frame.""" - return self._frame_samples - - def encode(self, pcm: bytes, capture_timestamp_us: int) -> list[tuple[bytes, int]]: - """Encode captured PCM into ``(frame_bytes, capture_timestamp_us)`` pairs. - - For PCM the input passes through unchanged. For FLAC/Opus the PCM is - buffered and emitted in codec-sized frames; each emitted frame is stamped - with the capture time of its first sample. - """ - if self._encoder is None: - return [(pcm, capture_timestamp_us)] if pcm else [] - - if not self._buffer: - self._buffer_head_us = capture_timestamp_us - self._buffer.extend(pcm) - - results: list[tuple[bytes, int]] = [] - chunk_size = self._frame_samples * self._stride - while len(self._buffer) >= chunk_size: - block = bytes(self._buffer[:chunk_size]) - del self._buffer[:chunk_size] - assert self._buffer_head_us is not None - frame_ts = self._buffer_head_us - self._buffer_head_us += round(self._frame_samples * 1_000_000 / self._sample_rate) - for encoded in self._encode_block(block): - results.append((encoded, frame_ts)) - return results - - def flush(self) -> list[bytes]: - """Flush the codec, returning any trailing frames.""" - if self._encoder is None: - return [] - return [data for packet in self._encoder.encode(None) if (data := bytes(packet))] - - def _encode_block(self, block: bytes) -> list[bytes]: - assert self._encoder is not None - av_mod = _get_av() - frame = av_mod.AudioFrame(format="s16", layout=self._layout, samples=self._frame_samples) - frame.sample_rate = self._sample_rate - frame.pts = self._pts - frame.time_base = Fraction(1, self._sample_rate) - self._pts += self._frame_samples - frame.planes[0].update(block) - return [data for packet in self._encoder.encode(frame) if (data := bytes(packet))] - - -def _get_av() -> ModuleType: - """Import PyAV lazily so non-source commands do not require it eagerly.""" - import av # noqa: PLC0415 - - return av diff --git a/sendspin/tui/app.py b/sendspin/tui/app.py index ad6eb63..1c1c337 100644 --- a/sendspin/tui/app.py +++ b/sendspin/tui/app.py @@ -13,6 +13,8 @@ if TYPE_CHECKING: from aiosendspin.models.metadata import SessionUpdateMetadata + from aiosendspin.noise.keys import Identity + from aiosendspin.noise.trust_store import ClientPairingStore from sendspin.volume_controller import VolumeController @@ -22,7 +24,6 @@ from aiosendspin.models.core import ( GroupUpdateServerPayload, ServerCommandPayload, - ServerHelloPayload, ServerStatePayload, StreamStartMessage, ) @@ -237,6 +238,8 @@ class AppArgs: audio_device: AudioDevice client_id: str client_name: str + identity: Identity + pairing_store: ClientPairingStore settings: ClientSettings url: str | None = None url_from_settings: bool = False @@ -334,9 +337,10 @@ def _create_client(self) -> SendspinClient: assert self._audio_handler is not None return SendspinClient( - client_id=args.client_id, + identity=args.identity, client_name=args.client_name, roles=roles, + pairing_store=args.pairing_store, device_info=get_device_info( manufacturer=args.manufacturer, product_name=args.product_name, @@ -365,7 +369,6 @@ def _attach_client(self) -> None: self._client.add_controller_state_listener(self._handle_server_state), self._client.add_server_command_listener(self._handle_server_command), self._client.add_color_listener(self._handle_color_update), - self._client.add_server_hello_listener(self._handle_server_hello), ] self._audio_handler.attach_client(self._client) @@ -793,7 +796,11 @@ def _handle_metadata_update(self, payload: ServerStatePayload) -> None: assert self._ui is not None state = self._state ui = self._ui - if payload.metadata is None or not state.update_metadata(payload.metadata): + if ( + payload.metadata is None + or isinstance(payload.metadata, UndefinedField) + or not state.update_metadata(payload.metadata) + ): return with ui.batch_update(): @@ -819,7 +826,8 @@ def _clear_visualizer_timelines(self) -> None: def _handle_color_update(self, payload: ServerStatePayload) -> None: """Forward a color@v1 palette payload to the UI.""" assert self._ui is not None - self._ui.update_palette(payload.color) + if not isinstance(payload.color, UndefinedField): + self._ui.update_palette(payload.color) def _persist_color_mode(self, mode: ColorMode) -> None: """Persist the user's color theme choice.""" @@ -855,7 +863,7 @@ def _handle_server_state(self, payload: ServerStatePayload) -> None: assert self._ui is not None state = self._state ui = self._ui - if not payload.controller: + if not payload.controller or isinstance(payload.controller, UndefinedField): return controller = payload.controller @@ -952,20 +960,6 @@ def _server_now_us(self) -> int: assert self._client is not None return self._client.compute_server_time(self._client.now_us()) - def _handle_server_hello(self, payload: ServerHelloPayload) -> None: - """Hide the visualizer panel when the server didn't activate visualizer@v1.""" - if not self._visualizer_enabled: - return - if Roles.VISUALIZER.value in payload.active_roles: - return - logger.warning( - "Server did not activate %s (active_roles=%s); hiding the visualizer panel.", - Roles.VISUALIZER.value, - payload.active_roles, - ) - if self._ui is not None: - self._ui.set_visualizer_enabled(False) - def _handle_stream_start(self, message: StreamStartMessage) -> None: """Record which visualizer types the server negotiated for this stream.""" if self._ui is None: diff --git a/tests/daemon/test_daemon.py b/tests/daemon/test_daemon.py index b781e2f..bf0fe7d 100644 --- a/tests/daemon/test_daemon.py +++ b/tests/daemon/test_daemon.py @@ -5,6 +5,8 @@ from types import SimpleNamespace from aiosendspin.models.types import PlayerCommand +from aiosendspin.noise.keys import Identity +from aiosendspin.noise.trust_store import InMemoryClientPairingStore from sendspin.daemon.daemon import DaemonArgs, SendspinDaemon from sendspin.settings import ClientSettings @@ -36,6 +38,8 @@ def _make_daemon(tmp_path: Path, *, settings_volume: int, settings_muted: bool) audio_device=SimpleNamespace(index=0, name="Fake Device"), client_id="test-client", client_name="Test Client", + identity=Identity.generate(), + pairing_store=InMemoryClientPairingStore(), settings=settings, use_mpris=False, ) diff --git a/tests/test_source_stream.py b/tests/test_source_stream.py index 705b697..ad6a767 100644 --- a/tests/test_source_stream.py +++ b/tests/test_source_stream.py @@ -4,40 +4,51 @@ import asyncio -from aiosendspin.models.source import ClientStreamStartSource, SourceCommandPayload -from aiosendspin.models.types import AudioCodec, SourceCommand, SourceSignal +from aiosendspin.models.core import ServerCommandPayload +from aiosendspin.models.source import SourceCommandServerPayload +from aiosendspin.models.types import AudioCodec, SignalState from sendspin.source_stream import SourceStreamConfig, SourceStreamer -class _FakeClient: - """Records the source-related calls a SourceStreamer makes.""" - +class _FakeCapture: def __init__(self) -> None: - self.stream_starts: list[ClientStreamStartSource] = [] - self.stream_ends = 0 - self.chunks: list[tuple[bytes, int]] = [] - self.signals: list[SourceSignal | None] = [] - self._clock = 1_000_000 + self.starts = 0 + self.stops = 0 + self.frames: list[bytes] = [] + self.start_gate: asyncio.Event | None = None - def now_us(self) -> int: - self._clock += 1000 - return self._clock + async def start(self) -> None: + self.starts += 1 + if self.start_gate is not None: + await self.start_gate.wait() - async def send_client_stream_start(self, source: ClientStreamStartSource) -> None: - self.stream_starts.append(source) + async def stop(self) -> None: + self.stops += 1 - async def send_client_stream_end(self) -> None: - self.stream_ends += 1 + async def feed(self, pcm: bytes) -> None: + self.frames.append(pcm) - async def send_source_audio_chunk(self, data: bytes, *, capture_timestamp_us: int) -> bool: - self.chunks.append((data, capture_timestamp_us)) - return True - async def send_source_state(self, *, signal: SourceSignal | None = None) -> None: +class _FakeConnection: + def __init__(self) -> None: + self.signals: list[SignalState] = [] + + async def send_source_signal(self, signal: SignalState) -> None: self.signals.append(signal) +class _FakeClient: + """Records the source-related calls a SourceStreamer makes.""" + + def __init__(self) -> None: + self.capture = _FakeCapture() + self._admitted_connection = _FakeConnection() + + def create_source_capture(self, _audio_format: object) -> _FakeCapture: + return self.capture + + def _config(*, codec: AudioCodec = AudioCodec.PCM, line_sense: bool = False) -> SourceStreamConfig: return SourceStreamConfig( codec=codec, @@ -61,8 +72,7 @@ async def test_begin_stream_announces_format_and_starts() -> None: """Beginning a stream sends client_stream/start and marks streaming active.""" streamer, client = _make() await streamer._begin_stream() # noqa: SLF001 - assert len(client.stream_starts) == 1 - assert client.stream_starts[0].codec == AudioCodec.PCM + assert client.capture.starts == 1 assert streamer._streaming.is_set() # noqa: SLF001 @@ -71,7 +81,7 @@ async def test_end_stream_sends_end_and_stops() -> None: streamer, client = _make() await streamer._begin_stream() # noqa: SLF001 await streamer._end_stream() # noqa: SLF001 - assert client.stream_ends == 1 + assert client.capture.stops == 1 assert not streamer._streaming.is_set() # noqa: SLF001 @@ -81,12 +91,11 @@ async def test_send_frame_streams_only_when_active() -> None: pcm = b"\x01\x02\x03\x04" * 16 await streamer._send_frame(pcm) # noqa: SLF001 (not started yet) - assert client.chunks == [] + assert client.capture.frames == [] await streamer._begin_stream() # noqa: SLF001 await streamer._send_frame(pcm) # noqa: SLF001 - assert len(client.chunks) == 1 - assert client.chunks[0][0] == pcm # PCM passthrough + assert client.capture.frames == [pcm] async def test_line_sense_reports_signal_changes() -> None: @@ -102,19 +111,38 @@ async def test_line_sense_reports_signal_changes() -> None: streamer._maybe_report_signal(silence) # noqa: SLF001 await asyncio.sleep(0.05) # let the scheduled send_source_state tasks run - assert client.signals == [SourceSignal.PRESENT, SourceSignal.ABSENT] + assert client._admitted_connection.signals == [SignalState.PRESENT, SignalState.ABSENT] async def test_handle_source_command_dispatches_start_stop() -> None: """A server start command begins streaming; a stop command ends it.""" streamer, client = _make() - streamer.handle_source_command(SourceCommandPayload(command=SourceCommand.START)) + streamer.handle_source_command( + ServerCommandPayload(source=SourceCommandServerPayload(command="start")) + ) await asyncio.sleep(0.05) assert streamer._streaming.is_set() # noqa: SLF001 - assert len(client.stream_starts) == 1 + assert client.capture.starts == 1 - streamer.handle_source_command(SourceCommandPayload(command=SourceCommand.STOP)) + streamer.handle_source_command( + ServerCommandPayload(source=SourceCommandServerPayload(command="stop")) + ) await asyncio.sleep(0.05) assert not streamer._streaming.is_set() # noqa: SLF001 - assert client.stream_ends == 1 + assert client.capture.stops == 1 + + +async def test_stop_waits_for_in_progress_start() -> None: + """A quick stop cannot be overtaken by an unfinished start.""" + streamer, client = _make() + client.capture.start_gate = asyncio.Event() + + start = asyncio.create_task(streamer._begin_stream()) # noqa: SLF001 + await asyncio.sleep(0) + stop = asyncio.create_task(streamer._end_stream()) # noqa: SLF001 + client.capture.start_gate.set() + await asyncio.gather(start, stop) + + assert not streamer.streaming + assert client.capture.stops == 1 diff --git a/tests/test_source_utils.py b/tests/test_source_utils.py index 0a42891..86394eb 100644 --- a/tests/test_source_utils.py +++ b/tests/test_source_utils.py @@ -5,16 +5,10 @@ import math import struct -import pytest -from aiosendspin.models.source import ClientStreamStartSource -from aiosendspin.models.types import AudioCodec -from aiosendspin.server.roles.source.group import SourceDecoder - -from sendspin.source_utils import SourceEncoder, calc_level +from sendspin.source_utils import calc_level RATE = 48000 CHANNELS = 2 -STRIDE = 2 * CHANNELS def _sine_pcm(duration_ms: int, freq: float = 440.0) -> bytes: @@ -35,65 +29,3 @@ def test_calc_level_silence_is_zero() -> None: def test_calc_level_signal_is_positive() -> None: """A real signal produces a positive normalized level.""" assert 0.0 < calc_level(_sine_pcm(20)) <= 1.0 - - -def test_pcm_encoder_passes_through() -> None: - """PCM encoding passes bytes through unchanged with no header.""" - encoder = SourceEncoder( - codec=AudioCodec.PCM, channels=CHANNELS, sample_rate=RATE, frame_samples=960 - ) - assert encoder.codec_header is None - pcm = _sine_pcm(20) - assert encoder.encode(pcm, 5000) == [(pcm, 5000)] - assert encoder.encode(b"", 6000) == [] - - -@pytest.mark.parametrize("codec", [AudioCodec.FLAC, AudioCodec.OPUS]) -def test_compressed_encoder_produces_header_and_frames(codec: AudioCodec) -> None: - """FLAC/Opus encoding yields a codec header and encoded frames.""" - encoder = SourceEncoder( - codec=codec, channels=CHANNELS, sample_rate=RATE, frame_samples=RATE * 20 // 1000 - ) - assert encoder.codec_header is not None - frames = encoder.encode(_sine_pcm(200), 1_000_000) - frames.extend((tail, 0) for tail in encoder.flush()) - assert frames - assert all(isinstance(data, bytes) and data for data, _ in frames) - - -@pytest.mark.parametrize("codec", [AudioCodec.PCM, AudioCodec.FLAC, AudioCodec.OPUS]) -def test_encode_round_trips_through_server_decoder(codec: AudioCodec) -> None: - """Frames encoded by the CLI decode back to PCM on the server side.""" - pcm = _sine_pcm(500) - encoder = SourceEncoder( - codec=codec, channels=CHANNELS, sample_rate=RATE, frame_samples=RATE * 20 // 1000 - ) - frame_bytes = RATE * 20 // 1000 * STRIDE - encoded: list[bytes] = [] - offset = 0 - ts = 1_000_000 - while offset < len(pcm): - block = pcm[offset : offset + frame_bytes] - offset += frame_bytes - encoded.extend(data for data, _ in encoder.encode(block, ts)) - ts += 20_000 - encoded.extend(encoder.flush()) - - decoder = SourceDecoder( - ClientStreamStartSource( - codec=codec, - channels=CHANNELS, - sample_rate=RATE, - bit_depth=16, - codec_header=encoder.codec_header, - ) - ) - decoded = bytearray() - for frame in encoded: - for chunk in decoder.decode(frame): - decoded += chunk - - in_samples = len(pcm) // STRIDE - out_samples = len(decoded) // STRIDE - # PCM is exact; FLAC/Opus carry small codec delay/padding, so allow a tolerance. - assert out_samples == pytest.approx(in_samples, abs=RATE // 10) diff --git a/tests/tui/test_role_negotiation.py b/tests/tui/test_role_negotiation.py deleted file mode 100644 index 7ceffe6..0000000 --- a/tests/tui/test_role_negotiation.py +++ /dev/null @@ -1,69 +0,0 @@ -from __future__ import annotations - -from pathlib import Path -from types import SimpleNamespace - -from aiosendspin.models.core import ServerHelloPayload -from aiosendspin.models.types import ConnectionReason, Roles - -from sendspin.settings import ClientSettings -from sendspin.tui.app import AppArgs, SendspinApp - - -class _FakeUI: - def __init__(self) -> None: - self.visualizer_enabled_calls: list[bool] = [] - - def set_visualizer_enabled(self, enabled: bool) -> None: - self.visualizer_enabled_calls.append(enabled) - - -def _make_app(tmp_path: Path) -> SendspinApp: - args = AppArgs( - audio_device=SimpleNamespace(index=0, name="Fake Device"), - client_id="test-client", - client_name="Test Client", - settings=ClientSettings(_settings_file=tmp_path / "settings.json"), - use_mpris=False, - ) - return SendspinApp(args) - - -def _payload(active_roles: list[str]) -> ServerHelloPayload: - return ServerHelloPayload( - server_id="srv", - name="srv", - version=1, - active_roles=active_roles, - connection_reason=ConnectionReason.DISCOVERY, - ) - - -def test_server_hello_without_visualizer_role_hides_panel(tmp_path: Path) -> None: - app = _make_app(tmp_path) - app._visualizer_enabled = True - app._ui = _FakeUI() - - app._handle_server_hello(_payload(active_roles=["player@v1", "controller@v1"])) - - assert app._ui.visualizer_enabled_calls == [False] - - -def test_server_hello_with_visualizer_role_leaves_panel(tmp_path: Path) -> None: - app = _make_app(tmp_path) - app._visualizer_enabled = True - app._ui = _FakeUI() - - app._handle_server_hello(_payload(active_roles=[Roles.VISUALIZER.value, "player@v1"])) - - assert app._ui.visualizer_enabled_calls == [] - - -def test_server_hello_ignored_when_visualizer_disabled(tmp_path: Path) -> None: - app = _make_app(tmp_path) - app._visualizer_enabled = False - app._ui = _FakeUI() - - app._handle_server_hello(_payload(active_roles=["player@v1"])) - - assert app._ui.visualizer_enabled_calls == [] diff --git a/tests/tui/test_volume_state.py b/tests/tui/test_volume_state.py index 61c6f12..8c3002e 100644 --- a/tests/tui/test_volume_state.py +++ b/tests/tui/test_volume_state.py @@ -5,6 +5,8 @@ from types import SimpleNamespace from aiosendspin.models.types import PlayerCommand +from aiosendspin.noise.keys import Identity +from aiosendspin.noise.trust_store import InMemoryClientPairingStore from sendspin.settings import ClientSettings from sendspin.tui.app import AppArgs, AppState, SendspinApp @@ -52,6 +54,8 @@ def _make_app(tmp_path: Path) -> SendspinApp: audio_device=SimpleNamespace(index=0, name="Fake Device"), client_id="test-client", client_name="Test Client", + identity=Identity.generate(), + pairing_store=InMemoryClientPairingStore(), settings=_make_settings(tmp_path), use_mpris=False, ) diff --git a/uv.lock b/uv.lock index 0506d21..0c93587 100644 --- a/uv.lock +++ b/uv.lock @@ -98,17 +98,20 @@ wheels = [ [[package]] name = "aiosendspin" -version = "6.0.1" +version = "9.1.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aiohttp" }, + { name = "cpace" }, + { name = "cryptography" }, { name = "mashumaro" }, + { name = "noiseprotocol" }, { name = "orjson" }, { name = "zeroconf" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/18/89/14c04309b2ee46095df4ce5fcd70554b26cf0c275526995a1c4aa9d59525/aiosendspin-6.0.1.tar.gz", hash = "sha256:a0c066fd7619113a643954aaa5265fc209d447589e7fe8ed09b19b070d0ed745", size = 160231, upload-time = "2026-05-31T14:54:18.649Z" } +sdist = { url = "https://files.pythonhosted.org/packages/81/51/b732cf80d93bf9c9c16f3a2bbe1de98e249a6e8c00ce3eed0c6540fa9c1e/aiosendspin-9.1.0.tar.gz", hash = "sha256:1cf7d500f970eb561ea9d3215935daa70c5c15cf8e4744c9364c7d2d5d2aca81", size = 249007, upload-time = "2026-08-11T14:43:34.973Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b6/96/bf436baac06826d2ccf5d41d871aa81c6fbb5e85d14b2cc9dd2bc4783041/aiosendspin-6.0.1-py3-none-any.whl", hash = "sha256:4bfbb3bdd68dc27d4d14ff8a4c34cb87b7ba9664d50908e1f117c545d9c7e86c", size = 187968, upload-time = "2026-05-31T14:54:17.037Z" }, + { url = "https://files.pythonhosted.org/packages/ff/ed/1b7ec599fbf0507ac7e436c044bfd6848da27920653a8bbf6c3ebceb3d33/aiosendspin-9.1.0-py3-none-any.whl", hash = "sha256:e0eefbe0ca0f88c5d098ca80d4d8ee53ac552859dc995e6f7e39ca9aa8fde1aa", size = 275527, upload-time = "2026-08-11T14:43:33.274Z" }, ] [package.optional-dependencies] @@ -117,6 +120,10 @@ server = [ { name = "numpy" }, { name = "pillow" }, ] +source = [ + { name = "av" }, + { name = "numpy" }, +] [[package]] name = "aiosendspin-mpris" @@ -355,6 +362,68 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "cpace" +version = "0.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cryptography" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/02/e9/0aa114fdf26b4112222ddbfd190197aa5d11bce21b4747cc1889998eead5/cpace-0.1.0.tar.gz", hash = "sha256:049d30b4389c965cb2d98551f2f7361382bfa342afc5d53b6dc16b84a5759ad2", size = 60404, upload-time = "2026-07-13T21:07:15.01Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fa/e8/949c45844ad0d65d0112c2c4fa11cc8a7c4359fe2a20667e00c930860bef/cpace-0.1.0-py3-none-any.whl", hash = "sha256:9fabb60a711a85934225be4081b3f5994e1ca4cd1335c5b9171770c8f1a2fda3", size = 9456, upload-time = "2026-07-13T21:07:13.703Z" }, +] + +[[package]] +name = "cryptography" +version = "50.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cffi", marker = "platform_python_implementation != 'PyPy'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/de/41/6cbdcf9142d00fe82836fbb51e503e58088575cf7a0fe1dbff6695bf0840/cryptography-50.0.0.tar.gz", hash = "sha256:eeac2acb5a20ed25e0ad6d1df9891a520b78b404266b6d11778f25d5d691a6c9", size = 880201, upload-time = "2026-07-31T14:25:10.11Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c5/5c/59086b4aac5e879d38ddbcf74e4be7ade89cebc3eb199a55da998c3bb46a/cryptography-50.0.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:031e2d5dd4bb9caa3ca9c82e5a197fd8ae680232cee62603d1a813f3f07e3d03", size = 4001252, upload-time = "2026-07-31T14:23:33.331Z" }, + { url = "https://files.pythonhosted.org/packages/57/ef/8f2df13c7216bcad3e1c74e07f6e193d93e998e114f524a53877c9af27ad/cryptography-50.0.0-cp311-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:fd9192b7b70c573d7f214eb1ae35e00d359f6f5e4b27c7e21e30de1fc6204645", size = 4719554, upload-time = "2026-07-31T14:23:35.611Z" }, + { url = "https://files.pythonhosted.org/packages/d9/41/029086c34d91052fc3b88bcc8056f709a7c915c7a23b235a54eb800b1c97/cryptography-50.0.0-cp311-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:06a32a980526a6ab9a4b9bf8f7385800791e2bb960903cb6b530e4817509a3b7", size = 4702130, upload-time = "2026-07-31T14:23:37.635Z" }, + { url = "https://files.pythonhosted.org/packages/7d/ff/b6ce0954962e7f7b969f850a883744197bb3910bdfd7b6da162eab7d9f68/cryptography-50.0.0-cp311-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:a1b30560f2acc95aa8b2e06e716a13dbfc97314747b80d9707e307f77b40d6b3", size = 4725244, upload-time = "2026-07-31T14:23:39.471Z" }, + { url = "https://files.pythonhosted.org/packages/06/1e/63a1027cb7fec360a182208e1b7767d5aa1fe57be3d6aa856e69a321edc0/cryptography-50.0.0-cp311-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:8d89f3976b10b4ce31118de72329025f70d2c6ead14a8217c5514dd2c6d5a78f", size = 5342265, upload-time = "2026-07-31T14:23:41.286Z" }, + { url = "https://files.pythonhosted.org/packages/6b/72/a1116d683a6d7ece94590013882515de087edf9ef0e6292aae615a44df73/cryptography-50.0.0-cp311-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:b42a28c1844fd9de8f3f7d540e36b66f3a9c83fceac7170ebc7a6a19edd9dcae", size = 4734609, upload-time = "2026-07-31T14:23:43.139Z" }, + { url = "https://files.pythonhosted.org/packages/15/37/36a9c479bbe49acea2636c7fd3360d20f7b7e079c300352011c44850b181/cryptography-50.0.0-cp311-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:900131fafd8aead39ac7dd3a7e833be754c17a95cfd91221636949fe4eb0aa8a", size = 4356517, upload-time = "2026-07-31T14:23:44.939Z" }, + { url = "https://files.pythonhosted.org/packages/32/98/8a151d64367204cbc63ec65d37502f1d9c53cf4bfc6ec3c532614dbec60d/cryptography-50.0.0-cp311-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:07949c449a1abcf60d1ee6e88956d89404c7df3c8258f46589e912988e551987", size = 4724529, upload-time = "2026-07-31T14:23:46.93Z" }, + { url = "https://files.pythonhosted.org/packages/22/f6/ec13b470172126464a86bf54d2294a46d29837fc51ba3e45d4047946fb5e/cryptography-50.0.0-cp311-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:f89831ef99dd7dd169ab06d63a831adb9e20a87aac6d380266bbda5823349169", size = 5299852, upload-time = "2026-07-31T14:23:48.851Z" }, + { url = "https://files.pythonhosted.org/packages/da/3a/f05e32c99d440c9bb891ea0e36c9091891e36be5a9a87ab2ee6ea20729f6/cryptography-50.0.0-cp311-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:82148ec5bddac30b51a5b3c1945075f896fa022cb93f8e4a01e9f6ee95292c5f", size = 4734462, upload-time = "2026-07-31T14:23:50.861Z" }, + { url = "https://files.pythonhosted.org/packages/ca/dc/bd72b26be8953f80625f63151efd38eee71c76ca6cf591c08ff34615a79e/cryptography-50.0.0-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:1489e263a8048bb8b6a8bac662eb2d402ea5d2b7b4699b72f385f1e2772db105", size = 4852708, upload-time = "2026-07-31T14:23:52.715Z" }, + { url = "https://files.pythonhosted.org/packages/27/20/c930314a2ab476d15dec966ec87e2e9637bb02b06106b12c0396c57bb603/cryptography-50.0.0-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:7cec5b856506da6defb290f30c9ee687d5f5e8cb0bd3f6459dde43b0b4fa40ef", size = 5004179, upload-time = "2026-07-31T14:23:54.887Z" }, + { url = "https://files.pythonhosted.org/packages/32/2e/c9db68a0c4bfa28e310707527c0ee3a2bd254104d2e02e68f368e197aa4c/cryptography-50.0.0-cp311-abi3-win_amd64.whl", hash = "sha256:bd1c592e4d5974f0d08d4888e432157adba757c66da0246918e43677fafa2d30", size = 3840395, upload-time = "2026-07-31T14:23:56.677Z" }, + { url = "https://files.pythonhosted.org/packages/c3/fb/951032a3bf22a5697c83183fb6294a4843772947a70e616c57b3ff5f522e/cryptography-50.0.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:49e7d93abdbd2990caced757e5fade25302f719c3c8fb6e6fff2dde98999fc41", size = 3989258, upload-time = "2026-07-31T14:23:58.881Z" }, + { url = "https://files.pythonhosted.org/packages/d4/67/91eb047e69c5e845f2f14b8a2e4a1aab0f283cb885531e9e22c8adb176bc/cryptography-50.0.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:19736989797678c6af1e55cd49055cdbcb55d8f6b5583ac5335f933aba9101dc", size = 4700648, upload-time = "2026-07-31T14:24:00.702Z" }, + { url = "https://files.pythonhosted.org/packages/30/82/85f0f7425c856b9f96459411eb12e74ef72df9caf6f8f15bf23a33ff131f/cryptography-50.0.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:80b63928fa35083b33966ce1efb70e5b9607181e49dcd1c22c8c005e319f667f", size = 4682442, upload-time = "2026-07-31T14:24:02.538Z" }, + { url = "https://files.pythonhosted.org/packages/1a/28/b555a365adff1cca2fbe7b9e487d68a40de6bc67ff2cb587473eb43de0e7/cryptography-50.0.0-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:d58c3db7cd6eed54e6c06744db55456b65ebd7492ddeae9c1e93cfca7aa857d3", size = 4707596, upload-time = "2026-07-31T14:24:04.394Z" }, + { url = "https://files.pythonhosted.org/packages/72/d8/f52538140cc719df62a01cf87d1c7142318d235817109d6f4054d7c352d6/cryptography-50.0.0-cp314-cp314t-manylinux_2_28_ppc64le.whl", hash = "sha256:df2a58a472f332225671c35b0a830208b86d004f82baa8530fa3782c85646533", size = 5314552, upload-time = "2026-07-31T14:24:06.31Z" }, + { url = "https://files.pythonhosted.org/packages/38/14/6120e5bd7c5aa022ad15424ba4d5c5269d0d9448ed4d55e492ea91e3c1c4/cryptography-50.0.0-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:11b74db56cdbe3cdee6e3f6982ecb70334fa10dce99ed58bf7894aaaa3b2a037", size = 4717113, upload-time = "2026-07-31T14:24:08.349Z" }, + { url = "https://files.pythonhosted.org/packages/fa/71/190bf38c3ee2e0f8efc9860ae100c9df4169742eef274b91e7aa1cb133b9/cryptography-50.0.0-cp314-cp314t-manylinux_2_31_armv7l.whl", hash = "sha256:f59e38625469987d7ef6d495323c55e7db6c212eaf6112267e0d3b565a2e9c9f", size = 4338580, upload-time = "2026-07-31T14:24:10.227Z" }, + { url = "https://files.pythonhosted.org/packages/3a/63/504ccfbbe61fd8aa983f7f146399cdf034c72c2fc55f5b2dfdcdcdb20c99/cryptography-50.0.0-cp314-cp314t-manylinux_2_34_aarch64.whl", hash = "sha256:ecfed7367f965a0328cfbdd70da860f15441f002f613185668c6e6ebf5a0ac11", size = 4707038, upload-time = "2026-07-31T14:24:12.169Z" }, + { url = "https://files.pythonhosted.org/packages/01/77/2cf79bbfc4d12ca106437a6e170d6aaa01a373e93093118aaaef0e801bd4/cryptography-50.0.0-cp314-cp314t-manylinux_2_34_ppc64le.whl", hash = "sha256:9aa87839c383bdbab6ef865787a1fb877af8dd03464c4400322726feaaadfc6d", size = 5273110, upload-time = "2026-07-31T14:24:14.38Z" }, + { url = "https://files.pythonhosted.org/packages/e5/45/8aae2972c520145377ea3559a605a899bebe227bf070b33cdb445929a9b9/cryptography-50.0.0-cp314-cp314t-manylinux_2_34_x86_64.whl", hash = "sha256:6ba6a53445bd3cfa809ef3ef5f1589aa6ba08784a1d962bf47d0940e871dab1c", size = 4716439, upload-time = "2026-07-31T14:24:16.415Z" }, + { url = "https://files.pythonhosted.org/packages/7b/20/4fe50b619a48c2525cc46e2dbc1ac490708d704be5d467bdaac6dc955682/cryptography-50.0.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:3f5735ffe4996d28b809371756219f5354864902a3b9e7c0b9ee87041209fc9c", size = 4837383, upload-time = "2026-07-31T14:24:18.553Z" }, + { url = "https://files.pythonhosted.org/packages/92/91/3a31366e183343d3703f8995c095f5734676bd6938118047e50fcf279eb4/cryptography-50.0.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:1b4a266766514614f8aa60416e71f2fc6e575d36e7bdc90f644fadb2f4b75b95", size = 4985772, upload-time = "2026-07-31T14:24:20.385Z" }, + { url = "https://files.pythonhosted.org/packages/74/9a/02ffe35b2853d121689871eb5dce862092562b3a1ed5cc98f1aaed441506/cryptography-50.0.0-cp314-cp314t-win_amd64.whl", hash = "sha256:12b9c6996425c76ea6c457ace4f3073e715b8c545add07cd1a8f3a4f90691269", size = 3816291, upload-time = "2026-07-31T14:24:22.125Z" }, + { url = "https://files.pythonhosted.org/packages/03/37/73d005be173aff344af30e9fd2a576575cb2391a7101d9cd3842e1fa8cce/cryptography-50.0.0-cp39-abi3-macosx_11_0_arm64.whl", hash = "sha256:ccdc4a71a4dabae05de219404f9f4abc38e3b58422177ff93d0da05967dafa07", size = 4036009, upload-time = "2026-07-31T14:24:24.122Z" }, + { url = "https://files.pythonhosted.org/packages/ff/c6/7a6202a534e32103a285b7834a120869557fe198d51d7cfe59754c8bda9c/cryptography-50.0.0-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:910e1d2668e7de9648f2bcee30e180db2a6b15c30f887d7c4c93ddf96e3992e3", size = 4745252, upload-time = "2026-07-31T14:24:26.118Z" }, + { url = "https://files.pythonhosted.org/packages/85/4f/0fa8c2f4428198f15d9ff8d63400e27afbf94ce833f6108da1eb3753f945/cryptography-50.0.0-cp39-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:a91296cb61e8df6f86d0c19cc4068228da256bf59bf86049fbd821084565327f", size = 4728939, upload-time = "2026-07-31T14:24:27.994Z" }, + { url = "https://files.pythonhosted.org/packages/d1/63/54dd723490ba2dc09b299682c10b38db38f159728bcaae8c591b8af2f22d/cryptography-50.0.0-cp39-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:e722f16708d854fe924790e051061f6704a472c3bac347b6fd88033ea8dd0dc5", size = 4748483, upload-time = "2026-07-31T14:24:30.254Z" }, + { url = "https://files.pythonhosted.org/packages/1d/dd/7c77d26285cc7f6991efce64a0f5b4f9383bfa5dd8c5033003eaf7db4cdb/cryptography-50.0.0-cp39-abi3-manylinux_2_28_ppc64le.whl", hash = "sha256:d764dcf130c428ef66786f866dd750f53182bc608813489915e9fc106bb0c82f", size = 5367599, upload-time = "2026-07-31T14:24:32.457Z" }, + { url = "https://files.pythonhosted.org/packages/46/c9/f60aed34c013f317f92817b6c171c2d22a78270fa41109bd4b08af26b194/cryptography-50.0.0-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:105110f43a471dbd0060b9c9516cb8a6a79233631a04cc2ba16f28323ac6e025", size = 4762647, upload-time = "2026-07-31T14:24:34.599Z" }, + { url = "https://files.pythonhosted.org/packages/be/f3/f9a0173b139372c3a48ed98154b45cc6b9de17c789d5ab552e621c293609/cryptography-50.0.0-cp39-abi3-manylinux_2_31_armv7l.whl", hash = "sha256:828743d939e9629bc267b8e2d08d8bb67cd4319c771a33d4b18b22dd8fb7440a", size = 4385197, upload-time = "2026-07-31T14:24:36.647Z" }, + { url = "https://files.pythonhosted.org/packages/d8/36/83bb81f6e569bc38e1e4a7bc80f29b46bb9601920bc455fc8e888f5d5742/cryptography-50.0.0-cp39-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:2a8183b489dc1f7f80f135780fadc1108f14b31b8a40411c7a5b17425f65f28b", size = 4748095, upload-time = "2026-07-31T14:24:39.493Z" }, + { url = "https://files.pythonhosted.org/packages/6b/16/d3008eff98c764979865834c3d386d4fd041b5f52e7f34fc29ac1a5eb515/cryptography-50.0.0-cp39-abi3-manylinux_2_34_ppc64le.whl", hash = "sha256:6e7d61120573a7f2cd94cc095f9e81f6967c61ccdf194285aa143ecec8e0b708", size = 5325948, upload-time = "2026-07-31T14:24:41.556Z" }, + { url = "https://files.pythonhosted.org/packages/9c/f8/d97f9603efda3888187bfdb893f26c41be4735c10631d05d284ee6b047c4/cryptography-50.0.0-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:37fdb0d0111f1e2ff07139dfb79f1b49531f8e213c46f1163dd7642979b58c47", size = 4762400, upload-time = "2026-07-31T14:24:43.636Z" }, + { url = "https://files.pythonhosted.org/packages/64/a2/4615c8f7d81a00b1d6e6afe19f694e1543582349fb5f4076f6cb5dc36485/cryptography-50.0.0-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:c87f62a3d3b9888ed0fdde100ec06aa61ca9cd44bad9057d1dff9a516b5f5bb9", size = 4878208, upload-time = "2026-07-31T14:24:45.522Z" }, + { url = "https://files.pythonhosted.org/packages/d2/1a/efcfb02f91407149a0dacffffab791f7e19bf6385f63b3666dc8b5e5c9c8/cryptography-50.0.0-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:65c2c3add92b45fd0709db8594536aea39c2a67af0e27ffcf049c498501140b7", size = 5037050, upload-time = "2026-07-31T14:24:47.697Z" }, + { url = "https://files.pythonhosted.org/packages/57/30/4a22984d4f1bdfb8c054f07a92bc176b97a3134cc1d6c4b3bffb1f3688b4/cryptography-50.0.0-cp39-abi3-win_amd64.whl", hash = "sha256:d24fead1d4d076e1bfb006dcec392074a3cd8d7b4fc8a595aa64073b2b7a96ba", size = 3874135, upload-time = "2026-07-31T14:24:50.085Z" }, +] + [[package]] name = "dbus-next" version = "0.2.3" @@ -700,6 +769,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/88/b2/d0896bdcdc8d28a7fc5717c305f1a861c26e18c05047949fb371034d98bd/nodeenv-1.10.0-py2.py3-none-any.whl", hash = "sha256:5bb13e3eed2923615535339b3c620e76779af4cb4c6a90deccc9e36b274d3827", size = 23438, upload-time = "2025-12-20T14:08:52.782Z" }, ] +[[package]] +name = "noiseprotocol" +version = "0.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cryptography" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/76/17/fcf8a90dcf36fe00b475e395f34d92f42c41379c77b25a16066f63002f95/noiseprotocol-0.3.1.tar.gz", hash = "sha256:b092a871b60f6a8f07f17950dc9f7098c8fe7d715b049bd4c24ee3752b90d645", size = 16890, upload-time = "2020-11-25T19:06:48.938Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9d/e1/76e4694201d67b93a6f1644b2588b4a3d965419fe189416e3496cf415db5/noiseprotocol-0.3.1-py3-none-any.whl", hash = "sha256:2e1a603a38439636cf0ffd8b3e8b12cee27d368a28b41be7dbe568b2abb23111", size = 20546, upload-time = "2020-03-03T18:51:28.095Z" }, +] + [[package]] name = "numpy" version = "2.4.4" @@ -1276,7 +1357,7 @@ name = "sendspin" version = "0.0.0" source = { editable = "." } dependencies = [ - { name = "aiosendspin", extra = ["server"] }, + { name = "aiosendspin", extra = ["server", "source"] }, { name = "aiosendspin-mpris" }, { name = "av" }, { name = "numpy" }, @@ -1305,7 +1386,7 @@ test = [ [package.metadata] requires-dist = [ - { name = "aiosendspin", extras = ["server"], specifier = "~=6.0.1" }, + { name = "aiosendspin", extras = ["server", "source"], specifier = "~=9.1.0" }, { name = "aiosendspin-mpris", specifier = "~=2.1.1" }, { name = "av", specifier = ">=15.0.0" }, { name = "codespell", marker = "extra == 'test'", specifier = "==2.4.1" }, From cc54fd96655920050f0aa7846195385051daa922 Mon Sep 17 00:00:00 2001 From: Jorge Pulido Date: Tue, 18 Aug 2026 08:57:31 +0200 Subject: [PATCH 4/6] fix: type disconnect cleanup callback --- sendspin/daemon/daemon.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/sendspin/daemon/daemon.py b/sendspin/daemon/daemon.py index 6112e02..ad39532 100644 --- a/sendspin/daemon/daemon.py +++ b/sendspin/daemon/daemon.py @@ -205,7 +205,11 @@ async def _run_server_initiated(self) -> None: client = self._create_client() self._attach_client(client) - client.add_disconnect_listener(lambda: create_task(self._handle_disconnect())) + + def handle_disconnect() -> None: + create_task(self._handle_disconnect()) + + client.add_disconnect_listener(handle_disconnect) self._listener = ClientListener( client_id=self._args.client_id, From 55439c6930db0b51df569107220ce7f22c9743ef Mon Sep 17 00:00:00 2001 From: Jorge Pulido Date: Tue, 18 Aug 2026 14:30:22 +0200 Subject: [PATCH 5/6] feat(source): support raw ALSA capture devices --- README.md | 5 +++- sendspin/audio_devices.py | 41 ++++++++++++++++++++++++++-- sendspin/cli.py | 29 ++++++++++++++------ sendspin/source_stream.py | 54 ++++++++++++++++++++++++++++++++----- tests/test_audio_devices.py | 18 ++++++++++++- tests/test_source_stream.py | 19 +++++++++++++ 6 files changed, 148 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 017a10d..ce7475a 100644 --- a/README.md +++ b/README.md @@ -310,6 +310,9 @@ sendspin source # Pick a specific server, input device, and codec sendspin source --url ws://192.168.1.50:8927/sendspin --device 2 --codec flac +# Capture directly from an ALSA PCM device +sendspin source --device hw:CARD=USB,DEV=0 + # Stream a 440 Hz sine test tone (no capture hardware needed) sendspin source --input sine ``` @@ -323,7 +326,7 @@ sendspin audio-devices inputs Key options: - `--input {linein,sine}` — capture from a real input device, or generate a sine test tone. Defaults to `linein`; passing `--device` implies `linein`. -- `--device` — input device index or name (see `audio-devices inputs`). +- `--device` — input device index, name, or raw ALSA PCM name (see `audio-devices inputs`; raw ALSA capture requires `arecord`). - `--codec {pcm,opus,flac}` — codec used to encode captured audio before sending (default `pcm`). Capture is 16-bit. - `--sample-rate` / `--channels` — capture format (default 48000 Hz, 2 channels). - `--line-sense` — report input signal presence to the server via `client/state`; the server may use it to decide when to start/stop the source. diff --git a/sendspin/audio_devices.py b/sendspin/audio_devices.py index 45396b0..9ab12d9 100644 --- a/sendspin/audio_devices.py +++ b/sendspin/audio_devices.py @@ -84,11 +84,20 @@ def query_devices() -> list[AudioDevice]: class InputDevice: """Represents an audio input (capture) device.""" - index: int + index: int | None name: str input_channels: int sample_rate: float is_default: bool + alsa_device_name: str | None = None + + @property + def device_id(self) -> int | str: + """Return the identifier to pass to sounddevice APIs.""" + if self.alsa_device_name is not None: + return self.alsa_device_name + assert self.index is not None + return self.index def query_input_devices() -> list[InputDevice]: @@ -270,9 +279,18 @@ def list_alsa_devices() -> list[tuple[str, str]]: Returns a list of (device_name, description) tuples for output devices. Returns an empty list if aplay is not available. """ + return _list_alsa_devices("aplay") + + +def list_alsa_input_devices() -> list[tuple[str, str]]: + """List ALSA capture PCM devices from ``arecord -L``.""" + return _list_alsa_devices("arecord") + + +def _list_alsa_devices(command: str) -> list[tuple[str, str]]: try: result = subprocess.run( - ["aplay", "-L"], # noqa: S607 + [command, "-L"], # noqa: S607 capture_output=True, text=True, timeout=5, @@ -300,6 +318,25 @@ def list_alsa_devices() -> list[tuple[str, str]]: return devices +def resolve_input_device(device_arg: str | None) -> InputDevice: + """Resolve an input device by index, name prefix, or raw ALSA name.""" + devices = query_input_devices() + if device_arg is None: + device = next((device for device in devices if device.is_default), None) + elif device_arg.isnumeric(): + device = next((device for device in devices if device.index == int(device_arg)), None) + else: + device = next((device for device in devices if device.name.startswith(device_arg)), None) + + if device is None and device_arg is not None and sys.platform.startswith("linux"): + if device_arg in {name for name, _ in list_alsa_input_devices()}: + return InputDevice(None, device_arg, 2, 48000.0, False, device_arg) + + if device is None: + raise ValueError(f"Audio input device '{device_arg or 'default'}' not found.") + return device + + def resolve_audio_device(device_arg: str | None) -> AudioDevice: """Resolve audio device from a CLI argument. diff --git a/sendspin/cli.py b/sendspin/cli.py index 5e0f69e..6823a64 100644 --- a/sendspin/cli.py +++ b/sendspin/cli.py @@ -144,7 +144,7 @@ def list_audio_devices() -> None: def list_input_devices() -> None: """List all available audio input (capture) devices.""" try: - from sendspin.audio_devices import query_input_devices + from sendspin.audio_devices import list_alsa_input_devices, query_input_devices except OSError as e: if "PortAudio library not found" in str(e): print(PORTAUDIO_NOT_FOUND_MESSAGE) @@ -168,6 +168,13 @@ def list_input_devices() -> None: f" Channels: {device.input_channels}, " f"Sample rate: {device.sample_rate} Hz" ) + alsa_devices = list_alsa_input_devices() + if alsa_devices: + print("\nALSA capture devices:") + for name, description in alsa_devices: + print(f" {name}") + if description: + print(f" {description}") if devices: default = next((d for d in devices if d.is_default), devices[0]) print(f"\nTo capture from an input device:\n sendspin source --device {default.index}") @@ -340,7 +347,10 @@ def _add_source_parser(subparsers: argparse._SubParsersAction[argparse.ArgumentP "--device", dest="source_device", default=None, - help="Input device index or name (see 'sendspin audio-devices inputs')", + help=( + "Input device index, name, or raw ALSA device name " + "(see 'sendspin audio-devices inputs')" + ), ) source_parser.add_argument( "--codec", @@ -946,14 +956,17 @@ async def display_pin(pin: str | None) -> None: LOGGER.info("Source client '%s' -> %s (%s, %s)", client_id, url, input_kind, codec.value) capture_task = asyncio.create_task(streamer.run()) + connection_task = asyncio.create_task(_source_connection_loop(client, url, streamer)) try: - await _source_connection_loop(client, url, streamer) + done, _ = await asyncio.wait( + {capture_task, connection_task}, return_when=asyncio.FIRST_COMPLETED + ) + for task in done: + task.result() finally: - capture_task.cancel() - try: - await capture_task - except asyncio.CancelledError: - pass + for task in (capture_task, connection_task): + task.cancel() + await asyncio.gather(capture_task, connection_task, return_exceptions=True) await client.disconnect() await settings.flush() return 0 diff --git a/sendspin/source_stream.py b/sendspin/source_stream.py index a819567..2be36f2 100644 --- a/sendspin/source_stream.py +++ b/sendspin/source_stream.py @@ -13,6 +13,7 @@ import logging import math import struct +import sys from dataclasses import dataclass from typing import TYPE_CHECKING @@ -162,6 +163,16 @@ async def _stream_linein(self) -> None: import sounddevice # noqa: PLC0415 cfg = self._config + if cfg.device is not None: + from sendspin.audio_devices import resolve_input_device # noqa: PLC0415 + + device = resolve_input_device(cfg.device) + if device.alsa_device_name is not None: + await self._stream_alsa(device.alsa_device_name) + return + device_id: int | str | None = device.device_id + else: + device_id = None loop = asyncio.get_running_loop() queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=32) @@ -174,13 +185,12 @@ def _callback(indata: object, _frames: int, _time: object, status: object) -> No except asyncio.QueueFull: logger.debug("Source capture queue full; dropping frame") - device = cfg.device if cfg.device is None else _resolve_device(cfg.device) stream = sounddevice.RawInputStream( samplerate=cfg.sample_rate, channels=cfg.channels, dtype="int16", blocksize=cfg.samples_per_frame, - device=device, + device=device_id, callback=_callback, ) with stream: @@ -194,7 +204,39 @@ def _callback(indata: object, _frames: int, _time: object, status: object) -> No data = await queue.get() await self._send_frame(data) - -def _resolve_device(device: str) -> int | str: - """Resolve an input device argument to a sounddevice identifier.""" - return int(device) if device.isnumeric() else device + async def _stream_alsa(self, device: str) -> None: + """Capture a raw ALSA PCM directly with arecord.""" + cfg = self._config + process = await asyncio.create_subprocess_exec( + "arecord", + "-q", + "-D", + device, + "-t", + "raw", + "-f", + "S16_LE", + "-r", + str(cfg.sample_rate), + "-c", + str(cfg.channels), + stdout=asyncio.subprocess.PIPE, + stderr=sys.stderr, + ) + assert process.stdout is not None + frame_bytes = cfg.samples_per_frame * cfg.channels * 2 + try: + while True: + try: + data = await process.stdout.readexactly(frame_bytes) + except asyncio.IncompleteReadError as exc: + if exc.partial: + logger.warning("Discarding %d trailing ALSA PCM bytes", len(exc.partial)) + break + await self._send_frame(data) + if await process.wait() != 0: + raise RuntimeError(f"arecord exited with status {process.returncode}") + finally: + if process.returncode is None: + process.terminate() + await process.wait() diff --git a/tests/test_audio_devices.py b/tests/test_audio_devices.py index dc52e42..74d7663 100644 --- a/tests/test_audio_devices.py +++ b/tests/test_audio_devices.py @@ -7,7 +7,7 @@ import sounddevice import sendspin.audio_devices as _mod -from sendspin.audio_devices import _try_alsa_device +from sendspin.audio_devices import _try_alsa_device, resolve_input_device def test_try_alsa_device_returns_device_when_portaudio_accepts(): @@ -118,3 +118,19 @@ def test_try_alsa_device_accepts_alsa_only_device(): assert result is not None assert result.alsa_device_name == "bluealsa" + + +def test_resolve_input_device_accepts_alsa_only_device(): + """A capture PCM listed by ALSA can be selected by its raw name.""" + with ( + patch.object(_mod.sys, "platform", "linux"), + patch.object(_mod, "query_input_devices", return_value=[]), + patch.object( + _mod, + "list_alsa_input_devices", + return_value=[("hw:CARD=USB,DEV=0", "USB Audio")], + ), + ): + result = resolve_input_device("hw:CARD=USB,DEV=0") + + assert result.device_id == "hw:CARD=USB,DEV=0" diff --git a/tests/test_source_stream.py b/tests/test_source_stream.py index ad6a767..964539a 100644 --- a/tests/test_source_stream.py +++ b/tests/test_source_stream.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +from unittest.mock import AsyncMock, patch from aiosendspin.models.core import ServerCommandPayload from aiosendspin.models.source import SourceCommandServerPayload @@ -146,3 +147,21 @@ async def test_stop_waits_for_in_progress_start() -> None: assert not streamer.streaming assert client.capture.stops == 1 + + +async def test_alsa_capture_uses_arecord() -> None: + """Raw ALSA names bypass PortAudio and stream arecord's PCM output.""" + streamer, client = _make() + process = AsyncMock() + process.stdout.readexactly = AsyncMock( + side_effect=[b"\x01\x02" * 16, asyncio.IncompleteReadError(b"", 16)] + ) + process.wait = AsyncMock(return_value=0) + process.returncode = 0 + + await streamer._begin_stream() # noqa: SLF001 + with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=process)) as spawn: + await streamer._stream_alsa("hw:CARD=USB,DEV=0") # noqa: SLF001 + + assert spawn.await_args.args[:4] == ("arecord", "-q", "-D", "hw:CARD=USB,DEV=0") + assert client.capture.frames == [b"\x01\x02" * 16] From ec50a94c653e38c8c93e266a1880829d33aeb81c Mon Sep 17 00:00:00 2001 From: Jorge Pulido Date: Tue, 18 Aug 2026 20:57:57 +0200 Subject: [PATCH 6/6] fix(source): align ALSA capture periods --- sendspin/source_stream.py | 2 ++ tests/test_source_stream.py | 1 + 2 files changed, 3 insertions(+) diff --git a/sendspin/source_stream.py b/sendspin/source_stream.py index 2be36f2..9e4b5ab 100644 --- a/sendspin/source_stream.py +++ b/sendspin/source_stream.py @@ -220,6 +220,8 @@ async def _stream_alsa(self, device: str) -> None: str(cfg.sample_rate), "-c", str(cfg.channels), + "--period-size", + str(cfg.samples_per_frame), stdout=asyncio.subprocess.PIPE, stderr=sys.stderr, ) diff --git a/tests/test_source_stream.py b/tests/test_source_stream.py index 964539a..cbaca17 100644 --- a/tests/test_source_stream.py +++ b/tests/test_source_stream.py @@ -164,4 +164,5 @@ async def test_alsa_capture_uses_arecord() -> None: await streamer._stream_alsa("hw:CARD=USB,DEV=0") # noqa: SLF001 assert spawn.await_args.args[:4] == ("arecord", "-q", "-D", "hw:CARD=USB,DEV=0") + assert spawn.await_args.args[-2:] == ("--period-size", "960") assert client.capture.frames == [b"\x01\x02" * 16]