| """Самодиагностика 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) |
|
|
|
|
| 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()) |
|
|
