wehub-resource-sync
154 lines
4.6 KiB
Python
154 lines
4.6 KiB
Python
import importlib.util
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SERVICE_PATH = (
|
|
REPO_ROOT
|
|
/ "examples"
|
|
/ "industrial_data_pretraining"
|
|
/ "fun_asr_nano"
|
|
/ "serve_vllm.py"
|
|
)
|
|
|
|
|
|
def load_service_module(monkeypatch):
|
|
fastapi_stub = types.ModuleType("fastapi")
|
|
|
|
class FastAPIStub:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def on_event(self, *args, **kwargs):
|
|
return lambda func: func
|
|
|
|
def post(self, *args, **kwargs):
|
|
return lambda func: func
|
|
|
|
def websocket(self, *args, **kwargs):
|
|
return lambda func: func
|
|
|
|
fastapi_stub.FastAPI = FastAPIStub
|
|
fastapi_stub.File = lambda *args, **kwargs: None
|
|
fastapi_stub.Form = lambda *args, **kwargs: None
|
|
fastapi_stub.UploadFile = object
|
|
fastapi_stub.WebSocket = object
|
|
fastapi_stub.WebSocketDisconnect = Exception
|
|
|
|
responses_stub = types.ModuleType("fastapi.responses")
|
|
|
|
class JSONResponseStub:
|
|
def __init__(self, content=None):
|
|
self.content = content
|
|
|
|
responses_stub.JSONResponse = JSONResponseStub
|
|
|
|
funasr_stub = types.ModuleType("funasr")
|
|
funasr_stub.AutoModel = object
|
|
|
|
nano_stub = types.ModuleType("funasr.models.fun_asr_nano.inference_vllm")
|
|
nano_stub.FunASRNanoVLLM = object
|
|
|
|
vad_stub = types.ModuleType("funasr.models.fsmn_vad_streaming.dynamic_vad")
|
|
vad_stub.DynamicStreamingVAD = object
|
|
|
|
monkeypatch.setitem(sys.modules, "fastapi", fastapi_stub)
|
|
monkeypatch.setitem(sys.modules, "fastapi.responses", responses_stub)
|
|
monkeypatch.setitem(sys.modules, "uvicorn", types.ModuleType("uvicorn"))
|
|
monkeypatch.setitem(sys.modules, "soundfile", types.ModuleType("soundfile"))
|
|
monkeypatch.setitem(sys.modules, "torch", types.ModuleType("torch"))
|
|
monkeypatch.setitem(sys.modules, "funasr", funasr_stub)
|
|
monkeypatch.setitem(sys.modules, "funasr.models", types.ModuleType("funasr.models"))
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"funasr.models.fun_asr_nano",
|
|
types.ModuleType("funasr.models.fun_asr_nano"),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules, "funasr.models.fun_asr_nano.inference_vllm", nano_stub
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"funasr.models.fsmn_vad_streaming",
|
|
types.ModuleType("funasr.models.fsmn_vad_streaming"),
|
|
)
|
|
monkeypatch.setitem(
|
|
sys.modules, "funasr.models.fsmn_vad_streaming.dynamic_vad", vad_stub
|
|
)
|
|
|
|
module_name = "serve_vllm_under_test"
|
|
sys.modules.pop(module_name, None)
|
|
spec = importlib.util.spec_from_file_location(module_name, SERVICE_PATH)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_openai_verbose_json_keeps_segment_speaker(monkeypatch):
|
|
module = load_service_module(monkeypatch)
|
|
|
|
response = module.build_openai_verbose_json(
|
|
{
|
|
"duration": 1.2,
|
|
"text": "hello",
|
|
"segments": [
|
|
{
|
|
"start": 0.0,
|
|
"end": 1.2,
|
|
"text": "hello",
|
|
"words": [{"word": "hello", "start": 0.0, "end": 1.2}],
|
|
"speaker": "SPK0",
|
|
}
|
|
],
|
|
},
|
|
language="en",
|
|
)
|
|
|
|
assert response["segments"][0]["speaker"] == "SPK0"
|
|
|
|
|
|
def test_process_audio_downmixes_stereo_before_resampling(monkeypatch):
|
|
module = load_service_module(monkeypatch)
|
|
captured = {}
|
|
|
|
librosa_stub = types.ModuleType("librosa")
|
|
|
|
def fake_resample(audio, orig_sr, target_sr):
|
|
assert audio.ndim == 1
|
|
captured["resample_input"] = audio.copy()
|
|
assert orig_sr == 8000
|
|
assert target_sr == 16000
|
|
return np.repeat(audio, 2)
|
|
|
|
librosa_stub.resample = fake_resample
|
|
monkeypatch.setitem(sys.modules, "librosa", librosa_stub)
|
|
|
|
class EngineStub:
|
|
def generate(self, inputs, **kwargs):
|
|
captured["engine_input"] = inputs[0]
|
|
return [{"text": "ok"}]
|
|
|
|
module._engine = EngineStub()
|
|
stereo_audio = np.column_stack(
|
|
[np.zeros(8000, dtype=np.float32), np.ones(8000, dtype=np.float32)]
|
|
)
|
|
|
|
result = module.process_audio(
|
|
stereo_audio,
|
|
sr=8000,
|
|
use_vad=False,
|
|
use_spk=False,
|
|
use_timestamp=False,
|
|
)
|
|
|
|
assert np.allclose(captured["resample_input"], 0.5)
|
|
assert captured["engine_input"].shape == (16000,)
|
|
assert np.allclose(captured["engine_input"], 0.5)
|
|
assert result["duration"] == 1.0
|
|
assert result["text"] == "ok"
|