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
|
"""Shared code for HTTP servers."""
import argparse
from pathlib import Path
from typing import Union
from flask import Flask, jsonify, redirect, request
from wyoming.client import AsyncClient
from wyoming.info import Describe, Info
def get_argument_parser() -> argparse.ArgumentParser:
"""Create argument parser with shared arguments."""
parser = argparse.ArgumentParser()
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=5000)
parser.add_argument("--uri", help="URI of Wyoming service")
parser.add_argument(
"--debug", action="store_true", help="Print DEBUG logs to console"
)
return parser
def get_app(
name: str, openapi_config_path: Union[str, Path], args: argparse.Namespace
) -> Flask:
"""Create Flask app with default endpoints."""
app = Flask(name)
@app.route("/")
def redirect_to_api():
return redirect("/api")
@app.route("/api/info", methods=["GET"])
async def api_info():
uri = request.args.get("uri", args.uri)
if not uri:
raise ValueError("URI is required")
async with AsyncClient.from_uri(uri) as client:
await client.write_event(Describe().event())
while True:
event = await client.read_event()
if event is None:
raise RuntimeError("Client disconnected")
if Info.is_type(event.type):
info = Info.from_event(event)
return jsonify(info.to_dict())
@app.errorhandler(Exception)
async def handle_error(err):
"""Return error as text."""
return (f"{err.__class__.__name__}: {err}", 500)
return app
|
