Compare commits
10 Commits
227d415c24
...
931c6d4776
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
931c6d4776 | ||
|
|
6c16883135 | ||
|
|
f77375422a | ||
|
|
ee145b4071 | ||
|
|
89ebad6c15 | ||
|
|
a6bdc02484 | ||
|
|
3bbe8a96f3 | ||
|
|
31ac90d2c9 | ||
|
|
bc0da57625 | ||
|
|
623c51a19d |
@@ -11,15 +11,38 @@ tuimble --host mumble.example.com --user myname
|
||||
|
||||
| Key | Action |
|
||||
|-----|--------|
|
||||
| `q` | Quit |
|
||||
| `space` | Push-to-talk (configurable) |
|
||||
| `F1` | Toggle self-deafen |
|
||||
| `F4` | Push-to-talk (configurable) |
|
||||
| `Enter` | Send message |
|
||||
| `Up` | Previous sent message (in chat input) |
|
||||
| `Down` | Next sent message (in chat input) |
|
||||
| `q` | Quit |
|
||||
| `Ctrl+C` | Quit |
|
||||
|
||||
## Push-to-Talk Modes
|
||||
|
||||
- **hold** — hold key to transmit, release to stop (default)
|
||||
- **toggle** — press to start, press again to stop
|
||||
- **toggle** — press to start, press again to stop (default)
|
||||
- **hold** — hold key to transmit, release to stop (requires evdev)
|
||||
|
||||
## Profiling
|
||||
|
||||
```sh
|
||||
tuimble --cprofile # saves to ~/.config/tuimble/profile.prof
|
||||
tuimble --cprofile /tmp/tuimble.prof # saves to custom path
|
||||
```
|
||||
|
||||
Profile data is dumped every 30 seconds and on exit, so snapshots
|
||||
are available even after a crash or `kill`. Output is standard `.prof`
|
||||
format:
|
||||
|
||||
```sh
|
||||
python3 -m pstats /tmp/tuimble.prof # interactive explorer
|
||||
# or install snakeviz for a browser-based flamegraph:
|
||||
# pip install snakeviz && snakeviz /tmp/tuimble.prof
|
||||
```
|
||||
|
||||
Note: cProfile captures the main thread only. Background workers
|
||||
started with `@work(thread=True)` are not included.
|
||||
|
||||
## Configuration
|
||||
|
||||
|
||||
@@ -1,13 +1,67 @@
|
||||
"""Entry point for tuimble."""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="tuimble",
|
||||
description="TUI client for Mumble",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--cprofile",
|
||||
nargs="?",
|
||||
const=None,
|
||||
default=False,
|
||||
metavar="FILE",
|
||||
help="run under cProfile, saving to FILE "
|
||||
"(default: ~/.config/tuimble/profile.prof)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
from tuimble.app import TuimbleApp
|
||||
|
||||
app = TuimbleApp()
|
||||
app.run()
|
||||
|
||||
if args.cprofile is not False:
|
||||
_run_profiled(app, args.cprofile)
|
||||
else:
|
||||
app.run()
|
||||
|
||||
|
||||
def _run_profiled(app, dest):
|
||||
"""Run the app under cProfile with periodic 30s dumps."""
|
||||
import cProfile
|
||||
from pathlib import Path
|
||||
from threading import Event, Thread
|
||||
|
||||
if dest is None:
|
||||
from tuimble.config import CONFIG_DIR
|
||||
|
||||
dest = CONFIG_DIR / "profile.prof"
|
||||
else:
|
||||
dest = Path(dest)
|
||||
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
prof = cProfile.Profile()
|
||||
stop = Event()
|
||||
|
||||
def _periodic_dump():
|
||||
while not stop.wait(30):
|
||||
prof.dump_stats(str(dest))
|
||||
|
||||
dumper = Thread(target=_periodic_dump, daemon=True)
|
||||
dumper.start()
|
||||
|
||||
try:
|
||||
prof.enable()
|
||||
app.run()
|
||||
finally:
|
||||
prof.disable()
|
||||
stop.set()
|
||||
prof.dump_stats(str(dest))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -53,6 +53,7 @@ class StatusBar(Static):
|
||||
|
||||
ptt_active = reactive(False)
|
||||
connected = reactive(False)
|
||||
self_deaf = reactive(False)
|
||||
server_info = reactive("")
|
||||
|
||||
def render(self) -> str:
|
||||
@@ -67,17 +68,20 @@ class StatusBar(Static):
|
||||
|
||||
if self.ptt_active:
|
||||
ptt_sym = "[#e0af68]\u25cf[/]"
|
||||
ptt_full = f"{ptt_sym} TX"
|
||||
ptt_full = f"{ptt_sym} [bold]TX[/bold]"
|
||||
else:
|
||||
ptt_sym = "[#565f89]\u25cb[/]"
|
||||
ptt_full = f"{ptt_sym} idle"
|
||||
|
||||
deaf_sym = "[#f7768e]\u2298[/]" if self.self_deaf else ""
|
||||
deaf_full = "[#f7768e]\u2298[/] deaf" if self.self_deaf else ""
|
||||
|
||||
if w < 40:
|
||||
return f" {conn_sym} {ptt_sym}"
|
||||
return f" {conn_sym} {deaf_sym}{ptt_sym}"
|
||||
if w < 60:
|
||||
return f" {conn_full} {ptt_full}"
|
||||
return f" {conn_full} {deaf_full}{' ' if deaf_full else ''}{ptt_full}"
|
||||
info = f" [dim]{self.server_info}[/]" if self.server_info else ""
|
||||
return f" {conn_full} {ptt_full}{info}"
|
||||
return f" {conn_full} {deaf_full}{' ' if deaf_full else ''}{ptt_full}{info}"
|
||||
|
||||
|
||||
class ChannelTree(Static):
|
||||
@@ -187,7 +191,7 @@ class ChatLog(RichLog):
|
||||
"""Message log."""
|
||||
|
||||
def __init__(self, **kwargs) -> None:
|
||||
super().__init__(markup=True, **kwargs)
|
||||
super().__init__(markup=True, wrap=True, **kwargs)
|
||||
|
||||
|
||||
# -- main app ----------------------------------------------------------------
|
||||
@@ -238,6 +242,7 @@ class TuimbleApp(App):
|
||||
"""
|
||||
|
||||
BINDINGS = [
|
||||
("f1", "toggle_deaf", "Deafen"),
|
||||
("q", "quit", "Quit"),
|
||||
("ctrl+c", "quit", "Quit"),
|
||||
]
|
||||
@@ -255,6 +260,9 @@ class TuimbleApp(App):
|
||||
username=srv.username,
|
||||
password=srv.password,
|
||||
)
|
||||
self._history: list[str] = []
|
||||
self._history_idx: int = -1
|
||||
self._history_draft: str = ""
|
||||
acfg = self._config.audio
|
||||
self._audio = AudioPipeline(
|
||||
sample_rate=acfg.sample_rate,
|
||||
@@ -357,6 +365,8 @@ class TuimbleApp(App):
|
||||
if not text:
|
||||
return
|
||||
event.input.clear()
|
||||
self._history.append(text)
|
||||
self._history_idx = -1
|
||||
|
||||
if not self._client.connected:
|
||||
self._show_error("not connected")
|
||||
@@ -406,27 +416,72 @@ class TuimbleApp(App):
|
||||
tree = self.query_one("#sidebar", ChannelTree)
|
||||
tree.set_state(channels, users_by_channel)
|
||||
|
||||
# -- deafen --------------------------------------------------------------
|
||||
|
||||
def action_toggle_deaf(self) -> None:
|
||||
"""Toggle self-deafen."""
|
||||
deafened = not self._audio.deafened
|
||||
self._audio.deafened = deafened
|
||||
self._client.set_self_deaf(deafened)
|
||||
status = self.query_one("#status", StatusBar)
|
||||
status.self_deaf = deafened
|
||||
chatlog = self.query_one("#chatlog", ChatLog)
|
||||
if deafened:
|
||||
chatlog.write("[#f7768e]\u2298 deafened[/]")
|
||||
else:
|
||||
chatlog.write("[#9ece6a]\u2713 undeafened[/]")
|
||||
|
||||
# -- PTT -----------------------------------------------------------------
|
||||
|
||||
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()
|
||||
"""Handle input history navigation and PTT key events."""
|
||||
focused = self.focused
|
||||
if isinstance(focused, Input) and event.key in ("up", "down"):
|
||||
inp = focused
|
||||
if event.key == "up":
|
||||
if not self._history:
|
||||
event.prevent_default()
|
||||
return
|
||||
if self._history_idx == -1:
|
||||
self._history_draft = inp.value
|
||||
self._history_idx = len(self._history) - 1
|
||||
elif self._history_idx > 0:
|
||||
self._history_idx -= 1
|
||||
inp.value = self._history[self._history_idx]
|
||||
inp.cursor_position = len(inp.value)
|
||||
else: # down
|
||||
if self._history_idx == -1:
|
||||
event.prevent_default()
|
||||
return
|
||||
if self._history_idx < len(self._history) - 1:
|
||||
self._history_idx += 1
|
||||
inp.value = self._history[self._history_idx]
|
||||
else:
|
||||
self._ptt.key_down()
|
||||
elif isinstance(self._ptt, TogglePtt):
|
||||
if event.key == ptt_key:
|
||||
self._ptt.toggle()
|
||||
self._history_idx = -1
|
||||
inp.value = self._history_draft
|
||||
inp.cursor_position = len(inp.value)
|
||||
event.prevent_default()
|
||||
return
|
||||
|
||||
if event.key != self._config.ptt.key:
|
||||
return
|
||||
|
||||
if isinstance(self._ptt, TogglePtt):
|
||||
self._ptt.toggle()
|
||||
elif isinstance(self._ptt, KittyPtt):
|
||||
# Kitty hold-mode requires key-release events that Textual
|
||||
# does not expose; kept for explicit backend="kitty" only.
|
||||
self._ptt.key_down()
|
||||
|
||||
def _on_ptt_change(self, transmitting: bool) -> None:
|
||||
self._audio.capturing = transmitting
|
||||
status = self.query_one("#status", StatusBar)
|
||||
status.ptt_active = transmitting
|
||||
chatlog = self.query_one("#chatlog", ChatLog)
|
||||
if transmitting:
|
||||
chatlog.write("[#e0af68]● transmitting[/]")
|
||||
else:
|
||||
chatlog.write("[dim]○ stopped transmitting[/dim]")
|
||||
|
||||
# -- resize --------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -39,6 +39,7 @@ class AudioPipeline:
|
||||
self._input_stream = None
|
||||
self._output_stream = None
|
||||
self._capturing = False
|
||||
self._deafened = False
|
||||
|
||||
def start(self):
|
||||
"""Open audio streams."""
|
||||
@@ -84,6 +85,14 @@ class AudioPipeline:
|
||||
def capturing(self, value: bool):
|
||||
self._capturing = value
|
||||
|
||||
@property
|
||||
def deafened(self) -> bool:
|
||||
return self._deafened
|
||||
|
||||
@deafened.setter
|
||||
def deafened(self, value: bool):
|
||||
self._deafened = value
|
||||
|
||||
def _capture_callback(self, indata, frames, time_info, status):
|
||||
"""Called by sounddevice when input data is available."""
|
||||
if status:
|
||||
@@ -98,6 +107,9 @@ class AudioPipeline:
|
||||
"""Called by sounddevice when output buffer needs data."""
|
||||
if status:
|
||||
log.warning("playback status: %s", status)
|
||||
if self._deafened:
|
||||
outdata[:] = b"\x00" * len(outdata)
|
||||
return
|
||||
try:
|
||||
pcm = self._playback_queue.get_nowait()
|
||||
n = min(len(pcm), len(outdata))
|
||||
@@ -116,6 +128,8 @@ class AudioPipeline:
|
||||
|
||||
def queue_playback(self, pcm_data: bytes):
|
||||
"""Queue raw PCM data for playback (16-bit, mono, 48kHz)."""
|
||||
if self._deafened:
|
||||
return
|
||||
try:
|
||||
self._playback_queue.put_nowait(pcm_data)
|
||||
except queue.Full:
|
||||
|
||||
@@ -181,6 +181,14 @@ class MumbleClient:
|
||||
if self._mumble and self._connected:
|
||||
self._mumble.channels[channel_id].move_in()
|
||||
|
||||
def set_self_deaf(self, deaf: bool):
|
||||
"""Toggle self-deafen on the server."""
|
||||
if self._mumble and self._connected:
|
||||
if deaf:
|
||||
self._mumble.users.myself.deafen()
|
||||
else:
|
||||
self._mumble.users.myself.undeafen()
|
||||
|
||||
# -- pymumble callbacks (run on pymumble thread) -------------------------
|
||||
|
||||
def _register_callbacks(self):
|
||||
|
||||
@@ -28,8 +28,8 @@ class AudioConfig:
|
||||
|
||||
@dataclass
|
||||
class PttConfig:
|
||||
key: str = "space"
|
||||
mode: str = "hold" # hold | toggle
|
||||
key: str = "f4"
|
||||
mode: str = "toggle" # toggle | hold (hold requires evdev)
|
||||
backend: str = "auto" # auto | kitty | evdev | toggle
|
||||
|
||||
|
||||
|
||||
@@ -139,6 +139,12 @@ def detect_backend(callback: Callback, preference: str = "auto") -> PttBackend:
|
||||
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)
|
||||
# auto: Textual does not expose key-release events, so kitty
|
||||
# hold-mode is not viable. Try evdev, fall back to toggle.
|
||||
try:
|
||||
import evdev as _evdev # noqa: F811
|
||||
|
||||
_ = _evdev.list_devices()
|
||||
return EvdevPtt(callback)
|
||||
except Exception:
|
||||
return TogglePtt(callback)
|
||||
|
||||
@@ -10,6 +10,7 @@ def test_default_construction():
|
||||
assert ap._input_device is None
|
||||
assert ap._output_device is None
|
||||
assert ap.capturing is False
|
||||
assert ap.deafened is False
|
||||
|
||||
|
||||
def test_custom_construction():
|
||||
@@ -89,6 +90,37 @@ def test_queue_playback_overflow_drops():
|
||||
assert ap._playback_queue.qsize() == ap._playback_queue.maxsize
|
||||
|
||||
|
||||
def test_deafened_toggle():
|
||||
ap = AudioPipeline()
|
||||
assert ap.deafened is False
|
||||
ap.deafened = True
|
||||
assert ap.deafened is True
|
||||
ap.deafened = False
|
||||
assert ap.deafened is False
|
||||
|
||||
|
||||
def test_queue_playback_discards_when_deafened():
|
||||
"""Incoming PCM is dropped when deafened."""
|
||||
ap = AudioPipeline()
|
||||
ap.deafened = True
|
||||
ap.queue_playback(b"\x42" * 100)
|
||||
assert ap._playback_queue.qsize() == 0
|
||||
|
||||
|
||||
def test_playback_callback_silence_when_deafened():
|
||||
"""Playback callback writes silence when deafened, even with queued data."""
|
||||
ap = AudioPipeline()
|
||||
frame_bytes = FRAME_SIZE * 2
|
||||
# Queue data before deafening
|
||||
pcm = b"\x42" * frame_bytes
|
||||
ap.queue_playback(pcm)
|
||||
ap.deafened = True
|
||||
|
||||
outdata = bytearray(b"\xff" * frame_bytes)
|
||||
ap._playback_callback(outdata, FRAME_SIZE, None, None)
|
||||
assert outdata == bytearray(frame_bytes) # all zeros
|
||||
|
||||
|
||||
def test_stop_without_start():
|
||||
"""Stop on unstarted pipeline should not raise."""
|
||||
ap = AudioPipeline()
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
"""Tests for MumbleClient dispatcher and callback wiring."""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from tuimble.client import MumbleClient
|
||||
|
||||
|
||||
@@ -39,3 +41,29 @@ def test_dispatch_skips_none_callback():
|
||||
client = MumbleClient(host="localhost")
|
||||
# Should not raise
|
||||
client._dispatch(None)
|
||||
|
||||
|
||||
def test_set_self_deaf_calls_pymumble():
|
||||
client = MumbleClient(host="localhost")
|
||||
client._connected = True
|
||||
myself = MagicMock()
|
||||
users = MagicMock()
|
||||
users.myself = myself
|
||||
mumble = MagicMock()
|
||||
mumble.users = users
|
||||
client._mumble = mumble
|
||||
|
||||
client.set_self_deaf(True)
|
||||
myself.deafen.assert_called_once()
|
||||
myself.undeafen.assert_not_called()
|
||||
|
||||
myself.reset_mock()
|
||||
client.set_self_deaf(False)
|
||||
myself.undeafen.assert_called_once()
|
||||
myself.deafen.assert_not_called()
|
||||
|
||||
|
||||
def test_set_self_deaf_noop_when_disconnected():
|
||||
client = MumbleClient(host="localhost")
|
||||
# Should not raise when not connected
|
||||
client.set_self_deaf(True)
|
||||
|
||||
@@ -8,7 +8,7 @@ def test_default_config():
|
||||
assert cfg.server.host == "localhost"
|
||||
assert cfg.server.port == 64738
|
||||
assert cfg.audio.sample_rate == 48000
|
||||
assert cfg.ptt.mode == "hold"
|
||||
assert cfg.ptt.mode == "toggle"
|
||||
|
||||
|
||||
def test_server_config():
|
||||
@@ -19,5 +19,5 @@ def test_server_config():
|
||||
|
||||
def test_ptt_config_defaults():
|
||||
ptt = PttConfig()
|
||||
assert ptt.key == "space"
|
||||
assert ptt.key == "f4"
|
||||
assert ptt.backend == "auto"
|
||||
|
||||
Reference in New Issue
Block a user