# 视频帧描述与摘要生成示例
本示例将介绍如何使用API实现以下功能：
- 使用UDTF（用户定义表函数）提取关键视频帧并生成对应的帧描述。
- 应用UDAF（用户定义聚合函数）将多个帧描述综合成连贯的视频摘要。
#### 环境准备
本示例需要以下Python包：
```
huawei-aura-frame
huawei-aura-connectorapi
torch
transformers
```
#### 操作步骤
1. 创建服务器连接。
   ```
   import os
   from aura_frame.multimodal import ai_lake
   from contextlib import contextmanager
   # Set the target database name
   target_database = "xxxxx"
   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()
   ```
   
2. 定义UDF生成视频帧描述。 本示例使用Hugging Face的[Salesforce/blip-image-captioning-base](https://huggingface.co/Salesforce/blip-image-captioning-base)模型作为图像到文本生成模型。
   由于数据库中的Python UDF目前不提供可靠的网络连接和下载功能，建议将原始模型下载到本地并通过OBS上传压缩文件（如.zip）。
   可以通过以下Python脚本在本地机器上下载Salesforce/blip-image-captioning-base：
   ```
   from huggingface_hub import snapshot_download
   local_dir = "blip-image-captioning-base"
   snapshot_download(
       repo_id="Salesforce/blip-image-captioning-base",
       local_dir=local_dir,
       local_dir_use_symlinks=False,
   )
   ```
   ```
   class VideoContentAnalyzer:
       def __init__(self, model_name="./blip-image-captioning-base"):
           import torch
           from transformers import BlipProcessor, BlipForConditionalGeneration
           self.device = "cuda" if torch.cuda.is_available() else "cpu"
           self.processor = BlipProcessor.from_pretrained(model_name)
           self.model = BlipForConditionalGeneration.from_pretrained(
               model_name,
               torch_dtype=torch.float16 if self.device == "cuda" else torch.float32
           ).to(self.device)
           self.model.eval()
       def __call__(self, frame: bytes, frame_index: int) -> str:
           import json
           analyse = self.analyze_frame(frame, frame_index)
           return json.dumps({"description": analyse})
       def analyze_frame(self, frame: bytes, frame_index: int):
           """Analyze the video frame content."""
           try:
               # Generate frame description.
               frame_description = self._generate_frame_description(frame)
               # Analyze content features.
               content_features = self._analyze_content_features(frame_description)
               analyse = {
                   'frame_index': frame_index,
                   'description': frame_description,
                   'content_type': content_features['content_type'],
                   'key_objects': content_features['key_objects'],
                   'scene_context': content_features['scene_context']
               }
           except Exception as e:
               # Current frame processing failed; return None.
               analyse = {}
           return analyse
       def _generate_frame_description(self, frame: bytes):
           import io
           from PIL import Image
           import torch
           pil_img = Image.open(io.BytesIO(frame)).convert("RGB")
           """Generate a single-frame description."""
           inputs = self.processor(images=pil_img, return_tensors="pt").to(self.device)
           with torch.no_grad():
               outputs = self.model.generate(
                   **inputs,
                   max_length=50,
                   num_beams=5,
                   early_stopping=True
               )
           description = self.processor.decode(outputs[0], skip_special_tokens=True)
           return description.strip()
       def _analyze_content_features(self, description):
           description_lower = description.lower()
           content_categories = {
               'educational': ['tutorial', 'lesson', 'explain', 'teach', 'learn', 'education'],
               'entertainment': ['funny', 'comedy', 'entertainment', 'show', 'performance'],
               'nature': ['outdoor', 'nature', 'landscape', 'mountain', 'forest', 'animal'],
               'sports': ['sport', 'game', 'player', 'team', 'match', 'competition'],
               'food': ['food', 'cooking', 'recipe', 'meal', 'restaurant', 'kitchen'],
               'technology': ['computer', 'tech', 'device', 'electronic', 'software'],
               'people': ['person', 'people', 'man', 'woman', 'child', 'group']
           }
           detected_categories = []
           for category, keywords in content_categories.items():
               if any(keyword in description_lower for keyword in keywords):
                   detected_categories.append(category)
           common_objects = [
               'person', 'people', 'man', 'woman', 'child', 'car', 'building',
               'tree', 'house', 'street', 'water', 'sky', 'food', 'animal'
           ]
           key_objects = [obj for obj in common_objects if obj in description_lower]
           if any(word in description_lower for word in ['indoors', 'inside', 'room']):
               scene_context = 'indoor'
           elif any(word in description_lower for word in ['outdoors', 'outside', 'park', 'street']):
               scene_context = 'outdoor'
           else:
               scene_context = 'unknown'
           return {
               'content_type': detected_categories[:2] if detected_categories else ['general'],
               'key_objects': key_objects[:5],
               'scene_context': scene_context        
           }
   ```
   
3. 定义UDAF从多个帧描述生成视频摘要。 以下示例使用Hugging Face的[google-t5/t5-small](https://huggingface.co/google-t5/t5-small)模型作为文本到文本生成模型。
   由于数据库中的Python UDF目前不提供可靠的网络连接和下载功能，建议将原始模型下载到本地并通过OBS上传压缩文件（如.zip）。
   ```
   from collections import Counter
   import typing
   class VideoSummaryGenerator:
       """Video Summary Generator"""
       def __init__(self, model):
           # Initialize the text summarization model.
           self.summarizer = pipeline(
               "text-generation",
               model=model,
               device=0 if torch.cuda.is_available() else -1
           )
       def generate_summary(self, frame_analyses):
           """Generate a video summary"""
           all_descriptions = [analysis['description'] for analysis in frame_analyses]
           if not all_descriptions:
               return "Unable to analyze the video content."
           # Analyze the characteristics of the video content.
           content_analysis = self._analyze_video_content(frame_analyses)
           # Generate a summary
           summary = self._generate_detailed_summary(all_descriptions, content_analysis)
           return summary
       def _analyze_video_content(self, frame_analyses):
           """Analyze the overall video content."""
           # all_descriptions = ' '.join([analysis['description'] for analysis in frame_analyses]).lower()
           # Extract the key information
           content_types = []
           all_objects = []
           for analysis in frame_analyses:
               content_types.extend(analysis['content_type'])
               all_objects.extend(analysis['key_objects'])
           # Calculate frequency
           common_content_types = [item for item, count in Counter(content_types).most_common(2)]
           common_objects = [item for item, count in Counter(all_objects).most_common(5)]
           return {
               'primary_categories': common_content_types,
               'dominant_objects': common_objects,
               'total_frames_analyzed': len(frame_analyses)
           }
       def _create_description_text(self, descriptions, content_analysis):
           """Create descriptive text."""
           desc_counter = Counter(descriptions)
           representative_descs = [desc for desc, count in desc_counter.most_common(3)]
           text = "Video Content Analysis:\n"
           text += f"Primary Categories: {', '.join(content_analysis['primary_categories'])}\n"
           text += f"Key Objects: {', '.join(content_analysis['dominant_objects'][:3])}\n"
           text += "Representative Scenes:\n"
           for i, desc in enumerate(representative_descs, 1):
               text += f"{i}. {desc}\n"
           return text
       def _generate_concise_summary(self, content_analysis):
           """Generate a concise summary"""
           primary_category = content_analysis['primary_categories'][0] if content_analysis['primary_categories'] else "General content"
           main_objects = ', '.join(content_analysis['dominant_objects'][:2])
           templates = [
               f"This is a {primary_category} video that includes elements like {main_objects}.",
               f"The video presents a {primary_category} scenario, with its focus on {main_objects}.",
               f"The content centers on {primary_category}, highlighting elements such as {main_objects}."
           ]
           import random
           return random.choice(templates)
       def _generate_detailed_summary(self, description_text, content_analysis):
           # Use the summary model to generate a more detailed description.
           try:
               summary = self.summarizer(
                   description_text,
                   max_length=150,
                   min_length=50,
                   do_sample=False
               )[0]['summary_text']
               return summary
           except Exception as e:
               # Failed to generate detailed information, switched to producing concise summary.
               return self._generate_concise_summary(content_analysis)
       def _generate_engaging_summary(self, content_analysis):
           """Generate an engaging summary"""
           primary_category = content_analysis['primary_categories'][0] if content_analysis['primary_categories'] else "Awesome"
           main_objects = '、'.join(content_analysis['dominant_objects'][:2])
           engaging_templates = [
               f" Don't miss out! This {primary_category} video takes you deep into the fascinating world of {main_objects}!",
               f" A spectacular showcase! A visual feast exploring the unique charm of {main_objects} in a {primary_category} setting!",
               f" Watch now! This video perfectly captures the {primary_category} moment of {main_objects}!"
           ]
           import random
           return random.choice(engaging_templates)
   class VideoSummarySystem:
       """Video Summary Generation System"""
       def __init__(self, model="./t5-small"):
           self.summary_generator = VideoSummaryGenerator(model)
           self.frame_descriptions = []
       def accumulate(self, description: str) -> None:
           import json
           self.frame_descriptions.append(json.loads(description))
       def finish(self) -> str:
           import json
           summary = self.summary_generator.generate_summary(self.frame_descriptions)
           result = {
               'video_summary': summary,
               'frames_analyzed': len(self.frame_descriptions),
               'content_categories': list(set(
                   cat for analysis in self.frame_descriptions
                   for cat in analysis['content_type']
               )),
               'key_objects': list(set(
                   obj for analysis in self.frame_descriptions
                   for obj in analysis['key_objects']
               ))[:8],
               'processing_details': {
                   'total_frames': len(self.frame_descriptions),
                   'successful_analyses': len(self.frame_descriptions)
               }
           }
           return json.dumps(result)
       @property
       def aggregate_state(self) -> typing.Dict[str, typing.Any]:
           return {
               "frame_descriptions": self.frame_descriptions,
           }
       def merge(self, other_state: typing.Dict[str, typing.Any]) -> None:
           self.frame_descriptions += other_state.get("frame_descriptions")
   ```
   
4. 执行创建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("VideoContentAnalyzer", database=target_database)
       except Exception as e:
           print("UDF doesn't exists.")
       con.create_scalar_function(
           VideoContentAnalyzer,
           database=target_database,
           packages=["transformers", "torch", "pillow"],
           imports=["./blip-image-captioning-base"]
       )
       try:
           con.delete_function("VideoSummarySystem", database=target_database)
       except Exception as e:
           print("UDF doesn't exists.")
       con.create_agg_function(
           VideoSummarySystem,
           database=target_database,
           packages=["transformers", "torch"],
           imports=["./t5-small"]
       )
   ```
   
5. 以aura_frame API形式使用Scalar UDF, UDAF，得到最后视频的摘要内容。
   这里需要先将需要处理的视频文件上传到obs://xxxxx/xxxxx/（read_video_frames传入path参数）某一路径下，然后通过dataframe进行处理。如果视频不是mp4文件，还需根据文件类型在read_video_frames中的file_extensions中添加对应类型名称。
   ```
   from aura_frame.multimodal.function import AggregateFnBuilder
   with create_connect() as con:
       ds = con.read_video_frames("obs://xxxxx/xxxxx/",image_height=800, image_width=400, image_format='jpg', include_paths=True, include_timestamp=True, file_extensions=["mp4"])
       ds.show(limit=1)
       # use UDF
       udf = con.get_function("VideoContentAnalyzer", database=target_database)
       ds = ds.map_batches(
           fn=udf,
           on=[ds.image["data"], ds.frame_index],
           as_col="descriptions",
           num_cpus=1,
           memory=1024,
           arch='x86',
           concurrency=1,
           model_name="./blip-image-captioning-base"
       )
       ds.select_columns([ds.frame_index, ds.descriptions]).show(1)
       # use UDAF for summary
       udaf = con.get_function("VideoSummarySystem", database=target_database)
       agg_builder = AggregateFnBuilder(
           fn=udaf,
           on=[ds.descriptions],
           as_col="describe",
           num_cpus=1,
           memory=1024,
           arch='x86'
       )
       summary = ds.aggregate(agg_builder).execute()
       for row in summary.itertuples():    
           print(row[1])
   ```