| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142 |
- """Interactive device picker using questionary or CLI fallback."""
- import sys
- from typing import Iterable
- from .pulse import Device
- def _build_choices(devices: list[Device], preselected: set[str]) -> list:
- """Build questionary checkbox choices.
- questionary preselects via ``Choice(..., checked=True)``; passing a
- ``default`` list to ``checkbox`` is rejected by its validator.
- """
- import questionary
- return [
- questionary.Choice(
- f"[{'OUT' if d.kind == 'output' else 'IN'}] {d.description} ({d.name})",
- value=d.name,
- checked=d.name in preselected,
- )
- for d in devices
- ]
- def select_sources(devices: list[Device], preselected: Iterable[str] = ()) -> list[str]:
- """Interactively select audio sources.
- If questionary is available and stdin is a TTY, show a checkbox UI.
- Otherwise, show a numbered menu.
- Args:
- devices: List of available Device objects.
- preselected: Source names that should be pre-checked.
- Returns:
- List of selected source names in device order, deduplicated.
- Raises:
- RuntimeError: If non-TTY and user provides invalid input with no valid context.
- """
- preselected_set = set(preselected)
- # Try questionary UI first
- if sys.stdin.isatty():
- try:
- import questionary
- except ImportError:
- questionary = None
- if questionary is not None:
- try:
- selected = questionary.checkbox(
- "Select audio sources to capture:",
- choices=_build_choices(devices, preselected_set),
- ).ask()
- except KeyboardInterrupt:
- return list(preselected)
- if selected is None:
- # User aborted (Ctrl+C)
- return list(preselected)
- # Preserve device order
- result = []
- seen = set()
- for device in devices:
- if device.name in selected and device.name not in seen:
- result.append(device.name)
- seen.add(device.name)
- return result
- # CLI fallback: show numbered menu and read input
- print("\nAvailable audio sources:")
- for i, device in enumerate(devices, 1):
- kind_label = "OUT" if device.kind == "output" else "IN"
- marker = "✓" if device.name in preselected_set else " "
- print(f" {marker} {i:2}. [{kind_label}] {device.description}")
- while True:
- prompt = (
- "\nSelect sources by number (space/comma-separated), "
- "'all', 'none', or blank to keep current:"
- )
- print(prompt)
- try:
- line = input("> ").strip()
- except EOFError:
- raise RuntimeError(
- "No TTY and EOF reached. Please provide --sources with comma-separated names "
- "(e.g. --sources 'alsa_input.usb-...,alsa_output.usb-...')"
- )
- if not line:
- # Keep preselected
- result = []
- seen = set()
- for device in devices:
- if device.name in preselected_set and device.name not in seen:
- result.append(device.name)
- seen.add(device.name)
- return result
- if line.lower() == "all":
- return [d.name for d in devices]
- if line.lower() == "none":
- return []
- # Parse indices
- try:
- # Accept space or comma as separator
- indices_str = line.replace(",", " ")
- indices = [int(x) for x in indices_str.split()]
- # Validate all indices first
- selected_indices = set()
- for idx in indices:
- if 1 <= idx <= len(devices):
- selected_indices.add(idx)
- else:
- print(
- f"Invalid index: {idx}. "
- f"Please enter numbers between 1 and {len(devices)}."
- )
- break
- else:
- # All indices were valid; return devices in device list order
- result = []
- seen = set()
- for i, device in enumerate(devices, 1):
- if i in selected_indices and device.name not in seen:
- result.append(device.name)
- seen.add(device.name)
- return result
- except ValueError:
- print(
- "Invalid input. Please enter space/comma-separated numbers, "
- "'all', 'none', or blank."
- )
|