feat(scene-engine): add text-guided scene editing

docs/source/features/generative_sim/scene_engine.md

@@ -27,10 +27,27 @@ python -m embodichain scene-engine \
     --output_root /path/to/scene_output
 ```
 
+## Scene Editing
+
+Edit an existing valid Scene Engine output with an instruction:
+
+```bash
+embodichain scene-engine \
+    --output_root /path/to/scene_output \
+    --edit_prompt "add a red cup to the front-center of the tabletop"
+```
+
+`--image` and `--edit_prompt` may also be provided together. Scene Engine then
+generates the image-based scene first and applies the edit to that export. An
+edit-only invocation requires an existing `scene_export` directory. The edit
+overwrites its `scene_config.json`, `scene_graph.json`, `scene.json`, and final
+`mesh_assets`; intermediate generation and edit artifacts remain available for
+debugging.
+
 ## Configuration
 
-Scene Engine reads the LLM, segmentation, and geometry-generation settings
-from `embodichain/gen_sim/.env`:
+Scene Engine reads the LLM, segmentation, image-generation, and
+geometry-generation settings from `embodichain/gen_sim/.env`:
 
 ```bash
 OPENAI_API_KEY="your-api-key"
@@ -43,7 +60,13 @@ SCENE_ENGINE_IMAGE_SEGMENTATION_BASE_URL="http://host:port"
 SCENE_ENGINE_IMAGE_SEGMENTATION_TIMEOUT_S=30
 SCENE_ENGINE_IMAGE_SEGMENTATION_MAX_ATTEMPTS=3
 SCENE_ENGINE_IMAGE_SEGMENTATION_HEALTH_PATH="/health"
-SCENE_ENGINE_IMAGE_SEGMENTATION_SINGLE_OBJECT_PATH="/predict"
+SCENE_ENGINE_IMAGE_SEGMENTATION_BY_PROMPT_PATH="/segment_by_prompt"
+
+SCENE_ENGINE_IMAGE_GENERATION_BASE_URL="http://host:port"
+SCENE_ENGINE_IMAGE_GENERATION_TIMEOUT_S=120
+SCENE_ENGINE_IMAGE_GENERATION_MAX_ATTEMPTS=3
+SCENE_ENGINE_IMAGE_GENERATION_HEALTH_PATH="/health"
+SCENE_ENGINE_IMAGE_GENERATION_BY_PROMPT_PATH="/generate_image_by_prompt"
 
 SCENE_ENGINE_GEOMETRY_GENERATION_BASE_URL="http://host:port"
 SCENE_ENGINE_GEOMETRY_GENERATION_TIMEOUT_S=600
@@ -66,9 +89,12 @@ The important final outputs are:
 scene_output/
 |-- scene_understanding/     # Object analysis, masks, and stage JSON
 |-- scene_generation/        # Generated, SimReady, and layout-debug artifacts
+|-- scene_editing/           # Present after edits; generated asset/debug artifacts
 `-- scene_export/
     |-- mesh_assets/         # Final GLBs
-    `-- scene_config.json    # Exported scene description
+    |-- scene_config.json    # Exported z-up scene description
+    |-- scene_graph.json     # Table support and planar relation graph
+    `-- scene.json           # Scene Engine object metadata and y-up poses
 ```
 
 Validate the export without opening a window:

embodichain/gen_sim/scene_engine/cli/start.py

@@ -21,15 +21,31 @@
 from pathlib import Path
 
 from embodichain.gen_sim.scene_engine.pipeline.generate import generate_scene_from_image
+from embodichain.gen_sim.scene_engine.pipeline.edit import edit_scene
 
 _SUPPORTED_IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png"}
 
 
 def cli_scene_engine(
-    image: str | Path,
+    image: str | Path | None,
     output_root: str | Path,
+    *,
+    edit_prompt: str | None = None,
 ) -> None:
-    """Generate one scene using the required ``gen_sim/.env`` settings."""
+    """Generate a scene from an image, edit an export, or do both in sequence."""
+    resolved_output_root = Path(output_root).expanduser().resolve()
+    if edit_prompt is not None:
+        edit_prompt = edit_prompt.strip()
+        if not edit_prompt:
+            raise ValueError("Edit prompt must not be empty.")
+
+    if image is None:
+        if edit_prompt is None:
+            raise ValueError("Provide --image, --edit_prompt, or both.")
+        edit_scene(output_root=resolved_output_root, edit_prompt=edit_prompt)
+        print("Successfully completed!")
+        return
+
     resolved_image_path = Path(image).expanduser().resolve()
     if not resolved_image_path.exists():
         raise FileNotFoundError(f"Image input not found: {resolved_image_path}")
@@ -40,37 +56,46 @@ def cli_scene_engine(
             "Image input must have one of these extensions: .jpg, .jpeg, .png"
         )
 
-    resolved_output_root = Path(output_root).expanduser().resolve()
     resolved_output_root.mkdir(parents=True, exist_ok=True)
-
     generate_scene_from_image(
         image_path=resolved_image_path,
         output_root=resolved_output_root,
     )
+    if edit_prompt is not None:
+        edit_scene(
+            output_root=resolved_output_root,
+            edit_prompt=edit_prompt,
+        )
     print("Successfully completed!")
 
 
 def main(argv: Sequence[str] | None = None) -> None:
     parser = argparse.ArgumentParser(
         prog="embodichain scene-engine",
-        description="Generate a Scene Engine export from one input image.",
+        description="Generate a Scene Engine export, edit one, or do both.",
         epilog="Service settings are read from embodichain/gen_sim/.env.",
     )
     parser.add_argument(
         "--image",
         type=str,
-        required=True,
-        help="Path to the required input image file (.jpg, .jpeg, or .png)",
+        required=False,
+        help="Optional input image file (.jpg, .jpeg, or .png)",
     )
     parser.add_argument(
         "--output_root",
         type=str,
         required=True,
         help="Path to the output directory",
     )
+    parser.add_argument(
+        "--edit_prompt",
+        type=str,
+        default=None,
+        help="Text instruction for editing an existing or newly generated output root",
+    )
     args = parser.parse_args(argv)
 
-    cli_scene_engine(args.image, args.output_root)
+    cli_scene_engine(args.image, args.output_root, edit_prompt=args.edit_prompt)
 
 
 if __name__ == "__main__":

embodichain/gen_sim/scene_engine/clients/image_generation.py

@@ -0,0 +1,172 @@
+# ----------------------------------------------------------------------------
+# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ----------------------------------------------------------------------------
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import requests
+
+from embodichain.gen_sim.scene_engine.configs.environment import (
+    read_scene_engine_env_values,
+)
+
+
+class ImageGenerationClient:
+    """Manage the Image Generation Server connection."""
+
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        timeout_s: int,
+        max_attempts: int,
+        health_path: str,
+        generate_image_by_prompt_path: str,
+        session: requests.Session | None = None,
+    ) -> None:
+        self._base_url = base_url.rstrip("/")
+        self._timeout_s = timeout_s
+        self._max_attempts = max_attempts
+        self._health_path = health_path
+        self._generate_image_by_prompt_path = generate_image_by_prompt_path
+        self._session = session or requests.Session()
+
+    @classmethod
+    def from_dotenv(cls) -> "ImageGenerationClient":
+        """Create a client from its required ``gen_sim/.env`` settings."""
+        return cls(**_load_dotenv_config())
+
+    def check_health(self) -> None:
+        last_error: requests.RequestException | RuntimeError | None = None
+        for _ in range(self._max_attempts):
+            try:
+                response = self._session.get(
+                    self._url(self._health_path),
+                    timeout=10,  # Use a shorter timeout for avoiding long waits.
+                )
+                response.raise_for_status()
+                response_data = response.json()
+                if (
+                    not isinstance(response_data, dict)
+                    or response_data.get("ok") is not True
+                ):
+                    raise RuntimeError(
+                        "Image Generation Server health response does not contain ok=true."
+                    )
+                return
+            except (requests.RequestException, ValueError, RuntimeError) as exc:
+                last_error = exc
+
+        assert last_error is not None
+        raise RuntimeError(
+            "Image Generation Server health check failed after "
+            f"{self._max_attempts} attempts."
+        ) from last_error
+
+    def close(self) -> None:
+        self._session.close()
+
+    def generate_image_by_prompt(
+        self,
+        *,
+        prompt: str,
+        output_path: str | Path,
+    ) -> Path:
+        """Generate one PNG image from ``prompt`` and save it to ``output_path``."""
+        prompt = prompt.strip()
+        if not prompt:
+            raise ValueError("Image generation prompt must not be empty.")
+
+        resolved_output_path = Path(output_path).expanduser().resolve()
+        resolved_output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        last_error: Exception | None = None
+        for _ in range(self._max_attempts):
+            try:
+                response = self._session.post(
+                    self._url(self._generate_image_by_prompt_path),
+                    json={"prompt": prompt},
+                    timeout=self._timeout_s,
+                )
+                response.raise_for_status()
+                content_type = response.headers.get("content-type", "").split(";")[0]
+                if content_type != "image/png":
+                    raise RuntimeError(
+                        "Image Generation Server response is not a PNG image."
+                    )
+                resolved_output_path.write_bytes(response.content)
+                return resolved_output_path
+            except (requests.RequestException, RuntimeError) as exc:
+                last_error = exc
+
+        assert last_error is not None
+        raise RuntimeError(
+            "Image Generation Server request failed after "
+            f"{self._max_attempts} attempts."
+        ) from last_error
+
+    def _url(self, path: str) -> str:
+        return f"{self._base_url}/{path.lstrip('/')}"
+
+
+def _load_dotenv_config() -> dict[str, Any]:
+    values = read_scene_engine_env_values(
+        "SCENE_ENGINE_IMAGE_GENERATION_BASE_URL",
+        "SCENE_ENGINE_IMAGE_GENERATION_TIMEOUT_S",
+        "SCENE_ENGINE_IMAGE_GENERATION_MAX_ATTEMPTS",
+        "SCENE_ENGINE_IMAGE_GENERATION_HEALTH_PATH",
+        "SCENE_ENGINE_IMAGE_GENERATION_BY_PROMPT_PATH",
+    )
+    try:
+        timeout_s = int(values["SCENE_ENGINE_IMAGE_GENERATION_TIMEOUT_S"])
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            "SCENE_ENGINE_IMAGE_GENERATION_TIMEOUT_S must be an integer."
+        ) from exc
+    if timeout_s < 1:
+        raise ValueError("SCENE_ENGINE_IMAGE_GENERATION_TIMEOUT_S must be at least 1.")
+
+    try:
+        max_attempts = int(values["SCENE_ENGINE_IMAGE_GENERATION_MAX_ATTEMPTS"])
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            "SCENE_ENGINE_IMAGE_GENERATION_MAX_ATTEMPTS must be an integer."
+        ) from exc
+    if max_attempts < 1:
+        raise ValueError(
+            "SCENE_ENGINE_IMAGE_GENERATION_MAX_ATTEMPTS must be at least 1."
+        )
+
+    string_keys = (
+        "SCENE_ENGINE_IMAGE_GENERATION_BASE_URL",
+        "SCENE_ENGINE_IMAGE_GENERATION_HEALTH_PATH",
+        "SCENE_ENGINE_IMAGE_GENERATION_BY_PROMPT_PATH",
+    )
+    for key in string_keys:
+        if not values[key].strip():
+            raise ValueError(f"Scene Engine .env key {key} must be a non-empty string.")
+
+    return {
+        "base_url": values["SCENE_ENGINE_IMAGE_GENERATION_BASE_URL"].strip(),
+        "timeout_s": timeout_s,
+        "max_attempts": max_attempts,
+        "health_path": values["SCENE_ENGINE_IMAGE_GENERATION_HEALTH_PATH"].strip(),
+        "generate_image_by_prompt_path": values[
+            "SCENE_ENGINE_IMAGE_GENERATION_BY_PROMPT_PATH"
+        ].strip(),
+    }

embodichain/gen_sim/scene_engine/clients/image_segmentation.py

@@ -36,14 +36,14 @@ def __init__(
         timeout_s: int,
         max_attempts: int,
         health_path: str,
-        segment_single_object_path: str,
+        segment_by_prompt_path: str,
         session: requests.Session | None = None,
     ) -> None:
         self._base_url = base_url.rstrip("/")
         self._timeout_s = timeout_s
         self._max_attempts = max_attempts
         self._health_path = health_path
-        self._segment_single_object_path = segment_single_object_path
+        self._segment_by_prompt_path = segment_by_prompt_path
         self._session = session or requests.Session()
 
     @classmethod
@@ -96,7 +96,7 @@ def segment_single_object(
             try:
                 with resolved_image_path.open("rb") as image_file:
                     response = self._session.post(
-                        self._url(self._segment_single_object_path),
+                        self._url(self._segment_by_prompt_path),
                         data={"prompt": prompt},
                         files={"image": (resolved_image_path.name, image_file)},
                         timeout=self._timeout_s,
@@ -138,7 +138,7 @@ def _load_dotenv_config() -> dict[str, Any]:
         "SCENE_ENGINE_IMAGE_SEGMENTATION_TIMEOUT_S",
         "SCENE_ENGINE_IMAGE_SEGMENTATION_MAX_ATTEMPTS",
         "SCENE_ENGINE_IMAGE_SEGMENTATION_HEALTH_PATH",
-        "SCENE_ENGINE_IMAGE_SEGMENTATION_SINGLE_OBJECT_PATH",
+        "SCENE_ENGINE_IMAGE_SEGMENTATION_BY_PROMPT_PATH",
     )
     try:
         timeout_s = int(values["SCENE_ENGINE_IMAGE_SEGMENTATION_TIMEOUT_S"])
@@ -165,7 +165,7 @@ def _load_dotenv_config() -> dict[str, Any]:
     string_keys = (
         "SCENE_ENGINE_IMAGE_SEGMENTATION_BASE_URL",
         "SCENE_ENGINE_IMAGE_SEGMENTATION_HEALTH_PATH",
-        "SCENE_ENGINE_IMAGE_SEGMENTATION_SINGLE_OBJECT_PATH",
+        "SCENE_ENGINE_IMAGE_SEGMENTATION_BY_PROMPT_PATH",
     )
     for key in string_keys:
         if not values[key].strip():
@@ -176,8 +176,8 @@ def _load_dotenv_config() -> dict[str, Any]:
         "timeout_s": timeout_s,
         "max_attempts": max_attempts,
         "health_path": values["SCENE_ENGINE_IMAGE_SEGMENTATION_HEALTH_PATH"].strip(),
-        "segment_single_object_path": values[
-            "SCENE_ENGINE_IMAGE_SEGMENTATION_SINGLE_OBJECT_PATH"
+        "segment_by_prompt_path": values[
+            "SCENE_ENGINE_IMAGE_SEGMENTATION_BY_PROMPT_PATH"
         ].strip(),
     }