handler.py

from typing import Dict
from faster_whisper import WhisperModel
import io
import re

class EndpointHandler:
    def __init__(self, model_dir=None):
        # The compute_type is set to "float16" for efficient GPU computation
        # For "int8" computation on CPU, the compute_type would be "int8"
        compute_type = "float16"

        # Initialize WhisperModel with large-v2 model size and specified compute_type
        model_size = "large-v2" if model_dir is None else model_dir
        self.model = WhisperModel(model_size, device="cuda", compute_type=compute_type)

    def __call__(self, data: Dict) -> Dict[str, str]:
        audio_bytes = data["inputs"]
        audio_file = io.BytesIO(audio_bytes)

        # Transcribe audio file with a smaller beam size for faster inference
        # Note: Adjust beam_size based on desired accuracy vs speed trade-off
        beam_size = 1
        segments, info = self.model.transcribe(audio_file, beam_size=beam_size)

        # Aggregate transcribed text and remove any extra spaces
        text = " ".join(segment.text.strip() for segment in segments)
        text = re.sub(' +', ' ', text)

        language_code = info.language
        language_prob = info.language_probability

        result = {
            "text": text,
            "language": language_code,
            "language_probability": language_prob
        }
        
        return result