commit 836018d146d070666df550800484ae2522e9aec3 Author: Username Date: Tue Feb 24 11:44:06 2026 +0100 feat: scaffold tuimble TUI mumble client Core modules: TUI app (textual), mumble protocol client, audio pipeline (sounddevice + opus), push-to-talk with kitty protocol / evdev / toggle backends. Config via TOML. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7842b70 --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +__pycache__/ +*.pyc +*.pyo +.venv/ +*.egg-info/ +dist/ +build/ +.eggs/ +*.egg +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..4ad4e00 --- /dev/null +++ b/Makefile @@ -0,0 +1,26 @@ +.PHONY: setup run lint test clean + +VENV := .venv +PIP := $(VENV)/bin/pip +PY := $(VENV)/bin/python + +setup: $(VENV)/bin/activate + +$(VENV)/bin/activate: + python3 -m venv $(VENV) + $(PIP) install --upgrade pip + $(PIP) install -e ".[dev]" + +run: setup + $(PY) -m tuimble + +lint: setup + $(VENV)/bin/ruff check src/ tests/ + $(VENV)/bin/ruff format --check src/ tests/ + +test: setup + $(VENV)/bin/pytest -v + +clean: + rm -rf $(VENV) dist build *.egg-info .pytest_cache .ruff_cache + find . -type d -name __pycache__ -exec rm -rf {} + diff --git a/PROJECT.md b/PROJECT.md new file mode 100644 index 0000000..a3866a0 --- /dev/null +++ b/PROJECT.md @@ -0,0 +1,28 @@ +# tuimble + +## Purpose + +A terminal-based Mumble client with full voice support. Fills the gap +between heavyweight GUI clients and no-client-at-all for terminal users. + +## Success Criteria + +- Connect to any standard Mumble server +- Transmit and receive voice audio +- Push-to-talk with sub-100ms latency +- Browse channels, see users, send/receive text +- Works in Kitty, WezTerm, Ghostty, foot; degrades gracefully elsewhere + +## Constraints + +- Python 3.11+ (async features, tomllib) +- Minimal dependencies (5 core libraries) +- Single-user, single-server (no multi-server) +- Linux-first (evdev fallback is Linux-only) + +## Non-Goals + +- GUI or web interface +- Server administration +- Audio recording/playback to file +- Plugin system diff --git a/README.md b/README.md new file mode 100644 index 0000000..c484d43 --- /dev/null +++ b/README.md @@ -0,0 +1,44 @@ +# tuimble + +TUI Mumble client with voice support and push-to-talk. + +## Features + +- Terminal-based UI (Textual) +- Voice transmission with Opus codec +- Push-to-talk via Kitty keyboard protocol, evdev, or toggle +- Channel browsing and text chat + +## Quick Start + +```sh +make setup +make run +``` + +## Configuration + +```sh +mkdir -p ~/.config/tuimble +``` + +`~/.config/tuimble/config.toml`: + +```toml +[server] +host = "mumble.example.com" +port = 64738 +username = "myname" + +[ptt] +key = "space" +mode = "hold" +backend = "auto" +``` + +## Requirements + +- Python 3.11+ +- libopus +- portaudio (libportaudio2) +- A Mumble server to connect to diff --git a/ROADMAP.md b/ROADMAP.md new file mode 100644 index 0000000..c6bd093 --- /dev/null +++ b/ROADMAP.md @@ -0,0 +1,30 @@ +# Roadmap + +## Phase 1 — Foundation (current) + +- [x] Project scaffold +- [ ] Basic TUI layout (channels, chat, status) +- [ ] Mumble server connection +- [ ] Text chat (send/receive) + +## Phase 2 — Voice + +- [ ] Audio pipeline (capture + playback) +- [ ] Opus encode/decode integration +- [ ] Push-to-talk (Kitty protocol) +- [ ] PTT fallbacks (evdev, toggle) + +## Phase 3 — Polish + +- [ ] Channel tree navigation +- [ ] User list with status indicators +- [ ] Volume control +- [ ] Server certificate handling +- [ ] Config file hot-reload + +## Phase 4 — Robustness + +- [ ] Reconnection handling +- [ ] Error recovery +- [ ] Audio device hot-swap +- [ ] Comprehensive test suite diff --git a/TASKLIST.md b/TASKLIST.md new file mode 100644 index 0000000..f4094ca --- /dev/null +++ b/TASKLIST.md @@ -0,0 +1,12 @@ +# Task List + +## Active + +- [ ] Wire TUI to MumbleClient (connect on startup, display state) +- [ ] Implement text message send/receive in chat log + +## Pending + +- [ ] Audio pipeline integration with MumbleClient +- [ ] PTT wiring (key events -> audio.capturing toggle) +- [ ] Channel tree population from server data diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..1499721 --- /dev/null +++ b/TODO.md @@ -0,0 +1,7 @@ +# TODO + +- Investigate pymumble_py3 vs pymumble fork status +- Check Textual key release event API surface +- Test opuslib availability on Debian 12 +- Determine if sounddevice needs ALSA or PulseAudio config +- Consider TOML config validation (pydantic?) diff --git a/docs/CHEATSHEET.md b/docs/CHEATSHEET.md new file mode 100644 index 0000000..a511ea4 --- /dev/null +++ b/docs/CHEATSHEET.md @@ -0,0 +1,12 @@ +# Cheatsheet + +``` +make setup setup venv + deps +make run launch tuimble +make test run tests +make lint check code style + +q quit +space push-to-talk (hold) +Enter send message +``` diff --git a/docs/DEBUG.md b/docs/DEBUG.md new file mode 100644 index 0000000..ce09d8e --- /dev/null +++ b/docs/DEBUG.md @@ -0,0 +1,45 @@ +# Debugging + +## Verbose Mode + +```sh +tuimble --verbose # (planned) +``` + +## Common Issues + +### No audio devices found + +```sh +python3 -c "import sounddevice; print(sounddevice.query_devices())" +``` + +Check that portaudio is installed and devices are accessible. + +### opuslib import error + +```sh +python3 -c "import opuslib" +``` + +If this fails, install libopus: `apt install libopus0` + +### Kitty protocol not detected + +Verify terminal supports the Kitty keyboard protocol. +Known working: Kitty, WezTerm, Ghostty, foot. + +### evdev permission denied + +```sh +ls -la /dev/input/event* +groups # should include 'input' +``` + +Add user to `input` group: `sudo usermod -aG input $USER` + +### Connection refused + +- Verify server address and port +- Check firewall allows outbound TCP/UDP to port 64738 +- Test with: `nc -zv 64738` diff --git a/docs/INSTALL.md b/docs/INSTALL.md new file mode 100644 index 0000000..5697dce --- /dev/null +++ b/docs/INSTALL.md @@ -0,0 +1,52 @@ +# Installation + +## Prerequisites + +### System Libraries + +```sh +# Debian/Ubuntu +sudo apt install libopus0 libopus-dev portaudio19-dev + +# Fedora/RHEL +sudo dnf install opus opus-devel portaudio-devel + +# Arch +sudo pacman -S opus portaudio +``` + +### Python + +Python 3.11 or later required. + +## Setup + +```sh +git clone ~/git/tuimble +cd ~/git/tuimble +make setup +``` + +## Verify + +```sh +make test +tuimble --version +``` + +## Optional: evdev PTT + +For Linux evdev push-to-talk (physical key detection): + +```sh +pip install tuimble[evdev] +sudo usermod -aG input $USER +# Log out and back in for group change +``` + +## Symlink + +```sh +ln -sf ~/git/tuimble/.venv/bin/tuimble ~/.local/bin/tuimble +which tuimble +``` diff --git a/docs/USAGE.md b/docs/USAGE.md new file mode 100644 index 0000000..1662616 --- /dev/null +++ b/docs/USAGE.md @@ -0,0 +1,27 @@ +# Usage + +## Running + +```sh +tuimble # uses ~/.config/tuimble/config.toml +tuimble --host mumble.example.com --user myname +``` + +## Key Bindings + +| Key | Action | +|-----|--------| +| `q` | Quit | +| `space` | Push-to-talk (configurable) | +| `Enter` | Send message | +| `Ctrl+C` | Quit | + +## Push-to-Talk Modes + +- **hold** — hold key to transmit, release to stop (default) +- **toggle** — press to start, press again to stop + +## Configuration + +See `~/.config/tuimble/config.toml`. All fields are optional; +defaults connect to localhost:64738. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c8fe48e --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,40 @@ +[build-system] +requires = ["setuptools>=68.0", "setuptools-scm"] +build-backend = "setuptools.build_meta" + +[project] +name = "tuimble" +version = "0.1.0" +description = "TUI Mumble client with voice support" +requires-python = ">=3.11" +dependencies = [ + "textual>=1.0.0", + "pymumble>=1.6", + "sounddevice>=0.5.0", + "tomli>=2.0.0;python_version<'3.11'", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "ruff>=0.8.0", +] +evdev = [ + "evdev>=1.7.0", +] + +[project.scripts] +tuimble = "tuimble.__main__:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.ruff] +line-length = 88 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/src/tuimble/__init__.py b/src/tuimble/__init__.py new file mode 100644 index 0000000..fe01c58 --- /dev/null +++ b/src/tuimble/__init__.py @@ -0,0 +1,3 @@ +"""TUI Mumble client with voice support.""" + +__version__ = "0.1.0" diff --git a/src/tuimble/__main__.py b/src/tuimble/__main__.py new file mode 100644 index 0000000..57304f5 --- /dev/null +++ b/src/tuimble/__main__.py @@ -0,0 +1,14 @@ +"""Entry point for tuimble.""" + +import sys + + +def main(): + from tuimble.app import TuimbleApp + + app = TuimbleApp() + app.run() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/tuimble/app.py b/src/tuimble/app.py new file mode 100644 index 0000000..c84c2cb --- /dev/null +++ b/src/tuimble/app.py @@ -0,0 +1,133 @@ +"""TUI application built on Textual.""" + +from __future__ import annotations + +from textual import events +from textual.app import App, ComposeResult +from textual.containers import Horizontal, Vertical +from textual.reactive import reactive +from textual.widgets import Footer, Header, Input, RichLog, Static + +from tuimble.config import Config, load_config +from tuimble.ptt import KittyPtt, TogglePtt, detect_backend + + +class StatusBar(Static): + """Connection and PTT status indicator.""" + + ptt_active = reactive(False) + connected = reactive(False) + + def render(self) -> str: + if self.connected: + conn = "\x1b[38;2;158;206;106m\u25cf connected\x1b[0m" + else: + conn = "\x1b[38;2;247;118;142m\u25cb disconnected\x1b[0m" + if self.ptt_active: + ptt = "\x1b[38;2;224;175;104m\u25cf TX\x1b[0m" + else: + ptt = "\x1b[38;2;86;95;137m\u25cb idle\x1b[0m" + return f" {conn} {ptt}" + + +class ChannelTree(Static): + """Channel and user list.""" + + def render(self) -> str: + hdr = "\x1b[38;2;169;177;214m Channels\x1b[0m" + empty = "\x1b[38;2;86;95;137m \u2514\u2500 (not connected)\x1b[0m" + return f"{hdr}\n{empty}" + + +class ChatLog(RichLog): + """Message log.""" + pass + + +class TuimbleApp(App): + """Main application.""" + + TITLE = "tuimble" + CSS = """ + Screen { + background: #1a1a2e; + color: #a9b1d6; + } + #main { + height: 1fr; + } + #sidebar { + width: 24; + border-right: solid #292e42; + padding: 0 1; + } + #chat-area { + width: 1fr; + } + #chatlog { + height: 1fr; + border-bottom: solid #292e42; + scrollbar-size: 1 1; + } + #input { + dock: bottom; + height: 3; + border-top: solid #292e42; + } + #status { + dock: bottom; + height: 1; + background: #16161e; + } + Footer { + background: #16161e; + } + Header { + background: #16161e; + } + """ + + BINDINGS = [ + ("q", "quit", "Quit"), + ("ctrl+c", "quit", "Quit"), + ] + + def __init__(self): + super().__init__() + self._config: Config = load_config() + self._ptt = detect_backend(self._on_ptt_change, self._config.ptt.backend) + + def compose(self) -> ComposeResult: + yield Header() + with Horizontal(id="main"): + yield ChannelTree(id="sidebar") + with Vertical(id="chat-area"): + yield ChatLog(id="chatlog") + yield Input(placeholder="message...", id="input") + yield StatusBar(id="status") + yield Footer() + + def on_mount(self): + chatlog = self.query_one("#chatlog", ChatLog) + chatlog.write("[dim]tuimble v0.1.0[/dim]") + chatlog.write("[dim]press q to quit[/dim]") + + def on_key(self, event: events.Key) -> None: + """Handle PTT key events.""" + ptt_key = self._config.ptt.key + + if isinstance(self._ptt, KittyPtt): + if event.key == ptt_key: + is_release = getattr(event, "key_type", None) + if is_release == "release": + self._ptt.key_up() + else: + self._ptt.key_down() + elif isinstance(self._ptt, TogglePtt): + if event.key == ptt_key: + self._ptt.toggle() + + def _on_ptt_change(self, transmitting: bool): + """Called when PTT state changes.""" + status = self.query_one("#status", StatusBar) + status.ptt_active = transmitting diff --git a/src/tuimble/audio.py b/src/tuimble/audio.py new file mode 100644 index 0000000..1262a6b --- /dev/null +++ b/src/tuimble/audio.py @@ -0,0 +1,131 @@ +"""Audio capture and playback pipeline. + +Handles microphone input, speaker output, and Opus encoding/decoding. +Designed to run alongside the async event loop via callback-based I/O. +""" + +from __future__ import annotations + +import logging +import queue + +log = logging.getLogger(__name__) + +SAMPLE_RATE = 48000 +CHANNELS = 1 +FRAME_SIZE = 960 # 20ms at 48kHz +DTYPE = "int16" + + +class AudioPipeline: + """Manages audio input/output streams and Opus codec.""" + + def __init__( + self, + sample_rate: int = SAMPLE_RATE, + frame_size: int = FRAME_SIZE, + input_device: int | None = None, + output_device: int | None = None, + ): + self._sample_rate = sample_rate + self._frame_size = frame_size + self._input_device = input_device + self._output_device = output_device + + self._capture_queue: queue.Queue[bytes] = queue.Queue(maxsize=50) + self._playback_queue: queue.Queue[bytes] = queue.Queue(maxsize=50) + + self._encoder = None + self._decoder = None + self._input_stream = None + self._output_stream = None + self._capturing = False + + def start(self): + """Initialize codec and open audio streams.""" + import opuslib + import sounddevice as sd + + self._encoder = opuslib.Encoder( + self._sample_rate, CHANNELS, opuslib.APPLICATION_VOIP + ) + self._decoder = opuslib.Decoder(self._sample_rate, CHANNELS) + + self._output_stream = sd.RawOutputStream( + samplerate=self._sample_rate, + channels=CHANNELS, + dtype=DTYPE, + device=self._output_device, + blocksize=self._frame_size, + callback=self._playback_callback, + ) + self._output_stream.start() + + self._input_stream = sd.RawInputStream( + samplerate=self._sample_rate, + channels=CHANNELS, + dtype=DTYPE, + device=self._input_device, + blocksize=self._frame_size, + callback=self._capture_callback, + ) + self._input_stream.start() + + log.info("audio pipeline started (rate=%d)", self._sample_rate) + + def stop(self): + """Close audio streams.""" + for stream in (self._input_stream, self._output_stream): + if stream is not None: + stream.stop() + stream.close() + self._input_stream = None + self._output_stream = None + log.info("audio pipeline stopped") + + @property + def capturing(self) -> bool: + return self._capturing + + @capturing.setter + def capturing(self, value: bool): + self._capturing = value + + def _capture_callback(self, indata, frames, time_info, status): + """Called by sounddevice when input data is available.""" + if status: + log.warning("capture status: %s", status) + if self._capturing and self._encoder: + try: + encoded = self._encoder.encode(bytes(indata), self._frame_size) + self._capture_queue.put_nowait(encoded) + except queue.Full: + pass + + def _playback_callback(self, outdata, frames, time_info, status): + """Called by sounddevice when output buffer needs data.""" + if status: + log.warning("playback status: %s", status) + try: + data = self._playback_queue.get_nowait() + if self._decoder: + pcm = self._decoder.decode(data, self._frame_size) + outdata[:] = pcm[: len(outdata)] + else: + outdata[:] = b"\x00" * len(outdata) + except queue.Empty: + outdata[:] = b"\x00" * len(outdata) + + def get_encoded_frame(self) -> bytes | None: + """Retrieve next encoded frame for transmission.""" + try: + return self._capture_queue.get_nowait() + except queue.Empty: + return None + + def queue_playback(self, opus_data: bytes): + """Queue encoded audio data for playback.""" + try: + self._playback_queue.put_nowait(opus_data) + except queue.Full: + pass diff --git a/src/tuimble/client.py b/src/tuimble/client.py new file mode 100644 index 0000000..971234f --- /dev/null +++ b/src/tuimble/client.py @@ -0,0 +1,156 @@ +"""Mumble protocol client. + +Wraps pymumble to provide an async-friendly interface for connecting +to Mumble servers, handling channels, users, and voice data. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass + +log = logging.getLogger(__name__) + + +@dataclass +class User: + session_id: int + name: str + channel_id: int + mute: bool = False + deaf: bool = False + self_mute: bool = False + self_deaf: bool = False + + +@dataclass +class Channel: + channel_id: int + name: str + parent_id: int + description: str = "" + + +class MumbleClient: + """Async wrapper around pymumble.""" + + def __init__( + self, + host: str, + port: int = 64738, + username: str = "tuimble-user", + password: str = "", + ): + self._host = host + self._port = port + self._username = username + self._password = password + self._mumble = None + self._connected = False + self._loop: asyncio.AbstractEventLoop | None = None + + self.on_text_message = None # callback(sender, message) + self.on_user_state = None # callback(user) + self.on_sound_received = None # callback(user, pcm_data) + + @property + def connected(self) -> bool: + return self._connected + + @property + def users(self) -> dict[int, User]: + if not self._mumble: + return {} + result = {} + for sid, u in self._mumble.users.items(): + result[sid] = User( + session_id=sid, + name=u["name"], + channel_id=u.get("channel_id", 0), + mute=u.get("mute", False), + deaf=u.get("deaf", False), + self_mute=u.get("self_mute", False), + self_deaf=u.get("self_deaf", False), + ) + return result + + @property + def channels(self) -> dict[int, Channel]: + if not self._mumble: + return {} + result = {} + for cid, ch in self._mumble.channels.items(): + result[cid] = Channel( + channel_id=cid, + name=ch["name"], + parent_id=ch.get("parent", 0), + description=ch.get("description", ""), + ) + return result + + async def connect(self): + """Connect to the Mumble server.""" + import pymumble_py3 as pymumble + + self._loop = asyncio.get_running_loop() + self._mumble = pymumble.Mumble( + self._host, + self._username, + port=self._port, + password=self._password, + reconnect=True, + ) + self._mumble.set_codec_profile("audio") + self._mumble.callbacks.set_callback( + pymumble.constants.PYMUMBLE_CLBK_TEXTMESSAGERECEIVED, + self._on_text_message, + ) + self._mumble.callbacks.set_callback( + pymumble.constants.PYMUMBLE_CLBK_SOUNDRECEIVED, + self._on_sound_received, + ) + + # pymumble runs its own thread + self._mumble.start() + self._mumble.is_ready() + self._connected = True + log.info("connected to %s:%d as %s", self._host, self._port, self._username) + + async def disconnect(self): + """Disconnect from the server.""" + if self._mumble: + self._mumble.stop() + self._connected = False + log.info("disconnected") + + def send_text(self, message: str, channel: bool = True): + """Send a text message to the current channel.""" + if self._mumble and self._connected: + ch = self._mumble.channels[self._mumble.users.myself["channel_id"]] + ch.send_text_message(message) + + def send_audio(self, opus_data: bytes): + """Send encoded audio to the server.""" + if self._mumble and self._connected: + self._mumble.sound_output.add_sound(opus_data) + + def join_channel(self, channel_id: int): + """Move to a different channel.""" + if self._mumble and self._connected: + self._mumble.channels[channel_id].move_in() + + def _on_text_message(self, message): + if self.on_text_message and self._loop: + actor = message.actor + users = self._mumble.users + name = users[actor]["name"] if actor in users else "?" + self._loop.call_soon_threadsafe( + self.on_text_message, name, message.message + ) + + def _on_sound_received(self, user, sound_chunk): + if self.on_sound_received and self._loop: + self._loop.call_soon_threadsafe( + self.on_sound_received, user, sound_chunk.pcm + ) diff --git a/src/tuimble/config.py b/src/tuimble/config.py new file mode 100644 index 0000000..37900e7 --- /dev/null +++ b/src/tuimble/config.py @@ -0,0 +1,64 @@ +"""Configuration management.""" + +from dataclasses import dataclass, field +from pathlib import Path + +CONFIG_DIR = Path.home() / ".config" / "tuimble" +CONFIG_FILE = CONFIG_DIR / "config.toml" + +DEFAULT_PORT = 64738 + + +@dataclass +class ServerConfig: + host: str = "localhost" + port: int = DEFAULT_PORT + username: str = "tuimble-user" + password: str = "" + channel: str = "" + + +@dataclass +class AudioConfig: + input_device: int | None = None + output_device: int | None = None + sample_rate: int = 48000 + frame_size: int = 960 # 20ms at 48kHz + + +@dataclass +class PttConfig: + key: str = "space" + mode: str = "hold" # hold | toggle + backend: str = "auto" # auto | kitty | evdev | toggle + + +@dataclass +class Config: + server: ServerConfig = field(default_factory=ServerConfig) + audio: AudioConfig = field(default_factory=AudioConfig) + ptt: PttConfig = field(default_factory=PttConfig) + + +def load_config(path: Path | None = None) -> Config: + """Load configuration from TOML file, falling back to defaults.""" + path = path or CONFIG_FILE + if not path.exists(): + return Config() + + try: + import tomllib + except ImportError: + import tomli as tomllib + + with open(path, "rb") as f: + data = tomllib.load(f) + + cfg = Config() + if "server" in data: + cfg.server = ServerConfig(**data["server"]) + if "audio" in data: + cfg.audio = AudioConfig(**data["audio"]) + if "ptt" in data: + cfg.ptt = PttConfig(**data["ptt"]) + return cfg diff --git a/src/tuimble/ptt.py b/src/tuimble/ptt.py new file mode 100644 index 0000000..e64a706 --- /dev/null +++ b/src/tuimble/ptt.py @@ -0,0 +1,144 @@ +"""Push-to-talk input handling. + +Backend priority: + 1. Kitty keyboard protocol (terminal-native, no extra deps) + 2. evdev (Linux /dev/input, requires 'input' group) + 3. Toggle mode (press-on / press-off, universal fallback) +""" + +from __future__ import annotations + +import asyncio +import enum +import logging +from abc import ABC, abstractmethod +from typing import Callable + +log = logging.getLogger(__name__) + +Callback = Callable[[bool], None] # True = transmitting, False = idle + + +class PttState(enum.Enum): + IDLE = "idle" + TRANSMITTING = "transmitting" + + +class PttBackend(ABC): + """Base class for push-to-talk backends.""" + + def __init__(self, callback: Callback): + self._callback = callback + self._state = PttState.IDLE + + @property + def state(self) -> PttState: + return self._state + + @property + def transmitting(self) -> bool: + return self._state is PttState.TRANSMITTING + + def _set_state(self, transmitting: bool): + new = PttState.TRANSMITTING if transmitting else PttState.IDLE + if new != self._state: + self._state = new + self._callback(transmitting) + + @abstractmethod + async def start(self): ... + + @abstractmethod + async def stop(self): ... + + +class KittyPtt(PttBackend): + """PTT via Kitty keyboard protocol key release events. + + Relies on the TUI framework (Textual) forwarding key_down/key_up. + This backend is driven externally by the app's key event handlers. + """ + + async def start(self): + log.info("kitty protocol PTT active") + + async def stop(self): + self._set_state(False) + + def key_down(self): + self._set_state(True) + + def key_up(self): + self._set_state(False) + + +class EvdevPtt(PttBackend): + """PTT via Linux evdev input subsystem.""" + + def __init__(self, callback: Callback, key_code: int = 57): + super().__init__(callback) + self._key_code = key_code # 57 = KEY_SPACE + self._task: asyncio.Task | None = None + + async def start(self): + try: + import evdev + except ImportError: + raise RuntimeError("evdev not installed; pip install tuimble[evdev]") + + devices = [evdev.InputDevice(p) for p in evdev.list_devices()] + kbd = None + for dev in devices: + caps = dev.capabilities(verbose=False) + if 1 in caps and self._key_code in caps[1]: + kbd = dev + break + + if kbd is None: + raise RuntimeError("no suitable input device found") + + log.info("evdev PTT active on %s", kbd.path) + self._task = asyncio.create_task(self._poll(kbd)) + + async def _poll(self, dev): + import evdev + + async for event in dev.async_read_loop(): + if event.type == evdev.ecodes.EV_KEY and event.code == self._key_code: + if event.value == 1: # press + self._set_state(True) + elif event.value == 0: # release + self._set_state(False) + + async def stop(self): + if self._task: + self._task.cancel() + self._task = None + self._set_state(False) + + +class TogglePtt(PttBackend): + """PTT via toggle: press once to start, press again to stop.""" + + async def start(self): + log.info("toggle PTT active") + + async def stop(self): + self._set_state(False) + + def toggle(self): + self._set_state(not self.transmitting) + + +def detect_backend(callback: Callback, preference: str = "auto") -> PttBackend: + """Select the best available PTT backend.""" + if preference == "kitty": + return KittyPtt(callback) + if preference == "evdev": + return EvdevPtt(callback) + if preference == "toggle": + return TogglePtt(callback) + + # auto: try kitty first (will be validated at runtime by the app), + # then evdev, then toggle + return KittyPtt(callback) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..0754580 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,23 @@ +"""Tests for configuration module.""" + +from tuimble.config import Config, PttConfig, ServerConfig + + +def test_default_config(): + cfg = Config() + assert cfg.server.host == "localhost" + assert cfg.server.port == 64738 + assert cfg.audio.sample_rate == 48000 + assert cfg.ptt.mode == "hold" + + +def test_server_config(): + srv = ServerConfig(host="mumble.example.com", port=12345) + assert srv.host == "mumble.example.com" + assert srv.port == 12345 + + +def test_ptt_config_defaults(): + ptt = PttConfig() + assert ptt.key == "space" + assert ptt.backend == "auto" diff --git a/tests/test_ptt.py b/tests/test_ptt.py new file mode 100644 index 0000000..b8f517f --- /dev/null +++ b/tests/test_ptt.py @@ -0,0 +1,37 @@ +"""Tests for push-to-talk backends.""" + +from tuimble.ptt import KittyPtt, PttState, TogglePtt + + +def test_kitty_ptt_press_release(): + states = [] + ptt = KittyPtt(lambda tx: states.append(tx)) + assert ptt.state is PttState.IDLE + + ptt.key_down() + assert ptt.state is PttState.TRANSMITTING + assert states == [True] + + ptt.key_up() + assert ptt.state is PttState.IDLE + assert states == [True, False] + + +def test_kitty_ptt_no_duplicate_events(): + states = [] + ptt = KittyPtt(lambda tx: states.append(tx)) + + ptt.key_down() + ptt.key_down() # repeat, no state change + assert states == [True] + + +def test_toggle_ptt(): + states = [] + ptt = TogglePtt(lambda tx: states.append(tx)) + + ptt.toggle() + assert ptt.transmitting is True + ptt.toggle() + assert ptt.transmitting is False + assert states == [True, False]