Compare commits

...

10 Commits

Author SHA1 Message Date
Username
931c6d4776 docs: add up/down key bindings to usage 2026-02-24 13:28:08 +01:00
Username
6c16883135 feat: add chat input history navigation (up/down) 2026-02-24 13:28:03 +01:00
Username
f77375422a add --cprofile flag with periodic dump 2026-02-24 13:18:10 +01:00
Username
ee145b4071 app: log transmit state changes to chatlog 2026-02-24 13:09:03 +01:00
Username
89ebad6c15 app: add F1 self-deafen toggle and bold TX indicator 2026-02-24 13:02:38 +01:00
Username
a6bdc02484 client: add set_self_deaf server notification 2026-02-24 13:02:09 +01:00
Username
3bbe8a96f3 audio: add deafened property to suppress playback
When deafened, playback callback writes silence and
queue_playback discards incoming PCM.
2026-02-24 12:58:39 +01:00
Username
31ac90d2c9 fix: ptt auto-detect falls back to toggle mode
Textual does not expose key-release events, so KittyPtt
hold-mode never received key_up and stayed transmitting.
Auto-detect now tries evdev first, then falls back to
toggle (press-on/press-off). Default mode changed to toggle.
2026-02-24 12:50:40 +01:00
Username
bc0da57625 ptt: change default key from space to f4
Space conflicts with chat input. F4 is unused by default
in most terminal emulators and won't interfere with typing.
2026-02-24 12:46:45 +01:00
Username
623c51a19d fix: enable word wrap in chat log
RichLog defaults wrap=False, causing long messages to extend
beyond the visible area instead of wrapping at the widget edge.
2026-02-24 12:42:39 +01:00
10 changed files with 249 additions and 29 deletions

View File

@@ -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

View File

@@ -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__":

View File

@@ -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 --------------------------------------------------------------

View File

@@ -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:

View File

@@ -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):

View File

@@ -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

View File

@@ -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)

View File

@@ -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()

View File

@@ -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)

View File

@@ -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"