change to oss

OSS_MIGRATION_SUMMARY.md

@@ -0,0 +1,79 @@
+# OSS 集成修改总结
+
+## 修改内容
+
+### 1. 新增文件
+
+- **oss_utils.py**: 新建的 OSS 工具模块，包含：
+  - OSS 客户端初始化
+  - 文件列表获取 (`list_oss_files`)
+  - 文件下载 (`download_oss_file`) 
+  - 文件存在检查 (`oss_file_exists`)
+  - 临时目录管理 (`get_user_tmp_dir`, `cleanup_user_tmp_dir`)
+
+- **.env.example**: OSS 环境变量配置示例
+
+### 2. 修改的文件
+
+#### simulation.py
+- **导入 OSS 模块**: 添加了 `oss_utils` 的导入
+- **重写 `stream_simulation_results` 函数**:
+  - 支持从 OSS 读取图像文件
+  - 使用用户会话级别的临时目录
+  - 改进的错误处理和日志记录
+- **重写 `create_video_segment` 函数**:
+  - 使用用户特定的临时目录
+  - 更好的目录管理
+- **新增 `process_remaining_oss_images` 函数**:
+  - 处理 OSS 上剩余的图像文件
+- **改进 `convert_to_h264` 函数**:
+  - 更好的 ffmpeg 路径查找
+  - 改进的错误处理
+
+#### app.py  
+- **导入 OSS 模块**: 添加了 `oss_utils` 相关导入
+- **重写 `run_simulation` 函数**:
+  - 不再检查本地目录是否存在
+  - 从 OSS 下载最终视频文件
+  - 使用临时目录管理
+- **更新 `cleanup_session` 函数**:
+  - 添加了用户临时目录清理
+
+#### requirements.txt
+- **添加 OSS 依赖**: `oss2>=2.15.0`
+
+#### README.md
+- **完全重写**: 添加了 OSS 配置说明和使用指南
+
+## 主要变化
+
+### 1. 数据源变更
+- **之前**: 从本地文件系统读取图像和视频
+- **现在**: 从阿里云 OSS 读取数据
+
+### 2. 临时文件管理
+- **之前**: 使用固定的系统目录
+- **现在**: 为每个用户会话创建独立的临时目录
+
+### 3. 错误处理
+- **之前**: 基本的错误处理
+- **现在**: 更全面的错误处理和日志记录
+
+### 4. 文件下载
+- **之前**: 直接读取本地文件
+- **现在**: 从 OSS 流式下载文件到临时目录
+
+## 配置要求
+
+使用此修改后的代码需要：
+
+1. 安装 `oss2` Python 包
+2. 配置以下环境变量：
+   - `OSS_ACCESS_KEY_ID`
+   - `OSS_ACCESS_KEY_SECRET` 
+   - `OSS_ENDPOINT`
+   - `OSS_BUCKET_NAME`
+
+## 向后兼容性
+
+这些修改保持了 API 的兼容性，但改变了数据源。如果需要保持本地文件系统的支持，可以在 `oss_utils.py` 中添加回退逻辑。

README.md

@@ -1,12 +1,72 @@
----
-title: InternNav Eval Demo
-emoji: 👁
-colorFrom: yellow
-colorTo: gray
-sdk: gradio
-sdk_version: 5.38.2
-app_file: app.py
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# InternNav 评估演示
+
+这是一个基于 Gradio 的 InternNav 模型推理演示应用，支持从阿里云 OSS 读取视频数据。
+
+## 功能特性
+
+- 🤖 支持多种导航模型 (rdp, cma)
+- 🎯 支持多种模式 (vlnPE, vlnCE)
+- 🎬 实时流式视频输出
+- ☁️ 从阿里云 OSS 读取数据
+- � 用户访问日志记录
+- 🔒 IP 频率限制保护
+
+## 环境配置
+
+### 1. 安装依赖
+
+```bash
+pip install -r requirements.txt
+```
+
+### 2. 配置环境变量
+
+复制 `.env.example` 为 `.env` 并填入您的 OSS 配置：
+
+```bash
+cp .env.example .env
+```
+
+编辑 `.env` 文件：
+
+```env
+OSS_ACCESS_KEY_ID=your_access_key_id
+OSS_ACCESS_KEY_SECRET=your_access_key_secret
+OSS_ENDPOINT=your_oss_endpoint
+OSS_BUCKET_NAME=your_bucket_name
+BACKEND_URL=http://47.95.6.204:51001
+```
+
+### 3. 运行应用
+
+```bash
+python app.py
+```
+
+## 文件结构
+
+```
+├── app.py              # 主应用入口
+├── config.py           # 配置文件
+├── backend_api.py      # 后端 API 交互
+├── simulation.py       # 仿真和视频处理 (支持 OSS)
+├── oss_utils.py        # OSS 工具函数
+├── logging_utils.py    # 日志工具
+├── ui_components.py    # UI 组件
+├── requirements.txt    # Python 依赖
+├── assets/            # 静态资源
+└── tmp/               # 临时文件目录
+```
+
+## OSS 集成
+
+应用现在完全支持从阿里云 OSS 读取数据：
+
+- 图像文件从 OSS 流式下载
+- 视频文件从 OSS 下载到本地临时目录
+- 自动清理用户会话的临时文件
+- 支持断点续传和错误恢复
+
+## 配置参考
+
+查看 Hugging Face Spaces 配置文档：https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -6,6 +6,7 @@ from backend_api import submit_to_backend, get_task_status, get_task_result
 from logging_utils import log_access, log_submission, is_request_allowed
 from simulation import stream_simulation_results, convert_to_h264
 from ui_components import update_history_display, update_scene_display, update_log_display, get_scene_instruction
+from oss_utils import download_oss_file, get_user_tmp_dir, cleanup_user_tmp_dir, oss_file_exists
 import os
 from datetime import datetime
 
@@ -16,40 +17,62 @@ def run_simulation(scene, model, mode, prompt, history, request: gr.Request):
     scene_desc = SCENE_CONFIGS.get(scene, {}).get("description", scene)
     user_ip = request.client.host if request else "unknown"
     session_id = request.session_hash
+    
     if not is_request_allowed(user_ip):
         log_submission(scene, prompt, model, user_ip, "IP blocked temporarily")
         raise gr.Error("Too many requests from this IP. Please wait and try again one minute later.")
-    # 传递model和mode给后端
-    #submission_result = submit_to_backend(scene, prompt, user=model)  # 可根据后端接口调整
+    
+    # 提交任务到后端
     submission_result = submit_to_backend(scene, prompt, mode, model, user_ip)
     if submission_result.get("status") != "pending":
         log_submission(scene, prompt, model, user_ip, "Submission failed")
         raise gr.Error(f"Submission failed: {submission_result.get('message', 'unknown issue')}")
+    
     try:
         task_id = submission_result["task_id"]
         SESSION_TASKS[session_id] = task_id
         gr.Info(f"Simulation started, task_id: {task_id}")
+        
         import time
         time.sleep(5)
         status = get_task_status(task_id)
-        result_folder = status.get("result", "")
+        # OSS上的结果文件夹路径，不再检查本地路径是否存在
+        result_folder = status.get("result", f"gradio_demo/tasks/{task_id}")
+        
     except Exception as e:
         log_submission(scene, prompt, model, user_ip, str(e))
         raise gr.Error(f"error occurred when parsing submission result from backend: {str(e)}")
-    if not os.path.exists(result_folder):
-        log_submission(scene, prompt, model, user_ip, "Result folder provided by backend doesn't exist")
-        raise gr.Error(f"Result folder provided by backend doesn't exist: <PATH>{result_folder}")
+
+    # 流式输出视频片段（从OSS读取）
     try:
-        for video_path in stream_simulation_results(result_folder, task_id):
+        for video_path in stream_simulation_results(result_folder, task_id, request):
             if video_path:
                 yield video_path, history
     except Exception as e:
         log_submission(scene, prompt, model, user_ip, str(e))
         raise gr.Error(f"流式输出过程中出错: {str(e)}")
+
+    # 获取最终任务状态
     status = get_task_status(task_id)
     if status.get("status") == "completed":
-        video_path = os.path.join(status.get("result"), "output.mp4")
-        video_path = convert_to_h264(video_path)
+        # 从OSS下载最终视频文件
+        oss_video_path = os.path.join(result_folder, "output.mp4")  # 或者根据实际的视频文件名调整
+        user_dir = get_user_tmp_dir(session_id)
+        local_video_path = os.path.join(user_dir, task_id, "output.mp4")
+        
+        try:
+            # 尝试下载视频文件
+            if oss_file_exists(oss_video_path):
+                download_oss_file(oss_video_path, local_video_path)
+                video_path = convert_to_h264(local_video_path)
+            else:
+                # 如果OSS上没有最终视频，使用最后一个片段
+                gr.Info("Final video not found in OSS, using last segment")
+                video_path = None
+                
+        except Exception as e:
+            print(f"Error downloading final video: {e}")
+            video_path = None
         new_entry = {
             "timestamp": timestamp,
             "scene": scene,
@@ -70,11 +93,8 @@ def run_simulation(scene, model, mode, prompt, history, request: gr.Request):
         yield None, history
     elif status.get("status") == "terminated":
         log_submission(scene, prompt, model, user_ip, "terminated")
-        video_path = os.path.join(result_folder, "output.mp4")
-        if os.path.exists(video_path):
-            return f"⚠️ 任务 {task_id} 被终止，已生成部分结果", video_path, history
-        else:
-            return f"⚠️ 任务 {task_id} 被终止，未生成结果", None, history
+        # 对于终止的任务，不再检查本地文件
+        yield None, history
     else:
         log_submission(scene, prompt, model, user_ip, "missing task's status from backend")
         raise gr.Error("missing task's status from backend")
@@ -90,6 +110,9 @@ def cleanup_session(request: gr.Request):
             requests.post(f"{BACKEND_URL}/predict/terminate/{task_id}", timeout=3)
         except Exception:
             pass
+    
+    # 清理用户临时目录
+    cleanup_user_tmp_dir(session_id)
 
 def record_access(request: gr.Request):
     user_ip = request.client.host if request else "unknown"

config.py

@@ -2,7 +2,7 @@
 # 配置相关：API、场景等
 import os
 
-BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:8000")
+BACKEND_URL = os.getenv("BACKEND_URL", "47.95.6.204:51001")
 API_ENDPOINTS = {
     "submit_task": f"{BACKEND_URL}/predict/video",
     "query_status": f"{BACKEND_URL}/predict/task",

oss_utils.py

@@ -0,0 +1,64 @@
+# oss_utils.py
+# OSS相关工具函数
+import os
+import oss2
+from typing import List
+import shutil
+
+# OSS配置
+OSS_CONFIG = {
+    "access_key_id": os.getenv("OSS_ACCESS_KEY_ID"),
+    "access_key_secret": os.getenv("OSS_ACCESS_KEY_SECRET"),
+    "endpoint": os.getenv("OSS_ENDPOINT"),
+    "bucket_name": os.getenv("OSS_BUCKET_NAME")
+}
+
+# 初始化OSS客户端
+auth = oss2.Auth(OSS_CONFIG["access_key_id"], OSS_CONFIG["access_key_secret"])
+bucket = oss2.Bucket(auth, OSS_CONFIG["endpoint"], OSS_CONFIG["bucket_name"])
+
+# 临时文件根目录
+TMP_ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tmp")
+os.makedirs(TMP_ROOT, exist_ok=True)
+
+def list_oss_files(folder_path: str) -> List[str]:
+    """列出OSS文件夹中的所有文件"""
+    files = []
+    try:
+        for obj in oss2.ObjectIterator(bucket, prefix=folder_path):
+            if not obj.key.endswith('/'):  # 排除目录本身
+                files.append(obj.key)
+        return sorted(files, key=lambda x: os.path.splitext(x)[0]) 
+    except Exception as e:
+        print(f"Error listing OSS files: {str(e)}")
+        return []
+
+def download_oss_file(oss_path: str, local_path: str):
+    """从OSS下载文件到本地"""
+    try:
+        # 确保本地目录存在
+        os.makedirs(os.path.dirname(local_path), exist_ok=True)
+        bucket.get_object_to_file(oss_path, local_path)
+    except Exception as e:
+        print(f"Error downloading file {oss_path}: {str(e)}")
+        raise
+
+def oss_file_exists(oss_path: str) -> bool:
+    """检查OSS文件是否存在"""
+    try:
+        return bucket.object_exists(oss_path)
+    except Exception as e:
+        print(f"Error checking if file exists in OSS: {str(e)}")
+        return False
+
+def get_user_tmp_dir(session_hash: str) -> str:
+    """获取用户临时目录"""
+    user_dir = os.path.join(TMP_ROOT, str(session_hash))
+    os.makedirs(user_dir, exist_ok=True)
+    return user_dir
+
+def cleanup_user_tmp_dir(session_hash: str):
+    """清理用户临时目录"""
+    user_dir = os.path.join(TMP_ROOT, str(session_hash))
+    if os.path.exists(user_dir):
+        shutil.rmtree(user_dir)

requirements.txt

@@ -2,3 +2,4 @@ gradio>=4.0.0
 requests>=2.28.0
 opencv-python>=4.6.0
 numpy>=1.21.0
+oss2>=2.15.0

simulation.py

@@ -8,46 +8,94 @@ 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
 
-def stream_simulation_results(result_folder: str, task_id: str, fps: int = 6):
-    result_folder = os.path.join(result_folder, "images")
-    os.makedirs(result_folder, exist_ok=True)
+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
+    frames_per_segment = fps * 2  # 每2秒输出一段
     processed_files = set()
     width, height = 0, 0
     last_status_check = 0
-    status_check_interval = 5
+    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_images(result_folder, processed_files, frame_buffer)
+                # 确保处理完所有已生成的图片
+                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)
+                    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
-        current_files = sorted(
-            [f for f in os.listdir(result_folder) if f.lower().endswith(('.png', '.jpg', '.jpeg'))],
-            key=lambda x: os.path.splitext(x)[0]
-        )
-        new_files = [f for f in current_files if f not in processed_files]
-        has_new_frames = False
-        for filename in new_files:
-            try:
-                img_path = os.path.join(result_folder, filename)
-                frame = cv2.imread(img_path)
-                if frame is not None:
-                    if width == 0:
-                        height, width = frame.shape[:2]
-                    frame_buffer.append(frame)
+
+        # 从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")
                     processed_files.add(filename)
                     has_new_frames = True
             except Exception:
@@ -60,41 +108,73 @@ def stream_simulation_results(result_folder: str, task_id: str, fps: int = 6):
     if max_time <= 0:
         raise gr.Error("timeout 240s")
 
-def create_video_segment(frames: List[np.ndarray], fps: int, width: int, height: int) -> str:
-    os.makedirs("/opt/gradio_demo/tasks/video_chunk", exist_ok=True)
-    segment_name = f"/opt/gradio_demo/tasks/video_chunk/output_{uuid.uuid4()}.mp4"
+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_images(result_folder: str, processed_files: set, frame_buffer: List[np.ndarray]):
-    current_files = sorted(
-        [f for f in os.listdir(result_folder) if f.lower().endswith(('.png', '.jpg', '.jpeg'))],
-        key=lambda x: os.path.splitext(x)[0]
-    )
-    new_files = [f for f in current_files if f not in processed_files]
-    for filename in new_files:
-        try:
-            img_path = os.path.join(result_folder, filename)
-            frame = cv2.imread(img_path)
-            if frame is not None:
-                frame_buffer.append(frame)
-                processed_files.add(filename)
-        except Exception:
-            pass
+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_bin = "/root/anaconda3/envs/gradio/bin/ffmpeg"
-    if not os.path.exists(ffmpeg_bin):
-        ffmpeg_bin = shutil.which("ffmpeg")
+    
+    # 查找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,
@@ -105,11 +185,13 @@ def convert_to_h264(video_path):
         "-movflags", "+faststart",
         video_path_h264
     ]
-    import subprocess
+    
     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
+        raise gr.Error(f"转换过程中发生错误: {str(e)}")