FunASR/funasr/models/qwen3_asr/model.py

import logging
import os
import time
from importlib.metadata import PackageNotFoundError
from importlib.metadata import requires as _package_requires
from importlib.metadata import version as _package_version
from typing import List, Optional, Union

import numpy as np
import torch
import torch.nn as nn

from funasr.register import tables


def _qwen_asr_transformers_specifier():
    try:
        from packaging.requirements import InvalidRequirement, Requirement
    except ImportError:
        return None

    requirements = _package_requires("qwen-asr") or []
    for requirement_text in requirements:
        try:
            requirement = Requirement(requirement_text)
        except InvalidRequirement:
            continue
        if requirement.name.lower() == "transformers":
            return requirement.specifier
    return None


def _qwen_asr_install_command(qwen_asr_version, transformers_specifier=None):
    if transformers_specifier:
        transformers_requirement = f"transformers{transformers_specifier}"
        return (
            f'pip install -U "qwen-asr=={qwen_asr_version}" '
            f'"{transformers_requirement}" accelerate'
        )
    return 'pip install -U "qwen-asr" transformers accelerate'


def _check_qwen3_asr_dependencies():
    try:
        qwen_asr_version = _package_version("qwen-asr")
    except PackageNotFoundError as e:
        raise ImportError(
            'qwen-asr package is required for Qwen3-ASR. Install with: pip install -U "qwen-asr"'
        ) from e

    try:
        transformers_version = _package_version("transformers")
    except PackageNotFoundError as e:
        raise ImportError(
            "transformers is required by qwen-asr. "
            f"Install with: {_qwen_asr_install_command(qwen_asr_version)}"
        ) from e

    transformers_specifier = _qwen_asr_transformers_specifier()
    if transformers_specifier:
        try:
            from packaging.version import InvalidVersion, Version

            Version(transformers_version)
            is_compatible = transformers_version in transformers_specifier
        except InvalidVersion:
            is_compatible = True

        if not is_compatible:
            install_command = _qwen_asr_install_command(qwen_asr_version, transformers_specifier)
            raise ImportError(
                "Qwen3-ASR dependency mismatch: "
                f"qwen-asr=={qwen_asr_version} requires transformers{transformers_specifier}, "
                f"but the active environment has transformers=={transformers_version}. "
                "This can trigger qwen_asr errors such as "
                "`AttributeError: 'Qwen3ASRConfig' object has no attribute 'thinker_config'`. "
                f"Run: {install_command}"
            )


# qwen-asr's validate_language() only accepts canonical full names ("Chinese", "English",
# ...), but FunASR documents short/ISO codes ("zh", "en", "auto") as valid language hints.
# Defined at module level so the lookup table is built once, not on every inference call.
_ISO_LANG_ALIASES = {
    "zh": "Chinese", "zh-cn": "Chinese", "zho": "Chinese", "cmn": "Chinese",
    "en": "English", "yue": "Cantonese", "ar": "Arabic", "de": "German",
    "fr": "French", "es": "Spanish", "pt": "Portuguese", "id": "Indonesian",
    "it": "Italian", "ko": "Korean", "ru": "Russian", "th": "Thai",
    "vi": "Vietnamese", "ja": "Japanese", "tr": "Turkish", "hi": "Hindi",
    "ms": "Malay", "nl": "Dutch", "sv": "Swedish", "da": "Danish",
    "fi": "Finnish", "pl": "Polish", "cs": "Czech", "fil": "Filipino",
    "fa": "Persian", "el": "Greek", "ro": "Romanian", "hu": "Hungarian",
    "mk": "Macedonian",
}


@tables.register("model_classes", "Qwen3ASR")
@tables.register("model_classes", "Qwen/Qwen3-ASR-1.7B")
@tables.register("model_classes", "Qwen/Qwen3-ASR-0.6B")
class Qwen3ASR(nn.Module):
    """Qwen3-ASR: Large Language Model based ASR supporting 52 languages.

    Wraps the qwen-asr package's Qwen3ASRModel for use within FunASR's AutoModel interface.
    Supports auto language detection, contextual recognition, and optional forced alignment
    for character-level timestamps.

    Requirements:
        pip install -U "qwen-asr==0.0.6" "transformers==4.57.6" accelerate

    Models:
        - Qwen/Qwen3-ASR-0.6B (lighter, ~4GB GPU memory)
        - Qwen/Qwen3-ASR-1.7B (more accurate, ~8GB GPU memory)
    """

    def __init__(self, **kwargs):
        """Initialize Qwen3ASR.
        
            Args:
                **kwargs: Additional keyword arguments.
            """
        super().__init__()
        model_path = kwargs.get("model_path", kwargs.get("model", "Qwen/Qwen3-ASR-1.7B"))
        device = kwargs.get("device", "cuda:0")
        dtype = kwargs.get("dtype", "bf16")
        hub = kwargs.get("hub", "ms")
        max_new_tokens = kwargs.get("max_new_tokens", 512)
        max_inference_batch_size = kwargs.get("max_inference_batch_size", 32)
        forced_aligner = kwargs.get("forced_aligner", None)
        forced_aligner_kwargs = kwargs.get("forced_aligner_kwargs", None)

        self._dtype_map = {"bf16": torch.bfloat16, "fp16": torch.float16, "fp32": torch.float32}
        self._device = device
        self._placeholder = nn.Parameter(torch.empty(0))

        model_path = self._resolve_model_path(model_path, hub, kwargs)
        self.model_path = model_path

        _check_qwen3_asr_dependencies()
        try:
            from qwen_asr import Qwen3ASRModel
        except ImportError as e:
            # Only catch if the package itself is missing, not if its dependencies are broken
            if "qwen_asr" in str(e):
                raise ImportError(
                    'qwen-asr package is required. Install with: pip install -U "qwen-asr"'
                ) from e
            raise e

        torch_dtype = self._dtype_map.get(dtype, torch.bfloat16)
        fa_kwargs = None
        if forced_aligner:
            fa_kwargs = forced_aligner_kwargs or {}
            fa_kwargs.setdefault("dtype", torch_dtype)
            fa_kwargs.setdefault("device_map", device)

        self.qwen3_asr_model = Qwen3ASRModel.from_pretrained(
            model_path,
            dtype=torch_dtype,
            device_map=device,
            forced_aligner=forced_aligner,
            forced_aligner_kwargs=fa_kwargs,
            max_inference_batch_size=max_inference_batch_size,
            max_new_tokens=max_new_tokens,
        )
        logging.info(f"Qwen3ASR model loaded from {model_path}")

    def _resolve_model_path(self, model_path, hub, kwargs):
        """Resolve model path: use local if exists, otherwise download from hub.

        Args:
            model_path (str): Model name or local path.
            hub (str): "ms" for ModelScope, "hf" for HuggingFace.
            kwargs (dict): Additional options (model_revision, etc.)

        Returns:
            str: Resolved local path to model files.
        """
        if os.path.exists(model_path):
            return model_path

        if hub in ("ms", "modelscope"):
            try:
                from modelscope.hub.snapshot_download import snapshot_download
                model_revision = kwargs.get("model_revision", "master")
                local_path = snapshot_download(model_path, revision=model_revision)
                logging.info(f"Downloaded from ModelScope: {model_path} -> {local_path}")
                return local_path
            except Exception as e:
                logging.warning(f"ModelScope download failed: {e}, falling back to HuggingFace path")

        return model_path

    def forward(self, **kwargs):
        """Forward pass for training.
        
            Args:
                **kwargs: Additional keyword arguments.
            """
        raise NotImplementedError("Qwen3ASR only supports inference mode")

    def inference(
        self,
        data_in,
        data_lengths=None,
        key: list = None,
        tokenizer=None,
        frontend=None,
        **kwargs,
    ):
        """Run Qwen3-ASR speech recognition.

        Args:
            data_in: Audio input. Accepts:
                - list of file paths/URLs
                - list of (numpy_array, sample_rate) tuples
                - single numpy array or torch Tensor
            data_lengths: Not used.
            key (list): Sample identifiers.
            tokenizer: Not used (Qwen3-ASR has internal tokenizer).
            frontend: Not used (Qwen3-ASR has internal audio processing).
            **kwargs: Runtime parameters:
                - language (str): Language hint (e.g. "Chinese", "English") or None for auto-detect.
                - return_time_stamps (bool): Return character-level timestamps (requires forced_aligner).
                - output_timestamp (bool): Same as return_time_stamps (for pipeline compatibility).
                - context (str): Context prompt for contextual recognition.

        Returns:
            tuple: (results, meta_data) where results is list of dicts:
                - "key" (str): Sample ID
                - "text" (str): Recognized text (with punctuation)
                - "language" (str): Detected language (if available)
                - "timestamp" (list): [[start_ms, end_ms], ...] (if timestamps enabled)
        """
        meta_data = {}
        time1 = time.perf_counter()

        language = kwargs.get("language", None)
        # Normalize FunASR's documented short/ISO codes (and "auto") to qwen-asr full names
        # so a hint like language="zh" doesn't raise "Unsupported language: Zh".
        if language is not None:
            _lk = str(language).strip().lower()
            if _lk in ("auto", "none", ""):
                language = None
            else:
                language = _ISO_LANG_ALIASES.get(_lk, language)
        return_time_stamps = kwargs.get("return_time_stamps", False) or kwargs.get("output_timestamp", False)
        context = kwargs.get("context", "")

        if isinstance(data_in, (list, tuple)):
            audio_inputs = []
            for item in data_in:
                if isinstance(item, np.ndarray):
                    audio_inputs.append((item.astype(np.float32), 16000))
                elif isinstance(item, torch.Tensor):
                    audio_inputs.append((item.cpu().numpy().astype(np.float32), 16000))
                else:
                    audio_inputs.append(item)
        elif isinstance(data_in, str):
            audio_inputs = [data_in]
        elif isinstance(data_in, torch.Tensor):
            audio_np = data_in.cpu().numpy().astype(np.float32)
            if audio_np.ndim == 1:
                audio_inputs = [(audio_np, 16000)]
            else:
                audio_inputs = [(audio_np[i], 16000) for i in range(audio_np.shape[0])]
        elif isinstance(data_in, np.ndarray):
            if data_in.ndim == 1:
                audio_inputs = [(data_in.astype(np.float32), 16000)]
            else:
                audio_inputs = [(data_in[i].astype(np.float32), 16000) for i in range(data_in.shape[0])]
        else:
            audio_inputs = [data_in]

        time2 = time.perf_counter()
        meta_data["load_data"] = f"{time2 - time1:0.3f}"

        # If timestamps requested but forced_aligner not configured, fall back to no timestamps
        if return_time_stamps and self.qwen3_asr_model.forced_aligner is None:
            logging.warning("return_time_stamps requires forced_aligner. Skipping timestamps. "
                           "Initialize with forced_aligner='Qwen/Qwen3-ForcedAligner-0.6B' to enable.")
            return_time_stamps = False

        results = self.qwen3_asr_model.transcribe(
            audio=audio_inputs,
            context=context,
            language=language,
            return_time_stamps=return_time_stamps,
        )

        time3 = time.perf_counter()
        meta_data["batch_data_time"] = time3 - time2

        output = []
        for i, r in enumerate(results):
            k = key[i] if key and i < len(key) else f"sample_{i}"
            result_dict = {"key": k, "text": r.text}
            if r.language:
                result_dict["language"] = r.language
            if return_time_stamps and r.time_stamps is not None:
                result_dict["timestamp"] = [
                    [int(ts.start_time), int(ts.end_time)]
                    for ts in r.time_stamps.items
                ]
            output.append(result_dict)

        return output, meta_data