feat: add voice pitch shifting
F6/F7 adjust outgoing voice pitch in 1-semitone steps (-12 to +12). PitchShifter integrates into the capture path at dequeue time so the PortAudio callback is never blocked. Status bar shows current pitch when non-zero. Config reload treats pitch as a safe change.
This commit is contained in:
@@ -11,6 +11,7 @@ dependencies = [
|
||||
"textual>=1.0.0",
|
||||
"pymumble>=1.6",
|
||||
"sounddevice>=0.5.0",
|
||||
"numpy>=1.24.0",
|
||||
"tomli>=2.0.0;python_version<'3.11'",
|
||||
]
|
||||
|
||||
|
||||
@@ -118,6 +118,7 @@ class StatusBar(Static):
|
||||
server_info = reactive("")
|
||||
output_vol = reactive(100)
|
||||
input_vol = reactive(100)
|
||||
pitch = reactive(0)
|
||||
|
||||
@staticmethod
|
||||
def _vol_bar(pct: int) -> str:
|
||||
@@ -157,9 +158,14 @@ class StatusBar(Static):
|
||||
f" [dim]out[/]{self._vol_bar(self.output_vol)}"
|
||||
f" [dim]in[/]{self._vol_bar(self.input_vol)}"
|
||||
)
|
||||
if self.pitch != 0:
|
||||
sign = "+" if self.pitch > 0 else ""
|
||||
pitch_str = f" [dim]pitch[/]{sign}{self.pitch}"
|
||||
else:
|
||||
pitch_str = ""
|
||||
info = f" [dim]{self.server_info}[/]" if self.server_info else ""
|
||||
deaf = f"{deaf_full} " if deaf_full else ""
|
||||
return f" {conn_full} {deaf}{ptt_full}{vol}{info}"
|
||||
return f" {conn_full} {deaf}{ptt_full}{vol}{pitch_str}{info}"
|
||||
|
||||
|
||||
class ChannelTree(Static):
|
||||
@@ -412,6 +418,8 @@ class TuimbleApp(App):
|
||||
("f2", "cycle_output_volume", "Vol Out"),
|
||||
("f3", "cycle_input_volume", "Vol In"),
|
||||
("f5", "reload_config", "Reload"),
|
||||
("f6", "pitch_down", "Pitch-"),
|
||||
("f7", "pitch_up", "Pitch+"),
|
||||
("q", "quit", "Quit"),
|
||||
("ctrl+c", "quit", "Quit"),
|
||||
]
|
||||
@@ -431,6 +439,7 @@ class TuimbleApp(App):
|
||||
)
|
||||
self._audio.input_gain = acfg.input_gain
|
||||
self._audio.output_gain = acfg.output_gain
|
||||
self._audio.pitch = acfg.pitch
|
||||
self._device_monitor = DeviceMonitor(self._on_device_change)
|
||||
self._pending_reload: Config | None = None
|
||||
self._tree_refresh_timer = None
|
||||
@@ -469,6 +478,7 @@ class TuimbleApp(App):
|
||||
status = self.query_one("#status", StatusBar)
|
||||
status.output_vol = int(self._audio.output_gain * 100)
|
||||
status.input_vol = int(self._audio.input_gain * 100)
|
||||
status.pitch = int(self._audio.pitch)
|
||||
|
||||
chatlog = self.query_one("#chatlog", ChatLog)
|
||||
chatlog.write("[dim]tuimble v0.1.0[/dim]")
|
||||
@@ -687,6 +697,7 @@ class TuimbleApp(App):
|
||||
)
|
||||
self._audio.input_gain = acfg.input_gain
|
||||
self._audio.output_gain = acfg.output_gain
|
||||
self._audio.pitch = acfg.pitch
|
||||
|
||||
if self._client.connected:
|
||||
self._start_audio()
|
||||
@@ -763,6 +774,26 @@ class TuimbleApp(App):
|
||||
chatlog = self.query_one("#chatlog", ChatLog)
|
||||
chatlog.write(f"[dim]input volume {pct}%[/dim]")
|
||||
|
||||
# -- pitch ----------------------------------------------------------------
|
||||
|
||||
def action_pitch_down(self) -> None:
|
||||
"""Decrease voice pitch by 1 semitone."""
|
||||
self._audio.pitch = max(-12.0, self._audio.pitch - 1.0)
|
||||
st = int(self._audio.pitch)
|
||||
status = self.query_one("#status", StatusBar)
|
||||
status.pitch = st
|
||||
chatlog = self.query_one("#chatlog", ChatLog)
|
||||
chatlog.write(f"[dim]pitch {st:+d} semitones[/dim]")
|
||||
|
||||
def action_pitch_up(self) -> None:
|
||||
"""Increase voice pitch by 1 semitone."""
|
||||
self._audio.pitch = min(12.0, self._audio.pitch + 1.0)
|
||||
st = int(self._audio.pitch)
|
||||
status = self.query_one("#status", StatusBar)
|
||||
status.pitch = st
|
||||
chatlog = self.query_one("#chatlog", ChatLog)
|
||||
chatlog.write(f"[dim]pitch {st:+d} semitones[/dim]")
|
||||
|
||||
# -- config reload --------------------------------------------------------
|
||||
|
||||
def _detect_config_changes(
|
||||
@@ -782,7 +813,7 @@ class TuimbleApp(App):
|
||||
safe.append(f"ptt {key}: {old_ptt[key]} -> {new_ptt[key]}")
|
||||
|
||||
# Audio: gains are safe; hardware settings require restart
|
||||
safe_audio = {"input_gain", "output_gain"}
|
||||
safe_audio = {"input_gain", "output_gain", "pitch"}
|
||||
old_aud = dataclasses.asdict(old.audio)
|
||||
new_aud = dataclasses.asdict(new.audio)
|
||||
for key in old_aud:
|
||||
@@ -809,9 +840,11 @@ class TuimbleApp(App):
|
||||
|
||||
self._audio.input_gain = new.audio.input_gain
|
||||
self._audio.output_gain = new.audio.output_gain
|
||||
self._audio.pitch = new.audio.pitch
|
||||
status = self.query_one("#status", StatusBar)
|
||||
status.input_vol = int(new.audio.input_gain * 100)
|
||||
status.output_vol = int(new.audio.output_gain * 100)
|
||||
status.pitch = int(new.audio.pitch)
|
||||
|
||||
def _apply_restart_changes(self, new: Config) -> None:
|
||||
"""Apply changes that require reconnect/audio restart."""
|
||||
@@ -852,6 +885,7 @@ class TuimbleApp(App):
|
||||
)
|
||||
self._audio.input_gain = acfg.input_gain
|
||||
self._audio.output_gain = acfg.output_gain
|
||||
self._audio.pitch = acfg.pitch
|
||||
self._config = new
|
||||
if self._client.connected:
|
||||
self._start_audio()
|
||||
@@ -897,6 +931,7 @@ class TuimbleApp(App):
|
||||
self._apply_safe_changes(new)
|
||||
self._config.audio.input_gain = new.audio.input_gain
|
||||
self._config.audio.output_gain = new.audio.output_gain
|
||||
self._config.audio.pitch = new.audio.pitch
|
||||
|
||||
if safe:
|
||||
for change in safe:
|
||||
|
||||
@@ -13,6 +13,8 @@ import queue
|
||||
import threading
|
||||
from typing import Callable
|
||||
|
||||
from tuimble.modulator import PitchShifter
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
SAMPLE_RATE = 48000
|
||||
@@ -110,6 +112,7 @@ class AudioPipeline:
|
||||
self._deafened = False
|
||||
self._input_gain = 1.0
|
||||
self._output_gain = 1.0
|
||||
self._pitch_shifter = PitchShifter(sample_rate)
|
||||
|
||||
def start(self) -> None:
|
||||
"""Open audio streams."""
|
||||
@@ -185,6 +188,14 @@ class AudioPipeline:
|
||||
def output_gain(self, value: float):
|
||||
self._output_gain = max(0.0, min(2.0, value))
|
||||
|
||||
@property
|
||||
def pitch(self) -> float:
|
||||
return self._pitch_shifter.semitones
|
||||
|
||||
@pitch.setter
|
||||
def pitch(self, value: float):
|
||||
self._pitch_shifter.semitones = value
|
||||
|
||||
def _capture_callback(self, indata, frames, time_info, status) -> None:
|
||||
"""Called by sounddevice when input data is available."""
|
||||
if status:
|
||||
@@ -219,15 +230,20 @@ class AudioPipeline:
|
||||
def get_capture_frame(self, timeout: float = 0.0) -> bytes | None:
|
||||
"""Retrieve next captured PCM frame for transmission.
|
||||
|
||||
Pitch shifting runs here (worker thread) rather than in the
|
||||
PortAudio callback, which must return within the frame period.
|
||||
|
||||
Args:
|
||||
timeout: Seconds to wait for a frame. 0 returns immediately.
|
||||
"""
|
||||
try:
|
||||
if timeout > 0:
|
||||
return self._capture_queue.get(timeout=timeout)
|
||||
return self._capture_queue.get_nowait()
|
||||
pcm = self._capture_queue.get(timeout=timeout)
|
||||
else:
|
||||
pcm = self._capture_queue.get_nowait()
|
||||
except queue.Empty:
|
||||
return None
|
||||
return self._pitch_shifter.process(pcm)
|
||||
|
||||
def queue_playback(self, pcm_data: bytes) -> None:
|
||||
"""Queue raw PCM data for playback (16-bit, mono, 48kHz)."""
|
||||
|
||||
@@ -32,6 +32,7 @@ class AudioConfig:
|
||||
frame_size: int = 960 # 20ms at 48kHz
|
||||
input_gain: float = 1.0
|
||||
output_gain: float = 1.0
|
||||
pitch: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
Reference in New Issue
Block a user