AuraVision/app.py at main · Rfannn/AuraVision · GitHub

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
import logging
import os
import sys
from functools import wraps
from typing import Any

import pyaudio
from flask import Flask, render_template, request, jsonify, send_from_directory
from flask_socketio import SocketIO
from flask_cors import CORS

from config import (
    SECRET_KEY,
    AUTH_TOKEN,
    FLASK_HOST,
    FLASK_PORT,
    FLASK_DEBUG,
    LOG_LEVEL,
    BASE_DIR,
)

logging.basicConfig(
    level=getattr(logging, LOG_LEVEL, logging.ERROR),
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger("auravision")

app = Flask(__name__, static_folder="static")
app.config["SECRET_KEY"] = SECRET_KEY
CORS(app)
socketio = SocketIO(app, cors_allowed_origins="*", async_mode="threading", logger=False, engineio_logger=False)

detected_lang: str = "en"


def require_token(f: Any) -> Any:
    @wraps(f)
    def decorated(*args: Any, **kwargs: Any) -> Any:
        if AUTH_TOKEN:
            token = request.headers.get("X-Auth-Token", "")
            if token != AUTH_TOKEN:
                return jsonify(success=False, error="Unauthorized"), 401
        return f(*args, **kwargs)
    return decorated


@app.route("/")
def index() -> Any:
    return render_template("index.html")


@app.route("/favicon.ico")
def favicon() -> Any:
    return send_from_directory(app.static_folder, "favicon.svg", mimetype="image/svg+xml")


@app.route("/mics")
def list_mics() -> Any:
    pa = pyaudio.PyAudio()
    mics = []
    default_idx = -1
    try:
        default_info = pa.get_default_input_device_info()
        default_idx = default_info["index"]
    except OSError:
        pass
    for i in range(pa.get_device_count()):
        info = pa.get_device_info_by_index(i)
        if info["maxInputChannels"] > 0:
            mics.append({
                "index": i,
                "name": info["name"],
                "channels": int(info["maxInputChannels"]),
                "rate": int(info["defaultSampleRate"]),
            })
    pa.terminate()
    return jsonify(mics=mics, default=default_idx)


@app.route("/update", methods=["POST"])
@require_token
def update() -> Any:
    global detected_lang

    data = request.get_json(silent=True)
    if not data or "text" not in data:
        return jsonify(success=False, error="No text provided"), 400

    text = data["text"].strip()
    lang = data.get("lang", "en")
    partial = data.get("partial", False)

    if not text:
        return jsonify(success=False, error="Empty text"), 400

    if lang != detected_lang:
        detected_lang = lang
        socketio.emit("lang_change", {"lang": lang})

    logger.info("[%s] %s%s", lang, text, " (partial)" if partial else "")

    event = "partial_text" if partial else "update_text"
    socketio.emit(event, {"text": text, "lang": lang})
    return jsonify(success=True)


@app.route("/health")
def health() -> Any:
    return jsonify(status="ok")


@socketio.on("connect")
def on_connect() -> None:
    logger.debug("Client connected")
    socketio.emit("lang_change", {"lang": detected_lang})


@socketio.on("disconnect")
def on_disconnect() -> None:
    logger.debug("Client disconnected")


@socketio.on("set_mic")
def on_set_mic(data: dict[str, Any]) -> None:
    logger.info("Mic change requested: %s", data)
    socketio.emit("set_mic", data)


if __name__ == "__main__":
    socketio.run(app, host=FLASK_HOST, port=FLASK_PORT, debug=FLASK_DEBUG)