# 语音识别示例
本示例展示如何定义Scalar UDF进行语音识别，将音频二进制数据通过Whisper模型转换为文本，体现多模态数据处理中UDF的使用能力。
#### 场景描述
在音频数据处理场景中，需要将存储在数据库表中的音频二进制数据通过语音识别模型（Whisper）转换为文本。通过Class形式的Scalar UDF，在__init__中加载Whisper模型，在__call__中对每段音频执行识别并返回文本结果。本示例从sample-files.com下载一段真实语音音频作为测试数据，写入数据库表后供UDF处理。
#### 约束与限制
- UDF运行时需通过pip安装transformers包，请确保运行时环境网络可用。
- Scalar UDF的__init__参数仅支持传入常量。
- 注册UDF前建议先调用delete_function清理同名函数，避免重复执行时报错。
 
#### 前提条件
- 已安装Aura DataFrame SDK。具体操作，请参见[Aura DataFrame SDK获取与安装](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_09_0026.html)。
- 已创建AuraJobV2端点，并获取端点名称。具体操作，请参见[创建Aura引擎端点](https://support.huaweicloud.com/usermanual-aidatalake/aidatalake_01_0049.html)。
- 已创建LakeFormation实例，并在该实例中创建Catalog及数据库。具体操作，请参见[创建LakeFormation实例并规划元数据](https://support.huaweicloud.com/qs-lakeformation/lakeformation_02_0001.html)。
- 已创建并行文件系统。具体操作，请参见[创建并行文件系统](https://support.huaweicloud.com/usermanual-obs/obs_13_0002.html)。
- 代码中涉及os.getenv()中的参数已配置到环境变量。
- 运行时环境要求Python版本为3.11。
 
#### 准备工作
本示例使用Whisper语音识别模型，需要准备模型文件、音频文件和Python依赖。
1. 安装Python依赖。
   ```
   pip install torch==2.6.0
   pip install transformers==5.3.0
   pip install av
   pip install scipy
   ```
   
2. 安装huggingface命令行工具。
   ```
   pip install huggingface-hub
   ```
   
3. 下载测试音频文件到示例脚本同级目录，重命名为test_audio.wav： 下载一段WAV格式的语音文件，建议10秒以内。
   
4. 下载Whisper base模型文件到示例脚本同级目录。
   ```
   hf download openai/whisper-base --local-dir ./whisper_base
   ```
   如果hf download命令无法使用，可通过浏览器手动下载：
   1. 在Hugging Face网站搜索openai/whisper-base，进入模型主页。
   
   2. 单击"Files and versions"标签页。
   
   3. 下载以下文件，保存到本地whisper_base目录下：
      - config.json
      
      - pytorch_model.bin
      
      - preprocessor_config.json
      
      - tokenizer_config.json
      
      - tokenizer.json
      
      - vocab.json
      
      - merges.txt
      
      - normalizer.json
      
      - special_tokens_map.json
      
      - added_tokens.json
      
      - generation_config.json
       
   
   4. 将whisper_base目录放置在与示例脚本同级目录下。
    
 
#### 示例
本示例定义一个基于Whisper模型的语音识别Scalar UDF，将测试音频写入数据库后，通过map_batches方法调用UDF将音频二进制数据转换为文本。
1. 定义Class UDF类SpeechRecognitionUDF，在__init__中通过transformers加载Whisper语音识别模型，在__call__中用av库解码音频二进制数据并重采样到16kHz后执行识别。
2. 连接Aura服务并设置UDF工作空间。
3. 读取测试音频文件，建表并插入音频二进制数据。
4. 删除已存在的同名UDF（支持重复执行）。
5. 注册Scalar UDF到目标数据库，指定依赖包和模型文件。
6. 加载audio_data数据集。
7. 通过map_batches方法调用UDF进行语音识别，生成hyp_text列。
8. 选择相关列并执行查询打印结果。
9. 关闭Aura连接。
```
import os
import ibis
import aura_frame
from aura_frame.multimodal import ai_lake
# 定义Scalar UDF：语音识别
class SpeechRecognitionUDF:
    def __init__(self, model_path: str) -> None:
        import torch
        from transformers import pipeline
        self._pipeline = pipeline(
            "automatic-speech-recognition",
            model=model_path,
            device=0 if torch.cuda.is_available() else -1,
        )
    def __call__(self, audios: bytes) -> str:
        import numpy as np
        import av
        import io
        from scipy.signal import resample_poly
        container = av.open(io.BytesIO(audios))
        stream = container.streams.audio[0]
        frames = []
        for frame in container.decode(stream):
            frames.append(frame.to_ndarray())
        if not frames:
            raise ValueError("Audio data is None.")
        audio_np = np.concatenate(frames, axis=1)
        if audio_np.ndim > 1:
            audio_np = audio_np.mean(axis=0)
        if audio_np.dtype == np.int16:
            audio_np = audio_np.astype(np.float32) / 32768.0
        elif audio_np.dtype == np.int32:
            audio_np = audio_np.astype(np.float32) / 2147483648.0
        elif audio_np.dtype != np.float32:
            audio_np = audio_np.astype(np.float32)
        # 重采样到 16kHz（Whisper 要求）
        target_sr = 16000
        if stream.rate != target_sr:
            audio_np = resample_poly(audio_np, target_sr, stream.rate)
        result = self._pipeline({"raw": audio_np, "sampling_rate": target_sr})
        return result["text"]
try:
    # 读取测试音频文件
    audio_path = "test_audio.wav"
    with open(audio_path, "rb") as f:
        audio_bytes = f.read()
    # 建表并插入音频数据
    table_name = "audio_data"
    table_schema = ibis.schema({"id": "int32", "audio_data": "binary"})
    con.sql(f'DROP TABLE IF EXISTS "{target_database}"."{table_name}";')
    con.create_table(
        table_name,
        schema=table_schema,
        database=target_database,
        table_format="orc",
        if_not_exists=True,
    )
    hex_str = audio_bytes.hex()
    con.sql(f'''
    INSERT INTO "{target_database}"."{table_name}" (id, audio_data) VALUES
        (1, '\\x{hex_str}'::bytea)
    ''')
    print("音频数据写入完成")
    # 删除已存在的同名UDF（支持重复执行）
    con.delete_function("speechrecognitionudf", database=target_database, if_it_exists=True)
    # 注册Scalar UDF：语音识别
    con.create_scalar_function(
        SpeechRecognitionUDF,
        database=target_database,
        packages=["transformers==5.3.0"],
        imports=["./whisper_base"],
    )
    # 加载音频数据集
    ds = con.load_dataset("audio_data", database=target_database)
    # 使用Scalar UDF进行语音识别
    udf = con.get_function("speechrecognitionudf", database=target_database)
    ds = ds.map_batches(
        fn=udf,
        on=[ds.audio_data],
        as_col="hyp_text",
        num_cpus=2,
        memory=4096,
        arch="x86",
        model_path="./whisper_base",
    )
    ds = ds.select_columns(ds.id, ds.audio_data, ds.hyp_text)
    df = ds.execute()
    print(df)
finally:
    con.close()
```
#### 相关文档
- [Session APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0135.html)
- [Dataset APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0132.html)
- [Table APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0133.html)
- [User-Defined Function APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0134.html)