simulation.py

# simulation.py
# 仿真与视频相关
import os
import time
import uuid
import cv2
import numpy as np
from typing import List
import gradio as gr
from backend_api import get_task_status
from oss_utils import list_oss_files, download_oss_file, get_user_tmp_dir, test_oss_access, clean_oss_result_path

def stream_simulation_results(result_folder: str, task_id: str, request: gr.Request, fps: int = 6):
    """

    流式输出仿真结果，从OSS读取图片

    

    参数:

        result_folder: OSS上包含生成图片的文件夹路径

        task_id: 后端任务ID用于状态查询

        request: Gradio请求对象

        fps: 输出视频的帧率

        

    生成:

        生成的视频文件路径 (分段输出)

    """
    # 初始化变量
    image_folder = os.path.join(result_folder, "images")
    frame_buffer: List[np.ndarray] = []
    frames_per_segment = fps * 2  # 每2秒输出一段
    processed_files = set()
    width, height = 0, 0
    last_status_check = 0
    status_check_interval = 5  # 每5秒检查一次后端状态
    max_time = 240

    # 创建临时目录存储下载的图片
    user_dir = get_user_tmp_dir(request.session_hash)
    local_image_dir = os.path.join(user_dir, task_id, "images")
    os.makedirs(local_image_dir, exist_ok=True)

    while max_time > 0:
        max_time -= 1
        current_time = time.time()
        
        # 定期检查后端状态
        if current_time - last_status_check > status_check_interval:
            status = get_task_status(task_id)
            print(f"Session {request.session_hash}, status: {status}")
            if status.get("status") == "completed":
                # 确保处理完所有已生成的图片
                process_remaining_oss_images(image_folder, local_image_dir, processed_files, frame_buffer)
                if frame_buffer:
                    yield create_video_segment(frame_buffer, fps, width, height, request)
                break
            elif status.get("status") == "failed":
                raise gr.Error(f"任务执行失败: {status.get('result', '未知错误')}")
            elif status.get("status") == "terminated":
                break
            last_status_check = current_time

        # 从OSS获取文件列表
        try:
            oss_files = list_oss_files(image_folder)
            new_files = [f for f in oss_files if f not in processed_files]
            has_new_frames = False
            
            for oss_path in new_files:
                try:
                    # 下载文件到本地
                    filename = os.path.basename(oss_path)
                    local_path = os.path.join(local_image_dir, filename)
                    download_oss_file(oss_path, local_path)
                    
                    # 读取图片
                    frame = cv2.imread(local_path)
                    if frame is not None:
                        if width == 0:  # 第一次获取图像尺寸
                            height, width = frame.shape[:2]
                        
                        frame_buffer.append(frame)
                        processed_files.add(oss_path)
                        has_new_frames = True
                except Exception as e:
                    print(f"Error processing {oss_path}: {e}")

            # 如果有新帧且积累够指定帧数，输出视频片段
            if has_new_frames and len(frame_buffer) >= frames_per_segment:
                segment_frames = frame_buffer[:frames_per_segment]
                frame_buffer = frame_buffer[frames_per_segment:]
                yield create_video_segment(segment_frames, fps, width, height, request)

        except Exception as e:
            print(f"Error accessing OSS: {e}")

        time.sleep(1)  # 避免过于频繁检查
    
    if max_time <= 0:
        raise gr.Error("timeout 240s")

def create_video_segment(frames: List[np.ndarray], fps: int, width: int, height: int, request: gr.Request) -> str:
    """创建视频片段"""
    user_dir = get_user_tmp_dir(request.session_hash)
    video_chunk_dir = os.path.join(user_dir, "video_chunks")
    os.makedirs(video_chunk_dir, exist_ok=True)
    
    segment_name = os.path.join(video_chunk_dir, f"output_{uuid.uuid4()}.mp4")
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(segment_name, fourcc, fps, (width, height))
    
    for frame in frames:
        out.write(frame)
    out.release()
    
    return segment_name

def process_remaining_oss_images(oss_folder: str, local_dir: str, processed_files: set, frame_buffer: List[np.ndarray]):
    """处理OSS上剩余的图片"""
    try:
        oss_files = list_oss_files(oss_folder)
        new_files = [f for f in oss_files if f not in processed_files]
        
        for oss_path in new_files:
            try:
                # 下载文件到本地
                filename = os.path.basename(oss_path)
                local_path = os.path.join(local_dir, filename)
                download_oss_file(oss_path, local_path)
                
                # 读取图片
                frame = cv2.imread(local_path)
                if frame is not None:
                    frame_buffer.append(frame)
                    processed_files.add(oss_path)
            except Exception as e:
                print(f"Error processing remaining {oss_path}: {e}")
    except Exception as e:
        print(f"Error accessing OSS for remaining files: {e}")

def convert_to_h264(video_path):
    """

    将视频转换为 H.264 编码的 MP4 格式

    生成新文件路径在原路径基础上添加 _h264 后缀

    """
    import shutil
    import subprocess
    
    base, ext = os.path.splitext(video_path)
    video_path_h264 = f"{base}_h264.mp4"
    
    # 查找ffmpeg
    ffmpeg_bin = shutil.which("ffmpeg")
    if ffmpeg_bin is None:
        # 尝试常见的安装路径
        possible_paths = [
            "/root/anaconda3/envs/gradio/bin/ffmpeg",
            "/usr/bin/ffmpeg",
            "/usr/local/bin/ffmpeg"
        ]
        for path in possible_paths:
            if os.path.exists(path):
                ffmpeg_bin = path
                break
    
    if ffmpeg_bin is None:
        raise RuntimeError("❌ 找不到 ffmpeg，请确保其已安装并在 PATH 中")
    
    ffmpeg_cmd = [
        ffmpeg_bin,
        "-i", video_path,
        "-c:v", "libx264",
        "-preset", "slow",
        "-crf", "23",
        "-c:a", "aac",
        "-movflags", "+faststart",
        video_path_h264
    ]
    
    try:
        result = subprocess.run(ffmpeg_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        if not os.path.exists(video_path_h264):
            raise FileNotFoundError(f"⚠️ H.264 文件未生成: {video_path_h264}")
        return video_path_h264
    except subprocess.CalledProcessError as e:
        raise gr.Error(f"FFmpeg 转换失败: {e.stderr}")
    except Exception as e:
        raise gr.Error(f"转换过程中发生错误: {str(e)}")

def create_final_video_from_oss_images(result_folder: str, task_id: str, request: gr.Request, fps: int = 6) -> str:
    """

    从OSS上的所有图片拼接成最终完整视频

    

    参数:

        result_folder: OSS上的结果文件夹路径

        task_id: 任务ID

        request: Gradio请求对象

        fps: 视频帧率 (6帧每秒)

        

    返回:

        最终视频文件路径

    """
    # 使用统一的路径清理函数
    cleaned_result_folder = clean_oss_result_path(result_folder, task_id)
    
    # 获取图片文件夹路径
    image_folder = os.path.join(cleaned_result_folder, "images").replace('\\', '/')
    user_dir = get_user_tmp_dir(request.session_hash)
    local_image_dir = os.path.join(user_dir, task_id, "final_images")
    os.makedirs(local_image_dir, exist_ok=True)
    
    # 添加调试信息
    print(f"🔍 DEBUG: original result_folder = {result_folder}")
    print(f"🔍 DEBUG: cleaned_result_folder = {cleaned_result_folder}")
    print(f"🔍 DEBUG: task_id = {task_id}")
    print(f"🔍 DEBUG: image_folder = {image_folder}")
    print(f"🔍 DEBUG: user_dir = {user_dir}")
    print(f"🔍 DEBUG: local_image_dir = {local_image_dir}")
    
    try:
        # 首先测试OSS访问
        print(f"🔍 DEBUG: Testing OSS access for task {task_id}")
        test_oss_access(task_id)
        
        # 1. 获取OSS上的所有图片文件
        print(f"🔍 DEBUG: Listing files in OSS folder: {image_folder}")
        oss_files = list_oss_files(image_folder)
        print(f"🔍 DEBUG: Total files found in OSS: {len(oss_files)}")
        print(f"🔍 DEBUG: First 10 files: {oss_files[:10]}")
        
        image_files = [f for f in oss_files if f.lower().endswith(('.png', '.jpg', '.jpeg'))]
        print(f"🔍 DEBUG: Image files found: {len(image_files)}")
        print(f"🔍 DEBUG: First 5 image files: {image_files[:5]}")
        
        if not image_files:
            # 如果没有找到图片，尝试直接在cleaned_result_folder中查找
            print(f"🔍 DEBUG: No images in {image_folder}, trying {cleaned_result_folder}")
            oss_files_direct = list_oss_files(cleaned_result_folder)
            print(f"🔍 DEBUG: Files in result_folder: {len(oss_files_direct)}")
            print(f"🔍 DEBUG: Direct files: {oss_files_direct[:10]}")
            
            # 查找所有子目录中的图片
            all_image_files = [f for f in oss_files_direct if f.lower().endswith(('.png', '.jpg', '.jpeg'))]
            print(f"🔍 DEBUG: All image files in result_folder: {len(all_image_files)}")
            
            # 尝试不同的路径组合
            alternative_paths = [
                f"gradio_demo/tasks/{task_id}/images",
                f"gradio_demo/tasks/{task_id}",
                f"tasks/{task_id}/images", 
                f"tasks/{task_id}",
                # 根据您确认的路径格式添加
                f"gradio_demo/tasks/{task_id}/image",  # 可能是单数形式
                f"gradio_demo/tasks/{task_id}/imgs",   # 可能是缩写
            ]
            
            for alt_path in alternative_paths:
                print(f"🔍 DEBUG: Trying alternative path: {alt_path}")
                alt_files = list_oss_files(alt_path)
                alt_images = [f for f in alt_files if f.lower().endswith(('.png', '.jpg', '.jpeg'))]
                print(f"🔍 DEBUG: Found {len(alt_images)} images in {alt_path}")
                if alt_images:
                    all_image_files = alt_images
                    print(f"🔍 DEBUG: Using images from alternative path: {alt_path}")
                    break
            
            if all_image_files:
                image_files = all_image_files
                print(f"🔍 DEBUG: Using image files from alternative search: {len(image_files)}")
            else:
                raise gr.Error("No images found in OSS for final video creation")
        
        # 2. 按文件名排序确保时间顺序正确
        image_files.sort(key=lambda x: os.path.splitext(os.path.basename(x))[0])
        print(f"🔍 DEBUG: Sorted image files (first 5): {[os.path.basename(f) for f in image_files[:5]]}")
        
        # 3. 下载所有图片到本地
        frames = []
        width, height = 0, 0
        
        print(f"🔍 DEBUG: Starting to download {len(image_files)} images...")
        for i, oss_path in enumerate(image_files):
            try:
                filename = os.path.basename(oss_path)
                local_path = os.path.join(local_image_dir, filename)
                download_oss_file(oss_path, local_path)
                
                # 读取图片
                frame = cv2.imread(local_path)
                if frame is not None:
                    if width == 0:  # 第一次获取图像尺寸
                        height, width = frame.shape[:2]
                        print(f"🔍 DEBUG: Image dimensions: {width}x{height}")
                    frames.append(frame)
                    
                    if (i + 1) % 10 == 0:  # 每10张图片输出一次进度
                        print(f"🔍 DEBUG: Downloaded {i + 1}/{len(image_files)} images")
                        
            except Exception as e:
                print(f"❌ Error processing {oss_path}: {e}")
                continue
        
        print(f"🔍 DEBUG: Successfully loaded {len(frames)} frames")
        
        if not frames:
            raise gr.Error("No valid images could be processed for final video")
        
        # 4. 创建最终视频
        final_video_dir = os.path.join(user_dir, task_id, "final")
        os.makedirs(final_video_dir, exist_ok=True)
        final_video_path = os.path.join(final_video_dir, "final_video.mp4")
        
        print(f"🔍 DEBUG: Creating video at: {final_video_path}")
        print(f"🔍 DEBUG: Video parameters: {len(frames)} frames, {fps} fps, {width}x{height}")
        
        # 使用OpenCV创建视频
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        out = cv2.VideoWriter(final_video_path, fourcc, fps, (width, height))
        
        for frame in frames:
            out.write(frame)
        out.release()
        
        # 5. 转换为H.264格式
        print(f"🔍 DEBUG: Converting to H.264...")
        h264_video_path = convert_to_h264(final_video_path)
        
        print(f"✅ Final video created: {h264_video_path} with {len(frames)} frames at {fps} fps")
        return h264_video_path
        
    except Exception as e:
        print(f"❌ Error creating final video from OSS images: {e}")
        print(f"❌ Exception type: {type(e).__name__}")
        import traceback
        print(f"❌ Traceback: {traceback.format_exc()}")
        raise gr.Error(f"Failed to create final video: {str(e)}")