File size: 7,017 Bytes
ed81d38 8879b17 ed81d38 8879b17 ed81d38 8879b17 ed81d38 8879b17 ed81d38 | 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 | """Самодиагностика MuScriptor: проверить сервер, прокси и саму транскрипцию.
Запуск в ноутбуке (пока пускач работает в соседней ячейке):
python3 plus/selfcheck.py
или с явными портами:
python3 plus/selfcheck.py --server 8222 --proxy 8223 --url https://...
"""
import argparse
import json
import math
import os
import struct
import sys
import time
import urllib.error
import urllib.request
import wave
PROBE = "/tmp/muscriptor_probe.wav"
PROBE_FREQS = (523.25, 659.25, 783.99) # C5, E5, G5
def make_probe(path=PROBE, seconds=2.0, rate=44100):
"""Короткий аккорд: три ноты с затуханием, чтобы было что транскрибировать."""
total = int(seconds * rate)
buffer = [0.0] * total
for index, freq in enumerate(PROBE_FREQS):
start = int((0.15 + 0.35 * index) * rate)
length = int(0.9 * rate)
for i in range(length):
if start + i >= total:
break
envelope = math.exp(-4.0 * i / rate)
buffer[start + i] += 0.28 * envelope * (
math.sin(2 * math.pi * freq * i / rate)
+ 0.35 * math.sin(4 * math.pi * freq * i / rate)
+ 0.15 * math.sin(6 * math.pi * freq * i / rate))
with wave.open(path, "w") as handle:
handle.setnchannels(1)
handle.setsampwidth(2)
handle.setframerate(rate)
handle.writeframes(b"".join(struct.pack("<h", int(max(-1.0, min(1.0, v)) * 32767))
for v in buffer))
return path
BYPASS = {"serveo-skip-browser-warning": "true"}
def get(url, timeout=20):
started = time.time()
try:
request = urllib.request.Request(url, headers=BYPASS)
with urllib.request.urlopen(request, timeout=timeout) as answer:
body = answer.read()
return answer.status, body, time.time() - started
except urllib.error.HTTPError as problem:
return problem.code, problem.read()[:400], time.time() - started
except Exception as problem:
return None, str(problem).encode("utf-8"), time.time() - started
def post_transcribe(base, audio_path, timeout=900):
import urllib.request
boundary = "----selfcheck%d" % int(time.time())
with open(audio_path, "rb") as handle:
audio = handle.read()
body = b"".join([
("--%s\r\n" % boundary).encode(),
b'Content-Disposition: form-data; name="file"; filename="probe.wav"\r\n',
b"Content-Type: audio/wav\r\n\r\n", audio, b"\r\n",
("--%s--\r\n" % boundary).encode()])
request = urllib.request.Request(
base + "/transcribe", data=body,
headers={"Content-Type": "multipart/form-data; boundary=%s" % boundary})
started = time.time()
pings = 0
notes = 0
text = ""
try:
with urllib.request.urlopen(request, timeout=timeout) as answer:
while True:
chunk = answer.read(4096)
if not chunk:
break
piece = chunk.decode("utf-8", "ignore")
text += piece
pings += piece.count(": ping")
notes = text.count('"type": "start"')
except Exception as problem:
return {"error": str(problem), "seconds": round(time.time() - started, 1),
"pings": pings, "notes": notes}
return {"error": None, "seconds": round(time.time() - started, 1),
"pings": pings, "notes": notes, "pings_seen": ": ping" in text}
def main():
parser = argparse.ArgumentParser(description="Проверка запущенного MuScriptor")
parser.add_argument("--server", type=int, default=8222)
parser.add_argument("--proxy", type=int, default=8223)
parser.add_argument("--url", default="", help="публичная ссылка, если есть")
parser.add_argument("--skip-transcribe", action="store_true")
args = parser.parse_args()
audio = make_probe()
print("пробный файл: %s (%.1f с)" % (audio, os.path.getsize(audio) / 1024 / 32))
print("\n[1] сервер :%d" % args.server)
status, body, spent = get("http://127.0.0.1:%d/health" % args.server)
print(" /health → %s за %.1f с %s" % (status, spent, body[:80]))
print("\n[2] прокси :%d" % args.proxy)
status, body, spent = get("http://127.0.0.1:%d/instruments" % args.proxy)
count = 0
try:
count = len(json.loads(body.decode("utf-8")).get("instruments", []))
except Exception:
pass
print(" /instruments → %s за %.1f с, список из %d инструментов" % (status, spent, count))
if status != 200 or count == 0:
print(" ВЫВОД: интерфейсу нечего показать в панели Instruments,"
" прокси или сервер не отдают /instruments")
if args.url:
print("\n[3] публичная ссылка")
status, body, spent = get(args.url.rstrip("/") + "/instruments", timeout=30)
text = body.decode("utf-8", "ignore")
print(" ссылка/instruments → %s за %.1f с" % (status, spent))
if status != 200:
print(" ВЫВОД: снаружи не отвечает, туннель не смотрит на прокси")
elif "instruments" not in text:
print(" ВЫВОД: вместо JSON пришла страница провайдера"
" (заглушка), API работать не будет")
else:
try:
count = len(json.loads(text).get("instruments", []))
print(" снаружи видно %d инструментов" % count)
except Exception:
print(" ВЫВОД: ответ снаружи не разобрать как JSON")
if not args.skip_transcribe:
print("\n[4] сама транскрипция через прокси")
result = post_transcribe("http://127.0.0.1:%d" % args.proxy, audio)
if result["error"]:
print(" ошибка: %s за %.1f с" % (result["error"], result["seconds"]))
else:
print(" ответ за %.1f с | нот: %d | пульсов по дороге: %d"
% (result["seconds"], result["notes"], result["pings"]))
if result["notes"] == 0:
print(" ВЫВОД: поток пришёл, но нот нет — смотри muscriptor_server.log")
elif result["pings"] == 0:
print(" ВНИМАНИЕ: нот %d, но пульсов не было; на длинном файле"
" туннель может закрыть соединение" % result["notes"])
print("\nготово")
if __name__ == "__main__":
sys.exit(main())
|