1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
"""Identify which TTS backend answers at a URL (remote-server probing).
The hub keeps locally-managed backends distinct from externally-run ones: a
server this tool started is tagged "[local]", and a server found by probing a
configured remote URL (``*_REMOTE_URL`` in ``app/converter/config.py``) is
tagged "[remote]". To know that a remote URL really hosts the backend we
think it does (and not some other HTTP service), each backend exposes a small
identity check over plain HTTP:
* audio.cpp ``GET /health`` -> ``{"status": "ok"}`` and ``GET /v1/models``
-> ``{"data": [{"id": ...}, ...]}``.
* faster ``GET /health`` -> a JSON object with a ``model_loaded`` key.
* qwen-tts a Gradio app: ``GET /info`` -> ``named_endpoints`` containing
the endpoint names the converter calls (``/run_instruct`` /
``/run_custom_voice`` / ``/generate_custom_voice`` for the
CustomVoice demo; ``/run_voice_clone`` / ``/generate_voice_clone``
for the Base demo; ``/run_voice_design`` for the VoiceDesign
demo).
``identify_server`` returns one of the IDENTITY_* constants, or None when the
URL does not answer or answers as something unrecognized. It is stdlib-only
(urllib) and deliberately imports nothing from the other backend modules, so
it stays cheap to import alongside ``backends.common``.
"""
import json
import urllib.parse
import urllib.request
from typing import Optional
from backends import common
IDENTITY_AUDIOCPP = "audiocpp"
IDENTITY_FASTER = "faster"
IDENTITY_QWEN_CUSTOM = "qwen-custom"
IDENTITY_QWEN_CLONE = "qwen-clone"
IDENTITY_QWEN_DESIGN = "qwen-design"
# Endpoint names the converter resolves for each qwen demo server (see
# converter.clients QwenTTSClient). Mirror them here so identification matches
# exactly what the converter would call.
_QWEN_CUSTOM_ENDPOINTS = (
"/run_instruct", "/run_custom_voice", "/generate_custom_voice")
_QWEN_CLONE_ENDPOINTS = ("/run_voice_clone", "/generate_voice_clone")
_QWEN_DESIGN_ENDPOINTS = ("/run_voice_design",)
DEFAULT_TIMEOUT = 3.0
def identify_server(url: str, timeout: float = DEFAULT_TIMEOUT) -> Optional[str]:
"""Return the backend identity answering at URL, or None.
A cheap TCP-connect gate runs first (``common.server_running``) so a dead
or unrouteable host returns quickly; the HTTP probes only run when
something is listening. Returns None when the URL is empty/unparsable,
unreachable, or answers as none of the known backends.
"""
if not url:
return None
base = url.rstrip("/")
if not common.server_running(url):
return None
identity = _identify_health(base, timeout)
if identity is not None:
return identity
return _identify_gradio(base, timeout)
def _get_json(url: str, timeout: float) -> Optional[dict]:
"""GET URL and parse a JSON object, or None on any error."""
try:
with urllib.request.urlopen(url, timeout=timeout) as response:
payload = json.loads(response.read().decode("utf-8"))
except (OSError, ValueError):
return None
return payload if isinstance(payload, dict) else None
def _identify_health(base: str, timeout: float) -> Optional[str]:
"""Identify audio.cpp / faster from their ``/health`` responses."""
payload = _get_json(f"{base}/health", timeout)
if payload is None:
return None
# faster's /health reports model load state under "model_loaded".
if "model_loaded" in payload:
return IDENTITY_FASTER
# audio.cpp's /health reports {"status": "ok"}; confirm it also serves
# the /v1/models catalog (id-bearing entries) to avoid mistaking some
# other service that happens to return {"status": "ok"}.
if payload.get("status") == "ok":
models = _get_json(f"{base}/v1/models", timeout)
entries = models.get("data") if models is not None else None
if isinstance(entries, list) and entries \
and any(isinstance(e, dict) and e.get("id") for e in entries):
return IDENTITY_AUDIOCPP
return None
def _identify_gradio(base: str, timeout: float) -> Optional[str]:
"""Identify a qwen-tts Gradio demo from its ``/info`` named endpoints."""
payload = _get_json(f"{base}/info", timeout)
if payload is None:
return None
endpoints = payload.get("named_endpoints")
if not isinstance(endpoints, dict):
return None
if any(name in endpoints for name in _QWEN_CUSTOM_ENDPOINTS):
return IDENTITY_QWEN_CUSTOM
if any(name in endpoints for name in _QWEN_CLONE_ENDPOINTS):
return IDENTITY_QWEN_CLONE
if any(name in endpoints for name in _QWEN_DESIGN_ENDPOINTS):
return IDENTITY_QWEN_DESIGN
return None
def _canonical_host(host: str) -> str:
"""Fold the loopback aliases so "localhost" and "127.0.0.1" compare equal."""
return "127.0.0.1" if host in ("localhost", "::1", "[::1]") else host
def same_endpoint(url_a: str, url_b: str) -> bool:
"""True when URL_A and URL_B address the same host and port.
Scheme and path are ignored (127.0.0.1:8080 and http://127.0.0.1:8080/
are the same server), and the loopback names are folded together
("localhost:8080" equals "127.0.0.1:8080") — the config's remote-URL
defaults point at the managed servers, so a user writing either form
must not get their own server double-counted as "[remote]".
Returns False when either URL is empty/unparsable.
"""
if not url_a or not url_b:
return False
try:
a = urllib.parse.urlsplit(url_a)
b = urllib.parse.urlsplit(url_b)
except ValueError:
return False
host_a = _canonical_host(a.hostname or "127.0.0.1")
host_b = _canonical_host(b.hostname or "127.0.0.1")
port_a = a.port or (443 if (a.scheme or "http") == "https" else 80)
port_b = b.port or (443 if (b.scheme or "http") == "https" else 80)
return host_a == host_b and port_a == port_b
def health_payload(url: str, timeout: float = DEFAULT_TIMEOUT) -> Optional[dict]:
"""Return the server's ``/health`` JSON document, or None.
A cheaper, raw check than ``identify_server``: used by the run view's
background poll to tell "process alive" from "server answering" without
probing every identity endpoint.
"""
if not url:
return None
return _get_json(f"{url.rstrip('/')}/health", timeout)
def faster_model_loaded(url: str, timeout: float = DEFAULT_TIMEOUT) -> bool:
"""True when a faster server at URL reports its model loaded.
faster's ``/health`` answers with a ``model_loaded`` flag only after the
weights are resident, so this is the "truly ready" signal used while
waiting for a started server to become usable.
"""
payload = health_payload(url, timeout)
return bool(payload and payload.get("model_loaded"))
|
