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.
This commit is contained in:
12
.gitignore
vendored
Normal file
12
.gitignore
vendored
Normal file
@@ -0,0 +1,12 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.pyo
|
||||
.venv/
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
.eggs/
|
||||
*.egg
|
||||
.pytest_cache/
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
26
Makefile
Normal file
26
Makefile
Normal file
@@ -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 {} +
|
||||
28
PROJECT.md
Normal file
28
PROJECT.md
Normal file
@@ -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
|
||||
44
README.md
Normal file
44
README.md
Normal file
@@ -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
|
||||
30
ROADMAP.md
Normal file
30
ROADMAP.md
Normal file
@@ -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
|
||||
12
TASKLIST.md
Normal file
12
TASKLIST.md
Normal file
@@ -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
|
||||
7
TODO.md
Normal file
7
TODO.md
Normal file
@@ -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?)
|
||||
12
docs/CHEATSHEET.md
Normal file
12
docs/CHEATSHEET.md
Normal file
@@ -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
|
||||
```
|
||||
45
docs/DEBUG.md
Normal file
45
docs/DEBUG.md
Normal file
@@ -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 <host> 64738`
|
||||
52
docs/INSTALL.md
Normal file
52
docs/INSTALL.md
Normal file
@@ -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 <repo-url> ~/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
|
||||
```
|
||||
27
docs/USAGE.md
Normal file
27
docs/USAGE.md
Normal file
@@ -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.
|
||||
40
pyproject.toml
Normal file
40
pyproject.toml
Normal file
@@ -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"]
|
||||
3
src/tuimble/__init__.py
Normal file
3
src/tuimble/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""TUI Mumble client with voice support."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
14
src/tuimble/__main__.py
Normal file
14
src/tuimble/__main__.py
Normal file
@@ -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())
|
||||
133
src/tuimble/app.py
Normal file
133
src/tuimble/app.py
Normal file
@@ -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
|
||||
131
src/tuimble/audio.py
Normal file
131
src/tuimble/audio.py
Normal file
@@ -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
|
||||
156
src/tuimble/client.py
Normal file
156
src/tuimble/client.py
Normal file
@@ -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
|
||||
)
|
||||
64
src/tuimble/config.py
Normal file
64
src/tuimble/config.py
Normal file
@@ -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
|
||||
144
src/tuimble/ptt.py
Normal file
144
src/tuimble/ptt.py
Normal file
@@ -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)
|
||||
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
23
tests/test_config.py
Normal file
23
tests/test_config.py
Normal file
@@ -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"
|
||||
37
tests/test_ptt.py
Normal file
37
tests/test_ptt.py
Normal file
@@ -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]
|
||||
Reference in New Issue
Block a user