kodjo/MyTTSClient
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working on testing the server

Browse Source
This commit is contained in:
Kodjo Sossouvi
2025-09-08 22:59:12 +02:00
parent 2ce6d22f89
commit ddcc5f5e04
6 changed files with 190 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
requirements.txt Normal file
View File

@@ -0,0 +1,23 @@
cffi==1.17.1
click==8.2.1
comtypes==1.4.12
iniconfig==2.1.0
markdown-it-py==4.0.0
mdurl==0.1.2
numpy==2.3.2
packaging==25.0
pluggy==1.6.0
psutil==7.0.0
PyAudio==0.2.14
pycaw==20240210
pycparser==2.22
Pygments==2.19.2
pytest==8.4.1
PyYAML==6.0.2
rich==14.1.0
scipy==1.16.1
shellingham==1.5.4
sounddevice==0.5.2
typer==0.17.3
typing_extensions==4.15.0
wyoming==1.7.2

5
src/cli.py
View File

@@ -9,15 +9,14 @@ audio_app = typer.Typer(help="Audio device operations")
# Add commands to subcommands
server_app.command("check")(server.check)
server_app.command("test")(server.test)
audio_app.command("list")(audio.list_devices)
audio_app.command("test")(audio.test_device)
audio_app.command("config")(audio.config_info)
audio_app.command("install")(audio.install)
# Register subcommands
app.add_typer(server_app, name="server")
app.add_typer(audio_app, name="audio")
if __name__ == "__main__":
app()
app()

127
src/commands/audio.py
View File

@@ -3,7 +3,8 @@ import numpy as np
import os
from typing import Optional, List
from ..devices import get_audio_devices_windows, install_audio_cmdlets, get_audio_devices_windows_from_pnp_devices
from ..devices import get_audio_devices_windows, install_audio_cmdlets, get_audio_devices_windows_from_pnp_devices, \
get_audio_devices_linux
from ..core.utils import *
from ..config import AppConfig
@@ -48,8 +49,6 @@ def _get_device_type(instance_id: str) -> str:
def _get_short_device_id(instance_id: str) -> str:
"""Get device name based on InstanceId."""
"SWD\MMDEVAPI\{0.0.0.00000000}.{20434736-BDB3-4CE2-A56B-5DF61D4DD593}"
if len(instance_id) == 68:
return instance_id[31:-1]
elif len(instance_id) == 55:
@@ -124,8 +123,23 @@ def list_devices(
else:
linux_devices = get_linux_audio_devices()
display_linux_devices(linux_devices, config, pulse, sort_config)
if native and not is_wsl:
typer.echo(typer.style("Native is applicable only when WSL is detected.", fg=typer.colors.RED))
return
if all_devices and native:
typer.echo(typer.style("All devices is not applicable with native mode.", fg=typer.colors.RED))
return
result = get_audio_devices_linux()
if not check_result(result):
return
linux_devices = result.result
# apply sorting and filtering
linux_devices = filter_and_sort(linux_devices, filter_info, sort_info, desc_info)
display_as_table(linux_devices)
# Show legend
typer.echo(f"\nLegend:")
@@ -141,30 +155,6 @@ def install():
typer.echo(result.stdout)
def get_linux_audio_devices() -> list:
"""Get Linux audio devices using sounddevice."""
try:
devices = sd.query_devices()
linux_devices = []
for i, device in enumerate(devices):
hostapi_name = sd.query_hostapis(device['hostapi'])['name']
linux_devices.append(LinuxAudioDevice(
device_id=i,
name=device['name'],
max_input_channels=device['max_input_channels'],
max_output_channels=device['max_output_channels'],
default_samplerate=device['default_samplerate'],
hostapi_name=hostapi_name
))
return linux_devices
except Exception as e:
typer.echo(typer.style(f"Error querying Linux audio devices: {e}", fg=typer.colors.RED, bold=True))
raise typer.Exit(1)
def display_linux_devices(devices: list,
config: AppConfig,
show_pulse: bool = False,
@@ -251,6 +241,7 @@ def test_device(
device: Optional[int] = typer.Option(None, "--device", "-d", help="Device ID to test (default: configured device)"),
duration: float = typer.Option(3.0, "--duration", help="Recording duration in seconds"),
save: bool = typer.Option(False, "--save", help="Save recording to WAV file"),
play: bool = typer.Option(False, "--play", help="Play recorded audio through default speakers"),
config_file: Optional[str] = typer.Option(None, "--config", "-c", help="Path to config file")
):
"""Test audio recording from a specific device."""
@@ -265,7 +256,11 @@ def test_device(
typer.echo("=" * 20)
# Get device info
devices_list = get_linux_audio_devices()
result = get_audio_devices_linux()
if not check_result(result):
return
devices_list = result.result
if test_device_id is not None:
if test_device_id >= len(devices_list):
@@ -319,72 +314,20 @@ def test_device(
else:
typer.echo(typer.style(" Status: Good signal level", fg=typer.colors.GREEN))
# Play recorded audio if requested
if play:
typer.echo(f"\n{typer.style('Playing recorded audio...', fg=typer.colors.CYAN, bold=True)}")
sd.play(audio_data, samplerate=config.audio.sample_rate)
sd.wait()
typer.echo(typer.style("Playback completed!", fg=typer.colors.CYAN))
# Save if requested
if save:
filename = f"test_device_{test_device_id or 'default'}_{int(duration)}s.wav"
# Note: WyomingAudioRecorder is not defined in the current code
# This would need to be implemented or the save functionality modified
typer.echo(f" Save functionality needs to be implemented")
from scipy.io import wavfile
wavfile.write(filename, config.audio.sample_rate, audio_int16)
typer.echo(typer.style(f"Audio saved to: {filename}", fg=typer.colors.MAGENTA, bold=True))
except Exception as e:
typer.echo(typer.style(f"Recording failed: {e}", fg=typer.colors.RED, bold=True))
raise typer.Exit(1)
def config_info(
config_file: Optional[str] = typer.Option(None, "--config", "-c", help="Path to config file")
):
"""Show current audio configuration."""
# Load configuration
config = AppConfig.load(config_file)
typer.echo(typer.style("Audio Configuration", fg=typer.colors.BLUE, bold=True))
typer.echo("=" * 21)
# Show current settings
typer.echo(f"\nCurrent settings:")
typer.echo(f" Sample rate: {typer.style(f'{config.audio.sample_rate} Hz', fg=typer.colors.CYAN)}")
typer.echo(f" Channels: {typer.style(str(config.audio.channels), fg=typer.colors.CYAN)}")
if config.audio.device is not None:
typer.echo(f" Device: {typer.style(str(config.audio.device), fg=typer.colors.CYAN)}")
# Show device details
try:
devices_list = get_linux_audio_devices()
if config.audio.device < len(devices_list):
device = devices_list[config.audio.device]
typer.echo(f"\nConfigured device details:")
typer.echo(f" Name: {device['name']}")
typer.echo(f" Input channels: {device['max_input_channels']}")
typer.echo(f" Default rate: {int(device['default_samplerate'])} Hz")
if device['max_input_channels'] == 0:
typer.echo(typer.style(" Warning: This device has no input channels!", fg=typer.colors.RED))
else:
typer.echo(typer.style(f" Warning: Configured device {config.audio.device} not found!",
fg=typer.colors.RED, bold=True))
except Exception as e:
typer.echo(typer.style(f" Error getting device info: {e}", fg=typer.colors.RED))
else:
typer.echo(f" Device: {typer.style('default', fg=typer.colors.CYAN)}")
# Show default device info
try:
default_device = sd.default.device
if hasattr(default_device, '__len__') and len(default_device) >= 2:
default_input = int(default_device[0])
else:
default_input = int(default_device)
devices_list = get_linux_audio_devices()
device = devices_list[default_input]
typer.echo(f"\nDefault input device:")
typer.echo(f" ID: {default_input}")
typer.echo(f" Name: {device['name']}")
typer.echo(f" Input channels: {device['max_input_channels']}")
except Exception as e:
typer.echo(typer.style(f" Error getting default device: {e}", fg=typer.colors.RED))
# Show configuration source info

119
src/commands/server.py
View File

@@ -4,7 +4,61 @@ from contextlib import closing
import typer
from typing import Optional
from wyoming.audio import AudioStart, AudioChunk, AudioStop
from ..config import AppConfig
import io
import wave
import numpy as np
import sounddevice as sd
import asyncio
from wyoming.client import AsyncTcpClient
from wyoming.asr import Transcribe, Transcript
async def _async_transcribe(host: str, port: int, timeout: float, pcm_bytes: bytes, lang: str) -> Optional[str]:
"""Stream raw PCM data to Wyoming ASR and return transcript text."""
# Instantiate the async TCP client
client = AsyncTcpClient(host, port)
# Audio parameters
rate = 16000
width = 2 # 16-bit
channels = 1
# The client instance is an async context manager.
async with client:
# 1. Send transcription request
await client.write_event(Transcribe(language=lang).event())
# 2. Start the audio stream
await client.write_event(AudioStart(rate, width, channels).event())
# 3. Send audio chunks
chunk_size = 2048 # A reasonable chunk size
for i in range(0, len(pcm_bytes), chunk_size):
chunk_bytes = pcm_bytes[i:i + chunk_size]
await client.write_event(AudioChunk(audio=chunk_bytes, rate=rate, width=width, channels=channels).event())
# 4. Stop the audio stream
await client.write_event(AudioStop().event())
# 5. Read events until a transcript arrives
transcript_text = None
try:
while True:
event = await asyncio.wait_for(client.read_event(), timeout=timeout)
if event is None:
break
if Transcript.is_type(event):
tr = Transcript.from_event(event)
transcript_text = tr.text
break
except asyncio.TimeoutError:
typer.echo(typer.style("Connection timed out waiting for transcript.", fg=typer.colors.YELLOW))
return transcript_text
def check_wyoming_server(host: str, port: int, timeout: float = 3.0) -> tuple[bool, float | None, str | None]:
@@ -57,4 +111,67 @@ def check(
typer.echo(typer.style("Wyoming server unreachable!", fg=typer.colors.RED, bold=True))
typer.echo(f"Server: {final_host}:{final_port}")
typer.echo(typer.style(f"Error: {error}", fg=typer.colors.RED))
raise typer.Exit(1)
raise typer.Exit(1)
def test(
duration: float = typer.Option(3.0, "--duration", help="Recording duration in seconds"),
lang: str = typer.Option("fr", "--lang", help="Language code: 'fr' or 'en'"),
host: Optional[str] = typer.Option(None, "--host", "-h", help="Wyoming server host"),
port: Optional[int] = typer.Option(None, "--port", "-p", help="Wyoming server port"),
timeout: Optional[float] = typer.Option(None, "--timeout", "-t", help="Connection timeout in seconds"),
config_file: Optional[str] = typer.Option(None, "--config", "-c", help="Path to config file")
):
"""Record from default microphone, send to Wyoming ASR server, and print transcription."""
# Load configuration
config = AppConfig.load(config_file)
final_host = host or config.server.host
final_port = port or config.server.port
final_timeout = timeout or config.server.timeout
# Validate language (two-letter code)
lang = (lang or "fr").strip().lower()
if lang not in ("fr", "en"):
typer.echo(typer.style("Invalid --lang. Use 'fr' or 'en'.", fg=typer.colors.RED))
raise typer.Exit(2)
# Check server reachability first
reachable, latency, err = check_wyoming_server(final_host, final_port, final_timeout)
if not reachable:
typer.echo(typer.style(f"Cannot reach Wyoming server at {final_host}:{final_port}: {err}", fg=typer.colors.RED))
raise typer.Exit(1)
# Record audio (16 kHz mono float32)
sample_rate = 16000
channels = 1
typer.echo(typer.style("Recording...", fg=typer.colors.GREEN, bold=True))
try:
frames = int(duration * sample_rate)
audio = sd.rec(frames, samplerate=sample_rate, channels=channels, dtype="float32")
sd.wait()
except Exception as e:
typer.echo(typer.style(f"Audio recording failed: {e}", fg=typer.colors.RED))
raise typer.Exit(1)
# Convert to PCM16 bytes directly, no need for WAV wrapper
audio_int16 = np.clip(audio.flatten() * 32767.0, -32768, 32767).astype(np.int16)
pcm_bytes = audio_int16.tobytes()
# Send to Wyoming ASR (async)
try:
typer.echo(typer.style(f"Connecting to {final_host}:{final_port} (lang={lang})...", fg=typer.colors.CYAN))
# Run the async helper
transcript_text = asyncio.run(
_async_transcribe(final_host, final_port, final_timeout, pcm_bytes, lang)
)
if transcript_text:
typer.echo(typer.style("\nTranscription:", fg=typer.colors.GREEN, bold=True))
typer.echo(transcript_text)
else:
typer.echo(typer.style("No transcription received.", fg=typer.colors.YELLOW))
except Exception as e:
typer.echo(typer.style(f"ASR request failed: {e}", fg=typer.colors.RED))
raise typer.Exit(1)

6
src/core/Expando.py
View File

@@ -8,8 +8,10 @@ class Expando:
You can then access the property using dot '.' (ex. obj.prop1.prop2)
"""
def __init__(self, props):
self._props = props
def __init__(self, props=None, **kwargs):
self._props = props.copy() if props else {}
if kwargs:
self._props.update(kwargs)
def __getattr__(self, item):
if item not in self._props:

9
src/core/utils.py
View File

@@ -50,7 +50,14 @@ class SelectConf:
class ProcessResult:
result: Any
error: str = None
@property
def stderr(self):
return self.error
@property
def returncode(self):
return 0 if self.result is not None else 1
def sort_by(items: list[Expando], sort_conf: SortConf):
"""Sort a list of items by a given property."""