Compare commits
2 Commits
404c632cfc
...
973a485996
| Author | SHA1 | Date | |
|---|---|---|---|
| 973a485996 | |||
| cc3c0dad7b |
@@ -19,7 +19,6 @@ services:
|
|||||||
ports:
|
ports:
|
||||||
- "8091:8091"
|
- "8091:8091"
|
||||||
- "8092:8092"
|
- "8092:8092"
|
||||||
- "8094:8094"
|
|
||||||
volumes:
|
volumes:
|
||||||
- /home/guru/vllm/data/root:/root
|
- /home/guru/vllm/data/root:/root
|
||||||
- ./deploy/qwen3_tts.yaml:/deploy/qwen3_tts.yaml:ro
|
- ./deploy/qwen3_tts.yaml:/deploy/qwen3_tts.yaml:ro
|
||||||
@@ -31,7 +30,7 @@ services:
|
|||||||
command: ["/bin/bash", "/entrypoint.sh"]
|
command: ["/bin/bash", "/entrypoint.sh"]
|
||||||
restart: unless-stopped
|
restart: unless-stopped
|
||||||
healthcheck:
|
healthcheck:
|
||||||
test: ["CMD", "curl", "-sf", "http://localhost:8091/health"]
|
test: ["CMD", "curl", "-sf", "http://localhost:8095/health"]
|
||||||
interval: 30s
|
interval: 30s
|
||||||
timeout: 10s
|
timeout: 10s
|
||||||
retries: 5
|
retries: 5
|
||||||
@@ -61,10 +60,11 @@ services:
|
|||||||
- ./deploy/qwen3_tts_clone.yaml:/deploy/qwen3_tts_clone.yaml:ro
|
- ./deploy/qwen3_tts_clone.yaml:/deploy/qwen3_tts_clone.yaml:ro
|
||||||
- ./docker/entrypoint_clone.sh:/entrypoint_clone.sh:ro
|
- ./docker/entrypoint_clone.sh:/entrypoint_clone.sh:ro
|
||||||
- ./docker/patch_qwen3_tts_runtime.py:/patch_qwen3_tts_runtime.py:ro
|
- ./docker/patch_qwen3_tts_runtime.py:/patch_qwen3_tts_runtime.py:ro
|
||||||
|
- ./docker/ws_log_proxy.py:/ws_log_proxy.py:ro
|
||||||
command: ["/bin/bash", "/entrypoint_clone.sh"]
|
command: ["/bin/bash", "/entrypoint_clone.sh"]
|
||||||
restart: "no"
|
restart: "no"
|
||||||
healthcheck:
|
healthcheck:
|
||||||
test: ["CMD", "curl", "-sf", "http://localhost:8091/health"]
|
test: ["CMD", "curl", "-sf", "http://localhost:8095/health"]
|
||||||
interval: 30s
|
interval: 30s
|
||||||
timeout: 10s
|
timeout: 10s
|
||||||
retries: 5
|
retries: 5
|
||||||
|
|||||||
@@ -7,9 +7,9 @@
|
|||||||
set -e
|
set -e
|
||||||
|
|
||||||
MODEL="${QWEN3_TTS_MODEL:-Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice}"
|
MODEL="${QWEN3_TTS_MODEL:-Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice}"
|
||||||
PORT=8091
|
PORT=8091 # external-facing proxy port (clients connect here)
|
||||||
|
VLLM_PORT=8095 # internal vllm-omni port
|
||||||
UI_PORT=8092
|
UI_PORT=8092
|
||||||
WS_LOG_PORT=8094
|
|
||||||
|
|
||||||
# --- Runtime-Patches fuer Qwen3-TTS anwenden ---
|
# --- Runtime-Patches fuer Qwen3-TTS anwenden ---
|
||||||
if [ -f /patch_qwen3_tts_runtime.py ]; then
|
if [ -f /patch_qwen3_tts_runtime.py ]; then
|
||||||
@@ -21,28 +21,29 @@ if [ -f /voice_clone_ui.py ]; then
|
|||||||
python3 /voice_clone_ui.py \
|
python3 /voice_clone_ui.py \
|
||||||
--host 0.0.0.0 \
|
--host 0.0.0.0 \
|
||||||
--port "$UI_PORT" \
|
--port "$UI_PORT" \
|
||||||
--api-base "http://localhost:${PORT}" \
|
--api-base "http://localhost:${VLLM_PORT}" \
|
||||||
--clone-api-base "http://qwen3-tts-clone:8093" \
|
--clone-api-base "http://qwen3-tts-clone:8093" \
|
||||||
--model "$MODEL" &
|
--model "$MODEL" &
|
||||||
UI_PID=$!
|
UI_PID=$!
|
||||||
echo "[ui] Voice-Cloning-UI laeuft auf Port ${UI_PORT}"
|
echo "[ui] Voice-Cloning-UI laeuft auf Port ${UI_PORT}"
|
||||||
fi
|
fi
|
||||||
|
|
||||||
# --- WebSocket-Logging-Proxy im Container starten ---
|
# --- WebSocket-Proxy auf dem client-facing PORT starten ---
|
||||||
|
# Proxy sitzt vor vllm-omni und fuegt Stille-Fuellen zwischen Saetzen ein.
|
||||||
if [ -f /ws_log_proxy.py ]; then
|
if [ -f /ws_log_proxy.py ]; then
|
||||||
python3 /ws_log_proxy.py \
|
python3 /ws_log_proxy.py \
|
||||||
--host 0.0.0.0 \
|
--host 0.0.0.0 \
|
||||||
--port "$WS_LOG_PORT" \
|
--port "$PORT" \
|
||||||
--upstream "ws://localhost:${PORT}/v1/audio/speech/stream" &
|
--upstream "ws://localhost:${VLLM_PORT}/v1/audio/speech/stream" &
|
||||||
WS_LOG_PID=$!
|
WS_PROXY_PID=$!
|
||||||
echo "[ws-log] Proxy laeuft auf Port ${WS_LOG_PORT}"
|
echo "[ws-proxy] Proxy laeuft auf Port ${PORT} -> vllm-omni Port ${VLLM_PORT}"
|
||||||
fi
|
fi
|
||||||
|
|
||||||
# --- Server im Hintergrund starten ---
|
# --- vllm-omni Server intern auf VLLM_PORT starten ---
|
||||||
vllm-omni serve "$MODEL" \
|
vllm-omni serve "$MODEL" \
|
||||||
--omni \
|
--omni \
|
||||||
--host 0.0.0.0 \
|
--host 0.0.0.0 \
|
||||||
--port "$PORT" \
|
--port "$VLLM_PORT" \
|
||||||
--deploy-config /deploy/qwen3_tts.yaml \
|
--deploy-config /deploy/qwen3_tts.yaml \
|
||||||
--gpu-memory-utilization 0.10 \
|
--gpu-memory-utilization 0.10 \
|
||||||
--trust-remote-code &
|
--trust-remote-code &
|
||||||
@@ -51,10 +52,10 @@ SERVER_PID=$!
|
|||||||
# --- Warmup im Hintergrund, sobald der Server gesund ist ---
|
# --- Warmup im Hintergrund, sobald der Server gesund ist ---
|
||||||
(
|
(
|
||||||
for _ in $(seq 1 150); do
|
for _ in $(seq 1 150); do
|
||||||
if curl -sf "http://localhost:${PORT}/health" >/dev/null 2>&1; then
|
if curl -sf "http://localhost:${VLLM_PORT}/health" >/dev/null 2>&1; then
|
||||||
echo "[warmup] Server gesund — sende Warmup-Request (kompiliert CUDA-Graphen)..."
|
echo "[warmup] Server gesund — sende Warmup-Request (kompiliert CUDA-Graphen)..."
|
||||||
t0=$SECONDS
|
t0=$SECONDS
|
||||||
curl -sf -X POST "http://localhost:${PORT}/v1/audio/speech" \
|
curl -sf -X POST "http://localhost:${VLLM_PORT}/v1/audio/speech" \
|
||||||
-H 'Content-Type: application/json' \
|
-H 'Content-Type: application/json' \
|
||||||
-d "{\"model\":\"${MODEL}\",\"input\":\"System wird aufgewärmt.\",\"voice\":\"Ryan\",\"language\":\"German\",\"response_format\":\"wav\"}" \
|
-d "{\"model\":\"${MODEL}\",\"input\":\"System wird aufgewärmt.\",\"voice\":\"Ryan\",\"language\":\"German\",\"response_format\":\"wav\"}" \
|
||||||
-o /dev/null \
|
-o /dev/null \
|
||||||
@@ -67,5 +68,5 @@ SERVER_PID=$!
|
|||||||
) &
|
) &
|
||||||
|
|
||||||
# --- Auf den Server-Prozess warten (PID 1 Signalweiterleitung) ---
|
# --- Auf den Server-Prozess warten (PID 1 Signalweiterleitung) ---
|
||||||
trap 'kill "$SERVER_PID" "${UI_PID:-}" "${WS_LOG_PID:-}" 2>/dev/null || true' TERM INT
|
trap 'kill "$SERVER_PID" "${UI_PID:-}" "${WS_PROXY_PID:-}" 2>/dev/null || true' TERM INT
|
||||||
wait "$SERVER_PID"
|
wait "$SERVER_PID"
|
||||||
|
|||||||
@@ -2,33 +2,43 @@
|
|||||||
set -e
|
set -e
|
||||||
|
|
||||||
MODEL="${QWEN3_TTS_CLONE_MODEL:-}"
|
MODEL="${QWEN3_TTS_CLONE_MODEL:-}"
|
||||||
PORT="${QWEN3_TTS_CLONE_PORT:-8091}"
|
PORT="${QWEN3_TTS_CLONE_PORT:-8091}" # external-facing proxy port
|
||||||
|
VLLM_PORT=8095 # internal vllm-omni port
|
||||||
|
|
||||||
if [ -z "$MODEL" ]; then
|
if [ -z "$MODEL" ]; then
|
||||||
echo "[clone] QWEN3_TTS_CLONE_MODEL ist nicht gesetzt." >&2
|
echo "[clone] QWEN3_TTS_CLONE_MODEL ist nicht gesetzt." >&2
|
||||||
exit 2
|
exit 2
|
||||||
fi
|
fi
|
||||||
|
|
||||||
echo "[clone] starte Clone-Modell: ${MODEL} auf Port ${PORT}"
|
echo "[clone] starte Clone-Modell: ${MODEL} auf Port ${PORT} (vllm intern: ${VLLM_PORT})"
|
||||||
|
|
||||||
if [ -f /patch_qwen3_tts_runtime.py ]; then
|
if [ -f /patch_qwen3_tts_runtime.py ]; then
|
||||||
python3 /patch_qwen3_tts_runtime.py
|
python3 /patch_qwen3_tts_runtime.py
|
||||||
fi
|
fi
|
||||||
|
|
||||||
|
# Proxy auf client-facing PORT; leitet an internen VLLM_PORT weiter.
|
||||||
|
if [ -f /ws_log_proxy.py ]; then
|
||||||
|
python3 /ws_log_proxy.py \
|
||||||
|
--host 0.0.0.0 \
|
||||||
|
--port "$PORT" \
|
||||||
|
--upstream "ws://localhost:${VLLM_PORT}/v1/audio/speech/stream" &
|
||||||
|
echo "[ws-proxy] Proxy läuft auf Port ${PORT} -> vllm-omni Port ${VLLM_PORT}"
|
||||||
|
fi
|
||||||
|
|
||||||
vllm-omni serve "$MODEL" \
|
vllm-omni serve "$MODEL" \
|
||||||
--omni \
|
--omni \
|
||||||
--host 0.0.0.0 \
|
--host 0.0.0.0 \
|
||||||
--port "$PORT" \
|
--port "$VLLM_PORT" \
|
||||||
--deploy-config /deploy/qwen3_tts_clone.yaml \
|
--deploy-config /deploy/qwen3_tts_clone.yaml \
|
||||||
--trust-remote-code &
|
--trust-remote-code &
|
||||||
SERVER_PID=$!
|
SERVER_PID=$!
|
||||||
|
|
||||||
(
|
(
|
||||||
for _ in $(seq 1 150); do
|
for _ in $(seq 1 150); do
|
||||||
if curl -sf "http://localhost:${PORT}/health" >/dev/null 2>&1; then
|
if curl -sf "http://localhost:${VLLM_PORT}/health" >/dev/null 2>&1; then
|
||||||
echo "[warmup] Clone gesund — sende Warmup-Request (kompiliert CUDA-Graphen)..."
|
echo "[warmup] Clone gesund — sende Warmup-Request (kompiliert CUDA-Graphen)..."
|
||||||
t0=$SECONDS
|
t0=$SECONDS
|
||||||
curl -sf -X POST "http://localhost:${PORT}/v1/audio/speech" \
|
curl -sf -X POST "http://localhost:${VLLM_PORT}/v1/audio/speech" \
|
||||||
-H 'Content-Type: application/json' \
|
-H 'Content-Type: application/json' \
|
||||||
--max-time 180 \
|
--max-time 180 \
|
||||||
-d "{\"model\":\"${MODEL}\",\"input\":\"System wird aufgewärmt.\",\"language\":\"German\",\"response_format\":\"wav\",\"task_type\":\"VoiceDesign\",\"instructions\":\"Eine ruhige Stimme.\"}" \
|
-d "{\"model\":\"${MODEL}\",\"input\":\"System wird aufgewärmt.\",\"language\":\"German\",\"response_format\":\"wav\",\"task_type\":\"VoiceDesign\",\"instructions\":\"Eine ruhige Stimme.\"}" \
|
||||||
|
|||||||
@@ -1,83 +1,350 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
"""Logging WebSocket proxy for Qwen3-TTS streaming calls."""
|
"""Combined HTTP + WebSocket proxy for Qwen3-TTS on a single port.
|
||||||
|
|
||||||
|
Routes on one port (default 8091):
|
||||||
|
POST /v1/audio/speech → REST proxy to internal vllm-omni
|
||||||
|
GET /v1/audio/speech/stream → WebSocket (PCM-paced, REST-sourced)
|
||||||
|
GET /health → health proxy to internal vllm-omni
|
||||||
|
|
||||||
|
Why REST-sourced WebSocket?
|
||||||
|
The vllm-omni streaming codec drops the onset of short utterances.
|
||||||
|
The batch REST endpoint (/v1/audio/speech) uses full decode and renders
|
||||||
|
every word cleanly. This proxy keeps the client-facing WS protocol intact
|
||||||
|
while sourcing audio sentence-by-sentence from REST, pacing the result
|
||||||
|
into one continuous real-time PCM stream via pcm_pacer.
|
||||||
|
|
||||||
|
Why instructions is NOT forwarded:
|
||||||
|
With a cloned ICL voice (ref_audio) the voice character comes from the
|
||||||
|
reference; an extra style prompt makes the Talker emit EOS far too early
|
||||||
|
and clips short words ("Hallo." → 0.16 s vs 0.40 s without).
|
||||||
|
"""
|
||||||
import argparse
|
import argparse
|
||||||
import asyncio
|
import asyncio
|
||||||
import datetime as dt
|
import datetime as dt
|
||||||
import json
|
import json
|
||||||
|
import math
|
||||||
|
import re
|
||||||
|
|
||||||
|
import aiohttp
|
||||||
import websockets
|
import websockets
|
||||||
|
from aiohttp import web
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
SAMPLE_RATE = 24_000
|
||||||
|
BYTES_PER_SAMPLE = 2
|
||||||
|
BYTES_PER_SEC = SAMPLE_RATE * BYTES_PER_SAMPLE # 48 000
|
||||||
|
|
||||||
|
SEND_CHUNK = 960 # 20 ms at 48 kHz
|
||||||
|
FRAME_SECS = SEND_CHUNK / BYTES_PER_SEC # 0.020 s
|
||||||
|
|
||||||
|
LEADING_MS = 100
|
||||||
|
LEADING_FRAMES = math.ceil(LEADING_MS / 1000 / FRAME_SECS)
|
||||||
|
TRAILING_MS = 200
|
||||||
|
TRAILING_FRAMES = math.ceil(TRAILING_MS / 1000 / FRAME_SECS)
|
||||||
|
|
||||||
|
# Per-sentence synthesis timeout. A healthy sentence renders in <2 s.
|
||||||
|
SYNTH_TIMEOUT_S = 30
|
||||||
|
# Direct REST proxy timeout — longer for potentially bigger inputs.
|
||||||
|
REST_PROXY_TIMEOUT_S = 120
|
||||||
|
|
||||||
|
SILENCE_CHUNK = bytes(SEND_CHUNK)
|
||||||
|
|
||||||
|
SPLIT_RE = re.compile(r"(?<=[.!?])\s+|(?<=[。!?])")
|
||||||
|
MIN_SENTENCE_LEN = 2
|
||||||
|
|
||||||
|
# Fields forwarded to the internal REST synthesis request.
|
||||||
|
# `instructions` deliberately omitted — see module docstring.
|
||||||
|
REST_PASSTHROUGH = ("model", "voice", "task_type", "language",
|
||||||
|
"seed", "speaker_embedding")
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
_internal_base_url: str = "" # e.g. http://localhost:8095
|
||||||
|
_internal_ws_url: str = "" # e.g. ws://localhost:8095/v1/audio/speech/stream
|
||||||
|
|
||||||
|
|
||||||
def now():
|
def now() -> str:
|
||||||
return dt.datetime.now().isoformat(timespec="seconds")
|
return dt.datetime.now().isoformat(timespec="seconds")
|
||||||
|
|
||||||
|
|
||||||
def shorten(value, limit=500):
|
class SentenceSplitter:
|
||||||
text = value if isinstance(value, str) else repr(value)
|
"""Incremental splitter: buffers text, yields complete sentences."""
|
||||||
return text if len(text) <= limit else text[:limit] + "..."
|
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self.buf = ""
|
||||||
|
|
||||||
|
def add(self, text: str) -> list[str]:
|
||||||
|
self.buf += text
|
||||||
|
out: list[str] = []
|
||||||
|
while True:
|
||||||
|
m = SPLIT_RE.search(self.buf)
|
||||||
|
if not m:
|
||||||
|
break
|
||||||
|
idx = m.end()
|
||||||
|
head = self.buf[:idx].strip()
|
||||||
|
if len(head) < MIN_SENTENCE_LEN:
|
||||||
|
break
|
||||||
|
self.buf = self.buf[idx:]
|
||||||
|
if head:
|
||||||
|
out.append(head)
|
||||||
|
return out
|
||||||
|
|
||||||
|
def flush(self) -> list[str]:
|
||||||
|
s = self.buf.strip()
|
||||||
|
self.buf = ""
|
||||||
|
return [s] if s else []
|
||||||
|
|
||||||
|
|
||||||
async def proxy(client_ws, upstream_url):
|
async def synth_rest(http: aiohttp.ClientSession, cfg: dict, text: str) -> bytes:
|
||||||
conn_id = f"ws-{id(client_ws):x}"
|
"""Synthesise one sentence via the internal batch REST endpoint → raw PCM."""
|
||||||
print(f"[{now()}] {conn_id} open -> {upstream_url}", flush=True)
|
payload = {"input": text, "response_format": "pcm"}
|
||||||
async with websockets.connect(upstream_url, max_size=None) as upstream_ws:
|
for k in REST_PASSTHROUGH:
|
||||||
async def client_to_upstream():
|
if cfg.get(k) is not None:
|
||||||
async for msg in client_ws:
|
payload[k] = cfg[k]
|
||||||
if isinstance(msg, str):
|
async with http.post(f"{_internal_base_url}/v1/audio/speech",
|
||||||
try:
|
json=payload) as resp:
|
||||||
data = json.loads(msg)
|
resp.raise_for_status()
|
||||||
typ = data.get("type")
|
return await resp.read()
|
||||||
if typ == "session.config":
|
|
||||||
log_data = dict(data)
|
|
||||||
if "speaker_embedding" in log_data:
|
|
||||||
emb = log_data["speaker_embedding"] or []
|
|
||||||
log_data["speaker_embedding"] = f"<{len(emb)} floats>"
|
|
||||||
print(f"[{now()}] {conn_id} C->S config {json.dumps(log_data, ensure_ascii=False, sort_keys=True)}", flush=True)
|
|
||||||
elif typ == "input.text":
|
|
||||||
print(f"[{now()}] {conn_id} C->S text {data.get('text')!r}", flush=True)
|
|
||||||
else:
|
|
||||||
print(f"[{now()}] {conn_id} C->S {shorten(msg)}", flush=True)
|
|
||||||
except Exception:
|
|
||||||
print(f"[{now()}] {conn_id} C->S {shorten(msg)}", flush=True)
|
|
||||||
else:
|
|
||||||
print(f"[{now()}] {conn_id} C->S binary {len(msg)} bytes", flush=True)
|
|
||||||
await upstream_ws.send(msg)
|
|
||||||
|
|
||||||
async def upstream_to_client():
|
|
||||||
async for msg in upstream_ws:
|
|
||||||
if isinstance(msg, str):
|
|
||||||
try:
|
|
||||||
data = json.loads(msg)
|
|
||||||
print(f"[{now()}] {conn_id} S->C {data}", flush=True)
|
|
||||||
except Exception:
|
|
||||||
print(f"[{now()}] {conn_id} S->C {shorten(msg)}", flush=True)
|
|
||||||
else:
|
|
||||||
print(f"[{now()}] {conn_id} S->C binary {len(msg)} bytes", flush=True)
|
|
||||||
await client_ws.send(msg)
|
|
||||||
|
|
||||||
await asyncio.gather(client_to_upstream(), upstream_to_client())
|
|
||||||
|
|
||||||
|
|
||||||
async def handler(client_ws):
|
# ── REST proxy ───────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
async def handle_rest(request: web.Request) -> web.StreamResponse:
|
||||||
|
"""Forward POST /v1/audio/speech to the internal vllm-omni REST API."""
|
||||||
|
body = await request.read()
|
||||||
|
ct = request.headers.get("Content-Type", "application/json")
|
||||||
|
timeout = aiohttp.ClientTimeout(total=REST_PROXY_TIMEOUT_S)
|
||||||
|
async with aiohttp.ClientSession(timeout=timeout) as http:
|
||||||
|
async with http.post(
|
||||||
|
f"{_internal_base_url}/v1/audio/speech",
|
||||||
|
data=body,
|
||||||
|
headers={"Content-Type": ct},
|
||||||
|
) as resp:
|
||||||
|
out_ct = resp.headers.get("Content-Type", "application/octet-stream")
|
||||||
|
response = web.StreamResponse(
|
||||||
|
status=resp.status,
|
||||||
|
headers={"Content-Type": out_ct},
|
||||||
|
)
|
||||||
|
await response.prepare(request)
|
||||||
|
async for chunk in resp.content.iter_chunked(65536):
|
||||||
|
await response.write(chunk)
|
||||||
|
await response.write_eof()
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
async def handle_health(request: web.Request) -> web.Response:
|
||||||
|
"""Proxy GET /health to internal vllm-omni."""
|
||||||
try:
|
try:
|
||||||
await proxy(client_ws, handler.upstream_url)
|
async with aiohttp.ClientSession() as http:
|
||||||
except Exception as exc:
|
async with http.get(f"{_internal_base_url}/health") as resp:
|
||||||
print(f"[{now()}] proxy error: {exc}", flush=True)
|
return web.Response(status=resp.status, text=await resp.text())
|
||||||
|
except Exception:
|
||||||
|
return web.Response(status=503, text="unavailable")
|
||||||
|
|
||||||
|
|
||||||
|
# ── WebSocket PCM session ────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
async def proxy_session_pcm(client_ws: web.WebSocketResponse,
|
||||||
|
conn_id: str, cfg: dict) -> None:
|
||||||
|
"""REST-sourced, paced, single continuous PCM stream per session."""
|
||||||
|
pcm_queue: asyncio.Queue = asyncio.Queue()
|
||||||
|
sentence_queue: asyncio.Queue = asyncio.Queue()
|
||||||
|
splitter = SentenceSplitter()
|
||||||
|
started = {"audio": False}
|
||||||
|
|
||||||
|
def enqueue_pcm(data: bytes) -> None:
|
||||||
|
for i in range(0, len(data), SEND_CHUNK):
|
||||||
|
piece = data[i:i + SEND_CHUNK]
|
||||||
|
if len(piece) < SEND_CHUNK:
|
||||||
|
piece = piece + bytes(SEND_CHUNK - len(piece))
|
||||||
|
pcm_queue.put_nowait(piece)
|
||||||
|
|
||||||
|
async def pcm_pacer() -> bool:
|
||||||
|
"""Drain pcm_queue at 48 kHz. Returns True = fully drained, False = disconnected."""
|
||||||
|
loop = asyncio.get_event_loop()
|
||||||
|
t0 = loop.time()
|
||||||
|
sent = 0
|
||||||
try:
|
try:
|
||||||
await client_ws.close()
|
while True:
|
||||||
except Exception:
|
target = t0 + sent / BYTES_PER_SEC
|
||||||
pass
|
delay = target - loop.time()
|
||||||
|
if delay > 0.001:
|
||||||
|
await asyncio.sleep(delay)
|
||||||
|
try:
|
||||||
|
chunk = pcm_queue.get_nowait()
|
||||||
|
if chunk is None:
|
||||||
|
break
|
||||||
|
except asyncio.QueueEmpty:
|
||||||
|
chunk = SILENCE_CHUNK
|
||||||
|
await client_ws.send_bytes(chunk)
|
||||||
|
sent += len(chunk)
|
||||||
|
except Exception as exc:
|
||||||
|
print(f"[{now()}] {conn_id} PACER send failed after {sent} B "
|
||||||
|
f"({sent/BYTES_PER_SEC:.2f}s): {exc!r}", flush=True)
|
||||||
|
return False
|
||||||
|
print(f"[{now()}] {conn_id} PACER done, {sent} B "
|
||||||
|
f"({sent/BYTES_PER_SEC:.2f}s)", flush=True)
|
||||||
|
return True
|
||||||
|
|
||||||
|
pacer_task = asyncio.create_task(pcm_pacer())
|
||||||
|
|
||||||
|
async def synth_worker() -> None:
|
||||||
|
timeout = aiohttp.ClientTimeout(total=SYNTH_TIMEOUT_S)
|
||||||
|
async with aiohttp.ClientSession(timeout=timeout) as http:
|
||||||
|
while True:
|
||||||
|
text = await sentence_queue.get()
|
||||||
|
if text is None:
|
||||||
|
break
|
||||||
|
try:
|
||||||
|
pcm = await synth_rest(http, cfg, text)
|
||||||
|
except Exception as exc:
|
||||||
|
print(f"[{now()}] {conn_id} REST synth failed for "
|
||||||
|
f"{text!r}: {exc!r}", flush=True)
|
||||||
|
continue
|
||||||
|
print(f"[{now()}] {conn_id} REST {text!r} -> {len(pcm)} B "
|
||||||
|
f"({len(pcm)/BYTES_PER_SEC:.2f}s)", flush=True)
|
||||||
|
if not started["audio"]:
|
||||||
|
started["audio"] = True
|
||||||
|
await client_ws.send_str(json.dumps(
|
||||||
|
{"type": "audio.start", "format": "pcm",
|
||||||
|
"sample_rate": SAMPLE_RATE}))
|
||||||
|
for _ in range(LEADING_FRAMES):
|
||||||
|
pcm_queue.put_nowait(SILENCE_CHUNK)
|
||||||
|
enqueue_pcm(pcm)
|
||||||
|
for _ in range(TRAILING_FRAMES):
|
||||||
|
pcm_queue.put_nowait(SILENCE_CHUNK)
|
||||||
|
pcm_queue.put_nowait(None)
|
||||||
|
|
||||||
|
worker_task = asyncio.create_task(synth_worker())
|
||||||
|
|
||||||
|
async def cleanup() -> None:
|
||||||
|
for t in (worker_task, pacer_task):
|
||||||
|
if not t.done():
|
||||||
|
t.cancel()
|
||||||
|
await asyncio.gather(worker_task, pacer_task, return_exceptions=True)
|
||||||
|
|
||||||
|
try:
|
||||||
|
async for msg in client_ws:
|
||||||
|
if msg.type == aiohttp.WSMsgType.TEXT:
|
||||||
|
try:
|
||||||
|
data = json.loads(msg.data)
|
||||||
|
typ = data.get("type")
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
if typ == "input.text":
|
||||||
|
text = data.get("text", "")
|
||||||
|
print(f"[{now()}] {conn_id} C text {text!r}", flush=True)
|
||||||
|
for sent in splitter.add(text):
|
||||||
|
sentence_queue.put_nowait(sent)
|
||||||
|
elif typ == "input.done":
|
||||||
|
print(f"[{now()}] {conn_id} C input.done", flush=True)
|
||||||
|
for sent in splitter.flush():
|
||||||
|
sentence_queue.put_nowait(sent)
|
||||||
|
sentence_queue.put_nowait(None)
|
||||||
|
drained_ok = await pacer_task
|
||||||
|
if not worker_task.done():
|
||||||
|
worker_task.cancel()
|
||||||
|
if drained_ok:
|
||||||
|
try:
|
||||||
|
await client_ws.send_str(
|
||||||
|
json.dumps({"type": "session.done"}))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return
|
||||||
|
elif msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.ERROR):
|
||||||
|
break
|
||||||
|
finally:
|
||||||
|
await cleanup()
|
||||||
|
|
||||||
|
|
||||||
async def main():
|
async def proxy_session_passthrough(client_ws: web.WebSocketResponse,
|
||||||
|
conn_id: str, first_msg_data: str) -> None:
|
||||||
|
"""Transparent WS passthrough to streaming upstream (non-PCM clients)."""
|
||||||
|
async with websockets.connect(_internal_ws_url, max_size=None) as upstream:
|
||||||
|
await upstream.send(first_msg_data)
|
||||||
|
|
||||||
|
async def c2u() -> None:
|
||||||
|
async for msg in client_ws:
|
||||||
|
if msg.type == aiohttp.WSMsgType.TEXT:
|
||||||
|
await upstream.send(msg.data)
|
||||||
|
elif msg.type == aiohttp.WSMsgType.BINARY:
|
||||||
|
await upstream.send(msg.data)
|
||||||
|
else:
|
||||||
|
break
|
||||||
|
|
||||||
|
async def u2c() -> None:
|
||||||
|
async for m in upstream:
|
||||||
|
if isinstance(m, str):
|
||||||
|
await client_ws.send_str(m)
|
||||||
|
else:
|
||||||
|
await client_ws.send_bytes(m)
|
||||||
|
|
||||||
|
await asyncio.gather(c2u(), u2c())
|
||||||
|
|
||||||
|
|
||||||
|
async def handle_ws(request: web.Request) -> web.WebSocketResponse:
|
||||||
|
conn_id = f"ws-{id(request):x}"
|
||||||
|
client_ws = web.WebSocketResponse(max_msg_size=0)
|
||||||
|
await client_ws.prepare(request)
|
||||||
|
print(f"[{now()}] {conn_id} open", flush=True)
|
||||||
|
|
||||||
|
# First client message must be session.config.
|
||||||
|
first = await client_ws.receive()
|
||||||
|
if first.type != aiohttp.WSMsgType.TEXT:
|
||||||
|
return client_ws
|
||||||
|
try:
|
||||||
|
cfg = json.loads(first.data)
|
||||||
|
except Exception:
|
||||||
|
cfg = {}
|
||||||
|
|
||||||
|
if cfg.get("response_format", "pcm") == "pcm":
|
||||||
|
log = dict(cfg)
|
||||||
|
if log.get("speaker_embedding"):
|
||||||
|
log["speaker_embedding"] = f"<{len(log['speaker_embedding'])} floats>"
|
||||||
|
print(f"[{now()}] {conn_id} config "
|
||||||
|
f"{json.dumps(log, ensure_ascii=False, sort_keys=True)}", flush=True)
|
||||||
|
await proxy_session_pcm(client_ws, conn_id, cfg)
|
||||||
|
else:
|
||||||
|
print(f"[{now()}] {conn_id} non-PCM -> passthrough", flush=True)
|
||||||
|
await proxy_session_passthrough(client_ws, conn_id, first.data)
|
||||||
|
|
||||||
|
return client_ws
|
||||||
|
|
||||||
|
|
||||||
|
# ── main ─────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
global _internal_base_url, _internal_ws_url
|
||||||
|
|
||||||
parser = argparse.ArgumentParser()
|
parser = argparse.ArgumentParser()
|
||||||
parser.add_argument("--host", default="0.0.0.0")
|
parser.add_argument("--host", default="0.0.0.0")
|
||||||
parser.add_argument("--port", type=int, default=8094)
|
parser.add_argument("--port", type=int, default=8091)
|
||||||
parser.add_argument("--upstream", default="ws://localhost:8091/v1/audio/speech/stream")
|
parser.add_argument("--upstream", default="ws://localhost:8095/v1/audio/speech/stream")
|
||||||
|
parser.add_argument("--rest", default=None,
|
||||||
|
help="Internal REST base URL (default: derived from --upstream)")
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
handler.upstream_url = args.upstream
|
|
||||||
print(f"WS log proxy: ws://{args.host}:{args.port}/v1/audio/speech/stream -> {args.upstream}", flush=True)
|
_internal_ws_url = args.upstream
|
||||||
async with websockets.serve(handler, args.host, args.port, max_size=None):
|
if args.rest:
|
||||||
await asyncio.Future()
|
_internal_base_url = args.rest.rstrip("/")
|
||||||
|
else:
|
||||||
|
http = args.upstream.replace("ws://", "http://").replace("wss://", "https://")
|
||||||
|
_internal_base_url = http.rsplit("/v1/", 1)[0]
|
||||||
|
|
||||||
|
print(f"Proxy on {args.host}:{args.port} "
|
||||||
|
f"(REST: {_internal_base_url}/v1/audio/speech "
|
||||||
|
f"WS: {_internal_ws_url})", flush=True)
|
||||||
|
|
||||||
|
app = web.Application()
|
||||||
|
app.router.add_get("/v1/audio/speech/stream", handle_ws)
|
||||||
|
app.router.add_post("/v1/audio/speech", handle_rest)
|
||||||
|
app.router.add_get("/health", handle_health)
|
||||||
|
|
||||||
|
runner = web.AppRunner(app, access_log=None)
|
||||||
|
await runner.setup()
|
||||||
|
site = web.TCPSite(runner, args.host, args.port)
|
||||||
|
await site.start()
|
||||||
|
print(f"Listening on http+ws://{args.host}:{args.port}", flush=True)
|
||||||
|
await asyncio.Future()
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
@@ -2,7 +2,7 @@
|
|||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
cd "$(dirname "$0")/.."
|
cd "$(dirname "$0")/.."
|
||||||
|
|
||||||
HEALTH_URL="http://localhost:8091/health"
|
HEALTH_URL="" # leer → nutzt docker inspect health status
|
||||||
MAX_WAIT="${MAX_WAIT:-180}"
|
MAX_WAIT="${MAX_WAIT:-180}"
|
||||||
[ -f .env ] && source .env
|
[ -f .env ] && source .env
|
||||||
MODEL="${QWEN3_TTS_CLONE_MODEL:-Qwen/Qwen3-TTS-12Hz-1.7B-Base}"
|
MODEL="${QWEN3_TTS_CLONE_MODEL:-Qwen/Qwen3-TTS-12Hz-1.7B-Base}"
|
||||||
@@ -25,16 +25,22 @@ printf " API http://localhost:8091
|
|||||||
"
|
"
|
||||||
|
|
||||||
while (( SECONDS - t0 < MAX_WAIT )); do
|
while (( SECONDS - t0 < MAX_WAIT )); do
|
||||||
if curl -sf "$HEALTH_URL" >/dev/null 2>&1; then
|
STATUS=$(docker inspect qwen3-tts-clone --format='{{.State.Health.Status}}' 2>/dev/null || true)
|
||||||
printf "✓ Clone bereit: http://localhost:8091 [%ds]
|
if [ "$STATUS" = "healthy" ]; then
|
||||||
|
printf "✓ Clone bereit: ws://localhost:8091 [%ds]
|
||||||
" $(( SECONDS - t0 ))
|
" $(( SECONDS - t0 ))
|
||||||
exit 0
|
exit 0
|
||||||
fi
|
fi
|
||||||
|
if [ "$STATUS" = "unhealthy" ]; then
|
||||||
|
printf "✗ Clone-Container unhealthy. Logs:
|
||||||
|
" >&2
|
||||||
|
docker logs --tail=20 qwen3-tts-clone 2>&1 >&2 || true
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
if ! docker ps --filter "name=qwen3-tts-clone" --filter "status=running" -q | grep -q .; then
|
if ! docker ps --filter "name=qwen3-tts-clone" --filter "status=running" -q | grep -q .; then
|
||||||
printf "✗ Clone-Container abgestürzt. Logs:
|
printf "✗ Clone-Container abgestürzt. Logs:
|
||||||
" >&2
|
" >&2
|
||||||
docker logs --tail=40 qwen3-tts-clone 2>&1 >&2 || true
|
docker logs --tail=20 qwen3-tts-clone 2>&1 >&2 || true
|
||||||
exit 1
|
exit 1
|
||||||
fi
|
fi
|
||||||
sleep 2
|
sleep 2
|
||||||
|
|||||||
@@ -2,7 +2,7 @@
|
|||||||
set -e
|
set -e
|
||||||
cd "$(dirname "$0")/.."
|
cd "$(dirname "$0")/.."
|
||||||
|
|
||||||
HEALTH_URL="http://localhost:8091/health"
|
HEALTH_URL="" # leer → nutzt docker inspect health status
|
||||||
MAX_WAIT=180
|
MAX_WAIT=180
|
||||||
# .env aus dem Projektroot laden (nur falls Variable nicht schon im Shell-Env gesetzt)
|
# .env aus dem Projektroot laden (nur falls Variable nicht schon im Shell-Env gesetzt)
|
||||||
[ -f .env ] && source .env
|
[ -f .env ] && source .env
|
||||||
@@ -59,13 +59,19 @@ LOG_PID=$!
|
|||||||
|
|
||||||
# --- Health-Check-Schleife ---
|
# --- Health-Check-Schleife ---
|
||||||
while (( SECONDS - t0 < MAX_WAIT )); do
|
while (( SECONDS - t0 < MAX_WAIT )); do
|
||||||
if curl -sf "$HEALTH_URL" >/dev/null 2>&1; then
|
STATUS=$(docker inspect qwen3-tts --format='{{.State.Health.Status}}' 2>/dev/null || true)
|
||||||
|
if [ "$STATUS" = "healthy" ]; then
|
||||||
kill "$LOG_PID" 2>/dev/null
|
kill "$LOG_PID" 2>/dev/null
|
||||||
wait "$LOG_PID" 2>/dev/null || true
|
wait "$LOG_PID" 2>/dev/null || true
|
||||||
printf "\n✓ Bereit: API http://localhost:8091 UI http://localhost:8092 WS-Log ws://localhost:8094/v1/audio/speech/stream [%ds]\n" $(( SECONDS - t0 ))
|
printf "\n✓ Bereit: API ws://localhost:8091 UI http://localhost:8092 [%ds]\n" $(( SECONDS - t0 ))
|
||||||
exit 0
|
exit 0
|
||||||
fi
|
fi
|
||||||
# Abbruch wenn Container schon gestoppt ist
|
if [ "$STATUS" = "unhealthy" ]; then
|
||||||
|
kill "$LOG_PID" 2>/dev/null
|
||||||
|
printf "\n✗ Container unhealthy. Logs:\n"
|
||||||
|
docker logs --tail=20 qwen3-tts 2>&1 | strip_ansi | grep -E "ERROR|ValueError|Exception" | head -5
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
if ! docker ps --filter "name=qwen3-tts" --filter "status=running" -q | grep -q .; then
|
if ! docker ps --filter "name=qwen3-tts" --filter "status=running" -q | grep -q .; then
|
||||||
kill "$LOG_PID" 2>/dev/null
|
kill "$LOG_PID" 2>/dev/null
|
||||||
printf "\n✗ Container abgestürzt. Logs:\n"
|
printf "\n✗ Container abgestürzt. Logs:\n"
|
||||||
|
|||||||
Reference in New Issue
Block a user