SenseVoice/SenseVoiceSmall-RKNN2/server.py

import os
import time
import logging
import re
print(f"Initial logging._nameToLevel: {logging._nameToLevel}")
from pathlib import Path
from typing import List, Dict, Any, Optional

import soundfile as sf
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# Ensure sensevoice_rknn.py is in the same directory or PYTHONPATH
# Add the directory of this script to sys.path if sensevoice_rknn is not found directly
import sys
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.append(str(SCRIPT_DIR))

try:
    from sensevoice_rknn import WavFrontend, SenseVoiceInferenceSession, FSMNVad, languages
except ImportError as e:
    logging.error(f"Error importing from sensevoice_rknn.py: {e}")
    logging.error("Please ensure sensevoice_rknn.py is in the same directory as server.py or in your PYTHONPATH.")
    # Fallback for critical components if import fails, to allow FastAPI to at least start and show an error
    class WavFrontend:
        def __init__(self, *args, **kwargs): raise NotImplementedError("WavFrontend not loaded")
        def get_features(self, *args, **kwargs): raise NotImplementedError("WavFrontend not loaded")
    class SenseVoiceInferenceSession:
        def __init__(self, *args, **kwargs): raise NotImplementedError("SenseVoiceInferenceSession not loaded")
        def __call__(self, *args, **kwargs): raise NotImplementedError("SenseVoiceInferenceSession not loaded")
    class FSMNVad:
        def __init__(self, *args, **kwargs): raise NotImplementedError("FSMNVad not loaded")
        def segments_offline(self, *args, **kwargs): raise NotImplementedError("FSMNVad not loaded")
        class Vad:
            def all_reset_detection(self, *args, **kwargs): raise NotImplementedError("FSMNVad not loaded")
        vad = Vad()

    languages = {"en": 4} # Default fallback

app = FastAPI()

# Logging will be handled by Uvicorn's default configuration or a custom log_config if provided to uvicorn.run
# Get a logger instance for application-specific logs if needed
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO) # Set level for this specific logger

# --- Model Configuration & Loading ---
MODEL_BASE_PATH = Path(__file__).resolve().parent

# These paths should match those used in sensevoice_rknn.py's main function
# or be configurable if they differ.
MVN_PATH = MODEL_BASE_PATH / "am.mvn"
EMBEDDING_NPY_PATH = MODEL_BASE_PATH / "embedding.npy"
ENCODER_RKNN_PATH = MODEL_BASE_PATH / "sense-voice-encoder.rknn"
BPE_MODEL_PATH = MODEL_BASE_PATH / "chn_jpn_yue_eng_ko_spectok.bpe.model"
VAD_CONFIG_DIR = MODEL_BASE_PATH # Assuming fsmn-config.yaml and fsmnvad-offline.onnx are here

# Global model instances
w_frontend: Optional[WavFrontend] = None
asr_model: Optional[SenseVoiceInferenceSession] = None
vad_model: Optional[FSMNVad] = None

@app.on_event("startup")
def load_models():
    global w_frontend, asr_model, vad_model
    logging.info("Loading models...")
    start_time = time.time()
    try:
        if not MVN_PATH.exists():
            raise FileNotFoundError(f"CMVN file not found: {MVN_PATH}")
        w_frontend = WavFrontend(cmvn_file=str(MVN_PATH))

        if not EMBEDDING_NPY_PATH.exists() or not ENCODER_RKNN_PATH.exists() or not BPE_MODEL_PATH.exists():
            raise FileNotFoundError(
                f"One or more ASR model files not found: "
                f"Embedding: {EMBEDDING_NPY_PATH}, Encoder: {ENCODER_RKNN_PATH}, BPE: {BPE_MODEL_PATH}"
            )
        asr_model = SenseVoiceInferenceSession(
            embedding_model_file=str(EMBEDDING_NPY_PATH),
            encoder_model_file=str(ENCODER_RKNN_PATH),
            bpe_model_file=str(BPE_MODEL_PATH),
            # Assuming default device_id and num_threads as in sensevoice_rknn.py's main
            device_id=-1, 
            intra_op_num_threads=4 
        )

        # Check for VAD model files (fsmn-config.yaml, fsmnvad-offline.onnx)
        if not (VAD_CONFIG_DIR / "fsmn-config.yaml").exists() or not (VAD_CONFIG_DIR / "fsmnvad-offline.onnx").exists():
             raise FileNotFoundError(f"VAD config or model not found in {VAD_CONFIG_DIR}")
        vad_model = FSMNVad(config_dir=str(VAD_CONFIG_DIR))
        
        logging.info(f"Models loaded successfully in {time.time() - start_time:.2f} seconds.")
    except FileNotFoundError as e:
        logging.error(f"Model loading failed: {e}")
        # Keep models as None, endpoints will raise errors
    except Exception as e:
        logging.error(f"An unexpected error occurred during model loading: {e}")
        # Keep models as None

class TranscribeRequest(BaseModel):
    audio_file_path: str
    language: str = "en"  # Default to English
    use_itn: bool = False

class Segment(BaseModel):
    start_time_s: float
    end_time_s: float
    text: str

class TranscribeResponse(BaseModel):
    full_transcription: str
    segments: List[Segment]

@app.post("/transcribe", response_model=str)
async def transcribe_audio(request: TranscribeRequest):
    if w_frontend is None or asr_model is None or vad_model is None:
        logging.error("Models not loaded. Transcription cannot proceed.")
        raise HTTPException(status_code=503, detail="Models are not loaded. Please check server logs.")

    audio_path = Path(request.audio_file_path)
    if not audio_path.exists() or not audio_path.is_file():
        logging.error(f"Audio file not found: {audio_path}")
        raise HTTPException(status_code=404, detail=f"Audio file not found: {audio_path}")

    try:
        waveform, sample_rate = sf.read(
            str(audio_path),
            dtype="float32",
            always_2d=True
        )
    except Exception as e:
        logging.error(f"Error reading audio file {audio_path}: {e}")
        raise HTTPException(status_code=400, detail=f"Could not read audio file: {e}")

    if sample_rate != 16000:
        # Basic resampling could be added here if needed, or just raise an error
        logging.warning(f"Audio sample rate is {sample_rate}Hz, expected 16000Hz. Results may be suboptimal.")
        # For now, we proceed but log a warning. For critical applications, convert or reject.

    logging.info(f"Processing audio: {audio_path}, Duration: {len(waveform) / sample_rate:.2f}s, Channels: {waveform.shape[1]}")

    lang_code = languages.get(request.language.lower())
    if lang_code is None:
        logging.warning(f"Unsupported language: {request.language}. Defaulting to 'en'. Supported: {list(languages.keys())}")
        lang_code = languages.get("en", 0) # Fallback to 'en' or 'auto' if 'en' isn't in languages

    all_segments_text: List[str] = []
    detailed_segments: List[Segment] = []
    processing_start_time = time.time()

    for channel_id in range(waveform.shape[1]):
        channel_data = waveform[:, channel_id]
        logging.info(f"Processing channel {channel_id + 1}/{waveform.shape[1]}")
        
        try:
            # Ensure channel_data is 1D for VAD if it expects that
            speech_segments = vad_model.segments_offline(channel_data) # segments_offline expects 1D array
        except Exception as e:
            logging.error(f"VAD processing failed for channel {channel_id}: {e}")
            # Optionally skip this channel or raise an error for the whole request
            continue # Skip to next channel

        for part_idx, part in enumerate(speech_segments):
            start_sample = int(part[0] * 16)  # VAD returns ms, convert to samples (16 samples/ms for 16kHz)
            end_sample = int(part[1] * 16)
            segment_audio = channel_data[start_sample:end_sample]

            if len(segment_audio) == 0:
                logging.info(f"Empty audio segment for channel {channel_id}, part {part_idx}. Skipping.")
                continue
            
            try:
                # Ensure get_features expects 1D array
                audio_feats = w_frontend.get_features(segment_audio) 
                # ASR model expects batch dimension, add [None, ...]
                asr_result_text_raw = asr_model(
                    audio_feats[None, ...],
                    language=lang_code,
                    use_itn=request.use_itn,
                )
                # Remove tags like <|en|>, <|HAPPY|>, etc.
                asr_result_text_cleaned = re.sub(r"<\|[^\|]+\|>", "", asr_result_text_raw).strip()
                
                segment_start_s = part[0] / 1000.0
                segment_end_s = part[1] / 1000.0
                logging.info(f"[Ch{channel_id}] [{segment_start_s:.2f}s - {segment_end_s:.2f}s] Raw: {asr_result_text_raw} Cleaned: {asr_result_text_cleaned}")
                all_segments_text.append(asr_result_text_cleaned)
                detailed_segments.append(Segment(start_time_s=segment_start_s, end_time_s=segment_end_s, text=asr_result_text_cleaned))
            except Exception as e:
                logging.error(f"ASR processing failed for segment {part_idx} in channel {channel_id}: {e}")
                # Optionally add a placeholder or skip this segment's text
                detailed_segments.append(Segment(start_time_s=part[0]/1000.0, end_time_s=part[1]/1000.0, text="[ASR_ERROR]"))

        vad_model.vad.all_reset_detection() # Reset VAD state for next channel or call

    full_transcription = " ".join(all_segments_text).strip()
    logging.info(f"Transcription complete in {time.time() - processing_start_time:.2f}s. Result: {full_transcription}")

    return full_transcription

if __name__ == "__main__":
    import uvicorn

    MINIMAL_LOGGING_CONFIG = {
        "version": 1,
        "disable_existing_loggers": False, # Let other loggers (like our app logger) exist
        "formatters": {
            "default": {
                "()": "uvicorn.logging.DefaultFormatter",
                "fmt": "%(levelprefix)s %(message)s",
                "use_colors": None,
            },
        },
        "handlers": {
            "default": {
                "formatter": "default",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stderr",
            },
        },
        "loggers": {
            "uvicorn": { # Uvicorn's own operational logs
                "handlers": ["default"],
                "level": logging.INFO, # Explicitly use integer
                "propagate": False,
            },
            "uvicorn.error": { # Logs for errors within Uvicorn
                "handlers": ["default"],
                "level": logging.INFO, # Explicitly use integer
                "propagate": False,
            },
            # We are deliberately not configuring uvicorn.access here for simplicity
            # It might default to INFO or be silent if not configured and no parent handler catches it.
        },
        # Ensure our application logger also works if needed
        __name__: {
            "handlers": ["default"],
            "level": logging.INFO,
            "propagate": False,
        }
    }

    logger.info(f"Attempting to run Uvicorn with minimal explicit log_config.")
    uvicorn.run(app, host="0.0.0.0", port=8000, log_config=MINIMAL_LOGGING_CONFIG)