From 836018d146d070666df550800484ae2522e9aec3 Mon Sep 17 00:00:00 2001 From: Username Date: Tue, 24 Feb 2026 11:44:06 +0100 Subject: [PATCH] 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. --- .gitignore | 12 ++++ Makefile | 26 +++++++ PROJECT.md | 28 ++++++++ README.md | 44 ++++++++++++ ROADMAP.md | 30 ++++++++ TASKLIST.md | 12 ++++ TODO.md | 7 ++ docs/CHEATSHEET.md | 12 ++++ docs/DEBUG.md | 45 ++++++++++++ docs/INSTALL.md | 52 ++++++++++++++ docs/USAGE.md | 27 +++++++ pyproject.toml | 40 +++++++++++ src/tuimble/__init__.py | 3 + src/tuimble/__main__.py | 14 ++++ src/tuimble/app.py | 133 ++++++++++++++++++++++++++++++++++ src/tuimble/audio.py | 131 +++++++++++++++++++++++++++++++++ src/tuimble/client.py | 156 ++++++++++++++++++++++++++++++++++++++++ src/tuimble/config.py | 64 +++++++++++++++++ src/tuimble/ptt.py | 144 +++++++++++++++++++++++++++++++++++++ tests/__init__.py | 0 tests/test_config.py | 23 ++++++ tests/test_ptt.py | 37 ++++++++++ 22 files changed, 1040 insertions(+) create mode 100644 .gitignore create mode 100644 Makefile create mode 100644 PROJECT.md create mode 100644 README.md create mode 100644 ROADMAP.md create mode 100644 TASKLIST.md create mode 100644 TODO.md create mode 100644 docs/CHEATSHEET.md create mode 100644 docs/DEBUG.md create mode 100644 docs/INSTALL.md create mode 100644 docs/USAGE.md create mode 100644 pyproject.toml create mode 100644 src/tuimble/__init__.py create mode 100644 src/tuimble/__main__.py create mode 100644 src/tuimble/app.py create mode 100644 src/tuimble/audio.py create mode 100644 src/tuimble/client.py create mode 100644 src/tuimble/config.py create mode 100644 src/tuimble/ptt.py create mode 100644 tests/__init__.py create mode 100644 tests/test_config.py create mode 100644 tests/test_ptt.py 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]