Adolfo Reyna
146 lines
4.4 KiB
Python
Executable File
146 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import http.client
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
DEFAULT_URL = os.environ.get("KSAY_URL", "http://127.0.0.1:7332")
|
|
LAUNCHD_LABEL = os.environ.get("KSAY_LAUNCHD_LABEL", "com.local.ksay-kokoro")
|
|
|
|
|
|
def read_stdin_if_needed(text_parts: list[str]) -> str:
|
|
if text_parts:
|
|
return " ".join(text_parts)
|
|
if not sys.stdin.isatty():
|
|
return sys.stdin.read()
|
|
return ""
|
|
|
|
|
|
def request_json(method: str, path: str, payload: dict | None = None) -> dict:
|
|
data = None
|
|
headers = {}
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
headers["content-type"] = "application/json"
|
|
req = urllib.request.Request(
|
|
f"{DEFAULT_URL}{path}",
|
|
data=data,
|
|
headers=headers,
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=120) as res:
|
|
return json.loads(res.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read().decode("utf-8", errors="replace")
|
|
try:
|
|
return json.loads(body)
|
|
except json.JSONDecodeError:
|
|
raise RuntimeError(f"ksay server returned HTTP {exc.code}: {body}") from exc
|
|
|
|
|
|
def try_wake_service() -> None:
|
|
domain = f"gui/{os.getuid()}"
|
|
subprocess.run(
|
|
["launchctl", "kickstart", "-k", f"{domain}/{LAUNCHD_LABEL}"],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def post_say(payload: dict) -> dict:
|
|
try:
|
|
return request_json("POST", "/say", payload)
|
|
except (
|
|
ConnectionError,
|
|
ConnectionResetError,
|
|
http.client.RemoteDisconnected,
|
|
urllib.error.URLError,
|
|
):
|
|
try_wake_service()
|
|
time.sleep(1.0)
|
|
return request_json("POST", "/say", payload)
|
|
|
|
|
|
def play_audio(path: str) -> None:
|
|
subprocess.run(["/usr/bin/afplay", path], check=True)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
prog="ksay",
|
|
description="Fast warm Kokoro TTS, intended as a neural replacement for macOS say.",
|
|
)
|
|
parser.add_argument("text", nargs="*", help="Text to speak. Reads stdin when omitted.")
|
|
parser.add_argument("-v", "--voice", default=os.environ.get("KSAY_VOICE", "af_heart"))
|
|
parser.add_argument("-r", "--rate", type=float, default=None, help="Compatibility alias; maps words/minute-ish values to speed.")
|
|
parser.add_argument("--speed", type=float, default=None, help="Kokoro speed multiplier.")
|
|
parser.add_argument("--lang-code", default=os.environ.get("KSAY_LANG_CODE", "a"))
|
|
parser.add_argument("-o", "--output", help="Write WAV to this path.")
|
|
parser.add_argument("--no-play", action="store_true", help="Generate the file without playing it.")
|
|
parser.add_argument("--json", action="store_true", help="Print the server response as JSON.")
|
|
parser.add_argument("--status", action="store_true", help="Show warm server health.")
|
|
parser.add_argument("--start", action="store_true", help="Kick the launchd service and wait for health.")
|
|
args = parser.parse_args()
|
|
|
|
if args.status:
|
|
print(json.dumps(request_json("GET", "/health"), indent=2))
|
|
return 0
|
|
|
|
if args.start:
|
|
try_wake_service()
|
|
for _ in range(60):
|
|
try:
|
|
print(json.dumps(request_json("GET", "/health"), indent=2))
|
|
return 0
|
|
except Exception:
|
|
time.sleep(1)
|
|
print("ksay service did not become healthy within 60s.", file=sys.stderr)
|
|
return 1
|
|
|
|
text = read_stdin_if_needed(args.text).strip()
|
|
if not text:
|
|
parser.error("text is required, or pipe text on stdin")
|
|
|
|
speed = args.speed
|
|
if speed is None:
|
|
speed = 1.0
|
|
if args.rate:
|
|
speed = max(0.5, min(2.0, args.rate / 180.0))
|
|
|
|
result = post_say(
|
|
{
|
|
"text": text,
|
|
"voice": args.voice,
|
|
"speed": speed,
|
|
"langCode": args.lang_code,
|
|
"output": args.output,
|
|
}
|
|
)
|
|
if not result.get("ok"):
|
|
print(result.get("error", "ksay failed"), file=sys.stderr)
|
|
return 1
|
|
|
|
if args.json:
|
|
print(json.dumps(result, indent=2))
|
|
else:
|
|
print(result["filePath"])
|
|
|
|
if not args.no_play:
|
|
play_audio(result["filePath"])
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|