# 语音识别示例
本案例介绍如何定义Vectorized Scalar UDF来进行语音识别、如何定义UDAF来进行聚合统计和可视化。
1. 从aura_frame中引入高阶类型（图片、音频、视频）： 
   ```
   from aura_frame.multimodal.types.image import Image, display_image
   from aura_frame.multimodal.types.audio import Audio
   from aura_frame.multimodal.types.vectorized import PandasVector
   ```
   
   
2. 定义Class Vectorized Scalar UDF来进行语音识别，这里使用到外部包transformers、torch： 
   ```
   class SpeechRecognitionUDF:
       def __init__(self, model_path: str) -> None:
           import os
           from transformers import pipeline
           import torch
           model_dir = os.path.abspath(model_path)
           self._asr = pipeline(
               "automatic-speech-recognition",
               model=model_dir,
               device=0 if torch.cuda.is_available() else -1,
           )
       def __call__(self, audios: bytes) -> str:
           import numpy as np
           import av
           import io
           container = av.open(io.BytesIO(audios))
           stream = container.streams.audio[0]
           frames = []
           for frame in container.decode(stream):
               frames.append(frame.to_ndarray())  # shape: (channels, samples)
           if not frames:
               raise ValueError("Audio data is None.")
           audio_np = np.concatenate(frames, axis=1)
           if audio_np.ndim > 1 and audio_np.shape[0] > 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)
           result = self._asr({"array": audio_np, "sampling_rate": stream.rate})
           return result["text"]
   ```
   
   
3. 定义UDAF来进行TF-IDF词频统计和词云图片生成，这里使用到外部包wordcloud、matplotlib： 
   ```
   class SpeakerWordsUDAF:
       def __init__(self, top_n: int = 40):
           from typing import List
           self._texts: List[str] = []
           self._top_n = top_n
           self._chapter_id = None
           # tiny built-in stopword list (extend as needed)
           self._stopwords = {
               "the", "a", "an", "and", "or", "but",
               "also", "about", "this", "that", "these", "those",
               "his", "her", "its", "their", "our", "your",
               "of", "to", "in", "on", "for", "at", "by", "with",
               "from", "as", "is", "are", "was", "were", "be", "been",
               "it", "he", "she", "they", "we", "you", "i",
               "not", "do", "did", "does", "had", "have", "has",
               "how", "what", "when", "where", "which", "who", "whom",
               "would", "could", "should", "will", "can", "may",
           }
       @property
       def aggregate_state(self):
           return {
               "texts": self._texts,
               "chapter_id": self._chapter_id,
           }
       def accumulate(self, chapter_id: int, hyp_text: str):
           if hyp_text:
               self._texts.append(hyp_text)
           if chapter_id:
               self._chapter_id = chapter_id
       def merge(self, other_state):
           if other_texts := other_state.get("texts"):
               self._texts.extend(other_texts)
           if other_chapterid := other_state.get("chapter_id"):
               if self._chapter_id:
                   # for one SpeakerWordsUDAF instance, it should process texts with same chapter_id
                   assert self._chapter_id == other_chapterid
               else:
                   # CN get chapter_id from DN
                   self._chapter_id = other_chapterid
       def _tokenize(self, text: str):
           # simple whitespace tokenizer + basic cleaning
           tokens = []
           for t in text.split():
               t = t.strip().lower()
               # drop short tokens and non-alpha tokens
               if len(t) < 3:
                   continue
               if not t.isalpha():
                   continue
               if t in self._stopwords:
                   continue
               tokens.append(t)
           return tokens
       def _generate_wordcloud(self, frequencies: dict) -> Image:
           from io import BytesIO
           import matplotlib.pyplot as plt
           from wordcloud import WordCloud
           try:
               # Create figure
               fig, ax = plt.subplots(figsize=(10, 5))
               # Generate WordCloud
               wc = WordCloud(width=800, height=400, background_color="white") \
                   .generate_from_frequencies(frequencies)
               # Draw
               ax.imshow(wc, interpolation='bilinear')
               ax.axis("off")
               ax.set_title(f"WordCloud for Chapter {self._chapter_id}", fontsize=18, pad=20)
               # Save to in-memory bytes
               buf = BytesIO()
               fig.savefig(buf, format="png", bbox_inches="tight")
               buf.seek(0)
               return Image(data=buf.getvalue(), format="PNG")
           except Exception as e:
               # You may log or re-raise here
               raise
           finally:
               # Cleanup: ALWAYS close the figure
               if fig is not None:
                   plt.close(fig)
       def finish(self) -> Image:
           import math
           if not self._texts:
               return None
           # 1) tokenize all docs
           docs = [self._tokenize(txt) for txt in self._texts]
           num_docs = len(docs)
           # 2) document frequency
           df = {}
           for tokens in docs:
               for tok in set(tokens):
                   df[tok] = df.get(tok, 0) + 1
           # 3) tf-idf per doc → aggregate
           tfidf_sums = {}
           tfidf_counts = {}
           for tokens in docs:
               if not tokens:
                   continue
               # term freq in this doc
               tf = {}
               for tok in tokens:
                   tf[tok] = tf.get(tok, 0) + 1
               doc_len = len(tokens)
               for tok, f in tf.items():
                   tf_val = f / doc_len
                   # idf: log(N/df) + 1
                   idf_val = math.log(num_docs / df[tok]) + 1.0
                   tfidf = tf_val * idf_val
                   tfidf_sums[tok] = tfidf_sums.get(tok, 0.0) + tfidf
                   tfidf_counts[tok] = tfidf_counts.get(tok, 0) + 1
           # 4) average tf-idf per term
           avg_tfidf = {
               tok: (tfidf_sums[tok] / tfidf_counts[tok])
               for tok in tfidf_sums
           }
           # 5) sort & top-N
           sorted_terms = sorted(
               avg_tfidf.items(),
               key=lambda kv: kv[1],
               reverse=True
           )[: self._top_n]
           out = {word: float(score) for word, score in sorted_terms}
           return self._generate_wordcloud(out)
   ```
   
   
4. 建立远程连接，需要用到外部包"huawei-aura-connectorapi"： 
   ```
   from contextlib import contextmanager
   from aura_frame.multimodal import ai_lake
   import os
   target_database = "xxxxx"
   @contextmanager
   def create_connect():
       con = ai_lake.connect(
           aura_endpoint=os.getenv("aura_endpoint"),
           aura_endpoint_name=os.getenv("aura_endpoint_name"),
           aura_workspace_id=os.getenv("aura_workspace_id"),
           lf_catalog_name=os.getenv("lf_catalog_name"),
           access_key=os.getenv("access_key"),
           secret_key=os.getenv("secret_key"),
           default_database=target_database,
           use_single_cn_mode=True
       )
       print(f"Establish ai_lake.connect: {con}")
       try:
           yield con
       finally:
           # release resource
           con.close()
   ```
   
   
5. 注册Scalar UDF、UDAF： 
   ```
   with create_connect() as con:
       con.set_function_staging_workspace(
           obs_directory_base=os.getenv("obs_directory_base"),
           obs_bucket_name=os.getenv("obs_bucket_name"),
           obs_server=os.getenv("obs_server"),
           access_key=os.getenv("access_key"),
           secret_key=os.getenv("secret_key"))
       print("finish set_function_staging_workspace")
       
       try:
           con.delete_function("SpeechRecognitionUDF", database=target_database)
       except Exception as e:
           print("UDF doesn't exists.")
       con.create_scalar_function(
           SpeechRecognitionUDF,
           database=target_database,
           packages=["transformers", "torch"],
           imports=["./facebook_wav2vec2_base_960h"]
       )
       try:
           con.delete_function("SpeakerWordsUDAF", database=target_database)
       except Exception as e:
           print("UDF doesn't exists.")
       con.create_agg_function(
           SpeakerWordsUDAF,
           database=target_database,
           packages=["wordcloud", "matplotlib"]
       )
   ```
   
   
6. 音频文件需要手动上传至"obs://xxxxx/xxxxx/"路径下（read_audios的path参数），根据音频文件类型的区别，根据需要在file_extension添加对应类型，然后以aura_frame API形式使用Scalar UDF, UDAF，得到最后的分组统计词云图片： 
   ```
   from aura_frame.multimodal.function import AggregateFnBuilder
   with create_connect() as con:
       # load dataset
       ds = con.read_audios("obs://xxxxx/xxxxx/", start=0, end=-1, sample_rate=16000, include_paths=False, file_extensions=["wav"])
       ds.show(limit=1)
       ds = ds.limit(1)
       # use Scalar UDF for ASR
       udf = con.get_function("SpeechRecognitionUDF", database=target_database)
       ds = ds.map_batches(
           fn=udf,
           on=[ds.audio["data"]],
           as_col="hyp_text",
           num_cpus=1,
           memory=1024,
           arch='x86',
           concurrency=1,
           model_path="./facebook_wav2vec2_base_960h"
       )
       ds.show(limit=1)
       ds = ds.add_column(id=1)
       ds.select_columns([ds.id, ds.hyp_text]).show(1)
       # use UDAF for summary
       udaf_builder = AggregateFnBuilder(
           fn=udaf,
           on=[ds.id, ds.hyp_text],
           as_col="tf_idf_image",
           num_cpus=1,
           memory=1024,
           arch='x86',
       )
       summary = ds.aggregate(
           udaf_builder,
           by=ds.id,
       )
       summary.show(limit=1)
       df = summary.execute()
       print(df)
       display_image(df["tf_idf_image"].tolist())
   ```