"""Blender Render Space 的 Gradio 界面、公开 API 和 MCP 入口。""" from __future__ import annotations # ZeroGPU 要求 spaces 在可能加载 CUDA 的库之前导入。 import spaces # noqa: F401 import json import logging import os from collections.abc import Iterator from typing import Any, Callable, TypeVar import gradio as gr import service from blender_runtime import ensure_blender from render_config import RenderConfig, parse_config LOGGER = logging.getLogger(__name__) T = TypeVar("T") CSS = """ .gradio-container { max-width: 1480px !important; margin: 0 auto; } #render-hero { background: linear-gradient(125deg, #102331, #1b4148 62%, #2c6062); border: 1px solid #31565b; border-radius: 20px; padding: 30px 32px; margin: 8px 0 18px; color: #f3fbfc; } #render-hero .eyebrow { font-size: 12px; letter-spacing: .15em; color: #9be3d8; } #render-hero h1 { color: #f3fbfc; font-size: clamp(27px, 4vw, 39px); margin: 7px 0 9px; } #render-hero p { color: #c7dcdf; margin: 0; line-height: 1.75; } #render-hero .pills { display: flex; flex-wrap: wrap; gap: 8px; margin-top: 19px; } #render-hero .pill { font-size: 12px; background: #ffffff12; border: 1px solid #ffffff26; border-radius: 30px; padding: 5px 12px; color: #e4f5f4; } .workflow-note { border-left: 3px solid #3d9e91; padding-left: 12px; } .section-caption { margin-top: 8px; margin-bottom: 0; } #job-actions { align-items: stretch; } @media (max-width: 640px) { #render-hero { padding: 24px 20px; border-radius: 14px; } } """ HERO = """
BLENDER RENDER SPACE

把场景变成作品

上传 Blender 工程与资源，用 JSON 驱动图片或动画渲染。
按批次保存帧，通过 API 与 MCP 接入你的创作 Agent。

默认 GPU + Eevee CPU / Cycles 可选 逐批续渲 图片 / 视频 API + MCP
""" def _invoke(function: Callable[..., T], *args: Any) -> T: """把服务异常转换成可读的 Gradio / API 错误，保留后台诊断日志。""" try: return function(*args) except gr.Error: raise except Exception as exc: LOGGER.exception("Render service request failed: %s", function.__name__) raise gr.Error(str(exc) or "操作失败，请查看 Space 日志。") from exc def _job_id(value: str) -> str: """任务 ID 也是访问任务的凭据，禁止将空字符串传入服务。""" value = (value or "").strip() if not value: raise gr.Error("请先准备任务，或填入之前保存的任务 ID。") return value def environment_info() -> dict[str, Any]: """Inspect the runtime, Blender installation and selected rendering backend. Returns: Environment information without allocating a ZeroGPU lease. """ return _invoke(service.environment_info) def config_schema() -> dict[str, Any]: """Get the JSON Schema for every supported render configuration option. Returns: The authoritative configuration schema, defaults and validation limits. """ return _invoke(service.config_schema) def prepare_job( files: list[str], config_json: str, config_file: str | None = None ) -> tuple[str, dict[str, Any]]: """Upload and validate a Blender project and create a resumable render job. Args: files: Uploaded .blend files, resource ZIP archives and supporting assets. config_json: Render configuration encoded as a JSON object. config_file: Optional uploaded JSON configuration file. Returns: A private job ID and the initial job status. Keep the ID for later calls. """ if not files: raise gr.Error("请上传 .blend 工程，或包含工程及资源的 ZIP 文件。") return _invoke(service.prepare_job, files, config_json or "{}", config_file) def inspect_job(job_id: str) -> dict[str, Any]: """Inspect a Blender scene, cameras and referenced assets before rendering. Args: job_id: The private job ID returned by prepare_job. Returns: Scene inspection results and the current job status. """ return _invoke(service.inspect_job, _job_id(job_id)) def render_chunk(job_id: str) -> dict[str, Any]: """Render the next bounded batch and retain completed frames for resuming. Args: job_id: The private job ID returned by prepare_job. Returns: Updated job status. Call again while pending frames remain, then finalize. """ return _invoke(service.render_chunk, _job_id(job_id)) def _render_remaining(job_id: str) -> Iterator[dict[str, Any]]: """界面专用连续模式：按所选后端逐批渲染，错误或取消后立即结束。""" job_id = _job_id(job_id) current = _invoke(service.job_status, job_id) yield current halted = {"cancelled", "cancelling", "canceled", "failed", "error"} if current.get("status") in {"cancelled", "cancelling", "canceled"}: gr.Warning("任务已停止，请先点击「恢复任务」，再启动连续渲染。") return if current.get("error") or current.get("status") in halted: gr.Warning("任务存在错误。请先检查日志并处理原因，再恢复任务；连续模式不会自动重试。") return if current.get("status") == "rendering": gr.Warning("此任务正在渲染，请等待当前批次完成。") return while current.get("remaining_frames"): # 每轮先检查取消标志；绝不自动清除标志或重试配额错误。 current = _invoke(service.job_status, job_id) if current.get("error") or current.get("status") in halted: yield current return remaining = current.get("remaining_frames", []) if not remaining: yield current return previous_count = len(remaining) current = _invoke(service.render_chunk, job_id) yield current if current.get("error") or current.get("status") in halted: return pending = current.get("remaining_frames", []) if pending and len(pending) >= previous_count: # 避免异常状态无进展却反复租用 GPU。 gr.Warning("本批次未减少剩余帧，连续渲染已停止。请检查任务状态与日志。") return def job_status(job_id: str) -> dict[str, Any]: """Read render progress without waiting for the Blender execution queue. Args: job_id: The private job ID returned by prepare_job. Returns: Current job state and completed-frame progress. """ return _invoke(service.job_status, _job_id(job_id)) def cancel_job(job_id: str) -> dict[str, Any]: """Request cancellation while preserving already completed render frames. Args: job_id: The private job ID returned by prepare_job. Returns: Cancellation status. Use job_status to observe the final state. """ return _invoke(service.cancel_job, _job_id(job_id)) def resume_job(job_id: str) -> dict[str, Any]: """Resume a stopped job while retaining successfully rendered frames. Args: job_id: The private job ID returned by prepare_job. Returns: Updated job state. Call render_chunk to render the next batch. """ return _invoke(service.resume_job, _job_id(job_id)) def finalize_job( job_id: str, ) -> tuple[dict[str, Any], list[str], str | None, str | None]: """Package rendered frames and encode a video when the job is complete. Args: job_id: The private job ID returned by prepare_job. Returns: A result summary, downloadable files, optional image and optional video. """ return _invoke(service.finalize_job, _job_id(job_id)) def cpu_self_test() -> tuple[dict[str, Any], list[str], str | None]: """Verify default CPU Eevee rendering without allocating a GPU.""" return _invoke(service.cpu_self_test) def gpu_self_test() -> tuple[dict[str, Any], list[str], str | None]: """Test actual Blender GPU rendering with a small built-in scene. Returns: Diagnostic results, downloadable logs and an optional test render. A failed GPU test does not silently switch the render job to CPU. """ return _invoke(service.gpu_self_test) def eevee_gpu_self_test(gpu_size: str = "large", graphics_backend: str = "opengl") -> tuple[dict[str, Any], list[str], str | None]: """Check GPU Eevee with a real render in the current server environment.""" return _invoke(service.eevee_gpu_self_test, gpu_size, graphics_backend) def _preset(mode: str) -> str: """从同一配置模型生成示例，避免界面默认值与后端校验发生偏差。""" data = RenderConfig().model_dump(mode="json") data["mode"] = mode return json.dumps(data, indent=2, ensure_ascii=False) def _render_options(config_json: str, engine: str, device: str, gpu_size: str = "large", graphics_backend: str = "opengl") -> str: # Update only render choices; retain camera, animation and quality settings. data = json.loads(config_json or "{}") data.update(engine=engine, device=device, gpu_size=gpu_size, graphics_backend=graphics_backend) return parse_config(json.dumps(data)).model_dump_json(indent=2) def build_app() -> gr.Blocks: """构建界面；导入本模块不会下载 Blender，也不会申请 GPU。""" with gr.Blocks(title="Blender Render Space", delete_cache=(3600, 86400)) as demo: gr.HTML(HERO) gr.Markdown( "**默认 GPU + Eevee**，CPU 和 Cycles 仍可选择。" "开始正式任务前，请先运行「GPU / Eevee 自检」确认实际出图。" "引擎和计算设备可以自由选择。GPU + Eevee 是否可用取决于服务器的图形接口，" "可在「环境自检」中检测；不需要用户在自己电脑安装硬件。", elem_classes="workflow-note", ) with gr.Tabs(): with gr.Tab("渲染工作台"): with gr.Row(): with gr.Column(scale=5, min_width=320): gr.Markdown("### 01 · 上传工程", elem_classes="section-caption") upload = gr.File( label="Blender 工程与资源", file_count="multiple", type="filepath", height=190, ) gr.Markdown( "推荐上传**包含 .blend 与纹理目录的 ZIP**，保留相对路径；" "也可以先在 Blender 中打包外部资源，再上传 .blend。" "多个单独上传文件的目录结构不会自动保留。" ) config_upload = gr.File( label="渲染配置 JSON（可选）", file_types=[".json"], type="filepath", ) gr.Markdown("### 02 · 设置渲染", elem_classes="section-caption") with gr.Row(): image_preset = gr.Button("图片配置", size="sm") video_preset = gr.Button("视频配置", size="sm") config_editor = gr.Code( value=_preset("image"), language="json", label="JSON 配置编辑器", lines=19, max_lines=32, interactive=True, ) gr.Markdown( "上传 JSON 文件后，将使用文件中的配置。" "使用编辑器时，请先清空上方 JSON 上传框。" "`batch_size` 控制每次提交的帧数；建议先用低分辨率、" "少量采样验证，再提高画质。" ) with gr.Row(): engine_choice = gr.Dropdown([("Eevee（默认）", "EEVEE"), ("Cycles", "CYCLES")], value="EEVEE", label="渲染引擎") device_choice = gr.Dropdown([("GPU（默认）", "AUTO"), ("CPU", "CPU")], value="AUTO", label="计算设备") with gr.Row(): gpu_size_choice = gr.Dropdown([("large · 48 GB", "large"), ("xlarge · 96 GB · 2 倍配额", "xlarge")], value="large", label="ZeroGPU 规格") graphics_backend_choice = gr.Dropdown([("OpenGL", "opengl"), ("Vulkan", "vulkan")], value="opengl", label="GPU Eevee 图形接口") apply_options = gr.Button("将引擎与设备应用到 JSON", size="sm") gr.Markdown("选择引擎或设备会自动更新编辑器中的 JSON；创建任务以 JSON 为准。上传 JSON 文件时，请直接修改文件中的 engine / device。GPU + Eevee 可先运行专用自检。") prepare = gr.Button("准备渲染任务", variant="primary", size="lg") with gr.Column(scale=6, min_width=320): gr.Markdown("### 03 · 检查与渲染", elem_classes="section-caption") job = gr.Textbox( label="任务 ID", placeholder="准备后自动填入，也可粘贴已有任务 ID", interactive=True, buttons=["copy"], ) gr.Markdown( "保存任务 ID，便于 API 调用和续渲。" "任务 ID 可访问对应结果，请勿公开。Space 重启可能清除临时任务。" ) with gr.Row(elem_id="job-actions"): inspect = gr.Button("检查场景与资源") render = gr.Button("渲染下一批", variant="primary") render_remaining = gr.Button("连续渲染剩余帧", variant="primary") with gr.Row(): refresh = gr.Button("刷新进度", size="sm") cancel = gr.Button("停止任务", variant="stop", size="sm") resume = gr.Button("恢复任务", size="sm") status = gr.JSON(label="任务状态 / 场景检查", value={"status": "请先准备任务"}) gr.Markdown( "每次点击「渲染下一批」处理一个批次；剩余帧可继续提交。" "「连续渲染剩余帧」会逐批提交，遇到错误或配额不足立即结束，" "不会自动重试。**仅选择 GPU 时消耗 GPU 配额；可随时点击「停止任务」。**" "API / Agent 也可按状态自动循环。" "停止后保留已完成帧；点击「恢复任务」后再提交渲染。" ) gr.Markdown("### 04 · 导出作品", elem_classes="section-caption") finalize = gr.Button("整理结果 / 合成视频", variant="primary") result = gr.JSON(label="导出结果") downloads = gr.File( label="下载结果与日志", file_count="multiple", interactive=False ) with gr.Tabs(): with gr.Tab("图片预览"): image = gr.Image( label="渲染图片", type="filepath", interactive=False, height=340, buttons=["download", "fullscreen"], ) with gr.Tab("视频预览"): video = gr.Video(label="渲染视频", interactive=False, height=340) with gr.Tab("环境自检"): gr.Markdown( "### 先验证运行环境\n\n" "「读取环境」不会申请 GPU。" "「运行 GPU 自检」会尝试实际渲染一个小场景，可能消耗 ZeroGPU 配额。" "自检失败时，请下载日志检查原因，或选择常驻 GPU Space。\n\n" "后端由 Space 环境变量 `RENDER_BACKEND` 选择：" "`auto`、`zerogpu`、`gpu` 或 `cpu`。" "GPU + Eevee 是默认配置，但可用性必须通过实际出图确认。任务 JSON 的 device=CPU 始终绕过 GPU 租约；GPU 不进行静默 CPU 回退。" ) with gr.Row(): env_button = gr.Button("读取环境") cpu_probe_button = gr.Button("运行 CPU / Eevee 自检") probe_button = gr.Button("运行 GPU / Cycles 自检") eevee_probe_button = gr.Button("运行 GPU / Eevee 自检", variant="primary") gr.Markdown("GPU / Eevee 自检使用工作台当前选择的 ZeroGPU 规格和图形接口；xlarge 消耗 2 倍配额，规格更大不代表具备图形能力。") env_output = gr.JSON(label="环境信息") with gr.Row(): with gr.Column(): probe_output = gr.JSON(label="自检报告") probe_files = gr.File( label="下载自检日志", file_count="multiple", interactive=False ) probe_image = gr.Image( label="自检渲染图", type="filepath", interactive=False, height=300 ) with gr.Tab("Agent / API"): gr.Markdown( "### 可直接调用的渲染工具\n\n" "API 名称与下表一致。点击页面底部 **Use via API** 查看当前 Space " "地址、参数格式与代码示例。所有接口也作为 MCP 工具提供。\n\n" "| API | 用途 |\n| --- | --- |\n" "| `environment_info` | 检查后端与安装信息 |\n" "| `config_schema` | 获取完整 JSON Schema |\n" "| `cpu_self_test` | 验证可选 CPU Eevee 渲染 |\n" "| `eevee_gpu_self_test` | 检测当前服务器的 GPU + Eevee 实际出图能力 |\n" "| `gpu_self_test` | 验证 Blender 实际 GPU 渲染 |\n" "| `prepare_job` | 上传工程和配置，创建任务 |\n" "| `inspect_job` | 检查场景、相机与资源 |\n" "| `render_chunk` | 渲染下一批，保存已完成帧 |\n" "| `job_status` | 不等待渲染队列，读取进度 |\n" "| `cancel_job` | 请求停止，保留已完成帧 |\n" "| `resume_job` | 将停止的任务恢复为可继续状态 |\n" "| `finalize_job` | 导出结果、日志并合成视频 |\n\n" "**建议调用顺序：** 获取配置 Schema → 准备任务 → 检查场景 → " "循环调用 `render_chunk` → 确认完成 → 调用 `finalize_job`。" "遇到错误先读取状态与日志；不要无条件重复申请 GPU。\n\n" "### MCP 连接\n\n" "Streamable HTTP：`https://<你的-Space-域名>.hf.space/gradio_api/mcp/`\n\n" "兼容 SSE：`https://<你的-Space-域名>.hf.space/gradio_api/mcp/sse`\n\n" "MCP 工具使用上传后的文件引用；Agent 本机路径不会自动出现在 Space 中。" "上传本地大文件优先使用 `gradio_client.handle_file`，" "后续操作只需传任务 ID。私有 Space 的客户端还需要 Hugging Face 令牌。" ) schema_button = gr.Button("读取配置 JSON Schema") schema_output = gr.JSON(label="配置 Schema") # Blender 操作共用一个执行队列，防止多个进程争用同一个 GPU 或输出目录。 blender_queue = {"concurrency_id": "blender", "concurrency_limit": 1} prepare.click( prepare_job, inputs=[upload, config_editor, config_upload], outputs=[job, status], api_name="prepare_job", **blender_queue, ) inspect.click(inspect_job, job, status, api_name="inspect_job", **blender_queue) render.click(render_chunk, job, status, api_name="render_chunk", **blender_queue) render_remaining.click( _render_remaining, job, status, api_visibility="private", **blender_queue, ) # 进度与停止请求不进入渲染队列，否则长渲染会阻塞取消操作。 refresh.click(job_status, job, status, api_name="job_status", queue=False) cancel.click(cancel_job, job, status, api_name="cancel_job", queue=False) resume.click(resume_job, job, status, api_name="resume_job", queue=False) finalize.click( finalize_job, job, [result, downloads, image, video], api_name="finalize_job", **blender_queue, ) env_button.click(environment_info, None, env_output, api_name="environment_info", queue=False) schema_button.click(config_schema, None, schema_output, api_name="config_schema", queue=False) option_inputs = [config_editor, engine_choice, device_choice, gpu_size_choice, graphics_backend_choice] for choice in (engine_choice, device_choice, gpu_size_choice, graphics_backend_choice): choice.input(_render_options, option_inputs, config_editor, queue=False, api_visibility="private") eevee_probe_button.click(eevee_gpu_self_test, [gpu_size_choice, graphics_backend_choice], [probe_output, probe_files, probe_image], api_name="eevee_gpu_self_test", **blender_queue) apply_options.click(_render_options, option_inputs, config_editor, queue=False, api_visibility="private") cpu_probe_button.click(cpu_self_test, None, [probe_output, probe_files, probe_image], api_name="cpu_self_test", **blender_queue) probe_button.click( gpu_self_test, None, [probe_output, probe_files, probe_image], api_name="gpu_self_test", **blender_queue, ) # 纯界面快捷操作不作为 Agent 工具暴露。 image_preset.click( lambda: _preset("image"), None, config_editor, queue=False, api_visibility="private" ) video_preset.click( lambda: _preset("video"), None, config_editor, queue=False, api_visibility="private" ) return demo.queue(default_concurrency_limit=1, max_size=32) demo = build_app() if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") # 安装与解压在申请 GPU 之前完成，避免把下载时间算进 ZeroGPU 租约。 LOGGER.info("正在准备 Blender 运行时（首次运行需要下载并校验安装包）") binary = ensure_blender() LOGGER.info("Blender 运行时已就绪：%s", binary) service.PUBLIC_ROOT.mkdir(parents=True, exist_ok=True) demo.launch( server_name=os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0"), server_port=int(os.environ.get("PORT", os.environ.get("GRADIO_SERVER_PORT", "7860"))), mcp_server=True, max_file_size="1gb", theme=gr.themes.Soft(primary_hue="teal", neutral_hue="slate"), css=CSS, # 仅发布服务显式导出的文件；不允许浏览整个任务目录或上传源文件。 allowed_paths=[str(service.PUBLIC_ROOT.resolve())], footer_links=["api"], show_error=False, )