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)