feat(scene-engine): add semantic scene generation and editing

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/geometry_generation.py

@@ -25,6 +25,8 @@
 
 import requests
 
+from embodichain.gen_sim.scene_engine.errors import SceneServiceError
+
 from embodichain.gen_sim.scene_engine.configs.environment import (
     read_scene_engine_env_values,
 )
@@ -77,7 +79,7 @@ def check_health(self) -> None:
                 last_error = exc
 
         assert last_error is not None
-        raise RuntimeError(
+        raise SceneServiceError(
             "Geometry Generation Server health check failed after "
             f"{self._max_attempts} attempts."
         ) from last_error
@@ -91,6 +93,7 @@ def generate_objects(
         image_path: str | Path,
         object_masks: list[tuple[str, Path]],
         output_root: str | Path,
+        seed: int | None = None,
     ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
         """Generate objects through the geometry server's mask-list endpoint.
 
@@ -125,6 +128,7 @@ def generate_objects(
         response_data, response_objects = self._request_objects(
             image_path=resolved_image_path,
             object_masks=resolved_object_masks,
+            seed=seed,
         )
 
         resolved_output_root = Path(output_root).expanduser().resolve()
@@ -158,6 +162,7 @@ def _request_objects(
         *,
         image_path: Path,
         object_masks: list[tuple[str, Path]],
+        seed: int | None,
     ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
         last_error: Exception | None = None
         for _ in range(self._max_attempts):
@@ -171,6 +176,7 @@ def _request_objects(
                     ]
                     response = self._session.post(
                         self._url(self._generate_objects_path),
+                        data=(None if seed is None else {"seed": str(int(seed))}),
                         files=[
                             (
                                 "image",
@@ -201,6 +207,11 @@ def _request_objects(
                         "Geometry Generation Server response is not valid JSON."
                     ) from exc
                 response_data = self._wait_for_task_if_needed(response_data)
+                if seed is not None and _response_seed(response_data) != int(seed):
+                    raise RuntimeError(
+                        "Geometry Generation Server did not acknowledge the "
+                        f"requested seed {int(seed)}."
+                    )
                 response_objects = _parse_objects_response(
                     response_data,
                     object_ids=[object_id for object_id, _ in object_masks],
@@ -210,7 +221,7 @@ def _request_objects(
                 last_error = exc
 
         assert last_error is not None
-        raise RuntimeError(
+        raise SceneServiceError(
             "Geometry Generation Server request failed after "
             f"{self._max_attempts} attempts."
         ) from last_error
@@ -294,7 +305,7 @@ def _download_glb(self, mesh_path: str, output_path: Path) -> None:
                 last_error = exc
 
         assert last_error is not None
-        raise RuntimeError(
+        raise SceneServiceError(
             "Geometry Generation Server GLB download failed after "
             f"{self._max_attempts} attempts: {mesh_path}"
         ) from last_error
@@ -374,6 +385,25 @@ def _parse_objects_response(
     return parsed_objects
 
 
+def _response_seed(value: object) -> int | None:
+    """Return a seed acknowledged by a geometry response envelope."""
+    if not isinstance(value, dict):
+        return None
+    candidates = [value.get("seed")]
+    for key in ("result", "metadata"):
+        nested = value.get(key)
+        if isinstance(nested, dict):
+            candidates.append(nested.get("seed"))
+    for candidate in candidates:
+        if candidate is None or isinstance(candidate, bool):
+            continue
+        try:
+            return int(candidate)
+        except (TypeError, ValueError):
+            continue
+    return None
+
+
 def _parse_numeric_list(
     value: object,
     *,

embodichain/gen_sim/scene_engine/clients/image_generation.py

@@ -0,0 +1,192 @@
+# ----------------------------------------------------------------------------
+# 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.errors import SceneServiceError
+
+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 SceneServiceError(
+            "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,
+        seed: int | None = None,
+    ) -> 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,
+                        **({} if seed is None else {"seed": int(seed)}),
+                    },
+                    timeout=self._timeout_s,
+                )
+                response.raise_for_status()
+                if seed is not None:
+                    acknowledged = response.headers.get(
+                        "x-generation-seed",
+                        response.headers.get("x-seed"),
+                    )
+                    try:
+                        acknowledged_seed = int(acknowledged)
+                    except (TypeError, ValueError):
+                        acknowledged_seed = None
+                    if acknowledged_seed != int(seed):
+                        raise RuntimeError(
+                            "Image Generation Server did not acknowledge the "
+                            f"requested seed {int(seed)}."
+                        )
+                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 SceneServiceError(
+            "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(),
+    }